Attribute herder/overlay heap in the periodic memory report

crates/app/src/app/lifecycle.rs

@@ -1968,7 +1968,17 @@ impl App {
         let peer_count = overlay.peer_count();
         tracing::info!(peer_count, "Overlay network started");
 
-        *self.overlay.write().await = Some(Arc::new(overlay));
+        let overlay = Arc::new(overlay);
+
+        // Register the overlay as a memory reporter so the periodic memory report
+        // folds in its per-subsystem heap components (flood gate, peer maps,
+        // known/banned peer sets). Weak so it does not extend the overlay's
+        // lifetime (#3845).
+        self.ledger_manager
+            .register_memory_reporter(Arc::downgrade(&overlay)
+                as std::sync::Weak<dyn henyey_common::memory::MemoryReporter>);
+
+        *self.overlay.write().await = Some(overlay);
 
         // Grab the tracking flag handle so synchronous callbacks can update it.
         if let Some(ref om) = *self.overlay.read().await {

crates/app/src/app/mod.rs

@@ -2102,6 +2102,16 @@ impl App {
                 tracing::warn!(error = %err, "Failed to store local quorum set");
             }
         }
+
+        // Register the herder as a memory reporter so the periodic memory report
+        // folds in its per-subsystem heap components (tx queue, SCP-slot
+        // retention, fetching envelopes, quorum + externalize-lag maps). Weak so
+        // the herder's `Arc<LedgerManager>` back-ref doesn't form a leak-cycle
+        // (#3845).
+        ledger_manager
+            .register_memory_reporter(Arc::downgrade(&herder)
+                as std::sync::Weak<dyn henyey_common::memory::MemoryReporter>);
+
         herder
     }
 

crates/common/src/memory.rs

@@ -47,6 +47,58 @@ impl ComponentMemory {
     }
 }
 
+/// A subsystem that can report its heap footprint as named components.
+///
+/// Implemented by long-lived subsystems (herder, overlay, …) whose owned
+/// allocations live outside the ledger manager's own report call site and so
+/// cannot be attributed by [`crate::memory`] helpers alone. The ledger
+/// manager holds a registry of `Weak<dyn MemoryReporter>` and folds each
+/// reporter's components into the periodic memory report (see #3845).
+///
+/// Implementations MUST follow the same discipline as the built-in
+/// components: `memory_components()` is O(1) per component (read
+/// capacities/lengths/counters, never iterate entries), conservative
+/// (slight over-count acceptable), and excludes `Arc`-shared ledger state to
+/// avoid double-counting. It MUST NOT acquire any lock already held by the
+/// ledger-close path that invokes it.
+///
+/// Object-safe: used only through `dyn MemoryReporter`.
+pub trait MemoryReporter: Send + Sync {
+    /// Return this subsystem's per-component heap estimates.
+    fn memory_components(&self) -> Vec<ComponentMemory>;
+}
+
+/// Estimate heap bytes for a `BTreeMap` with `len` entries.
+///
+/// `BTreeMap` stores entries in B-tree nodes (Rust's `B = 6`, so each node
+/// holds up to `2*B - 1 = 11` key/value pairs). We approximate the footprint
+/// as the key/value payload plus per-node bookkeeping (child pointers + a
+/// length field). The result is monotonic in `len` and conservative.
+///
+/// `BTreeSet<K>` is a `BTreeMap<K, ()>`, so pass `value_size = 0`.
+pub fn btreemap_heap_bytes(len: usize, key_size: usize, value_size: usize) -> usize {
+    if len == 0 {
+        return 0;
+    }
+    // 2*B - 1 with B = 6 (libstd's BTree branching factor).
+    const NODE_CAPACITY: usize = 11;
+    let nodes = len.div_ceil(NODE_CAPACITY);
+    let payload = len * (key_size + value_size);
+    // Over-count with the internal-node layout for every node (each holds up
+    // to NODE_CAPACITY+1 child pointers plus a length field) — conservative.
+    let per_node_overhead =
+        (NODE_CAPACITY + 1) * std::mem::size_of::<usize>() + std::mem::size_of::<u16>();
+    payload + nodes * per_node_overhead
+}
+
+/// Estimate heap bytes for a `VecDeque` with the given capacity and element size.
+///
+/// `VecDeque` backs its ring buffer with a single contiguous allocation of
+/// `capacity` elements, so the footprint is `capacity * element_size`.
+pub fn vecdeque_heap_bytes(capacity: usize, element_size: usize) -> usize {
+    capacity * element_size
+}
+
 /// Estimate heap bytes for a `HashMap` with the given capacity and entry sizes.
 ///
 /// Accounts for hashbrown's internal layout: each entry stores the key and
@@ -114,4 +166,33 @@ mod tests {
         assert_eq!(cm.name, "test");
         assert!((cm.heap_mb() - 1.0).abs() < 0.001);
     }
