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/// §4.3 (since 1.1): delete the repo's rows, then walk the tree and record a
318/// row only for a file whose bytes' hash equals its live doc's `file_hash` —
319/// the cache may say "known" only about bytes the store already holds. A file
320/// with no live doc, or whose bytes differ from what was ingested, gets no row,
321/// so the next sweep still sees it as a candidate and drift still reports it.
322/// Returns the number of files walked, recorded or not.
323pub fn rebuild_file_stats(
324    store: &Store,
325    repo_id: &str,
326    fs: &dyn FileSystem,
327    root: &Path,
328) -> Result<usize> {
329    store.conn().execute(
330        "DELETE FROM file_stats WHERE repo_id = ?1",
331        params![repo_id],
332    )?;
333    let live: HashMap<String, Vec<u8>> = {
334        let mut stmt = store.conn().prepare(
335            "SELECT path, file_hash FROM docs
336             WHERE repo_id = ?1 AND deleted_commit IS NULL AND file_hash IS NOT NULL",
337        )?;
338        let rows = stmt.query_map(params![repo_id], |r| Ok((r.get(0)?, r.get(1)?)))?;
339        rows.collect::<std::result::Result<_, _>>()?
340    };
341    let paths = fs.walk_markdown(root)?;
342    for path in &paths {
343        let Some(want) = live.get(path) else {
344            continue; // no live doc: nothing the cache may vouch for
345        };
346        if let Some(hash) = hash_file(fs, root, path)? {
347            if hash[..] == want[..] {
348                record_file_stat(store, repo_id, fs, root, path, &hash)?;
349            }
350        }
351    }
352    Ok(paths.len())
353}
354
355/// The cache as the fixtures project it: `(path, mtime_ns, size, hash hex)`
356/// by `path`.
357pub fn file_stats_rows(store: &Store, repo_id: &str) -> Result<Vec<(String, i64, i64, String)>> {
358    let mut rows: Vec<(String, i64, i64, String)> = load_cache(store, repo_id)?
359        .into_iter()
360        .map(|c| (c.path, c.mtime_ns, c.size, hex(&c.hash)))
361        .collect();
362    rows.sort_by(|a, b| a.0.as_bytes().cmp(b.0.as_bytes()));
363    Ok(rows)
364}
365
366#[cfg(test)]
367mod tests {
368    use super::*;
369    use crate::fs::MemFileSystem;
370    use omgbase_store::SequentialMinter;
371
372    const TS: &str = "2026-09-26T10:00:00.000Z";
373    const ROOT: &str = "/r";
374
375    fn row(path: &str, mtime_ns: i64, content: &str) -> CacheRow {
376        CacheRow {
377            path: path.to_owned(),
378            mtime_ns,
379            size: content.len() as i64,
380            hash: sha256(content.as_bytes()),
381        }
382    }
383
384    fn entry(path: &str, mtime_ns: i64, content: &str) -> DiskEntry {
385        DiskEntry {
386            path: path.to_owned(),
387            stat: FileStat {
388                mtime_ns,
389                size: content.len() as i64,
390            },
391        }
392    }
393
394    #[test]
395    fn plan_decisions() {
396        let cache = [
397            row("a.md", 1, "A"),
398            row("b.md", 2, "B"),
399            row("gone.md", 3, "G"),
400        ];
401        let disk = [
402            entry("new.md", 9, "N"),
403            entry("a.md", 1, "A"), // unchanged stat: not a candidate
404            entry("b.md", 5, "B"), // touched, same bytes: refreshed
405        ];
406        let mut hashed = Vec::new();
407        let plan = sweep_plan(&cache, &disk, &mut |p: &str| {
408            hashed.push(p.to_owned());
409            Ok(sha256(match p {
410                "new.md" => b"N",
411                "b.md" => b"B",
412                _ => b"?",
413            }))
414        })
415        .unwrap();
416        assert_eq!(plan.candidates, ["new.md", "b.md"]);
417        assert_eq!(plan.changed, ["new.md"]);
418        assert_eq!(plan.refreshed, ["b.md"]);
419        assert_eq!(plan.deletions, ["gone.md"]);
420        assert_eq!(
421            hashed,
422            ["new.md", "b.md"],
423            "only candidates are hashed, in order"
424        );
425        assert_eq!(plan.to_ingest(), ["new.md", "gone.md"]);
426        assert_eq!(
427            plan.to_json(),
428            serde_json::json!({"candidates": ["new.md", "b.md"], "changed": ["new.md"], "deletions": ["gone.md"], "refreshed": ["b.md"]})
429        );
430        // Size change alone is a candidate; a differing hash is changed.
431        let plan = sweep_plan(
432            &[row("a.md", 1, "A")],
433            &[entry("a.md", 1, "AB")],
434            &mut |_| Ok(sha256(b"AB")),
435        )
436        .unwrap();
437        assert_eq!(plan.changed, ["a.md"]);
438        let empty = sweep_plan(&[], &[], &mut |_| unreachable!()).unwrap();
439        assert_eq!(empty, SweepPlan::default());
440    }
441
442    #[test]
443    fn sweep_drift_and_rebuild_over_a_mem_fs() {
444        let mut store =
445            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
446        let repo = store.create_repo("fixture").unwrap();
447        let mut fs = MemFileSystem::new();
448        let root = Path::new(ROOT);
449        fs.set("a.md", "# A\n", 1);
450        fs.set("d/b.md", "# B\n", 2);
451        let cfg = Config::default();
452
453        let drift = detect_disk_drift(&store, &repo, &fs, root).unwrap();
454        assert_eq!(
455            drift,
456            DiskDrift {
457                changed: 0,
458                deleted: 0,
459                untracked: 2
460            }
461        );
462
463        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
464        assert_eq!((r.scanned, r.candidates, r.changed), (2, 2, true));
465        assert_eq!(r.checkpoint.ingested, ["a.md", "d/b.md"]);
466        assert_eq!(r.to_json()["scanned"], 2);
467        let stats = file_stats_rows(&store, &repo).unwrap();
468        assert_eq!(stats.len(), 2);
469        assert_eq!(stats[0].0, "a.md");
470        assert_eq!((stats[0].1, stats[0].2), (1, 4));
471        assert_eq!(stats[0].3, hex(&sha256(b"# A\n")));
472        assert!(
473            detect_disk_drift(&store, &repo, &fs, root)
474                .unwrap()
475                .is_clean()
476        );
477
478        // Quiet sweep: nothing hashed, nothing changed.
479        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
480        assert_eq!((r.scanned, r.candidates, r.changed), (2, 0, false));
481        assert!(r.checkpoint.ingested.is_empty());
482
483        // A touch without an edit: a candidate, refreshed, no ingest.
484        fs.set("a.md", "# A\n", 10);
485        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
486        assert_eq!((r.scanned, r.candidates, r.changed), (2, 1, false));
487        assert!(
488            r.checkpoint.suppressed.is_empty(),
489            "a refreshed path is not even observed"
490        );
491        assert_eq!(file_stats_rows(&store, &repo).unwrap()[0].1, 10);
492
493        // An edit and a deletion.
494        fs.set("a.md", "# A2\n", 11);
495        fs.remove("d/b.md");
496        assert_eq!(
497            detect_disk_drift(&store, &repo, &fs, root).unwrap(),
498            DiskDrift {
499                changed: 1,
500                deleted: 1,
501                untracked: 0
502            }
503        );
504        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
505        assert_eq!(r.checkpoint.ingested, ["a.md"]);
506        assert_eq!(r.checkpoint.deleted, ["d/b.md"]);
507        assert!(r.changed);
508        let stats = file_stats_rows(&store, &repo).unwrap();
509        assert_eq!(stats.len(), 1);
510        assert_eq!(stats[0].3, hex(&sha256(b"# A2\n")));
511
512        // Rebuild from scratch (1.1): a.md's bytes match its live doc and get a
513        // row; d/b.md is back on disk but its doc is tombstoned, so it gets no
514        // row and stays visible as untracked. The count is the walk, not the rows.
515        fs.set("d/b.md", "# B\n", 3);
516        assert_eq!(rebuild_file_stats(&store, &repo, &fs, root).unwrap(), 2);
517        let stats = file_stats_rows(&store, &repo).unwrap();
518        assert_eq!(stats.len(), 1);
519        assert_eq!(stats[0].0, "a.md");
520        assert_eq!(
521            detect_disk_drift(&store, &repo, &fs, root).unwrap(),
522            DiskDrift {
523                changed: 0,
524                deleted: 0,
525                untracked: 1
526            },
527            "a rebuild does not hide an untracked file"
528        );
529        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
530        assert_eq!(r.checkpoint.ingested, ["d/b.md"]);
531        assert_eq!(file_stats_rows(&store, &repo).unwrap().len(), 2);
532
533        // Rebuild over a pending edit: the edited file gets no row, drift and
534        // the sweep still see it; the rebuild's count still walks it.
535        fs.set("a.md", "# A3\n", 12);
536        assert_eq!(rebuild_file_stats(&store, &repo, &fs, root).unwrap(), 2);
537        let stats = file_stats_rows(&store, &repo).unwrap();
538        assert_eq!(stats.len(), 1);
539        assert_eq!(stats[0].0, "d/b.md");
540        assert_eq!(
541            detect_disk_drift(&store, &repo, &fs, root).unwrap(),
542            DiskDrift {
543                changed: 1,
544                deleted: 0,
545                untracked: 0
546            },
547            "a rebuild keeps a pending edit visible"
548        );
549        let r = freshness_sweep(&mut store, &repo, &fs, root, TS, None, &cfg).unwrap();
550        assert_eq!(r.checkpoint.ingested, ["a.md"]);
551        assert_eq!(
552            file_stats_rows(&store, &repo).unwrap()[0].3,
553            hex(&sha256(b"# A3\n"))
554        );
555
556        // record_file_stat deletes the row of a vanished file.
557        fs.remove("d/b.md");
558        record_file_stat(&store, &repo, &fs, root, "d/b.md", &[0; 32]).unwrap();
559        assert_eq!(file_stats_rows(&store, &repo).unwrap().len(), 1);
560    }
561}
omgbase_sync/freshness.rs

omgbase_sync/
freshness.rs
