fix(storage,node): path-ancestor & shared-stamp fixes + node hardening batch

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

crates/node/src/handlers/network_event/blobs.rs

@@ -55,14 +55,27 @@ 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,
                         size = size,
                         "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,

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,