diff --git a/crates/store/src/state/lifecycle.rs b/crates/store/src/state/lifecycle.rs index e40d46388..e2d5c562e 100644 --- a/crates/store/src/state/lifecycle.rs +++ b/crates/store/src/state/lifecycle.rs @@ -3,7 +3,6 @@ use std::num::NonZeroUsize; use std::path::Path; use std::sync::Arc; -use std::sync::atomic::AtomicUsize; use arc_swap::ArcSwap; use miden_node_tracing::spawn::spawn_blocking_in_current_span; @@ -32,15 +31,7 @@ use crate::state::loader::{ verify_tree_consistency, }; use crate::state::writer::{WriteRequest, WriteWorker, WriterTask}; -use crate::state::{ - BlockCache, - BlockWriter, - ProofCache, - ProofWriter, - SnapshotGuard, - State, - StateSnapshot, -}; +use crate::state::{BlockCache, BlockWriter, ProofCache, ProofWriter, State, StateSnapshot}; use crate::{COMPONENT, DataDirectory, DatabaseOptions}; /// Awaits a spawned load task, forwarding its result. @@ -276,9 +267,6 @@ impl State { let block_cache = BlockCache::new(BLOCK_CACHE_CAPACITY); let proof_cache = ProofCache::new(PROOF_CACHE_CAPACITY); - // Shared counter of live snapshot generations, for observability. - let snapshots_live = Arc::new(AtomicUsize::new(0)); - // Create the initial snapshot from reader views of the just-loaded trees. let initial_snapshot = Arc::new(StateSnapshot::new( nullifier_tree @@ -289,7 +277,6 @@ impl State { forest .reader() .map_err(|e| StateInitializationError::AccountStateForestIoError(e.as_report()))?, - SnapshotGuard::new(Arc::clone(&snapshots_live), latest_block_num), )); let latest_snapshot = Arc::new(ArcSwap::from(initial_snapshot)); @@ -309,7 +296,6 @@ impl State { account_tree, blockchain, forest, - snapshots_live, apply_block_thread_priority, ); let state = Self { diff --git a/crates/store/src/state/mod.rs b/crates/store/src/state/mod.rs index 3a519e36a..ec61c1131 100644 --- a/crates/store/src/state/mod.rs +++ b/crates/store/src/state/mod.rs @@ -21,8 +21,8 @@ pub use lifecycle::LoadedState; use miden_protocol::block::BlockNumber; pub use replica::{BlockCache, BlockNotification, ProofCache, ProofNotification}; use tokio::sync::watch; +use view::StateSnapshot; pub use view::{ScopedBlockNum, ScopedBlockRange, StateView, TransactionInputs}; -use view::{SnapshotGuard, StateSnapshot}; pub use writer::{BlockWriter, ProofWriter, WriterTask}; use crate::blocks::BlockStore; diff --git a/crates/store/src/state/view/mod.rs b/crates/store/src/state/view/mod.rs index 334c4a511..a86c6d661 100644 --- a/crates/store/src/state/view/mod.rs +++ b/crates/store/src/state/view/mod.rs @@ -26,12 +26,7 @@ mod scoped; pub use scoped::{ScopedBlockNum, ScopedBlockRange}; mod snapshot; -pub(in crate::state) use snapshot::{ - PublishedGenerations, - SNAPSHOTS_LIVE_WARN_THRESHOLD, - SnapshotGuard, - StateSnapshot, -}; +pub(in crate::state) use snapshot::{PublishedGenerations, StateSnapshot}; mod account; mod batch_inputs; @@ -50,8 +45,7 @@ pub use transaction_inputs::TransactionInputs; /// /// Obtained from [`State::view`]; create one per request and drop it when the request completes. /// Holding a view pins a snapshot generation (and thereby the `RocksDB` snapshots backing the -/// trees), so it must not be stored in long-lived structs; leaked or slow readers are reported by -/// the store's snapshot-lifetime warnings. +/// trees), so it must not be stored in long-lived structs. /// /// Reads that are technically not block-scoped (e.g. content-addressed note scripts) also live /// here so that every read path flows through a single, consistently-scoped type. @@ -94,9 +88,7 @@ impl State { /// /// Work in the closure should be kept to low-complexity compute over the view, ideally with no /// I/O and no other `.await` points. Anything slower holds the pinned snapshot, and therefore - /// its underlying `RocksDB` snapshot, for as long as it runs. The snapshot's lifetime is logged - /// as a warning if held too long, but that is a backstop, not a substitute for keeping closures - /// short. + /// its underlying `RocksDB` snapshot, for as long as it runs. pub async fn with_view(&self, f: impl AsyncFnOnce(&StateView) -> R) -> R { let view = self.view(); f(&view).await diff --git a/crates/store/src/state/view/snapshot.rs b/crates/store/src/state/view/snapshot.rs index 50616f2d3..87e05adc3 100644 --- a/crates/store/src/state/view/snapshot.rs +++ b/crates/store/src/state/view/snapshot.rs @@ -1,23 +1,22 @@ //! In-memory snapshot machinery for lock-free reads. //! //! Readers access the store's tree state through immutable [`StateSnapshot`] snapshots published -//! by the block writer after each committed block. [`SnapshotGuard`] tracks how many snapshot -//! generations are pinned by readers, since each generation pins a `RocksDB` snapshot. The writer -//! additionally remembers each published generation in [`PublishedGenerations`], whose oldest -//! still-pinned height feeds snapshot-aware history pruning, since SQLite reads have no -//! point-in-time protection equivalent to the `RocksDB` snapshots backing the trees. +//! by the block writer after each committed block. The writer remembers each published generation +//! in [`PublishedGenerations`]. The oldest still-pinned height feeds snapshot-aware history +//! pruning, since SQLite reads have no point-in-time protection equivalent to the `RocksDB` +//! snapshots backing the trees. +//! +//! Everything here operates on whole generations. Readers acquire a snapshot through +//! [`StateView`](super::StateView), the request-scoped handle. use std::collections::VecDeque; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, OnceLock, Weak}; +use std::sync::{Arc, Weak}; use std::time::{Duration, Instant}; -use miden_node_tracing::{debug, warn}; use miden_protocol::block::nullifier_tree::NullifierTree; use miden_protocol::block::{BlockNumber, Blockchain}; use miden_protocol::crypto::merkle::smt::LargeSmt; -use crate::COMPONENT; use crate::account_state_forest::{ AccountStateForest, AccountStateForestBackendReader, @@ -26,31 +25,16 @@ use crate::account_state_forest::{ use crate::accounts::AccountTreeWithHistory; use crate::state::loader::TreeStorageReader; -/// Time held past supersession above which [`SnapshotGuard`] logs a warning on release. -/// -/// Readers are expected to be request-scoped, so a superseded generation outliving several block -/// intervals indicates a slow or leaked reader pinning a `RocksDB` snapshot (see -/// [`SnapshotGuard`]). The clock starts at supersession rather than creation: the latest -/// generation is always pinned by the published pointer, so time spent as the current generation -/// (idle chains, shutdown) says nothing about reader behaviour. -const SNAPSHOT_SUPERSEDED_WARN_THRESHOLD: Duration = Duration::from_secs(10); - -/// Number of live snapshot generations above which the block writer logs a warning after -/// publishing a new snapshot. -/// -/// Steady state is 1-2 generations: the just-published snapshot plus predecessors briefly pinned -/// by in-flight requests. A sustained higher count means slow or leaked readers are holding old -/// generations alive (see [`SnapshotGuard`]). -pub(in crate::state) const SNAPSHOTS_LIVE_WARN_THRESHOLD: u64 = 4; - -/// Upper bound on how far the snapshot-aware pruning tip may lag the chain tip. +/// Upper bound, in blocks, on how far history pruning may trail the chain tip. /// -/// History pruning keys off the oldest live snapshot generation (see -/// [`PublishedGenerations::prune_tip`]), so a leaked or pathologically slow reader would -/// otherwise stall pruning indefinitely. Beyond this -/// many blocks of lag the writer prunes anyway, accepting the historical-read race for that reader -/// (which the snapshot-lifetime warnings have long since reported). One full retention window, so -/// worst-case retained history is bounded at twice the window. +/// Pruning normally waits for the oldest live snapshot generation, so SQLite history that a +/// pinned generation can still read is not deleted (see [`PublishedGenerations::advance`]). +/// Without a bound, one leaked reader would block pruning forever. Once a generation falls more +/// than this many blocks behind the tip, pruning stops waiting for it. Rows inside that reader's +/// retention window may then be deleted while it still holds the generation, so its historical +/// SQLite reads may observe missing data. Its tree reads are unaffected: the generation still +/// pins the `RocksDB` snapshots backing the trees. The cap equals one retention window, so +/// retained history never exceeds two windows. const SNAPSHOT_PRUNE_LAG_CAP: u32 = HISTORICAL_BLOCK_RETENTION; // PUBLISHED GENERATIONS @@ -61,14 +45,30 @@ const SNAPSHOT_PRUNE_LAG_CAP: u32 = HISTORICAL_BLOCK_RETENTION; /// /// Owned exclusively by the writer — no locks or shared state. Liveness is not tracked /// separately: a [`Weak`] per generation asks the snapshot's own [`Arc`] refcount, which is the -/// ground truth for "some reader can still see this height". Dead and no-longer-relevant entries -/// are discarded on each [`Self::prune_tip`] call (once per applied block), which bounds the -/// deque to roughly [`SNAPSHOT_PRUNE_LAG_CAP`] entries even when a reader leaks its snapshot. +/// ground truth for "some reader can still see this height". Dead entries are discarded on each +/// [`Self::advance`] call (once per applied block); pinned entries are kept regardless of age so +/// the true oldest pinned height stays observable, which bounds the deque to one entry per live +/// snapshot generation (each entry is small next to the pinned snapshot itself). /// /// Generic over the pinned type for testability; the writer uses `T = StateSnapshot`. pub(in crate::state) struct PublishedGenerations { /// Published generations in ascending height order. - entries: VecDeque<(BlockNumber, Weak)>, + entries: VecDeque>, +} + +/// One published snapshot generation tracked by [`PublishedGenerations`]. +struct Generation { + height: BlockNumber, + /// The time when a successor generation was published. `None` while this is the latest + /// generation. + /// + /// [`PublishedGenerations::record`] sets this field on the previous back entry. A + /// generation's age is measured from this point rather than from its publication, because + /// the writer itself keeps the latest generation alive through the published pointer: on an + /// idle chain the latest generation can stay alive for hours with no reader holding it. Only + /// time spent alive after a successor exists is evidence that a reader holds the generation. + superseded_at: Option, + pinned: Weak, } impl PublishedGenerations { @@ -76,110 +76,78 @@ impl PublishedGenerations { Self { entries: VecDeque::new() } } - /// Records a newly published generation. Heights must be recorded in ascending order. + /// Records a newly published generation and marks the previous latest generation as superseded. + /// Heights must be recorded in ascending order. pub(in crate::state) fn record(&mut self, height: BlockNumber, pinned: &Arc) { debug_assert!( - self.entries.back().is_none_or(|(back, _)| *back < height), + self.entries.back().is_none_or(|back| back.height < height), "generation {height} published out of order", ); - self.entries.push_back((height, Arc::downgrade(pinned))); + // The new generation supersedes the previous back entry. + if let Some(previous) = self.entries.back_mut() { + previous.superseded_at = Some(Instant::now()); + } + self.entries.push_back(Generation { + height, + superseded_at: None, + pinned: Arc::downgrade(pinned), + }); } - /// Returns the effective chain tip for history pruning. + /// Returns the number of recorded generations. /// - /// The store's SQLite reads are scoped only by an upper block bound, with no point-in-time - /// protection equivalent to the `RocksDB` snapshots backing the trees. Pruning therefore - /// treats the oldest still-pinned generation as the tip: a generation pinned at height `H` - /// keeps the same retention window it had when `H` was the tip, and pruning simply lags until - /// it is released. The lag is capped at [`SNAPSHOT_PRUNE_LAG_CAP`] blocks so a leaked reader - /// cannot stall pruning indefinitely: entries below the cap's floor are discarded despite - /// still being pinned, as are entries that are no longer pinned. - pub(in crate::state) fn prune_tip(&mut self, chain_tip: BlockNumber) -> BlockNumber { - let lag_floor = chain_tip.as_u32().saturating_sub(SNAPSHOT_PRUNE_LAG_CAP); - while let Some((height, pinned)) = self.entries.front() { - // Drop entries below the lag floor or that are no longer pinned. - if height.as_u32() < lag_floor || pinned.strong_count() == 0 { - self.entries.pop_front(); - } else { - break; - } - } - // Return the prune tip, which is the oldest pinned generation or the chain tip. - self.entries.front().map_or(chain_tip, |(height, _)| (*height).min(chain_tip)) + /// Only [`Self::advance`] discards dead generations, so the count is exact directly after + /// `advance` and is an upper bound on the live count between calls. + pub(in crate::state) fn live(&self) -> usize { + self.entries.len() } -} -// SNAPSHOT GUARD -// ================================================================================================ - -/// RAII member of [`StateSnapshot`] that tracks the number of live snapshot generations. -/// -/// [`StateSnapshot`] is dropped exactly when the last [`Arc`] reference to it is released, so the -/// shared counter reports how many distinct snapshot generations are currently pinned by readers. -/// A sustained count above 1-2 means slow readers are holding old generations alive. Each -/// generation pins a `RocksDB` snapshot, which delays garbage collection of superseded key -/// versions during compaction (compaction itself keeps running); the retained garbage grows with -/// write churn for as long as the snapshot is held and is reclaimed once it is released. A held -/// generation also holds back SQLite history pruning (see [`PublishedGenerations::prune_tip`]). -/// -/// Readers are expected to be request-scoped, so a superseded generation should be released well -/// within a block interval. Outliving supersession by more than -/// [`SNAPSHOT_SUPERSEDED_WARN_THRESHOLD`] is logged at warn level; a generation that is never -/// superseded (the latest at shutdown) is released silently regardless of age. -pub(in crate::state) struct SnapshotGuard { - live: Arc, - created_at: Instant, - /// Set by the writer when a newer generation replaces this one as the published snapshot. - superseded_at: OnceLock, - block_num: BlockNumber, -} - -impl SnapshotGuard { - pub(in crate::state) fn new(live: Arc, block_num: BlockNumber) -> Self { - live.fetch_add(1, Ordering::Relaxed); - Self { - live, - created_at: Instant::now(), - superseded_at: OnceLock::new(), - block_num, + /// Discards generations no longer pinned by any reader and reports on those that remain. + /// + /// The prune tip is the effective chain tip for history pruning. The store's SQLite reads are + /// scoped only by an upper block bound, with no point-in-time protection equivalent to the + /// `RocksDB` snapshots backing the trees. Pruning therefore treats the oldest still-pinned + /// generation as the tip: a generation pinned at height `H` keeps the same retention window it + /// had when `H` was the tip, and pruning simply lags until it is released. The lag is capped + /// at [`SNAPSHOT_PRUNE_LAG_CAP`] blocks so a leaked reader cannot stall pruning indefinitely: + /// generations below the cap's floor no longer hold pruning back, but stay recorded so + /// [`GenerationsStatus::oldest_pinned`] keeps reporting them for as long as they are pinned. + pub(in crate::state) fn advance(&mut self, chain_tip: BlockNumber) -> GenerationsStatus { + self.entries.retain(|generation| generation.pinned.strong_count() > 0); + let oldest = self.entries.front(); + let oldest_pinned = oldest.map(|generation| generation.height); + let oldest_superseded_for = + oldest.and_then(|generation| generation.superseded_at).map(|at| at.elapsed()); + // The prune tip is the oldest pinned generation within the lag cap, or the chain tip. + let lag_floor = chain_tip.as_u32().saturating_sub(SNAPSHOT_PRUNE_LAG_CAP); + let prune_tip = self + .entries + .iter() + .map(|generation| generation.height) + .find(|height| height.as_u32() >= lag_floor) + .map_or(chain_tip, |height| height.min(chain_tip)); + GenerationsStatus { + prune_tip, + oldest_pinned, + oldest_superseded_for, } } - - /// Marks this generation as superseded by a newer published snapshot, starting the clock - /// against which slow readers are measured. Only the first call takes effect. - pub(in crate::state) fn mark_superseded(&self) { - let _ = self.superseded_at.set(Instant::now()); - } } -impl Drop for SnapshotGuard { - fn drop(&mut self) { - let remaining = self.live.fetch_sub(1, Ordering::Relaxed) - 1; - let lifetime_ms = u64::try_from(self.created_at.elapsed().as_millis()).unwrap_or(u64::MAX); - let block_num = self.block_num.as_u32(); - let superseded_for = self.superseded_at.get().map(Instant::elapsed); - if let Some(superseded_for) = - superseded_for.filter(|held| *held > SNAPSHOT_SUPERSEDED_WARN_THRESHOLD) - { - let superseded_for_ms = u64::try_from(superseded_for.as_millis()).unwrap_or(u64::MAX); - warn!( - target: COMPONENT, - "State snapshot held for excessive time after supersession", - block.number = block_num, - snapshot.lifetime_ms = lifetime_ms, - snapshot.superseded_for_ms = superseded_for_ms, - snapshots.live = remaining - ); - } else { - debug!( - target: COMPONENT, - "State snapshot released", - block.number = block_num, - snapshot.lifetime_ms = lifetime_ms, - snapshots.live = remaining - ); - } - } +/// Per-block report on the still-pinned snapshot generations; see +/// [`PublishedGenerations::advance`]. +pub(in crate::state) struct GenerationsStatus { + /// The effective chain tip for history pruning: the oldest still-pinned generation within + /// [`SNAPSHOT_PRUNE_LAG_CAP`], or the chain tip when none is pinned. + pub(in crate::state) prune_tip: BlockNumber, + /// The oldest generation still pinned by any reader, regardless of the lag cap. `None` when no + /// generation is pinned. + pub(in crate::state) oldest_pinned: Option, + /// The time since the oldest pinned generation was superseded. `None` when that generation is + /// still the latest generation, or when no generation is pinned. In steady state the value is + /// `None` or much smaller than one block interval. A value that grows across blocks means a + /// slow or leaked reader pins an old generation. + pub(in crate::state) oldest_superseded_for: Option, } // STATE SNAPSHOT @@ -191,6 +159,11 @@ impl Drop for SnapshotGuard { /// [`AccountStateForestBackendReader`]), so any number of readers can access the data concurrently /// without holding a lock and without blocking the writer. /// +/// A pinned snapshot pins the `RocksDB` snapshots that back the trees. A pinned `RocksDB` +/// snapshot delays garbage collection of superseded key versions during compaction; compaction +/// itself continues. A pinned snapshot also holds back SQLite history pruning (see +/// [`PublishedGenerations::advance`]). +/// /// The writer and lifecycle can *construct* snapshots via [`Self::new`], but the fields are only /// readable within the view module tree: every read outside it must go through /// [`StateView`](super::StateView). @@ -199,34 +172,24 @@ pub(in crate::state) struct StateSnapshot { pub(super) blockchain: Blockchain, pub(super) account_tree: AccountTreeWithHistory, pub(super) forest: AccountStateForest, - /// Keeps the live-snapshot count accurate; see [`SnapshotGuard`]. - guard: SnapshotGuard, } impl StateSnapshot { - /// Assembles a snapshot from reader views of the trees and the guard tracking its generation. + /// Assembles a snapshot from reader views of the trees. pub(in crate::state) fn new( nullifier_tree: NullifierTree>, blockchain: Blockchain, account_tree: AccountTreeWithHistory, forest: AccountStateForest, - guard: SnapshotGuard, ) -> Self { Self { nullifier_tree, blockchain, account_tree, forest, - guard, } } - /// Marks this snapshot as superseded by a newer published generation; see - /// [`SnapshotGuard::mark_superseded`]. - pub(in crate::state) fn mark_superseded(&self) { - self.guard.mark_superseded(); - } - /// Returns the latest block number. pub(in crate::state) fn latest_block_num(&self) -> BlockNumber { self.blockchain @@ -240,12 +203,14 @@ mod tests { use super::*; #[test] - fn prune_tip_tracks_oldest_pinned_height_across_out_of_order_drops() { + fn advance_tracks_oldest_pinned_height_across_out_of_order_drops() { let mut published = PublishedGenerations::::new(); let tip = BlockNumber::from(100); // No live generations: prune at the tip. - assert_eq!(published.prune_tip(tip), tip); + let status = published.advance(tip); + assert_eq!(status.prune_tip, tip); + assert_eq!(status.oldest_pinned, None); let gen_97 = Arc::new(97); let gen_98 = Arc::new(98); @@ -253,24 +218,35 @@ mod tests { published.record(BlockNumber::from(97), &gen_97); published.record(BlockNumber::from(98), &gen_98); published.record(BlockNumber::from(99), &gen_99); - assert_eq!(published.prune_tip(tip), BlockNumber::from(97)); + let status = published.advance(tip); + assert_eq!(status.prune_tip, BlockNumber::from(97)); + assert_eq!(status.oldest_pinned, Some(BlockNumber::from(97))); + assert_eq!(published.live(), 3); + // Recording generation 98 superseded generation 97. + assert!(status.oldest_superseded_for.is_some()); // Dropping a middle generation leaves the oldest unchanged. drop(gen_98); - assert_eq!(published.prune_tip(tip), BlockNumber::from(97)); + assert_eq!(published.advance(tip).prune_tip, BlockNumber::from(97)); drop(gen_97); - assert_eq!(published.prune_tip(tip), BlockNumber::from(99)); + let status = published.advance(tip); + assert_eq!(status.prune_tip, BlockNumber::from(99)); + assert_eq!(published.live(), 1); + // Generation 99 is the latest generation and is not superseded. + assert!(status.oldest_superseded_for.is_none()); // A pinned generation never advances pruning past the tip. - assert_eq!(published.prune_tip(BlockNumber::from(98)), BlockNumber::from(98)); + assert_eq!(published.advance(BlockNumber::from(98)).prune_tip, BlockNumber::from(98)); drop(gen_99); - assert_eq!(published.prune_tip(tip), tip); + let status = published.advance(tip); + assert_eq!(status.prune_tip, tip); + assert_eq!(status.oldest_pinned, None); } #[test] - fn prune_tip_discards_leaked_entries_below_the_lag_floor() { + fn advance_caps_prune_lag_but_keeps_reporting_the_leaked_oldest() { let mut published = PublishedGenerations::::new(); let leaked = Arc::new(1); published.record(BlockNumber::from(1), &leaked); @@ -278,11 +254,32 @@ mod tests { // While the leaked generation is within the lag cap it holds pruning back; near genesis the // lag floor saturates to zero. let tip = BlockNumber::from(SNAPSHOT_PRUNE_LAG_CAP); - assert_eq!(published.prune_tip(tip), BlockNumber::from(1)); - - // Once the tip advances past the cap it is discarded despite still being pinned, and no - // longer holds pruning back. + let status = published.advance(tip); + assert_eq!(status.prune_tip, BlockNumber::from(1)); + assert_eq!(status.oldest_pinned, Some(BlockNumber::from(1))); + // No successor generation was recorded, so the leaked generation is not superseded. + assert!(status.oldest_superseded_for.is_none()); + + // Once the tip advances past the cap it no longer holds pruning back, but is still reported + // as the oldest pinned generation for as long as it is pinned. let tip = BlockNumber::from(SNAPSHOT_PRUNE_LAG_CAP + 2); - assert_eq!(published.prune_tip(tip), tip); + let status = published.advance(tip); + assert_eq!(status.prune_tip, tip); + assert_eq!(status.oldest_pinned, Some(BlockNumber::from(1))); + + // A newer pinned generation above the floor becomes the prune tip while the leaked one + // still drives the reported lag. + let gen_recent = Arc::new(2); + let recent_height = BlockNumber::from(SNAPSHOT_PRUNE_LAG_CAP + 1); + published.record(recent_height, &gen_recent); + let status = published.advance(tip); + assert_eq!(status.prune_tip, recent_height); + assert_eq!(status.oldest_pinned, Some(BlockNumber::from(1))); + // Recording the newer generation superseded the leaked generation. + assert!(status.oldest_superseded_for.is_some()); + + drop(leaked); + let status = published.advance(tip); + assert_eq!(status.oldest_pinned, Some(recent_height)); } } diff --git a/crates/store/src/state/writer/worker.rs b/crates/store/src/state/writer/worker.rs index 77ec55176..92296de30 100644 --- a/crates/store/src/state/writer/worker.rs +++ b/crates/store/src/state/writer/worker.rs @@ -1,6 +1,5 @@ //! The write worker: single-task owner of the store's mutable trees. -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Once}; use arc_swap::ArcSwap; @@ -38,12 +37,7 @@ use crate::db::{Db, NoteRecord}; use crate::errors::{ApplyBlockError, InvalidBlockError}; use crate::state::block_lifecycle::{BlockLifecycle, lifecycle_events_enabled}; use crate::state::loader::TreeStorage; -use crate::state::view::{ - PublishedGenerations, - SNAPSHOTS_LIVE_WARN_THRESHOLD, - SnapshotGuard, - StateSnapshot, -}; +use crate::state::view::{PublishedGenerations, StateSnapshot}; use crate::state::{BlockCache, BlockNotification}; use crate::{COMPONENT, HistoricalError, LOG_TARGET}; @@ -71,8 +65,6 @@ pub(in crate::state) struct WriteWorker { blockchain: Blockchain, /// The mutable account state forest owned by this writer. forest: AccountStateForest, - /// Shared counter of live snapshot generations, for observability. - snapshots_live: Arc, /// Writer-local log of published generations; its oldest still-pinned height feeds the /// snapshot-aware history-pruning tip. published_generations: PublishedGenerations, @@ -110,7 +102,6 @@ impl WriteWorker { account_tree: AccountTreeWithHistory, blockchain: Blockchain, forest: AccountStateForest, - snapshots_live: Arc, apply_block_thread_priority: bool, ) -> Self { // Seed the generation log with the initial snapshot so its readers hold back pruning @@ -138,7 +129,6 @@ impl WriteWorker { account_tree, blockchain, forest, - snapshots_live, published_generations, apply_pool, } @@ -232,7 +222,18 @@ impl WriteWorker { // generation rather than the actual tip: unlike the `RocksDB`-backed trees, SQLite reads // have no point-in-time protection, so pruning lags while pinned views can still reach // the history and catches up once they are released. - let prune_tip = self.published_generations.prune_tip(block_num); + let generations = self.published_generations.advance(block_num); + let snapshot_lag = generations + .oldest_pinned + .map_or(0, |oldest| block_num.as_u32() - oldest.as_u32()); + let oldest_superseded_for_ms = generations + .oldest_superseded_for + .map_or(0, |superseded| u64::try_from(superseded.as_millis()).unwrap_or(u64::MAX)); + miden_span_record!( + snapshots.lag_blocks = snapshot_lag, + snapshots.oldest_superseded_for_ms = oldest_superseded_for_ms + ); + let prune_tip = generations.prune_tip; let resolved_note_ids = self .db .apply_block( @@ -256,13 +257,11 @@ impl WriteWorker { ); // Atomically publish the new state. Readers that call `snapshot()` after this point will - // see the updated state. Readers holding the old snapshot continue unaffected, but are on - // the clock: a superseded generation held too long is reported on release. + // see the updated state. Readers that hold the old snapshot are not affected. self.published_generations.record(block_num, &snapshot); - self.latest_snapshot.swap(snapshot).mark_superseded(); + self.latest_snapshot.swap(snapshot); - let snapshots_live = self.check_live_snapshots(block_num); - miden_span_record!(snapshots.live = snapshots_live); + miden_span_record!(snapshots.live = self.published_generations.live()); // Push to cache and notify replica subscribers. self.block_cache @@ -281,24 +280,6 @@ impl WriteWorker { Ok(()) } - /// Returns the number of live snapshot generations, warning when slow readers are pinning too - /// many old generations in memory. - /// - /// The count is returned rather than recorded here because `miden_span_record!` must be used - /// within a `#[miden_instrument]` function. - fn check_live_snapshots(&self, block_num: BlockNumber) -> u64 { - let snapshots_live = self.snapshots_live.load(Ordering::Relaxed) as u64; - if snapshots_live > SNAPSHOTS_LIVE_WARN_THRESHOLD { - warn!( - target: COMPONENT, - "too many live state snapshots; slow readers are pinning old generations", - block.number = block_num, - snapshots.live = snapshots_live - ); - } - snapshots_live - } - /// Computes the note records and all tree and forest mutations for a block, without mutating /// any state, and serializes the signed block. /// @@ -404,7 +385,6 @@ impl WriteWorker { self.blockchain.clone(), self.account_tree.reader(), self.forest.reader().expect("forest snapshot creation should not fail"), - SnapshotGuard::new(Arc::clone(&self.snapshots_live), block_num), )) }) } diff --git a/crates/tracing/src/attribute.rs b/crates/tracing/src/attribute.rs index 7903504b9..fe1154db1 100644 --- a/crates/tracing/src/attribute.rs +++ b/crates/tracing/src/attribute.rs @@ -76,9 +76,9 @@ const NUMBER_FIELD_NAMES: &[&str] = &[ "retry.delay_ms", "shutdown.grace_period_ms", "snapshot.block_num", - "snapshot.lifetime_ms", - "snapshot.superseded_for_ms", + "snapshots.lag_blocks", "snapshots.live", + "snapshots.oldest_superseded_for_ms", "subscription.idle_ms", "subscription.stall_timeout_ms", "sync.block_gap",