laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-30 19:46:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

node: make the shared context the sole store of immutable state

Browse Source 
Remove the duplicated immutable fields (config, identity, startup_epoch,
started_at, is_leaf_only, max_connections/peers/links) from the Node
struct so the Arc<NodeContext> bundle is the single source of truth.
Previously Node owned these fields and a parallel context copy, kept in
lockstep by rebuild_context() at every mutation site — pure overhead that
existed only because of the duplication.

- Replace rebuild_context() with replace_context(): a clone-edit-swap of
  the whole Arc. The per-instance context stays immutable; mutation swaps
  the Arc. This is the sole runtime mutation path (constructors, leaf_only,
  update_peers).
- Add Copy-returning accessors startup_epoch() and max_connections()/
  max_peers()/max_links(); migrate the remaining direct field readers onto
  the accessors. node_addr()/npub()/Debug now read identity/is_leaf_only
  from the context.
- update_peers reads the pre-update peer set from the live context Arc
  before building a fresh Config + context and swapping — preserving the
  read-before-write ordering its mutation-window test depends on.
- Remove the test-only set_max_* setters; tests set the limits on Config at
  construction instead (new make_node_with_max_peers/links helpers).
- Add a ci-local guard that fails if the Node struct re-declares a bundled
  field, so the single-store invariant can't silently regress.

