flint_lean/

sync.rs

//! The sync verb (v1, HITL — decided §9 Q4), plus the scoped form the
//! `.flint/sync` sentinel invokes (boundary-verbs plan §2.2, D4).
//! Harness- or sentinel-invoked, never background.
//!
//! Sync BEGINS with a full scan: "locally dirty" means dirty per THAT
//! scan against the baseline — never per the last barrier's snapshot
//! (otherwise sync honors a remote delete over the agent's un-scanned
//! latest work: the review's steady-state destruction finding). Policy:
//! locally-dirty wins; remote deletions apply only to locally-clean
//! paths; adds/changes fetch through the store; every skipped apply is
//! a surfaced conflict, never silent.
//!
//! **The scope rule (D4) is a correctness rule, not an optimization.**
//! A scoped sync advances `inst_base` only for the paths it actually
//! applied or verified in scope, and leaves `baseline.seq` /
//! `baseline.manifest_etag` UNTOUCHED. `inst_base` is the three-way
//! merge base; if a scoped sync advanced the whole merge base to
//! bucket-current, every out-of-scope foreign change would look
//! already-integrated to the next merge and would be silently lost from
//! the inbox flow forever. With D4 those changes remain "foreign" and
//! flow through the normal merge → inbox → consume path at the next
//! barrier, untouched from today.

use std::collections::BTreeMap;

use serde::Serialize;

use flint_store::{crc64_nvme, crc64_to_b64, GenerationStamps, PosixStamps, StoreError};

use super::barrier::{mtime_nanos_of, mtime_of, write_file_atomic_in};
use super::state::{BaselineEntry, ConflictRecord};
use super::{inbox, manifest, now_unix, scan, LeanError, LeanResult, Syncer};

#[derive(Debug, Default, Serialize)]
pub struct SyncReport {
    pub applied: Vec<String>,
    pub deleted: Vec<String>,
    pub conflicts: Vec<String>,
    pub seq: u64,
    /// Remote changes seen but deferred to the inbox flow because they
    /// fell outside the requested scope (D4). Zero for a whole-tree
    /// sync.
    #[serde(default)]
    pub out_of_scope_foreign: usize,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scope: Option<Vec<String>>,
}

/// A normalized scope: path prefixes and exact paths, matched on
/// COMPONENT boundaries (`"in"` never matches `internal/`).
#[derive(Debug, Clone)]
pub struct Scope {
    entries: Vec<String>,
}

impl Scope {
    pub fn new(raw: &[String]) -> Scope {
        let mut entries = vec![];
        for e in raw.iter().take(super::sentinel::MAX_SCOPE_ENTRIES) {
            if e.len() > super::sentinel::MAX_SCOPE_ENTRY_LEN {
                continue;
            }
            let norm = e.trim_matches('/').replace('\\', "/");
            if norm.is_empty() || norm.split('/').any(|c| c == ".." || c == ".") {
                continue;
            }
            entries.push(norm);
        }
        Scope { entries }
    }

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    pub fn entries(&self) -> &[String] {
        &self.entries
    }

    /// Component-boundary match: an entry covers itself exactly and
    /// everything beneath it, never a sibling sharing a name prefix.
    pub fn covers(&self, path: &str) -> bool {
        self.entries.iter().any(|e| {
            path == e.as_str()
                || path.strip_prefix(e.as_str()).map(|r| r.starts_with('/')).unwrap_or(false)
        })
    }
}

impl Syncer {
    /// Whole-tree sync — exactly as shipped, including advancing
    /// `seq`/`manifest_etag`.
    pub async fn sync(&mut self) -> LeanResult<SyncReport> {
        self.sync_scoped(None).await
    }

