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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
//! 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.
///
/// ```
/// let s = kevy_embedded::Store::open(kevy_embedded::Config::default())?;
/// s.set(b"u:1", b"a")?;
/// let frozen = s.snapshot();
/// s.set(b"u:2", b"b")?; // lands after the freeze
/// assert_eq!(frozen.keys_prefix(b"u:").len(), 1);
/// assert_eq!(s.snapshot().keys_prefix(b"u:").len(), 2);
/// # Ok::<(), kevy_embedded::KevyError>(())
/// ```
#[derive(Debug)]
pub struct Snapshot {
views: Vec<SnapshotView>,
}
/// One entry from [`Snapshot::each_prefix`] / [`Snapshot::keys_prefix`].
///
/// ```
/// let s = kevy_embedded::Store::open(kevy_embedded::Config::default())?;
/// s.set(b"u:1", b"a")?;
/// let entries = s.snapshot().keys_prefix(b"u:");
/// assert_eq!((entries[0].key.as_slice(), entries[0].ttl_ms), (&b"u:1"[..], None));
/// # Ok::<(), kevy_embedded::KevyError>(())
/// ```
#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub struct SnapshotEntry {
/// The key.
///
/// ```
/// let s = kevy_embedded::Store::open(kevy_embedded::Config::default())?;
/// s.set(b"u:1", b"a")?;
/// s.set(b"other", b"b")?;
/// let keys: Vec<Vec<u8>> = s.snapshot().keys_prefix(b"u:").into_iter().map(|e| e.key).collect();
/// assert_eq!(keys, [b"u:1".to_vec()]);
/// # Ok::<(), kevy_embedded::KevyError>(())
/// ```
pub key: Vec<u8>,
/// Remaining TTL in ms at freeze time (`None` = no expiry).
///
/// ```
/// let s = kevy_embedded::Store::open(kevy_embedded::Config::default())?;
/// s.set_with_ttl(b"u:1", b"a", std::time::Duration::from_secs(60))?;
/// let ttl = s.snapshot().keys_prefix(b"u:")[0].ttl_ms;
/// assert!(ttl.is_some_and(|ms| ms > 50_000 && ms <= 60_000));
/// # Ok::<(), kevy_embedded::KevyError>(())
/// ```
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 }
}
}