blockworx_server/
store.rs

1//! The durable log: one append-only table, written by the single-writer
2//! task and nobody else.
3
4use std::{
5    path::Path,
6    time::{SystemTime, UNIX_EPOCH},
7};
8
9use anyhow::{Context, Result, ensure};
10use blockworx_doc::{
11    commit::{Commit, CommitEnvelope},
12    encode,
13    rev::Rev,
14};
15use rusqlite::Connection;
16
17/// Payloads are CBOR, the same bytes the wire carries (decided
18/// 2026-08-16). The cost, accepted with the decision: rows are opaque to
19/// `sqlite3` and `ripgrep`, so "when did this route change" needs a tool
20/// that decodes rather than a grep.
21const SCHEMA: &str = "
22    CREATE TABLE IF NOT EXISTS commits (
23        rev       INTEGER PRIMARY KEY,
24        wall_time INTEGER NOT NULL,
25        payload   BLOB NOT NULL
26    );
27";
28
29pub struct Store {
30    connection: Connection,
31}
32
33impl Store {
34    /// # Errors
35    /// The file cannot be opened or the schema cannot be applied. A
36    /// missing file is created, not an error: that is an empty document.
37    pub fn open(path: &Path) -> Result<Self> {
38        let connection = Connection::open(path)
39            .with_context(|| format!("opening the log at {}", path.display()))?;
40        // WAL so a reader (a future dump tool) never blocks the writer.
41        connection
42            .execute_batch(&format!("PRAGMA journal_mode = WAL;{SCHEMA}"))
43            .context("applying the schema")?;
44        Ok(Self { connection })
45    }
46
47    /// # Errors
48    /// The insert fails. The caller must not publish the commit if so —
49    /// a commit the server has broadcast must survive a restart.
50    pub fn append(&self, rev: Rev, commit: &Commit) -> Result<()> {
51        let payload = encode::to_bytes(&CommitEnvelope::CommitV1(commit.clone()));
52        let rev = i64::try_from(rev.get()).context("a rev beyond i64 — the log outlived SQLite")?;
53        self.connection
54            .execute(
55                "INSERT INTO commits (rev, wall_time, payload) VALUES (?1, ?2, ?3)",
56                (rev, now_millis(), payload),
57            )
58            .with_context(|| format!("appending rev {rev}"))?;
59        Ok(())
60    }
61
62    /// Every commit in rev order, with the rev the row claims.
63    ///
64    /// The rev comes back as a plain integer rather than a [`Rev`]: revs
65    /// are minted by the fold and nothing outside it may forge one, so the
66    /// caller's job is to *check* that replaying reproduces this number,
67    /// not to trust it.
68    ///
69    /// # Errors
70    /// A row that will not decode. Never skipped — half-loading a log
71    /// written by a newer build forks the document silently.
72    pub fn replay(&self) -> Result<Vec<(u64, Commit)>> {
73        let mut statement = self
74            .connection
75            .prepare("SELECT rev, payload FROM commits ORDER BY rev")?;
76        let rows = statement.query_map([], |row| {
77            Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
78        })?;
79
80        let mut log = Vec::new();
81        for row in rows {
82            let (rev, payload) = row?;
83            let rev = u64::try_from(rev).context("a negative rev in the log")?;
84            let CommitEnvelope::CommitV1(commit) = encode::from_bytes(&payload)
85                .with_context(|| format!("decoding the commit at rev {rev}"))?;
86            log.push((rev, commit));
87        }
88        Ok(log)
89    }
90}
91
92/// UTC millis at acceptance. Display only — it orders nothing, because
93/// `Rev` does.
94fn now_millis() -> i64 {
95    SystemTime::now()
96        .duration_since(UNIX_EPOCH)
97        .map(|since| since.as_millis())
98        .unwrap_or_default() as i64
99}
100
101/// Fold a stored log into the authority's state, checking as it goes that
102/// replay reproduces the revs the rows claim.
103///
104/// # Errors
105/// A row that will not decode, will not fold, or whose rev disagrees with
106/// what replaying produced. All three are hard startup errors rather than
107/// skipped rows: a log this process cannot reproduce is one it must not
108/// serve.
109pub fn replay_into(store: &Store, host: &mut blockworx_doc::session::Host) -> Result<()> {
110    for (claimed, commit) in store.replay()? {
111        let minted = host
112            .ingest(&commit)
113            .with_context(|| format!("folding the commit at rev {claimed}"))?;
114        ensure!(
115            minted.get() == claimed,
116            "the log claims rev {claimed} where replaying produces {} — the log has a gap",
117            minted.get(),
118        );
119    }
120    Ok(())
121}
122
123#[cfg(test)]
124mod tests {
125    use super::*;
126    use blockworx_doc::{
127        block_model::{BlockInit, Icon, LabelInit},
128        geometry::{FracVal, GridPoint, GridRect, GridSize},
129        id::BlockId,
130        opcode::{Crud, OpCodes},
131        session::Host,
132        values::{LabelSide, Role},
133    };
134    use uuid::Uuid;
135
136    fn block(byte: u8) -> BlockId {
137        BlockId::from_uuid(Uuid::from_bytes([byte; 16]))
138    }
139
140    fn label(name: &str) -> LabelInit {
141        LabelInit {
142            name: name.into(),
143            side: LabelSide::Bottom,
144            offset: FracVal::from(0.0),
145            hidden: false,
146        }
147    }
148
149    fn create(byte: u8, name: &str) -> Commit {
150        Commit::new(
151            format!("Added {name}"),
152            vec![OpCodes::Block(
153                block(byte),
154                Crud::Create(BlockInit {
155                    parent: BlockId::NULL,
156                    rect: GridRect {
157                        top_left: GridPoint { x: 1, y: 2 },
158                        size: GridSize { w: 3, h: 4 },
159                    },
160                    locked: false,
161                    role: Role::default(),
162                    title: label(name),
163                    type_label: label("kind"),
164                    icon: Icon::default(),
165                }),
166            )],
167        )
168    }
169
170    fn temp_db() -> (tempfile::TempDir, std::path::PathBuf) {
171        let dir = tempfile::tempdir().expect("a temp dir");
172        let path = dir.path().join("log.db");
173        (dir, path)
174    }
175
176    /// A missing file is an empty document, not an error.
177    #[test]
178    fn a_fresh_database_replays_to_the_empty_document() {
179        let (_dir, path) = temp_db();
180        let store = Store::open(&path).expect("a missing file is created");
181        let mut host = Host::default();
182        replay_into(&store, &mut host).expect("an empty log folds");
183        assert_eq!(host.rev(), Rev::ZERO);
184    }
185
186    /// The phase's durability claim: what the server accepted is what a
187    /// restart reproduces, byte for byte.
188    #[test]
189    fn a_restart_refolds_the_same_document() {
190        let (_dir, path) = temp_db();
191        let mut host = Host::default();
192        {
193            let store = Store::open(&path).expect("the log opens");
194            for (byte, name) in [(1, "one"), (2, "two"), (3, "three")] {
195                let commit = create(byte, name);
196                let accepted = host.accept(commit).expect("the fold accepts");
197                store
198                    .append(accepted.rev(), accepted.commit())
199                    .expect("append");
200                host.publish(accepted);
201            }
202        }
203
204        let reopened = Store::open(&path).expect("the log reopens");
205        let mut refolded = Host::default();
206        replay_into(&reopened, &mut refolded).expect("the log refolds");
207
208        assert_eq!(refolded.rev(), host.rev());
209        assert_eq!(
210            refolded.state().content_hash(),
211            host.state().content_hash(),
212            "a restart must reproduce the document exactly, not merely structurally",
213        );
214    }
215
216    /// A row this build cannot decode stops the server rather than being
217    /// skipped: a half-loaded log is a forked document.
218    #[test]
219    fn a_corrupt_row_is_a_startup_error() {
220        let (_dir, path) = temp_db();
221        {
222            let store = Store::open(&path).expect("the log opens");
223            let mut host = Host::default();
224            let accepted = host.accept(create(1, "one")).expect("the fold accepts");
225            store
226                .append(accepted.rev(), accepted.commit())
227                .expect("append");
228            host.publish(accepted);
229            store
230                .connection
231                .execute(
232                    "UPDATE commits SET payload = ?1 WHERE rev = 1",
233                    (vec![0xffu8; 4],),
234                )
235                .expect("corrupting the row");
236        }
237
238        let store = Store::open(&path).expect("the log reopens");
239        let mut host = Host::default();
240        assert!(replay_into(&store, &mut host).is_err());
241    }
242
243    /// A log with a hole is refused: replaying it would silently
244    /// renumber every commit after the gap.
245    #[test]
246    fn a_gap_in_the_log_is_a_startup_error() {
247        let (_dir, path) = temp_db();
248        {
249            let store = Store::open(&path).expect("the log opens");
250            let mut host = Host::default();
251            for (byte, name) in [(1, "one"), (2, "two")] {
252                let accepted = host.accept(create(byte, name)).expect("the fold accepts");
253                store
254                    .append(accepted.rev(), accepted.commit())
255                    .expect("append");
256                host.publish(accepted);
257            }
258            store
259                .connection
260                .execute("DELETE FROM commits WHERE rev = 1", ())
261                .expect("punching a hole");
262        }
263
264        let store = Store::open(&path).expect("the log reopens");
265        let mut host = Host::default();
266        let refused = replay_into(&store, &mut host).expect_err("a gap must not fold");
267        assert!(refused.to_string().contains("gap"), "{refused}");
268    }
269}
blockworx_server/store.rs

blockworx_server/
store.rs
