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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
//! Point-in-time snapshot views — the freeze half of COW serialization.
//!
//! [`Store::collect_snapshot`] walks the keyspace once and shallow-clones
//! every live entry: keys and string values copy their bytes (≤22 B inline
//! = a 24 B memcpy), collection values bump an `Arc` refcount. The pause is
//! O(n) at nanoseconds per entry — independent of collection sizes and of
//! disk speed. The returned [`SnapshotView`] is `Send`: hand it to a
//! background thread and serialize at leisure while the store keeps
//! mutating (writes copy-on-write via `Arc::make_mut`, deletions just drop
//! one strong ref — the view's data stays alive until it is dropped).
//!
//! TTLs are resolved to remaining-milliseconds at collect time, so the view
//! is a consistent instant: an entry that expires *after* the collect still
//! appears with the remaining TTL it had at that instant.
#[cfg(not(feature = "std"))]
use crate::nostd_prelude::*;
use crate::value::Value;
use crate::{SmallBytes, Store, now_ns, remaining_ms};
/// A frozen, `Send` view of one store's live entries at a single instant.
pub struct SnapshotView {
entries: Vec<(SmallBytes, Value, Option<u64>)>,
/// Hash field TTLs frozen with the view.
hfttl: Vec<(SmallBytes, SmallBytes, u64)>,
/// Tiering view pinning: every vlog file that existed at
/// collect time. A cold stub cloned into the view can only
/// reference these, and a pinned file survives compaction until the
/// last Arc drops — so the view's offsets stay valid for its whole
/// life, however long the serializer thread takes.
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
pins: Vec<std::sync::Arc<kevy_vlog::VlogFile>>,
/// Row-segment pins — the seg-backed stubs' serializer-thread
/// read path, same doctrine as the vlog pins.
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
seg_pins: Vec<(u32, std::sync::Arc<kevy_seg::Seg>)>,
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
seg_files: Vec<(u32, String)>,
}
// Compile-time guarantee that a view can cross to a serializer thread.
const _: () = {
const fn assert_send<T: Send>() {}
assert_send::<SnapshotView>();
};
impl SnapshotView {
/// Visit every entry as `(key, &value, ttl_ms)` — the same shape as
/// [`Store::snapshot_each`], so serializers take either source.
pub fn each<F: FnMut(&[u8], &Value, Option<u64>)>(&self, mut f: F) {
for (k, v, ttl) in &self.entries {
f(k.as_slice(), v, *ttl);
}
}
/// Visit the frozen hash field TTLs.
pub fn each_hash_ttl<F: FnMut(&[u8], &[u8], u64)>(&self, mut f: F) {
for (k, field, d) in &self.hfttl {
f(k.as_slice(), field.as_slice(), *d);
}
}
/// Number of entries frozen in the view.
pub fn len(&self) -> usize {
self.entries.len()
}
/// Whether the view holds zero entries.
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
/// Decode a cold stub's record against the view's pinned vlog files
/// into a fresh hot [`Value`] — the serializer-thread read path:
/// no store access, no promotion, memory bound = this one value.
/// `None` when `v` is hot. A stub naming an unpinned file, a failed
/// read, or a bad decode is a process bug (the vlog is per-boot and
/// this process pinned every file at collect time) — surfaced
/// loudly, never healed silently.
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
pub fn materialize_cold(&self, key: &[u8], v: &Value) -> Option<Value> {
let Value::Cold(c) = v else { return None };
if c.is_seg() {
let payload = self
.seg_pins
.iter()
.find(|(q, _)| *q == c.seg_ix())
.expect("segrows: view stub references a segment pinned at collect time")
.1
.get(key)
.expect("segrows: pinned segment read failed — refused, not healed")
.expect("segrows: stub points at a record the segment does not hold");
return Some(
crate::tier_codec::decode(c.type_tag, payload)
.expect("segrows: cold row decode failed — process bug"),
);
}
let file = self
.pins
.iter()
.find(|f| f.id() == c.file_id)
.expect("tier: view stub references a file pinned at collect time");
let (_key, payload) = file
.read(c.vref())
.expect("tier: pinned vlog read failed — per-boot spill file, this is a process bug");
Some(
crate::tier_codec::decode(c.type_tag, payload)
.expect("tier: cold record decode failed — process bug"),
)
}
/// No tier backend on this target — `Value::Cold` cannot exist.
#[cfg(not(all(feature = "std", not(target_arch = "wasm32"))))]
pub fn materialize_cold(&self, _key: &[u8], _v: &Value) -> Option<Value> {
None
}
/// The frozen row segments' `(seq, file)` identities.
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
pub fn row_seg_files(&self) -> Vec<(u32, String)> {
self.seg_files.clone()
}
#[cfg(not(all(feature = "std", not(target_arch = "wasm32"))))]
pub fn row_seg_files(&self) -> Vec<(u32, String)> {
Vec::new()
}
}
impl Store {
/// Freeze a point-in-time [`SnapshotView`] of every live entry.
///
/// O(n) shallow: per entry one key clone + one [`Value`] clone (string
/// bytes copied, collections refcount-bumped) + the TTL resolved to
/// remaining millis. Expired-but-unreaped entries are skipped, matching
/// [`Store::snapshot_each`].
pub fn collect_snapshot(&self) -> SnapshotView {
let now = now_ns();
let mut entries = Vec::with_capacity(self.map.len());
for (k, e) in &self.map {
if e.is_expired_at(now) {
continue;
}
let ttl = e.expire_at_ns.map(|ns| remaining_ms(ns, now));
entries.push((k.clone(), e.value.clone(), ttl));
}
let mut hfttl = Vec::new();
self.hash_ttl_each(|k, f, d| {
hfttl.push((
crate::SmallBytes::from_slice(k),
crate::SmallBytes::from_slice(f),
d,
));
});
SnapshotView {
entries,
hfttl,
// View pinning: capture ALL current vlog file pins with
// the view — the frozen stubs above can only reference
// files that exist at this instant.
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
pins: self.tier_pins(),
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
seg_pins: self.segrow_pins(),
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
seg_files: self.row_seg_files(),
}
}
}