use std::collections::VecDeque;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock, Weak};
use std::time::{Duration, Instant};
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,
HISTORICAL_BLOCK_RETENTION,
};
use crate::accounts::AccountTreeWithHistory;
use crate::state::loader::TreeStorageReader;
const SNAPSHOT_SUPERSEDED_WARN_THRESHOLD: Duration = Duration::from_secs(10);
pub(in crate::state) const SNAPSHOTS_LIVE_WARN_THRESHOLD: u64 = 4;
const SNAPSHOT_PRUNE_LAG_CAP: u32 = HISTORICAL_BLOCK_RETENTION;
pub(in crate::state) struct PublishedGenerations<T = StateSnapshot> {
entries: VecDeque<(BlockNumber, Weak<T>)>,
}
impl<T> PublishedGenerations<T> {
pub(in crate::state) fn new() -> Self {
Self { entries: VecDeque::new() }
}
pub(in crate::state) fn record(&mut self, height: BlockNumber, pinned: &Arc<T>) {
debug_assert!(
self.entries.back().is_none_or(|(back, _)| *back < height),
"generation {height} published out of order",
);
self.entries.push_back((height, Arc::downgrade(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() {
if height.as_u32() < lag_floor || pinned.strong_count() == 0 {
self.entries.pop_front();
} else {
break;
}
}
self.entries.front().map_or(chain_tip, |(height, _)| (*height).min(chain_tip))
}
}
pub(in crate::state) struct SnapshotGuard {
live: Arc<AtomicUsize>,
created_at: Instant,
superseded_at: OnceLock<Instant>,
block_num: BlockNumber,
}
impl SnapshotGuard {
pub(in crate::state) fn new(live: Arc<AtomicUsize>, block_num: BlockNumber) -> Self {
live.fetch_add(1, Ordering::Relaxed);
Self {
live,
created_at: Instant::now(),
superseded_at: OnceLock::new(),
block_num,
}
}
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);
tracing::warn!(
target: COMPONENT,
block_num,
snapshot.lifetime_ms = lifetime_ms,
snapshot.superseded_for_ms = superseded_for_ms,
snapshots.live = remaining,
"state snapshot held for excessive time after supersession",
);
} else {
tracing::debug!(
target: COMPONENT,
block_num,
snapshot.lifetime_ms = lifetime_ms,
snapshots.live = remaining,
"state snapshot released",
);
}
}
}
pub(in crate::state) struct StateSnapshot {
pub(super) nullifier_tree: NullifierTree<LargeSmt<TreeStorageReader>>,
pub(super) blockchain: Blockchain,
pub(super) account_tree: AccountTreeWithHistory<TreeStorageReader>,
pub(super) forest: AccountStateForest<AccountStateForestBackendReader>,
guard: SnapshotGuard,
}
impl StateSnapshot {
pub(in crate::state) fn new(
nullifier_tree: NullifierTree<LargeSmt<TreeStorageReader>>,
blockchain: Blockchain,
account_tree: AccountTreeWithHistory<TreeStorageReader>,
forest: AccountStateForest<AccountStateForestBackendReader>,
guard: SnapshotGuard,
) -> Self {
Self {
nullifier_tree,
blockchain,
account_tree,
forest,
guard,
}
}
pub(in crate::state) fn mark_superseded(&self) {
self.guard.mark_superseded();
}
pub(in crate::state) fn latest_block_num(&self) -> BlockNumber {
self.blockchain
.chain_tip()
.expect("chain should always have at least the genesis block")
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn prune_tip_tracks_oldest_pinned_height_across_out_of_order_drops() {
let mut published = PublishedGenerations::<u32>::new();
let tip = BlockNumber::from(100);
assert_eq!(published.prune_tip(tip), tip);
let gen_97 = Arc::new(97);
let gen_98 = Arc::new(98);
let gen_99 = Arc::new(99);
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));
drop(gen_98);
assert_eq!(published.prune_tip(tip), BlockNumber::from(97));
drop(gen_97);
assert_eq!(published.prune_tip(tip), BlockNumber::from(99));
assert_eq!(published.prune_tip(BlockNumber::from(98)), BlockNumber::from(98));
drop(gen_99);
assert_eq!(published.prune_tip(tip), tip);
}
#[test]
fn prune_tip_discards_leaked_entries_below_the_lag_floor() {
let mut published = PublishedGenerations::<u32>::new();
let leaked = Arc::new(1);
published.record(BlockNumber::from(1), &leaked);
let tip = BlockNumber::from(SNAPSHOT_PRUNE_LAG_CAP);
assert_eq!(published.prune_tip(tip), BlockNumber::from(1));
let tip = BlockNumber::from(SNAPSHOT_PRUNE_LAG_CAP + 2);
assert_eq!(published.prune_tip(tip), tip);
}
}