1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
//! 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)
}
}