+
+    #[test]
+    fn test_btreemap_heap_bytes_zero() {
+        assert_eq!(btreemap_heap_bytes(0, 32, 64), 0);
+    }
+
+    #[test]
+    fn test_btreemap_heap_bytes_monotonic() {
+        // Empty is zero; any entry is strictly positive; more entries is more.
+        let one = btreemap_heap_bytes(1, 32, 64);
+        let many = btreemap_heap_bytes(100, 32, 64);
+        assert!(one > 0);
+        assert!(many > one);
+        // Larger key/value sizes cost strictly more for the same len.
+        assert!(btreemap_heap_bytes(100, 64, 128) > btreemap_heap_bytes(100, 32, 64));
+    }
+
+    #[test]
+    fn test_btreemap_heap_bytes_set_like() {
+        // A BTreeSet is a BTreeMap<K, ()>; value_size 0 still counts the keys.
+        assert!(btreemap_heap_bytes(50, 8, 0) > 0);
+    }
+
+    #[test]
+    fn test_vecdeque_heap_bytes() {
+        assert_eq!(vecdeque_heap_bytes(0, 16), 0);
+        assert_eq!(vecdeque_heap_bytes(64, 16), 1024);
+        assert!(vecdeque_heap_bytes(128, 16) > vecdeque_heap_bytes(64, 16));
+    }
 }

crates/herder/src/externalize_lag.rs

@@ -52,6 +52,30 @@ impl ExternalizeLagTracker {
         }
     }
 
+    /// Estimate the heap footprint of the lag-tracking maps (#3845).
+    ///
+    /// O(number of tracked nodes) — reading only map/deque capacities. Each
+    /// per-node deque is capped at `MAX_LAG_SAMPLES`.
+    pub fn estimate_heap_bytes(&self) -> usize {
+        use henyey_common::memory::{hashmap_heap_bytes, vecdeque_heap_bytes};
+        // node_lag: HashMap<NodeId, VecDeque<Duration>>.
+        let mut total = hashmap_heap_bytes(
+            self.node_lag.capacity(),
+            std::mem::size_of::<NodeId>(),
+            std::mem::size_of::<VecDeque<Duration>>(),
+        );
+        for samples in self.node_lag.values() {
+            total += vecdeque_heap_bytes(samples.capacity(), std::mem::size_of::<Duration>());
+        }
+        // first_externalize: HashMap<SlotIndex, Instant>.
+        total += hashmap_heap_bytes(
+            self.first_externalize.capacity(),
+            std::mem::size_of::<SlotIndex>(),
+            std::mem::size_of::<Instant>(),
+        );
+        total
+    }
+
     /// Record an externalize event for a slot.
     ///
     /// - On the first call per slot, sets `first_externalize[slot] = now`.
@@ -440,4 +464,23 @@ mod tests {
         let json = serde_json::to_value(&without_lag).unwrap();
         assert!(json["lag_ms"].is_null());
     }
+
+    /// #3845: `estimate_heap_bytes` is 0 for a fresh tracker and grows as slots
+    /// and per-node lag samples are recorded.
+    #[test]
+    fn test_externalize_lag_estimate_heap_bytes_grows() {
+        let mut tracker = ExternalizeLagTracker::new();
+        assert_eq!(tracker.estimate_heap_bytes(), 0);
+
+        let self_node = make_node(1);
+        let peer = make_node(2);
+        let t0 = Instant::now();
+        tracker.record_event(100, &self_node, true, t0);
+        tracker.record_event(100, &peer, false, t0 + Duration::from_millis(50));
+        let after = tracker.estimate_heap_bytes();
+        assert!(
+            after > 0,
+            "recording a slot + peer lag sample must cost more than 0"
+        );
+    }
 }

crates/herder/src/fetching_envelopes.rs

@@ -744,6 +744,37 @@ impl FetchingEnvelopes {
         self.stats.read().clone()
     }
 
