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