omgbase_sync/
checkpoint.rs

1//! Checkpoints (`spec/sync/README.md` §4.1): one `checkpoints` row per batch
2//! observed through the filesystem fast path or the adapter driver, then the
3//! pool sweep — the row and the sweep in **one** transaction, so a batch
4//! never leaves a checkpoint whose sweep did not run (or the reverse). Each
5//! member's commit is still its own transaction (`spec/store` §5.4).
6
7use std::path::Path;
8
9use omgbase_reconcile::Config;
10use omgbase_store::{BatchItem, BatchOutcome, Store};
11use rusqlite::params;
12use serde_json::Value;
13
14use crate::error::Result;
15use crate::fs::FileSystem;
16
17/// What a checkpoint produced, in batch order.
18#[derive(Clone, Debug, Default, PartialEq, Eq)]
19pub struct CheckpointResult {
20    pub checkpoint_id: String,
21    /// Paths that produced observed commits.
22    pub ingested: Vec<String>,
23    /// Echoes: the hash already matched.
24    pub suppressed: Vec<String>,
25    /// Gone members that had a live doc.
26    pub deleted: Vec<String>,
27    /// Ingested but carrying git conflict markers.
28    pub conflicted: Vec<String>,
29}
30
31impl CheckpointResult {
32    /// `{ checkpoint_id, ingested, suppressed, deleted, conflicted }`.
33    #[must_use]
34    pub fn to_json(&self) -> Value {
35        serde_json::json!({
36            "checkpoint_id": self.checkpoint_id,
37            "ingested": self.ingested,
38            "suppressed": self.suppressed,
39            "deleted": self.deleted,
40            "conflicted": self.conflicted,
41        })
42    }
43}
44
45/// The `files` column of a checkpoint: `[[path, old_hash | null, new_hash |
46/// null], …]` in batch order.
47#[must_use]
48pub fn files_json(outcomes: &[BatchOutcome]) -> Value {
49    Value::Array(
50        outcomes
51            .iter()
52            .map(|o| match o {
53                BatchOutcome::Deleted(d) => {
54                    serde_json::json!([d.path, d.old_hash_hex, Value::Null])
55                }
56                BatchOutcome::Observed(ob) => {
57                    serde_json::json!([ob.path, ob.old_hash_hex, ob.new_hash_hex])
58                }
59            })
60            .collect(),
61    )
62}
63
64/// §4.1: bucket the outcomes, **mint `cp`**, insert the row and sweep the
65/// pool at `ts` (`spec/store` §5.5) in one transaction: a failing sweep
66/// rolls the row back.
67pub fn finish_checkpoint(
68    store: &mut Store,
69    repo_id: &str,
70    outcomes: &[BatchOutcome],
71    ts: &str,
72    git_head: Option<&str>,
73) -> Result<CheckpointResult> {
74    let checkpoint_id = store.mint("cp")?;
75    let mut result = CheckpointResult {
76        checkpoint_id: checkpoint_id.clone(),
77        ..CheckpointResult::default()
78    };
79    for o in outcomes {
80        match o {
81            BatchOutcome::Deleted(d) => {
82                if d.doc_id.is_some() {
83                    result.deleted.push(d.path.clone());
84                }
85            }
86            BatchOutcome::Observed(ob) => {
87                if ob.echo {
88                    result.suppressed.push(ob.path.clone());
89                } else if ob.conflicted {
90                    result.conflicted.push(ob.path.clone());
91                } else {
92                    result.ingested.push(ob.path.clone());
93                }
94            }
95        }
96    }
97    let tx = store.transaction()?;
98    tx.execute(
99        "INSERT INTO checkpoints (id, repo_id, ts, files, git_head) VALUES (?1, ?2, ?3, ?4, ?5)",
100        params![
101            checkpoint_id,
102            repo_id,
103            ts,
104            files_json(outcomes).to_string(),
105            git_head
106        ],
107    )?;
108    omgbase_store::derived::sweep_pool(&tx, ts)?;
109    tx.commit()?;
110    Ok(result)
111}
112
113/// §4.1 `process_checkpoint`: read each path under `root` (`None` when
114/// absent), `observe_batch`, record the checkpoint.
115#[allow(clippy::too_many_arguments)]
116pub fn process_checkpoint(
117    store: &mut Store,
118    repo_id: &str,
119    fs: &dyn FileSystem,
120    root: &Path,
121    paths: &[String],
122    ts: &str,
123    git_head: Option<&str>,
124    config: &Config,
125) -> Result<CheckpointResult> {
126    let mut items = Vec::with_capacity(paths.len());
127    for path in paths {
128        items.push(BatchItem {
129            path: path.clone(),
130            source: fs.read(root, path)?,
131        });
132    }
133    let outcomes = store.observe_batch(repo_id, &items, ts, config)?;
134    finish_checkpoint(store, repo_id, &outcomes, ts, git_head)
135}
136
137#[cfg(test)]
138mod tests {
139    use super::*;
140    use crate::fs::MemFileSystem;
141    use omgbase_store::SequentialMinter;
142    use rusqlite::Connection;
143
144    const TS: &str = "2026-09-26T10:00:00.000Z";
145
146    #[test]
147    fn checkpoint_buckets_and_files_column() {
148        let mut store =
149            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
150        let repo = store.create_repo("fixture").unwrap();
151        let mut fs = MemFileSystem::new();
152        fs.set("a.md", "# A\n\nOne.\n", 1);
153        fs.set("c.md", "<<<<<<< HEAD\nx\n=======\ny\n>>>>>>> b\n", 2);
154        let root = Path::new("/r");
155        let paths: Vec<String> = ["a.md", "c.md", "gone.md"]
156            .iter()
157            .map(|s| (*s).to_owned())
158            .collect();
159        let r = process_checkpoint(
160            &mut store,
161            &repo,
162            &fs,
163            root,
164            &paths,
165            TS,
166            None,
167            &Config::default(),
168        )
169        .unwrap();
170        assert_eq!(r.checkpoint_id, "cp_0");
171        assert_eq!(r.ingested, ["a.md"]);
172        assert_eq!(r.conflicted, ["c.md"]);
173        assert!(r.deleted.is_empty(), "nothing was live at gone.md");
174        assert!(r.suppressed.is_empty());
175
176        // Echo + a real deletion, with a git head.
177        fs.remove("c.md");
178        let r2 = process_checkpoint(
179            &mut store,
180            &repo,
181            &fs,
182            root,
183            &paths,
184            TS,
185            Some("abc123"),
186            &Config::default(),
187        )
188        .unwrap();
189        assert_eq!(r2.checkpoint_id, "cp_1");
190        assert_eq!(r2.suppressed, ["a.md"]);
191        assert_eq!(r2.deleted, ["c.md"]);
192        assert_eq!(r2.to_json()["deleted"], serde_json::json!(["c.md"]));
193
194        let rows: Vec<(String, String, Option<String>)> = {
195            let mut stmt = store
196                .conn()
197                .prepare(
198                    "SELECT id, files, git_head FROM checkpoints WHERE repo_id = ?1 ORDER BY rowid",
199                )
200                .unwrap();
201            stmt.query_map(params![repo], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
202                .unwrap()
203                .map(|r| r.unwrap())
204                .collect()
205        };
206        assert_eq!(rows.len(), 2);
207        let files: Value = serde_json::from_str(&rows[0].1).unwrap();
208        assert_eq!(files.as_array().unwrap().len(), 3);
209        assert_eq!(files[0][0], "a.md");
210        assert_eq!(files[0][1], Value::Null);
211        assert_eq!(files[0][2].as_str().unwrap().len(), 64);
212        assert_eq!(files[2], serde_json::json!(["gone.md", null, null]));
213        assert_eq!(rows[1].2.as_deref(), Some("abc123"));
214        let files2: Value = serde_json::from_str(&rows[1].1).unwrap();
215        assert_eq!(files2[0][1], files2[0][2], "an echo repeats the hash");
216        assert_eq!(files2[1][2], Value::Null);
217        assert!(files2[1][1].is_string(), "the gone member's prior hash");
218        assert!(!rows[0].1.contains(' '), "compact JSON like JSON.stringify");
219    }
220
221    // ---- transaction hygiene: the statement stream ---------------------------------------
222    //
223    // `Connection::trace_v2` (rusqlite's `trace` feature, dev-only) records every
224    // statement this thread runs; a write "rides a transaction" when a `BEGIN`
225    // precedes it with no `COMMIT`/`ROLLBACK` between, and a `COMMIT` follows.
226
227    thread_local! {
228        static TRACE: std::cell::RefCell<Vec<String>> = const { std::cell::RefCell::new(Vec::new()) };
229    }
230
231    fn record_stmt(e: rusqlite::trace::TraceEvent<'_>) {
232        if let rusqlite::trace::TraceEvent::Stmt(_, sql) = e {
233            TRACE.with(|t| t.borrow_mut().push(sql.trim().to_owned()));
234        }
235    }
236
237    /// Start (or restart) recording this thread's statements on `conn`.
238    fn start_trace(conn: &Connection) {
239        TRACE.with(|t| t.borrow_mut().clear());
240        conn.trace_v2(
241            rusqlite::trace::TraceEventCodes::SQLITE_TRACE_STMT,
242            Some(record_stmt),
243        );
244    }
245
246    fn traced() -> Vec<String> {
247        TRACE.with(|t| t.borrow().clone())
248    }
249
250    /// The index of the `BEGIN` opening the transaction the first statement
251    /// matching `needle` ran in — `None` when it ran in autocommit (no open
252    /// `BEGIN`, or the transaction rolled back). Panics when nothing matches.
253    fn tx_of(log: &[String], needle: &str) -> Option<usize> {
254        let i = log
255            .iter()
256            .position(|s| s.contains(needle))
257            .unwrap_or_else(|| panic!("no statement matching {needle:?} in {log:#?}"));
258        let begin = log[..i].iter().rposition(|s| s.starts_with("BEGIN"))?;
259        if log[begin..i]
260            .iter()
261            .any(|s| s.starts_with("COMMIT") || s.starts_with("ROLLBACK"))
262        {
263            return None;
264        }
265        let end = log[i..]
266            .iter()
267            .position(|s| {
268                s.starts_with("COMMIT") || s.starts_with("ROLLBACK") || s.starts_with("BEGIN")
269            })
270            .map(|k| i + k)?;
271        log[end].starts_with("COMMIT").then_some(begin)
272    }
273
274    fn checkpoint_count(store: &Store) -> i64 {
275        store
276            .conn()
277            .query_row("SELECT count(*) FROM checkpoints", [], |r| r.get(0))
278            .unwrap()
279    }
280
281    /// §4.1: the `checkpoints` row and the pool sweep are one transaction.
282    #[test]
283    fn checkpoint_row_and_sweep_share_one_transaction() {
284        let mut store =
285            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
286        let repo = store.create_repo("fixture").unwrap();
287        start_trace(store.conn());
288        let r = finish_checkpoint(&mut store, &repo, &[], TS, None).unwrap();
289        assert_eq!(r.checkpoint_id, "cp_0");
290        let log = traced();
291        let row = tx_of(&log, "INSERT INTO checkpoints");
292        assert!(row.is_some(), "the row rides a transaction: {log:#?}");
293        assert_eq!(
294            row,
295            tx_of(&log, "DELETE FROM resurrection_pool WHERE expires_ts"),
296            "… the same one as the sweep: {log:#?}"
297        );
298        assert_eq!(checkpoint_count(&store), 1);
299        assert!(store.conn().is_autocommit(), "nothing left open");
300
301        // Through `process_checkpoint` too (a member commit is its own
302        // transaction; the row + sweep follow in theirs).
303        let mut fs = MemFileSystem::new();
304        fs.set("a.md", "# A\n\nOne.\n", 1);
305        start_trace(store.conn());
306        let paths = vec!["a.md".to_owned()];
307        process_checkpoint(
308            &mut store,
309            &repo,
310            &fs,
311            Path::new("/r"),
312            &paths,
313            TS,
314            None,
315            &Config::default(),
316        )
317        .unwrap();
318        let log = traced();
319        let member = tx_of(&log, "INSERT INTO commits");
320        let row = tx_of(&log, "INSERT INTO checkpoints");
321        assert!(member.is_some() && row.is_some(), "{log:#?}");
322        assert_ne!(member, row, "the member's commit is its own transaction");
323        assert_eq!(
324            row,
325            tx_of(&log, "DELETE FROM resurrection_pool WHERE expires_ts")
326        );
327        assert_eq!(checkpoint_count(&store), 2);
328    }
329
330    /// Failure injection: a sweep that fails rolls the checkpoint row back,
331    /// leaving neither half.
332    #[test]
333    fn a_failing_sweep_leaves_no_checkpoint_row() {
334        let mut store =
335            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
336        let repo = store.create_repo("fixture").unwrap();
337        store
338            .conn()
339            .execute_batch("DROP TABLE resurrection_pool")
340            .unwrap();
341        start_trace(store.conn());
342        let err = finish_checkpoint(&mut store, &repo, &[], TS, None).unwrap_err();
343        assert!(
344            err.to_string().contains("resurrection_pool"),
345            "the sweep failed: {err}"
346        );
347        assert_eq!(checkpoint_count(&store), 0, "the row was rolled back");
348        assert!(store.conn().is_autocommit(), "the transaction is closed");
349        let log = traced();
350        assert!(
351            log.iter().any(|s| s.starts_with("INSERT INTO checkpoints")),
352            "the row was attempted: {log:#?}"
353        );
354        assert!(
355            log.iter().any(|s| s.starts_with("ROLLBACK")),
356            "… and rolled back: {log:#?}"
357        );
358        assert!(!log.iter().any(|s| s.starts_with("COMMIT")));
359    }
360}
omgbase_sync/checkpoint.rs

omgbase_sync/
checkpoint.rs