    pub async fn sync_scoped(&mut self, scope: Option<Vec<String>>) -> LeanResult<SyncReport> {
        // An all-rejected scope is REFUSED, never widened. `Scope::new`
        // silently drops entries that are too long, past
        // MAX_SCOPE_ENTRIES, or contain `.`/`..` — so a caller naming
        // three malformed paths used to get `None`, and `in_scope`'s
        // `.unwrap_or(true)` turned that into the WHOLE TREE. A
        // whole-tree sync re-derives against the remote manifest and
        // DELETES local files for remotely-deleted paths, so the failure
        // mode of a typo was maximum privilege. An error must not
        // return a legal value.
        let scope = match scope {
            None => None,
            Some(raw) => {
                let s = Scope::new(&raw);
                if s.is_empty() {
                    return Err(LeanError::State(format!(
                        "sync scope named {} entr{} and NONE survived validation — \
                         refusing, because an empty scope widens to the WHOLE TREE",
                        raw.len(),
                        if raw.len() == 1 { "y" } else { "ies" }
                    )));
                }
                Some(s)
            }
        };
        let in_scope = |path: &str| scope.as_ref().map(|s| s.covers(path)).unwrap_or(true);
        let mut report = SyncReport::default();
        report.scope = scope.as_ref().map(|s| s.entries().to_vec());
        let mut baseline = self.state.load_baseline()?;

        // 1. The scan comes FIRST; dirt is judged against it alone.
        let scanned = scan::scan(&self.cfg.root)?;
        let classified = scan::classify(&scanned, &baseline);
        let locally_dirty = |path: &str| {
            classified.uploads.contains(path)
                || classified.deletes.contains(path)
                || classified.first_absence.contains(path)
        };

        // 2. Remote truth: manifest + inbox overlay (an inbox entry is
        //    a write the manifest has not re-cited yet).
        let loaded = manifest::load(self.store.as_ref(), &self.cfg).await?;
        let (theirs, _metag) = match loaded {
            Some(l) => (l.manifest, Some(l.etag)),
            None => (Default::default(), None),
        };
        let ib = inbox::load(self.store.as_ref(), &self.cfg).await?;
        let mut remote: BTreeMap<String, String> =
            theirs.entries.iter().map(|(p, e)| (p.clone(), e.etag.clone())).collect();
        // The writer's CRC for each remote etag, overlaid in the same
        // order as the etags so the two maps always describe the same
        // object: the manifest's for a cited entry, the inbox entry's
        // for an overlay (`None` if that writer had none to give).
        let mut remote_crc: BTreeMap<String, Option<String>> = theirs
            .entries
            .iter()
            .map(|(p, e)| (p.clone(), Some(e.crc64_b64.clone())))
            .collect();
        for e in &ib.doc.entries {
            remote.insert(e.path.clone(), e.etag.clone());
            remote_crc.insert(e.path.clone(), e.crc64_b64.clone());
        }

        /// Paths whose `inst_base` this sync is entitled to advance:
        /// under D4, exactly those it applied or verified in scope.
        struct Advanced(std::collections::BTreeSet<String>);
        let mut advanced = Advanced(Default::default());

        // 3. Apply adds/changes (remote differs from OUR merge base).
        for (path, etag) in &remote {
            let base = baseline.inst_base.get(path);
            let unchanged_remotely = base.map(|b| b == etag).unwrap_or(false);
            if unchanged_remotely {
                advanced.0.insert(path.clone());
                continue;
            }
            if !in_scope(path) {
                // D4: NOT integrated, NOT advanced — it stays foreign
                // and reaches this workspace through the next barrier's
                // merge → inbox → consume path.
                report.out_of_scope_foreign += 1;
                continue;
            }
            if baseline.entries.get(path).map(|b| &b.etag == etag).unwrap_or(false) {
                advanced.0.insert(path.clone()); // already integrated (e.g. our own publish)
                continue;
            }
            if locally_dirty(path) {
                // The phantom-conflict rule (§2.2): `sync` saves the
                // baseline only at the end, so a crash mid-apply
                // followed by a re-honor makes already-applied paths
                // scan dirty against the stale baseline. Declaring a
                // conflict there would report a conflict for a path
                // whose local bytes ARE the remote bytes, and the path
                // would then re-publish as a spurious generation bump.
                // Compare content identity first.
                // The remote entry's crc must come from the object we
                // would actually apply, not from the manifest: when
                // remote truth is an INBOX overlay, the manifest entry
                // is a generation behind and its crc would never match.
                // One HEAD, only on the dirty-vs-remote path.
                let remote_meta = match self.store.head(&self.cfg.file_key(path)).await {
                    Ok(m) if m.etag == *etag => Some(m),
                    _ => None,
                };
                // The remote's CRC: the backend's attestation when it
                // offers one, else the writer's for this etag — client-
                // computed, so it exists on a backend that attests none
                // (Ozone). Without either the bytes cannot be judged
                // identical, and the path is a conflict as before.
                let want: Option<String> = remote_meta.as_ref().and_then(|m| {
                    m.crc64_b64.clone().or_else(|| remote_crc.get(path).cloned().flatten())
                });
                let local_path = self.cfg.root.join(path);
                let local_crc =
                    std::fs::read(&local_path).ok().map(|b| crc64_to_b64(crc64_nvme(&b)));
                let identical = want.is_some() && want == local_crc;
                if identical {
                    let st = std::fs::metadata(&local_path)?;
                    let stamps =
                        remote_meta.as_ref().and_then(|m| GenerationStamps::from_meta(&m.meta));
                    baseline.entries.insert(
                        path.clone(),
                        BaselineEntry {
                            etag: etag.clone(),
                            generation: stamps
                                .map(|s| s.generation)
                                .or_else(|| theirs.entries.get(path).map(|e| e.generation))
                                .unwrap_or(0),
                            size: st.len(),
                            mtime_unix: mtime_of(&st),
                            mtime_nanos: Some(mtime_nanos_of(&st)),
                            crc64_b64: local_crc,
                        },
                    );
                    advanced.0.insert(path.clone());
                    report.applied.push(path.clone());
                    continue;
                }
                self.state.append_conflict(&ConflictRecord {
                    path: path.clone(),
                    foreign_etag: etag.clone(),
                    preserved_key: None, // remote version stays in the bucket
                    kind: "sync-dirty".into(),
                    at_unix: now_unix(),
                })?;
                report.conflicts.push(path.clone());
                continue;
            }
            let key = self.cfg.file_key(path);
            let fetched = self.store.get_whole(&key, Some(etag)).await;
            let (meta, body) = match fetched {
                Ok(ok) => ok,
                Err(StoreError::PreconditionFailed(_)) | Err(StoreError::NotFound(_)) => {
                    continue; // superseded mid-sync; the next sync sees the newer truth
                }
                Err(e) => return Err(e.into()),
            };
            let mode = PosixStamps::from_meta(&meta.meta).map(|p| p.mode);
            // VERIFIED before it is written, exactly as checkout's fresh
            // fetch is: against the writer's CRC for this etag (the
            // manifest's for a cited entry, the gateway's for an inbox
            // overlay), else the backend's attestation when it offers
            // one. What the baseline records is OURS, over the bytes
            // written; the next citation repair cites that.
            let got = crc64_to_b64(crc64_nvme(&body));
            let want =
                remote_crc.get(path).cloned().flatten().or_else(|| meta.crc64_b64.clone());
            if let Some(want) = want {
                if want != got {
                    return Err(LeanError::State(format!(
                        "sync: {path} (etag {etag}) is cited with CRC-64 {want}, but the bytes \
                         fetched under that etag hash to {got} — the object is corrupt or the \
                         store returned the wrong bytes; refusing to apply it (nothing written)"
                    )));
                }
            }
            if let Err(e) = write_file_atomic_in(&self.cfg.root, path, &body, mode) {
                // Containment refusal: surfaced, never a wedge.
                self.state.append_conflict(&ConflictRecord {
                    path: path.clone(),
                    foreign_etag: etag.clone(),
                    preserved_key: None,
                    kind: format!("sync-refused-containment: {e}"),
                    at_unix: now_unix(),
                })?;
                report.conflicts.push(path.clone());
                continue;
            }
            let local = self.cfg.root.join(path);
            let st = std::fs::metadata(&local)?;
            let stamps = GenerationStamps::from_meta(&meta.meta);
            baseline.entries.insert(
                path.clone(),
                BaselineEntry {
                    etag: meta.etag.clone(),
                    generation: stamps.map(|s| s.generation).unwrap_or(0),
                    size: st.len(),
                    mtime_unix: mtime_of(&st),
                    mtime_nanos: Some(mtime_nanos_of(&st)),
                    crc64_b64: Some(got),
                },
            );
            advanced.0.insert(path.clone());
            report.applied.push(path.clone());
        }

        // 4. Remote deletions: in our merge base, gone from the
        //    manifest, and not overlaid by an inbox entry — apply only
        //    on locally-clean paths.
        let base_paths: Vec<String> = baseline.inst_base.keys().cloned().collect();
        for path in base_paths {
            if remote.contains_key(&path) {
                continue;
            }
            if !in_scope(&path) {
                report.out_of_scope_foreign += 1;
                continue;
            }
            let local = self.cfg.root.join(&path);
            if !local.exists() {
                baseline.entries.remove(&path);
                advanced.0.insert(path.clone());
                continue;
            }
            if locally_dirty(&path) {
                self.state.append_conflict(&ConflictRecord {
                    path: path.clone(),
                    foreign_etag: String::new(),
                    preserved_key: None,
                    kind: "sync-remote-delete-vs-dirty".into(),
                    at_unix: now_unix(),
                })?;
                report.conflicts.push(path.clone());
                continue;
            }
            std::fs::remove_file(&local)?;
            baseline.entries.remove(&path);
            advanced.0.insert(path.clone());
            report.deleted.push(path.clone());
        }

        // 5. Advance the merge base.
        //
        //    Whole-tree: to the manifest we synced against, exactly as
        //    shipped. Scoped (D4): ONLY for the paths this sync applied
        //    or verified, and seq/manifest_etag stay put — otherwise
        //    every out-of-scope foreign change reads as
        //    already-integrated at the next merge and is lost.
        let theirs_base: BTreeMap<String, String> =
            theirs.entries.iter().map(|(p, e)| (p.clone(), e.etag.clone())).collect();
        //    Except where an inbox OVERLAY hid the manifest's version. This
        //    sync judged those paths against the overlay's etag, never the
        //    manifest's, so the base must not claim the manifest's: an
        //    entry can be OLDER than the manifest (the commit that cited
        //    past it drops it only after its CAS), and advancing there
        //    skipped a change the tree never received — for good, since
        //    nothing later compares against it again (the model's
        //    LeanBarrierLeaseSyncOverlayStale). Left where it was, the next
        //    barrier's merge sees the manifest's change as foreign and
        //    brings it in.
        let hidden: std::collections::BTreeSet<&String> = ib
            .doc
            .entries
            .iter()
            .map(|e| &e.path)
            .filter(|p| theirs_base.get(*p) != remote.get(*p))
            .collect();
        match &scope {
            None => {
                let before = std::mem::take(&mut baseline.inst_base);
                baseline.inst_base = theirs_base.clone();
                for path in &hidden {
                    match before.get(*path) {
                        Some(etag) => {
                            baseline.inst_base.insert((*path).clone(), etag.clone());
                        }
                        None => {
                            baseline.inst_base.remove(*path);
                        }
                    }
                }
                baseline.seq = theirs.seq;
                report.seq = theirs.seq;
            }
            Some(_) => {
                for path in &advanced.0 {
                    if hidden.contains(path) {
                        continue;
                    }
                    match theirs_base.get(path) {
                        Some(etag) => {
                            baseline.inst_base.insert(path.clone(), etag.clone());
                        }
                        None => {
                            baseline.inst_base.remove(path);
                        }
                    }
                }
                report.seq = baseline.seq;
            }
        }
        let rescan = scan::scan(&self.cfg.root)?;
        baseline.prev_scan = rescan.keys().cloned().collect();
        // Materialised files before the baseline that vouches for them.
        self.state.sync_tree()?;
        self.state.save_baseline(&baseline)?;
        self.trace("sync", serde_json::json!({"scoped": scope.is_some(), "applied": report.applied.len(),
            "deleted": report.deleted.len(), "conflicts": report.conflicts.len(), "seq": report.seq,
            "hidden": hidden.len()}));
        Ok(report)
    }
}