1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
//! v2.4 — public point-in-time snapshot view.
//!
//! [`Store::snapshot`] freezes a consistent COW view of the WHOLE
//! keyspace: all shard write locks are taken in shard order (the same
//! deterministic-order discipline as `atomic_all_shards`, so the two
//! can't deadlock against each other), every shard's O(n)-shallow
//! view is collected inside that single window, then the locks drop.
//! Writers block only for the collection (~8 ns/entry), not for the
//! caller's subsequent iteration.
//!
//! The primary consumer: rebuilding derived state after a
//! `FeedError::Resync` — freeze a view, note `changes_tail()`, scan
//! your prefix from the view, resume the feed from the noted cursor.
use kevy_store::SnapshotView;
use crate::store::{Store, lock_write};
/// A frozen, consistent point-in-time view of the whole store.
pub struct Snapshot {
views: Vec<SnapshotView>,
}
/// One entry from [`Snapshot::each_prefix`] / [`Snapshot::keys_prefix`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SnapshotEntry {
/// The key.
pub key: Vec<u8>,
/// Remaining TTL in ms at freeze time (`None` = no expiry).
pub ttl_ms: Option<u64>,
}
impl Snapshot {
/// Visit every entry under `prefix` as `(key, value, ttl_ms)` —
/// the raw [`kevy_store::Value`] borrow, for callers that want the
/// typed payload without copies.
pub fn each_prefix<F: FnMut(&[u8], &kevy_store::Value, Option<u64>)>(
&self,
prefix: &[u8],
mut f: F,
) {
for view in &self.views {
view.each(|k, v, ttl| {
if k.starts_with(prefix) {
f(k, v, ttl);
}
});
}
}
/// Collect the keys (+ TTLs) under `prefix`, unordered across
/// shards.
pub fn keys_prefix(&self, prefix: &[u8]) -> Vec<SnapshotEntry> {
let mut out = Vec::new();
self.each_prefix(prefix, |k, _, ttl| {
out.push(SnapshotEntry { key: k.to_vec(), ttl_ms: ttl });
});
out
}
/// Total entries in the view.
pub fn len(&self) -> usize {
let mut n = 0;
for view in &self.views {
view.each(|_, _, _| n += 1);
}
n
}
/// Whether the view is empty.
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
impl Store {
/// Freeze a consistent point-in-time [`Snapshot`] of the whole
/// keyspace (see the module doc for the locking discipline).
pub fn snapshot(&self) -> Snapshot {
// Deterministic shard order (same as atomic_all_shards) — hold
// ALL write locks across the collection so no write lands
// between shard freezes.
let guards: Vec<_> = self.shards.iter().map(|s| lock_write(s)).collect();
let views = guards.iter().map(|g| g.store.collect_snapshot()).collect();
drop(guards);
Snapshot { views }
}
}