kevy_store/tier_serve.rs
1//! The tiering READ funnel + the no-promote peek surface.
2//!
3//! Three read lanes come through here:
4//!
5//! - **Client reads** (`tier_serve`): the promotion gate — the first
6//! materializing access serves decoded bytes without installing
7//! (probation `touched` mark), the second promotes.
8//! - **Bulk whole-value reads** ([`Store::peek_scope`]): digest,
9//! scope-move, exports. Inside the scope every cold materializing
10//! read serves via pread WITHOUT setting the mark and WITHOUT
11//! promoting — a bulk sweep is not an access signal.
12//! - **Row-field reads** ([`Store::peek_hash_fields`] /
13//! [`Store::peek_hash_rows`]): hydration and index backfill. ONE
14//! record read + ONE decode per cold ROW extracts every requested
15//! field (never one pread per field); a page's cold rows coalesce —
16//! sorted by `(file_id, offset)` — into one [`ColdBatchReader`]
17//! batch (io_uring: one submit-and-wait for N READ SQEs; poller /
18//! embedded: an ordered `read_at` loop).
19//!
20//! Counter contract (asserted by the D1/D3 gates): `peek_preads_total`
21//! += 1 per cold ROW peeked; `batch_submissions_total` += the reader's
22//! kernel submission count per page with ≥1 cold row.
23
24#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
25mod enabled {
26 use std::sync::Arc;
27
28 use kevy_vlog::{VlogFile, VlogRef, verify_image};
29
30 use crate::value::{COLD_TAG_HASH, ColdRef, Value};
31 use crate::{Entry, Store, StoreError};
32 use kevy_bytes::SmallBytes;
33
34 /// One planned cold-record read in a [`Store::peek_hash_rows`]
35 /// batch. The pinned file keeps the record readable even if a
36 /// compaction retires the file mid-batch.
37 pub struct ColdRead {
38 /// Pinned vlog file the record lives in.
39 pub file: Arc<VlogFile>,
40 /// Record address; the image to fetch is `vref.disk_len()`
41 /// bytes at `vref.offset`.
42 pub vref: VlogRef,
43 }
44
45 /// The read-issuance half of a cold batch: fetch every record
46 /// image, in `reads` order. [`SyncColdRead`] is the ordered
47 /// positional-read loop (poller reactors + embedded); the server's
48 /// io_uring backend submits the batch to a secondary ring instead.
49 pub trait ColdBatchReader {
50 /// Fetch each `reads[i]`'s raw image (`vref.disk_len()` bytes
51 /// at `vref.offset`, unverified — the store runs
52 /// [`verify_image`] + decode on completion). Returns the images
53 /// plus the number of kernel submissions made (1 for the sync
54 /// loop, ceil(n / ring entries) for a ring).
55 fn read_batch(&mut self, reads: &[ColdRead]) -> std::io::Result<(Vec<Vec<u8>>, u64)>;
56 }
57
58 /// The default reader: one ordered `pread` per record.
59 pub struct SyncColdRead;
60
61 impl ColdBatchReader for SyncColdRead {
62 fn read_batch(&mut self, reads: &[ColdRead]) -> std::io::Result<(Vec<Vec<u8>>, u64)> {
63 let mut images = Vec::with_capacity(reads.len());
64 for r in reads {
65 images.push(r.file.read_image(r.vref)?);
66 }
67 Ok((images, 1))
68 }
69 }
70
71 /// One peeked row: the per-field values of a live hash
72 /// (`Ok(Some(..))`, one `Option` per requested field), a missing
73 /// key (`Ok(None)`), or a non-hash (`Err(WrongType)`).
74 pub type PeekRow = Result<Option<Vec<Option<Vec<u8>>>>, StoreError>;
75
76 /// Stage-1 verdict for one peeked key (zero IO — the stub's tag
77 /// answers WRONGTYPE without a pread).
78 enum Probe {
79 Missing,
80 WrongType,
81 Hot,
82 ColdHash(ColdRef),
83 }
84
85 /// Extract `fields` from a live hot entry, hmget-shaped. `Err` on a
86 /// non-hash; `Cold` never reaches here (callers resolve it first).
87 fn hot_hash_fields(
88 e: &Entry,
89 fields: &[&[u8]],
90 ) -> Result<Vec<Option<Vec<u8>>>, StoreError> {
91 match &e.value {
92 Value::Hash(h) => Ok(fields.iter().map(|f| h.get(*f).map(SmallBytes::to_vec)).collect()),
93 Value::SegHash(h) => {
94 Ok(fields.iter().map(|f| h.get(f).map(SmallBytes::to_vec)).collect())
95 }
96 Value::SmallHashInline(h) => {
97 Ok(fields.iter().map(|f| h.get(f).map(<[u8]>::to_vec)).collect())
98 }
99 _ => Err(StoreError::WrongType),
100 }
101 }
102
103 impl Store {
104 /// Stage-2 funnel for WRITE paths: a live Cold entry whose tag
105 /// matches `want` is promoted in place; a mismatch is WRONGTYPE
106 /// with zero preads. Hot values / absent keys pass through.
107 pub(crate) fn tier_resolve(&mut self, key: &[u8], want: u8) -> Result<(), StoreError> {
108 if !self.cold_backing {
109 return Ok(());
110 }
111 let tag = match self.live_entry(key) {
112 Some(Entry { value: Value::Cold(c), .. }) => c.type_tag,
113 _ => return Ok(()),
114 };
115 if tag != want {
116 return Err(StoreError::WrongType);
117 }
118 self.promote_in_place(key);
119 Ok(())
120 }
121
122 /// Stage-2 funnel for READ paths — `live_entry` plus the
123 /// promotion gate. On a live Cold entry with a matching tag:
124 /// first materializing access decodes into the serve scratch
125 /// (no install, probation mark set); the second promotes. A tag
126 /// mismatch is WRONGTYPE with zero preads. Hot/absent =
127 /// `live_entry` verbatim. Inside [`Store::peek_scope`] the gate
128 /// is bypassed: serve via scratch, mark untouched, no promote.
129 pub(crate) fn tier_serve(&mut self, key: &[u8], want: u8) -> Result<Option<&Entry>, StoreError> {
130 if !self.cold_backing {
131 return Ok(self.live_entry(key));
132 }
133 let cold = match self.live_entry(key) {
134 None => return Ok(None),
135 Some(e) => match &e.value {
136 Value::Cold(c) => Some((c.type_tag, c.touched != 0)),
137 _ => None,
138 },
139 };
140 match cold {
141 None => Ok(self.live_entry(key)),
142 Some((tag, _)) if tag != want => Err(StoreError::WrongType),
143 Some(_) if self.tier_peek => self.tier_serve_cold(key, true),
144 Some((_, true)) => {
145 self.promote_in_place(key);
146 Ok(self.live_entry(key))
147 }
148 Some((_, false)) => self.tier_serve_cold(key, false),
149 }
150 }
151
152 /// Cold serve without installing: decode the record into the
153 /// scratch entry and hand a reference to it. The scratch
154 /// mirrors the live entry's TTL so callers that read
155 /// `expire_at_ns` behave identically. `peek = false` is the
156 /// gate's first touch (probation mark set); `peek = true` is a
157 /// bulk read (mark untouched, peek pread counted).
158 fn tier_serve_cold(&mut self, key: &[u8], peek: bool) -> Result<Option<&Entry>, StoreError> {
159 let (cref, expire) = {
160 let e = self.map.get_mut(key).expect("probed live above");
161 let Value::Cold(c) = &mut e.value else { unreachable!("cold checked above") };
162 if !peek {
163 c.touched = 1;
164 }
165 (*c, e.expire_at_ns)
166 };
167 let value = self.tier_read_record(key, cref);
168 if peek
169 && !cref.is_seg()
170 && let Some(t) = self.tier.as_mut()
171 {
172 t.peek_preads_total += 1;
173 }
174 let mut entry = Entry::new(value, None);
175 entry.expire_at_ns = expire;
176 entry.set_weight(u64::from(cref.weight));
177 self.tier_scratch = Some(entry);
178 Ok(self.tier_scratch.as_ref())
179 }
180
181 /// Read + decode one cold record (bumps the pread counter). A
182 /// vlog read/decode failure is a process bug by the vlog's
183 /// per-boot doctrine — surfaced loudly, never healed silently.
184 pub(crate) fn tier_read_record(&mut self, key: &[u8], cref: ColdRef) -> Value {
185 if cref.is_seg() {
186 return self.segrow_read(cref, key);
187 }
188 let t = self.tier.as_mut().expect("tier enabled");
189 t.preads_total += 1;
190 let (_key, payload) = t
191 .vlog
192 .read(cref.vref())
193 .expect("tier: vlog read failed — per-boot spill file, this is a process bug");
194 crate::tier_codec::decode(cref.type_tag, payload)
195 .expect("tier: cold record decode failed — process bug")
196 }
197
198 /// `&self` peek for the zero-copy shared lane and COPY: decode
199 /// a fresh owned value from the record WITHOUT installing,
200 /// promoting, or setting the probation mark (documented: the
201 /// shared lane pays a pread until a `&mut`-path access
202 /// promotes). `None` when the value is not Cold. (Counter-free:
203 /// the shared lane is `&self`; the `&mut` peeks carry the
204 /// counters the gates assert on.)
205 pub(crate) fn tier_peek_value(&self, key: &[u8], v: &Value) -> Option<Value> {
206 let Value::Cold(c) = v else { return None };
207 if c.is_seg() {
208 return Some(self.segrow_read(*c, key));
209 }
210 let t = self.tier.as_ref().expect("cold value ⇒ tiering on");
211 let (_key, payload) = t
212 .vlog
213 .read(c.vref())
214 .expect("tier: vlog read failed — per-boot spill file, this is a process bug");
215 Some(
216 crate::tier_codec::decode(c.type_tag, payload)
217 .expect("tier: cold record decode failed — process bug"),
218 )
219 }
220
221 /// Run `f` in bulk-read (no-promote peek) mode: every cold
222 /// materializing read inside serves via pread WITHOUT setting
223 /// the probation mark and WITHOUT promoting. The whole-value
224 /// peek for digest / scope-move / export sweeps — a bulk
225 /// sweep must never thrash the hot tier.
226 pub fn peek_scope<R>(&mut self, f: impl FnOnce(&mut Self) -> R) -> R {
227 let prev = self.tier_peek;
228 self.tier_peek = true;
229 let r = f(self);
230 self.tier_peek = prev;
231 r
232 }
233
234 /// Row peek: `fields` of the hash at `key`, without
235 /// promotion and without advancing the touched gate. A hot row
236 /// reads as `hmget` does; a COLD hash stub costs ONE record
237 /// read + ONE decode for all fields. `Ok(None)` = missing key;
238 /// `Err(WrongType)` = non-hash (zero preads when cold — the
239 /// stage-1 tag answers).
240 pub fn peek_hash_fields(
241 &mut self,
242 key: &[u8],
243 fields: &[&[u8]],
244 ) -> PeekRow {
245 self.purge_hash_ttl(key);
246 match self.peek_probe(key) {
247 Probe::Missing => Ok(None),
248 Probe::WrongType => Err(StoreError::WrongType),
249 Probe::Hot => {
250 let e = self.live_entry(key).expect("probed live above");
251 Ok(Some(hot_hash_fields(e, fields)?))
252 }
253 Probe::ColdHash(cref) => {
254 let value = self.tier_read_record(key, cref);
255 if !cref.is_seg()
256 && let Some(t) = self.tier.as_mut()
257 {
258 t.peek_preads_total += 1;
259 }
260 let Value::Hash(h) = &value else {
261 unreachable!("hash-tagged record decodes to a hash")
262 };
263 Ok(Some(fields.iter().map(|f| h.get(*f).map(SmallBytes::to_vec)).collect()))
264 }
265 }
266 }
267
268 /// The seg-backed half of a cold plan, served synchronously
269 /// (one fence descent each — batching belongs to the vlog's
270 /// pread model). Returns the vlog-backed remainder.
271 fn serve_seg_rows(
272 &mut self,
273 out: &mut [PeekRow],
274 plan: Vec<(usize, ColdRef)>,
275 fields: &[&[u8]],
276 keys: &[&[u8]],
277 ) -> Vec<(usize, ColdRef)> {
278 let mut kept = Vec::with_capacity(plan.len());
279 for (row, c) in plan {
280 if !c.is_seg() {
281 kept.push((row, c));
282 continue;
283 }
284 let value = self.segrow_read(c, keys[row]);
285 let Value::Hash(h) = &value else {
286 unreachable!("hash-tagged record decodes to a hash")
287 };
288 out[row] = Ok(Some(fields.iter().map(|f| h.get(*f).map(SmallBytes::to_vec)).collect()));
289 }
290 kept
291 }
292
293 /// Stage-1 classification for the peeks — the same
294 /// probe-then-act two-phase shape `tier_serve` uses (borrow
295 /// split; the second `live_entry` on the hot arm mirrors it).
296 fn peek_probe(&mut self, key: &[u8]) -> Probe {
297 match self.live_entry(key) {
298 None => Probe::Missing,
299 Some(e) => match &e.value {
300 Value::Cold(c) if c.type_tag == COLD_TAG_HASH => Probe::ColdHash(*c),
301 Value::Cold(_) => Probe::WrongType,
302 Value::Hash(_) | Value::SegHash(_) | Value::SmallHashInline(_) => Probe::Hot,
303 _ => Probe::WrongType,
304 },
305 }
306 }
307
308 /// Page peek: `keys` × `fields` with every cold row
309 /// coalesced — sorted by `(file_id, offset)` — into ONE
310 /// [`ColdBatchReader`] batch, decoded once per row, results in
311 /// input order. Per-key result mirrors
312 /// [`Store::peek_hash_fields`]. No promotion, no gate
313 /// advancement; hot rows read exactly as `hmget` does.
314 pub fn peek_hash_rows(
315 &mut self,
316 keys: &[&[u8]],
317 fields: &[&[u8]],
318 reader: &mut dyn ColdBatchReader,
319 ) -> Vec<PeekRow> {
320 let mut out = Vec::with_capacity(keys.len());
321 let mut plan: Vec<(usize, ColdRef)> = Vec::new();
322 for (i, key) in keys.iter().enumerate() {
323 self.purge_hash_ttl(key);
324 match self.peek_probe(key) {
325 Probe::Missing => out.push(Ok(None)),
326 Probe::WrongType => out.push(Err(StoreError::WrongType)),
327 Probe::Hot => {
328 let e = self.live_entry(key).expect("probed live above");
329 out.push(hot_hash_fields(e, fields).map(Some));
330 }
331 Probe::ColdHash(cref) => {
332 plan.push((i, cref));
333 out.push(Ok(None)); // patched after the batch
334 }
335 }
336 }
337 if !plan.is_empty() {
338 self.peek_cold_batch(&mut out, plan, fields, reader, keys);
339 }
340 out
341 }
342
343 /// The cold half of [`Store::peek_hash_rows`]: sort, batch-read
344 /// through `reader`, verify + decode each record ONCE, extract
345 /// all fields, patch results back in original row order.
346 fn peek_cold_batch(
347 &mut self,
348 out: &mut [PeekRow],
349 plan: Vec<(usize, ColdRef)>,
350 fields: &[&[u8]],
351 reader: &mut dyn ColdBatchReader,
352 keys: &[&[u8]],
353 ) {
354 let mut plan = self.serve_seg_rows(out, plan, fields, keys);
355 if plan.is_empty() {
356 return;
357 }
358 plan.sort_by_key(|(_, c)| (c.file_id, c.offset));
359 let t = self.tier.as_mut().expect("cold value ⇒ tiering on");
360 let reads: Vec<ColdRead> = plan
361 .iter()
362 .map(|(_, c)| ColdRead {
363 file: t.vlog.pin(c.file_id).expect("live stub names a live file"),
364 vref: c.vref(),
365 })
366 .collect();
367 let (images, submissions) = reader
368 .read_batch(&reads)
369 .expect("tier: vlog batch read failed — per-boot spill file, this is a process bug");
370 assert_eq!(images.len(), reads.len(), "reader must return one image per read");
371 t.preads_total += plan.len() as u64;
372 t.peek_preads_total += plan.len() as u64;
373 t.batch_submissions_total += submissions;
374 for (((row, cref), image), read) in plan.into_iter().zip(images).zip(&reads) {
375 let (_key, frame) = verify_image(cref.vref(), image)
376 .expect("tier: cold record image verify failed — process bug");
377 let payload = read
378 .file
379 .decompress(&frame)
380 .expect("tier: cold record decompress failed — process bug");
381 let value = crate::tier_codec::decode(cref.type_tag, payload)
382 .expect("tier: cold record decode failed — process bug");
383 let Value::Hash(h) = &value else {
384 unreachable!("hash-tagged record decodes to a hash")
385 };
386 out[row] = Ok(Some(fields.iter().map(|f| h.get(*f).map(SmallBytes::to_vec)).collect()));
387 }
388 }
389 }
390}
391
392#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
393pub use enabled::{ColdBatchReader, ColdRead, PeekRow, SyncColdRead};
394
395/// Funnel/peek passthroughs for builds without the tier backend
396/// (no_std / wasm): `Value::Cold` cannot be constructed there, so the
397/// funnel degenerates to `live_entry` and the peeks to plain hot reads.
398#[cfg(not(all(feature = "std", not(target_arch = "wasm32"))))]
399mod disabled {
400 use alloc::vec::Vec;
401
402 use crate::value::Value;
403 use crate::{Entry, Store, StoreError};
404 use kevy_bytes::SmallBytes;
405
406 impl Store {
407 #[inline]
408 pub(crate) fn tier_resolve(&mut self, _key: &[u8], _want: u8) -> Result<(), StoreError> {
409 Ok(())
410 }
411
412 #[inline]
413 pub(crate) fn tier_serve(&mut self, key: &[u8], _want: u8) -> Result<Option<&Entry>, StoreError> {
414 Ok(self.live_entry(key))
415 }
416
417 #[inline]
418 pub(crate) fn tier_peek_value(&self, _key: &[u8], _v: &Value) -> Option<Value> {
419 None
420 }
421
422 /// No tier backend on this target — `f` runs with nothing to
423 /// peek past.
424 #[inline]
425 pub fn peek_scope<R>(&mut self, f: impl FnOnce(&mut Self) -> R) -> R {
426 f(self)
427 }
428
429 /// No tier backend on this target — the plain hot-row read.
430 pub fn peek_hash_fields(
431 &mut self,
432 key: &[u8],
433 fields: &[&[u8]],
434 ) -> Result<Option<Vec<Option<Vec<u8>>>>, StoreError> {
435 self.purge_hash_ttl(key);
436 match self.live_entry(key) {
437 None => Ok(None),
438 Some(e) => match &e.value {
439 Value::Hash(h) => {
440 Ok(Some(fields.iter().map(|f| h.get(*f).map(SmallBytes::to_vec)).collect()))
441 }
442 Value::SegHash(h) => {
443 Ok(Some(fields.iter().map(|f| h.get(f).map(SmallBytes::to_vec)).collect()))
444 }
445 Value::SmallHashInline(h) => {
446 Ok(Some(fields.iter().map(|f| h.get(f).map(<[u8]>::to_vec)).collect()))
447 }
448 _ => Err(StoreError::WrongType),
449 },
450 }
451 }
452 }
453}