+    /// Estimate the heap footprint of the per-slot envelope buffers (#3845).
+    ///
+    /// O(number of buffered slots) — bounded by `max_future_slots` — reading
+    /// only map/set/vec capacities. Envelope payloads themselves are carried by
+    /// the SCP layer and are not double-counted here; only the inline
+    /// `ScpEnvelope`/`FetchingEntry` struct footprints are included.
+    pub fn estimate_heap_bytes(&self) -> usize {
+        use henyey_common::memory::{
+            btreemap_heap_bytes, hashmap_heap_bytes, hashset_heap_bytes, vec_heap_bytes,
+        };
+        let slots = self.slots.read();
+        // Outer BTreeMap<SlotIndex, SlotEnvelopes>.
+        let mut total = btreemap_heap_bytes(
+            slots.len(),
+            std::mem::size_of::<SlotIndex>(),
+            std::mem::size_of::<SlotEnvelopes>(),
+        );
+        // Per-slot inner collections (same-module private field access).
+        for slot in slots.values() {
+            total += hashset_heap_bytes(slot.discarded.capacity(), std::mem::size_of::<Hash256>());
+            total += hashset_heap_bytes(slot.processed.capacity(), std::mem::size_of::<Hash256>());
+            total += hashmap_heap_bytes(
+                slot.fetching.capacity(),
+                std::mem::size_of::<Hash256>(),
+                std::mem::size_of::<FetchingEntry>(),
+            );
+            total += vec_heap_bytes(slot.ready.capacity(), std::mem::size_of::<ScpEnvelope>());
+        }
+        total
+    }
+
     /// Trim stale data while preserving state for slots after catchup.
     /// Called after catchup to release memory from stale data.
     pub fn trim_stale(&self, keep_after_slot: SlotIndex) {
@@ -3076,4 +3107,22 @@ mod tests {
             "duplicate dep arrival must not re-trigger broadcast"
         );
     }
+
+    /// #3845: `estimate_heap_bytes` is 0 with no buffered slots and grows once
+    /// slots hold envelopes.
+    #[test]
+    fn test_fetching_envelopes_estimate_heap_bytes_grows() {
+        let fetching = FetchingEnvelopes::with_defaults(Box::new(|_, _| false));
+        assert_eq!(fetching.estimate_heap_bytes(), 0, "empty buffers cost 0");
+
+        fetching.test_insert_ready(100, vec![make_envelope(100, 1), make_envelope(100, 2)]);
+        let with_slot = fetching.estimate_heap_bytes();
+        assert!(with_slot > 0, "a buffered slot must cost more than 0");
+
+        fetching.test_insert_ready(101, vec![make_envelope(101, 3)]);
+        assert!(
+            fetching.estimate_heap_bytes() > with_slot,
+            "a second buffered slot must increase the estimate"
+        );
+    }
 }

crates/herder/src/herder.rs

@@ -560,6 +560,48 @@ pub struct Herder {
     is_applying_flag: std::sync::OnceLock<Arc<std::sync::atomic::AtomicBool>>,
 }
 
+impl henyey_common::memory::MemoryReporter for Herder {
+    /// Report the herder-owned heap components that live outside the ledger
+    /// manager's own report call site (#3845): the transaction queue, the
+    /// fetching-envelope buffers, the SCP slot/quorum trackers, and the
+    /// externalize-lag maps. Each estimate is O(1)/O(bounded) and takes only
+    /// short-lived read locks — none re-enters the ledger — so it is safe to
+    /// call from the ledger-close report path.
+    fn memory_components(&self) -> Vec<henyey_common::memory::ComponentMemory> {
+        use henyey_common::memory::ComponentMemory;
+        vec![
+            ComponentMemory::new(
+                "herder_tx_queue",
+                self.tx_queue.estimate_heap_bytes() as u64,
+                self.tx_queue.len() as u64,
+            ),
+            ComponentMemory::new(
+                "herder_fetching_envelopes",
+                self.fetching_envelopes.estimate_heap_bytes() as u64,
+                0,
+            ),
+            ComponentMemory::new(
+                "herder_scp_slots",
+                self.slot_quorum_tracker.read().estimate_heap_bytes() as u64,
+                0,
+            ),
+            {
+                let quorum = self.quorum_tracker.read();
+                ComponentMemory::new(
+                    "herder_quorum",
+                    quorum.estimate_heap_bytes() as u64,
+                    quorum.tracked_node_count() as u64,
+                )
+            },
+            ComponentMemory::new(
+                "herder_externalize_lag",
+                self.scp_driver.estimate_externalize_lag_heap_bytes() as u64,
+                0,
+            ),
+        ]
+    }
+}
+
 impl Herder {
     /// Create a new Herder (observer mode, no secret key).
     pub fn new(
@@ -16511,4 +16553,23 @@ mod fetching_envelopes_routing_tests {
             "slot 100 should be purged from SCP (below purge boundary, not checkpoint)"
         );
     }
+
+    /// #3845: the `MemoryReporter` impl exposes exactly the five named herder
+    /// components, so the periodic memory report can attribute them.
+    #[test]
+    fn test_herder_memory_components_names() {
+        use henyey_common::memory::MemoryReporter;
+        let herder = make_test_herder();
+        let names: Vec<&str> = herder.memory_components().iter().map(|c| c.name).collect();
+        assert_eq!(
+            names,
+            vec![
+                "herder_tx_queue",
+                "herder_fetching_envelopes",
+                "herder_scp_slots",
+                "herder_quorum",
+                "herder_externalize_lag",
+            ]
+        );
+    }
 }