mkit_server/sql/
schema.rs

1//! The physical schema and its versioned, forward-only migrations.
2//!
3//! Every statement runs on `rusqlite` and on Durable Object `SQLite`: no
4//! `ATTACH`, no `PRAGMA`, no transaction control. The version lives in the
5//! one-row `mkit_schema` table (not `PRAGMA user_version`, which Durable
6//! Objects do not allow).
7//!
8//! Logical layout changes (new key classes, new row kinds) need **no**
9//! physical migration: they are key layouts, versioned by the `v` row
10//! (`store::keys::layout_version`). A physical migration changes only the
11//! `kv` table's shape.
12//!
13//! Physical v1: one `kv` table keyed by `(part, key)`. `part` is the
14//! [`Partition::encode`](crate::Partition::encode) bytes, so one native
15//! file holds every partition (a D34 shard is a `part` value); a Durable
16//! Object holds one partition, and the column is constant there. It is a
17//! `BLOB`, not `TEXT`: the encoding's components end in `0x00`, which
18//! `SQLite` text functions treat as a terminator.
19//!
20//! Physical v2 adds `kv_timers`, a partial index over the timer rows
21//! (`w 00 …`) that lets a backend find each partition's earliest timer.
22//! It is index-only, but a binary built before v2 refuses a v2 database
23//! ("schema is newer than this binary"): roll back only to a v2 binary.
24
25use super::{SqlConn, SqlError, SqlValue, TxFn, count};
26use crate::store::StoreError;
27
28/// One physical migration: its statements run in one transaction, which
29/// then records `version`.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct Migration {
32    /// The schema version this migration reaches.
33    pub version: u32,
34    /// Its statements, in order. Each is idempotent (`IF NOT EXISTS`).
35    pub statements: &'static [&'static str],
36}
37
38/// The version table, created before anything reads it.
39pub const BOOTSTRAP: &str = "CREATE TABLE IF NOT EXISTS mkit_schema \
40     (id INTEGER PRIMARY KEY CHECK (id = 1), version INTEGER NOT NULL)";
41
42const READ_VERSION: &str = "SELECT version FROM mkit_schema WHERE id = 1";
43const WRITE_VERSION: &str = "INSERT INTO mkit_schema (id, version) VALUES (1, ?1) \
44     ON CONFLICT (id) DO UPDATE SET version = excluded.version";
45
46/// Every migration, in ascending version order.
47pub const MIGRATIONS: &[Migration] = &[
48    Migration {
49        version: 1,
50        statements: &[
51            "CREATE TABLE IF NOT EXISTS kv (part BLOB NOT NULL, key BLOB NOT NULL, \
52         value BLOB NOT NULL, PRIMARY KEY (part, key)) WITHOUT ROWID",
53        ],
54    },
55    Migration {
56        version: 2,
57        statements: &[
58            "CREATE INDEX IF NOT EXISTS kv_timers ON kv (key, part) WHERE key >= x'7700' AND key < x'7701'",
59        ],
60    },
61];
62
63/// The schema version this binary expects: the last migration's.
64pub const SCHEMA_VERSION: u32 = 2;
65
66/// The version recorded in the database: 0 for a new one.
67fn stored_version<C: SqlConn>(conn: &C) -> Result<u32, SqlError> {
68    match conn.query(READ_VERSION, &[])?.first() {
69        None => Ok(0),
70        Some(row) => u32::try_from(count(row, 0)?).map_err(|_| SqlError::Corrupt("schema version")),
71    }
72}
73
74/// Check an existing database without applying migrations or writing schema rows.
75///
76/// # Errors
77/// The database's version is not exactly this binary's, or it cannot be read.
78pub fn require_current<C: SqlConn>(conn: &C) -> Result<u32, StoreError> {
79    if conn
80        .query(
81            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'mkit_schema'",
82            &[],
83        )?
84        .is_empty()
85    {
86        return Err(StoreError::Unsupported(format!(
87            "database schema version 0 differs from binary version {SCHEMA_VERSION}; export with a matching binary"
88        ).into()));
89    }
90    let version = stored_version(conn)?;
91    if version != SCHEMA_VERSION {
92        return Err(StoreError::Unsupported(format!(
93            "database schema version {version} differs from binary version {SCHEMA_VERSION}; export with a matching binary"
94        ).into()));
95    }
96    Ok(version)
97}
98
99enum Step {
100    Applied,
101    Done(u32),
102}
103
104/// Bring the database to [`SCHEMA_VERSION`]; returns the version reached.
105/// Each missing migration runs in its own transaction, which re-reads the
106/// version first, so concurrent openers apply it once. Idempotent.
107///
108/// # Errors
109/// [`StoreError::Unsupported`] if the database records a newer version than
110/// this binary's (there is no downgrade; nothing is changed); the engine's
111/// error otherwise.
112pub fn migrate<C: SqlConn>(conn: &C) -> Result<u32, StoreError> {
113    loop {
114        let step: TxFn<C, Step> = Box::new(|c: C| {
115            c.exec(BOOTSTRAP, &[])?;
116            let current = stored_version(&c)?;
117            let Some(next) = MIGRATIONS.iter().find(|m| m.version > current) else {
118                return Ok(Step::Done(current));
119            };
120            for statement in next.statements {
121                c.exec(statement, &[])?;
122            }
123            c.exec(WRITE_VERSION, &[SqlValue::Integer(next.version.into())])?;
124            Ok(Step::Applied)
125        });
126        match conn.transaction(step)? {
127            Step::Applied => {}
128            Step::Done(version) if version > SCHEMA_VERSION => {
129                return Err(StoreError::Unsupported(
130                    "database schema is newer than this binary".into(),
131                ));
132            }
133            Step::Done(version) => return Ok(version),
134        }
135    }
136}
mkit_server/sql/schema.rs

mkit_server/sql/
schema.rs
