omgbase_sync/
freshness.rs

1//! The freshness sweep (`spec/sync/README.md` §4.3): the `file_stats` cache
2//! against a filesystem snapshot — as a pure plan ([`sweep_plan`]) and as the
3//! I/O around it; read-only disk drift; the cache rebuild.
4
5use std::collections::{HashMap, HashSet};
6use std::path::Path;
7
8use omgbase_format::hash::{hex, sha256};
9use omgbase_reconcile::Config;
10use omgbase_store::Store;
11use rusqlite::params;
12use serde_json::Value;
13
14use crate::checkpoint::{CheckpointResult, process_checkpoint};
15use crate::error::Result;
16use crate::fs::{FileStat, FileSystem};
17
18/// A `file_stats` row.
19#[derive(Clone, Debug, PartialEq, Eq)]
20pub struct CacheRow {
21    pub path: String,
22    pub mtime_ns: i64,
23    pub size: i64,
24    pub hash: [u8; 32],
25}
26
27/// One walked file with its stat.
28#[derive(Clone, Debug, PartialEq, Eq)]
29pub struct DiskEntry {
30    pub path: String,
31    pub stat: FileStat,
32}
33
34/// The §4.3 decisions over a snapshot.
35#[derive(Clone, Debug, Default, PartialEq, Eq)]
36pub struct SweepPlan {
37    /// Paths not in the cache or whose `(mtime_ns, size)` differs, in walk order.
38    pub candidates: Vec<String>,
39    /// Candidates whose bytes' hash differs from the cache (or have no row).
40    pub changed: Vec<String>,
41    /// Cached paths not on disk, in cache order.
42    pub deletions: Vec<String>,
43    /// Candidates whose hash matched: only the stat is refreshed.
44    pub refreshed: Vec<String>,
45    /// The hash of every candidate (for the cache refresh).
46    pub hashes: HashMap<String, [u8; 32]>,
47}
48
49impl SweepPlan {
50    /// `{ candidates, changed, deletions, refreshed }` (the `pure.json` shape).
51    #[must_use]
52    pub fn to_json(&self) -> Value {
53        serde_json::json!({
54            "candidates": self.candidates,
55            "changed": self.changed,
56            "deletions": self.deletions,
57            "refreshed": self.refreshed,
58        })
59    }
60
61    /// The paths a checkpoint processes: `changed ++ deletions`.
62    #[must_use]
63    pub fn to_ingest(&self) -> Vec<String> {
64        self.changed
65            .iter()
66            .chain(self.deletions.iter())
67            .cloned()
68            .collect()
69    }
70}
71
72/// §4.3 steps 1–2 as a pure function: `hash_of(path)` is called once per
73/// candidate, in candidate order.
74pub fn sweep_plan(
75    cache: &[CacheRow],
76    disk: &[DiskEntry],
77    hash_of: &mut dyn FnMut(&str) -> Result<[u8; 32]>,
78) -> Result<SweepPlan> {
79    let cached: HashMap<&str, &CacheRow> = cache.iter().map(|r| (r.path.as_str(), r)).collect();
80    let seen: HashSet<&str> = disk.iter().map(|d| d.path.as_str()).collect();
81    let mut plan = SweepPlan::default();
82    for d in disk {
83        let differs = cached
84            .get(d.path.as_str())
85            .is_none_or(|c| c.mtime_ns != d.stat.mtime_ns || c.size != d.stat.size);
86        if differs {
87            plan.candidates.push(d.path.clone());
88        }
89    }
90    for c in cache {
91        if !seen.contains(c.path.as_str()) {
92            plan.deletions.push(c.path.clone());
93        }
94    }
95    for path in &plan.candidates {
96        let hash = hash_of(path)?;
97        plan.hashes.insert(path.clone(), hash);
98        if cached.get(path.as_str()).is_none_or(|c| c.hash != hash) {
99            plan.changed.push(path.clone());
100        } else {
101            plan.refreshed.push(path.clone());
102        }
103    }
104    Ok(plan)
105}
106
107/// The repo's `file_stats` rows in row order.
108pub fn load_cache(store: &Store, repo_id: &str) -> Result<Vec<CacheRow>> {
109    let mut stmt = store.conn().prepare(
110        "SELECT path, mtime_ns, size, hash FROM file_stats WHERE repo_id = ?1 ORDER BY rowid",
111    )?;
112    let rows = stmt.query_map(params![repo_id], |r| {
113        let hash: Vec<u8> = r.get(3)?;
114        Ok(CacheRow {
115            path: r.get(0)?,
116            mtime_ns: r.get(1)?,
117            size: r.get(2)?,
118            hash: hash.try_into().unwrap_or([0; 32]),
119        })
120    })?;
121    Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
122}
123
124/// The walk with each file's stat (a file that vanished between the walk and
125/// the stat is skipped).
126pub fn snapshot(fs: &dyn FileSystem, root: &Path) -> Result<Vec<DiskEntry>> {
127    let mut out = Vec::new();
128    for path in fs.walk_markdown(root)? {
129        if let Some(stat) = fs.stat(root, &path)? {
130            out.push(DiskEntry { path, stat });
131        }
132    }
133    Ok(out)
134}
135
136/// Hash the file at `path` (`sha256` of its bytes); `None` when absent.
137fn hash_file(fs: &dyn FileSystem, root: &Path, path: &str) -> Result<Option<[u8; 32]>> {
138    Ok(fs.read(root, path)?.map(|s| sha256(s.as_bytes())))
139}
140
141/// §4.3 step 4 `record_file_stat`: upsert the fresh stat and `hash`; a path
142/// that vanished meanwhile deletes its row.
143pub fn record_file_stat(
144    store: &Store,
145    repo_id: &str,
146    fs: &dyn FileSystem,
147    root: &Path,
148    path: &str,
149    hash: &[u8; 32],
150) -> Result<()> {
151    match fs.stat(root, path)? {
152        None => {
153            store.conn().execute(
154                "DELETE FROM file_stats WHERE repo_id = ?1 AND path = ?2",
155                params![repo_id, path],
156            )?;
157        }
158        Some(st) => {
159            store.conn().execute(
160                "INSERT INTO file_stats (repo_id, path, mtime_ns, size, hash) VALUES (?1, ?2, ?3, ?4, ?5)
161                 ON CONFLICT(repo_id, path) DO UPDATE SET mtime_ns = excluded.mtime_ns, size = excluded.size, hash = excluded.hash",
162                params![repo_id, path, st.mtime_ns, st.size, &hash[..]],
163            )?;
164        }
165    }
166    Ok(())
167}
168
169/// The checkpoint result plus the scan counters.
170#[derive(Clone, Debug, PartialEq, Eq)]
171pub struct SweepResult {
172    pub checkpoint: CheckpointResult,
173    /// Files walked.
174    pub scanned: usize,
175    /// Stat-mismatched (or new) files hashed.
176    pub candidates: usize,
177    /// `ingested`, `deleted` or `conflicted` non-empty.
178    pub changed: bool,
179}
180
181impl SweepResult {
182    /// The checkpoint fields plus `scanned`, `candidates`, `changed`.
183    #[must_use]
184    pub fn to_json(&self) -> Value {
185        let mut v = self.checkpoint.to_json();
186        v["scanned"] = Value::from(self.scanned);
187        v["candidates"] = Value::from(self.candidates);
188        v["changed"] = Value::from(self.changed);
189        v
190    }
191}
192
193/// §4.3: plan over the cache and the snapshot, refresh touched-not-edited
194/// stats, checkpoint `changed ++ deletions` at `ts` (with `git_head` on the
195/// row), refresh the cache for the changed paths and drop the deletions'
196/// rows.
197pub fn freshness_sweep(
198    store: &mut Store,
199    repo_id: &str,
200    fs: &dyn FileSystem,
201    root: &Path,
202    ts: &str,
203    git_head: Option<&str>,
204    config: &Config,
205) -> Result<SweepResult> {
206    let cache = load_cache(store, repo_id)?;
207    let disk = snapshot(fs, root)?;
208    let plan = {
209        let mut hash_of = |path: &str| -> Result<[u8; 32]> {
210            Ok(hash_file(fs, root, path)?.unwrap_or_else(|| sha256(b"")))
211        };
212        sweep_plan(&cache, &disk, &mut hash_of)?
213    };
214    let fresh: HashMap<&str, FileStat> = disk.iter().map(|d| (d.path.as_str(), d.stat)).collect();
215    for path in &plan.refreshed {
216        if let Some(st) = fresh.get(path.as_str()) {
217            store.conn().execute(
218                "UPDATE file_stats SET mtime_ns = ?1, size = ?2 WHERE repo_id = ?3 AND path = ?4",
219                params![st.mtime_ns, st.size, repo_id, path],
220            )?;
221        }
222    }
223    let checkpoint = process_checkpoint(
224        store,
225        repo_id,
226        fs,
227        root,
228        &plan.to_ingest(),
229        ts,
230        git_head,
231        config,
232    )?;
233    for path in &plan.changed {
234        let hash = plan.hashes.get(path).copied().unwrap_or([0; 32]);
235        record_file_stat(store, repo_id, fs, root, path, &hash)?;
236    }
237    for path in &plan.deletions {
238        store.conn().execute(
239            "DELETE FROM file_stats WHERE repo_id = ?1 AND path = ?2",
240            params![repo_id, path],
241        )?;
242    }
243    let changed = !checkpoint.ingested.is_empty()
244        || !checkpoint.deleted.is_empty()
245        || !checkpoint.conflicted.is_empty();
246    Ok(SweepResult {
247        checkpoint,
248        scanned: disk.len(),
249        candidates: plan.candidates.len(),
250        changed,
251    })
252}
253
254/// How the database disagrees with the disk (§4.3 "Disk drift").
255#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
256pub struct DiskDrift {
257    /// Candidates whose live doc's `file_hash` differs from the bytes' hash.
258    pub changed: usize,
259    /// Live docs whose path is not on disk.
260    pub deleted: usize,
261    /// Candidates with no live doc at that path.
262    pub untracked: usize,
263}
264
265impl DiskDrift {
266    #[must_use]
267    pub fn is_clean(&self) -> bool {
268        self.changed == 0 && self.deleted == 0 && self.untracked == 0
269    }
270}
271
272/// Read-only: the same cache and snapshot, counted, touching nothing.
273pub fn detect_disk_drift(
274    store: &Store,
275    repo_id: &str,
276    fs: &dyn FileSystem,
277    root: &Path,
278) -> Result<DiskDrift> {
279    let cache = load_cache(store, repo_id)?;
280    let cached: HashMap<&str, &CacheRow> = cache.iter().map(|r| (r.path.as_str(), r)).collect();
281    let docs: HashMap<String, Option<Vec<u8>>> = {
282        let mut stmt = store.conn().prepare(
283            "SELECT path, file_hash FROM docs WHERE repo_id = ?1 AND deleted_commit IS NULL",
284        )?;
285        let rows = stmt.query_map(params![repo_id], |r| Ok((r.get(0)?, r.get(1)?)))?;
286        rows.collect::<std::result::Result<_, _>>()?
287    };
288    let disk = snapshot(fs, root)?;
289    let seen: HashSet<&str> = disk.iter().map(|d| d.path.as_str()).collect();
290    let mut drift = DiskDrift::default();
291    for d in &disk {
292        let differs = cached
293            .get(d.path.as_str())
294            .is_none_or(|c| c.mtime_ns != d.stat.mtime_ns || c.size != d.stat.size);
295        if !differs {
296            continue;
297        }
298        match docs.get(&d.path) {
299            None => drift.untracked += 1,
300            Some(file_hash) => {
301                let on_disk = hash_file(fs, root, &d.path)?;
302                let same = matches!((file_hash, on_disk), (Some(h), Some(od)) if h[..] == od[..]);
303                if !same {
304                    drift.changed += 1;
305                }
306            }
307        }
308    }
309    for path in docs.keys() {
310        if !seen.contains(path.as_str()) {
311            drift.deleted += 1;
312        }
313    }
314    Ok(drift)
315}
316
317/// Delete the repo's rows, re-stat and re-hash every walked file; returns how
318/// many.
319pub fn rebuild_file_stats(
320    store: &Store,
321    repo_id: &str,
322    fs: &dyn FileSystem,
323    root: &Path,
324) -> Result<usize> {
325    store.conn().execute(
326        "DELETE FROM file_stats WHERE repo_id = ?1",
327        params![repo_id],
328    )?;
329    let paths = fs.walk_markdown(root)?;
330    for path in &paths {
331        if let Some(hash) = hash_file(fs, root, path)? {
332            record_file_stat(store, repo_id, fs, root, path, &hash)?;
333        }
334    }
335    Ok(paths.len())
336}
337
338/// The cache as the fixtures project it: `(path, mtime_ns, size, hash hex)`
339/// by `path`.
340pub fn file_stats_rows(store: &Store, repo_id: &str) -> Result<Vec<(String, i64, i64, String)>> {
341    let mut rows: Vec<(String, i64, i64, String)> = load_cache(store, repo_id)?
342        .into_iter()
343        .map(|c| (c.path, c.mtime_ns, c.size, hex(&c.hash)))
344        .collect();
345    rows.sort_by(|a, b| a.0.as_bytes().cmp(b.0.as_bytes()));
346    Ok(rows)
347}
348
349#[cfg(test)]
350mod tests {
351    use super::*;
352    use crate::fs::MemFileSystem;
353    use omgbase_store::SequentialMinter;
354
355    const TS: &str = "2026-09-26T10:00:00.000Z";
356    const ROOT: &str = "/r";
357
358    fn row(path: &str, mtime_ns: i64, content: &str) -> CacheRow {
359        CacheRow {
360            path: path.to_owned(),
361            mtime_ns,
362            size: content.len() as i64,
363            hash: sha256(content.as_bytes()),
364        }
365    }
366
367    fn entry(path: &str, mtime_ns: i64, content: &str) -> DiskEntry {
368        DiskEntry {
369            path: path.to_owned(),
370            stat: FileStat {
371                mtime_ns,
372                size: content.len() as i64,
373            },
374        }
375    }
376
377    #[test]
378    fn plan_decisions() {
379        let cache = [
380            row("a.md", 1, "A"),
381            row("b.md", 2, "B"),
382            row("gone.md", 3, "G"),
383        ];
384        let disk = [
385            entry("new.md", 9, "N"),
386            entry("a.md", 1, "A"), // unchanged stat: not a candidate
387            entry("b.md", 5, "B"), // touched, same bytes: refreshed
388        ];
389        let mut hashed = Vec::new();
390        let plan = sweep_plan(&cache, &disk, &mut |p: &str| {
391            hashed.push(p.to_owned());
392            Ok(sha256(match p {
393                "new.md" => b"N",
394                "b.md" => b"B",
395                _ => b"?",
396            }))
397        })
398        .unwrap();
399        assert_eq!(plan.candidates, ["new.md", "b.md"]);
400        assert_eq!(plan.changed, ["new.md"]);
401        assert_eq!(plan.refreshed, ["b.md"]);
402        assert_eq!(plan.deletions, ["gone.md"]);
403        assert_eq!(
404            hashed,
405            ["new.md", "b.md"],
406            "only candidates are hashed, in order"
407        );
408        assert_eq!(plan.to_ingest(), ["new.md", "gone.md"]);
409        assert_eq!(
410            plan.to_json(),
411            serde_json::json!({"candidates": ["new.md", "b.md"], "changed": ["new.md"], "deletions": ["gone.md"], "refreshed": ["b.md"]})
412        );
413        // Size change alone is a candidate; a differing hash is changed.
414        let plan = sweep_plan(
415            &[row("a.md", 1, "A")],
416            &[entry("a.md", 1, "AB")],
417            &mut |_| Ok(sha256(b"AB")),
418        )
419        .unwrap();
420        assert_eq!(plan.changed, ["a.md"]);
421        let empty = sweep_plan(&[], &[], &mut |_| unreachable!()).unwrap();
422        assert_eq!(empty, SweepPlan::default());
423    }
424
425    #[test]
426    fn sweep_drift_and_rebuild_over_a_mem_fs() {
427        let mut store =
428            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
429        let repo = store.create_repo("fixture").unwrap();
430        let mut fs = MemFileSystem::new();
431        let root = Path::new(ROOT);
432        fs.set("a.md", "# A\n", 1);
433        fs.set("d/b.md", "# B\n", 2);
434        let cfg = Config::default();
435
436        let drift = detect_disk_drift(&store, &repo, &fs, root).unwrap();
437        assert_eq!(
438            drift,
439            DiskDrift {
440                changed: 0,
441                deleted: 0,
442                untracked: 2
443            }
444        );
445
446        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
447        assert_eq!((r.scanned, r.candidates, r.changed), (2, 2, true));
448        assert_eq!(r.checkpoint.ingested, ["a.md", "d/b.md"]);
449        assert_eq!(r.to_json()["scanned"], 2);
450        let stats = file_stats_rows(&store, &repo).unwrap();
451        assert_eq!(stats.len(), 2);
452        assert_eq!(stats[0].0, "a.md");
453        assert_eq!((stats[0].1, stats[0].2), (1, 4));
454        assert_eq!(stats[0].3, hex(&sha256(b"# A\n")));
455        assert!(
456            detect_disk_drift(&store, &repo, &fs, root)
457                .unwrap()
458                .is_clean()
459        );
460
461        // Quiet sweep: nothing hashed, nothing changed.
462        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
463        assert_eq!((r.scanned, r.candidates, r.changed), (2, 0, false));
464        assert!(r.checkpoint.ingested.is_empty());
465
466        // A touch without an edit: a candidate, refreshed, no ingest.
467        fs.set("a.md", "# A\n", 10);
468        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
469        assert_eq!((r.scanned, r.candidates, r.changed), (2, 1, false));
470        assert!(
471            r.checkpoint.suppressed.is_empty(),
472            "a refreshed path is not even observed"
473        );
474        assert_eq!(file_stats_rows(&store, &repo).unwrap()[0].1, 10);
475
476        // An edit and a deletion.
477        fs.set("a.md", "# A2\n", 11);
478        fs.remove("d/b.md");
479        assert_eq!(
480            detect_disk_drift(&store, &repo, &fs, root).unwrap(),
481            DiskDrift {
482                changed: 1,
483                deleted: 1,
484                untracked: 0
485            }
486        );
487        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
488        assert_eq!(r.checkpoint.ingested, ["a.md"]);
489        assert_eq!(r.checkpoint.deleted, ["d/b.md"]);
490        assert!(r.changed);
491        let stats = file_stats_rows(&store, &repo).unwrap();
492        assert_eq!(stats.len(), 1);
493        assert_eq!(stats[0].3, hex(&sha256(b"# A2\n")));
494
495        // Rebuild from scratch.
496        fs.set("d/b.md", "# B\n", 3);
497        assert_eq!(rebuild_file_stats(&store, &repo, &fs, root).unwrap(), 2);
498        assert_eq!(file_stats_rows(&store, &repo).unwrap().len(), 2);
499        assert_eq!(
500            detect_disk_drift(&store, &repo, &fs, root).unwrap(),
501            DiskDrift {
502                changed: 0,
503                deleted: 0,
504                untracked: 0
505            },
506            "cache says unchanged, so the untracked b.md is invisible to drift"
507        );
508
509        // record_file_stat deletes the row of a vanished file.
510        fs.remove("d/b.md");
511        record_file_stat(&store, &repo, &fs, root, "d/b.md", &[0; 32]).unwrap();
512        assert_eq!(file_stats_rows(&store, &repo).unwrap().len(), 1);
513    }
514}
omgbase_sync/freshness.rs

omgbase_sync/
freshness.rs
