From 9acfee3ddcd7b5453a843fdc66543070c71c702c Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Wed, 5 Aug 2026 18:36:20 -0400 Subject: [PATCH] fix(dash-spv): release clean storage segments below the committed height during long scans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A long backfill pinned every item it ever downloaded. `SegmentCache` holds 50_000 items per segment and only evicts once more than `MAX_ACTIVE_SEGMENTS` (10) are resident, so a 350k-block scan — which spans just 7-8 segments — never tripped the LRU. Every compact filter and every full decoded block (GCS false positives, ~12k blocks at a 27k-script watch set) stayed in RAM until the process died. `persist` wrote segments to disk and marked them Clean but never freed them. Add a caller-declared committed-height watermark. After `persist`, any segment lying entirely at or below it is dropped from memory, leaving the on-disk file as the source of truth; the next read reloads it through the lazy path already used for any non-resident segment. This is safe because `Clean` is reachable only via `Segment::load` from an existing file or a successful `Segment::persist`, so a clean segment is byte-identical to its backing file and releasing it is invisible to readers. Dirty segments (contents exist only in memory), the frontier segment (still being written), and partially committed segments are always kept. The watermark is wired where each cache's owner already knows the answer: - blocks, from the in-order `take_next_ordered_block` drain, which applies blocks to every interested wallet before advancing; - filters, from batch commit, which happens only once every matched block in the batch has been downloaded and applied. It may move backwards — a wallet rescan rolls it back before re-reading lower heights — which only narrows the release window. `truncate_above` clamps it and `clear` drops it so it never outlives its data. Header caches are deliberately left alone: their random-access path is the separate `header_hash_index` map, which segment release cannot shrink, and their ranges are read at arbitrary depths during scanning. The mechanism is generic, so they can opt in later with one call. New trait methods carry default no-op bodies, so out-of-tree implementors are unaffected and the change is drop-in for the platform AAR build. Co-Authored-By: Claude Fable 5 --- dash-spv/src/storage/blocks.rs | 58 +++++ dash-spv/src/storage/filters.rs | 15 ++ dash-spv/src/storage/mod.rs | 8 + dash-spv/src/storage/segments.rs | 337 +++++++++++++++++++++++++++ dash-spv/src/sync/blocks/manager.rs | 18 ++ dash-spv/src/sync/filters/manager.rs | 12 + 6 files changed, 448 insertions(+) diff --git a/dash-spv/src/storage/blocks.rs b/dash-spv/src/storage/blocks.rs index 27324d5c9..34b90a76f 100644 --- a/dash-spv/src/storage/blocks.rs +++ b/dash-spv/src/storage/blocks.rs @@ -33,6 +33,17 @@ pub trait BlockStorage: Send + Sync + 'static { /// A crash between `truncate_above` and `persist` may leave orphaned segment /// files on disk and cause the storage to reopen at the pre-truncation tip. async fn truncate_above(&mut self, target_height: CoreBlockHeight) -> StorageResult<()>; + + /// Declare the highest block height that has been applied to every + /// interested wallet, allowing the storage to stop holding those block + /// bodies in memory. + /// + /// Blocks at or below this height stay readable: `load_block` reloads them + /// from disk on demand. The watermark may move backwards when a rescan + /// re-processes lower heights. + /// + /// Defaults to a no-op for implementations that hold no in-memory cache. + async fn set_committed_height(&mut self, _height: CoreBlockHeight) {} } /// Persistent storage for full blocks using segmented files. @@ -81,6 +92,10 @@ impl BlockStorage for PersistentBlockStorage { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.blocks.write().await.truncate_above(target_height).await } + + async fn set_committed_height(&mut self, height: u32) { + self.blocks.write().await.set_committed_height(height); + } } #[cfg(test)] @@ -143,6 +158,49 @@ mod tests { assert_eq!(storage.load_block(3).await.unwrap(), None); } + /// Block bodies are the dominant memory consumer during a long backfill. + /// Once applied, they must leave memory while staying loadable from disk — + /// `handle_sync_event` re-reads stored blocks by height on resume/rescan. + #[tokio::test] + async fn test_committed_blocks_are_released_but_still_loadable() { + let temp_dir = TempDir::new().unwrap(); + let mut storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap(); + + // One block in each of segments 0, 1 and 2 (50_000 heights per segment). + // Carry real transaction payloads so the reload proves the block bodies + // round-trip, not merely that a slot is occupied. + let txs = vec![dashcore::Transaction::dummy_empty()]; + let low = HashedBlock::dummy(10, txs.clone()); + let mid = HashedBlock::dummy(50_010, txs.clone()); + let tip = HashedBlock::dummy(100_010, txs); + + storage.store_block(10, low.clone()).await.unwrap(); + storage.store_block(50_010, mid.clone()).await.unwrap(); + storage.store_block(100_010, tip.clone()).await.unwrap(); + storage.persist(temp_dir.path()).await.unwrap(); + + assert_eq!(storage.blocks.read().await.resident_segment_ids(), vec![0, 1, 2]); + + // Everything below segment 2 has been applied to the wallets. + storage.set_committed_height(99_999).await; + storage.persist(temp_dir.path()).await.unwrap(); + + // Segments 0 and 1 are gone from memory; the frontier segment stays. + assert_eq!( + storage.blocks.read().await.resident_segment_ids(), + vec![2], + "applied block segments must be released" + ); + + // All three blocks still load, byte-identically, via the disk fallback. + assert_eq!(storage.load_block(10).await.unwrap(), Some(low)); + assert_eq!(storage.load_block(50_010).await.unwrap(), Some(mid)); + assert_eq!(storage.load_block(100_010).await.unwrap(), Some(tip)); + + // Gaps inside a released segment still report absent, not sentinel data. + assert_eq!(storage.load_block(11).await.unwrap(), None); + } + #[tokio::test] async fn test_returns_none_for_gaps() { let temp_dir = TempDir::new().unwrap(); diff --git a/dash-spv/src/storage/filters.rs b/dash-spv/src/storage/filters.rs index 3578a26ea..249b966b9 100644 --- a/dash-spv/src/storage/filters.rs +++ b/dash-spv/src/storage/filters.rs @@ -51,6 +51,17 @@ pub trait FilterStorage: Send + Sync + 'static { /// A crash between `truncate_above` and `persist` may leave orphaned segment /// files on disk and cause the storage to reopen at the pre-truncation tip. async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()>; + + /// Declare the highest filter height that has been scanned and committed + /// for every wallet, allowing the storage to stop holding those filters in + /// memory. + /// + /// Filters at or below this height stay readable: `load_filters` reloads + /// them from disk on demand. The watermark may move backwards when a + /// rescan rolls the scan position back. + /// + /// Defaults to a no-op for implementations that hold no in-memory cache. + async fn set_committed_height(&mut self, _height: u32) {} } pub struct PersistentFilterStorage { @@ -110,6 +121,10 @@ impl FilterStorage for PersistentFilterStorage { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.filters.write().await.truncate_above(target_height).await } + + async fn set_committed_height(&mut self, height: u32) { + self.filters.write().await.set_committed_height(height); + } } #[cfg(test)] diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index 248ab10d0..70a851acb 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -386,6 +386,10 @@ impl filters::FilterStorage for DiskStorageManager { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.filters.write().await.truncate_above(target_height).await } + + async fn set_committed_height(&mut self, height: u32) { + self.filters.write().await.set_committed_height(height).await; + } } #[async_trait] @@ -401,6 +405,10 @@ impl BlockStorage for DiskStorageManager { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.blocks.write().await.truncate_above(target_height).await } + + async fn set_committed_height(&mut self, height: u32) { + self.blocks.write().await.set_committed_height(height).await; + } } #[async_trait] diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index f76566aa0..ec900899c 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -83,6 +83,14 @@ pub struct SegmentCache { /// Segment ids whose backing files must be removed on the next `persist`. /// Populated by `truncate_above` for segments that are dropped entirely. to_delete: HashSet, + /// Highest height whose items are durably persisted *and* fully consumed + /// by every reader, as declared by the owning sync manager through + /// [`SegmentCache::set_committed_height`]. + /// + /// `None` — the default — disables committed-height release entirely, so a + /// cache whose owner never declares a watermark keeps exactly the residency + /// behavior it had before this field existed. + committed_height: Option, } impl SegmentCache { @@ -98,6 +106,7 @@ impl SegmentCache { start_height: None, segments_dir: segments_dir.clone(), to_delete: HashSet::new(), + committed_height: None, }; // Building the metadata @@ -445,6 +454,15 @@ impl SegmentCache { self.tip_height = Some(target_height); + // The watermark must never claim heights that no longer exist: a + // rescan re-reads this range, and a stale high watermark would keep + // releasing segments it is actively refilling. + if let Some(committed) = self.committed_height { + if committed > target_height { + self.committed_height = Some(target_height); + } + } + Ok(()) } @@ -493,6 +511,8 @@ impl SegmentCache { self.evicted.clear(); self.tip_height = None; self.start_height = None; + // Nothing is stored, so nothing is committed. + self.committed_height = None; Ok(()) } @@ -527,6 +547,123 @@ impl SegmentCache { tracing::error!("Failed to persist segment with id {id}: {e}"); } } + + // After the writes above, so segments cleaned by *this* pass are + // eligible immediately. A segment whose persist failed is still Dirty + // and is therefore skipped. + let released = self.release_committed_segments(); + if released > 0 { + tracing::debug!( + "SegmentCache: released {} committed segment(s) below height {:?}; {} resident", + released, + self.committed_height, + self.segments.len(), + ); + } + } + + /// Declare the highest height whose items are durably persisted and fully + /// consumed, so the cache may stop holding them in memory. + /// + /// Segments lying *entirely* at or below `height` are released from memory + /// by the next [`SegmentCache::persist`], once they are `Clean`. The + /// on-disk file remains the source of truth and a later read of a released + /// height transparently reloads the segment through the same lazy path used + /// for any never-resident segment — the data a reader observes is + /// unchanged, only the resident-set size is. + /// + /// The watermark may move **backwards**: a wallet rescan rolls it back + /// before re-reading lower heights. A regression only ever narrows the + /// release window, so it is always safe. Callers therefore assign rather + /// than accumulate, and no monotonicity is enforced here. + pub fn set_committed_height(&mut self, height: u32) { + self.committed_height = Some(height); + } + + /// The committed-height watermark, or `None` when the owner has never + /// declared one (in which case no committed-height release occurs). + /// + /// Test-only for now — nothing reads the watermark back in production. + /// Promote it to an unconditional accessor when a caller needs it. + #[cfg(test)] + #[inline] + pub fn committed_height(&self) -> Option { + self.committed_height + } + + /// Sorted ids of the segments currently held in memory. Test-only hook for + /// asserting residency from the storage wrappers. + #[cfg(test)] + pub(crate) fn resident_segment_ids(&self) -> Vec { + let mut ids: Vec = self.segments.keys().copied().collect(); + ids.sort_unstable(); + ids + } + + /// Release the in-memory items of every fully-committed clean segment, + /// keeping the on-disk file as the source of truth. + /// + /// This is what stops a long backfill from pinning every segment it ever + /// touched. `MAX_ACTIVE_SEGMENTS` alone cannot: a scan spanning fewer than + /// ten segments never trips the LRU, so every downloaded item stays + /// resident until the process dies. + /// + /// Releasing a `Clean` segment is invisible to readers. `Clean` is reachable + /// only via [`Segment::load`] from an existing file or a successful + /// [`Segment::persist`], so a clean segment's items are byte-identical to + /// its backing file, and `get_segment_mut` reloads exactly those bytes on + /// the next access. + /// + /// Three classes are deliberately kept: + /// - `Dirty` segments — their contents exist *only* in memory, so dropping + /// them would lose data outright. + /// - The segment holding the sync frontier (`tip_height`) — it is still + /// being written, and releasing it would force a reload on the very next + /// store. + /// - Any segment not entirely at or below the watermark, so a partially + /// committed segment is never released out from under an active reader. + /// + /// Segments queued in `to_delete` cannot appear here: both `truncate_above` + /// and `clear` remove a segment from `self.segments` in the same step that + /// queues it, so the two sets are disjoint by construction. + /// + /// Returns the number of segments released. + fn release_committed_segments(&mut self) -> usize { + let Some(committed) = self.committed_height else { + return 0; + }; + + // Still-written frontier; never release it. + let frontier_segment = self.tip_height.map(Self::height_to_segment_id); + + let items_per_segment = Segment::::ITEMS_PER_SEGMENT as u64; + let committed = committed as u64; + + // Widened to u64 so a segment id near u32::MAX cannot overflow the + // range arithmetic; such a segment simply fails the comparison. + let releasable: Vec = self + .segments + .iter() + .filter(|(id, segment)| { + if segment.state != SegmentState::Clean { + return false; + } + + if Some(**id) == frontier_segment { + return false; + } + + let last_height = (**id as u64) * items_per_segment + items_per_segment - 1; + last_height <= committed + }) + .map(|(id, _)| *id) + .collect(); + + for id in &releasable { + self.segments.remove(id); + } + + releasable.len() } #[inline] @@ -1307,6 +1444,206 @@ mod tests { ); } + /// The headline guarantee: a committed, clean segment is dropped from + /// memory on persist, and a later read transparently reloads it from disk + /// byte-identically. + /// + /// This is the leak that pinned a 350k-block backfill in RAM: the scan + /// spans fewer than `MAX_ACTIVE_SEGMENTS`, so the LRU never fires and every + /// item stays resident until the process dies. + #[tokio::test] + async fn test_release_committed_segments_reloads_from_disk() { + let tmp_dir = TempDir::new().unwrap(); + + const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; + + // Dense across segments 0 and 1, plus a few items into segment 2 so the + // frontier lives above the range we expect to be released. + let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT * 2 + 5); + + let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); + cache.store_items_at_height(&items, 0).await.unwrap(); + cache.persist(tmp_dir.path()).await; + + // Baseline: without a watermark nothing is released, which is exactly + // the pre-fix behavior. + assert_eq!(cache.committed_height(), None); + assert_eq!(cache.segments.len(), 3, "no watermark declared, so nothing is released"); + + // Segments 0 and 1 are now fully committed (their last heights are + // ITEMS_PER_SEGMENT-1 and 2*ITEMS_PER_SEGMENT-1). + cache.set_committed_height(ITEMS_PER_SEGMENT * 2 - 1); + cache.persist(tmp_dir.path()).await; + + assert!(!cache.segments.contains_key(&0), "committed clean segment 0 must be released"); + assert!(!cache.segments.contains_key(&1), "committed clean segment 1 must be released"); + assert!(cache.segments.contains_key(&2), "frontier segment must stay resident"); + assert_eq!(cache.segments.len(), 1); + + // The cache-level watermarks are unaffected by residency. + assert_eq!(cache.start_height(), Some(0)); + assert_eq!(cache.tip_height(), Some(ITEMS_PER_SEGMENT * 2 + 4)); + + // Reading the released range reloads both segments from disk and + // returns exactly the bytes that were written. + let reread = cache.get_items(0..ITEMS_PER_SEGMENT * 2).await.unwrap(); + assert_eq!(reread, items[0..(ITEMS_PER_SEGMENT * 2) as usize]); + + // Single-item reads across the boundary agree too. + assert_eq!(cache.get_item(0).await.unwrap(), Some(items[0])); + assert_eq!( + cache.get_item(ITEMS_PER_SEGMENT).await.unwrap(), + Some(items[ITEMS_PER_SEGMENT as usize]) + ); + } + + /// Never release a segment that is dirty (its items exist only in memory), + /// the frontier segment (still being written), or a segment only partially + /// below the watermark. + #[tokio::test] + async fn test_release_skips_dirty_frontier_and_partial_segments() { + let tmp_dir = TempDir::new().unwrap(); + + const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; + + let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); + + // Sparse writes into segments 0, 1 and 2. + cache.store_items_at_height(&FilterHeader::dummy_batch(0..1), 10).await.unwrap(); + cache + .store_items_at_height(&FilterHeader::dummy_batch(1..2), ITEMS_PER_SEGMENT + 10) + .await + .unwrap(); + cache + .store_items_at_height(&FilterHeader::dummy_batch(2..3), ITEMS_PER_SEGMENT * 2 + 10) + .await + .unwrap(); + + // A watermark above everything, but nothing has been persisted yet, so + // all three segments are Dirty and none may be released. + cache.set_committed_height(u32::MAX); + let released = cache.release_committed_segments(); + assert_eq!(released, 0, "dirty segments must never be released"); + assert_eq!(cache.segments.len(), 3); + + cache.persist(tmp_dir.path()).await; + // persist() cleaned all three, then released every non-frontier one. + assert_eq!(cache.segments.keys().copied().collect::>(), vec![2]); + + // Now verify the partial-coverage rule: a watermark inside segment 0 + // does not release it, because its upper heights are not yet committed. + let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); + let _ = cache.get_segment_mut(&0).await.unwrap(); + assert!(cache.segments.contains_key(&0)); + + cache.set_committed_height(ITEMS_PER_SEGMENT - 2); // one short of the segment's last height + assert_eq!(cache.release_committed_segments(), 0, "partially committed segment must stay"); + assert!(cache.segments.contains_key(&0)); + + cache.set_committed_height(ITEMS_PER_SEGMENT - 1); // exactly the last height + assert_eq!(cache.release_committed_segments(), 1, "fully committed segment is releasable"); + assert!(!cache.segments.contains_key(&0)); + } + + /// The rescan path re-reads heights far below the watermark. Those segments + /// were released, so this exercises the lazy reload under the access + /// pattern `reset_for_rescan` / `start_download` produce. + #[tokio::test] + async fn test_released_segments_serve_a_rescan_reread() { + let tmp_dir = TempDir::new().unwrap(); + + const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; + + let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT * 2 + 5); + + let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); + cache.store_items_at_height(&items, 0).await.unwrap(); + cache.set_committed_height(ITEMS_PER_SEGMENT * 2 - 1); + cache.persist(tmp_dir.path()).await; + assert_eq!(cache.segments.len(), 1); + + // A wallet appears behind the scan: the manager rolls the watermark + // back and re-reads from a low height. + cache.set_committed_height(0); + let rescanned = cache.get_items(5..ITEMS_PER_SEGMENT + 5).await.unwrap(); + assert_eq!(rescanned, items[5..(ITEMS_PER_SEGMENT + 5) as usize]); + + // With the watermark rolled back, the re-read segments are retained + // rather than being dropped again underneath the rescan. + cache.persist(tmp_dir.path()).await; + assert!(cache.segments.contains_key(&0), "rolled-back watermark must stop re-release"); + + // And the data still round-trips after a full reopen. + let mut reloaded = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); + assert_eq!(reloaded.tip_height(), Some(ITEMS_PER_SEGMENT * 2 + 4)); + assert_eq!( + reloaded.get_items(0..ITEMS_PER_SEGMENT * 2 + 5).await.unwrap(), + items, + "released-then-reloaded data must survive a process restart byte-identically" + ); + } + + /// The watermark must never outlive the data it refers to: `truncate_above` + /// clamps it and `clear` drops it, so a later scan of the same range is not + /// released out from under itself. + #[tokio::test] + async fn test_committed_height_follows_truncate_and_clear() { + let tmp_dir = TempDir::new().unwrap(); + + let items = FilterHeader::dummy_batch(0..30); + + let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); + cache.store_items_at_height(&items, 0).await.unwrap(); + + cache.set_committed_height(25); + assert_eq!(cache.committed_height(), Some(25)); + + // Truncating above a height below the watermark clamps it. + cache.truncate_above(10).await.unwrap(); + assert_eq!(cache.committed_height(), Some(10)); + + // Truncating above the watermark leaves it alone. + cache.set_committed_height(5); + cache.truncate_above(8).await.unwrap(); + assert_eq!(cache.committed_height(), Some(5)); + + cache.clear().unwrap(); + assert_eq!(cache.committed_height(), None, "an empty cache has nothing committed"); + } + + /// A released segment must still accept new writes into its unused slots, + /// reloading the existing contents first rather than silently starting from + /// a blank segment (which would drop the persisted items on the next write). + #[tokio::test] + async fn test_store_into_released_segment_preserves_existing_items() { + let tmp_dir = TempDir::new().unwrap(); + + const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; + + let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); + cache.store_items_at_height(&FilterHeader::dummy_batch(0..10), 0).await.unwrap(); + cache + .store_items_at_height(&FilterHeader::dummy_batch(50..55), ITEMS_PER_SEGMENT) + .await + .unwrap(); + + cache.set_committed_height(ITEMS_PER_SEGMENT - 1); + cache.persist(tmp_dir.path()).await; + assert!(!cache.segments.contains_key(&0), "segment 0 released"); + + // Write into a gap in the released segment. + cache.store_items_at_height(&FilterHeader::dummy_batch(90..95), 100).await.unwrap(); + + // Both the reloaded originals and the new items are present. + assert_eq!(cache.get_items(0..10).await.unwrap(), FilterHeader::dummy_batch(0..10)); + assert_eq!(cache.get_items(100..105).await.unwrap(), FilterHeader::dummy_batch(90..95)); + + cache.persist(tmp_dir.path()).await; + let mut reloaded = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); + assert_eq!(reloaded.get_items(0..10).await.unwrap(), FilterHeader::dummy_batch(0..10)); + assert_eq!(reloaded.get_items(100..105).await.unwrap(), FilterHeader::dummy_batch(90..95)); + } + #[test] fn test_segment_insert_get() { let segment_id = 10; diff --git a/dash-spv/src/sync/blocks/manager.rs b/dash-spv/src/sync/blocks/manager.rs index 70df8624b..ab97e2fe0 100644 --- a/dash-spv/src/sync/blocks/manager.rs +++ b/dash-spv/src/sync/blocks/manager.rs @@ -77,6 +77,9 @@ impl BlocksManager SyncResult> { let mut events = Vec::new(); + // Highest height applied in this drain, used below to advance the + // storage's committed watermark exactly once. + let mut last_applied: Option = None; // Process blocks in height order using pipeline's ordering logic while let Some((block, height, interested)) = self.pipeline.take_next_ordered_block() { @@ -124,6 +127,7 @@ impl BlocksManager BlocksManager= batch_end { @@ -596,6 +601,13 @@ impl