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.
4
5use std::path::Path;
6
7use omgbase_reconcile::Config;
8use omgbase_store::{BatchItem, BatchOutcome, Store};
9use rusqlite::params;
10use serde_json::Value;
11
12use crate::error::Result;
13use crate::fs::FileSystem;
14
15/// What a checkpoint produced, in batch order.
16#[derive(Clone, Debug, Default, PartialEq, Eq)]
17pub struct CheckpointResult {
18    pub checkpoint_id: String,
19    /// Paths that produced observed commits.
20    pub ingested: Vec<String>,
21    /// Echoes: the hash already matched.
22    pub suppressed: Vec<String>,
23    /// Gone members that had a live doc.
24    pub deleted: Vec<String>,
25    /// Ingested but carrying git conflict markers.
26    pub conflicted: Vec<String>,
27}
28
29impl CheckpointResult {
30    /// `{ checkpoint_id, ingested, suppressed, deleted, conflicted }`.
31    #[must_use]
32    pub fn to_json(&self) -> Value {
33        serde_json::json!({
34            "checkpoint_id": self.checkpoint_id,
35            "ingested": self.ingested,
36            "suppressed": self.suppressed,
37            "deleted": self.deleted,
38            "conflicted": self.conflicted,
39        })
40    }
41}
42
43/// The `files` column of a checkpoint: `[[path, old_hash | null, new_hash |
44/// null], …]` in batch order.
45#[must_use]
46pub fn files_json(outcomes: &[BatchOutcome]) -> Value {
47    Value::Array(
48        outcomes
49            .iter()
50            .map(|o| match o {
51                BatchOutcome::Deleted(d) => {
52                    serde_json::json!([d.path, d.old_hash_hex, Value::Null])
53                }
54                BatchOutcome::Observed(ob) => {
55                    serde_json::json!([ob.path, ob.old_hash_hex, ob.new_hash_hex])
56                }
57            })
58            .collect(),
59    )
60}
61
62/// §4.1: bucket the outcomes, **mint `cp`**, insert the row, sweep the pool
63/// at `ts` (`spec/store` §5.5).
64pub fn finish_checkpoint(
65    store: &mut Store,
66    repo_id: &str,
67    outcomes: &[BatchOutcome],
68    ts: &str,
69    git_head: Option<&str>,
70) -> Result<CheckpointResult> {
71    let checkpoint_id = store.mint("cp");
72    let mut result = CheckpointResult {
73        checkpoint_id: checkpoint_id.clone(),
74        ..CheckpointResult::default()
75    };
76    for o in outcomes {
77        match o {
78            BatchOutcome::Deleted(d) => {
79                if d.doc_id.is_some() {
80                    result.deleted.push(d.path.clone());
81                }
82            }
83            BatchOutcome::Observed(ob) => {
84                if ob.echo {
85                    result.suppressed.push(ob.path.clone());
86                } else if ob.conflicted {
87                    result.conflicted.push(ob.path.clone());
88                } else {
89                    result.ingested.push(ob.path.clone());
90                }
91            }
92        }
93    }
94    store.conn().execute(
95        "INSERT INTO checkpoints (id, repo_id, ts, files, git_head) VALUES (?1, ?2, ?3, ?4, ?5)",
96        params![
97            checkpoint_id,
98            repo_id,
99            ts,
100            files_json(outcomes).to_string(),
101            git_head
102        ],
103    )?;
104    store.sweep_pool(ts)?;
105    Ok(result)
106}
107
108/// §4.1 `process_checkpoint`: read each path under `root` (`None` when
109/// absent), `observe_batch`, record the checkpoint.
110#[allow(clippy::too_many_arguments)]
111pub fn process_checkpoint(
112    store: &mut Store,
113    repo_id: &str,
114    fs: &dyn FileSystem,
115    root: &Path,
116    paths: &[String],
117    ts: &str,
118    git_head: Option<&str>,
119    config: &Config,
120) -> Result<CheckpointResult> {
121    let mut items = Vec::with_capacity(paths.len());
122    for path in paths {
123        items.push(BatchItem {
124            path: path.clone(),
125            source: fs.read(root, path)?,
126        });
127    }
128    let outcomes = store.observe_batch(repo_id, &items, ts, config)?;
129    finish_checkpoint(store, repo_id, &outcomes, ts, git_head)
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135    use crate::fs::MemFileSystem;
136    use omgbase_store::SequentialMinter;
137
138    const TS: &str = "2026-09-26T10:00:00.000Z";
139
140    #[test]
141    fn checkpoint_buckets_and_files_column() {
142        let mut store =
143            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
144        let repo = store.create_repo("fixture").unwrap();
145        let mut fs = MemFileSystem::new();
146        fs.set("a.md", "# A\n\nOne.\n", 1);
147        fs.set("c.md", "<<<<<<< HEAD\nx\n=======\ny\n>>>>>>> b\n", 2);
148        let root = Path::new("/r");
149        let paths: Vec<String> = ["a.md", "c.md", "gone.md"]
150            .iter()
151            .map(|s| (*s).to_owned())
152            .collect();
153        let r = process_checkpoint(
154            &mut store,
155            &repo,
156            &fs,
157            root,
158            &paths,
159            TS,
160            None,
161            &Config::default(),
162        )
163        .unwrap();
164        assert_eq!(r.checkpoint_id, "cp_0");
165        assert_eq!(r.ingested, ["a.md"]);
166        assert_eq!(r.conflicted, ["c.md"]);
167        assert!(r.deleted.is_empty(), "nothing was live at gone.md");
168        assert!(r.suppressed.is_empty());
169
170        // Echo + a real deletion, with a git head.
171        fs.remove("c.md");
172        let r2 = process_checkpoint(
173            &mut store,
174            &repo,
175            &fs,
176            root,
177            &paths,
178            TS,
179            Some("abc123"),
180            &Config::default(),
181        )
182        .unwrap();
183        assert_eq!(r2.checkpoint_id, "cp_1");
184        assert_eq!(r2.suppressed, ["a.md"]);
185        assert_eq!(r2.deleted, ["c.md"]);
186        assert_eq!(r2.to_json()["deleted"], serde_json::json!(["c.md"]));
187
188        let rows: Vec<(String, String, Option<String>)> = {
189            let mut stmt = store
190                .conn()
191                .prepare(
192                    "SELECT id, files, git_head FROM checkpoints WHERE repo_id = ?1 ORDER BY rowid",
193                )
194                .unwrap();
195            stmt.query_map(params![repo], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
196                .unwrap()
197                .map(|r| r.unwrap())
198                .collect()
199        };
200        assert_eq!(rows.len(), 2);
201        let files: Value = serde_json::from_str(&rows[0].1).unwrap();
202        assert_eq!(files.as_array().unwrap().len(), 3);
203        assert_eq!(files[0][0], "a.md");
204        assert_eq!(files[0][1], Value::Null);
205        assert_eq!(files[0][2].as_str().unwrap().len(), 64);
206        assert_eq!(files[2], serde_json::json!(["gone.md", null, null]));
207        assert_eq!(rows[1].2.as_deref(), Some("abc123"));
208        let files2: Value = serde_json::from_str(&rows[1].1).unwrap();
209        assert_eq!(files2[0][1], files2[0][2], "an echo repeats the hash");
210        assert_eq!(files2[1][2], Value::Null);
211        assert!(files2[1][1].is_string(), "the gone member's prior hash");
212        assert!(!rows[0].1.contains(' '), "compact JSON like JSON.stringify");
213    }
214}
omgbase_sync/checkpoint.rs

omgbase_sync/
checkpoint.rs