cargo test --lib 1291/0; clippy -D warnings and release build clean.
This commit is contained in:
Johnathan Corgan
2026-06-02 16:42:05 +00:00
parent 08b8b3908e
commit d9a4a7807c
16 changed files with 208 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/context.rs
+10 -17
View File
@@ -1,16 +1,17 @@
//! Shared immutable context bundle.
//!
//! [`NodeContext`] groups the [`Node`](super::Node)'s effectively-immutable
//! fields behind a single `Arc` so that handlers can eventually borrow a
//! cheap `&NodeContext` clone instead of `&self`.
//! fields behind a single `Arc` so that handlers can borrow a cheap
//! `&NodeContext` clone instead of `&self`.
//!
//! During the migration it is a *parallel, authoritative* copy of the
//! corresponding `Node` fields: both are kept in lockstep at the only three
//! mutation points — the constructor, [`update_peers`](super::Node::update_peers),
//! and the test-only `set_max_*` setters — via
//! [`Node::rebuild_context`](super::Node::rebuild_context). Readers migrate
//! onto the bundle incrementally; the duplicated `Node` fields are removed
//! once the last reader has moved over.
//! It is the **sole store** of these fields: the `Node` no longer keeps
//! duplicate copies. The bundle itself is immutable; the rare mutation of a
//! bundled field (the constructors, [`leaf_only`](super::Node::leaf_only),
//! and [`update_peers`](super::Node::update_peers)) is done by building a fresh
//! `NodeContext` and swapping the whole `Arc` via
//! [`Node::replace_context`](super::Node::replace_context). Readers reach the
//! fields through the `Node` accessors (`config()`, `identity()`,
//! `startup_epoch()`, `is_leaf_only()`, `max_*()`, `uptime()`).
use std::sync::Arc;
@@ -27,8 +28,6 @@ pub(crate) struct NodeContext {
pub identity: Identity,
/// Random epoch generated at startup for peer restart detection.
// Consumed by readers migrating in a later sub-PR.
#[allow(dead_code)]
pub startup_epoch: [u8; 8],
/// Instant when the node was created, for uptime reporting.
@@ -38,18 +37,12 @@ pub(crate) struct NodeContext {
pub is_leaf_only: bool,
/// Maximum connections (0 = unlimited).
// Consumed by readers migrating in a later sub-PR.
#[allow(dead_code)]
pub max_connections: usize,
/// Maximum peers (0 = unlimited).
// Consumed by readers migrating in a later sub-PR.
#[allow(dead_code)]
pub max_peers: usize,
/// Maximum links (0 = unlimited).
// Consumed by readers migrating in a later sub-PR.
#[allow(dead_code)]
pub max_links: usize,
}
src/node/handlers/handshake.rs
+5 -5
View File
@@ -181,7 +181,7 @@ impl Node {
let noise_msg1 = &packet.data[header.noise_msg1_offset..];
let msg2_response = match conn.receive_handshake_init(
our_keypair,
self.startup_epoch,
self.startup_epoch(),
noise_msg1,
packet.timestamp_ms,
) {
@@ -227,7 +227,7 @@ impl Node {
// handshake whose data frames subsequently fail decryption here).
// The late cap check inside promote_connection() is intentionally
// left in place as defense-in-depth.
if self.max_peers > 0 && self.peers.len() >= self.max_peers {
if self.max_peers() > 0 && self.peers.len() >= self.max_peers() {
let is_known_active = self.peers.contains_key(&peer_node_addr);
let is_pending_outbound = self.connections.iter().any(|(_, conn)| {
conn.expected_identity()
@@ -237,7 +237,7 @@ impl Node {
if !is_known_active && !is_pending_outbound {
debug!(
peer = %self.peer_display_name(&peer_node_addr),
max = self.max_peers,
max = self.max_peers(),
"Silent-dropping Msg1 at max_peers cap (early gate; no Msg2 sent)"
);
// `link_id` was allocated above but `conn` is still a local
@@ -1244,10 +1244,10 @@ impl Node {
}
// Normal promotion
if self.max_peers > 0 && self.peers.len() >= self.max_peers {
if self.max_peers() > 0 && self.peers.len() >= self.max_peers() {
let _ = self.index_allocator.free(our_index);
return Err(NodeError::MaxPeersExceeded {
max: self.max_peers,
max: self.max_peers(),
});
}
src/node/handlers/rekey.rs
+2 -2
View File
@@ -196,7 +196,7 @@ impl Node {
// Create IK initiator handshake directly (no PeerConnection)
let our_keypair = self.identity().keypair();
let mut hs = HandshakeState::new_initiator(our_keypair, peer_pubkey);
hs.set_local_epoch(self.startup_epoch);
hs.set_local_epoch(self.startup_epoch());
let noise_msg1 = match hs.write_message_1() {
Ok(msg) => msg,
@@ -529,7 +529,7 @@ impl Node {
// Create Noise XK initiator handshake
let our_keypair = self.identity().keypair();
let mut handshake = HandshakeState::new_xk_initiator(our_keypair, dest_pubkey);
handshake.set_local_epoch(self.startup_epoch);
handshake.set_local_epoch(self.startup_epoch());
let msg1 = match handshake.write_xk_message_1() {
Ok(m) => m,
src/node/handlers/session.rs
+3 -3
View File
@@ -507,7 +507,7 @@ impl Node {
}
let our_keypair = self.identity().keypair();
let mut handshake = HandshakeState::new_xk_responder(our_keypair);
handshake.set_local_epoch(self.startup_epoch);
handshake.set_local_epoch(self.startup_epoch());
if let Err(e) = handshake.read_xk_message_1(&setup.handshake_payload) {
debug!(error = %e, "Failed to process rekey XK msg1");
@@ -557,7 +557,7 @@ impl Node {
// Create XK responder handshake and process msg1
let our_keypair = self.identity().keypair();
let mut handshake = HandshakeState::new_xk_responder(our_keypair);
handshake.set_local_epoch(self.startup_epoch);
handshake.set_local_epoch(self.startup_epoch());
if let Err(e) = handshake.read_xk_message_1(&setup.handshake_payload) {
debug!(error = %e, "Failed to process Noise XK msg1 in SessionSetup");
@@ -1333,7 +1333,7 @@ impl Node {
// Create Noise XK initiator handshake
let our_keypair = self.identity().keypair();
let mut handshake = HandshakeState::new_xk_initiator(our_keypair, dest_pubkey);
handshake.set_local_epoch(self.startup_epoch);
handshake.set_local_epoch(self.startup_epoch());
let msg1 = handshake
.write_xk_message_1()
.map_err(|e| NodeError::SendFailed {
src/node/lifecycle.rs
+22 -18
View File
@@ -55,11 +55,14 @@ impl Node {
new_by_addr.insert(*identity.node_addr(), peer);
}
// Read the current peer set directly from the field: update_peers is the
// config-source mutation owner (it writes self.config.peers below and then
// rebuilds the context), so it manages config.peers directly rather than
// through the context accessor — same rationale as the write at line ~124.
// Read the current peer set from the context *before* the swap below:
// update_peers is the config-source mutation owner. It reads the current
// (pre-update) peer set here, builds a fresh Config + context, then swaps
// the whole Arc. Reading the live context Arc before the swap yields the
// pre-update set the diff needs (the `update_peers_races_*` canary depends
// on this ordering).
let current_by_addr: HashMap<NodeAddr, PeerConfig> = self
.context
.config
.peers()
.iter()
@@ -124,8 +127,9 @@ impl Node {
.map(|node_addr| new_by_addr[node_addr].clone())
.collect();
self.config.peers = new_by_addr.into_values().collect();
self.rebuild_context();
let mut new_config = (*self.context.config).clone();
new_config.peers = new_by_addr.into_values().collect();
self.replace_context(|ctx| ctx.config = std::sync::Arc::new(new_config));
for peer_config in added_configs {
outcome.added += 1;
@@ -487,7 +491,7 @@ impl Node {
// Start the Noise handshake and get message 1
let our_keypair = self.identity().keypair();
let noise_msg1 =
match connection.start_handshake(our_keypair, self.startup_epoch, current_time_ms) {
match connection.start_handshake(our_keypair, self.startup_epoch(), current_time_ms) {
Ok(msg) => msg,
Err(e) => {
// Clean up the index and link
@@ -698,7 +702,7 @@ impl Node {
debug!(
peer_npub = %traversal.peer_npub,
peers = self.peers.len(),
max_peers = self.max_peers,
max_peers = self.max_peers(),
"Dropping established NAT traversal: at capacity"
);
continue;
@@ -2069,16 +2073,16 @@ impl Node {
.connections
.len()
.saturating_add(self.pending_connects.len());
let connection_slots = if self.max_connections == 0 {
let connection_slots = if self.max_connections() == 0 {
usize::MAX
} else {
self.max_connections.saturating_sub(connection_used)
self.max_connections().saturating_sub(connection_used)
};
let peer_slots = if self.max_peers == 0 {
let peer_slots = if self.max_peers() == 0 {
usize::MAX
} else {
self.max_peers.saturating_sub(self.peers.len())
self.max_peers().saturating_sub(self.peers.len())
};
connection_slots.min(peer_slots)
@@ -2089,25 +2093,25 @@ impl Node {
.connections
.len()
.saturating_add(self.pending_connects.len());
if self.max_connections == 0 {
if self.max_connections() == 0 {
usize::MAX
} else {
self.max_connections.saturating_sub(used)
self.max_connections().saturating_sub(used)
}
}
fn outbound_link_slots(&self) -> usize {
if self.max_links == 0 {
if self.max_links() == 0 {
usize::MAX
} else {
self.max_links.saturating_sub(self.links.len())
self.max_links().saturating_sub(self.links.len())
}
}
fn path_candidate_attempt_budget(&self, peer_node_addr: &NodeAddr) -> usize {
if !self.peers.contains_key(peer_node_addr)
&& self.max_peers > 0
&& self.peers.len() >= self.max_peers
&& self.max_peers() > 0
&& self.peers.len() >= self.max_peers()
{
return 0;
}
src/node/mod.rs
+55 -100
View File
@@ -295,35 +295,18 @@ struct PendingConnect {
/// connection before authentication completes.
// Discovery lookup constants moved to config: node.discovery.attempt_timeouts_secs, node.discovery.ttl
pub struct Node {
// === Identity ===
/// This node's cryptographic identity.
identity: Identity,
/// Random epoch generated at startup for peer restart detection.
/// Exchanged inside Noise handshake messages so peers can detect restarts.
startup_epoch: [u8; 8],
/// Instant when the node was created, for uptime reporting.
started_at: std::time::Instant,
// === Configuration ===
/// Loaded configuration.
config: Config,
/// Shared immutable context bundle. A parallel, authoritative copy of the
/// effectively-immutable fields (config/identity/startup_epoch/started_at/
/// is_leaf_only/max_*), kept in lockstep with the `Node` fields via
/// `rebuild_context`. Readers migrate onto it in later sub-PRs; the
/// duplicated `Node` fields are removed once the last reader has moved.
// === Immutable Context ===
/// Shared immutable context bundle: the single source of truth for the
/// node's effectively-immutable state (config/identity/startup_epoch/
/// started_at/is_leaf_only/max_*). Mutated only by whole-`Arc` replacement
/// via `replace_context` at the constructors, `leaf_only`, and
/// `update_peers`; readers reach it through the accessors.
context: Arc<context::NodeContext>,
// === State ===
/// Node operational state.
state: NodeState,
/// Whether this is a leaf-only node.
is_leaf_only: bool,
// === Spanning Tree ===
/// Local spanning tree state.
tree_state: TreeState,
@@ -390,14 +373,6 @@ pub struct Node {
/// initiation timestamp (Unix ms). Prevents duplicate flood queries.
pending_lookups: HashMap<NodeAddr, handlers::discovery::PendingLookup>,
// === Resource Limits ===
/// Maximum connections (0 = unlimited).
max_connections: usize,
/// Maximum peers (0 = unlimited).
max_peers: usize,
/// Maximum links (0 = unlimited).
max_links: usize,
// === Counters ===
/// Next link ID to allocate.
next_link_id: u64,
@@ -656,13 +631,8 @@ impl Node {
));
Ok(Self {
identity,
startup_epoch,
started_at,
config,
context,
state: NodeState::Created,
is_leaf_only,
tree_state,
bloom_state,
coord_cache,
@@ -679,9 +649,6 @@ impl Node {
identity_cache: HashMap::new(),
pending_tun_packets: HashMap::new(),
pending_lookups: HashMap::new(),
max_connections,
max_peers,
max_links,
next_link_id: 1,
next_transport_id: 1,
stats: stats::NodeStats::new(),
@@ -816,13 +783,8 @@ impl Node {
));
Ok(Self {
identity,
startup_epoch,
started_at,
config,
context,
state: NodeState::Created,
is_leaf_only: false,
tree_state,
bloom_state,
coord_cache,
@@ -839,9 +801,6 @@ impl Node {
identity_cache: HashMap::new(),
pending_tun_packets: HashMap::new(),
pending_lookups: HashMap::new(),
max_connections,
max_peers,
max_links,
next_link_id: 1,
next_transport_id: 1,
stats: stats::NodeStats::new(),
@@ -901,9 +860,8 @@ impl Node {
/// Create a leaf-only node (simplified state).
pub fn leaf_only(config: Config) -> Result<Self, NodeError> {
let mut node = Self::new(config)?;
node.is_leaf_only = true;
node.bloom_state = BloomState::leaf_only(*node.identity.node_addr());
node.rebuild_context();
node.bloom_state = BloomState::leaf_only(*node.node_addr());
node.replace_context(|ctx| ctx.is_leaf_only = true);
Ok(node)
}
@@ -915,7 +873,7 @@ impl Node {
// Collect UDP configs with optional names to avoid borrow conflicts
let udp_instances: Vec<_> = self
.config
.config()
.transports
.udp
.iter()
@@ -933,13 +891,13 @@ impl Node {
#[cfg(unix)]
{
let eth_instances: Vec<_> = self
.config
.config()
.transports
.ethernet
.iter()
.map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
.collect();
let xonly = self.identity.pubkey();
let xonly = self.identity().pubkey();
for (name, eth_config) in eth_instances {
let transport_id = self.allocate_transport_id();
let mut eth =
@@ -951,7 +909,7 @@ impl Node {
// Create TCP transport instances
let tcp_instances: Vec<_> = self
.config
.config()
.transports
.tcp
.iter()
@@ -966,7 +924,7 @@ impl Node {
// Create Tor transport instances
let tor_instances: Vec<_> = self
.config
.config()
.transports
.tor
.iter()
@@ -983,7 +941,7 @@ impl Node {
#[cfg(bluer_available)]
{
let ble_instances: Vec<_> = self
.config
.config()
.transports
.ble
.iter()
@@ -1004,7 +962,7 @@ impl Node {
io,
packet_tx.clone(),
);
ble.set_local_pubkey(self.identity.pubkey().serialize());
ble.set_local_pubkey(self.identity().pubkey().serialize());
transports.push(TransportHandle::Ble(ble));
}
Err(e) => {
@@ -1126,12 +1084,17 @@ impl Node {
/// Get this node's NodeAddr.
pub fn node_addr(&self) -> &NodeAddr {
self.identity.node_addr()
self.context.identity.node_addr()
}
/// Get this node's npub.
pub fn npub(&self) -> String {
self.identity.npub()
self.context.identity.npub()
}
/// Get this node's startup epoch (random per-boot tag for restart detection).
pub fn startup_epoch(&self) -> [u8; 8] {
self.context.startup_epoch
}
/// Reload the host map if the backing `/etc/fips/hosts` file changed.
@@ -1174,22 +1137,16 @@ impl Node {
self.context.config.as_ref()
}
/// Rebuild the shared [`context::NodeContext`] from the current `Node`
/// fields. Called after any mutation of a bundled field (`update_peers`,
/// the `set_max_*` setters) so `self.context` stays equal to the `Node`
/// fields it mirrors. Cheap — the only deep copy is the (rare) `Config`
/// clone.
fn rebuild_context(&mut self) {
self.context = Arc::new(context::NodeContext::new(
Arc::new(self.config.clone()),
self.identity.clone(),
self.startup_epoch,
self.started_at,
self.is_leaf_only,
self.max_connections,
self.max_peers,
self.max_links,
));
/// Mutate the shared immutable context by building a fresh
/// [`context::NodeContext`] and swapping the whole `Arc`. The per-instance
/// context is never interior-mutated; this clone-edit-swap is the sole
/// runtime mutation path for the bundle (the constructors,
/// [`leaf_only`](Self::leaf_only), and [`update_peers`](Self::update_peers)).
/// Cheap — the only deep copy is the (rare) `Config` clone behind its `Arc`.
fn replace_context(&mut self, f: impl FnOnce(&mut context::NodeContext)) {
let mut ctx = (*self.context).clone();
f(&mut ctx);
self.context = Arc::new(ctx);
}
/// Calculate the effective IPv6 MTU that can be sent over FIPS.
@@ -1228,7 +1185,7 @@ impl Node {
return mtu;
}
// Fallback to config: try UDP first, then Ethernet
if let Some((_, cfg)) = self.config.transports.udp.iter().next() {
if let Some((_, cfg)) = self.config().transports.udp.iter().next() {
return cfg.mtu();
}
1280
@@ -1297,7 +1254,7 @@ impl Node {
let parent_id = *self.tree_state.my_declaration().parent_id();
let is_root = self.tree_state.is_root();
let max_fpr = self.config.node.bloom.max_inbound_fpr;
let max_fpr = self.config().node.bloom.max_inbound_fpr;
let mut total: f64 = 1.0; // count self
let mut child_count: u32 = 0;
let mut has_data = false;
@@ -1361,7 +1318,7 @@ impl Node {
None => true,
Some(last) => {
now.duration_since(last)
>= std::time::Duration::from_secs(self.config.node.mmp.log_interval_secs)
>= std::time::Duration::from_secs(self.config().node.mmp.log_interval_secs)
}
};
if should_log {
@@ -1482,16 +1439,19 @@ impl Node {
// === Resource Limits ===
/// Set the maximum number of connections (handshake phase).
pub fn set_max_connections(&mut self, max: usize) {
self.max_connections = max;
self.rebuild_context();
/// Maximum connections (handshake phase); 0 = unlimited.
pub fn max_connections(&self) -> usize {
self.context.max_connections
}
/// Set the maximum number of peers (authenticated).
pub fn set_max_peers(&mut self, max: usize) {
self.max_peers = max;
self.rebuild_context();
/// Maximum authenticated peers; 0 = unlimited.
pub fn max_peers(&self) -> usize {
self.context.max_peers
}
/// Maximum links; 0 = unlimited.
pub fn max_links(&self) -> usize {
self.context.max_links
}
/// Returns false when we are at or above the configured `max_peers`
@@ -1503,13 +1463,8 @@ impl Node {
/// Nostr-mediated NAT-traversal punch) from doing pointless work
/// when saturated.
pub(crate) fn outbound_admission_check(&self) -> bool {
self.max_peers == 0 || self.peers.len() < self.max_peers
}
/// Set the maximum number of links.
pub fn set_max_links(&mut self, max: usize) {
self.max_links = max;
self.rebuild_context();
let max_peers = self.context.max_peers;
max_peers == 0 || self.peers.len() < max_peers
}
// === Counts ===
@@ -1574,9 +1529,9 @@ impl Node {
/// Add a link.
pub fn add_link(&mut self, link: Link) -> Result<(), NodeError> {
if self.max_links > 0 && self.links.len() >= self.max_links {
if self.max_links() > 0 && self.links.len() >= self.max_links() {
return Err(NodeError::MaxLinksExceeded {
max: self.max_links,
max: self.max_links(),
});
}
let link_id = link.link_id();
@@ -1680,9 +1635,9 @@ impl Node {
return Err(NodeError::ConnectionAlreadyExists(link_id));
}
if self.max_connections > 0 && self.connections.len() >= self.max_connections {
if self.max_connections() > 0 && self.connections.len() >= self.max_connections() {
return Err(NodeError::MaxConnectionsExceeded {
max: self.max_connections,
max: self.max_connections(),
});
}
@@ -1817,7 +1772,7 @@ impl Node {
self.identity_cache
.insert(prefix, (node_addr, pubkey, Self::now_ms()));
// LRU eviction
let max = self.config.node.cache.identity_size;
let max = self.config().node.cache.identity_size;
if self.identity_cache.len() > max
&& let Some(oldest_key) = self
.identity_cache
@@ -1868,7 +1823,7 @@ impl Node {
/// Configured maximum identity cache size.
pub fn identity_cache_max(&self) -> usize {
self.config.node.cache.identity_size
self.config().node.cache.identity_size
}
/// Number of pending discovery lookups.
@@ -2287,7 +2242,7 @@ impl fmt::Debug for Node {
f.debug_struct("Node")
.field("node_addr", self.node_addr())
.field("state", &self.state)
.field("is_leaf_only", &self.is_leaf_only)
.field("is_leaf_only", &self.is_leaf_only())
.field("connections", &self.connection_count())
.field("peers", &self.peer_count())
.field("links", &self.link_count())
src/node/retry.rs
+1 -1
View File
@@ -235,7 +235,7 @@ impl Node {
if !self.outbound_admission_check() {
debug!(
peers = self.peers.len(),
max_peers = self.max_peers,
max_peers = self.max_peers(),
retry_pending = self.retry_pending.len(),
"Suppressing auto-reconnect retries: at capacity"
);
src/node/tests/acl.rs
+5 -5
View File
@@ -55,10 +55,10 @@ async fn test_inbound_msg1_denied_by_acl() {
std::fs::write(deny_path(&dir), format!("{}\n", node_a.npub())).unwrap();
node_b.reload_peer_acl().await;
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
let mut conn_a = PeerConnection::outbound(LinkId::new(1), peer_b_identity, 1000);
let noise_msg1 = conn_a
.start_handshake(node_a.identity.keypair(), node_a.startup_epoch, 1000)
.start_handshake(node_a.identity().keypair(), node_a.startup_epoch(), 1000)
.unwrap();
let wire_msg1 = build_msg1(SessionIndex::new(7), &noise_msg1);
let packet = ReceivedPacket::with_timestamp(
@@ -81,13 +81,13 @@ async fn test_outbound_msg2_denied_after_acl_reload() {
let node_b = make_node();
let transport_id = TransportId::new(1);
let remote_addr = TransportAddr::from_string("127.0.0.1:5001");
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
let link_id_a = node_a.allocate_link_id();
let mut conn_a = PeerConnection::outbound(link_id_a, peer_b_identity, 1000);
let our_index_a = node_a.index_allocator.allocate().unwrap();
let noise_msg1 = conn_a
.start_handshake(node_a.identity.keypair(), node_a.startup_epoch, 1000)
.start_handshake(node_a.identity().keypair(), node_a.startup_epoch(), 1000)
.unwrap();
conn_a.set_our_index(our_index_a);
conn_a.set_transport_id(transport_id);
@@ -113,7 +113,7 @@ async fn test_outbound_msg2_denied_after_acl_reload() {
let responder_epoch = [0x11; 8];
let noise_msg2 = conn_b
.receive_handshake_init(
node_b.identity.keypair(),
node_b.identity().keypair(),
responder_epoch,
&noise_msg1,
1000,
src/node/tests/bootstrap.rs
+5 -5
View File
@@ -62,9 +62,9 @@ async fn test_adopted_udp_traversal_completes_handshake() {
}
let peer_a_node_addr =
*PeerIdentity::from_pubkey_full(node_a.identity.pubkey_full()).node_addr();
*PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full()).node_addr();
let peer_b_node_addr =
*PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full()).node_addr();
*PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full()).node_addr();
assert_eq!(
node_a.peer_count(),
@@ -223,7 +223,7 @@ async fn test_third_peer_can_handshake_via_adopted_transport_socket() {
assert_eq!(pkt_at_b.data[0] & 0x0f, PHASE_MSG2);
node_b.handle_msg2(pkt_at_b).await;
let node_a_addr = *PeerIdentity::from_pubkey_full(node_a.identity.pubkey_full()).node_addr();
let node_a_addr = *PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full()).node_addr();
assert!(
node_b.get_peer(&node_a_addr).is_some(),
"node_b should first be connected to node_a via adopted transport"
@@ -237,7 +237,7 @@ async fn test_third_peer_can_handshake_via_adopted_transport_socket() {
.transports
.insert(transport_id_c, TransportHandle::Udp(transport_c));
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
let adopted_addr = TransportAddr::from_string(&handoff_result.local_addr.to_string());
node_c
.initiate_connection(transport_id_c, adopted_addr, peer_b_identity)
@@ -273,7 +273,7 @@ async fn test_third_peer_can_handshake_via_adopted_transport_socket() {
};
node_c.handle_msg2(pkt_at_c).await;
let node_c_addr = *PeerIdentity::from_pubkey_full(node_c.identity.pubkey_full()).node_addr();
let node_c_addr = *PeerIdentity::from_pubkey_full(node_c.identity().pubkey_full()).node_addr();
assert!(
node_b.get_peer(&node_c_addr).is_some(),
"node_b should promote node_c when node_c handshakes via adopted socket"
src/node/tests/discovery.rs
+1 -1
View File
@@ -1048,7 +1048,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() {
// Default attempt_timeouts_secs is [1, 2, 4, 8]. Confirm so the test
// cannot silently drift if the default changes.
assert_eq!(
node.config.node.discovery.attempt_timeouts_secs,
node.config().node.discovery.attempt_timeouts_secs,
vec![1, 2, 4, 8],
"test pins the [1,2,4,8] default; update the test if the default changes"
);
src/node/tests/handshake.rs
+23 -23
View File
@@ -49,7 +49,7 @@ async fn test_two_node_handshake_udp() {
// === Phase 1: Node A initiates handshake to Node B ===
// Create peer identity for B (must use full key for ECDH parity)
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
let peer_b_node_addr = *peer_b_identity.node_addr();
let link_id_a = node_a.allocate_link_id();
@@ -59,9 +59,9 @@ async fn test_two_node_handshake_udp() {
let our_index_a = node_a.index_allocator.allocate().unwrap();
// Start handshake (generates Noise IK msg1)
let our_keypair_a = node_a.identity.keypair();
let our_keypair_a = node_a.identity().keypair();
let noise_msg1 = conn_a
.start_handshake(our_keypair_a, node_a.startup_epoch, 1000)
.start_handshake(our_keypair_a, node_a.startup_epoch(), 1000)
.unwrap();
conn_a.set_our_index(our_index_a);
conn_a.set_transport_id(transport_id_a);
@@ -101,7 +101,7 @@ async fn test_two_node_handshake_udp() {
// Verify B promoted the inbound connection
let peer_a_node_addr =
*PeerIdentity::from_pubkey_full(node_a.identity.pubkey_full()).node_addr();
*PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full()).node_addr();
assert_eq!(
node_b.peer_count(),
1,
@@ -290,16 +290,16 @@ async fn test_run_rx_loop_handshake() {
// === Phase 1: Node A initiates handshake to Node B ===
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
let peer_b_node_addr = *peer_b_identity.node_addr();
let link_id_a = node_a.allocate_link_id();
let mut conn_a = PeerConnection::outbound(link_id_a, peer_b_identity, 1000);
let our_index_a = node_a.index_allocator.allocate().unwrap();
let our_keypair_a = node_a.identity.keypair();
let our_keypair_a = node_a.identity().keypair();
let noise_msg1 = conn_a
.start_handshake(our_keypair_a, node_a.startup_epoch, 1000)
.start_handshake(our_keypair_a, node_a.startup_epoch(), 1000)
.unwrap();
conn_a.set_our_index(our_index_a);
conn_a.set_transport_id(transport_id_a);
@@ -347,7 +347,7 @@ async fn test_run_rx_loop_handshake() {
// Verify Node B promoted the inbound connection via rx loop dispatch
let peer_a_node_addr =
*PeerIdentity::from_pubkey_full(node_a.identity.pubkey_full()).node_addr();
*PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full()).node_addr();
assert_eq!(
node_b.peer_count(),
@@ -474,9 +474,9 @@ async fn test_cross_connection_both_initiate() {
.insert(transport_id_b, TransportHandle::Udp(transport_b));
// Peer identities (must use full key for ECDH parity)
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
let peer_b_node_addr = *peer_b_identity.node_addr();
let peer_a_identity = PeerIdentity::from_pubkey_full(node_a.identity.pubkey_full());
let peer_a_identity = PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full());
let peer_a_node_addr = *peer_a_identity.node_addr();
// === Phase 1: Both nodes initiate handshakes (simulate auto_connect) ===
@@ -485,9 +485,9 @@ async fn test_cross_connection_both_initiate() {
let link_id_a_out = node_a.allocate_link_id();
let mut conn_a = PeerConnection::outbound(link_id_a_out, peer_b_identity, 1000);
let our_index_a = node_a.index_allocator.allocate().unwrap();
let our_keypair_a = node_a.identity.keypair();
let our_keypair_a = node_a.identity().keypair();
let noise_msg1_a = conn_a
.start_handshake(our_keypair_a, node_a.startup_epoch, 1000)
.start_handshake(our_keypair_a, node_a.startup_epoch(), 1000)
.unwrap();
conn_a.set_our_index(our_index_a);
conn_a.set_transport_id(transport_id_a);
@@ -515,9 +515,9 @@ async fn test_cross_connection_both_initiate() {
let link_id_b_out = node_b.allocate_link_id();
let mut conn_b = PeerConnection::outbound(link_id_b_out, peer_a_identity, 1000);
let our_index_b = node_b.index_allocator.allocate().unwrap();
let our_keypair_b = node_b.identity.keypair();
let our_keypair_b = node_b.identity().keypair();
let noise_msg1_b = conn_b
.start_handshake(our_keypair_b, node_b.startup_epoch, 1000)
.start_handshake(our_keypair_b, node_b.startup_epoch(), 1000)
.unwrap();
conn_b.set_our_index(our_index_b);
conn_b.set_transport_id(transport_id_b);
@@ -666,9 +666,9 @@ async fn test_stale_connection_cleanup() {
// Allocate session index and set transport info
let our_index = node.index_allocator.allocate().unwrap();
let our_keypair = node.identity.keypair();
let our_keypair = node.identity().keypair();
let _noise_msg1 = conn
.start_handshake(our_keypair, node.startup_epoch, past_time_ms)
.start_handshake(our_keypair, node.startup_epoch(), past_time_ms)
.unwrap();
conn.set_our_index(our_index);
conn.set_transport_id(transport_id);
@@ -744,9 +744,9 @@ async fn test_failed_connection_cleanup() {
let mut conn = PeerConnection::outbound(link_id, peer_identity, now_ms);
let our_index = node.index_allocator.allocate().unwrap();
let our_keypair = node.identity.keypair();
let our_keypair = node.identity().keypair();
let _noise_msg1 = conn
.start_handshake(our_keypair, node.startup_epoch, now_ms)
.start_handshake(our_keypair, node.startup_epoch(), now_ms)
.unwrap();
conn.set_our_index(our_index);
conn.set_transport_id(transport_id);
@@ -804,9 +804,9 @@ async fn test_msg1_stored_for_resend() {
let mut conn = PeerConnection::outbound(link_id, peer_identity, now_ms);
let our_index = node.index_allocator.allocate().unwrap();
let our_keypair = node.identity.keypair();
let our_keypair = node.identity().keypair();
let noise_msg1 = conn
.start_handshake(our_keypair, node.startup_epoch, now_ms)
.start_handshake(our_keypair, node.startup_epoch(), now_ms)
.unwrap();
conn.set_our_index(our_index);
conn.set_transport_id(transport_id);
@@ -814,7 +814,7 @@ async fn test_msg1_stored_for_resend() {
// Build wire msg1 and store it (as initiate_peer_connection does)
let wire_msg1 = build_msg1(our_index, &noise_msg1);
let resend_interval = node.config.node.rate_limit.handshake_resend_interval_ms;
let resend_interval = node.config().node.rate_limit.handshake_resend_interval_ms;
conn.set_handshake_msg1(wire_msg1.clone(), now_ms + resend_interval);
// Verify stored msg1 matches what was built
@@ -837,9 +837,9 @@ async fn test_resend_scheduling() {
let mut conn = PeerConnection::outbound(link_id, peer_identity, now_ms);
let our_index = node.index_allocator.allocate().unwrap();
let our_keypair = node.identity.keypair();
let our_keypair = node.identity().keypair();
let noise_msg1 = conn
.start_handshake(our_keypair, node.startup_epoch, now_ms)
.start_handshake(our_keypair, node.startup_epoch(), now_ms)
.unwrap();
conn.set_our_index(our_index);
conn.set_transport_id(transport_id);
src/node/tests/mod.rs
+21 -6
View File
@@ -27,14 +27,29 @@ pub(super) fn make_node() -> Node {
make_node_with(Config::new())
}
/// Build a test node from an explicit `Config`. Prefer this over poking
/// `node.config.*` after construction: immutable fields are mirrored into the
/// shared `NodeContext` at build time, so a post-construction field poke is
/// invisible to any reader that has migrated onto the `config()` accessor.
/// Build a test node from an explicit `Config`. Immutable state lives solely in
/// the shared `NodeContext`, built once at construction — there is no
/// post-construction field to poke, so set limits/config on the `Config` here.
pub(super) fn make_node_with(config: Config) -> Node {
Node::new(config).unwrap()
}
/// Build a test node with an explicit `max_peers` limit (replaces the removed
/// `set_max_peers` setter; resource limits are immutable post-construction).
pub(super) fn make_node_with_max_peers(max_peers: usize) -> Node {
let mut config = Config::new();
config.node.limits.max_peers = max_peers;
make_node_with(config)
}
/// Build a test node with an explicit `max_links` limit (replaces the removed
/// `set_max_links` setter; resource limits are immutable post-construction).
pub(super) fn make_node_with_max_links(max_links: usize) -> Node {
let mut config = Config::new();
config.node.limits.max_links = max_links;
make_node_with(config)
}
#[allow(dead_code)]
pub(super) fn make_node_addr(val: u8) -> NodeAddr {
let mut bytes = [0u8; 16];
@@ -65,9 +80,9 @@ pub(super) fn make_completed_connection(
let mut conn = PeerConnection::outbound(link_id, peer_identity, current_time_ms);
// Run initiator side of handshake
let our_keypair = node.identity.keypair();
let our_keypair = node.identity().keypair();
let msg1 = conn
.start_handshake(our_keypair, node.startup_epoch, current_time_ms)
.start_handshake(our_keypair, node.startup_epoch(), current_time_ms)
.unwrap();
// Run responder side to generate msg2
src/node/tests/session.rs
+7 -6
View File
@@ -1630,8 +1630,9 @@ fn test_coords_warmup_config_default() {
#[test]
fn test_identity_cache_lru_eviction() {
let mut node = make_node();
node.config.node.cache.identity_size = 2;
let mut config = crate::Config::new();
config.node.cache.identity_size = 2;
let mut node = make_node_with(config);
let id1 = Identity::generate();
let id2 = Identity::generate();
@@ -1794,7 +1795,7 @@ async fn test_session_handshake_timeout() {
let identity_b = Identity::generate();
let handshake =
HandshakeState::new_initiator(node.identity.keypair(), identity_b.pubkey_full());
HandshakeState::new_initiator(node.identity().keypair(), identity_b.pubkey_full());
let dest_addr = *identity_b.node_addr();
@@ -1811,7 +1812,7 @@ async fn test_session_handshake_timeout() {
assert!(node.sessions.contains_key(&dest_addr));
// Before timeout: session should remain
let timeout_secs = node.config.node.rate_limit.handshake_timeout_secs;
let timeout_secs = node.config().node.rate_limit.handshake_timeout_secs;
let before_timeout = 1000 + timeout_secs * 1000 - 1;
node.resend_pending_session_handshakes(before_timeout).await;
assert!(
@@ -1855,7 +1856,7 @@ async fn test_session_awaiting_msg3_timeout() {
assert!(node.sessions.contains_key(&src_addr));
// After timeout: session should be removed
let timeout_secs = node.config.node.rate_limit.handshake_timeout_secs;
let timeout_secs = node.config().node.rate_limit.handshake_timeout_secs;
let after_timeout = 1000 + timeout_secs * 1000 + 1;
node.resend_pending_session_handshakes(after_timeout).await;
assert!(
@@ -2301,7 +2302,7 @@ fn install_established_session_with_mmp(node: &mut Node, remote: &Identity) {
1000,
true,
);
entry.init_mmp(&node.config.node.session_mmp);
entry.init_mmp(&node.config().node.session_mmp);
node.sessions.insert(remote_addr, entry);
}
src/node/tests/spanning_tree.rs
+1 -1
View File
@@ -90,7 +90,7 @@ pub(super) async fn initiate_handshake(nodes: &mut [TestNode], i: usize, j: usiz
let our_index = initiator.node.index_allocator.allocate().unwrap();
let our_keypair = initiator.node.identity().keypair();
let noise_msg1 = conn
.start_handshake(our_keypair, initiator.node.startup_epoch, 1000)
.start_handshake(our_keypair, initiator.node.startup_epoch(), 1000)
.unwrap();
conn.set_our_index(our_index);
conn.set_transport_id(transport_id);
src/node/tests/unit.rs
+32 -37
View File
@@ -250,8 +250,7 @@ fn test_node_link_management() {
#[test]
fn test_node_link_limit() {
let mut node = make_node();
node.set_max_links(2);
let mut node = make_node_with_max_links(2);
for i in 0..2 {
let link_id = node.allocate_link_id();
@@ -383,9 +382,8 @@ fn test_node_cross_connection_resolution() {
#[test]
fn test_node_peer_limit() {
let mut node = make_node();
let mut node = make_node_with_max_peers(2);
let transport_id = TransportId::new(1);
node.set_max_peers(2);
// Add two peers via promotion
for i in 0..2 {
@@ -601,9 +599,9 @@ fn test_promote_cleans_up_pending_outbound_to_same_peer() {
let mut pending_conn =
PeerConnection::outbound(pending_link_id, peer_b_identity, pending_time_ms);
let our_keypair = node.identity.keypair();
let our_keypair = node.identity().keypair();
let _msg1 = pending_conn
.start_handshake(our_keypair, node.startup_epoch, pending_time_ms)
.start_handshake(our_keypair, node.startup_epoch(), pending_time_ms)
.unwrap();
let pending_index = node.index_allocator.allocate().unwrap();
@@ -640,9 +638,9 @@ fn test_promote_cleans_up_pending_outbound_to_same_peer() {
let mut completing_conn =
PeerConnection::outbound(completing_link_id, peer_b_identity, completing_time_ms);
let our_keypair = node.identity.keypair();
let our_keypair = node.identity().keypair();
let msg1 = completing_conn
.start_handshake(our_keypair, node.startup_epoch, completing_time_ms)
.start_handshake(our_keypair, node.startup_epoch(), completing_time_ms)
.unwrap();
// B responds
@@ -986,7 +984,7 @@ fn active_peer_same_path_discovery_refreshes_stale_peer() {
let transport_id = TransportId::new(1);
let current_addr = TransportAddr::from_string("127.0.0.1:9");
let stale_at = Node::now_ms().saturating_sub(
node.config
node.config()
.node
.heartbeat_interval_secs
.saturating_add(1)
@@ -1037,7 +1035,20 @@ async fn node_context_mirrors_config_and_immutable_facades() {
#[tokio::test]
async fn update_peers_races_new_alternative_without_dropping_active_peer() {
let mut node = make_node();
// The node's *current* (pre-update) peer set must contain `old_peer`, so it
// is baked into the Config at construction (immutable context = sole store).
let peer_full = Identity::generate();
let old_peer = crate::config::PeerConfig {
npub: peer_full.npub(),
alias: None,
addresses: vec![crate::config::PeerAddress::new("udp", "127.0.0.1:9")],
connect_policy: crate::config::ConnectPolicy::AutoConnect,
auto_reconnect: true,
via_nostr: false,
};
let mut config = Config::new();
config.peers = vec![old_peer.clone()];
let mut node = make_node_with(config);
let (packet_tx, packet_rx) = packet_channel(64);
node.packet_tx = Some(packet_tx.clone());
node.packet_rx = Some(packet_rx);
@@ -1056,7 +1067,6 @@ async fn update_peers_races_new_alternative_without_dropping_active_peer() {
node.transports
.insert(transport_id, TransportHandle::Udp(udp));
let peer_full = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full());
let peer_node_addr = *peer_identity.node_addr();
let current_addr = TransportAddr::from_string("127.0.0.1:9");
@@ -1076,14 +1086,6 @@ async fn update_peers_races_new_alternative_without_dropping_active_peer() {
),
);
let old_peer = crate::config::PeerConfig {
npub: peer_full.npub(),
alias: None,
addresses: vec![crate::config::PeerAddress::new("udp", "127.0.0.1:9")],
connect_policy: crate::config::ConnectPolicy::AutoConnect,
auto_reconnect: true,
via_nostr: false,
};
let new_peer = crate::config::PeerConfig {
addresses: vec![
crate::config::PeerAddress::new("udp", "127.0.0.1:9"),
@@ -1091,7 +1093,6 @@ async fn update_peers_races_new_alternative_without_dropping_active_peer() {
],
..old_peer.clone()
};
node.config.peers = vec![old_peer];
let outcome = node.update_peers(vec![new_peer]).await.unwrap();
@@ -1257,8 +1258,8 @@ fn test_schedule_reconnect_preserves_backoff() {
);
// With count=3, backoff should be 5s * 2^3 = 40s.
let base_ms = node.config.node.retry.base_interval_secs * 1000;
let max_ms = node.config.node.retry.max_backoff_secs * 1000;
let base_ms = node.config().node.retry.base_interval_secs * 1000;
let max_ms = node.config().node.retry.max_backoff_secs * 1000;
let expected_delay = state.backoff_ms(base_ms, max_ms);
assert_eq!(
state.retry_after_ms,
@@ -1293,8 +1294,8 @@ fn test_schedule_reconnect_fresh_state() {
"Fresh reconnect should start at count=0"
);
// Base delay: 5s * 2^0 = 5s
let base_ms = node.config.node.retry.base_interval_secs * 1000;
let max_ms = node.config.node.retry.max_backoff_secs * 1000;
let base_ms = node.config().node.retry.base_interval_secs * 1000;
let max_ms = node.config().node.retry.max_backoff_secs * 1000;
let expected_delay = state.backoff_ms(base_ms, max_ms);
assert_eq!(state.retry_after_ms, 1_000 + expected_delay);
}
@@ -1625,8 +1626,7 @@ fn inject_dummy_peers(node: &mut Node, count: usize) {
#[test]
fn outbound_admission_check_direct() {
// max_peers cap honored: above-cap returns false, below-cap returns true.
let mut node = make_node();
node.set_max_peers(3);
let mut node = make_node_with_max_peers(3);
assert!(node.outbound_admission_check(), "0/3 should be admissible");
inject_dummy_peers(&mut node, 2);
@@ -1643,8 +1643,7 @@ fn outbound_admission_check_direct() {
);
// No-cap sentinel: max_peers == 0 admits unconditionally.
let mut uncapped = make_node();
uncapped.set_max_peers(0);
let mut uncapped = make_node_with_max_peers(0);
assert!(uncapped.outbound_admission_check());
inject_dummy_peers(&mut uncapped, 50);
assert!(
@@ -1655,8 +1654,7 @@ fn outbound_admission_check_direct() {
#[tokio::test]
async fn process_pending_retries_gated_at_capacity() {
let mut node = make_node();
node.set_max_peers(2);
let mut node = make_node_with_max_peers(2);
inject_dummy_peers(&mut node, 2);
// Queue a retry that would otherwise be due.
@@ -1714,8 +1712,7 @@ async fn poll_nostr_discovery_established_gated_at_capacity() {
use crate::discovery::EstablishedTraversal;
use std::net::UdpSocket;
let mut node = make_node();
node.set_max_peers(2);
let mut node = make_node_with_max_peers(2);
inject_dummy_peers(&mut node, 2);
let bootstrap = Arc::new(NostrDiscovery::new_for_test());
@@ -1794,7 +1791,7 @@ async fn craft_and_send_msg1(
use crate::node::wire::build_msg1;
use crate::utils::index::SessionIndex;
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
let sender_pubkey_id = PeerIdentity::from_pubkey_full(sender_identity.pubkey_full());
let sender_node_addr = *sender_pubkey_id.node_addr();
@@ -1852,8 +1849,7 @@ async fn handle_msg1_silent_drops_at_cap_for_new_peer() {
use crate::config::UdpConfig;
use tokio::time::{Duration, timeout};
let mut node = make_node();
node.set_max_peers(2);
let mut node = make_node_with_max_peers(2);
inject_dummy_peers(&mut node, 2);
assert_eq!(node.peer_count(), 2, "precondition: at cap");
@@ -1942,8 +1938,7 @@ async fn handle_msg1_silent_drops_at_cap_for_new_peer() {
async fn handle_msg1_admits_existing_peer_at_cap() {
use crate::config::UdpConfig;
let mut node = make_node();
node.set_max_peers(2);
let mut node = make_node_with_max_peers(2);
inject_dummy_peers(&mut node, 1);
testing/ci-local.sh
+15
View File
@@ -233,6 +233,21 @@ run_build() {
record "clippy" 1
return 1
fi
# Guard: the effectively-immutable state lives solely in NodeContext. The
# Node struct must not re-declare a bundled field (config/identity/
# startup_epoch/started_at/is_leaf_only/node_profile/max_*) — a shadow field
# would silently reopen dual-store divergence between the struct and the
# context. Checks the struct *declaration*, so it is wrap-insensitive.
info "node-context single-store guard"
if awk '/^pub struct Node \{/,/^\}/' src/node/mod.rs \
| grep -qE '^[[:space:]]+(config|identity|startup_epoch|started_at|is_leaf_only|node_profile|max_connections|max_peers|max_links):'; then
fail "Node struct re-declares a bundled immutable field; it must live solely in NodeContext"
record "node-context-guard" 1
return 1
else
record "node-context-guard" 0
fi
}
# ── Stage 2: Unit Tests ───────────────────────────────────────────────────