rindle_sqlite/
storage.rs

1//! Spec `10` — **SQLite-backed operator scratch state** (the server spill path).
2//!
3//! This is the `#[cfg(feature = "sqlite")]` half of the former `rindle::storage` module,
4//! lifted into `rindle-sqlite`: `DatabaseStorage` / `OpStorage` over rusqlite, plus the
5//! `StorageValue` blob codec. The backend-agnostic `Storage` / `StorageValue` /
6//! `StorageProvider` / `MemoryStorage` / `StorageFactory` types stay in `rindle::storage`.
7//!
8//! Each engine gets its own [`DatabaseStorage`] (a private temp-file DB), so a single
9//! database hosts exactly one set of operators. Every operator is handed a unique,
10//! auto-incrementing `op_id` and all of them share one `storage(op, key) -> val` table —
11//! no per-tenant/client-group column is needed. `OpStorage` implements
12//! [`rindle::storage::Storage`]; `DatabaseStorage` implements
13//! [`rindle::storage::StorageProvider`], so a graph built with
14//! `StorageFactory::custom(Rc::new(database_storage))` spills operator state to SQLite.
15
16use std::cell::{Cell, RefCell};
17use std::rc::Rc;
18use std::sync::Arc;
19
20use rusqlite::{params, OptionalExtension};
21
22use rindle::storage::{ReduceAcc, Storage, StorageProvider, StorageValue};
23use rindle::value::{owned_row, OwnedRow, OwnedValue};
24use rindle::RindleError;
25
26// ---------------------------------------------------------------------------
27// DatabaseStorage (server) — spec 10 §4.4 / §5.2
28// ---------------------------------------------------------------------------
29
30#[derive(Clone, Copy, Debug)]
31pub struct DatabaseStorageOptions {
32    /// Number of point operations (`set`/`get`/`del`, matching the JS behavior)
33    /// before checkpointing the current scratch-state transaction.
34    pub commit_interval: u32,
35    /// Reserved for the production incremental-vacuum threshold. The current
36    /// implementation clears rows correctly but does not run compaction yet.
37    pub compaction_threshold_bytes: u64,
38}
39
40impl Default for DatabaseStorageOptions {
41    fn default() -> DatabaseStorageOptions {
42        DatabaseStorageOptions {
43            commit_interval: 5_000,
44            compaction_threshold_bytes: 50 * 1024 * 1024,
45        }
46    }
47}
48
49#[derive(Clone)]
50pub struct DatabaseStorage {
51    inner: Rc<DatabaseStorageInner>,
52}
53
54struct DatabaseStorageInner {
55    conn: rusqlite::Connection,
56    opts: DatabaseStorageOptions,
57    num_writes: Cell<u32>,
58    /// Monotonic operator-id allocator. Each operator in this database gets a unique
59    /// `op_id` (its keyspace inside the shared `storage` table); ids are never reused,
60    /// so a torn-down-then-rebuilt operator can never collide with its former rows.
61    next_op_id: Cell<i64>,
62}
63
64impl DatabaseStorage {
65    pub fn new_in_memory() -> rusqlite::Result<DatabaseStorage> {
66        DatabaseStorage::new(rusqlite::Connection::open_in_memory()?)
67    }
68
69    /// Open operator storage over a **private, on-disk temporary database** — SQLite's
70    /// `""` filename: a per-connection scratch DB created in the temp directory and
71    /// deleted automatically when the connection closes. This is the server **spill
72    /// path**: operator state can grow past RAM and overflow to disk, but it keeps the
73    /// durability-free scratch pragmas (`journal_mode = OFF`, `synchronous = OFF`,
74    /// `locking_mode = EXCLUSIVE`), so there are no journal writes or `fsync`s on the
75    /// hot path — pages only reach disk when the OS page cache evicts them.
76    pub fn new_temp_file() -> rusqlite::Result<DatabaseStorage> {
77        DatabaseStorage::new(rusqlite::Connection::open("")?)
78    }
79
80    pub fn new(conn: rusqlite::Connection) -> rusqlite::Result<DatabaseStorage> {
81        DatabaseStorage::with_options(conn, DatabaseStorageOptions::default())
82    }
83
84    pub fn with_options(
85        conn: rusqlite::Connection,
86        opts: DatabaseStorageOptions,
87    ) -> rusqlite::Result<DatabaseStorage> {
88        // Query Planner Stability Guarantee (see rindle-replica `parallel::open_wal2`): the shared
89        // `storage(op, key) -> val` table is hit by cached point ops re-bound with a fresh key on
90        // every set/get/del. QPSG keeps their plans value-independent so a re-bound seek is never
91        // re-parsed by this bedrock-SQLite build. Set at open, before `init` prepares anything or
92        // opens its `BEGIN`.
93        conn.set_db_config(
94            rusqlite::config::DbConfig::SQLITE_DBCONFIG_ENABLE_QPSG,
95            true,
96        )?;
97        let db = DatabaseStorage {
98            inner: Rc::new(DatabaseStorageInner {
99                conn,
100                opts,
101                num_writes: Cell::new(0),
102                next_op_id: Cell::new(1),
103            }),
104        };
105        db.init()?;
106        Ok(db)
107    }
108
109    /// Allocate one operator's isolated keyspace (a fresh `op_id`) over this database.
110    /// The store gets its own (undrained) error sink — for direct use outside a graph
111    /// (tests, benchmarks). Inside a graph the [`StorageProvider`] seam threads the
112    /// graph's `runtime_error` sink in via [`create_storage_with_sink`].
113    ///
114    /// [`create_storage_with_sink`]: rindle::storage::StorageProvider::create_storage_with_sink
115    pub fn create_storage(&self) -> OpStorage {
116        self.op_storage_with_sink(Rc::default())
117    }
118
119    fn op_storage_with_sink(&self, error_sink: Rc<RefCell<Option<RindleError>>>) -> OpStorage {
120        let op_id = self.inner.next_op_id.get();
121        self.inner.next_op_id.set(op_id + 1);
122        OpStorage {
123            db: self.clone(),
124            op_id,
125            error_sink,
126        }
127    }
128
129    pub fn checkpoint(&self) -> rusqlite::Result<()> {
130        self.conn().execute_batch("COMMIT; BEGIN;")?;
131        self.inner.num_writes.set(0);
132        Ok(())
133    }
134
135    fn conn(&self) -> &rusqlite::Connection {
136        &self.inner.conn
137    }
138
139    fn init(&self) -> rusqlite::Result<()> {
140        self.conn().execute_batch(
141            r#"
142            PRAGMA journal_mode = OFF;
143            PRAGMA synchronous = OFF;
144            PRAGMA temp_store = MEMORY;
145            PRAGMA locking_mode = EXCLUSIVE;
146            CREATE TABLE IF NOT EXISTS storage (
147                "op"  INTEGER NOT NULL,
148                "key" TEXT NOT NULL,
149                "val" BLOB NOT NULL,
150                PRIMARY KEY("op", "key")
151            ) WITHOUT ROWID;
152            BEGIN;
153            "#,
154        )
155    }
156
157    fn maybe_checkpoint(&self) -> rusqlite::Result<()> {
158        let interval = self.inner.opts.commit_interval;
159        if interval == 0 {
160            return Ok(());
161        }
162        let next = self.inner.num_writes.get().saturating_add(1);
163        if next >= interval {
164            self.checkpoint()?;
165        } else {
166            self.inner.num_writes.set(next);
167        }
168        Ok(())
169    }
170}
171
172impl Drop for DatabaseStorageInner {
173    fn drop(&mut self) {
174        let _ = self.conn.execute_batch("COMMIT;");
175    }
176}
177
178/// The graph's `StorageFactory` vends one [`OpStorage`] per stateful operator straight
179/// off the database, threading the graph's `runtime_error` sink in (WS02.3) so storage
180/// failures surface via `take_runtime_error` at the mutation boundary instead of aborting.
181impl StorageProvider for DatabaseStorage {
182    fn create_storage_with_sink(
183        &self,
184        error_sink: Rc<RefCell<Option<RindleError>>>,
185    ) -> Box<dyn Storage> {
186        Box::new(self.op_storage_with_sink(error_sink))
187    }
188}
189
190/// One operator's namespaced scratch map inside the shared SQLite `storage` table,
191/// keyed by this operator's unique `op_id`.
192pub struct OpStorage {
193    db: DatabaseStorage,
194    op_id: i64,
195    /// Where transient SQLite / decode errors are parked (WS02.3, spec 10 E14): the
196    /// `Storage` trait is infallible by design, so an op returns a safe sentinel
197    /// (None / empty / no-op) and parks the typed error here. The graph drains it via
198    /// `take_runtime_error` at the mutation boundary, so the process never aborts.
199    error_sink: Rc<RefCell<Option<RindleError>>>,
200}
201
202impl OpStorage {
203    /// Park a storage error (set-if-empty, first failure wins). The infallible
204    /// `Storage` method that called this then returns its safe sentinel.
205    fn park(&self, err: RindleError) {
206        let mut slot = self.error_sink.borrow_mut();
207        if slot.is_none() {
208            *slot = Some(err);
209        }
210    }
211}
212
213impl Storage for OpStorage {
214    fn set(&self, key: &str, value: StorageValue) {
215        if let Err(e) = self.db.maybe_checkpoint() {
216            self.park(RindleError::sqlite("checkpoint operator storage", e));
217            return;
218        }
219        let bytes = encode_storage_value(&value);
220        // `prepare_cached` so the steady-state hot path (one `set` per operator per push)
221        // reuses a compiled statement instead of re-parsing this SQL every call.
222        let set = || -> rusqlite::Result<()> {
223            self.db
224                .conn()
225                .prepare_cached(
226                    r#"
227                    INSERT INTO storage ("op", "key", "val")
228                    VALUES (?1, ?2, ?3)
229                    ON CONFLICT("op", "key")
230                    DO UPDATE SET "val" = excluded."val"
231                    "#,
232                )?
233                .execute(params![self.op_id, key, bytes])?;
234            Ok(())
235        };
236        if let Err(e) = set() {
237            self.park(RindleError::sqlite("set operator storage value", e));
238        }
239    }
240
241    fn get(&self, key: &str) -> Option<StorageValue> {
242        if let Err(e) = self.db.maybe_checkpoint() {
243            self.park(RindleError::sqlite("checkpoint operator storage", e));
244            return None;
245        }
246        let query = || -> rusqlite::Result<Option<Vec<u8>>> {
247            self.db
248                .conn()
249                .prepare_cached(
250                    r#"
251                    SELECT "val"
252                    FROM storage
253                    WHERE "op" = ?1 AND "key" = ?2
254                    "#,
255                )?
256                .query_row(params![self.op_id, key], |r| r.get(0))
257                .optional()
258        };
259        let bytes = match query() {
260            Ok(b) => b,
261            Err(e) => {
262                self.park(RindleError::sqlite("get operator storage value", e));
263                return None;
264            }
265        };
266        match bytes {
267            Some(b) => match decode_storage_value(&b) {
268                Ok(v) => Some(v),
269                Err(msg) => {
270                    self.park(RindleError::Storage(format!(
271                        "decode operator storage value: {msg}"
272                    )));
273                    None
274                }
275            },
276            None => None,
277        }
278    }
279
280    fn del(&self, key: &str) {
281        if let Err(e) = self.db.maybe_checkpoint() {
282            self.park(RindleError::sqlite("checkpoint operator storage", e));
283            return;
284        }
285        let del = || -> rusqlite::Result<()> {
286            self.db
287                .conn()
288                .prepare_cached(
289                    r#"
290                    DELETE FROM storage
291                    WHERE "op" = ?1 AND "key" = ?2
292                    "#,
293                )?
294                .execute(params![self.op_id, key])?;
295            Ok(())
296        };
297        if let Err(e) = del() {
298            self.park(RindleError::sqlite("delete operator storage value", e));
299        }
300    }
301
302    fn scan<'s>(&'s self, prefix: &str) -> Box<dyn Iterator<Item = (Box<str>, StorageValue)> + 's> {
303        // Checkpoint once up front — a scan does no writes, so there is nothing to
304        // re-check between pages. A failure parks and yields an empty scan, as before.
305        // The actual rows are then pulled lazily, a bounded page at a time, by
306        // `ScanCursor` rather than slurped into one `Vec` here.
307        if let Err(e) = self.db.maybe_checkpoint() {
308            self.park(RindleError::sqlite("checkpoint operator storage", e));
309            return Box::new(std::iter::empty());
310        }
311        Box::new(ScanCursor::new(self, prefix))
312    }
313
314    /// Drop this operator's whole namespace in one `DELETE` (the
315    /// [`Storage::clear`] fast path used by pipeline teardown), rather than the
316    /// default per-key scan+del. `op_id`s are never reused, so this only reclaims disk
317    /// — a rebuilt operator gets a fresh `op_id` and so an already-empty keyspace.
318    ///
319    /// Unlike the point ops, `clear` does NOT gate on `maybe_checkpoint`: a bulk delete
320    /// only shrinks the open transaction (it runs in the standing `BEGIN`), so there is
321    /// no checkpoint-failure path that could leave rows behind. Only the `DELETE` itself
322    /// can fail, and it parks like the others.
323    fn clear(&self) {
324        if let Err(e) = self.db.conn().execute(
325            r#"DELETE FROM storage WHERE "op" = ?1"#,
326            params![self.op_id],
327        ) {
328            self.park(RindleError::sqlite("clear operator storage", e));
329        }
330    }
331}
332
333/// How many `(key, value)` pairs a [`ScanCursor`] pulls from SQLite per round trip.
334/// `scan` used to materialize the *entire* matching range into one `Vec`; an operator over
335/// a large keyspace (a `reduce`/`cap` enumerating every group) could pin that whole range in
336/// RAM at once. The cursor pages the statement instead, buffering at most this many rows —
337/// bounded memory however big the operator's namespace has grown. Scratch scans are usually
338/// tiny (a handful of partition keys, spec 10 §5.1), so the common case is still a single
339/// round trip; only genuinely large scans pay for the extra seeks.
340const SCAN_PAGE: usize = 1024;
341
342/// Lazy, bounded-memory [`Storage::scan`] over `OpStorage`: an ascending prefix scan that
343/// fetches `SCAN_PAGE` rows per round trip and re-seeks with `"key" > last` for the next
344/// page, instead of reading the whole range into a `Vec` up front.
345///
346/// Keyset (not `OFFSET`) pagination is what lets the cursor honor the `scan` contract that
347/// the caller may `set`/`del` *while draining* (spec 10 §5.1, the reason `MemoryStorage`
348/// snapshots into a `Vec`): each page is a self-contained statement, dropped before a pair
349/// is yielded, so no read is pinned across the caller's loop. A concurrent write — which can
350/// `COMMIT; BEGIN` the standing txn via `maybe_checkpoint` — just lands the next page in a
351/// fresh snapshot that resumes strictly after `seek`; there is no live cursor to invalidate.
352struct ScanCursor<'s> {
353    store: &'s OpStorage,
354    /// Owned because the `prefix: &str` argument does not outlive the returned iterator.
355    /// Bounds the scan: paging stops at the first key that does not start with it.
356    prefix: Box<str>,
357    /// Exclusive lower bound for the *next* page (the last key already yielded). `None`
358    /// before the first page, whose bound is instead the inclusive `"key" >= prefix` start.
359    seek: Option<Box<str>>,
360    /// The current page, drained one pair per `next()` before the next page is fetched.
361    page: std::vec::IntoIter<(Box<str>, StorageValue)>,
362    /// Set once the range is exhausted (a short page), the prefix is passed, or an error is
363    /// parked — after which no further page is fetched.
364    done: bool,
365}
366
367impl<'s> ScanCursor<'s> {
368    fn new(store: &'s OpStorage, prefix: &str) -> ScanCursor<'s> {
369        ScanCursor {
370            store,
371            prefix: prefix.into(),
372            seek: None,
373            page: Vec::new().into_iter(),
374            done: false,
375        }
376    }
377
378    /// Fetch the next page into `self.page`, advancing `self.seek` (to the last key read) and
379    /// `self.done` (once the scan is finished). The SQLite work is delegated to `query_page`
380    /// so no borrow of `self` is held across these field updates.
381    fn fetch_page(&mut self) {
382        let (page, finished) =
383            Self::query_page(self.store, self.prefix.as_ref(), self.seek.as_deref());
384        if finished {
385            self.done = true;
386        }
387        // Resume the next page strictly after the last key we accepted.
388        if let Some((last_key, _)) = page.last() {
389            self.seek = Some(last_key.clone());
390        }
391        self.page = page.into_iter();
392    }
393
394    /// Read up to `SCAN_PAGE` `(key, value)` pairs, ordered ascending, with `"key"` bounded
395    /// below by `seek` (exclusive) or — on the first page (`seek == None`) — by `prefix`
396    /// (inclusive). Returns the decoded pairs and whether the scan is now finished: a short
397    /// page (range exhausted), the first key past the prefix, or a parked error all end it.
398    /// Any pairs decoded before an error are still returned (matching the old eager scan,
399    /// which yielded everything decoded before the failing row).
400    fn query_page(
401        store: &OpStorage,
402        prefix: &str,
403        seek: Option<&str>,
404    ) -> (Vec<(Box<str>, StorageValue)>, bool) {
405        let conn = store.db.conn();
406        // The first page seeks inclusively from the prefix; later pages strictly past the
407        // last key already yielded. Two cached statements (one per bound), reused per page.
408        let (sql, lower) = match seek {
409            None => (
410                r#"
411                SELECT "key", "val"
412                FROM storage
413                WHERE "op" = ?1 AND "key" >= ?2
414                ORDER BY "key"
415                LIMIT ?3
416                "#,
417                prefix,
418            ),
419            Some(last) => (
420                r#"
421                SELECT "key", "val"
422                FROM storage
423                WHERE "op" = ?1 AND "key" > ?2
424                ORDER BY "key"
425                LIMIT ?3
426                "#,
427                last,
428            ),
429        };
430        let mut stmt = match conn.prepare_cached(sql) {
431            Ok(s) => s,
432            Err(e) => {
433                store.park(RindleError::sqlite("prepare scan operator storage", e));
434                return (Vec::new(), true);
435            }
436        };
437        let rows = match stmt.query_map(params![store.op_id, lower, SCAN_PAGE as i64], |r| {
438            Ok((r.get::<_, String>(0)?, r.get::<_, Vec<u8>>(1)?))
439        }) {
440            Ok(r) => r,
441            Err(e) => {
442                store.park(RindleError::sqlite("scan operator storage", e));
443                return (Vec::new(), true);
444            }
445        };
446        let mut out = Vec::new();
447        let mut fetched = 0usize;
448        for row in rows {
449            fetched += 1;
450            let (key, bytes) = match row {
451                Ok(kv) => kv,
452                Err(e) => {
453                    store.park(RindleError::sqlite("read scan row", e));
454                    return (out, true);
455                }
456            };
457            if !key.starts_with(prefix) {
458                return (out, true);
459            }
460            match decode_storage_value(&bytes) {
461                Ok(value) => out.push((key.into_boxed_str(), value)),
462                Err(msg) => {
463                    store.park(RindleError::Storage(format!(
464                        "decode operator storage scan value: {msg}"
465                    )));
466                    return (out, true);
467                }
468            }
469        }
470        // A short page (fewer rows than the limit) means the range is exhausted.
471        (out, fetched < SCAN_PAGE)
472    }
473}
474
475impl Iterator for ScanCursor<'_> {
476    type Item = (Box<str>, StorageValue);
477
478    fn next(&mut self) -> Option<(Box<str>, StorageValue)> {
479        loop {
480            if let Some(pair) = self.page.next() {
481                return Some(pair);
482            }
483            if self.done {
484                return None;
485            }
486            // `fetch_page` either refills `self.page` or sets `self.done`; loop to drain the
487            // refilled page, or to return `None` now that the scan is finished.
488            self.fetch_page();
489        }
490    }
491}
492
493/// On-disk operator-state codec version (WS09.3). Prepended to every blob *before*
494/// the variant tag. The spill is a derived cache (never the source of truth), so the
495/// migration policy on a version mismatch is **discard and re-hydrate from source**,
496/// not in-place upgrade — bumping this is cheap precisely because recovery is rebuild.
497/// v3: `ReduceAcc.int_sum` widened i64 → i128 (design 226 §5.3).
498const STORAGE_FORMAT_VERSION: u8 = 3;
499
500fn encode_storage_value(value: &StorageValue) -> Vec<u8> {
501    let mut out = Vec::new();
502    out.push(STORAGE_FORMAT_VERSION);
503    match value {
504        StorageValue::Take { size, bound } => {
505            out.push(1);
506            put_u32(&mut out, *size);
507            match bound {
508                None => out.push(0),
509                Some(row) => {
510                    out.push(1);
511                    put_row(&mut out, row);
512                }
513            }
514        }
515        StorageValue::Bound(row) => {
516            out.push(2);
517            put_row(&mut out, row);
518        }
519        StorageValue::Cap { size, pks } => {
520            out.push(3);
521            put_u32(&mut out, *size);
522            put_len(&mut out, pks.len());
523            for pk in pks {
524                put_bytes(&mut out, pk.as_bytes());
525            }
526        }
527        StorageValue::Reduce { count, accs } => {
528            out.push(4);
529            put_i64(&mut out, *count);
530            put_len(&mut out, accs.len());
531            for acc in accs {
532                put_i128(&mut out, acc.int_sum);
533                put_f64(&mut out, acc.float_sum);
534                put_i64(&mut out, acc.non_null);
535                put_i64(&mut out, acc.float_count);
536            }
537        }
538    }
539    out
540}
541
542fn decode_storage_value(bytes: &[u8]) -> Result<StorageValue, String> {
543    let mut d = Decoder::new(bytes);
544    let version = d.u8()?;
545    if version != STORAGE_FORMAT_VERSION {
546        return Err(format!(
547            "operator state format v{version} unsupported (engine expects v{STORAGE_FORMAT_VERSION}); rebuild from source"
548        ));
549    }
550    let value = match d.u8()? {
551        1 => {
552            let size = d.u32()?;
553            let bound = match d.u8()? {
554                0 => None,
555                1 => Some(d.row()?),
556                _ => return Err("invalid Take bound tag".into()),
557            };
558            StorageValue::Take { size, bound }
559        }
560        2 => StorageValue::Bound(d.row()?),
561        3 => {
562            let size = d.u32()?;
563            let n = d.u32()? as usize;
564            let mut pks = Vec::with_capacity(n);
565            for _ in 0..n {
566                pks.push(d.string()?.into_boxed_str());
567            }
568            StorageValue::Cap { size, pks }
569        }
570        4 => {
571            let count = d.i64()?;
572            let n = d.u32()? as usize;
573            let mut accs = Vec::with_capacity(n);
574            for _ in 0..n {
575                accs.push(ReduceAcc {
576                    int_sum: d.i128()?,
577                    float_sum: d.f64()?,
578                    non_null: d.i64()?,
579                    float_count: d.i64()?,
580                });
581            }
582            StorageValue::Reduce { count, accs }
583        }
584        _ => return Err("invalid storage value tag".into()),
585    };
586    d.finish()?;
587    Ok(value)
588}
589
590fn put_row(out: &mut Vec<u8>, row: &OwnedRow) {
591    put_len(out, row.len());
592    for value in row.cells() {
593        put_value(out, &value.to_owned());
594    }
595}
596
597fn put_value(out: &mut Vec<u8>, value: &OwnedValue) {
598    match value {
599        // Operator-state rows are full server rows and never carry `Absent`; tag 6 keeps
600        // the codec total and round-trippable should one ever be persisted.
601        OwnedValue::Absent => out.push(6),
602        OwnedValue::Null => out.push(0),
603        OwnedValue::Bool(b) => {
604            out.push(1);
605            out.push(u8::from(*b));
606        }
607        OwnedValue::Int(i) => {
608            out.push(2);
609            out.extend_from_slice(&i.to_le_bytes());
610        }
611        OwnedValue::Float(f) => {
612            out.push(3);
613            out.extend_from_slice(&f.to_bits().to_le_bytes());
614        }
615        OwnedValue::Str(s) => {
616            out.push(4);
617            put_bytes(out, s.as_bytes());
618        }
619        OwnedValue::Json(s) => {
620            out.push(5);
621            put_bytes(out, s.as_bytes());
622        }
623    }
624}
625
626fn put_len(out: &mut Vec<u8>, len: usize) {
627    let len = u32::try_from(len).expect("operator storage value length fits in u32");
628    put_u32(out, len);
629}
630
631fn put_u32(out: &mut Vec<u8>, n: u32) {
632    out.extend_from_slice(&n.to_le_bytes());
633}
634
635fn put_i64(out: &mut Vec<u8>, n: i64) {
636    out.extend_from_slice(&n.to_le_bytes());
637}
638
639fn put_i128(out: &mut Vec<u8>, n: i128) {
640    out.extend_from_slice(&n.to_le_bytes());
641}
642
643fn put_f64(out: &mut Vec<u8>, n: f64) {
644    out.extend_from_slice(&n.to_bits().to_le_bytes());
645}
646
647fn put_bytes(out: &mut Vec<u8>, bytes: &[u8]) {
648    put_len(out, bytes.len());
649    out.extend_from_slice(bytes);
650}
651
652struct Decoder<'a> {
653    bytes: &'a [u8],
654    pos: usize,
655}
656
657impl<'a> Decoder<'a> {
658    fn new(bytes: &'a [u8]) -> Decoder<'a> {
659        Decoder { bytes, pos: 0 }
660    }
661
662    fn finish(&self) -> Result<(), String> {
663        if self.pos == self.bytes.len() {
664            Ok(())
665        } else {
666            Err("trailing storage codec bytes".into())
667        }
668    }
669
670    fn take(&mut self, len: usize) -> Result<&'a [u8], String> {
671        let end = self
672            .pos
673            .checked_add(len)
674            .ok_or_else(|| "storage codec length overflow".to_string())?;
675        let out = self
676            .bytes
677            .get(self.pos..end)
678            .ok_or_else(|| "truncated storage codec value".to_string())?;
679        self.pos = end;
680        Ok(out)
681    }
682
683    fn u8(&mut self) -> Result<u8, String> {
684        Ok(self.take(1)?[0])
685    }
686
687    fn u32(&mut self) -> Result<u32, String> {
688        let mut buf = [0; 4];
689        buf.copy_from_slice(self.take(4)?);
690        Ok(u32::from_le_bytes(buf))
691    }
692
693    fn i64(&mut self) -> Result<i64, String> {
694        let mut buf = [0; 8];
695        buf.copy_from_slice(self.take(8)?);
696        Ok(i64::from_le_bytes(buf))
697    }
698
699    fn i128(&mut self) -> Result<i128, String> {
700        let mut buf = [0; 16];
701        buf.copy_from_slice(self.take(16)?);
702        Ok(i128::from_le_bytes(buf))
703    }
704
705    fn u64(&mut self) -> Result<u64, String> {
706        let mut buf = [0; 8];
707        buf.copy_from_slice(self.take(8)?);
708        Ok(u64::from_le_bytes(buf))
709    }
710
711    fn f64(&mut self) -> Result<f64, String> {
712        Ok(f64::from_bits(self.u64()?))
713    }
714
715    fn bytes(&mut self) -> Result<&'a [u8], String> {
716        let len = self.u32()? as usize;
717        self.take(len)
718    }
719
720    fn string(&mut self) -> Result<String, String> {
721        String::from_utf8(self.bytes()?.to_vec())
722            .map_err(|_| "storage codec string is not UTF-8".to_string())
723    }
724
725    fn row(&mut self) -> Result<OwnedRow, String> {
726        let len = self.u32()? as usize;
727        let mut values = Vec::with_capacity(len);
728        for _ in 0..len {
729            values.push(self.value()?);
730        }
731        Ok(owned_row(values))
732    }
733
734    fn value(&mut self) -> Result<OwnedValue, String> {
735        match self.u8()? {
736            0 => Ok(OwnedValue::Null),
737            1 => match self.u8()? {
738                0 => Ok(OwnedValue::Bool(false)),
739                1 => Ok(OwnedValue::Bool(true)),
740                _ => Err("invalid bool tag".into()),
741            },
742            2 => Ok(OwnedValue::Int(self.i64()?)),
743            3 => Ok(OwnedValue::Float(f64::from_bits(self.u64()?))),
744            4 => Ok(OwnedValue::Str(Arc::from(self.string()?))),
745            5 => Ok(OwnedValue::Json(Arc::from(self.string()?))),
746            6 => Ok(OwnedValue::Absent),
747            _ => Err("invalid row value tag".into()),
748        }
749    }
750}
751
752// ---------------------------------------------------------------------------
753// WS02.3 / WS09.3 — operator-storage error sink + format versioning
754// ---------------------------------------------------------------------------
755
756/// A SQLite operator-storage failure must surface as a parked `RindleError` (drained by
757/// the graph at the mutation boundary) and return a safe sentinel — never `.expect`
758/// / abort. Here we force the WS09.3 format-version mismatch (a bumped/corrupt blob)
759/// and assert `get` returns `None` while parking a `RindleError::Storage`.
760#[cfg(test)]
761mod fault_tests {
762    use super::*;
763
764    #[test]
765    fn decode_version_mismatch_parks_error_not_panic() {
766        let db = DatabaseStorage::new_in_memory().expect("open in-memory db");
767        let sink: Rc<RefCell<Option<RindleError>>> = Rc::default();
768        let store = db.op_storage_with_sink(sink.clone());
769
770        // A valid round-trip parks nothing.
771        store.set(
772            "ok",
773            StorageValue::Bound(owned_row(vec![OwnedValue::Int(7)])),
774        );
775        assert!(matches!(store.get("ok"), Some(StorageValue::Bound(_))));
776        assert!(
777            sink.borrow().is_none(),
778            "a valid round-trip must not park an error"
779        );
780
781        // Inject a blob whose leading version byte is wrong (a future codec / corrupt
782        // store), into the store's own `op` namespace.
783        let bad_blob: Vec<u8> = vec![
784            STORAGE_FORMAT_VERSION.wrapping_add(1),
785            2, /* Bound tag */
786        ];
787        store
788            .db
789            .conn()
790            .execute(
791                r#"INSERT INTO storage ("op","key","val") VALUES (?1,?2,?3)"#,
792                params![store.op_id, "corrupt", bad_blob],
793            )
794            .expect("inject blob");
795
796        // `get` must NOT panic: it returns the safe sentinel (None) and parks a typed
797        // error for the graph to re-raise.
798        let got = store.get("corrupt");
799        assert!(
800            got.is_none(),
801            "a decode failure returns None, never a value"
802        );
803        let parked = sink.borrow();
804        assert!(
805            matches!(&*parked, Some(RindleError::Storage(msg)) if msg.contains("format v")),
806            "expected a parked RindleError::Storage about a format version, got {parked:?}"
807        );
808    }
809}
810
811/// Spec 10 §3.2 — `scan` pages through SQLite a bounded chunk at a time rather than
812/// materializing the whole range. These guard the keyset-paginated cursor across result
813/// sets larger than one page: a bad inter-page seek would drop, duplicate, mis-order, or
814/// bleed past the prefix. Data is sized off `SCAN_PAGE` so the test keeps spanning several
815/// pages if that constant ever changes.
816#[cfg(test)]
817mod scan_paging_tests {
818    use super::*;
819
820    fn reduce_count(v: Option<StorageValue>) -> Option<i64> {
821        match v {
822            Some(StorageValue::Reduce { count, .. }) => Some(count),
823            _ => None,
824        }
825    }
826
827    #[test]
828    fn scan_spans_many_pages_without_dropping_or_reordering_keys() {
829        let db = DatabaseStorage::new_in_memory().expect("open in-memory db");
830        let s = db.create_storage();
831
832        // Several pages of "a:" keys, then a small "b:" block immediately after so a
833        // prefix scan has to stop at a boundary that falls mid-page. Zero-padded to a fixed
834        // width so byte (BINARY) order matches numeric order — and matches `MemoryStorage`'s
835        // `BTreeMap<Box<str>>` order, the parity the two backends must keep.
836        let n = SCAN_PAGE * 2 + 7;
837        for i in 0..n {
838            s.set(
839                &format!("a:{i:06}"),
840                StorageValue::Reduce {
841                    count: i as i64,
842                    accs: Vec::new(),
843                },
844            );
845        }
846        for i in 0..5 {
847            s.set(
848                &format!("b:{i:06}"),
849                StorageValue::Reduce {
850                    count: -1,
851                    accs: Vec::new(),
852                },
853            );
854        }
855
856        // Whole-keyspace scan: every key exactly once, strictly ascending across pages.
857        let all: Vec<Box<str>> = s.scan("").map(|(k, _)| k).collect();
858        assert_eq!(all.len(), n + 5, "scan must yield every key across pages");
859        assert!(
860            all.windows(2).all(|w| w[0] < w[1]),
861            "paged scan keys must stay strictly ascending with no duplicates"
862        );
863
864        // Prefix scan that crosses page boundaries: exactly the `n` "a:" keys, in order,
865        // stopping before the "b:" block instead of running into the next page.
866        let a_keys: Vec<String> = s.scan("a:").map(|(k, _)| k.to_string()).collect();
867        let want: Vec<String> = (0..n).map(|i| format!("a:{i:06}")).collect();
868        assert_eq!(
869            a_keys, want,
870            "prefix scan dropped/added a key or crossed the prefix"
871        );
872
873        // Values round-trip on both sides of a page boundary (paging must not corrupt them).
874        for i in [0, SCAN_PAGE - 1, SCAN_PAGE, SCAN_PAGE + 1, n - 1] {
875            assert_eq!(
876                reduce_count(s.get(&format!("a:{i:06}"))),
877                Some(i as i64),
878                "value at index {i} (near a page boundary) failed to round-trip"
879            );
880        }
881    }
882
883    #[test]
884    fn empty_and_unmatched_prefix_scans_are_empty() {
885        let db = DatabaseStorage::new_in_memory().expect("open in-memory db");
886        let s = db.create_storage();
887
888        // Empty store: the cursor's first page is empty and the scan ends immediately.
889        assert_eq!(
890            s.scan("").count(),
891            0,
892            "scan of an empty store yields nothing"
893        );
894
895        s.set(
896            "a:0",
897            StorageValue::Reduce {
898                count: 0,
899                accs: Vec::new(),
900            },
901        );
902        // A prefix with no matches must yield nothing (the first key found is past it).
903        assert_eq!(
904            s.scan("z:").count(),
905            0,
906            "non-matching prefix yields nothing"
907        );
908    }
909}
rindle_sqlite/storage.rs

rindle_sqlite/
storage.rs
