From f10467fea7359dfa0b537ee2cbc62accc05dc140 Mon Sep 17 00:00:00 2001 From: Sandi Fatic Date: Mon, 6 Jul 2026 14:04:03 +0200 Subject: [PATCH 1/4] fix(storage): compare final path segment in is_ancestor_of; unify local Shared stamp authority is_ancestor_of iterated only the offsets between segments, so it never compared the ancestor candidate's deepest segment: `::a::x` reported as an ancestor of `::a::y::z`, and a single-segment path matched any deeper path regardless of content. is_descendant_of inherited both. Compare all of self's segments (depth()+1) as a true prefix instead, and add regression tests for the differing-final-segment and differing-single-segment cases. The local-write authorization for Shared entities lived in two hand-rolled copies (save_raw used stored-union-claimed; the delete path used stored only), which could silently drift on the auth boundary. Extract one authorize_local_shared_stamp helper and route both through it. Behavior is preserved: in the delete path metadata is loaded fresh from the index, so its writers already are the stored set and the union collapses to the same stored-membership check. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/storage/src/address.rs | 17 +++--- crates/storage/src/interface.rs | 85 +++++++++++++++++------------ crates/storage/src/tests/address.rs | 16 ++++++ 3 files changed, 73 insertions(+), 45 deletions(-) diff --git a/crates/storage/src/address.rs b/crates/storage/src/address.rs index 6235eebe7b..f64fb3c829 100644 --- a/crates/storage/src/address.rs +++ b/crates/storage/src/address.rs @@ -163,16 +163,13 @@ impl Path { if self.depth() >= other.depth() { return false; } - let mut last_offset = 0_usize; - - for &offset in &self.offsets { - if self.path[last_offset..offset as usize] != other.path[last_offset..offset as usize] { - return false; - } - last_offset = offset as usize; - } - - true + // Every one of self's segments must equal other's segment at the same + // position — all `depth() + 1` of them, including the deepest. `zip` + // stops at self's (shorter) length, so this checks self as a full + // segment-wise prefix of other. Iterating only the offsets between + // segments, as an earlier version did, silently skipped self's final + // segment: `::a::x` then wrongly read as an ancestor of `::a::y::z`. + self.segments().zip(other.segments()).all(|(a, b)| a == b) } /// Checks if this path is a descendant of another. diff --git a/crates/storage/src/interface.rs b/crates/storage/src/interface.rs index d7a879f397..93121e381b 100644 --- a/crates/storage/src/interface.rs +++ b/crates/storage/src/interface.rs @@ -211,6 +211,11 @@ enum RemoveMode { Relocate, } +/// A resolved local `Shared` stamp authorization: the writer set to persist +/// paired with the signer to record. Produced by +/// [`Interface::authorize_local_shared_stamp`]. +type SharedStampAuthorization = (BTreeMap, PublicKey); + /// The primary interface for the storage system. #[derive(Debug, Default, Clone)] #[non_exhaustive] @@ -2759,22 +2764,16 @@ impl Interface { } } - // If this is a local shared action by a writer, set the nonce. - // Note: unlike save_raw, here `metadata` was just loaded from the index - // a few lines above and represents the current stored state. There's - // no separate "claimed" set to union against — the executor must be - // in the stored writer set to authorize the delete. + // If this is a local shared action by a writer, set the nonce. Same + // authority rule as save_raw, via the shared helper. Here `metadata` + // was just loaded from the index above, so its writers already are the + // stored set — the helper's stored ∪ claimed union collapses to the + // stored membership check this delete requires. let shared_to_stamp = if let StorageType::Shared { - writers: stored, .. + writers: claimed, .. } = &metadata.storage_type { - let executor: calimero_primitives::identity::PublicKey = - crate::env::executor_id().into(); - if stored.contains_key(&executor) { - Some((stored.clone(), executor)) - } else { - None - } + Self::authorize_local_shared_stamp(child_id, claimed)? } else { None }; @@ -3527,6 +3526,41 @@ impl Interface { Ok(lww_pick(existing, incoming)) } + /// Decides whether the local executor may stamp a `Shared` mutation of + /// `id`, returning the writer set to persist and the signer to record, or + /// `None` if the executor is not authorized. + /// + /// Authority is the union of two writer sets: + /// - `claimed`: the writers carried in the metadata being written. On a + /// save this is the incoming action's own claimed set; on a delete it + /// is the set just loaded from the stored index (so `claimed` already + /// equals stored there, and the union below is a no-op). + /// - stored: the writers currently persisted in the index for `id`. + /// + /// Membership in EITHER set authorizes the stamp. The union is what lets a + /// writer rotate itself out: it is still in the stored set though absent + /// from the new claimed set, and the remote verifier also checks against + /// stored, so the signature still verifies there. + /// + /// Both the save and delete paths route through here so the local-write + /// authority rule lives in exactly one place and cannot drift between them; + /// each caller keeps its own nonce and any schema re-stamp. + fn authorize_local_shared_stamp( + id: Id, + claimed: &BTreeMap, + ) -> Result, StorageError> { + let executor: PublicKey = crate::env::executor_id().into(); + let stored_has_executor = >::get_metadata(id)? + .as_ref() + .map(|m| match &m.storage_type { + StorageType::Shared { writers, .. } => writers.contains_key(&executor), + _ => false, + }) + .unwrap_or(false); + let authorized = stored_has_executor || claimed.contains_key(&executor); + Ok(authorized.then(|| (claimed.clone(), executor))) + } + /// Saves raw serialized data with orphan checking. /// /// # Errors @@ -3603,13 +3637,8 @@ impl Interface { } // If this is a local shared action by a writer, set the nonce. - // - // Stamping authority is the union of (stored writers) and (action's claimed writers): - // - Stored: the writer set as currently persisted in the index. - // - Claimed: the writer set in the action's own metadata. - // Stamp if the executor is in EITHER. This is what enables rotate-self-out: - // a writer rotating themselves out has executor ∈ stored but ∉ claimed; the - // verifier on remote also uses stored, so the signature still verifies there. + // Authority (stored ∪ claimed) is decided by the shared helper; here + // `claimed` is the incoming action's own writer set. // // Same re-stamp-always rationale as the User arm above: a // re-write may carry the previously-stored real signature @@ -3620,21 +3649,7 @@ impl Interface { .. } = &metadata.storage_type { - let executor: calimero_primitives::identity::PublicKey = - crate::env::executor_id().into(); - let stored_has_executor = >::get_metadata(id)? - .as_ref() - .map(|m| match &m.storage_type { - StorageType::Shared { writers, .. } => writers.contains_key(&executor), - _ => false, - }) - .unwrap_or(false); - let claimed_has_executor = claimed_writers.contains_key(&executor); - if stored_has_executor || claimed_has_executor { - Some((claimed_writers.clone(), executor)) - } else { - None - } + Self::authorize_local_shared_stamp(id, claimed_writers)? } else { None }; diff --git a/crates/storage/src/tests/address.rs b/crates/storage/src/tests/address.rs index 6982464970..ab6a236abb 100644 --- a/crates/storage/src/tests/address.rs +++ b/crates/storage/src/tests/address.rs @@ -134,6 +134,17 @@ mod path__public_methods { assert!(!Path::new("::root::node") .unwrap() .is_ancestor_of(&Path::new("::root::another").unwrap())); + + // A strictly-shallower path whose *final* segment differs is not an + // ancestor — the deepest segment must be compared, not skipped. + assert!(!Path::new("::root::wrong") + .unwrap() + .is_ancestor_of(&Path::new("::root::node::leaf").unwrap())); + // A shallower single-segment path whose only segment differs is not an + // ancestor either. + assert!(!Path::new("::other") + .unwrap() + .is_ancestor_of(&Path::new("::root::node").unwrap())); } #[test] @@ -154,6 +165,11 @@ mod path__public_methods { assert!(!Path::new("::root::node") .unwrap() .is_descendant_of(&Path::new("::root::another").unwrap())); + + // A deeper path is not a descendant of one whose final segment differs. + assert!(!Path::new("::root::node::leaf") + .unwrap() + .is_descendant_of(&Path::new("::root::wrong").unwrap())); } #[test] From df1a289c3d17c3929a542e2bfc58c124472a818b Mon Sep 17 00:00:00 2001 From: Sandi Fatic Date: Mon, 6 Jul 2026 15:13:36 +0200 Subject: [PATCH 2/4] fix(node): reservoir bias, blob-id check, backfill cap, mailbox bound, sync debounce, map TTLs, log throttle, config knobs Address a batch of node-crate review findings: - utils: choose_stream consumed the first item before enumerating, biasing Algorithm R toward later elements (and a single-item stream matched anything). Enumerate the whole stream so replacement prob is 1/(idx+1). - blobs: verify a downloaded blob's content hash equals the requested id before reporting success; discard mismatched/tampered content. - namespace backfill: enforce MAX_BACKFILL_OPS on the receiver, not just the responder, so a misbehaving peer can't drive unbounded apply work. Shared const in sync::mod. - subscriptions: stop swallowing has_context store errors via unwrap_or_default(); log and bail instead of treating them as 'no context'. - namespace parent-pull: add bounded exponential backoff between peer attempts. - network_event_channel: lock-free time throttle on the capacity/backpressure warnings so sustained overload can't flood the log. - network_event_processor: bound NodeManager's mailbox and forward via try_send with bounded retry + a drop counter instead of do_send (which ignores capacity and grew the mailbox without limit). - heartbeat: per-context debounce on the 'peer is ahead' recovery sync so N ahead peers no longer spawn N syncs per 30s cycle. - heartbeat/readiness: TTL-evict the peer-keyed divergence_streak, behind_sync, and probe-response maps so they can't grow on peer churn. - run/constants: name the channel capacities in constants.rs instead of scattering magic numbers. - namespace: decompose the ~200-line governance-delta spawned future into a NamespaceDeltaApply context with run()/on_applied() methods. cargo test -p calimero-node: 431 passed. clippy + fmt clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/node/src/constants.rs | 25 + .../node/src/handlers/network_event/blobs.rs | 15 +- .../src/handlers/network_event/heartbeat.rs | 28 + .../src/handlers/network_event/namespace.rs | 478 ++++++++++-------- .../handlers/network_event/subscriptions.rs | 33 +- crates/node/src/manager.rs | 24 +- crates/node/src/manager/startup.rs | 10 + crates/node/src/network_event_channel.rs | 71 ++- crates/node/src/network_event_processor.rs | 107 +++- crates/node/src/readiness.rs | 9 + crates/node/src/run.rs | 23 +- .../node/src/sync/manager/namespace_sync.rs | 5 +- crates/node/src/sync/mod.rs | 7 + crates/node/src/utils.rs | 11 +- 14 files changed, 561 insertions(+), 285 deletions(-) diff --git a/crates/node/src/constants.rs b/crates/node/src/constants.rs index 2c2a4ef55d..5a027e3a39 100644 --- a/crates/node/src/constants.rs +++ b/crates/node/src/constants.rs @@ -27,6 +27,31 @@ pub const OLD_BLOBS_EVICTION_FREQUENCY_S: u64 = 300; /// Prevents log spam when buffer is under sustained pressure. pub const DELTA_BUFFER_DROP_WARNING_RATE_LIMIT_S: u64 = 5; +/// Capacity of the dedicated network-event channel between the network +/// manager and the NodeManager bridge. Sized to absorb burst traffic; the +/// bridge applies backpressure (and, past `NETWORK_EVENT_MAX_PENDING_RETRIES`, +/// drops with a metric) once full. +pub const NETWORK_EVENT_CHANNEL_SIZE: usize = 1000; +/// Upper bound on overflow events buffered in the network-event channel's +/// async retry path before a true drop — roughly one channel's worth. +pub const NETWORK_EVENT_MAX_PENDING_RETRIES: usize = 1000; + +/// Broadcast capacity for server-facing node events (supports many concurrent +/// WebSocket clients). +pub const EVENT_BROADCAST_CHANNEL_SIZE: usize = 256; +/// Buffer for queued context sync requests (absorbs bursts of context +/// joins/syncs). +pub const CTX_SYNC_CHANNEL_SIZE: usize = 64; +/// Buffer for queued namespace governance sync requests. +pub const NS_SYNC_CHANNEL_SIZE: usize = 16; +/// Buffer for queued namespace join requests. +pub const NS_JOIN_CHANNEL_SIZE: usize = 16; +/// Buffer for queued open-subgroup join requests. +pub const OPEN_SUBGROUP_JOIN_CHANNEL_SIZE: usize = 16; +/// Buffer for the execute path's locally-applied-delta notifications to the +/// node's in-memory DeltaStore. +pub const LOCAL_DELTA_CHANNEL_SIZE: usize = 256; + /// How often the gossipsub mesh-peer-count snapshot is logged per node /// (in seconds). The snapshot is CI-observable evidence of actual mesh /// state — the libp2p-gossipsub `Updating mesh, new mesh: {}` log reports diff --git a/crates/node/src/handlers/network_event/blobs.rs b/crates/node/src/handlers/network_event/blobs.rs index 8d4c65ae92..43e0e05e20 100644 --- a/crates/node/src/handlers/network_event/blobs.rs +++ b/crates/node/src/handlers/network_event/blobs.rs @@ -55,7 +55,7 @@ pub(super) fn handle_blob_downloaded( let reader = &blob_data[..]; match blobstore.put(reader).await { - Ok((stored_blob_id, _hash, size)) => { + Ok((stored_blob_id, _hash, size)) if stored_blob_id == blob_id => { info!( requested_blob_id = %blob_id, stored_blob_id = %stored_blob_id, @@ -63,6 +63,19 @@ pub(super) fn handle_blob_downloaded( "Blob stored successfully" ); } + Ok((stored_blob_id, _hash, size)) => { + // Blob ids are content hashes: the id the peer served under + // must equal the hash of the bytes it delivered. A mismatch + // means we were handed the wrong (or tampered) content, so + // the requested blob is NOT satisfied — never report success. + error!( + requested_blob_id = %blob_id, + stored_blob_id = %stored_blob_id, + %from_peer, + size = size, + "Downloaded blob content hash does not match the requested id; discarding" + ); + } Err(e) => { error!( blob_id = %blob_id, diff --git a/crates/node/src/handlers/network_event/heartbeat.rs b/crates/node/src/handlers/network_event/heartbeat.rs index ff149b5cd0..293a8e8b3d 100644 --- a/crates/node/src/handlers/network_event/heartbeat.rs +++ b/crates/node/src/handlers/network_event/heartbeat.rs @@ -1,11 +1,21 @@ use std::collections::HashSet; +use std::time::{Duration, Instant}; use actix::{AsyncContext, WrapFuture}; use calimero_primitives::context::ContextId; use tracing::{debug, error, info, warn}; +use crate::constants::HASH_HEARTBEAT_FREQUENCY_S; use crate::NodeManager; +/// Minimum spacing between hash-heartbeat "peer is ahead" recovery syncs for a +/// given context. A context-wide sync serves every peer, so this collapses the +/// per-peer heartbeat storm (N ahead peers → N syncs each ~30s cycle) to at +/// most one sync per context per cycle. Sized just under the heartbeat cadence +/// so a still-behind context can re-sync on the next cycle. +const HEARTBEAT_BEHIND_SYNC_DEBOUNCE: Duration = + Duration::from_secs(HASH_HEARTBEAT_FREQUENCY_S.saturating_sub(5)); + /// Cap on per-child `warn!` events emitted from the divergence dump. /// A wide context (e.g. UnorderedMap with hundreds of entries) could /// otherwise produce a log burst that overwhelms aggregation @@ -142,6 +152,7 @@ pub(super) fn handle_hash_heartbeat( our_hash, their_hash: their_root_hash, count, + last_seen: Instant::now(), }, ); @@ -292,6 +303,23 @@ pub(super) fn handle_hash_heartbeat( return; } + // Debounce per context: one recovery sync covers whatever any peer + // is ahead on, so don't spawn another within the window even as + // other peers' heartbeats keep reporting us behind. + let now = Instant::now(); + if let Some(last) = manager.behind_sync_at.get(&context_id) { + if now.duration_since(*last) < HEARTBEAT_BEHIND_SYNC_DEBOUNCE { + debug!( + %context_id, + ?source, + "Peer has heads we don't have, but a recovery sync ran recently; \ + skipping (debounced)" + ); + return; + } + } + let _ = manager.behind_sync_at.insert(context_id, now); + info!( %context_id, ?source, diff --git a/crates/node/src/handlers/network_event/namespace.rs b/crates/node/src/handlers/network_event/namespace.rs index fbdc812716..0e92467bb8 100644 --- a/crates/node/src/handlers/network_event/namespace.rs +++ b/crates/node/src/handlers/network_event/namespace.rs @@ -125,236 +125,247 @@ pub(super) fn handle_namespace_governance_delta( return; } - let context_client = this.clients.context.clone(); - let node_client = this.clients.node.clone(); - let node_state = this.state.clone(); - let network_client = this.managers.sync.network_client.clone(); - // Clone of the sync manager used to fire reconcile-via-anchor on - // a `MemberRemoved` / `MemberLeft` apply that reports state-hash - // divergence. Cloning is cheap (Arc-wrapped fields) and the - // spawned task needs an owned handle. - let sync_manager = this.managers.sync.clone(); - let sync_timeout = this.managers.sync.sync_config.timeout; - let pull_budget_max_peers = this.managers.sync.sync_config.parent_pull_additional_peers; - let pull_budget_duration = this.managers.sync.sync_config.parent_pull_budget; - let readiness_addr = this.readiness_addr.clone(); - // PR-6c Task 6c.8: capture the migration-heartbeat emitter address + a - // datastore handle so the apply path can drive an on-change heartbeat once - // the governance op applies (mirrors the `readiness_addr` capture). Cloning - // is cheap (Arc-wrapped) and the spawned future needs owned handles. - let migration_emitter_addr = this.migration_emitter_addr.clone(); - let migration_datastore = this.datastore.clone(); + // Bundle the handles the off-actor apply needs into one owned context, so + // this function stays a thin decode+route+spawn shell and the apply / + // side-effect steps live in named methods rather than a ~200-line future. + let apply = NamespaceDeltaApply { + context_client: this.clients.context.clone(), + node_client: this.clients.node.clone(), + node_state: this.state.clone(), + network_client: this.managers.sync.network_client.clone(), + sync_manager: this.managers.sync.clone(), + sync_timeout: this.managers.sync.sync_config.timeout, + pull_budget_max_peers: this.managers.sync.sync_config.parent_pull_additional_peers, + pull_budget_duration: this.managers.sync.sync_config.parent_pull_budget, + readiness_addr: this.readiness_addr.clone(), + migration_emitter_addr: this.migration_emitter_addr.clone(), + migration_datastore: this.datastore.clone(), + }; - let op_for_ack = op.clone(); - // Capture the signer for the peer-identity cache before `op` is - // consumed by `apply_signed_namespace_op` below. Signature was - // already verified above, so the identity is real; the recording - // itself is gated on `Applied` below to skip relayed duplicates and - // pending ops where the delivering peer didn't necessarily originate - // the message. + // Capture the signer for the peer-identity cache and a clone of the op for + // the ack before `op` is consumed by the apply. The signature was already + // verified above, so the identity is real; recording it is gated on + // `Applied` (see `on_applied`) to skip relayed duplicates and pending ops + // where the delivering peer didn't necessarily originate the message. let signer = op.signer; + let op_for_ack = op.clone(); let _ignored = ctx.spawn( async move { - let outcome = match context_client.apply_signed_namespace_op(op).await { - Ok(outcome) => outcome, - Err(err) => { - warn!(?err, %source, "failed to apply namespace governance delta"); - return; - } - }; + apply + .run(op, op_for_ack, signer, source, namespace_id) + .await + } + .into_actor(this), + ); +} - // Notify the ReadinessManager FSM that we've made local - // progress on this namespace. Without this signal, - // `state_per_namespace` stays empty forever, no beacons emit, - // and the readiness subsystem is inert (#2269 cursor[bot] - // HIGH-severity finding). `Pending` and `Duplicate` outcomes - // do NOT advance our applied count — Pending is waiting on - // parents (no real progress yet) and Duplicate is a re-deliver. - if let NamespaceApplyOutcome::Applied { divergence } = &outcome { - if let Some(addr) = &readiness_addr { - addr.do_send(crate::readiness::NamespaceOpApplied { namespace_id }); - } +/// Owned handles captured from [`NodeManager`] to apply one verified namespace +/// governance op off the actor and run its follow-up side effects. All fields +/// are cheap (Arc-backed) clones. +struct NamespaceDeltaApply { + context_client: calimero_context_client::client::ContextClient, + node_client: calimero_node_primitives::client::NodeClient, + node_state: crate::NodeState, + network_client: NetworkClient, + sync_manager: crate::sync::SyncManager, + sync_timeout: tokio::time::Duration, + pull_budget_max_peers: usize, + pull_budget_duration: tokio::time::Duration, + readiness_addr: Option>, + migration_emitter_addr: Option>, + migration_datastore: calimero_store::Store, +} - // PR-6c Task 6c.8: drive the migration-heartbeat emitter on the - // same applied edge. Recompute the node's facts from the (now - // updated) local governance state and post them — this seeds the - // namespace into the emitter (making its periodic keep-alive - // tick live) and edge-triggers an on-change heartbeat when the - // target schema / residue changed. Best-effort: a `None` address - // (emitter not yet mounted) drops the signal; the next applied - // op re-drives it. - if let Some(addr) = &migration_emitter_addr { - let facts = crate::migration_status::compute_namespace_migration_facts( - &migration_datastore, - namespace_id, - ); - addr.do_send(crate::migration_status::MigrationFactsUpdate { - namespace_id, - facts, - }); - } +impl NamespaceDeltaApply { + /// Apply the op and route on its outcome: on `Applied`, run the side + /// effects (see [`Self::on_applied`]) then ack; on `Pending`, proactively + /// backfill from the sender and parent-pull the rest of the mesh. `Pending` + /// and `Duplicate` neither advance our applied count nor ack. + async fn run( + self, + op: SignedNamespaceOp, + op_for_ack: SignedNamespaceOp, + signer: calimero_primitives::identity::PublicKey, + source: libp2p::PeerId, + namespace_id: [u8; 32], + ) { + let outcome = match self.context_client.apply_signed_namespace_op(op).await { + Ok(outcome) => outcome, + Err(err) => { + warn!(?err, %source, "failed to apply namespace governance delta"); + return; + } + }; - // Record the (peer, identity) pair now that the - // signature verified, the nonce was monotonic, and the - // op applied successfully. Consumed by anchor-preferred - // sync peer selection. See `NodeState::peer_identities`. - // - // `None`: this path doesn't carry the signer's group + - // role cheaply, so the observation updates only the - // in-memory reverse view. The durable cache is populated - // from the state-delta path (which has both) — the deltas - // that follow a member's governance op re-observe them - // there, so cold-start coverage is unaffected. - node_state.observe_peer_identity(source, signer, None); - - // Governance-pending active drain: a governance op that just applied may - // unblock state deltas previously buffered as `Unknown`. - // Without this hook the lazy on-state-delta drain - // deadlocks when the only state delta in flight is one - // waiting for that very governance op (e2e 3-node test - // reproduced this). - let drain_input = crate::handlers::state_delta::StateDeltaContext { - node_clients: crate::state::NodeClients { - context: context_client.clone(), - node: node_client.clone(), - }, - node_state: node_state.clone(), - network_client: network_client.clone(), - sync_timeout, - }; - crate::handlers::state_delta::drain_all_governance_pending(&drain_input).await; - // PR-6b Task 6b.5: a cascade-upgrade governance op that just - // applied bumps the group's target schema and is the catalyst - // for this node's lazy binary advance. Drain any absorbed - // straggler deltas the now-loaded reader can read (verbatim - // replay); no-op for contexts that haven't advanced. - crate::handlers::state_delta::drain_all_absorbed(&drain_input).await; + if let NamespaceApplyOutcome::Applied { divergence } = &outcome { + self.on_applied(divergence, source, signer, namespace_id) + .await; + } - // Reconcile-via-anchor: if `MemberRemoved` / `MemberLeft` - // apply reported state-hash divergence from the signed - // claims, pull canonical state from a trusted anchor. - // The sync manager method picks an anchor peer (from - // the group's trusted-anchor set, filtered through the - // verified peer-identity cache) and verifies the - // post-adoption hash against the signed expected. - // Best-effort: no anchor connected → logged and - // dropped; re-attempted on the next signed op or - // sync tick. - // - // Detached via `actix::spawn` so a multi-second - // Snapshot / HashComparison sync doesn't block the - // governance-apply task — the actor's mailbox needs - // to keep draining (other signed ops, acks, and the - // proactive backfill for `Pending`). `actix::spawn` - // is the right primitive here: the inner future - // touches non-`Send` types via the sync manager, so - // `tokio::spawn` (which requires `Send`) can't take - // it; `actix::spawn` schedules on the current arbiter - // and stays single-threaded with the actor. The - // arbiter's task queue runs the detached future - // concurrently with the actor's mailbox drain. - // Errors are already logged inside - // `reconcile_after_divergence`; the spawned task has - // no return value to consume. - if let Some(report) = divergence { - let sm = sync_manager.clone(); - let report = report.clone(); - drop(actix::spawn(async move { - sm.reconcile_after_divergence(report).await; - })); - } + // Emit a `SignedAck` on the same topic only when we've newly applied the + // op. `Pending` (waiting on parents) would lie about application, and + // `Duplicate` (already had it) would just inflate gossip with no + // observable change to the publisher's dedup-by-signer counting. + if matches!(outcome, NamespaceApplyOutcome::Applied { .. }) { + emit_namespace_ack( + &self.context_client, + &self.network_client, + namespace_id, + &op_for_ack, + ) + .await; + } - // Prompt key recovery on receipt (#2613). A gossiped Group - // op we couldn't decrypt is now buffered awaiting its key. - // Trigger a direct key pull immediately rather than waiting - // for a sync tick — crucially, this is the ONLY recovery - // trigger for a namespace/subgroup member that holds no - // local context (e.g. a Restricted-subgroup member just - // added via `add_group_members`): the per-context interval - // recovery never runs for it, and without a prompt pull it - // never decrypts the subgroup's ops (membership, - // `ContextRegistered`) and so can't see or join the - // subgroup. Cheap when nothing is awaiting (one op-log - // scan); tries multiple peers. Detached for the same - // non-`Send` reason as the reconcile above. - { - let sm = sync_manager.clone(); - drop(actix::spawn(async move { - sm.recover_missing_group_keys(namespace_id, None).await; - })); - } - } + // Proactive backfill fires ONLY for `Pending` — the DAG accepted the op + // but can't apply it until missing parents arrive. `Applied` is the + // happy path; `Duplicate` means we already have the op (common on + // gossip rebroadcast) and a backfill for it would pull the whole + // namespace for nothing. + // + // We MUST fetch from `source` first before handing off to + // `resolve_namespace_pending`: that helper seeds its `ParentPullBudget` + // with the initial peer already marked tried, so passing `source` + // straight to it means `source` is never actually queried — which in a + // 2-node mesh silently does nothing. Empty `delta_ids` means "give me + // everything for this namespace" on the responder side. + if matches!(outcome, NamespaceApplyOutcome::Pending) { + debug!( + %source, + namespace_id = %hex::encode(namespace_id), + "gossip governance op is pending; triggering proactive backfill" + ); + fetch_and_apply_namespace_backfill( + &self.context_client, + &self.node_client, + &self.network_client, + &self.sync_manager, + source, + namespace_id, + Vec::new(), + self.sync_timeout, + &self.node_state, + ) + .await; + resolve_namespace_pending( + &self.context_client, + &self.node_client, + &self.network_client, + &self.sync_manager, + source, + namespace_id, + self.sync_timeout, + self.pull_budget_max_peers, + self.pull_budget_duration, + &self.node_state, + ) + .await; + } - // Phase 4: emit a `SignedAck` on the same topic when we've - // newly applied the op. `Pending` (waiting on parents) and - // `Duplicate` (we already had it — likely already acked - // earlier) deliberately don't ack: Pending would lie about - // application, and Duplicate would just inflate gossip with - // no observable change to the publisher's dedup-by-signer - // counting. - if matches!(outcome, NamespaceApplyOutcome::Applied { .. }) { - emit_namespace_ack(&context_client, &network_client, namespace_id, &op_for_ack) - .await; - } + // Group-key delivery to a new joiner is no longer pushed from here onto + // the namespace governance DAG. The joiner pulls the key directly from a + // sync peer when it next syncs the namespace and finds it lacks the key + // (see `SyncManager::recover_missing_group_keys`), a durable retry + // instead of the old single fire-and-forget shot. + } - // Proactive backfill (#2198) fires ONLY for `Pending` — the - // DAG accepted the op but can't apply it until missing parents - // arrive. `Applied` is the steady-state happy path; `Duplicate` - // means we already have the op (very common on gossip, since - // every mesh peer rebroadcasts), and triggering a backfill for - // it would open a stream and request the full namespace state - // for nothing. - // - // NOTE: we MUST ask `source` first before handing off to - // `resolve_namespace_pending`. That helper seeds its - // `ParentPullBudget` with the initial peer marked as already - // tried, so passing `source` to it directly without a prior - // fetch means `source` never actually gets queried — which in - // a 2-node mesh (where no other peers exist) silently does - // nothing. Empty `delta_ids` means "give me everything for - // this namespace" on the responder side. - if matches!(outcome, NamespaceApplyOutcome::Pending) { - debug!( - %source, - namespace_id = %hex::encode(namespace_id), - "gossip governance op is pending; triggering proactive backfill" - ); - fetch_and_apply_namespace_backfill( - &context_client, - &node_client, - &network_client, - &sync_manager, - source, - namespace_id, - Vec::new(), - sync_timeout, - &node_state, - ) - .await; - resolve_namespace_pending( - &context_client, - &node_client, - &network_client, - &sync_manager, - source, - namespace_id, - sync_timeout, - pull_budget_max_peers, - pull_budget_duration, - &node_state, - ) - .await; - } + /// Side effects that run only when the op *newly* applied: nudge the + /// readiness FSM, drive an on-change migration heartbeat, record the + /// (peer, identity) pair, drain governance-pending / absorbed state deltas, + /// reconcile via a trusted anchor on reported divergence, and prompt group- + /// key recovery. + async fn on_applied( + &self, + divergence: &Option, + source: libp2p::PeerId, + signer: calimero_primitives::identity::PublicKey, + namespace_id: [u8; 32], + ) { + // Notify the ReadinessManager FSM that we've made local progress on this + // namespace. Without this signal `state_per_namespace` stays empty + // forever, no beacons emit, and the readiness subsystem is inert. + if let Some(addr) = &self.readiness_addr { + addr.do_send(crate::readiness::NamespaceOpApplied { namespace_id }); + } - // Group-key delivery to a new joiner is no longer pushed from - // here onto the namespace governance DAG. The joiner pulls the - // key directly from a sync peer when it next syncs the - // namespace and finds it lacks the key (see - // `SyncManager::recover_missing_group_keys`), which gives it a - // durable retry instead of the old single fire-and-forget shot. + // Drive the migration-heartbeat emitter on the same applied edge: + // recompute the node's facts from the now-updated local governance state + // and post them — seeding the namespace into the emitter (so its + // periodic keep-alive tick goes live) and edge-triggering an on-change + // heartbeat when the target schema / residue changed. Best-effort: a + // `None` address drops the signal and the next applied op re-drives it. + if let Some(addr) = &self.migration_emitter_addr { + let facts = crate::migration_status::compute_namespace_migration_facts( + &self.migration_datastore, + namespace_id, + ); + addr.do_send(crate::migration_status::MigrationFactsUpdate { + namespace_id, + facts, + }); } - .into_actor(this), - ); + + // Record the (peer, identity) pair now that the signature verified, the + // nonce was monotonic, and the op applied. Consumed by anchor-preferred + // sync peer selection. `None`: this path doesn't carry the signer's + // group + role cheaply, so it updates only the in-memory reverse view; + // the durable cache is populated from the state-delta path that follows. + self.node_state.observe_peer_identity(source, signer, None); + + // Governance-pending active drain: a governance op that just applied may + // unblock state deltas previously buffered as `Unknown`. Without this + // the lazy on-state-delta drain deadlocks when the only in-flight state + // delta is one waiting for that very governance op. + let drain_input = crate::handlers::state_delta::StateDeltaContext { + node_clients: crate::state::NodeClients { + context: self.context_client.clone(), + node: self.node_client.clone(), + }, + node_state: self.node_state.clone(), + network_client: self.network_client.clone(), + sync_timeout: self.sync_timeout, + }; + crate::handlers::state_delta::drain_all_governance_pending(&drain_input).await; + // A cascade-upgrade governance op that just applied bumps the group's + // target schema and is the catalyst for this node's lazy binary advance. + // Drain any absorbed straggler deltas the now-loaded reader can read + // (verbatim replay); no-op for contexts that haven't advanced. + crate::handlers::state_delta::drain_all_absorbed(&drain_input).await; + + // Reconcile-via-anchor: if a `MemberRemoved` / `MemberLeft` apply + // reported state-hash divergence from the signed claims, pull canonical + // state from a trusted anchor (verifying the post-adoption hash against + // the signed expected). Best-effort: no anchor connected → logged and + // dropped, re-attempted on the next signed op or sync tick. Detached via + // `actix::spawn` so a multi-second Snapshot / HashComparison sync doesn't + // block the actor's mailbox drain; the inner future touches non-`Send` + // sync-manager types, so `actix::spawn` (same arbiter, single-threaded) + // is required over `tokio::spawn`. Errors are logged inside + // `reconcile_after_divergence`. + if let Some(report) = divergence { + let sm = self.sync_manager.clone(); + let report = report.clone(); + drop(actix::spawn(async move { + sm.reconcile_after_divergence(report).await; + })); + } + + // Prompt key recovery on receipt: a gossiped Group op we couldn't + // decrypt is now buffered awaiting its key. Trigger a direct key pull + // immediately rather than waiting for a sync tick — this is the ONLY + // recovery trigger for a namespace/subgroup member that holds no local + // context (e.g. a Restricted-subgroup member just added via + // `add_group_members`): the per-context interval recovery never runs for + // it, and without a prompt pull it never decrypts the subgroup's ops and + // so can't see or join the subgroup. Cheap when nothing is awaiting; + // tries multiple peers. Detached for the same non-`Send` reason as the + // reconcile above. + { + let sm = self.sync_manager.clone(); + drop(actix::spawn(async move { + sm.recover_missing_group_keys(namespace_id, None).await; + })); + } + } } pub(super) fn handle_namespace_state_heartbeat( @@ -439,6 +450,14 @@ pub(super) fn handle_namespace_state_heartbeat( ); } +/// Base delay between namespace parent-pull peer attempts; doubled per attempt. +const PARENT_PULL_BASE_BACKOFF: std::time::Duration = std::time::Duration::from_millis(200); +/// Upper bound on the per-attempt backoff, regardless of attempt count. +const PARENT_PULL_MAX_BACKOFF: std::time::Duration = std::time::Duration::from_secs(2); +/// Cap on the exponential shift so `1 << shift` cannot overflow and the delay +/// saturates at [`PARENT_PULL_MAX_BACKOFF`]. +const PARENT_PULL_MAX_BACKOFF_SHIFT: u32 = 4; + /// Iterate other namespace-mesh peers asking for backfill until the local /// governance DAG has no more pending ops for this namespace, or the retry /// budget is exhausted. @@ -534,6 +553,16 @@ async fn resolve_namespace_pending( node_state, ) .await; + + // Back off between peer attempts so a namespace that stays pending does + // not spin a tight request loop against the mesh. Exponential with a + // cap; the scheduler's wall-clock `budget` still bounds total time (this + // sleep counts against it), so backoff only paces the attempts. + let shift = (scheduler.attempts() as u32).min(PARENT_PULL_MAX_BACKOFF_SHIFT); + let backoff = PARENT_PULL_BASE_BACKOFF + .saturating_mul(1_u32 << shift) + .min(PARENT_PULL_MAX_BACKOFF); + tokio::time::sleep(backoff).await; } } @@ -599,7 +628,20 @@ async fn fetch_and_apply_namespace_backfill( // so the apply loop is contiguous. let mut pending_divergences: Vec = Vec::new(); - for (delta_id, op_bytes) in deltas { + // Cap what we apply regardless of what the responder sent. A + // cooperating responder already trims to this bound, but a + // misbehaving one must not be able to drive unbounded apply work by + // overfilling the response. + if deltas.len() > crate::sync::MAX_BACKFILL_OPS { + warn!( + %peer, + namespace_id = %hex::encode(namespace_id), + received = deltas.len(), + cap = crate::sync::MAX_BACKFILL_OPS, + "namespace backfill response exceeds cap; applying only the first cap ops" + ); + } + for (delta_id, op_bytes) in deltas.into_iter().take(crate::sync::MAX_BACKFILL_OPS) { if let Ok(op) = borsh::from_slice::(&op_bytes) { match context_client.apply_signed_namespace_op(op).await { Ok(NamespaceApplyOutcome::Applied { divergence }) => { diff --git a/crates/node/src/handlers/network_event/subscriptions.rs b/crates/node/src/handlers/network_event/subscriptions.rs index b88da39782..ebd3d492c0 100644 --- a/crates/node/src/handlers/network_event/subscriptions.rs +++ b/crates/node/src/handlers/network_event/subscriptions.rs @@ -111,18 +111,27 @@ pub(super) fn handle_subscribed( return; }; - if !manager - .clients - .context - .has_context(&context_id) - .unwrap_or_default() - { - debug!( - %context_id, - %peer_id, - "Observed subscription to unknown context, ignoring.." - ); - return; + match manager.clients.context.has_context(&context_id) { + Ok(true) => {} + Ok(false) => { + debug!( + %context_id, + %peer_id, + "Observed subscription to unknown context, ignoring.." + ); + return; + } + Err(err) => { + // A store error is unknown state, not "no such context". Surface it + // and bail rather than silently treating the context as absent. + warn!( + %context_id, + %peer_id, + %err, + "has_context lookup failed while handling subscription; ignoring" + ); + return; + } } info!( diff --git a/crates/node/src/manager.rs b/crates/node/src/manager.rs index 051159fedd..45463f068f 100644 --- a/crates/node/src/manager.rs +++ b/crates/node/src/manager.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex}; -use std::time::Instant; +use std::time::{Duration, Instant}; use actix::{Actor, Addr}; use calimero_blobstore::BlobManager as BlobStore; @@ -32,8 +32,18 @@ pub(crate) struct DivergenceMark { pub(crate) our_hash: calimero_primitives::hash::Hash, pub(crate) their_hash: calimero_primitives::hash::Hash, pub(crate) count: u32, + /// When this entry was last observed. Lets the heartbeat tick evict marks + /// for peers that diverged and then went away without ever converging (a + /// convergence is what normally removes the entry), so the map can't grow + /// without bound on peer churn. + pub(crate) last_seen: Instant, } +/// Evict divergence / behind-sync bookkeeping for a (context, peer) or context +/// not seen within this window. Generous relative to the 30s heartbeat cadence: +/// only genuinely stale churned-away entries are reclaimed. +pub(crate) const HEARTBEAT_STATE_TTL: Duration = Duration::from_secs(300); + /// Main node orchestrator. /// /// **SRP Applied**: Clear separation of: @@ -98,6 +108,13 @@ pub struct NodeManager { /// surfacing a genuinely stuck split-brain. See [`DivergenceMark`]. pub(crate) divergence_streak: HashMap<(calimero_primitives::context::ContextId, libp2p::PeerId), DivergenceMark>, + /// Per-context timestamp of the last hash-heartbeat "peer is ahead" + /// recovery sync. A context-wide `sync` covers every peer at once, so + /// without this every ahead-peer's 30s heartbeat would spawn its own sync + /// (N peers → N syncs per cycle). Debounced to one per + /// `HEARTBEAT_BEHIND_SYNC_DEBOUNCE` window per context. Touched only + /// synchronously on the manager actor, so a plain `HashMap` suffices. + pub(crate) behind_sync_at: HashMap, /// Per-namespace timestamp of the last beacon-*triggered* governance /// sync (#2367). Caps beacon-divergence syncs to one per namespace /// per `NS_BEACON_SYNC_DEBOUNCE` window — beacons arrive every ~5s @@ -161,6 +178,7 @@ impl NodeManager { sync_session_tx, divergence_detected, divergence_streak: HashMap::new(), + behind_sync_at: HashMap::new(), ns_beacon_sync_debounce: Arc::new(Mutex::new(HashMap::new())), migration_status_cache: Arc::new(MigrationStatusCache::default()), migration_emitter_addr: None, @@ -172,6 +190,10 @@ impl Actor for NodeManager { type Context = actix::Context; fn started(&mut self, ctx: &mut Self::Context) { + // Bound the mailbox so the network-event bridge's `try_send` sees + // backpressure instead of growing it without limit. `do_send` callers + // (which ignore capacity) are unaffected. + ctx.set_mailbox_capacity(crate::network_event_processor::NODE_MANAGER_MAILBOX_CAPACITY); self.setup_startup_subscriptions(ctx); self.setup_maintenance_intervals(ctx); self.setup_hash_heartbeat_interval(ctx); diff --git a/crates/node/src/manager/startup.rs b/crates/node/src/manager/startup.rs index f79cac8a96..da884bea4f 100644 --- a/crates/node/src/manager/startup.rs +++ b/crates/node/src/manager/startup.rs @@ -171,6 +171,16 @@ impl NodeManager { let _handle = ctx.run_interval( Duration::from_secs(constants::HASH_HEARTBEAT_FREQUENCY_S), |act, ctx| { + // Reclaim heartbeat bookkeeping for peers/contexts that have gone + // quiet, so these maps can't grow without bound on peer churn. + // Both are touched only here and in the (synchronous) heartbeat + // handler on this actor, so a plain retain is race-free. + let now = std::time::Instant::now(); + act.divergence_streak + .retain(|_, mark| now.duration_since(mark.last_seen) < crate::manager::HEARTBEAT_STATE_TTL); + act.behind_sync_at + .retain(|_, last| now.duration_since(*last) < crate::manager::HEARTBEAT_STATE_TTL); + let context_client = act.clients.context.clone(); let node_client = act.clients.node.clone(); diff --git a/crates/node/src/network_event_channel.rs b/crates/node/src/network_event_channel.rs index 34cfe0ef52..f836f9efc1 100644 --- a/crates/node/src/network_event_channel.rs +++ b/crates/node/src/network_event_channel.rs @@ -32,6 +32,17 @@ use tokio::sync::mpsc; use tokio::time::timeout; use tracing::{debug, error, info, warn}; +/// Minimum spacing between repeated channel-pressure warnings, so sustained +/// overload does not turn them into a per-event log flood. +const PRESSURE_WARN_INTERVAL: Duration = Duration::from_secs(5); + +/// Process uptime in milliseconds, measured from a lazily captured base +/// instant. Backs the lock-free warning throttle. +fn process_uptime_millis() -> u64 { + static BASE: std::sync::OnceLock = std::sync::OnceLock::new(); + BASE.get_or_init(Instant::now).elapsed().as_millis() as u64 +} + /// Configuration for the network event channel. #[derive(Debug, Clone, Copy)] pub struct NetworkEventChannelConfig { @@ -120,6 +131,14 @@ pub struct NetworkEventChannelMetrics { /// retry path. Backs `max_pending_retries` and drives /// `backpressure_active`. Not a registered metric (operational state). pub pending_retries: Arc, + + /// Process-uptime (ms) of the last emitted "approaching capacity" warning. + /// Used to throttle that log line so sustained pressure can't flood it. + pub last_capacity_warn_ms: Arc, + + /// Process-uptime (ms) of the last emitted backpressure warning; throttles + /// that log line the same way. + pub last_backpressure_warn_ms: Arc, } impl NetworkEventChannelMetrics { @@ -190,6 +209,8 @@ impl NetworkEventChannelMetrics { processing_latency, high_watermark: Arc::new(AtomicU64::new(0)), pending_retries: Arc::new(AtomicU64::new(0)), + last_capacity_warn_ms: Arc::new(AtomicU64::new(0)), + last_backpressure_warn_ms: Arc::new(AtomicU64::new(0)), } } @@ -207,6 +228,26 @@ impl NetworkEventChannelMetrics { processing_latency: Histogram::new(exponential_buckets(0.0001, 2.0, 18)), high_watermark: Arc::new(AtomicU64::new(0)), pending_retries: Arc::new(AtomicU64::new(0)), + last_capacity_warn_ms: Arc::new(AtomicU64::new(0)), + last_backpressure_warn_ms: Arc::new(AtomicU64::new(0)), + } + } + + /// Returns `true` at most once per `min_interval` for the given throttle + /// slot, stamping it with the current process uptime. Lock-free; a slot of + /// `0` means "never warned" so the first call always passes. + fn throttle_warn(slot: &AtomicU64, min_interval: Duration) -> bool { + let now = process_uptime_millis(); + let interval = min_interval.as_millis() as u64; + let mut last = slot.load(Ordering::Relaxed); + loop { + if last != 0 && now.saturating_sub(last) < interval { + return false; + } + match slot.compare_exchange_weak(last, now, Ordering::AcqRel, Ordering::Relaxed) { + Ok(_) => return true, + Err(actual) => last = actual, + } } } @@ -279,7 +320,12 @@ impl NetworkEventSender { // Check warning threshold let fill_ratio = current_depth as f64 / max_capacity as f64; - if fill_ratio >= self.config.warning_threshold { + if fill_ratio >= self.config.warning_threshold + && NetworkEventChannelMetrics::throttle_warn( + &self.metrics.last_capacity_warn_ms, + PRESSURE_WARN_INTERVAL, + ) + { warn!( current_depth, max_capacity, @@ -335,12 +381,17 @@ impl NetworkEventSender { self.metrics.events_retried.inc(); self.metrics.backpressure_active.set(1); - warn!( - event_type, - pending, - channel_size = self.config.channel_size, - "Network event channel full - applying backpressure (retrying event)" - ); + if NetworkEventChannelMetrics::throttle_warn( + &self.metrics.last_backpressure_warn_ms, + PRESSURE_WARN_INTERVAL, + ) { + warn!( + event_type, + pending, + channel_size = self.config.channel_size, + "Network event channel full - applying backpressure (retrying event)" + ); + } let tx = self.tx.clone(); let metrics = self.metrics.clone(); @@ -451,6 +502,12 @@ impl NetworkEventReceiver { Some(timestamped.event) } + /// A clone of the shared channel metrics, so downstream stages (e.g. the + /// NodeManager bridge) can record into the same series. + pub fn metrics(&self) -> NetworkEventChannelMetrics { + self.metrics.clone() + } + /// Try to receive without blocking. pub fn try_recv(&mut self) -> Option { match self.rx.try_recv() { diff --git a/crates/node/src/network_event_processor.rs b/crates/node/src/network_event_processor.rs index 0974073e2f..b645392bc1 100644 --- a/crates/node/src/network_event_processor.rs +++ b/crates/node/src/network_event_processor.rs @@ -18,15 +18,50 @@ //! async spawn patterns that work within Actix's actor model. use std::sync::Arc; +use std::time::Duration; use actix::Addr; use calimero_network_primitives::messages::NetworkEvent; use tokio::sync::Notify; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; -use crate::network_event_channel::NetworkEventReceiver; +use crate::network_event_channel::{NetworkEventChannelMetrics, NetworkEventReceiver}; use crate::NodeManager; +/// Bound on NodeManager's actor mailbox. `try_send` (below) respects this, so +/// the network-event feed can no longer grow the mailbox without limit — unlike +/// `do_send`, which ignores capacity entirely. +pub(crate) const NODE_MANAGER_MAILBOX_CAPACITY: usize = 1024; + +/// How many times `forward_event` retries a full mailbox before dropping. +const MAILBOX_SEND_MAX_RETRIES: usize = 20; + +/// Delay between mailbox-full retries. `MAILBOX_SEND_MAX_RETRIES` × this bounds +/// how long a single event is held (and how long the source channel is left to +/// backpressure) before the event is dropped. +const MAILBOX_SEND_RETRY_BACKOFF: Duration = Duration::from_millis(25); + +/// Stable label for a network event, for logs and drop metrics. +fn event_type_str(event: &NetworkEvent) -> &'static str { + match event { + NetworkEvent::Message { .. } => "Message", + NetworkEvent::StreamOpened { .. } => "StreamOpened", + NetworkEvent::Subscribed { .. } => "Subscribed", + NetworkEvent::Unsubscribed { .. } => "Unsubscribed", + NetworkEvent::ListeningOn { .. } => "ListeningOn", + NetworkEvent::BlobRequested { .. } => "BlobRequested", + NetworkEvent::BlobProvidersFound { .. } => "BlobProvidersFound", + NetworkEvent::BlobDownloaded { .. } => "BlobDownloaded", + NetworkEvent::BlobDownloadFailed { .. } => "BlobDownloadFailed", + NetworkEvent::SpecializedNodeVerificationRequest { .. } => { + "SpecializedNodeVerificationRequest" + } + NetworkEvent::SpecializedNodeInvitationResponse { .. } => { + "SpecializedNodeInvitationResponse" + } + } +} + /// Bridge that forwards events from the channel to NodeManager. /// /// This ensures events are reliably delivered to the NodeManager actor, @@ -38,6 +73,10 @@ pub struct NetworkEventBridge { /// NodeManager actor address. node_manager: Addr, + /// Shared channel metrics, so a mailbox-full drop is counted in the same + /// `events_dropped` series as source-channel drops. + metrics: NetworkEventChannelMetrics, + /// Shutdown signal. shutdown: Arc, } @@ -46,6 +85,7 @@ impl NetworkEventBridge { /// Create a new bridge. pub fn new(receiver: NetworkEventReceiver, node_manager: Addr) -> Self { Self { + metrics: receiver.metrics(), receiver, node_manager, shutdown: Arc::new(Notify::new()), @@ -71,7 +111,7 @@ impl NetworkEventBridge { event = self.receiver.recv() => { match event { Some(event) => { - self.forward_event(event); + self.forward_event(event).await; } None => { info!("Network event channel closed, shutting down bridge"); @@ -95,31 +135,41 @@ impl NetworkEventBridge { } /// Forward a single event to NodeManager. - fn forward_event(&self, event: NetworkEvent) { - // Log event type for debugging - let event_type = match &event { - NetworkEvent::Message { .. } => "Message", - NetworkEvent::StreamOpened { .. } => "StreamOpened", - NetworkEvent::Subscribed { .. } => "Subscribed", - NetworkEvent::Unsubscribed { .. } => "Unsubscribed", - NetworkEvent::ListeningOn { .. } => "ListeningOn", - NetworkEvent::BlobRequested { .. } => "BlobRequested", - NetworkEvent::BlobProvidersFound { .. } => "BlobProvidersFound", - NetworkEvent::BlobDownloaded { .. } => "BlobDownloaded", - NetworkEvent::BlobDownloadFailed { .. } => "BlobDownloadFailed", - NetworkEvent::SpecializedNodeVerificationRequest { .. } => { - "SpecializedNodeVerificationRequest" - } - NetworkEvent::SpecializedNodeInvitationResponse { .. } => { - "SpecializedNodeInvitationResponse" - } - }; - + /// + /// Uses `try_send` against NodeManager's bounded mailbox instead of + /// `do_send`: `do_send` ignores the mailbox capacity, so draining the + /// bounded source channel straight into it let the mailbox grow without + /// limit. A short bounded retry rides out transient fullness (and, by not + /// pulling the next event, lets the source channel apply its own + /// backpressure); a mailbox that stays full means NodeManager is + /// overwhelmed, so the event is dropped with a counter rather than buffered + /// forever. + async fn forward_event(&self, event: NetworkEvent) { + let event_type = event_type_str(&event); debug!(event_type, "Forwarding network event to NodeManager"); - // Forward to NodeManager - this uses Actix's do_send which is reliable - // within the same Actix system - self.node_manager.do_send(event); + let mut event = event; + for _ in 0..MAILBOX_SEND_MAX_RETRIES { + match self.node_manager.try_send(event) { + Ok(()) => return, + Err(actix::dev::SendError::Closed(_)) => { + debug!(event_type, "NodeManager mailbox closed; dropping event"); + return; + } + Err(actix::dev::SendError::Full(returned)) => { + event = returned; + tokio::time::sleep(MAILBOX_SEND_RETRY_BACKOFF).await; + } + } + } + + // Sustained-full mailbox: drop with visibility rather than grow it. + self.metrics.events_dropped.inc(); + warn!( + event_type, + capacity = NODE_MANAGER_MAILBOX_CAPACITY, + "NodeManager mailbox full after retries; dropping network event" + ); } /// Graceful shutdown: drain and forward remaining events. @@ -132,8 +182,11 @@ impl NetworkEventBridge { if count > 0 { info!(count, "Forwarding remaining events before shutdown"); + // Best-effort flush on the way down: the mailbox-bounding concern + // does not apply during shutdown, so use `do_send` to hand every + // remaining event over without dropping any. for event in remaining_events { - self.forward_event(event); + self.node_manager.do_send(event); } } diff --git a/crates/node/src/readiness.rs b/crates/node/src/readiness.rs index 8bedcde5bf..25842c2477 100644 --- a/crates/node/src/readiness.rs +++ b/crates/node/src/readiness.rs @@ -496,6 +496,15 @@ impl ReadinessManager { // change, and collect the post-recompute `*Ready` snapshots // for emission. let now = Instant::now(); + + // Evict probe rate-limit stamps for (peer, namespace) pairs that have + // gone quiet, so this map can't grow without bound on peer churn. The + // stamps only gate probe responses within `beacon_interval / 2`, so any + // entry older than this TTL is safe to drop — a later probe re-inserts. + const PROBE_RESPONSE_TTL: Duration = Duration::from_secs(300); + self.last_probe_response_at + .retain(|_, at| now.duration_since(*at) < PROBE_RESPONSE_TTL); + let ttl = self.config.ttl_heartbeat; let cfg = self.config; let mut to_emit: Vec<([u8; 32], ReadinessState)> = Vec::new(); diff --git a/crates/node/src/run.rs b/crates/node/src/run.rs index 174b86f67f..4be7319dd6 100644 --- a/crates/node/src/run.rs +++ b/crates/node/src/run.rs @@ -181,7 +181,7 @@ pub async fn start(config: NodeConfig) -> eyre::Result<()> { // Create dedicated network event channel for reliable message delivery // This replaces LazyRecipient to avoid cross-arbiter message loss let channel_config = NetworkEventChannelConfig { - channel_size: 1000, // Configurable, handles burst patterns + channel_size: crate::constants::NETWORK_EVENT_CHANNEL_SIZE, warning_threshold: 0.8, // Log warning at 80% capacity stats_log_interval: Duration::from_secs(30), // On a full channel, apply backpressure (wait for capacity) rather @@ -189,7 +189,7 @@ pub async fn start(config: NodeConfig) -> eyre::Result<()> { send_timeout: Duration::from_secs(5), // Bound extra buffering from waiting overflow events to ~1x the // channel before escalating to a true drop. - max_pending_retries: 1000, + max_pending_retries: crate::constants::NETWORK_EVENT_MAX_PENDING_RETRIES, }; let (network_event_sender, network_event_receiver) = network_event_channel::channel(channel_config, &mut registry); @@ -234,15 +234,15 @@ pub async fn start(config: NodeConfig) -> eyre::Result<()> { )) .await?; - // Increased buffer sizes for better burst handling and concurrency - // 256 events: supports more concurrent WebSocket clients - // 64 sync requests: handles burst context joins/syncs - let (event_sender, _) = broadcast::channel(256); + // Channel capacities are named in `crate::constants` so the burst/concurrency + // budgets live in one place rather than as scattered magic numbers. + let (event_sender, _) = broadcast::channel(crate::constants::EVENT_BROADCAST_CHANNEL_SIZE); - let (ctx_sync_tx, ctx_sync_rx) = mpsc::channel(64); - let (ns_sync_tx, ns_sync_rx) = mpsc::channel(16); - let (ns_join_tx, ns_join_rx) = mpsc::channel(16); - let (open_subgroup_join_tx, open_subgroup_join_rx) = mpsc::channel(16); + let (ctx_sync_tx, ctx_sync_rx) = mpsc::channel(crate::constants::CTX_SYNC_CHANNEL_SIZE); + let (ns_sync_tx, ns_sync_rx) = mpsc::channel(crate::constants::NS_SYNC_CHANNEL_SIZE); + let (ns_join_tx, ns_join_rx) = mpsc::channel(crate::constants::NS_JOIN_CHANNEL_SIZE); + let (open_subgroup_join_tx, open_subgroup_join_rx) = + mpsc::channel(crate::constants::OPEN_SUBGROUP_JOIN_CHANNEL_SIZE); let sync_client = SyncClient::new(ctx_sync_tx, ns_sync_tx, ns_join_tx, open_subgroup_join_tx); @@ -250,7 +250,8 @@ pub async fn start(config: NodeConfig) -> eyre::Result<()> { // applied deltas so the in-memory DeltaStore stays current without // re-scanning the DB on every interval sync. Drained by a task // spawned once `node_state` is available (below). - let (local_delta_tx, mut local_delta_rx) = mpsc::channel(256); + let (local_delta_tx, mut local_delta_rx) = + mpsc::channel(crate::constants::LOCAL_DELTA_CHANNEL_SIZE); let node_client = NodeClient::new( datastore.clone(), diff --git a/crates/node/src/sync/manager/namespace_sync.rs b/crates/node/src/sync/manager/namespace_sync.rs index 4ccdfb7caf..41321d2fe2 100644 --- a/crates/node/src/sync/manager/namespace_sync.rs +++ b/crates/node/src/sync/manager/namespace_sync.rs @@ -21,6 +21,7 @@ use tokio::time; use tracing::{debug, info, warn}; use super::SyncManager; +use crate::sync::MAX_BACKFILL_OPS; impl SyncManager { /// Actively request governance catch-up from a specific peer whose @@ -364,10 +365,6 @@ impl SyncManager { let handle = store.handle(); let mut found = Vec::new(); - /// Maximum ops returned in a single backfill response to prevent - /// memory exhaustion from large namespace governance DAGs. - const MAX_BACKFILL_OPS: usize = 500; - if delta_ids.is_empty() { // Empty request = "give me everything for this namespace". let start = calimero_store::key::NamespaceGovOp::new(namespace_id, [0u8; 32]); diff --git a/crates/node/src/sync/mod.rs b/crates/node/src/sync/mod.rs index 59afe45303..dd1e83e518 100644 --- a/crates/node/src/sync/mod.rs +++ b/crates/node/src/sync/mod.rs @@ -35,6 +35,13 @@ mod blobs; mod config; pub(crate) mod delta_request; + +/// Maximum ops exchanged in a single namespace backfill response, capping +/// memory use from large namespace governance DAGs. Enforced on BOTH ends: the +/// responder never sends more, and the receiver never applies more, so a +/// misbehaving responder cannot push an unbounded batch either way. +pub(crate) const MAX_BACKFILL_OPS: usize = 500; + pub(crate) mod driver; mod hash_comparison; pub mod hash_comparison_protocol; diff --git a/crates/node/src/utils.rs b/crates/node/src/utils.rs index 648191b7b1..4e255f2568 100644 --- a/crates/node/src/utils.rs +++ b/crates/node/src/utils.rs @@ -25,12 +25,15 @@ pub(crate) async fn choose_stream( stream: impl Stream, rng: &mut impl Rng, ) -> Option { - let mut stream = pin!(stream); + let mut stream = pin!(stream.enumerate()); - let mut item = stream.next().await; - - let mut stream = stream.enumerate(); + let mut item = None; + // Algorithm R over the whole stream: the (idx+1)-th item (0-based `idx`) + // replaces the reservoir with probability 1/(idx+1). The first item + // (idx == 0) always fills the empty reservoir. Consuming the first element + // *before* enumerating — as an earlier version did — offset every + // probability by one and biased selection toward later elements. while let Some((idx, this)) = stream.next().await { if rng.gen_range(0..idx + 1) == 0 { item = Some(this); From ae0b7b9417ef9d74021f556080a1008ee873fc73 Mon Sep 17 00:00:00 2001 From: Sandi Fatic Date: Tue, 7 Jul 2026 07:17:54 +0200 Subject: [PATCH 3/4] fix(storage): skip redundant index read on the delete-path Shared stamp Address PR review: remove_child_from_inner already loads metadata (and thus the stored writer set) from the index right before authorizing the stamp, so authorize_local_shared_stamp re-fetching it via Index::get_metadata was a provably redundant read on every local Shared delete. Add a `stored` param: the delete path passes its already-loaded writers, the save path passes None (its `claimed` is the incoming action's set, not stored) and still looks up. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/storage/src/interface.rs | 34 ++++++++++++++++++++++----------- 1 file changed, 23 insertions(+), 11 deletions(-) diff --git a/crates/storage/src/interface.rs b/crates/storage/src/interface.rs index 93121e381b..03bea27c44 100644 --- a/crates/storage/src/interface.rs +++ b/crates/storage/src/interface.rs @@ -2767,13 +2767,14 @@ impl Interface { // If this is a local shared action by a writer, set the nonce. Same // authority rule as save_raw, via the shared helper. Here `metadata` // was just loaded from the index above, so its writers already are the - // stored set — the helper's stored ∪ claimed union collapses to the - // stored membership check this delete requires. + // stored set — pass them as `stored` so the helper skips a redundant + // index read, and the stored ∪ claimed union collapses to the stored + // membership check this delete requires. let shared_to_stamp = if let StorageType::Shared { writers: claimed, .. } = &metadata.storage_type { - Self::authorize_local_shared_stamp(child_id, claimed)? + Self::authorize_local_shared_stamp(child_id, claimed, Some(claimed))? } else { None }; @@ -3545,18 +3546,29 @@ impl Interface { /// Both the save and delete paths route through here so the local-write /// authority rule lives in exactly one place and cannot drift between them; /// each caller keeps its own nonce and any schema re-stamp. + /// + /// `stored`: the caller's already-loaded stored writer set, when it has one. + /// The delete path loads `metadata` from the index immediately before + /// calling, so its writers ARE the stored set — it passes `Some(..)` to + /// avoid a redundant index read. The save path's `claimed` is the incoming + /// action's set (not stored), so it passes `None` and the stored set is + /// looked up here. fn authorize_local_shared_stamp( id: Id, claimed: &BTreeMap, + stored: Option<&BTreeMap>, ) -> Result, StorageError> { let executor: PublicKey = crate::env::executor_id().into(); - let stored_has_executor = >::get_metadata(id)? - .as_ref() - .map(|m| match &m.storage_type { - StorageType::Shared { writers, .. } => writers.contains_key(&executor), - _ => false, - }) - .unwrap_or(false); + let stored_has_executor = match stored { + Some(stored) => stored.contains_key(&executor), + None => >::get_metadata(id)? + .as_ref() + .map(|m| match &m.storage_type { + StorageType::Shared { writers, .. } => writers.contains_key(&executor), + _ => false, + }) + .unwrap_or(false), + }; let authorized = stored_has_executor || claimed.contains_key(&executor); Ok(authorized.then(|| (claimed.clone(), executor))) } @@ -3649,7 +3661,7 @@ impl Interface { .. } = &metadata.storage_type { - Self::authorize_local_shared_stamp(id, claimed_writers)? + Self::authorize_local_shared_stamp(id, claimed_writers, None)? } else { None }; From d5b012c0f7e9118ba458acb981513ebd8da72a1b Mon Sep 17 00:00:00 2001 From: Sandi Fatic Date: Wed, 8 Jul 2026 12:23:23 +0200 Subject: [PATCH 4/4] build: bump crossbeam-epoch 0.9.18 -> 0.9.20 (RUSTSEC-2026-0204) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cargo-deny advisories fails on RUSTSEC-2026-0204 (invalid pointer deref in crossbeam-epoch's fmt::Display for Atomic/Shared), pulled transitively via rayon -> wasmer. Lockfile-only patch bump to the fixed 0.9.20; no source or API change. Affects master too (Cargo.lock was identical) — unblocks CI. Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b686ad7e3c..c852e71951 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2486,9 +2486,9 @@ dependencies = [ [[package]] name = "crossbeam-epoch" -version = "0.9.18" +version = "0.9.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" dependencies = [ "crossbeam-utils", ]