Skip to main content

datumesh_store/
sqlite.rs

1//! `SqliteStore`: the store in one SQLite file (since 0.4.0).
2
3use std::borrow::Cow;
4use std::path::{Path, PathBuf};
5use std::sync::Mutex;
6use std::time::{Duration, Instant};
7
8use rusqlite::types::Value as SqlValue;
9use rusqlite::{Connection, OpenFlags, OptionalExtension};
10
11use crate::error::{Db, Error, Result};
12use crate::slices::{check_entry, cmp_keys, is_hash, is_instant, version_slice, Constraint, Kind, Pattern};
13use crate::sqlite_sql::L;
14use crate::{Binding, Namespace, SliceMember, SliceValue, Store, Version};
15use datumesh_core::json::{self, Value};
16use datumesh_core::lexical::{check_dns_name, check_function_name, check_hex, check_instant};
17use datumesh_core::{
18    keys_from_response, sha256_hex, verify, Arg, Args, Cell, ErrorCode, FunctionDescriptor, Keys, Manifest, Receipt,
19    Resolver, Scalar, Slice, Type,
20};
21
22/// How a [`SqliteStore`] opens and uses its connections.
23///
24/// Builder style: `SqliteOptions::new().busy_timeout(Duration::from_secs(5))`.
25#[derive(Clone, Debug, PartialEq, Eq)]
26pub struct SqliteOptions {
27    max_size: usize,
28    busy_timeout: Duration,
29    durable: bool,
30}
31
32impl Default for SqliteOptions {
33    fn default() -> SqliteOptions {
34        SqliteOptions::new()
35    }
36}
37
38impl SqliteOptions {
39    /// At most 8 idle connections kept; a writer waits at most 30 seconds for the
40    /// database's write lock (and a reader for a lock held in exclusive mode); every commit
41    /// durable (`synchronous = FULL`).
42    pub fn new() -> SqliteOptions {
43        SqliteOptions { max_size: 8, busy_timeout: Duration::from_secs(30), durable: true }
44    }
45
46    /// The most idle connections kept open for reuse (at least 1). More may be open at
47    /// once, one per concurrent call; SQLite connections are cheap, and readers do not
48    /// wait for each other or for the writer (WAL).
49    pub fn max_size(mut self, n: usize) -> SqliteOptions {
50        self.max_size = n.max(1);
51        self
52    }
53
54    /// How long a call waits for a lock another connection (of this process or another)
55    /// holds before failing with [`Error::Database`] (`SQLITE_BUSY`; reads through the
56    /// [`Resolver`]: `None`). Writers take the database's one write lock for the duration
57    /// of a `put_*`, so this is the longest a writer waits behind the others.
58    pub fn busy_timeout(mut self, d: Duration) -> SqliteOptions {
59        self.busy_timeout = d;
60        self
61    }
62
63    /// Whether each commit is flushed to disk before it returns (`synchronous = FULL`, the
64    /// default), or only at checkpoints (`synchronous = NORMAL`: a power loss can lose the
65    /// latest commits, never consistency; much faster for bulk loads, like PostgreSQL's
66    /// `synchronous_commit = off`). Either way an application crash loses nothing
67    /// committed.
68    pub fn durable(mut self, yes: bool) -> SqliteOptions {
69        self.durable = yes;
70        self
71    }
72}
73
74/// The store in one SQLite file (bundled SQLite, no server), with the same behaviour as
75/// [`PgStore`](crate::PgStore): every method, error and rule of [`Store`] and
76/// [`Resolver`]. `Send + Sync`: share one per process (behind an `Arc`); each call takes a
77/// connection for its duration. Since 0.4.0.
78///
79/// The file is in WAL mode (readers never wait for the writer), with a busy timeout
80/// ([`SqliteOptions::busy_timeout`]). Every `put_*` is one transaction begun with `BEGIN
81/// IMMEDIATE`: it takes the database's write lock first, so its checks (a version later
82/// than the latest, a key not yet taken) and its inserts are one atomic step against every
83/// other writer, threads of this process or other processes on the same machine. The
84/// append-only rules are also triggers in the file, so they hold for any connection. (A
85/// SQLite file on a network file system is not supported: its locks are not reliable.)
86///
87/// Every `put_*` is one transaction. [`Store`] methods return errors: a malformed argument
88/// is [`Error::Invalid`] (no query is made), a database failure [`Error::Database`]. The
89/// [`Resolver`] methods answer both with `None`; a database failure is logged with
90/// `tracing` (target `datumesh_store`, level `error`).
91pub struct SqliteStore {
92    path: PathBuf,
93    options: SqliteOptions,
94    idle: Mutex<Vec<Connection>>,
95    /// The store's own writers queue here before taking the database's write lock: SQLite's
96    /// busy handler polls with sleeps of up to 100 ms, so threads contending for the lock
97    /// would mostly sleep. Other stores and processes still meet at the database's lock.
98    writer: Mutex<()>,
99    /// Opened with `open_existing`: never creates the file, never changes its journal mode.
100    existing: bool,
101}
102
103impl std::fmt::Debug for SqliteStore {
104    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105        f.debug_struct("SqliteStore").field("path", &self.path).finish_non_exhaustive()
106    }
107}
108
109/// A connection taken from the store, returned to it when dropped.
110struct Pooled<'a> {
111    store: &'a SqliteStore,
112    conn: Option<Connection>,
113}
114
115impl std::ops::Deref for Pooled<'_> {
116    type Target = Connection;
117    fn deref(&self) -> &Connection {
118        self.conn.as_ref().expect("a pooled connection")
119    }
120}
121
122impl Drop for Pooled<'_> {
123    fn drop(&mut self) {
124        if let Some(c) = self.conn.take() {
125            // A connection left inside a transaction (a failed rollback) is not reused.
126            if c.is_autocommit() {
127                if let Ok(mut idle) = self.store.idle.lock() {
128                    if idle.len() < self.store.options.max_size {
129                        idle.push(c);
130                    }
131                }
132            }
133        }
134    }
135}
136
137impl SqliteStore {
138    /// Opens the SQLite file at `path` with the default [`SqliteOptions`]; a missing file
139    /// is created (empty: apply [`migrate::sqlite::apply`](crate::migrate::sqlite::apply)
140    /// next), as a missing directory is not. [`open_existing`](Self::open_existing) creates
141    /// nothing.
142    pub fn open(path: impl AsRef<Path>) -> Result<SqliteStore> {
143        SqliteStore::open_with(path, SqliteOptions::new())
144    }
145
146    /// Opens the SQLite file at `path`, creating it when it is missing (its directory must
147    /// exist). One connection is opened at once (and
148    /// the file put in WAL mode), so a path that cannot be opened fails here
149    /// (`SQLITE_CANTOPEN`); the schema is not looked at, so a new or empty file is fine
150    /// (create the schema with [`migrate::sqlite::apply`](crate::migrate::sqlite::apply)).
151    pub fn open_with(path: impl AsRef<Path>, options: SqliteOptions) -> Result<SqliteStore> {
152        let store = SqliteStore {
153            path: path.as_ref().to_path_buf(),
154            options,
155            idle: Mutex::new(Vec::new()),
156            writer: Mutex::new(()),
157            existing: false,
158        };
159        let c = store.open_connection()?;
160        if let Ok(mut idle) = store.idle.lock() {
161            idle.push(c);
162        }
163        Ok(store)
164    }
165
166    /// Opens an existing SQLite file without changing anything: a path with no file is
167    /// [`Error::Database`] (`no such database file: <path>`, code `SQLITE_CANTOPEN`) and
168    /// nothing is created; the file's journal mode is left as it is. For checking a
169    /// database ([`migrate::sqlite::check`](crate::migrate::sqlite::check), the command's
170    /// `migrate --check`), which has no side effects. Writes through such a store work
171    /// as through any other. Since 0.4.0.
172    pub fn open_existing(path: impl AsRef<Path>) -> Result<SqliteStore> {
173        let path = path.as_ref();
174        if !path.is_file() {
175            return Err(Error::Database(crate::error::DatabaseError::no_such_file(path)));
176        }
177        let store = SqliteStore {
178            path: path.to_path_buf(),
179            options: SqliteOptions::new(),
180            idle: Mutex::new(Vec::new()),
181            writer: Mutex::new(()),
182            existing: true,
183        };
184        let c = store.open_connection()?;
185        if let Ok(mut idle) = store.idle.lock() {
186            idle.push(c);
187        }
188        Ok(store)
189    }
190
191    /// Opens a store named by a URL: `sqlite:<path>` or `sqlite://<path>` (so
192    /// `sqlite:///var/lib/datumesh/store.db` is an absolute path), or a plain file path.
193    /// Query parameters are not taken: the path is everything after the scheme.
194    pub fn connect(url: &str) -> Result<SqliteStore> {
195        SqliteStore::open(sqlite_path(url))
196    }
197
198    /// The database file's path.
199    pub fn path(&self) -> &Path {
200        &self.path
201    }
202
203    fn open_connection(&self) -> Result<Connection> {
204        let mut flags = OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_NO_MUTEX;
205        if !self.existing {
206            flags |= OpenFlags::SQLITE_OPEN_CREATE;
207        }
208        let c = Connection::open_with_flags(&self.path, flags).db()?;
209        c.busy_timeout(self.options.busy_timeout).db()?;
210        c.set_prepared_statement_cache_capacity(L::ALL.len() + 16);
211        if self.existing {
212            // Nothing about the file changes: no journal mode, no synchronous setting.
213            c.execute_batch("PRAGMA foreign_keys = ON").db()?;
214            return Ok(c);
215        }
216        // WAL: readers and the writer do not block each other, and a commit appends to the
217        // log. Changing the mode needs a lock the busy handler does not always wait for (a
218        // new file opened by several processes at once), so it is retried here.
219        let start = Instant::now();
220        loop {
221            match c.query_row("PRAGMA journal_mode = WAL", [], |r| r.get::<_, String>(0)) {
222                Ok(mode) if mode.eq_ignore_ascii_case("wal") => break,
223                Ok(mode) => return Err(Error::Config(format!("SQLite: journal mode WAL refused (it is {mode})"))),
224                Err(e) if is_busy(&e) && start.elapsed() < self.options.busy_timeout => {
225                    std::thread::sleep(Duration::from_millis(5));
226                }
227                Err(e) => return Err(e).db(),
228            }
229        }
230        let sync = if self.options.durable { "FULL" } else { "NORMAL" };
231        c.execute_batch(&format!("PRAGMA foreign_keys = ON; PRAGMA synchronous = {sync};")).db()?;
232        Ok(c)
233    }
234
235    /// A connection: an idle one, or a new one.
236    fn conn(&self) -> Result<Pooled<'_>> {
237        let idle = self.idle.lock().ok().and_then(|mut v| v.pop());
238        let conn = match idle {
239            Some(c) => c,
240            None => self.open_connection()?,
241        };
242        Ok(Pooled { store: self, conn: Some(conn) })
243    }
244
245    /// Prepares every SQL statement the store uses against the live schema, on a new
246    /// connection, and returns how many there are. Fails with the first statement the
247    /// schema cannot serve.
248    pub fn check_statements(&self) -> Result<usize> {
249        let c = self.open_connection()?;
250        for q in L::ALL {
251            c.prepare(q.sql()).db()?;
252        }
253        Ok(L::ALL.len())
254    }
255
256    /// A connection held for a migration (one transaction on it), or a catalog read.
257    pub(crate) fn raw(&self) -> Result<impl std::ops::Deref<Target = Connection> + '_> {
258        self.conn()
259    }
260
261    /// A [`Resolver`] lookup: a failure (logged by `run`) becomes `None`.
262    fn resolve<T>(&self, op: &'static str, f: impl FnOnce(&Connection) -> Result<Option<T>>) -> Option<T> {
263        self.run(op, f).ok().flatten()
264    }
265
266    /// Runs `f` on a connection; a database failure is logged.
267    fn run<T>(&self, op: &'static str, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
268        let result = self.conn().and_then(|c| f(&c));
269        if let Err(e @ Error::Database(_)) = &result {
270            tracing::error!(target: "datumesh_store", op, error = %e, "database failure");
271        }
272        result
273    }
274
275    /// Runs `f` in a write transaction (`BEGIN IMMEDIATE`: the write lock first), committed
276    /// when `f` succeeds and rolled back otherwise; a database failure is logged.
277    fn write<T>(&self, op: &'static str, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
278        self.run(op, |c| {
279            let v = self.locked_write(c, f)?;
280            self.bound_wal(c);
281            Ok(v)
282        })
283    }
284
285    /// `write`, after the store's other writers.
286    fn locked_write<T>(&self, c: &Connection, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
287        let _queue = self.writer.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
288        write(c, f)
289    }
290
291    /// Keeps the write-ahead log bounded. SQLite's automatic checkpoints copy the log into
292    /// the database but restart the log only when no reader is using it, which under a
293    /// steady load of concurrent readers may be never: the log then grows with every
294    /// commit. After a commit, when the log has outgrown `WAL_LIMIT`, this connection runs
295    /// a TRUNCATE checkpoint, which waits (the busy timeout) for the readers to move to the
296    /// latest snapshot, copies everything and empties the log. A failure (still busy) is
297    /// left to the next commit.
298    fn bound_wal(&self, c: &Connection) {
299        let mut wal = self.path.clone().into_os_string();
300        wal.push("-wal");
301        if std::fs::metadata(&wal).is_ok_and(|m| m.len() > WAL_LIMIT) {
302            if let Err(e) = c.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(())) {
303                tracing::debug!(target: "datumesh_store", error = %e, "checkpoint deferred");
304            }
305        }
306    }
307
308    /// A version from a query answering (tx, receipt_hash, asof) for (cell_id, asof).
309    fn asof_version(&self, op: &'static str, q: L, cell_id: &str, asof: &str) -> Result<Option<Version>> {
310        check_hex(cell_id, 64, "cell_id")?;
311        check_instant(asof, "asof")?;
312        self.run(op, |c| {
313            c.prepare_cached(q.sql())
314                .db()?
315                .query_row(rusqlite::params![cell_id, asof], |r| version_row(r, cell_id))
316                .optional()
317                .db()
318        })
319    }
320
321    /// A version and its receipt's bytes from a query answering (tx, receipt_hash, asof,
322    /// bytes) for (cell_id, asof).
323    fn asof_with_receipt(
324        &self,
325        op: &'static str,
326        q: L,
327        cell_id: &str,
328        asof: &str,
329    ) -> Result<Option<(Version, Vec<u8>)>> {
330        check_hex(cell_id, 64, "cell_id")?;
331        check_instant(asof, "asof")?;
332        self.run(op, |c| {
333            c.prepare_cached(q.sql())
334                .db()?
335                .query_row(rusqlite::params![cell_id, asof], |r| Ok((version_row(r, cell_id)?, r.get(3)?)))
336                .optional()
337                .db()
338        })
339    }
340}
341
342/// The size of the write-ahead log beyond which a commit checkpoints it and truncates it.
343const WAL_LIMIT: u64 = 64 << 20;
344
345/// Runs `f` in a `BEGIN IMMEDIATE` transaction on `c`.
346pub(crate) fn write<T>(c: &Connection, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
347    c.execute_batch("BEGIN IMMEDIATE").db()?;
348    let result = f(c).and_then(|v| c.execute_batch("COMMIT").db().map(|()| v));
349    if result.is_err() && !c.is_autocommit() {
350        let _ = c.execute_batch("ROLLBACK");
351    }
352    result
353}
354
355fn is_busy(e: &rusqlite::Error) -> bool {
356    matches!(e, rusqlite::Error::SqliteFailure(f, _) if f.code == rusqlite::ErrorCode::DatabaseBusy)
357}
358
359/// The path of a `sqlite:` URL (`sqlite:<path>`, `sqlite://<path>`), or `url` itself.
360pub(crate) fn sqlite_path(url: &str) -> &str {
361    match url.strip_prefix("sqlite:") {
362        Some(rest) => rest.strip_prefix("//").unwrap_or(rest),
363        None => url,
364    }
365}
366
367/// The name of a SQLite (extended) result code.
368pub(crate) fn result_code_name(code: std::os::raw::c_int) -> String {
369    let name = match code {
370        1 => "SQLITE_ERROR",
371        2 => "SQLITE_INTERNAL",
372        3 => "SQLITE_PERM",
373        4 => "SQLITE_ABORT",
374        5 => "SQLITE_BUSY",
375        6 => "SQLITE_LOCKED",
376        7 => "SQLITE_NOMEM",
377        8 => "SQLITE_READONLY",
378        9 => "SQLITE_INTERRUPT",
379        10 => "SQLITE_IOERR",
380        11 => "SQLITE_CORRUPT",
381        12 => "SQLITE_NOTFOUND",
382        13 => "SQLITE_FULL",
383        14 => "SQLITE_CANTOPEN",
384        15 => "SQLITE_PROTOCOL",
385        16 => "SQLITE_EMPTY",
386        17 => "SQLITE_SCHEMA",
387        18 => "SQLITE_TOOBIG",
388        19 => "SQLITE_CONSTRAINT",
389        20 => "SQLITE_MISMATCH",
390        21 => "SQLITE_MISUSE",
391        22 => "SQLITE_NOLFS",
392        23 => "SQLITE_AUTH",
393        24 => "SQLITE_FORMAT",
394        25 => "SQLITE_RANGE",
395        26 => "SQLITE_NOTADB",
396        261 => "SQLITE_BUSY_RECOVERY",
397        517 => "SQLITE_BUSY_SNAPSHOT",
398        773 => "SQLITE_BUSY_TIMEOUT",
399        262 => "SQLITE_LOCKED_SHAREDCACHE",
400        264 => "SQLITE_READONLY_RECOVERY",
401        520 => "SQLITE_READONLY_CANTLOCK",
402        270 => "SQLITE_CANTOPEN_NOTEMPDIR",
403        526 => "SQLITE_CANTOPEN_ISDIR",
404        782 => "SQLITE_CANTOPEN_FULLPATH",
405        267 => "SQLITE_CORRUPT_VTAB",
406        275 => "SQLITE_CONSTRAINT_CHECK",
407        531 => "SQLITE_CONSTRAINT_COMMITHOOK",
408        787 => "SQLITE_CONSTRAINT_FOREIGNKEY",
409        1043 => "SQLITE_CONSTRAINT_FUNCTION",
410        1299 => "SQLITE_CONSTRAINT_NOTNULL",
411        1555 => "SQLITE_CONSTRAINT_PRIMARYKEY",
412        1811 => "SQLITE_CONSTRAINT_TRIGGER",
413        2067 => "SQLITE_CONSTRAINT_UNIQUE",
414        2323 => "SQLITE_CONSTRAINT_VTAB",
415        2579 => "SQLITE_CONSTRAINT_ROWID",
416        3091 => "SQLITE_CONSTRAINT_DATATYPE",
417        _ => {
418            // Another extended code: its primary code's name and the number.
419            let primary = code & 0xff;
420            return if primary != code && primary != 0 {
421                format!("{}({code})", result_code_name(primary))
422            } else {
423                format!("SQLITE({code})")
424            };
425        }
426    };
427    name.to_string()
428}
429
430fn version_row(r: &rusqlite::Row<'_>, cell_id: &str) -> rusqlite::Result<Version> {
431    Ok(Version { cell_id: cell_id.to_string(), tx: r.get(0)?, receipt_hash: r.get(1)?, asof: r.get(2)? })
432}
433
434fn get_bytes(c: &Connection, q: L, key: &str) -> Result<Option<Vec<u8>>> {
435    c.prepare_cached(q.sql()).db()?.query_row([key], |r| r.get(0)).optional().db()
436}
437
438fn get_strings(c: &Connection, q: L, key: &str) -> Result<Vec<String>> {
439    let mut st = c.prepare_cached(q.sql()).db()?;
440    let rows = st.query_map([key], |r| r.get(0)).db()?;
441    rows.collect::<rusqlite::Result<Vec<String>>>().db()
442}
443
444fn exists(c: &Connection, q: L, key: &str) -> Result<bool> {
445    c.prepare_cached(q.sql()).db()?.query_row([key], |r| r.get(0)).db()
446}
447
448fn exec(c: &Connection, q: L, params: impl rusqlite::Params) -> Result<usize> {
449    c.prepare_cached(q.sql()).db()?.execute(params).db()
450}
451
452/// A stored value that does not parse back (a corrupt row).
453fn bad(what: &str) -> Error {
454    Error::Invalid(datumesh_core::Error::new(ErrorCode::CanonType, "$", format!("a stored {what}")))
455}
456
457/// A cell's arguments from the typed-argument index: a JSON array of `[name, type, value]`.
458fn args_of(text: &str) -> Result<Args> {
459    let Ok(Value::Array(list)) = json::parse(text.as_bytes()) else { return Err(bad("argument list")) };
460    let mut args = Args::new();
461    for a in &list {
462        let item = |i: usize| match a {
463            Value::Array(x) => x.get(i).and_then(Value::as_str).ok_or_else(|| bad("argument")),
464            _ => Err(bad("argument")),
465        };
466        let ty = Type::from_name(item(1)?).ok_or_else(|| bad("type"))?;
467        args.insert(item(0)?.to_string(), Arg::Value(Scalar::from_lexical(ty, item(2)?)?));
468    }
469    Ok(args)
470}
471
472/// The sort key of a typed value in the typed-argument index: for a number, a text that
473/// sorts (as bytes) as the decimal does; for every other type, its lexical form, which
474/// already sorts in the type's order.
475pub(crate) fn sort_key(ty: &str, lexical: &str) -> String {
476    if ty == "number" {
477        decimal_key(lexical)
478    } else {
479        lexical.to_string()
480    }
481}
482
483/// A decimal (SPEC 3.1 form: optional `-`, digits, optional `.` and digits) as a text that
484/// sorts as the number does: `0` for a negative number, `1` for zero, `2` for a positive
485/// one; then the decimal exponent `e` (the number is `0.d1d2... x 10^e`) plus 500, in three
486/// digits; then the significant digits `d1d2...`. Longer digit strings after a shared
487/// prefix are larger, as text and as numbers. For a negative number the exponent and the
488/// digits are complemented (`999 - x`, `9 - d`) and `~` appended, which reverses the order,
489/// prefixes included (`~` sorts after every digit). The specification's numbers have at
490/// most 128 characters, so `e` lies in -127..=128.
491pub(crate) fn decimal_key(lexical: &str) -> String {
492    let (negative, magnitude) = match lexical.strip_prefix('-') {
493        Some(m) => (true, m),
494        None => (false, lexical),
495    };
496    let (int, frac) = magnitude.split_once('.').unwrap_or((magnitude, ""));
497    let int = int.trim_start_matches('0');
498    let frac = frac.trim_end_matches('0');
499    let (exponent, digits): (i64, String) = if !int.is_empty() {
500        (int.len() as i64, format!("{int}{frac}").trim_end_matches('0').to_string())
501    } else {
502        let zeros = frac.len() - frac.trim_start_matches('0').len();
503        (-(zeros as i64), frac[zeros..].to_string())
504    };
505    if digits.is_empty() {
506        return "1".to_string();
507    }
508    let e = (exponent + 500).clamp(0, 999);
509    if negative {
510        let complement: String = digits.bytes().map(|d| char::from(b'9' - (d - b'0'))).collect();
511        format!("0{:03}{complement}~", 999 - e)
512    } else {
513        format!("2{e:03}{digits}")
514    }
515}
516
517impl Resolver for SqliteStore {
518    fn descriptor(&self, fn_hash: &str) -> Option<Cow<'_, [u8]>> {
519        if !is_hash(fn_hash) {
520            return None;
521        }
522        self.resolve("descriptor", |c| get_bytes(c, L::DescriptorGet, fn_hash)).map(Cow::Owned)
523    }
524
525    fn receipt(&self, hash: &str) -> Option<Cow<'_, [u8]>> {
526        if !is_hash(hash) {
527            return None;
528        }
529        self.resolve("receipt", |c| get_bytes(c, L::ReceiptGet, hash)).map(Cow::Owned)
530    }
531
532    /// The stored manifest of slice entry `entry` ([`Store::put_manifest`]), its canonical
533    /// form; `None` when there is none, or `entry` is not an entry.
534    fn manifest(&self, entry: &str) -> Option<Cow<'_, [u8]>> {
535        if check_entry(entry, "$").is_err() {
536            return None;
537        }
538        self.resolve("manifest", |c| get_bytes(c, L::ManifestGet, entry)).map(Cow::Owned)
539    }
540
541    /// `Some` only for a `stored` function, as [`PgStore`](crate::PgStore)'s: the value of
542    /// the receipt of [`stored_asof`](Store::stored_asof)'s version.
543    fn evaluate(&self, cell: &Cell, asof: &str) -> Option<Scalar> {
544        if !is_instant(asof) {
545            return None;
546        }
547        self.resolve("evaluate", |c| {
548            let Some(desc) = get_bytes(c, L::DescriptorGet, &cell.fn_hash)? else { return Ok(None) };
549            match FunctionDescriptor::from_json(&desc) {
550                Ok(d) if d.lang == "stored" => {}
551                Ok(_) => return Ok(None),
552                Err(e) => {
553                    tracing::error!(target: "datumesh_store", fn_hash = %cell.fn_hash, error = %e, "stored descriptor does not parse");
554                    return Ok(None);
555                }
556            }
557            let bytes: Option<Vec<u8>> = c
558                .prepare_cached(L::ValueAsof.sql())
559                .db()?
560                .query_row(rusqlite::params![cell.id(), asof], |r| r.get(0))
561                .optional()
562                .db()?;
563            let Some(bytes) = bytes else { return Ok(None) };
564            match Receipt::from_json(&bytes) {
565                Ok(r) => Ok(Some(r.value().clone())),
566                Err(e) => {
567                    tracing::error!(target: "datumesh_store", cell_id = %cell.id(), error = %e, "stored receipt does not parse");
568                    Ok(None)
569                }
570            }
571        })
572    }
573}
574
575impl Store for SqliteStore {
576    fn put_descriptor(&self, bytes: &[u8]) -> Result<String> {
577        let d = FunctionDescriptor::from_json(bytes)?;
578        let hash = d.hash();
579        let canonical = d.canonical();
580        self.run("put_descriptor", |c| {
581            if exists(c, L::DescriptorExists, &hash)? {
582                return Ok(());
583            }
584            self.locked_write(c, |c| exec(c, L::DescriptorPut, rusqlite::params![hash, canonical]).map(|_| ()))?;
585            self.bound_wal(c);
586            Ok(())
587        })?;
588        tracing::debug!(target: "datumesh_store", fn_hash = %hash, "put_descriptor");
589        Ok(hash)
590    }
591
592    fn put_receipt(&self, bytes: &[u8]) -> Result<String> {
593        // Steps 1-4 first: a receipt that is not one is refused without a query.
594        let r = Receipt::from_json(bytes)?;
595        let canonical = json::parse(bytes)?.to_jcs();
596        let hash = r.hash().to_string();
597        let cell = r.cell();
598        let cell_id = cell.id();
599        let host = r.signer_host().to_string();
600        let (mut entries, mut inputs): (Vec<String>, Vec<String>) =
601            r.inputs().iter().cloned().partition(|e| e.starts_with("slice:"));
602        entries.sort();
603        entries.dedup();
604        inputs.sort();
605        inputs.dedup();
606        let mut args = Vec::with_capacity(cell.args.len());
607        for (name, a) in &cell.args {
608            let Some(s) = a.scalar() else {
609                return Err(Error::Invalid(datumesh_core::Error::new(
610                    ErrorCode::Schema,
611                    "$.cell.args",
612                    "a receipt's cell has no open argument",
613                )));
614            };
615            let ty = s.ty().as_str();
616            args.push((name.clone(), ty, s.lexical().to_string(), sort_key(ty, s.lexical())));
617        }
618        self.run("put_receipt", |c| {
619            // Idempotent: the very receipt already stored verified when it was stored.
620            if let Some(stored) = get_bytes(c, L::ReceiptGet, &hash)? {
621                if stored == canonical {
622                    return Ok(());
623                }
624            }
625            // Steps 5-6 against the stored key document of the signer host (outside the
626            // write lock: verifying is the slow part).
627            let mut keys = Keys::new();
628            if let Some(doc) = get_bytes(c, L::NamespaceGet, &host)? {
629                keys.insert_response(&host, 200, Some(&doc));
630            }
631            let out = verify(bytes, &keys, None, None);
632            if !out.ok {
633                return Err(Error::Invalid(out.error().unwrap_or_else(|| {
634                    datumesh_core::Error::new(ErrorCode::Signature, "$", "the receipt did not verify")
635                })));
636            }
637            self.locked_write(c, |c| {
638                // Under the write lock: another writer may have stored it meanwhile.
639                if let Some(stored) = get_bytes(c, L::ReceiptGet, &hash)? {
640                    return if stored == canonical { Ok(()) } else { Err(Error::Conflict { hash: hash.clone() }) };
641                }
642                let value = r.value();
643                exec(
644                    c,
645                    L::ReceiptPut,
646                    rusqlite::params![
647                        hash,
648                        canonical,
649                        cell_id,
650                        r.tx(),
651                        cell.fn_hash,
652                        r.publisher(),
653                        r.asof(),
654                        value.ty().as_str(),
655                        value.lexical()
656                    ],
657                )?;
658                for i in &inputs {
659                    exec(c, L::EdgePut, rusqlite::params![hash, i])?;
660                }
661                for e in &entries {
662                    exec(c, L::SliceEdgePut, rusqlite::params![hash, e])?;
663                }
664                if !exists(c, L::CellExists, &cell_id)? {
665                    exec(
666                        c,
667                        L::CellPut,
668                        rusqlite::params![cell_id, cell.fn_hash, cell.ns, cell.fn_name, args.len() as i64],
669                    )?;
670                    for (name, ty, value, key) in &args {
671                        exec(c, L::CellArgPut, rusqlite::params![cell_id, name, ty, value, key, cell.fn_hash])?;
672                    }
673                }
674                Ok(())
675            })?;
676            self.bound_wal(c);
677            Ok(())
678        })?;
679        tracing::debug!(target: "datumesh_store", hash = %hash, cell_id = %cell_id, "put_receipt");
680        Ok(hash)
681    }
682
683    fn put_blob(&self, bytes: &[u8]) -> Result<String> {
684        let sha = sha256_hex(bytes);
685        let length = i64::try_from(bytes.len()).unwrap_or(i64::MAX);
686        self.run("put_blob", |c| {
687            if exists(c, L::BlobExists, &sha)? {
688                return Ok(());
689            }
690            self.locked_write(c, |c| exec(c, L::BlobPut, rusqlite::params![sha, bytes, length]).map(|_| ()))?;
691            self.bound_wal(c);
692            Ok(())
693        })?;
694        tracing::debug!(target: "datumesh_store", sha256 = %sha, length, "put_blob");
695        Ok(sha)
696    }
697
698    fn blob(&self, sha256: &str) -> Result<Option<Cow<'_, [u8]>>> {
699        check_hex(sha256, 64, "sha256")?;
700        Ok(self.run("blob", |c| get_bytes(c, L::BlobGet, sha256))?.map(Cow::Owned))
701    }
702
703    fn put_version(&self, cell_id: &str, tx: &str, receipt_hash: &str) -> Result<()> {
704        check_hex(cell_id, 64, "cell_id")?;
705        check_instant(tx, "tx")?;
706        check_hex(receipt_hash, 64, "receipt_hash")?;
707        self.write("put_version", |c| {
708            let meta: Option<(String, String)> = c
709                .prepare_cached(L::ReceiptMeta.sql())
710                .db()?
711                .query_row([receipt_hash], |r| Ok((r.get(0)?, r.get(1)?)))
712                .optional()
713                .db()?;
714            let Some((r_cell, r_tx)) = meta else {
715                return Err(Error::NotFound { what: "receipt", key: receipt_hash.to_string() });
716            };
717            if r_cell != cell_id {
718                return Err(Error::Mismatch { what: "cell_id", expected: cell_id.to_string(), found: r_cell });
719            }
720            if r_tx != tx {
721                return Err(Error::Mismatch { what: "tx", expected: tx.to_string(), found: r_tx });
722            }
723            if let Some(l) = get_strings(c, L::VersionLatestTx, cell_id)?.pop() {
724                if tx <= l.as_str() {
725                    return Err(Error::NotLater {
726                        what: "version",
727                        key: cell_id.to_string(),
728                        given: tx.to_string(),
729                        latest: l,
730                    });
731                }
732            }
733            exec(c, L::VersionPut, rusqlite::params![cell_id, tx, receipt_hash])?;
734            Ok(())
735        })?;
736        tracing::debug!(target: "datumesh_store", cell_id, tx, receipt_hash, "put_version");
737        Ok(())
738    }
739
740    fn version_asof(&self, cell_id: &str, asof: &str) -> Result<Option<Version>> {
741        self.asof_version("version_asof", L::VersionAsof, cell_id, asof)
742    }
743
744    fn versions(&self, cell_id: &str, from: &str, to: &str) -> Result<Vec<Version>> {
745        check_hex(cell_id, 64, "cell_id")?;
746        check_instant(from, "from")?;
747        check_instant(to, "to")?;
748        self.run("versions", |c| {
749            let mut st = c.prepare_cached(L::VersionRange.sql()).db()?;
750            let rows = st.query_map(rusqlite::params![cell_id, from, to], |r| version_row(r, cell_id)).db()?;
751            rows.collect::<rusqlite::Result<Vec<Version>>>().db()
752        })
753    }
754
755    fn latest(&self, cell_id: &str) -> Result<Option<Version>> {
756        check_hex(cell_id, 64, "cell_id")?;
757        self.run("latest", |c| {
758            c.prepare_cached(L::VersionLatest.sql())
759                .db()?
760                .query_row([cell_id], |r| version_row(r, cell_id))
761                .optional()
762                .db()
763        })
764    }
765
766    fn dependents(&self, receipt_hash: &str) -> Result<Vec<String>> {
767        check_hex(receipt_hash, 64, "receipt_hash")?;
768        self.run("dependents", |c| get_strings(c, L::Dependents, receipt_hash))
769    }
770
771    fn inputs(&self, receipt_hash: &str) -> Result<Vec<String>> {
772        check_hex(receipt_hash, 64, "receipt_hash")?;
773        self.run("inputs", |c| get_strings(c, L::Inputs, receipt_hash))
774    }
775
776    fn put_namespace(&self, host: &str, keydoc_bytes: &[u8], validated_at: &str) -> Result<()> {
777        keys_from_response(host, 200, Some(keydoc_bytes))?;
778        check_instant(validated_at, "validated_at")?;
779        let canonical = json::parse(keydoc_bytes)?.to_jcs();
780        self.write("put_namespace", |c| {
781            let at: Option<String> =
782                c.prepare_cached(L::NamespaceGet.sql()).db()?.query_row([host], |r| r.get(1)).optional().db()?;
783            if let Some(at) = at {
784                if validated_at < at.as_str() {
785                    return Err(Error::NotLater {
786                        what: "namespace",
787                        key: host.to_string(),
788                        given: validated_at.to_string(),
789                        latest: at,
790                    });
791                }
792            }
793            exec(c, L::NamespacePut, rusqlite::params![host, canonical, validated_at])?;
794            Ok(())
795        })?;
796        tracing::debug!(target: "datumesh_store", host, validated_at, "put_namespace");
797        Ok(())
798    }
799
800    fn namespace(&self, host: &str) -> Result<Option<Namespace>> {
801        check_dns_name(host, "host")?;
802        self.run("namespace", |c| {
803            c.prepare_cached(L::NamespaceGet.sql())
804                .db()?
805                .query_row([host], |r| {
806                    Ok(Namespace {
807                        host: host.to_string(),
808                        keydoc: r.get(0)?,
809                        validated_at: r.get(1)?,
810                        suspended_at: r.get(2)?,
811                    })
812                })
813                .optional()
814                .db()
815        })
816    }
817
818    fn bind_name(&self, fn_name: &str, fn_hash: &str, tx: &str) -> Result<()> {
819        check_function_name(fn_name, "fn_name")?;
820        check_hex(fn_hash, 64, "fn_hash")?;
821        check_instant(tx, "tx")?;
822        self.write("bind_name", |c| {
823            if !exists(c, L::DescriptorExists, fn_hash)? {
824                return Err(Error::NotFound { what: "descriptor", key: fn_hash.to_string() });
825            }
826            if let Some(l) = get_strings(c, L::NameLatest, fn_name)?.pop() {
827                if tx <= l.as_str() {
828                    return Err(Error::NotLater {
829                        what: "name",
830                        key: fn_name.to_string(),
831                        given: tx.to_string(),
832                        latest: l,
833                    });
834                }
835            }
836            exec(c, L::NamePut, rusqlite::params![fn_name, tx, fn_hash])?;
837            Ok(())
838        })?;
839        tracing::debug!(target: "datumesh_store", fn_name, fn_hash, tx, "bind_name");
840        Ok(())
841    }
842
843    fn resolve_name(&self, fn_name: &str, asof: &str) -> Result<Option<String>> {
844        check_function_name(fn_name, "fn_name")?;
845        check_instant(asof, "asof")?;
846        self.run("resolve_name", |c| {
847            c.prepare_cached(L::NameAsof.sql()).db()?.query_row([fn_name, asof], |r| r.get(0)).optional().db()
848        })
849    }
850
851    fn resolve_binding(&self, fn_name: &str, asof: &str) -> Result<Option<Binding>> {
852        check_function_name(fn_name, "fn_name")?;
853        check_instant(asof, "asof")?;
854        self.run("resolve_binding", |c| {
855            c.prepare_cached(L::BindingAsof.sql())
856                .db()?
857                .query_row([fn_name, asof], |r| Ok(Binding { fn_hash: r.get(0)?, tx: r.get(1)? }))
858                .optional()
859                .db()
860        })
861    }
862
863    fn put_computed(&self, cell_id: &str, tx: &str, receipt_hash: &str) -> Result<()> {
864        check_hex(cell_id, 64, "cell_id")?;
865        check_instant(tx, "tx")?;
866        check_hex(receipt_hash, 64, "receipt_hash")?;
867        let asof = self.write("put_computed", |c| {
868            let meta: Option<(String, String, String)> = c
869                .prepare_cached(L::ReceiptMeta.sql())
870                .db()?
871                .query_row([receipt_hash], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
872                .optional()
873                .db()?;
874            let Some((r_cell, r_tx, asof)) = meta else {
875                return Err(Error::NotFound { what: "receipt", key: receipt_hash.to_string() });
876            };
877            if r_cell != cell_id {
878                return Err(Error::Mismatch { what: "cell_id", expected: cell_id.to_string(), found: r_cell });
879            }
880            if r_tx != tx {
881                return Err(Error::Mismatch { what: "tx", expected: tx.to_string(), found: r_tx });
882            }
883            let stored: Option<String> = c
884                .prepare_cached(L::ComputedGet.sql())
885                .db()?
886                .query_row([cell_id, tx, asof.as_str()], |r| r.get(0))
887                .optional()
888                .db()?;
889            match stored {
890                Some(h) if h == receipt_hash => {}
891                Some(_) => return Err(Error::Conflict { hash: receipt_hash.to_string() }),
892                None => {
893                    exec(c, L::ComputedPut, rusqlite::params![cell_id, tx, asof, receipt_hash])?;
894                }
895            }
896            Ok(asof)
897        })?;
898        tracing::debug!(target: "datumesh_store", cell_id, tx, asof, receipt_hash, "put_computed");
899        Ok(())
900    }
901
902    fn computed_asof(&self, cell_id: &str, asof: &str) -> Result<Option<Version>> {
903        self.asof_version("computed_asof", L::ComputedAsof, cell_id, asof)
904    }
905
906    fn stored_asof(&self, cell_id: &str, asof: &str) -> Result<Option<Version>> {
907        self.asof_version("stored_asof", L::StoredAsof, cell_id, asof)
908    }
909
910    fn cell_dependents(&self, cell_id: &str) -> Result<Vec<String>> {
911        check_hex(cell_id, 64, "cell_id")?;
912        self.run("cell_dependents", |c| get_strings(c, L::CellDependents, cell_id))
913    }
914
915    /// One query: the version and its receipt's bytes.
916    fn computed_asof_with_receipt(&self, cell_id: &str, asof: &str) -> Result<Option<(Version, Cow<'_, [u8]>)>> {
917        Ok(self
918            .asof_with_receipt("computed_asof_with_receipt", L::ComputedAsofReceipt, cell_id, asof)?
919            .map(|(v, b)| (v, Cow::Owned(b))))
920    }
921
922    /// One query: the version and its receipt's bytes.
923    fn stored_asof_with_receipt(&self, cell_id: &str, asof: &str) -> Result<Option<(Version, Cow<'_, [u8]>)>> {
924        Ok(self
925            .asof_with_receipt("stored_asof_with_receipt", L::StoredAsofReceipt, cell_id, asof)?
926            .map(|(v, b)| (v, Cow::Owned(b))))
927    }
928
929    fn put_manifest(&self, bytes: &[u8]) -> Result<String> {
930        let m = Manifest::from_json(bytes)?;
931        let entry = m.entry();
932        let canonical = m.canonical();
933        let members = m.members();
934        let n = i64::try_from(members.len()).unwrap_or(i64::MAX);
935        self.run("put_manifest", |c| {
936            // The entry hashes the content: a stored entry has these very bytes.
937            if exists(c, L::ManifestExists, &entry)? {
938                return Ok(());
939            }
940            self.locked_write(c, |c| {
941                if exists(c, L::ManifestExists, &entry)? {
942                    return Ok(());
943                }
944                exec(
945                    c,
946                    L::ManifestPut,
947                    rusqlite::params![entry, m.slice().id(), m.members_hash(), m.slice().fn_hash(), n, canonical],
948                )?;
949                let mut st = c.prepare_cached(L::ManifestMemberPut.sql()).db()?;
950                for h in members {
951                    st.execute([entry.as_str(), h.as_str()]).db()?;
952                }
953                Ok(())
954            })?;
955            self.bound_wal(c);
956            Ok(())
957        })?;
958        tracing::debug!(target: "datumesh_store", entry = %entry, members = n, "put_manifest");
959        Ok(entry)
960    }
961
962    fn member_manifests(&self, receipt_hash: &str) -> Result<Vec<String>> {
963        check_hex(receipt_hash, 64, "receipt_hash")?;
964        self.run("member_manifests", |c| get_strings(c, L::MemberManifests, receipt_hash))
965    }
966
967    fn slice_entries(&self, receipt_hash: &str) -> Result<Vec<String>> {
968        check_hex(receipt_hash, 64, "receipt_hash")?;
969        self.run("slice_entries", |c| get_strings(c, L::SliceEntries, receipt_hash))
970    }
971
972    fn slice_dependents(&self, entry: &str) -> Result<Vec<String>> {
973        check_entry(entry, "entry")?;
974        self.run("slice_dependents", |c| get_strings(c, L::SliceDependents, entry))
975    }
976
977    fn slice_members(&self, slice: &Slice, asof: &str) -> Result<Vec<SliceMember>> {
978        check_instant(asof, "asof")?;
979        let v = slice.to_value();
980        if let Some((cell_id, from, to)) = version_slice(slice, &v)? {
981            return self.versions_between(&cell_id, &from, &to, asof);
982        }
983        let p = Pattern::of(&v)?;
984        self.run("slice_members", |c| {
985            let rows = rows(&p, c, asof, [L::SliceMembersAll, L::SliceMembersDrive], |r| {
986                let cell_id: String = r.get(4)?;
987                Ok((
988                    SliceMember {
989                        version: Version { cell_id, tx: r.get(0)?, receipt_hash: r.get(1)?, asof: r.get(2)? },
990                        receipt: r.get(3)?,
991                    },
992                    r.get::<_, String>(5)?,
993                ))
994            })?;
995            in_key_order(&p, rows, |m| &m.version.cell_id)
996        })
997    }
998
999    fn versions_between(&self, cell_id: &str, from: &str, to: &str, asof: &str) -> Result<Vec<SliceMember>> {
1000        check_hex(cell_id, 64, "cell_id")?;
1001        check_instant(from, "from")?;
1002        check_instant(to, "to")?;
1003        check_instant(asof, "asof")?;
1004        self.run("versions_between", |c| {
1005            let mut st = c.prepare_cached(L::VersionsBetween.sql()).db()?;
1006            let rows = st
1007                .query_map(rusqlite::params![cell_id, from, to, asof], |r| {
1008                    Ok(SliceMember { version: version_row(r, cell_id)?, receipt: r.get(3)? })
1009                })
1010                .db()?;
1011            rows.collect::<rusqlite::Result<Vec<SliceMember>>>().db()
1012        })
1013    }
1014
1015    fn slice_values(&self, slice: &Slice, asof: &str) -> Result<Vec<SliceValue>> {
1016        check_instant(asof, "asof")?;
1017        let v = slice.to_value();
1018        if let Some((cell_id, from, to)) = version_slice(slice, &v)? {
1019            return self.run("slice_values", |c| {
1020                let mut st = c.prepare_cached(L::VersionsBetweenValues.sql()).db()?;
1021                let rows = st.query_map(rusqlite::params![cell_id, from, to, asof], value_columns).db()?;
1022                let mut out = Vec::new();
1023                for r in rows {
1024                    out.push(value_of(r.db()?)?.0);
1025                }
1026                Ok(out)
1027            });
1028        }
1029        let p = Pattern::of(&v)?;
1030        self.run("slice_values", |c| {
1031            let rows = rows(&p, c, asof, [L::SliceValuesAll, L::SliceValuesDrive], value_columns)?;
1032            let mut values = Vec::with_capacity(rows.len());
1033            for row in rows {
1034                values.push(value_of(row)?);
1035            }
1036            in_key_order_parsed(&p, values, |m| &m.version.cell_id)
1037        })
1038    }
1039
1040    fn slice_changed_after(&self, slice: &Slice, asof: &str) -> Result<bool> {
1041        check_instant(asof, "asof")?;
1042        let v = slice.to_value();
1043        if let Some((cell_id, from, to)) = version_slice(slice, &v)? {
1044            return self.run("slice_changed_after", |c| {
1045                c.prepare_cached(L::VersionsChangedAfter.sql())
1046                    .db()?
1047                    .query_row(rusqlite::params![cell_id, from, to, asof], |r| r.get(0))
1048                    .db()
1049            });
1050        }
1051        let p = Pattern::of(&v)?;
1052        self.run("slice_changed_after", |c| changed_after(&p, c, asof))
1053    }
1054}
1055
1056/// The columns of a value row: (tx, receipt_hash, asof, value_type, value, cell_id,
1057/// arguments, ns, fn, fn_hash).
1058type ValueColumns = (String, String, String, String, String, String, String, String, String, String);
1059
1060fn value_columns(r: &rusqlite::Row<'_>) -> rusqlite::Result<ValueColumns> {
1061    Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?, r.get(7)?, r.get(8)?, r.get(9)?))
1062}
1063
1064/// A member's value and cell, rebuilt from the typed-argument index (with its arguments).
1065fn value_of(c: ValueColumns) -> Result<(SliceValue, Args)> {
1066    let (tx, receipt_hash, asof, ty, value, cell_id, args, ns, fn_name, fn_hash) = c;
1067    let ty = Type::from_name(&ty).ok_or_else(|| bad("type"))?;
1068    let args = args_of(&args)?;
1069    let cell = Cell { args: args.clone(), fn_hash, fn_name, ns };
1070    Ok((
1071        SliceValue {
1072            version: Version { cell_id, tx, receipt_hash, asof },
1073            value: Scalar::from_lexical(ty, &value)?,
1074            cell,
1075        },
1076        args,
1077    ))
1078}
1079
1080/// The constraints of a slice as the queries take them (`:cons`): `[name, kind, type, lo,
1081/// hi]`, a number range's ends as sort keys, a bound value's lexical form as `lo`.
1082fn constraints(p: &Pattern) -> String {
1083    let items: Vec<Value> = p
1084        .args
1085        .iter()
1086        .map(|a| {
1087            let (lo, hi) = match a.kind {
1088                Kind::Range => (sort_key(&a.ty, &a.lo), sort_key(&a.ty, &a.hi)),
1089                _ => (a.lo.clone(), a.hi.clone()),
1090            };
1091            Value::Array(
1092                [a.name.clone(), a.kind.as_str().to_string(), a.ty.clone(), lo, hi]
1093                    .into_iter()
1094                    .map(Value::String)
1095                    .collect(),
1096            )
1097        })
1098        .collect();
1099    String::from_utf8(Value::Array(items).to_jcs()).unwrap_or_default()
1100}
1101
1102/// Binds the named parameters a statement has, from `params` (skipping the others).
1103fn bind(st: &mut rusqlite::CachedStatement<'_>, params: &[(&str, SqlValue)]) -> Result<()> {
1104    for (name, value) in params {
1105        if let Some(i) = st.parameter_index(name).db()? {
1106            st.raw_bind_parameter(i, value).db()?;
1107        }
1108    }
1109    Ok(())
1110}
1111
1112/// The bound or ranged argument holding the fewest cells (probed, up to 10,001, when there
1113/// are several, or when `count` asks), else none (every cell of the function); and the
1114/// number of candidates when probed.
1115fn drive<'p>(p: &'p Pattern, c: &Connection, count: bool) -> Result<(Option<&'p Constraint>, Option<i64>)> {
1116    let probe = |a: Option<&Constraint>| -> Result<i64> {
1117        match a {
1118            None => c.prepare_cached(L::ProbeCells.sql()).db()?.query_row([&p.fn_hash], |r| r.get(0)).db(),
1119            Some(a) => {
1120                let (lo, hi) = (sort_key(&a.ty, &a.lo), sort_key(&a.ty, &a.hi));
1121                c.prepare_cached(L::SliceProbe.sql())
1122                    .db()?
1123                    .query_row(rusqlite::params![p.fn_hash, a.name, a.ty, lo, hi], |r| r.get(0))
1124                    .db()
1125            }
1126        }
1127    };
1128    let constrained: Vec<&Constraint> = p.args.iter().filter(|a| a.kind != Kind::Open).collect();
1129    Ok(match constrained.as_slice() {
1130        [] => (None, if count { Some(probe(None)?) } else { None }),
1131        [one] => (Some(*one), if count { Some(probe(Some(one))?) } else { None }),
1132        several => {
1133            let mut best: Option<(i64, &Constraint)> = None;
1134            for a in several {
1135                let n = probe(Some(a))?;
1136                if best.is_none_or(|(m, _)| n < m) {
1137                    best = Some((n, a));
1138                }
1139            }
1140            (best.map(|(_, a)| a), best.map(|(n, _)| n))
1141        }
1142    })
1143}
1144
1145/// The parameters of a slice query: the slice's, and the driving argument's range.
1146fn params(p: &Pattern, asof: &str, drive: Option<&Constraint>) -> Vec<(&'static str, SqlValue)> {
1147    let text = |s: &str| SqlValue::Text(s.to_string());
1148    let mut out = vec![
1149        (":fn_hash", text(&p.fn_hash)),
1150        (":ns", text(&p.ns)),
1151        (":fn", text(&p.fn_name)),
1152        (":nargs", SqlValue::Integer(p.args.len() as i64)),
1153        (":asof", text(asof)),
1154        (":cons", SqlValue::Text(constraints(p))),
1155    ];
1156    if let Some(a) = drive {
1157        out.push((":d_name", text(&a.name)));
1158        out.push((":d_type", text(&a.ty)));
1159        out.push((":d_lo", SqlValue::Text(sort_key(&a.ty, &a.lo))));
1160        out.push((":d_hi", SqlValue::Text(sort_key(&a.ty, &a.hi))));
1161    }
1162    out
1163}
1164
1165/// The rows of one of the member queries `qs` (all cells, driven), each made by `make`.
1166fn rows<T>(
1167    p: &Pattern,
1168    c: &Connection,
1169    asof: &str,
1170    qs: [L; 2],
1171    make: impl Fn(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
1172) -> Result<Vec<T>> {
1173    let (d, _) = drive(p, c, false)?;
1174    let q = if d.is_some() { qs[1] } else { qs[0] };
1175    let mut st = c.prepare_cached(q.sql()).db()?;
1176    bind(&mut st, &params(p, asof, d))?;
1177    let mut rows = st.raw_query();
1178    let mut out = Vec::new();
1179    while let Some(r) = rows.next().db()? {
1180        out.push(make(r).db()?);
1181    }
1182    Ok(out)
1183}
1184
1185/// The key of a member: the "type:value" of its open and ranged arguments, by name.
1186fn key_of(p: &Pattern, args: &Args) -> Vec<String> {
1187    p.args
1188        .iter()
1189        .filter(|a| a.kind != Kind::Value)
1190        .filter_map(|a| args.get(&a.name).and_then(Arg::scalar))
1191        .map(|s| format!("{}:{}", s.ty().as_str(), s.lexical()))
1192        .collect()
1193}
1194
1195/// Members with their argument lists (JSON), in key order (then by cell id).
1196fn in_key_order<T>(p: &Pattern, rows: Vec<(T, String)>, cell_of: impl Fn(&T) -> &String) -> Result<Vec<T>> {
1197    let mut keyed = Vec::with_capacity(rows.len());
1198    for (m, args) in rows {
1199        keyed.push((key_of(p, &args_of(&args)?), m));
1200    }
1201    keyed.sort_by(|(a, ma), (b, mb)| cmp_keys(a, b).then_with(|| cell_of(ma).cmp(cell_of(mb))));
1202    Ok(keyed.into_iter().map(|(_, m)| m).collect())
1203}
1204
1205/// Members with their parsed arguments, in key order (then by cell id).
1206fn in_key_order_parsed<T>(p: &Pattern, rows: Vec<(T, Args)>, cell_of: impl Fn(&T) -> &String) -> Result<Vec<T>> {
1207    let mut keyed: Vec<(Vec<String>, T)> = rows.into_iter().map(|(m, args)| (key_of(p, &args), m)).collect();
1208    keyed.sort_by(|(a, ma), (b, mb)| cmp_keys(a, b).then_with(|| cell_of(ma).cmp(cell_of(mb))));
1209    Ok(keyed.into_iter().map(|(_, m)| m).collect())
1210}
1211
1212/// Whether a matching cell has a stored version answering from after `asof`: from
1213/// whichever side is smaller (both probed, up to 10,001), the function's receipts with an
1214/// asof after `asof`, or the slice's candidate cells.
1215fn changed_after(p: &Pattern, c: &Connection, asof: &str) -> Result<bool> {
1216    let (d, candidates) = drive(p, c, true)?;
1217    let recent: i64 = c
1218        .prepare_cached(L::ProbeRecent.sql())
1219        .db()?
1220        .query_row(rusqlite::params![p.fn_hash, asof], |r| r.get(0))
1221        .db()?;
1222    let q = if recent <= candidates.unwrap_or(i64::MAX) {
1223        L::ChangedRecent
1224    } else if d.is_some() {
1225        L::ChangedCandDrive
1226    } else {
1227        L::ChangedCandAll
1228    };
1229    let d = if q == L::ChangedCandDrive { d } else { None };
1230    let mut st = c.prepare_cached(q.sql()).db()?;
1231    bind(&mut st, &params(p, asof, d))?;
1232    let mut rows = st.raw_query();
1233    Ok(rows.next().db()?.is_some())
1234}
1235
1236#[cfg(test)]
1237mod tests {
1238    use super::*;
1239    use crate::slices::cmp_decimal;
1240
1241    #[test]
1242    fn decimal_keys_sort_as_the_numbers() {
1243        let mut numbers: Vec<String> = [
1244            "0", "9", "10", "-1.5", "-1", "-1.25", "-1.11", "-1.1", "-0.1", "-0.099", "0.099", "0.1", "0.109", "0.11",
1245            "1", "1.1", "1.109", "1.11", "9.5", "99.999", "100", "-100", "-10", "-9", "123", "-123", "0.0001",
1246            "-0.0001", "1000000", "-1000000", "12.5", "12.05", "-12.5", "-12.05", "120", "-120", "1.2", "-1.2",
1247        ]
1248        .iter()
1249        .map(|s| s.to_string())
1250        .collect();
1251        numbers.push("9".repeat(128));
1252        numbers.push(format!("-{}", "9".repeat(127)));
1253        numbers.push(format!("1{}", "0".repeat(127)));
1254        numbers.push(format!("-1{}", "0".repeat(126)));
1255        numbers.push(format!("0.{}1", "0".repeat(125)));
1256        numbers.push(format!("-0.{}1", "0".repeat(124)));
1257        numbers.push(format!("{}.{}", "1".repeat(64), "1".repeat(63)));
1258        numbers.push(format!("-{}.{}", "1".repeat(63), "1".repeat(63)));
1259        for a in &numbers {
1260            assert!(a.len() <= 128, "{a}");
1261            for b in &numbers {
1262                assert_eq!(decimal_key(a).cmp(&decimal_key(b)), cmp_decimal(a, b), "{a} vs {b}");
1263            }
1264        }
1265        // Equal numbers, equal keys; the forms.
1266        assert_eq!(decimal_key("0"), "1");
1267        assert_eq!(decimal_key("100"), "25031");
1268        assert_eq!(decimal_key("0.0012"), "249812");
1269        assert_eq!(decimal_key("-1.5"), "049884~");
1270        assert_eq!(decimal_key(&"9".repeat(128)), format!("2628{}", "9".repeat(128)));
1271    }
1272
1273    #[test]
1274    fn result_codes_have_names() {
1275        assert_eq!(result_code_name(5), "SQLITE_BUSY");
1276        assert_eq!(result_code_name(1811), "SQLITE_CONSTRAINT_TRIGGER");
1277        assert_eq!(result_code_name(14), "SQLITE_CANTOPEN");
1278        assert_eq!(result_code_name(10 | (99 << 8)), "SQLITE_IOERR(25354)");
1279    }
1280
1281    #[test]
1282    fn sqlite_urls_name_paths() {
1283        assert_eq!(sqlite_path("sqlite:store.db"), "store.db");
1284        assert_eq!(sqlite_path("sqlite://store.db"), "store.db");
1285        assert_eq!(sqlite_path("sqlite:///var/lib/x.db"), "/var/lib/x.db");
1286        assert_eq!(sqlite_path("/var/lib/x.db"), "/var/lib/x.db");
1287        assert_eq!(sqlite_path("x.db"), "x.db");
1288    }
1289
1290    /// The plan of a statement, one line per step (`EXPLAIN QUERY PLAN`).
1291    fn plan(c: &Connection, sql: &str) -> Vec<String> {
1292        let mut st = c.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap();
1293        let n = st.parameter_count();
1294        for i in 1..=n {
1295            st.raw_bind_parameter(i, "x").unwrap();
1296        }
1297        let mut rows = st.raw_query();
1298        let mut out = Vec::new();
1299        while let Some(r) = rows.next().unwrap() {
1300            out.push(r.get::<_, String>(3).unwrap());
1301        }
1302        out
1303    }
1304
1305    /// The statements of bounded slices and of the probes look tables up by index only: no
1306    /// step scans a table or a whole index (SQLite's counterpart of the PostgreSQL tests'
1307    /// sequential-scan counters). Planned on the migrated schema with statistics of a
1308    /// loaded store's shape (100,000 cells of one function).
1309    #[test]
1310    fn bounded_slices_use_indexes() {
1311        let c = Connection::open_in_memory().unwrap();
1312        for m in crate::migrate::sqlite::MIGRATIONS {
1313            c.execute_batch(m.sql).unwrap();
1314        }
1315        // Statistics as ANALYZE leaves them for 100,000 cells, 2 arguments each.
1316        c.execute_batch(
1317            "ANALYZE sqlite_schema;
1318             INSERT INTO sqlite_stat1 (tbl, idx, stat) VALUES
1319               ('cells', 'cells_fn_hash', '100000 100000 1'),
1320               ('cells', 'sqlite_autoindex_cells_1', '100000 1'),
1321               ('cell_args', 'cell_args_key', '200000 200000 100000 100000 1 1'),
1322               ('cell_args', 'sqlite_autoindex_cell_args_1', '200000 2 1'),
1323               ('versions', 'sqlite_autoindex_versions_1', '100000 1 1'),
1324               ('versions', 'versions_receipt_hash', '100000 1'),
1325               ('receipts', 'sqlite_autoindex_receipts_1', '100000 1'),
1326               ('receipts', 'receipts_cell_tx', '100000 1 1'),
1327               ('receipts', 'receipts_fn_hash_asof', '100000 100000 1'),
1328               ('receipts', 'receipts_cell_asof', '100000 1 1');
1329             ANALYZE sqlite_schema;",
1330        )
1331        .unwrap();
1332        // A scan of the constraints (json_each) or of a subquery's rows is fine; of a table
1333        // or an index, not.
1334        let scans = |s: &str| s.starts_with("SCAN ") && !s.starts_with("SCAN k") && !s.starts_with("SCAN (subquery");
1335        for q in [
1336            L::SliceMembersDrive,
1337            L::SliceValuesDrive,
1338            L::ChangedCandDrive,
1339            L::ChangedRecent,
1340            L::SliceProbe,
1341            L::ProbeRecent,
1342            L::ProbeCells,
1343            L::VersionsBetween,
1344            L::VersionsBetweenValues,
1345            L::StoredAsof,
1346            L::ComputedAsof,
1347            L::CellDependents,
1348        ] {
1349            let steps = plan(&c, q.sql());
1350            assert!(!steps.iter().any(|s| scans(s)), "{q:?} scans:\n{steps:#?}");
1351        }
1352        // The unbounded ones walk the function's cells by index, and nothing else.
1353        for q in [L::SliceMembersAll, L::SliceValuesAll, L::ChangedCandAll] {
1354            let steps = plan(&c, q.sql());
1355            assert!(
1356                steps.iter().any(|s| s.contains("SEARCH c USING") && s.contains("cells_fn_hash")),
1357                "{q:?}: {steps:#?}"
1358            );
1359            assert!(!steps.iter().any(|s| scans(s)), "{q:?} scans:\n{steps:#?}");
1360        }
1361    }
1362}