From 9a2be4a3f39de981ce38d581e84c8edaada8b131 Mon Sep 17 00:00:00 2001 From: Tomer Weller Date: Thu, 27 Aug 2026 05:42:00 +0000 Subject: [PATCH] Attribute herder/overlay heap in the periodic memory report MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a Weak-reporter registry to LedgerManager so the every-64-ledger memory report folds in per-subsystem heap components from the herder (tx queue, SCP-slot retention, fetching envelopes, quorum + externalize -lag maps) and overlay (flood gate, peer maps, known/banned peer sets), shrinking the 97.5% unaccounted region the 42h soak documented. - henyey-common: object-safe MemoryReporter trait + btreemap/vecdeque heap helpers. - ledger: memory_reporters registry + register_memory_reporter (&self, Weak) + extracted collect_and_prune_reporters helper folded into build_memory_report after ledger-owned components (locks released first — no deadlock against reporter callbacks). - herder/overlay: O(1) estimate_heap_bytes on the owned collections and MemoryReporter impls; app wires both registrations so the impls have a production call path. Observability-only: read-only heap estimates and additive log fields; no observable-surface change. Refs #3845 Co-authored-by: Claude Code --- crates/app/src/app/lifecycle.rs | 12 +- crates/app/src/app/mod.rs | 10 ++ crates/common/src/memory.rs | 81 +++++++++++++ crates/herder/src/externalize_lag.rs | 43 +++++++ crates/herder/src/fetching_envelopes.rs | 49 ++++++++ crates/herder/src/herder.rs | 61 ++++++++++ crates/herder/src/quorum_tracker.rs | 74 ++++++++++++ crates/herder/src/scp_driver.rs | 8 ++ crates/herder/src/tx_queue/mod.rs | 80 +++++++++++++ crates/ledger/src/manager.rs | 145 ++++++++++++++++++++++++ crates/overlay/src/flood.rs | 41 +++++++ crates/overlay/src/manager/mod.rs | 107 +++++++++++++++++ 12 files changed, 710 insertions(+), 1 deletion(-) diff --git a/crates/app/src/app/lifecycle.rs b/crates/app/src/app/lifecycle.rs index 53890ec24..f9361d79e 100644 --- a/crates/app/src/app/lifecycle.rs +++ b/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); + + *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 { diff --git a/crates/app/src/app/mod.rs b/crates/app/src/app/mod.rs index 3c348e981..4c8647617 100644 --- a/crates/app/src/app/mod.rs +++ b/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` back-ref doesn't form a leak-cycle + // (#3845). + ledger_manager + .register_memory_reporter(Arc::downgrade(&herder) + as std::sync::Weak); + herder } diff --git a/crates/common/src/memory.rs b/crates/common/src/memory.rs index 8dbd2899b..2e1a5cc98 100644 --- a/crates/common/src/memory.rs +++ b/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` 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; +} + +/// 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` is a `BTreeMap`, 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::() + std::mem::size_of::(); + 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; 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)); + } } diff --git a/crates/herder/src/externalize_lag.rs b/crates/herder/src/externalize_lag.rs index 8023827b4..bae1b82c9 100644 --- a/crates/herder/src/externalize_lag.rs +++ b/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>. + let mut total = hashmap_heap_bytes( + self.node_lag.capacity(), + std::mem::size_of::(), + std::mem::size_of::>(), + ); + for samples in self.node_lag.values() { + total += vecdeque_heap_bytes(samples.capacity(), std::mem::size_of::()); + } + // first_externalize: HashMap. + total += hashmap_heap_bytes( + self.first_externalize.capacity(), + std::mem::size_of::(), + std::mem::size_of::(), + ); + 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" + ); + } } diff --git a/crates/herder/src/fetching_envelopes.rs b/crates/herder/src/fetching_envelopes.rs index 6199d82c0..ab640578d 100644 --- a/crates/herder/src/fetching_envelopes.rs +++ b/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. + let mut total = btreemap_heap_bytes( + slots.len(), + std::mem::size_of::(), + std::mem::size_of::(), + ); + // 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::()); + total += hashset_heap_bytes(slot.processed.capacity(), std::mem::size_of::()); + total += hashmap_heap_bytes( + slot.fetching.capacity(), + std::mem::size_of::(), + std::mem::size_of::(), + ); + total += vec_heap_bytes(slot.ready.capacity(), std::mem::size_of::()); + } + 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" + ); + } } diff --git a/crates/herder/src/herder.rs b/crates/herder/src/herder.rs index f7b76b2af..f80e1c56c 100644 --- a/crates/herder/src/herder.rs +++ b/crates/herder/src/herder.rs @@ -560,6 +560,48 @@ pub struct Herder { is_applying_flag: std::sync::OnceLock>, } +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 { + 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", + ] + ); + } } diff --git a/crates/herder/src/quorum_tracker.rs b/crates/herder/src/quorum_tracker.rs index d7d05cd6e..e5732bd2f 100644 --- a/crates/herder/src/quorum_tracker.rs +++ b/crates/herder/src/quorum_tracker.rs @@ -176,6 +176,23 @@ impl SlotQuorumTracker { self.slot_nodes.remove(&slot); } } + + /// Estimate the heap footprint of the per-slot node sets (#3845). + /// + /// O(number of tracked slots) — bounded by `max_slots` — reading only + /// map/set capacities. + pub fn estimate_heap_bytes(&self) -> usize { + use henyey_common::memory::{hashmap_heap_bytes, hashset_heap_bytes}; + let mut total = hashmap_heap_bytes( + self.slot_nodes.capacity(), + std::mem::size_of::(), + std::mem::size_of::>(), + ); + for nodes in self.slot_nodes.values() { + total += hashset_heap_bytes(nodes.capacity(), std::mem::size_of::()); + } + total + } } /// Metadata about a node in the transitive quorum graph. @@ -382,6 +399,29 @@ impl QuorumTracker { .get(node_id) .map(|info| &info.closest_validators) } + + /// Estimate the heap footprint of the transitive quorum map (#3845). + /// + /// O(number of tracked nodes) — reading only map/set capacities. The + /// per-node `Option` payloads are excluded (shared/cached + /// elsewhere); only the inline `NodeInfo` and each node's + /// `closest_validators` set are counted. + pub fn estimate_heap_bytes(&self) -> usize { + use henyey_common::memory::{btreemap_heap_bytes, hashmap_heap_bytes}; + let mut total = hashmap_heap_bytes( + self.quorum.capacity(), + std::mem::size_of::(), + std::mem::size_of::(), + ); + for info in self.quorum.values() { + total += btreemap_heap_bytes( + info.closest_validators.len(), + std::mem::size_of::(), + 0, + ); + } + total + } } fn for_each_quorum_node(quorum_set: &ScpQuorumSet, f: &mut F) @@ -653,4 +693,38 @@ mod tests { assert!(closest_d.contains(&node_b)); assert!(!closest_d.contains(&node_c)); } + + /// #3845: `SlotQuorumTracker::estimate_heap_bytes` is 0 when empty and grows + /// as slot/node entries are recorded. + #[test] + fn test_slot_quorum_tracker_estimate_heap_bytes_grows() { + let mut tracker = SlotQuorumTracker::new(None, 0); + assert_eq!(tracker.estimate_heap_bytes(), 0); + + tracker.record_envelope(100, make_node_id(2)); + let with_one = tracker.estimate_heap_bytes(); + assert!(with_one > 0); + + tracker.record_envelope(100, make_node_id(3)); + tracker.record_envelope(101, make_node_id(4)); + assert!(tracker.estimate_heap_bytes() > with_one); + } + + /// #3845: `QuorumTracker::estimate_heap_bytes` grows as the transitive + /// quorum map is expanded. + #[test] + fn test_quorum_tracker_estimate_heap_bytes_grows() { + let local = make_node_id(1); + let node_b = make_node_id(2); + let mut tracker = QuorumTracker::new(local.clone()); + let empty = tracker.estimate_heap_bytes(); + + tracker + .expand(&local, make_quorum_set(vec![node_b.clone()], 1)) + .expect("expand local"); + assert!( + tracker.estimate_heap_bytes() > empty, + "expanding the quorum map must increase the estimate" + ); + } } diff --git a/crates/herder/src/scp_driver.rs b/crates/herder/src/scp_driver.rs index 31f2bd648..6561d3015 100644 --- a/crates/herder/src/scp_driver.rs +++ b/crates/herder/src/scp_driver.rs @@ -1288,6 +1288,14 @@ impl ScpDriver { self.externalize_lag.read().get_lag_info_summary(&qset) } + /// Estimate the heap footprint of the externalize-lag tracker (#3845). + /// + /// Exposed so the herder's memory reporter can attribute the driver-owned + /// lag maps, which live outside the ledger manager's report call site. + pub fn estimate_externalize_lag_heap_bytes(&self) -> usize { + self.externalize_lag.read().estimate_heap_bytes() + } + /// Elapsed time since the first SCP activity was recorded for `slot`. /// Returns `None` if `record_slot_activity` was never called for this slot /// (e.g., catchup/fast-forward paths). diff --git a/crates/herder/src/tx_queue/mod.rs b/crates/herder/src/tx_queue/mod.rs index d7848f6e2..36e6b9e8e 100644 --- a/crates/herder/src/tx_queue/mod.rs +++ b/crates/herder/src/tx_queue/mod.rs @@ -2774,6 +2774,64 @@ impl TransactionQueue { self.store.read().is_empty() } + /// Estimate the heap footprint of this queue's owned collections (#3845). + /// + /// Reads capacities/lengths only — it never iterates entries (the + /// banned-transactions deque is bounded by the ban depth, so summing its + /// inner sets' capacities is a small constant). Each lock is taken and + /// released independently, so no two queue locks are held at once and the + /// documented `store → account_states → banned → seen` order cannot be + /// violated. The shared `Arc` payloads are excluded — + /// only the inline `QueuedTransaction` struct is counted — to avoid + /// double-counting. + pub fn estimate_heap_bytes(&self) -> usize { + use henyey_common::memory::{ + btreemap_heap_bytes, hashmap_heap_bytes, hashset_heap_bytes, vecdeque_heap_bytes, + }; + + // store: by_hash HashMap + fee_index BTreeSet (same-module field access). + let store_bytes = { + let store = self.store.read(); + hashmap_heap_bytes( + store.by_hash.capacity(), + std::mem::size_of::(), + std::mem::size_of::(), + ) + btreemap_heap_bytes(store.fee_index.len(), std::mem::size_of::(), 0) + }; + + // seen: HashSet. + let seen_bytes = { + let seen = self.seen.read(); + hashset_heap_bytes(seen.capacity(), std::mem::size_of::()) + }; + + // banned_transactions: VecDeque> (depth bounded by ban depth). + let banned_bytes = { + let banned = self.banned_transactions.read(); + let outer = + vecdeque_heap_bytes(banned.capacity(), std::mem::size_of::>()); + let inner: usize = banned + .iter() + .map(|s| hashset_heap_bytes(s.capacity(), std::mem::size_of::())) + .sum(); + outer + inner + }; + + // account_states: HashMap, AccountState> — add an estimate for + // the heap-allocated XDR-encoded AccountId keys. + let account_bytes = { + let states = self.account_states.read(); + const ACCOUNT_KEY_HEAP_BYTES: usize = 40; + hashmap_heap_bytes( + states.capacity(), + std::mem::size_of::>(), + std::mem::size_of::(), + ) + states.len() * ACCOUNT_KEY_HEAP_BYTES + }; + + store_bytes + seen_bytes + banned_bytes + account_bytes + } + /// Reset all lane-based and global eviction fee thresholds. /// /// Called whenever the queue is rebuilt or transactions are evicted/shifted @@ -11826,4 +11884,26 @@ mod broadcast_visitor_tests { // Age 5 should clamp into bucket [3] assert_eq!(stats.pending_txs_age, [0, 0, 0, 1]); } + + /// #3845: `estimate_heap_bytes` is ~0 on an empty queue and grows once + /// transactions (and bans) populate the owned collections. + #[test] + fn test_tx_queue_estimate_heap_bytes_grows() { + let queue = TransactionQueue::with_ban_depth(TxQueueConfig::default(), 3); + let empty = queue.estimate_heap_bytes(); + + let tx1 = make_test_envelope(200, 1); + assert_eq!(queue.try_add(tx1), TxQueueResult::Added); + let with_one = queue.estimate_heap_bytes(); + assert!( + with_one > empty, + "adding a transaction must increase the heap estimate ({with_one} !> {empty})" + ); + + // Banning populates the banned-transactions deque. + let mut tx2 = make_test_envelope(200, 1); + set_source(&mut tx2, 2); + queue.ban(&[Hash256::hash_xdr(&tx2)]); + assert!(queue.estimate_heap_bytes() >= with_one); + } } diff --git a/crates/ledger/src/manager.rs b/crates/ledger/src/manager.rs index 26ce8219e..6733f2d95 100644 --- a/crates/ledger/src/manager.rs +++ b/crates/ledger/src/manager.rs @@ -1641,6 +1641,19 @@ pub struct LedgerManager { /// Used to compute `stellar_ledger_age_closed_seconds` — the time elapsed /// between consecutive close_ledger calls, matching stellar-core's mLastClose. last_close_wall_time: Mutex, + + /// Registry of external subsystems that report per-component heap usage + /// into the periodic memory report (see #3845). + /// + /// Held as `Weak` because reporters (herder, overlay) themselves hold an + /// `Arc`; a strong back-ref would form a reference cycle and + /// leak — self-defeating for a memory-observability feature. Dead reporters + /// upgrade to `None` and are pruned in [`Self::build_memory_report`]. + /// + /// Interior-mutability (`&self`) registration via `register_memory_reporter` + /// mirrors the existing interpose-hook precedent, so subsystems can register + /// after the `LedgerManager` is behind an `Arc`. + memory_reporters: Mutex>>, } // Compile-time assertion: LedgerManager must be Send + Sync for spawn_blocking. @@ -1649,6 +1662,31 @@ const _: fn() = || { let _ = assert_send_sync:: as fn(); }; +/// Upgrade each registered weak memory reporter, collect its components, and +/// prune any that have been dropped (see #3845). +/// +/// Factored out as a free function so it can be unit-tested directly without a +/// heavy `LedgerManager` constructor. Only the registry lock is held, and it is +/// released *before* any `memory_components()` call — so a reporter is free to +/// touch the ledger from inside its report without a lock-order inversion, and +/// the prune never runs while a component guard is held. +fn collect_and_prune_reporters( + reporters: &Mutex>>, +) -> Vec { + // Snapshot the live reporters and drop dead weaks under the registry lock, + // then release it before invoking any reporter callback. + let live: Vec> = { + let mut guard = reporters.lock(); + guard.retain(|w| w.strong_count() > 0); + guard.iter().filter_map(|w| w.upgrade()).collect() + }; + let mut components = Vec::new(); + for reporter in live { + components.extend(reporter.memory_components()); + } + components +} + impl LedgerManager { /// Create a new ledger manager. /// @@ -1699,9 +1737,25 @@ impl LedgerManager { commit_publication_interpose: Mutex::new(None), invariant_manager: None, last_close_wall_time: Mutex::new(std::time::Instant::now()), + memory_reporters: Mutex::new(Vec::new()), } } + /// Register an external subsystem to contribute per-component heap + /// estimates to the periodic memory report (see #3845). + /// + /// Takes a `Weak` (see the `memory_reporters` field docs for the cycle + /// rationale) and `&self` so callers can register after the manager is + /// wrapped in an `Arc`. Registration order is not significant — the report + /// simply concatenates each live reporter's components after the + /// ledger-owned ones. + pub fn register_memory_reporter( + &self, + reporter: std::sync::Weak, + ) { + self.memory_reporters.lock().push(reporter); + } + /// Get the network ID. pub fn network_id(&self) -> &NetworkId { &self.network_id @@ -3415,6 +3469,12 @@ impl LedgerManager { } } + // External subsystem components (herder, overlay, …) registered via + // `register_memory_reporter`. Collected last, after all ledger-owned + // component guards above have been released, so no reporter callback can + // deadlock against a lock this method still holds (#3845). + components.extend(collect_and_prune_reporters(&self.memory_reporters)); + crate::memory_report::MemoryReport::new(ledger_seq, components) } @@ -12126,4 +12186,89 @@ mod tests { "V20 initial ledger-cost target size is 30 GB before V23" ); } + + // --- Memory reporter registry (#3845) --- + + use henyey_common::memory::{ComponentMemory, MemoryReporter}; + + /// A minimal reporter that yields one named component, for exercising the + /// weak-upgrade / merge / prune path without a heavy subsystem. + struct FakeReporter { + name: &'static str, + bytes: u64, + } + + impl MemoryReporter for FakeReporter { + fn memory_components(&self) -> Vec { + vec![ComponentMemory::new(self.name, self.bytes, 1)] + } + } + + #[test] + fn test_register_memory_reporter_appends_components() { + let reporters: Mutex>> = Mutex::new(Vec::new()); + let a: Arc = Arc::new(FakeReporter { + name: "fake_a", + bytes: 100, + }); + let b: Arc = Arc::new(FakeReporter { + name: "fake_b", + bytes: 200, + }); + reporters.lock().push(Arc::downgrade(&a)); + reporters.lock().push(Arc::downgrade(&b)); + + let components = collect_and_prune_reporters(&reporters); + let names: Vec<_> = components.iter().map(|c| c.name).collect(); + assert!(names.contains(&"fake_a")); + assert!(names.contains(&"fake_b")); + assert_eq!(components.iter().map(|c| c.bytes).sum::(), 300); + // Both weaks are still live, so nothing was pruned. + assert_eq!(reporters.lock().len(), 2); + } + + #[test] + fn test_dropped_memory_reporter_is_skipped() { + let reporters: Mutex>> = Mutex::new(Vec::new()); + let live: Arc = Arc::new(FakeReporter { + name: "live", + bytes: 42, + }); + reporters.lock().push(Arc::downgrade(&live)); + { + // This reporter is dropped at the end of the block, so its weak + // must upgrade to None and be pruned on the next collection. + let dead: Arc = Arc::new(FakeReporter { + name: "dead", + bytes: 999, + }); + reporters.lock().push(Arc::downgrade(&dead)); + assert_eq!(reporters.lock().len(), 2); + } + + let components = collect_and_prune_reporters(&reporters); + let names: Vec<_> = components.iter().map(|c| c.name).collect(); + assert_eq!(names, vec!["live"]); + // The dead weak was pruned from the registry. + assert_eq!(reporters.lock().len(), 1); + } + + #[test] + fn test_register_memory_reporter_via_manager() { + let lm = LedgerManager::new( + "Test SDF Network ; September 2015".to_string(), + LedgerManagerConfig::default(), + ); + let reporter: Arc = Arc::new(FakeReporter { + name: "registered", + bytes: 7, + }); + lm.register_memory_reporter(Arc::downgrade(&reporter)); + + let report = lm.build_memory_report(1); + assert!( + report.components.iter().any(|c| c.name == "registered"), + "registered reporter's component must appear in the memory report" + ); + } } diff --git a/crates/overlay/src/flood.rs b/crates/overlay/src/flood.rs index c0b75adf8..69300291d 100644 --- a/crates/overlay/src/flood.rs +++ b/crates/overlay/src/flood.rs @@ -404,6 +404,25 @@ impl FloodGate { } } + /// Estimate the heap footprint of the seen-message map (#3845). + /// + /// O(1): reads the entry count only. The per-entry `peers` sets are *not* + /// walked — that would be O(n) over up to `max_entries` (~1M) entries on + /// the ledger-close report path — so only the flat `IndexMap` payload (the + /// dominant term) is counted. Returns `(bytes, entry_count)`. + pub fn estimate_heap_bytes(&self) -> (u64, u64) { + use henyey_common::memory::hashmap_heap_bytes; + let len = self.map.lock().entries.len(); + // IndexMap ≈ a Vec of (K, V) entries plus a hash index table; the flat + // hashmap helper over the entry count approximates both. + let bytes = hashmap_heap_bytes( + len, + std::mem::size_of::(), + std::mem::size_of::(), + ); + (bytes as u64, len as u64) + } + /// Removes flood records from ledgers before `ledger_seq`. /// /// Matches upstream stellar-core's `clearBelow(maxLedger)` which removes @@ -1357,4 +1376,26 @@ mod tests { assert_eq!(keys[2], hc); assert_eq!(keys[3], hd); } + + /// #3845: `estimate_heap_bytes` is 0 on an empty flood gate and grows + /// monotonically as messages are recorded. + #[test] + fn test_flood_gate_estimate_heap_bytes_monotonic() { + let gate = FloodGate::new(); + let (bytes0, count0) = gate.estimate_heap_bytes(); + assert_eq!((bytes0, count0), (0, 0), "empty flood gate costs 0"); + + gate.record_local_broadcast(make_hash(1), 1); + let (bytes1, count1) = gate.estimate_heap_bytes(); + assert!(bytes1 > 0); + assert_eq!(count1, 1); + + gate.record_local_broadcast(make_hash(2), 1); + let (bytes2, count2) = gate.estimate_heap_bytes(); + assert!( + bytes2 > bytes1, + "a second message must increase the estimate" + ); + assert_eq!(count2, 2); + } } diff --git a/crates/overlay/src/manager/mod.rs b/crates/overlay/src/manager/mod.rs index 8ce4ef98d..4cc2a8322 100644 --- a/crates/overlay/src/manager/mod.rs +++ b/crates/overlay/src/manager/mod.rs @@ -341,6 +341,28 @@ impl KnownPeerSet { } } + /// Estimate the heap footprint of the known-peer collections (#3845). + /// + /// O(1): reads Vec/HashSet capacities only. The `String` hostnames inside + /// each `PeerAddress` are not walked; only the inline element footprints are + /// counted (conservative under-count of a small, bounded structure). + pub(super) fn estimate_heap_bytes(&self) -> usize { + use henyey_common::memory::{hashset_heap_bytes, vec_heap_bytes}; + vec_heap_bytes( + self.config_entries.capacity(), + std::mem::size_of::(), + ) + vec_heap_bytes( + self.resolved.capacity(), + std::mem::size_of::>(), + ) + vec_heap_bytes( + self.discovered.capacity(), + std::mem::size_of::(), + ) + hashset_heap_bytes( + self.discovered_keys.capacity(), + std::mem::size_of::(), + ) + } + /// Apply DNS resolution results. `results` must be positionally aligned /// with `config_entries`. On `Some(addr)`: updates resolution. On `None`: /// preserves last-good (does NOT clear). @@ -1361,6 +1383,67 @@ pub struct OverlayManager { broadcast_backpressure_warn_last_ms: AtomicU64, } +impl henyey_common::memory::MemoryReporter for OverlayManager { + /// Report the overlay-owned heap components that live outside the ledger + /// manager's own report call site (#3845): the flood gate's seen-message + /// map, the connected-peer maps, and the known/banned peer sets. Each + /// estimate is O(1) and takes only short-lived locks — none re-enters the + /// ledger — so it is safe to call from the ledger-close report path. + /// + /// `BanManager`/`PeerManager` DB caches are intentionally excluded (they are + /// not `OverlayManager` fields); a follow-up can add them if they prove + /// material. `DashMap` capacity is not exposed, so `len()` is used as a + /// conservative floor for the sharded maps. + fn memory_components(&self) -> Vec { + use henyey_common::memory::{hashmap_heap_bytes, hashset_heap_bytes, ComponentMemory}; + + let (flood_bytes, flood_count) = self.flood_gate.estimate_heap_bytes(); + + let peers_len = self.peers.len(); + let info_len = self.peer_info_cache.len(); + let ext_len = self.peer_latest_externalized.len(); + + let banned_bytes = { + let banned = self.banned_peers.read(); + hashset_heap_bytes(banned.capacity(), std::mem::size_of::()) + }; + let known_bytes = self.known_peers.read().estimate_heap_bytes(); + + vec![ + ComponentMemory::new("overlay_flood", flood_bytes, flood_count), + ComponentMemory::new( + "overlay_peers", + hashmap_heap_bytes( + peers_len, + std::mem::size_of::(), + std::mem::size_of::(), + ) as u64, + peers_len as u64, + ), + ComponentMemory::new( + "overlay_peer_info_cache", + hashmap_heap_bytes( + info_len, + std::mem::size_of::(), + std::mem::size_of::(), + ) as u64, + info_len as u64, + ), + ComponentMemory::new( + "overlay_peer_externalized", + hashmap_heap_bytes( + ext_len, + std::mem::size_of::(), + std::mem::size_of::(), + ) as u64, + ext_len as u64, + ), + ComponentMemory::new("overlay_banned_peers", banned_bytes as u64, 0), + ComponentMemory::new("overlay_known_peers", known_bytes as u64, 0), + ] + } +} + impl OverlayManager { /// Create a new overlay manager with the given configuration. pub fn new(config: OverlayConfig, local_node: LocalNode) -> Result { @@ -5989,4 +6072,28 @@ pub(crate) mod tests { "send_error_and_drop should report false for an unknown peer" ); } + + /// #3845: the `MemoryReporter` impl exposes exactly the six named overlay + /// components, so the periodic memory report can attribute them. + #[test] + fn test_overlay_memory_components_names() { + use henyey_common::memory::MemoryReporter; + let config = OverlayConfig::testnet(); + let secret = SecretKey::generate(); + let local_node = LocalNode::new_testnet(secret); + let manager = OverlayManager::new(config, local_node).unwrap(); + + let names: Vec<&str> = manager.memory_components().iter().map(|c| c.name).collect(); + assert_eq!( + names, + vec![ + "overlay_flood", + "overlay_peers", + "overlay_peer_info_cache", + "overlay_peer_externalized", + "overlay_banned_peers", + "overlay_known_peers", + ] + ); + } }