omgbase_sync/
recovery.rs

1//! Startup recovery (`spec/sync/README.md` §4.3 "Recovery"): first the
2//! pending materializations are replayed (store 14.0 — the files an `api`
3//! commit still owes are written, not mistaken for divergence), then every
4//! live doc's file is checked against its current revision; a divergent file
5//! is re-ingested as an observed commit, one document at a time (§9).
6
7use std::path::Path;
8
9use omgbase_format::hash::sha256;
10use omgbase_reconcile::Config;
11use omgbase_store::{Origin, Store};
12use rusqlite::params;
13use serde_json::Value;
14
15use crate::error::Result;
16use crate::fs::FileSystem;
17
18/// What recovery found.
19#[derive(Clone, Debug, Default, PartialEq, Eq)]
20pub struct RecoveryResult {
21    /// Step 0: paths whose pending materialization landed (not on the wire).
22    pub materialized: Vec<String>,
23    /// Paths re-ingested.
24    pub healed: Vec<String>,
25    /// Live docs whose file is gone.
26    pub missing: Vec<String>,
27}
28
29impl RecoveryResult {
30    #[must_use]
31    pub fn to_json(&self) -> Value {
32        serde_json::json!({ "healed": self.healed, "missing": self.missing })
33    }
34}
35
36/// Step 0: [`crate::freshness::materialize_pending`]. Then for every live
37/// doc (in row order): a missing file is `missing`; a file whose hash
38/// differs from the current revision's `rendered_hash` (or `file_hash` when
39/// there is no revision) is re-ingested with the reconciling resolver and
40/// listed `healed`.
41pub fn recover_repo(
42    store: &mut Store,
43    repo_id: &str,
44    fs: &dyn FileSystem,
45    root: &Path,
46    ts: &str,
47    config: &Config,
48) -> Result<RecoveryResult> {
49    let materialized = crate::freshness::materialize_pending(store, repo_id, fs, root)?;
50    type Row = (String, Option<Vec<u8>>, Option<Vec<u8>>);
51    let docs: Vec<Row> = {
52        let mut stmt = store.conn().prepare(
53            "SELECT d.path, d.file_hash, r.rendered_hash
54             FROM docs d LEFT JOIN revisions r ON r.rev_id = d.current_rev
55             WHERE d.repo_id = ?1 AND d.deleted_commit IS NULL ORDER BY d.rowid",
56        )?;
57        let rows = stmt.query_map(params![repo_id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
58        rows.collect::<std::result::Result<_, _>>()?
59    };
60    let mut result = RecoveryResult {
61        materialized,
62        ..RecoveryResult::default()
63    };
64    for (path, file_hash, rendered_hash) in docs {
65        let Some(content) = fs.read(root, &path)? else {
66            result.missing.push(path);
67            continue;
68        };
69        let on_disk = sha256(content.as_bytes());
70        let recorded = rendered_hash.or(file_hash);
71        if recorded.is_none_or(|r| r[..] != on_disk[..]) {
72            store.reconciling_ingest(
73                repo_id,
74                &path,
75                &content,
76                ts,
77                Origin::Observed,
78                None,
79                None,
80                config,
81            )?;
82            result.healed.push(path);
83        }
84    }
85    Ok(result)
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91    use crate::fs::MemFileSystem;
92    use omgbase_store::{BatchItem, SequentialMinter};
93
94    const TS: &str = "2026-09-26T10:00:00.000Z";
95
96    #[test]
97    fn heals_divergent_files_and_reports_missing() {
98        let mut store =
99            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
100        let repo = store.create_repo("fixture").unwrap();
101        let cfg = Config::default();
102        let items = [
103            BatchItem::observed("a.md", "# A\n\nOne.\n"),
104            BatchItem::observed("b.md", "# B\n"),
105            BatchItem::observed("c.md", "# C\n"),
106        ];
107        store.observe_batch(&repo, &items, TS, &cfg).unwrap();
108        let mut fs = MemFileSystem::new();
109        fs.set("a.md", "# A\n\nOne, edited.\n", 1);
110        fs.set("b.md", "# B\n", 2);
111        let root = Path::new("/r");
112        let r = recover_repo(&mut store, &repo, &fs, root, TS, &cfg).unwrap();
113        assert_eq!(r.healed, ["a.md"]);
114        assert_eq!(r.missing, ["c.md"]);
115        assert_eq!(r.to_json()["healed"], serde_json::json!(["a.md"]));
116        let commits: i64 = store
117            .conn()
118            .query_row(
119                "SELECT count(*) FROM commits WHERE repo_id = ?1",
120                params![repo],
121                |r| r.get(0),
122            )
123            .unwrap();
124        assert_eq!(commits, 4);
125        let origin: String = store
126            .conn()
127            .query_row("SELECT origin FROM commits WHERE seq = 4", [], |r| r.get(0))
128            .unwrap();
129        assert_eq!(origin, "observed");
130        let doc: String = store
131            .conn()
132            .query_row("SELECT doc_id FROM docs WHERE path = 'a.md'", [], |r| {
133                r.get(0)
134            })
135            .unwrap();
136        assert_eq!(
137            store.reconstruct(&doc).unwrap().as_deref(),
138            Some("# A\n\nOne, edited.\n")
139        );
140        // The heading kept its id: the reconciling resolver was used. Its
141        // `same` verdict is not persisted (store 14.2; through 14.1 this
142        // commit held one `same` row), so the commit's rows are the short
143        // paragraph's replacement alone — a fresh ingest would have written
144        // `inserted` for the heading too, under a new id.
145        let kinds: Vec<String> = {
146            let mut stmt = store
147                .conn()
148                .prepare("SELECT kind FROM dispositions WHERE commit_id = 'c_3' ORDER BY kind")
149                .unwrap();
150            stmt.query_map([], |r| r.get(0))
151                .unwrap()
152                .map(|r| r.unwrap())
153                .collect()
154        };
155        assert_eq!(kinds, ["deleted", "inserted"]);
156        let heading: String = store
157            .conn()
158            .query_row(
159                "SELECT block_id FROM blocks WHERE doc_id = ?1 AND type = 'heading' AND deleted_commit IS NULL",
160                params![doc],
161                |r| r.get(0),
162            )
163            .unwrap();
164        assert_eq!(heading, "b_0");
165        // Idempotent.
166        let again = recover_repo(&mut store, &repo, &fs, root, TS, &cfg).unwrap();
167        assert!(again.healed.is_empty());
168        assert_eq!(again.missing, ["c.md"]);
169    }
170
171    /// Step 0 (2.2): recovery replays the pending rows a crashed request left
172    /// before comparing files, so the file the api commit owed is written —
173    /// not counted as divergence and re-ingested as `observed`.
174    #[test]
175    fn recovery_materializes_pending_rows_first() {
176        /// Reads from the working tree, dies on the first write.
177        struct Crashed(omgbase_store::FsDocStore);
178        impl omgbase_store::DocStore for Crashed {
179            fn exists(&self, p: &str) -> bool {
180                self.0.exists(p)
181            }
182            fn read(&self, p: &str) -> omgbase_store::Result<Option<String>> {
183                self.0.read(p)
184            }
185            fn write(&mut self, p: &str, _b: &str) -> omgbase_store::Result<()> {
186                Err(omgbase_store::Error::Other(format!("crashed before {p}")))
187            }
188            fn rename(&mut self, f: &str, _t: &str) -> omgbase_store::Result<()> {
189                Err(omgbase_store::Error::Other(format!("crashed before {f}")))
190            }
191            fn remove(&mut self, p: &str) -> omgbase_store::Result<()> {
192                Err(omgbase_store::Error::Other(format!("crashed before {p}")))
193            }
194        }
195        let dir = std::env::temp_dir().join(format!(
196            "omgbase-sync-recover-{}-{}",
197            std::process::id(),
198            std::time::SystemTime::now()
199                .duration_since(std::time::UNIX_EPOCH)
200                .map(|d| d.as_nanos())
201                .unwrap_or(0)
202        ));
203        std::fs::create_dir_all(&dir).unwrap();
204        std::fs::write(dir.join("a.md"), "# A\n").unwrap();
205        let mut store =
206            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
207        let repo = store.create_repo("fixture").unwrap();
208        let cfg = Config::default();
209        let fs = crate::fs::RealFileSystem;
210        store
211            .observe_batch(&repo, &[BatchItem::observed("a.md", "# A\n")], TS, &cfg)
212            .unwrap();
213        let ctx = omgbase_store::DocOpContext {
214            repo_id: repo.clone(),
215            actor: Some("agent:test".to_owned()),
216            ts: TS.to_owned(),
217        };
218        let mut set = serde_json::Map::new();
219        set.insert("status".to_owned(), serde_json::json!("done"));
220        store
221            .docs_set_meta(
222                &ctx,
223                &mut Crashed(omgbase_store::FsDocStore::new(&dir)),
224                "a.md",
225                Some(&set),
226                &[],
227            )
228            .unwrap_err();
229        assert_eq!(store.pending_materializations(&repo).unwrap().len(), 1);
230
231        let r = recover_repo(&mut store, &repo, &fs, &dir, TS, &cfg).unwrap();
232        assert_eq!(r.materialized, ["a.md"]);
233        assert!(r.healed.is_empty(), "the file was owed, not divergent");
234        assert!(r.missing.is_empty());
235        assert_eq!(
236            std::fs::read_to_string(dir.join("a.md")).unwrap(),
237            "---\nstatus: done\n---\n\n# A\n"
238        );
239        assert!(store.pending_materializations(&repo).unwrap().is_empty());
240        assert_eq!(
241            r.to_json(),
242            serde_json::json!({ "healed": [], "missing": [] }),
243            "step 0 is not on the wire"
244        );
245        let _ = std::fs::remove_dir_all(&dir);
246    }
247}
omgbase_sync/recovery.rs

omgbase_sync/
recovery.rs
