Skip to main content

flint_lean/
manifest.rs

1//! The lean subtree manifest: one JSON document, CAS-written, the
2//! atomic commit point every reader trusts (checkout, sync, DR).
3//!
4//! Deliberately NOT `tier::manifest`: the hub's writer rebuilds the
5//! whole document from its local walk and cannot merge (its Entry has
6//! no author identity), which is the amputation engine the lean review
7//! reproduced three ways. This writer merges three-way:
8//! base = the manifest view at OUR last install, mine = this barrier's
9//! walk, theirs = bucket-current. Foreign entries are preserved and
10//! surfaced for the next consume; a local delete loses to a foreign
11//! modify (delete/modify resolves conservative — the formal model's
12//! GC-vs-merge counterexample is exactly this case).
13
14use std::collections::{BTreeMap, BTreeSet};
15
16use serde::{Deserialize, Serialize};
17
18use flint_store::{
19    crc64_nvme, GenerationStamps, ObjectMeta, ObjectStore, PutCondition, StoreError,
20};
21
22use super::{LeanConfig, LeanError, LeanResult};
23
24#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
25pub struct LeanEntry {
26    /// Object key (under `<prefix>/files/`).
27    pub key: String,
28    pub etag: String,
29    /// Full-object CRC-64/NVME of the cited bytes, in wire (base64)
30    /// form — computed by the flint client that MOVED the bytes (the
31    /// publisher over what it uploaded, a consume or checkout over what
32    /// it wrote), never copied from a backend header. It is the one
33    /// integrity check that holds on every backend: S3 attests a
34    /// checksum on the wire, Ozone attests none, and the manifest's
35    /// value describes the bytes either way. Every reader verifies a
36    /// fresh fetch against it before the file becomes visible.
37    pub crc64_b64: String,
38    pub size: u64,
39    pub mode: u32,
40    pub mtime_unix: i64,
41    pub generation: u64,
42    /// The publishing writer's lease epoch (0 = a gateway/HITL write).
43    pub epoch: u64,
44}
45
46#[derive(Debug, Clone, Default, Serialize, Deserialize)]
47pub struct LeanManifest {
48    pub seq: u64,
49    /// path -> entry. Paths are workspace-relative, '/'-separated.
50    pub entries: BTreeMap<String, LeanEntry>,
51    /// This workspace is published by exactly ONE writer, so an object
52    /// that has moved off its citation was moved by something that is
53    /// not the publisher.
54    ///
55    /// It exists because a reader's default arm on a 412 is an
56    /// explicit S3-wins adopt — right for a lean workspace, where an
57    /// object past its citation means a human wrote newer bytes that
58    /// ought to win. On a MIRROR that reading is exactly inverted: the
59    /// publisher is the only party entitled to write, so "past the
60    /// citation" means a stranger did, and adopting copies bytes no
61    /// manifest cites into a reader's tree with no error and no
62    /// conflict record. Composition drill C4 measured that against
63    /// forge's legible export.
64    ///
65    /// Set by the installing pass from config, never inherited.
66    #[serde(default)]
67    pub sole_writer: bool,
68    /// Which citation source installed this manifest (§2.4.1). Also
69    /// stamped on the object's metadata so it is readable by HEAD.
70    #[serde(default, skip_serializing_if = "Option::is_none")]
71    pub boundary_source: Option<String>,
72    /// What each published delete RETIRED, path -> the etag the document
73    /// cited before the delete, kept until the path is cited again (or
74    /// `TOMBSTONE_KEEP_SEQS` generations pass). Review 2026-09-18, H1e: a
75    /// writer that adopted a UI write while it was still pending, and
76    /// merges after another writer cited it and then deleted it knowingly,
77    /// otherwise sees only "baseline ≠ merge base, object at the key" —
78    /// the same view a delete that RACED the write leaves, where modify
79    /// rightly wins — and the citation repair (removed in step 5) resurrected
80    /// the deleted generation. The tombstone is the one fact that tells the
81    /// two apart.
82    /// Empty on every document written before this field existed.
83    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
84    pub tombstones: BTreeMap<String, Tombstone>,
85}
86
87/// A published delete's record in the document: the generation it retired
88/// and the generation of the document that published it.
89#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
90pub struct Tombstone {
91    pub etag: String,
92    pub seq: u64,
93}
94
95/// How many document generations a tombstone outlives its delete. A writer
96/// whose merge base is older than this can still re-cite what it holds —
97/// the residual, named in SAFETY.md §4.14.
98pub const TOMBSTONE_KEEP_SEQS: u64 = 10_000;
99
100impl LeanManifest {
101    /// COMPACT, not pretty. This document is O(entries) and it moves on
102    /// every path that matters: the pre-marker checkout GET, every
103    /// barrier merge GET and CAS PUT, and every gateway read verb. The
104    /// indentation was ~22% of those bytes and of the parse that
105    /// follows. `mc cat | jq` — the operator's and every rig's only
106    /// debugging surface for the CAS commit point — is unaffected.
107    pub fn to_bytes(&self) -> Vec<u8> {
108        serde_json::to_vec(self).expect("manifest serializes")
109    }
110    pub fn parse(bytes: &[u8]) -> Result<LeanManifest, String> {
111        serde_json::from_slice(bytes).map_err(|e| e.to_string())
112    }
113}
114
115/// The bucket-current manifest + its document ETag (the CAS token).
116/// The pointer document at `<prefix>/.flint/lean/current` — the ONLY
117/// mutable metadata object under the pointer layout (design of record:
118/// `docs/plans/flint-lean-manifest-pointer-design.md`).
119///
120/// Everything a reader needs before it decides whether to fetch entries
121/// at all lives here, so the barrier's idle tick reads a few hundred
122/// bytes instead of a document that runs to 264 MiB at 1M entries.
123#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
124pub struct Pointer {
125    /// The manifest generation this pointer publishes. Bumped by a
126    /// citation AND by a takeover rotation, which is why it is not the
127    /// same number as `entries_seq`.
128    pub seq: u64,
129    /// The single immutable object holding every entry of this
130    /// generation — the step-one layout. `None` on a CHUNKED pointer.
131    ///
132    /// Optional here and REQUIRED in the step-one binary's schema, and
133    /// that asymmetry is the migration: a step-one reader handed a
134    /// chunked pointer fails to parse it and refuses, rather than
135    /// decoding a pointer it half-understands. `manifest::load` maps a
136    /// missing object to `Ok(None)`, `None` means FIRST WRITE, and a
137    /// barrier answers that with `If-None-Match: *` — so every layout
138    /// change has to make an old reader refuse rather than conclude the
139    /// project is empty. **Do not give this field a serde default in a
140    /// released binary; that is what makes the refusal work.**
141    #[serde(default, skip_serializing_if = "Option::is_none")]
142    pub entries_key: Option<String>,
143    /// The `seq` the entries object was written under. A rotation
144    /// leaves this ALONE while bumping `seq`, which is exactly what
145    /// lets a follower skip the entries GET after a takeover.
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub entries_seq: Option<u64>,
148    /// The chunk list — the step-two layout. `Some(vec![])` is a
149    /// legitimately EMPTY project and is distinguishable from absence,
150    /// which is why this is an Option around a Vec rather than a bare
151    /// Vec with a default.
152    #[serde(default, skip_serializing_if = "Option::is_none")]
153    pub chunks: Option<Vec<super::chunk::ChunkRef>>,
154    /// Carried here as well as on the manifest because the pointer is
155    /// the authority for the fields it carries, and a reader that
156    /// reconstructs a manifest from it must not lose the flag that
157    /// decides whether it may adopt.
158    #[serde(default)]
159    pub sole_writer: bool,
160    #[serde(default, skip_serializing_if = "Option::is_none")]
161    pub boundary_source: Option<String>,
162    /// The epoch that installed this pointer. Diagnostic; the fence is
163    /// the CAS, not this field.
164    pub epoch: u64,
165    /// The content address of the document's tombstones in the chunked
166    /// layout (`chunk::TombstoneBody`, at `chunk_key`), `None` when the
167    /// document has none. The single-object layout carries them inline.
168    #[serde(default, skip_serializing_if = "Option::is_none")]
169    pub tombstones: Option<String>,
170}
171
172/// Where a pointer's entries actually live.
173#[derive(Debug)]
174pub enum Entries<'a> {
175    /// One generation object (step one).
176    Single { key: &'a str, seq: u64 },
177    /// A chunk list (step two). Empty = an empty project.
178    Chunked(&'a [super::chunk::ChunkRef]),
179}
180
181impl Pointer {
182    /// Resolve the layout, refusing anything ambiguous.
183    ///
184    /// BOTH forms present, or NEITHER, is malformed and must fail —
185    /// never pick one. A pointer carrying both would let two readers
186    /// that broke the tie differently disagree about the contents of
187    /// the same seq, which is the one thing a single visible object was
188    /// supposed to make impossible.
189    pub fn entries(&self) -> LeanResult<Entries<'_>> {
190        match (&self.entries_key, &self.chunks) {
191            (Some(k), None) => {
192                Ok(Entries::Single { key: k, seq: self.entries_seq.unwrap_or(self.seq) })
193            }
194            (None, Some(c)) => Ok(Entries::Chunked(c)),
195            (Some(_), Some(_)) => Err(super::LeanError::State(format!(
196                "manifest pointer at seq {} names BOTH a generation object and a chunk list; \
197                 refusing to guess which describes this manifest",
198                self.seq
199            ))),
200            (None, None) => Err(super::LeanError::State(format!(
201                "manifest pointer at seq {} names no entries at all — neither a generation \
202                 object nor a chunk list; an empty project is an empty chunk LIST, not a \
203                 missing one",
204                self.seq
205            ))),
206        }
207    }
208}
209
210/// What a writer must present to CAS. Carries the LAYOUT as well as the
211/// tag, because a workspace mid-migration has a legacy manifest and no
212/// pointer yet, and the two cases take different preconditions: an
213/// If-Match on the object the writer read, versus an If-None-Match on a
214/// pointer that must not already exist.
215#[derive(Debug, Clone, PartialEq)]
216pub struct ManifestHandle {
217    pub etag: String,
218    /// True when the document was read from the LEGACY single-object
219    /// key, i.e. this write is the migration.
220    pub legacy: bool,
221    /// The chunk list this write is superseding, empty if the workspace
222    /// is not chunked yet. Carried on the handle rather than re-read at
223    /// publish time because the caller has just loaded it, and because
224    /// a publish that had to GET the pointer again to find out what it
225    /// already knew would pay a request per barrier for nothing.
226    ///
227    /// This is what makes a publish O(changed): a chunk whose address
228    /// is already here is not sent.
229    pub prev_chunks: Vec<super::chunk::ChunkRef>,
230    /// The tombstones object the superseded pointer names, if any: retired
231    /// with the chunks when the new pointer names another (M1).
232    pub prev_tombstones: Option<String>,
233}
234
235/// What the legacy key is overwritten with once a workspace has moved to
236/// the pointer layout. It must NOT parse as a manifest: `LeanManifest`
237/// gives `seq` and `entries` no serde default, so an OLD `flint-sync`
238/// reading this gets `LeanError::State` and refuses.
239///
240/// Deleting the legacy key instead would be a data-loss shape, not a
241/// tidy-up: `manifest::load` maps a missing object to `Ok(None)`, `None`
242/// means FIRST WRITE, and a barrier answers that with
243/// `IfNoneMatchAny` — so an old syncer would conclude the project is
244/// empty and re-seed over it.
245pub const REFUSAL_DOC: &str = concat!(
246    "{\"moved\":\".flint/lean/current\",",
247    "\"note\":\"this workspace uses the manifest pointer layout; upgrade flint-sync\"}"
248);
249
250pub struct LoadedManifest {
251    pub manifest: LeanManifest,
252    /// The CAS handle: the POINTER's etag under the pointer layout, the
253    /// legacy object's etag before migration.
254    pub etag: String,
255    /// The object's own Last-Modified. The GET already carries it; a
256    /// caller that wants the coherence stamp must not pay a second
257    /// request for what this one threw away (review: U31).
258    pub last_modified_unix: Option<u64>,
259    /// `None` ⇒ this workspace is still on the legacy single-object
260    /// layout and the next write migrates it.
261    pub pointer: Option<Pointer>,
262}
263
264impl LoadedManifest {
265    /// The handle a writer presents to CAS over this read.
266    pub fn handle(&self) -> ManifestHandle {
267        ManifestHandle {
268            etag: self.etag.clone(),
269            legacy: self.pointer.is_none(),
270            prev_chunks: match self.pointer.as_ref().map(|p| p.entries()) {
271                Some(Ok(Entries::Chunked(c))) => c.to_vec(),
272                _ => Vec::new(),
273            },
274            prev_tombstones: self.pointer.as_ref().and_then(|p| p.tombstones.clone()),
275        }
276    }
277}
278
279/// Read the pointer alone. The barrier's idle tick wants exactly this
280/// and nothing else: a few hundred bytes that answer "did anything
281/// move, and were the ENTRIES part of it".
282pub struct LoadedPointer {
283    pub pointer: Pointer,
284    pub etag: String,
285    /// The pointer object's own Last-Modified — the citation's clock,
286    /// which /status reports and which the GET already carries.
287    pub last_modified_unix: Option<u64>,
288}
289
290pub async fn load_pointer(
291    store: &dyn ObjectStore,
292    cfg: &LeanConfig,
293) -> LeanResult<Option<LoadedPointer>> {
294    match store.get_whole(&cfg.current_key(), None).await {
295        Ok((meta, bytes)) => {
296            let pointer: Pointer = serde_json::from_slice(&bytes)
297                .map_err(|e| super::LeanError::State(format!("manifest pointer parse: {e}")))?;
298            Ok(Some(LoadedPointer { pointer, etag: meta.etag, last_modified_unix: meta.last_modified_unix }))
299        }
300        Err(StoreError::NotFound(_)) => Ok(None),
301        Err(e) => Err(e.into()),
302    }
303}
304
305/// The whole document: pointer, then the generation it names. Falls
306/// back to the legacy single-object key for a workspace that has not
307/// been migrated — which every existing workspace is, exactly once.
308/// How many times `load` will restart onto a newer generation before
309/// giving up. A restart happens only when a chunk we needed was swept
310/// AND the pointer moved under us, so it is bounded by publish rate,
311/// not by anything we control — three is generous for a workspace whose
312/// floor is measured in seconds.
313const LOAD_ATTEMPTS: u32 = 3;
314
315pub async fn load(
316    store: &dyn ObjectStore,
317    cfg: &LeanConfig,
318) -> LeanResult<Option<LoadedManifest>> {
319    load_from(store, cfg, None).await
320}
321
322/// [`load`], starting from a pointer read the caller ALREADY made
323/// (`Some(None)`: it 404'd), so the first attempt does not GET it again —
324/// the consume's cheap path reads the pointer and, when it has moved or is
325/// absent, needs the document behind that same read. A restart onto a
326/// newer generation reads the pointer afresh, as `load` always did.
327pub async fn load_from(
328    store: &dyn ObjectStore,
329    cfg: &LeanConfig,
330    mut first: Option<Option<LoadedPointer>>,
331) -> LeanResult<Option<LoadedManifest>> {
332  for attempt in 0..LOAD_ATTEMPTS {
333    let read = match first.take() {
334        Some(already) => already,
335        None => load_pointer(store, cfg).await?,
336    };
337    if let Some(LoadedPointer { pointer: p, etag, last_modified_unix: pointer_lm }) = read {
338        // A pointer naming an object that is not there is not an empty
339        // workspace — it is a broken one, and answering `None` would
340        // invite a re-seed over a live project. That rule holds one
341        // level deeper for chunks: a chunk list with a hole is a
342        // manifest missing entries, and a missing entry reads to every
343        // consumer as a file the agent deleted.
344        let (mut manifest, meta_lm) = match p.entries()? {
345            Entries::Single { key, .. } => {
346                let (meta, bytes) = match store.get_whole(key, None).await {
347                    Ok(v) => v,
348                    Err(StoreError::NotFound(_)) => {
349                        return Err(super::LeanError::State(format!(
350                            "manifest pointer at seq {} names {key}, which does not exist — \
351                             refusing to treat a broken pointer as an empty workspace",
352                            p.seq
353                        )))
354                    }
355                    Err(e) => return Err(e.into()),
356                };
357                let m = LeanManifest::parse(&bytes)
358                    .map_err(|e| super::LeanError::State(format!("manifest parse: {e}")))?;
359                (m, meta.last_modified_unix)
360            }
361            Entries::Chunked(refs) => {
362                // Sequential on purpose for now: a chunk fetch is
363                // ordinary and the interesting property is that a hole
364                // FAILS rather than silently shortening the manifest.
365                let mut bodies = Vec::with_capacity(refs.len());
366                let mut raced = false;
367                for r in refs {
368                    let key = cfg.chunk_key(&r.addr);
369                    match store.get_whole(&key, None).await {
370                        Ok((_, b)) => bodies.push(b.to_vec()),
371                        Err(StoreError::NotFound(_)) => {
372                            // A chunk we need is gone. Which of two very
373                            // different things happened is decided by
374                            // the POINTER, not by the chunk:
375                            //
376                            // - the pointer MOVED ⇒ we raced a sweep
377                            //   that collected a generation we were
378                            //   still reading. The current generation is
379                            //   a complete snapshot, so start over
380                            //   against it: coherent, just newer. This
381                            //   is what makes a reader safe for a
382                            //   DURATION rather than for `Retain`
383                            //   publishes (chunked design §8.2,
384                            //   machine-checked by
385                            //   `LeanChunkGCSlowReader`).
386                            // - the pointer is UNCHANGED ⇒ the live
387                            //   manifest genuinely has a hole, and there
388                            //   is nothing newer to restart onto. Fail,
389                            //   and never as `Ok(None)`, which every
390                            //   caller reads as FIRST WRITE.
391                            let moved = match load_pointer(store, cfg).await? {
392                                Some(now) => now.etag != etag,
393                                None => true,
394                            };
395                            if moved && attempt + 1 < LOAD_ATTEMPTS {
396                                raced = true;
397                                break;
398                            }
399                            return Err(super::LeanError::State(format!(
400                                "manifest pointer at seq {} names chunk {} (first key {:?}), \
401                                 which does not exist — refusing to serve a manifest with a \
402                                 hole in it as though those entries had been deleted",
403                                p.seq, r.addr, r.first
404                            )));
405                        }
406                        Err(e) => return Err(e.into()),
407                    }
408                }
409                if raced {
410                    continue;
411                }
412                let entries = super::chunk::assemble(refs, &bodies)?;
413                let tombstones = match &p.tombstones {
414                    None => BTreeMap::new(),
415                    Some(addr) => match store.get_whole(&cfg.chunk_key(addr), None).await {
416                        Ok((_, b)) => super::chunk::decode_tombstones(&b, addr)?,
417                        // The same two cases as a missing chunk, decided the
418                        // same way: by whether the pointer moved.
419                        Err(StoreError::NotFound(_)) => {
420                            let moved = match load_pointer(store, cfg).await? {
421                                Some(now) => now.etag != etag,
422                                None => true,
423                            };
424                            if moved && attempt + 1 < LOAD_ATTEMPTS {
425                                continue;
426                            }
427                            return Err(super::LeanError::State(format!(
428                                "manifest pointer at seq {} names tombstones {addr}, which do not exist",
429                                p.seq
430                            )));
431                        }
432                        Err(e) => return Err(e.into()),
433                    },
434                };
435                (
436                    LeanManifest {
437                        seq: p.seq,
438                        entries,
439                        sole_writer: p.sole_writer,
440                        boundary_source: p.boundary_source.clone(),
441                        tombstones,
442                    },
443                    None,
444                )
445            }
446        };
447        // The pointer is the authority for the fields it carries: a
448        // rotation moves `seq` without rewriting the entries object, so
449        // the generation's own copy is stale by construction.
450        manifest.seq = p.seq;
451        manifest.sole_writer = p.sole_writer;
452        manifest.boundary_source = p.boundary_source.clone();
453        return Ok(Some(LoadedManifest {
454            manifest,
455            etag,
456            // The CITATION's clock is when the pointer moved, not when
457            // the entries object happened to be written: a rotation
458            // republishes the same entries under a new generation.
459            last_modified_unix: pointer_lm.or(meta_lm),
460            pointer: Some(p),
461        }));
462    }
463    // No pointer: the pre-migration layout, which has no chunks and so
464    // nothing to race a sweep over. Answer it and leave the loop.
465    return match store.get_whole(&cfg.manifest_key(), None).await {
466        Ok((meta, bytes)) => {
467            let manifest = LeanManifest::parse(&bytes)
468                .map_err(|e| super::LeanError::State(format!("manifest parse: {e}")))?;
469            Ok(Some(LoadedManifest {
470                manifest,
471                etag: meta.etag,
472                last_modified_unix: meta.last_modified_unix,
473                pointer: None,
474            }))
475        }
476        Err(StoreError::NotFound(_)) => Ok(None),
477        Err(e) => Err(e.into()),
478    };
479  }
480  // Every attempt raced a sweep. Refusing is right: the alternative is
481  // to serve a manifest we know is short, and short reads as deleted.
482  Err(super::LeanError::State(format!(
483      "manifest load restarted {LOAD_ATTEMPTS} times without settling — a sweep is \
484       collecting generations faster than this reader can finish one"
485  )))
486}
487
488/// How many lost races a one-CAS edit re-reads and retries before it gives
489/// up `Contended`.
490pub const EDIT_CAS_ATTEMPTS: usize = 16;
491
492/// Why a one-CAS edit did not land.
493#[derive(Debug)]
494pub enum EditError<E> {
495    /// The edit refused the document it was shown.
496    Refused(E),
497    Store(LeanError),
498    /// Lost `EDIT_CAS_ATTEMPTS` races in a row.
499    Contended,
500}
501
502/// ONE manifest CAS for an edit by a party that holds no lease (P2): the
503/// gateway's UI saves, deletes and renames, and the ingress sweep's
504/// adoptions. Load the current document; let `edit` judge it and change a
505/// copy — the next generation, with the installing party's fields reset as
506/// `merge` resets them; keep tombstones to `TOMBSTONE_KEEP_SEQS`; CAS it in
507/// with epoch 0; on a lost race do it all again against what won, so every
508/// precondition is judged against the document the CAS replaces.
509pub async fn commit_edit<E>(
510    store: &dyn ObjectStore,
511    cfg: &LeanConfig,
512    flush: &str,
513    edit: impl Fn(Option<&LoadedManifest>, &mut LeanManifest) -> Result<(), E>,
514) -> Result<(), EditError<E>> {
515    // The document the last attempt sent, when its CAS answered 412 or
516    // 409. An SDK retries a conditional PUT whose response was lost, and
517    // the retry meets the object its first attempt wrote: if the document
518    // now IS the one sent, the edit committed, and it is answered so rather
519    // than judged again against itself (a delete would find its path gone
520    // and answer NoSuchFile; the model's GDelete acknowledges what it
521    // commits, `udel`). A peer's identical edit at the same seq (the same
522    // delete, raced) leaves the same document: committed either way.
523    let mut sent: Option<LeanManifest> = None;
524    for _ in 0..EDIT_CAS_ATTEMPTS {
525        let current = load(store, cfg).await.map_err(EditError::Store)?;
526        if let (Some(s), Some(c)) = (sent.as_ref(), current.as_ref()) {
527            if same_document(s, &c.manifest) {
528                return Ok(());
529            }
530        }
531        let mut doc = current.as_ref().map(|l| l.manifest.clone()).unwrap_or_default();
532        doc.seq += 1;
533        doc.sole_writer = false;
534        doc.boundary_source = None;
535        edit(current.as_ref(), &mut doc).map_err(EditError::Refused)?;
536        let seq = doc.seq;
537        doc.tombstones.retain(|_, t| seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
538        let expected = current.as_ref().map(|l| l.handle());
539        let before = current.as_ref().map(|l| &l.manifest);
540        match cas_write_retiring(store, cfg, &doc, before, expected.as_ref(), 0, flush, None).await {
541            Ok(_) => return Ok(()),
542            Err(LeanError::Store(StoreError::PreconditionFailed(_))) | Err(LeanError::Store(StoreError::Conflict(_))) => {
543                sent = Some(doc);
544                continue;
545            }
546            Err(e) => return Err(EditError::Store(e)),
547        }
548    }
549    Err(EditError::Contended)
550}
551
552/// Is `b` the document `a` is — the same generation with the same
553/// citations, tombstones and stamps? (`commit_edit`'s own-write check.)
554fn same_document(a: &LeanManifest, b: &LeanManifest) -> bool {
555    a.seq == b.seq
556        && a.entries == b.entries
557        && a.tombstones == b.tombstones
558        && a.sole_writer == b.sole_writer
559        && a.boundary_source == b.boundary_source
560}
561
562/// CAS-write the manifest document. `expected`: None ⇒ first write
563/// (If-None-Match:*); Some(etag) ⇒ If-Match. The caller owns 412
564/// handling (merge-retry or fence — never blind re-seed).
565pub async fn cas_write(
566    store: &dyn ObjectStore,
567    cfg: &LeanConfig,
568    m: &LeanManifest,
569    expected: Option<&ManifestHandle>,
570    epoch: u64,
571    flush_uuid: &str,
572) -> LeanResult<ObjectMeta> {
573    cas_write_stamped(store, cfg, m, expected, epoch, flush_uuid, None).await
574}
575
576/// `cas_write` with an explicit boundary-source stamp on the manifest
577/// OBJECT (§2.4.1): downstream consumers distinguish a declared-coherent
578/// citation from a forced possibly-torn one from the bucket alone, for
579/// the price of a HEAD rather than a GET of a manifest that can run to
580/// hundreds of MiB.
581pub async fn cas_write_stamped(
582    store: &dyn ObjectStore,
583    cfg: &LeanConfig,
584    m: &LeanManifest,
585    expected: Option<&ManifestHandle>,
586    epoch: u64,
587    flush_uuid: &str,
588    boundary_source: Option<&str>,
589) -> LeanResult<ObjectMeta> {
590    Ok(cas_write_refs(store, cfg, m, expected, epoch, flush_uuid, boundary_source).await?.0)
591}
592
593/// `cas_write_stamped`, also returning the chunk objects the new pointer
594/// references (its chunks and its tombstones object) — `None` on the
595/// single-object layout.
596async fn cas_write_refs(
597    store: &dyn ObjectStore,
598    cfg: &LeanConfig,
599    m: &LeanManifest,
600    expected: Option<&ManifestHandle>,
601    epoch: u64,
602    flush_uuid: &str,
603    boundary_source: Option<&str>,
604) -> LeanResult<(ObjectMeta, Option<Vec<String>>)> {
605    // The manifest's own document must agree with its object stamp:
606    // a reader that GETs and a reader that HEADs must never disagree
607    // about how coherent the citation claims to be.
608    // The layout switch. Every publisher routes through here — the
609    // barrier, the takeover rotation and the gateway's HITL CAS — so the
610    // decision lives at one site rather than three, and §6's "chunk
611    // server-side on receipt" falls out rather than needing its own
612    // path.
613    if cfg.chunked {
614        let (meta, refs) = cas_write_chunked_refs(
615            store,
616            cfg,
617            m,
618            expected,
619            expected.map(|h| h.prev_chunks.as_slice()).unwrap_or(&[]),
620            PublishStamps { epoch, flush_uuid, boundary_source },
621        )
622        .await?;
623        return Ok((meta, Some(refs)));
624    }
625    let mut m = m.clone();
626    if boundary_source.is_some() {
627        m.boundary_source = boundary_source.map(|s| s.to_string());
628    }
629    let m = &m;
630    let bytes = m.to_bytes();
631    let crc = crc64_nvme(&bytes);
632    let stamps = GenerationStamps {
633        generation: m.seq,
634        epoch,
635        flush_uuid: flush_uuid.to_string(),
636        // Stamp what the DOCUMENT carries, not merely what this call
637        // passed. An explicit source has already been written into the
638        // document above, so the two agree there either way; the case
639        // this closes is a `None` caller over a document that already
640        // names its clock — the gateway's HITL CAS relays a document it
641        // did not author. Stamping `None` on those left the object
642        // saying nothing while the document said `sentinel`, which is
643        // exactly the GET/HEAD divergence the contract above forbids.
644        boundary_source: m.boundary_source.clone(),
645        posix: None,
646    };
647
648    // 1. The ENTRIES, to a fresh immutable key. `IfNoneMatchAny` is the
649    //    whole safety story for this object: two writers that raced to
650    //    the same seq cannot both land, and the loser is told so here
651    //    rather than discovering it after it has already published.
652    let entries_key = cfg.generation_key(m.seq, flush_uuid);
653    match store
654        .put_whole(&entries_key, bytes.into(), &PutCondition::IfNoneMatchAny, &stamps, crc)
655        .await
656    {
657        Ok(_) => {}
658        // The key carries this writer's flush uuid, so this is not a
659        // race with another writer — it is a retry of OUR OWN write
660        // whose pointer CAS we never learned the outcome of. The object
661        // is byte-identical by construction; fall through to the CAS,
662        // which is where the race is actually decided.
663        Err(StoreError::PreconditionFailed(_)) | Err(StoreError::Conflict(_)) => {}
664        Err(e) => return Err(e.into()),
665    }
666
667    // 2. The POINTER, which is what makes the publish visible and what
668    //    decides the race. Everything above this line is invisible to
669    //    every reader; a crash here leaves an orphan generation object
670    //    that nobody reads and the sweep collects.
671    let pointer = Pointer {
672        seq: m.seq,
673        entries_key: Some(entries_key.clone()),
674        entries_seq: Some(m.seq),
675        // The step-one layout. `cas_write_chunked` is what produces a
676        // chunk list; this path stays byte-for-byte what it was.
677        chunks: None,
678        sole_writer: m.sole_writer,
679        boundary_source: m.boundary_source.clone(),
680        epoch,
681        tombstones: None,
682    };
683    Ok((put_pointer(store, cfg, &pointer, expected, &stamps).await?, None))
684}
685
686/// Who is publishing, and under what coherence claim. Grouped because
687/// they travel together through every publish path and are meaningless
688/// apart — and because a positional list of three scalars is how an
689/// epoch ends up where a flush uuid was meant to go.
690pub struct PublishStamps<'a> {
691    pub epoch: u64,
692    pub flush_uuid: &'a str,
693    pub boundary_source: Option<&'a str>,
694}
695
696/// Publish a manifest as a CHUNK LIST (chunked design §2).
697///
698/// The ordering is the atomicity story and is not negotiable: every
699/// chunk is durable BEFORE the pointer CAS, and the pointer is the only
700/// thing that ever becomes visible. A crash between the two leaves
701/// orphan chunks nobody references and no reader can observe — never a
702/// half-written manifest.
703///
704/// `prev` is the chunk list this publish is replacing. Chunks whose
705/// address is already in it are skipped ENTIRELY — not uploaded and
706/// re-conditioned, but never sent. That is where O(changed) actually
707/// comes from: without it, a one-file publish still ships every chunk's
708/// bytes over the wire and only the store's 412 saves the storage.
709pub async fn cas_write_chunked(
710    store: &dyn ObjectStore,
711    cfg: &LeanConfig,
712    m: &LeanManifest,
713    expected: Option<&ManifestHandle>,
714    prev: &[super::chunk::ChunkRef],
715    pub_stamps: PublishStamps<'_>,
716) -> LeanResult<ObjectMeta> {
717    Ok(cas_write_chunked_refs(store, cfg, m, expected, prev, pub_stamps).await?.0)
718}
719
720/// `cas_write_chunked`, also returning the chunk addresses the new pointer
721/// references, its tombstones object included.
722async fn cas_write_chunked_refs(
723    store: &dyn ObjectStore,
724    cfg: &LeanConfig,
725    m: &LeanManifest,
726    expected: Option<&ManifestHandle>,
727    prev: &[super::chunk::ChunkRef],
728    pub_stamps: PublishStamps<'_>,
729) -> LeanResult<(ObjectMeta, Vec<String>)> {
730    let PublishStamps { epoch, flush_uuid, boundary_source } = pub_stamps;
731    let mut m = m.clone();
732    if boundary_source.is_some() {
733        m.boundary_source = boundary_source.map(|s| s.to_string());
734    }
735    let m = &m;
736    let split =
737        super::chunk::split_with(&m.entries, cfg.chunk_target, cfg.chunk_min, cfg.chunk_max)?;
738    let have: std::collections::HashSet<&str> = prev.iter().map(|r| r.addr.as_str()).collect();
739
740    let stamps = GenerationStamps {
741        generation: m.seq,
742        epoch,
743        flush_uuid: flush_uuid.to_string(),
744        boundary_source: m.boundary_source.clone(),
745        posix: None,
746    };
747
748    for (r, body) in &split {
749        if have.contains(r.addr.as_str()) {
750            continue;
751        }
752        // UNCONDITIONAL, and that is a correctness requirement rather
753        // than laziness — `LeanChunkGCAdoptSkips` in the formal model
754        // fails without it.
755        //
756        // This chunk is not in `prev`, so nothing references it yet. It
757        // may nonetheless already EXIST: a crashed publish of ours left
758        // it, and content addressing means our retry produces the same
759        // key. Adopting it by reference alone would make an old object
760        // live without making it LOOK live, and the orphan sweep reads
761        // age — so the sweep would collect it as the orphan it still
762        // appears to be, between this call and our pointer CAS.
763        // Rewriting refreshes the age the sweep reads. The bytes are
764        // identical either way, so the write is idempotent; the cost
765        // falls only on adopted orphans, which are rare by
766        // construction, and chunks already in `prev` are skipped above.
767        let key = cfg.chunk_key(&r.addr);
768        let crc = crc64_nvme(body);
769        store.put_whole(&key, body.clone().into(), &PutCondition::Unconditional, &stamps, crc).await?;
770    }
771
772    // The tombstones, one object beside the chunks (rewritten like an
773    // adopted chunk: the sweep reads age, and the bytes are identical).
774    let tombstones = if m.tombstones.is_empty() {
775        None
776    } else {
777        let (addr, body) = super::chunk::encode_tombstones(&m.tombstones)?;
778        let crc = crc64_nvme(&body);
779        store.put_whole(&cfg.chunk_key(&addr), body.into(), &PutCondition::Unconditional, &stamps, crc).await?;
780        Some(addr)
781    };
782    let pointer = Pointer {
783        seq: m.seq,
784        entries_key: None,
785        entries_seq: None,
786        chunks: Some(split.into_iter().map(|(r, _)| r).collect()),
787        sole_writer: m.sole_writer,
788        boundary_source: m.boundary_source.clone(),
789        epoch,
790        tombstones: tombstones.clone(),
791    };
792    let meta = put_pointer(store, cfg, &pointer, expected, &stamps).await?;
793    let mut refs: Vec<String> = pointer.chunks.iter().flatten().map(|r| r.addr.clone()).collect();
794    refs.extend(tombstones);
795    Ok((meta, refs))
796}
797
798/// What a commit retired (M1): the handles its document no longer cites
799/// and the chunk objects its pointer no longer references. One immutable
800/// log object per commit, under `retired/`, keyed by the unix second it
801/// was written — the retirement's clock. The sweeps spare everything a
802/// log younger than `retire_grace_secs` names; a writer reaps the older
803/// ones (`untracked.rs`).
804#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
805pub struct RetireLog {
806    pub at_unix: u64,
807    #[serde(default, skip_serializing_if = "Vec::is_empty")]
808    pub handles: Vec<String>,
809    /// Chunk ADDRESSES (`LeanConfig::chunk_key` names the object).
810    #[serde(default, skip_serializing_if = "Vec::is_empty")]
811    pub chunks: Vec<String>,
812}
813
814pub fn retired_prefix(cfg: &LeanConfig) -> String {
815    format!("{}/{}/retired/", cfg.prefix, super::LEAN_DIR)
816}
817
818/// The unix second a retire log's KEY carries (`<at:020>-...`): a listing
819/// dates every log with no GET.
820pub fn retire_log_at(cfg: &LeanConfig, key: &str) -> Option<u64> {
821    key.strip_prefix(&retired_prefix(cfg))?.get(..20)?.parse().ok()
822}
823
824/// A commit, and the log of what it retired, written right after the CAS.
825/// A crash between the two leaves those handles and chunks to the sweeps'
826/// write-age rule — the shape before the retire age — and nothing worse.
827/// The log is best effort for the same reason: a failure to write it costs
828/// a lagging reader a re-resolve, never the commit.
829#[allow(clippy::too_many_arguments)]
830pub async fn cas_write_retiring(
831    store: &dyn ObjectStore,
832    cfg: &LeanConfig,
833    m: &LeanManifest,
834    before: Option<&LeanManifest>,
835    expected: Option<&ManifestHandle>,
836    epoch: u64,
837    flush_uuid: &str,
838    boundary_source: Option<&str>,
839) -> LeanResult<ObjectMeta> {
840    let (meta, refs) = cas_write_refs(store, cfg, m, expected, epoch, flush_uuid, boundary_source).await?;
841    let still: std::collections::HashSet<&str> = m.entries.values().map(|e| e.key.as_str()).collect();
842    let mut handles: Vec<String> = before
843        .map(|b| b.entries.values().map(|e| e.key.clone()).filter(|k| !still.contains(k.as_str())).collect())
844        .unwrap_or_default();
845    handles.sort();
846    handles.dedup();
847    let mut chunks: Vec<String> = vec![];
848    if let (Some(refs), Some(h)) = (refs.as_ref(), expected) {
849        let now: std::collections::HashSet<&str> = refs.iter().map(|a| a.as_str()).collect();
850        chunks = h
851            .prev_chunks
852            .iter()
853            .map(|r| r.addr.clone())
854            .chain(h.prev_tombstones.clone())
855            .filter(|a| !now.contains(a.as_str()))
856            .collect();
857        chunks.sort();
858        chunks.dedup();
859    }
860    if !handles.is_empty() || !chunks.is_empty() {
861        let at = super::now_unix();
862        let log = RetireLog { at_unix: at, handles, chunks };
863        let key = format!("{}{at:020}-{flush_uuid}-{}.json", retired_prefix(cfg), uuid::Uuid::new_v4());
864        let body = serde_json::to_vec(&log).map_err(|e| LeanError::State(format!("retire log: {e}")))?;
865        let crc = crc64_nvme(&body);
866        let stamps = GenerationStamps {
867            generation: m.seq,
868            epoch,
869            flush_uuid: flush_uuid.to_string(),
870            boundary_source: None,
871            posix: None,
872        };
873        if let Err(e) = store.put_whole(&key, body.into(), &PutCondition::IfNoneMatchAny, &stamps, crc).await {
874            eprintln!("flint-sync: could not log what seq {} retired ({e}); the sweeps judge it by write age", m.seq);
875        }
876    }
877    Ok(meta)
878}
879
880/// CAS the pointer. Split out because a takeover rotation writes ONLY
881/// this — no entries object, no multi-MB anything.
882async fn put_pointer(
883    store: &dyn ObjectStore,
884    cfg: &LeanConfig,
885    pointer: &Pointer,
886    expected: Option<&ManifestHandle>,
887    stamps: &GenerationStamps,
888) -> LeanResult<ObjectMeta> {
889    let body = serde_json::to_vec(pointer)
890        .map_err(|e| super::LeanError::State(format!("pointer encode: {e}")))?;
891    let crc = crc64_nvme(&body);
892    // A LEGACY handle is a migration: the writer read the old single
893    // object and the pointer does not exist yet, so the precondition is
894    // "no pointer", not "the pointer I read". A concurrent migration
895    // therefore loses here and retries against the layout that won,
896    // instead of both of them believing they installed it.
897    let cond = match expected {
898        Some(h) if !h.legacy => PutCondition::IfMatch(h.etag.clone()),
899        _ => PutCondition::IfNoneMatchAny,
900    };
901    let meta = store.put_whole(&cfg.current_key(), body.into(), &cond, stamps, crc).await?;
902    if let Some(h) = expected.filter(|h| h.legacy) {
903        // The workspace has moved. Poison the legacy key rather than
904        // delete it — see REFUSAL_DOC. Conditional on the very object
905        // this writer read, so a racing writer's view is never
906        // clobbered, and best effort: the refusal doc is decorative to
907        // every new reader, and the ONLY reader it speaks to is an old
908        // binary that must refuse rather than re-seed.
909        let bytes = REFUSAL_DOC.as_bytes().to_vec();
910        let crc = crc64_nvme(&bytes);
911        if let Err(e) = store
912            .put_whole(
913                &cfg.manifest_key(),
914                bytes.into(),
915                &PutCondition::IfMatch(h.etag.clone()),
916                stamps,
917                crc,
918            )
919            .await
920        {
921            eprintln!("flint-sync: could not poison the legacy manifest key after migration: {e}");
922        }
923    }
924    Ok(meta)
925}
926
927/// Bump the generation without touching the entries.
928///
929/// This is what a takeover does, and under the single-object layout it
930/// was a multi-MB GET + PUT per claim that also moved the ETag every
931/// other syncer's no-change early exit compares against — so one claim
932/// cost every follower a full fetch and parse of a document in which
933/// nothing had changed. Here it is one small CAS: `seq` moves, so an
934/// outstanding pointer handle goes stale exactly as before, and
935/// `entries_seq` does NOT, so a follower can see that the entries are
936/// the ones it already has.
937pub async fn rotate_for_takeover(
938    store: &dyn ObjectStore,
939    cfg: &LeanConfig,
940    epoch: u64,
941) -> LeanResult<Option<(LeanManifest, String)>> {
942    for _ in 0..3 {
943        // The legacy layout has no pointer to rotate: migrate by
944        // rewriting the document once, then rotations are cheap forever.
945        let Some(LoadedPointer { pointer: p, etag, .. }) = load_pointer(store, cfg).await? else {
946            let Some(loaded) = load(store, cfg).await? else {
947                return Ok(None);
948            };
949            let mut rotated = loaded.manifest.clone();
950            rotated.seq += 1;
951            let h = loaded.handle();
952            match cas_write(store, cfg, &rotated, Some(&h), epoch, "takeover-rotation").await {
953                Ok(meta) => return Ok(Some((rotated, meta.etag))),
954                Err(super::LeanError::Store(StoreError::PreconditionFailed(_) | StoreError::Conflict(_))) => continue,
955                Err(e) => return Err(e),
956            }
957        };
958        let next = Pointer { seq: p.seq + 1, epoch, ..p.clone() };
959        let stamps = GenerationStamps {
960            generation: next.seq,
961            epoch,
962            flush_uuid: "takeover-rotation".to_string(),
963            boundary_source: next.boundary_source.clone(),
964            posix: None,
965        };
966        let h = ManifestHandle { etag, legacy: false, prev_chunks: Vec::new(), prev_tombstones: None };
967        match put_pointer(store, cfg, &next, Some(&h), &stamps).await {
968            Ok(meta) => {
969                // The document the caller gets back is the standing one
970                // with the rotated seq — the entries were not read and
971                // did not need to be.
972                let mut m = LeanManifest::default();
973                m.seq = next.seq;
974                m.sole_writer = next.sole_writer;
975                m.boundary_source = next.boundary_source.clone();
976                return Ok(Some((m, meta.etag)));
977            }
978            // S3 answers a conditional write racing another on the key with 409,
979            // and a takeover whose acquire has LANDED must not fail on a lost race
980            // (review 2026-09-18, H2).
981            Err(super::LeanError::Store(StoreError::PreconditionFailed(_) | StoreError::Conflict(_))) => continue,
982            Err(e) => return Err(e),
983        }
984    }
985    Err(super::LeanError::State(
986        "takeover rotation lost 3 CAS races — a live writer is still publishing".into(),
987    ))
988}
989
990/// How many superseded generations to keep behind the live one.
991///
992/// Not zero, and not one. A reader GETs the pointer and then GETs the
993/// object it names, and those are two requests with a gap between them:
994/// reaping the generation a reader has just resolved would turn a
995/// perfectly ordinary read into a `NotFound`. A handful is a few hundred
996/// KiB at any realistic entry count and buys that gap several publishes.
997pub const KEEP_GENERATIONS: usize = 5;
998
999/// How long a generation NEWER than the live one is left alone before it
1000/// is treated as an orphan.
1001///
1002/// An object above the live generation is one of two things: a publish
1003/// that is in flight right now — its entries written, its pointer CAS
1004/// not yet landed — or the wreckage of one that died in that window. The
1005/// two are indistinguishable from a listing, and deleting the first
1006/// breaks a live publish, so age is the only honest discriminator. Well
1007/// past any plausible manifest PUT, including a 264 MiB one over a slow
1008/// link.
1009pub const ORPHAN_GRACE_SECS: u64 = 3600;
1010
1011/// Retire-age G's default (`LeanConfig::retire_grace_secs`): how long a
1012/// handle or chunk stays after the commit that stopped referencing it.
1013/// Past a large checkout's manifest load and a UI read's two GETs; a
1014/// reader that lags longer re-resolves the pointer, as it always has. It
1015/// costs storage: every version a commit replaced, kept this long.
1016pub const RETIRE_GRACE_SECS: u64 = 600;
1017
1018/// Delete superseded generation objects, and orphans left by a publish
1019/// that died between its entries PUT and its pointer CAS.
1020///
1021/// Immutable metadata that is never collected is a leak, and this one
1022/// grows by a whole manifest per publish.
1023///
1024/// The two cases are NOT symmetric, which the first cut of this got
1025/// wrong. BELOW the live generation, everything is superseded by
1026/// definition and the only question is how much slack to leave a reader
1027/// mid-resolve — that is `KEEP_GENERATIONS`. ABOVE it, an object may be
1028/// a publish still in flight, and a keep-window is exactly the wrong
1029/// tool: an orphan at a high seq sorts to the front and the window
1030/// protects it forever while costing a real generation its slot. Age
1031/// decides that case instead.
1032///
1033/// Best effort and off the critical path: a failure here costs storage,
1034/// never correctness, so it warns rather than failing a publish.
1035pub async fn sweep_generations(store: &dyn ObjectStore, cfg: &LeanConfig) -> LeanResult<usize> {
1036    let Some(lp) = load_pointer(store, cfg).await? else {
1037        return Ok(0);
1038    };
1039    // A chunked pointer names no generation object at all, so there is
1040    // nothing here to keep a window of. Chunks are shared across
1041    // generations and are collected by a different rule entirely
1042    // (chunked design §8.1) — sweeping them by "not named by the live
1043    // pointer" would delete history the window exists to preserve.
1044    let Some(live) = lp.pointer.entries_key else {
1045        // The workspace has MOVED to the chunk layout, so no pointer
1046        // names a generation object any more and every one of them is
1047        // superseded. Collect them past the grace rather than leaking
1048        // the whole pre-migration history forever — a reader still
1049        // resolving one revalidates and restarts (§8.2), and the grace
1050        // covers the window in which it might not have noticed yet.
1051        let prefix = format!("{}/{}/manifests/", cfg.prefix, super::LEAN_DIR);
1052        let now = super::now_unix();
1053        let mut removed = 0;
1054        for o in store.list(&prefix).await? {
1055            let age = o.last_modified_unix.map(|t| now.saturating_sub(t));
1056            if !age.is_some_and(|a| a >= cfg.orphan_grace_secs) {
1057                continue;
1058            }
1059            match store.delete(&o.key).await {
1060                Ok(()) => removed += 1,
1061                Err(e) => eprintln!("flint-sync: could not reap migrated generation {}: {e}", o.key),
1062            }
1063        }
1064        return Ok(removed);
1065    };
1066    let prefix = format!("{}/{}/manifests/", cfg.prefix, super::LEAN_DIR);
1067    let mut listed = store.list(&prefix).await?;
1068    // The key is `<seq:020>-<uuid>`, so lexical order IS generation
1069    // order and a plain sort is enough — the zero padding is doing this
1070    // job, which is why it is there.
1071    listed.sort_by(|a, b| a.key.cmp(&b.key));
1072    let now = super::now_unix();
1073    let live_ix = listed.iter().position(|o| o.key == live);
1074
1075    let mut doomed: Vec<String> = Vec::new();
1076    for (i, o) in listed.iter().enumerate() {
1077        if o.key == live {
1078            continue;
1079        }
1080        match live_ix {
1081            // Newer than the live pointer: in flight, or wreckage.
1082            Some(ix) if i > ix => {
1083                let age = o.last_modified_unix.map(|t| now.saturating_sub(t));
1084                // No timestamp ⇒ leave it. A store that cannot date its
1085                // objects gets a leak, not a deleted live publish.
1086                if age.is_some_and(|a| a > ORPHAN_GRACE_SECS) {
1087                    doomed.push(o.key.clone());
1088                }
1089            }
1090            // Older than the live pointer: superseded. Keep a window — by
1091            // count, and (M1) until the generation that superseded it is
1092            // `retire_grace_secs` old: at one UI save a second, the count
1093            // alone is seconds. No date on the successor ⇒ keep it.
1094            Some(ix) => {
1095                let superseded_for = listed[i + 1].last_modified_unix.map(|t| now.saturating_sub(t));
1096                if ix.saturating_sub(i) > KEEP_GENERATIONS
1097                    && superseded_for.is_some_and(|a| a >= cfg.retire_grace_secs)
1098                {
1099                    doomed.push(o.key.clone());
1100                }
1101            }
1102            // The pointer names an object that is not in the listing.
1103            // `load` already refuses that workspace; sweep nothing.
1104            None => return Ok(0),
1105        }
1106    }
1107    let mut removed = 0;
1108    for k in doomed {
1109        match store.delete(&k).await {
1110            Ok(()) => removed += 1,
1111            Err(e) => eprintln!("flint-sync: could not reap superseded generation {k}: {e}"),
1112        }
1113    }
1114    Ok(removed)
1115}
1116
1117/// Collect chunk objects that no live pointer references.
1118///
1119/// Built to `lean/formal/LeanChunkGC.tla`, which refuted the ordering
1120/// rule the design originally prescribed. Four things carry the safety
1121/// here and each has a mutation config that breaks it alone:
1122///
1123/// 1. **The reference set must not predate a CAS the delete follows.**
1124///    S3 cannot make the read and the delete atomic, so this fences
1125///    instead: the pointer's ETag is read before the listing and again
1126///    after, and the pass ABORTS if it moved. A publish landed, and we
1127///    no longer know whether it references a candidate.
1128/// 2. **A grace.** A chunk written and not yet referenced would
1129///    otherwise be collected out from under its own publish.
1130/// 3. **The grace outlives the longest publish** — a timing
1131///    assumption, which is why `orphan_grace_secs` is config and why
1132///    this comment names it rather than leaving it implied.
1133/// 4. **Adoption rewrites what it adopts** (`cas_write_chunked`), so
1134///    any chunk a concurrent publish could reference is FRESH and the
1135///    grace covers it. Without that rule an adopted orphan is old, and
1136///    the age this reads is a lie about whether it is in use.
1137///
1138/// Deliberately no history window, unlike `sweep_generations`. That
1139/// window existed partly so a reader mid-resolve could finish; readers
1140/// now revalidate against the pointer and restart onto the current
1141/// generation (§8.2), which removes the reason. Adding one back means
1142/// retaining pointer SNAPSHOTS so their chunk sets are enumerable
1143/// (§9) — chunks are shared, so "what the live pointer does not name"
1144/// is the only reference set available without them.
1145pub async fn sweep_chunks(store: &dyn ObjectStore, cfg: &LeanConfig) -> LeanResult<usize> {
1146    sweep_chunks_sparing(store, cfg, &std::collections::HashSet::new()).await
1147}
1148
1149/// `sweep_chunks`, sparing the chunk KEYS a young retire log names (M1):
1150/// a chunk a commit superseded less than `retire_grace_secs` ago, however
1151/// long ago it was written. A reader mid-load of the previous pointer
1152/// finds it.
1153pub async fn sweep_chunks_sparing(
1154    store: &dyn ObjectStore,
1155    cfg: &LeanConfig,
1156    spared: &std::collections::HashSet<String>,
1157) -> LeanResult<usize> {
1158    let Some(before) = load_pointer(store, cfg).await? else {
1159        return Ok(0);
1160    };
1161    let mut refs: std::collections::HashSet<String> = match before.pointer.entries()? {
1162        Entries::Chunked(c) => c.iter().map(|r| r.addr.clone()).collect(),
1163        // The single-object layout has no chunks; `sweep_generations`
1164        // owns it, and running both over one workspace would have each
1165        // reasoning about objects the other manages.
1166        Entries::Single { .. } => return Ok(0),
1167    };
1168    // The tombstone object lives beside the chunks and is referenced
1169    // the same way.
1170    if let Some(t) = &before.pointer.tombstones {
1171        refs.insert(t.clone());
1172    }
1173
1174    let prefix = format!("{}/{}/chunks/", cfg.prefix, super::LEAN_DIR);
1175    let listed = store.list(&prefix).await?;
1176
1177    // THE FENCE (rule 1). Nothing published across the reference read
1178    // and the listing, so the candidates and the references describe
1179    // the same moment. A publish that lands after this still cannot
1180    // lose a chunk: everything it references is either already in
1181    // `refs` or was just written or rewritten, and rule 4 guarantees
1182    // the latter are fresh.
1183    let after = load_pointer(store, cfg).await?;
1184    if after.map(|p| p.etag) != Some(before.etag) {
1185        return Ok(0);
1186    }
1187
1188    let now = super::now_unix();
1189    let mut removed = 0;
1190    for o in listed {
1191        let Some(addr) = o.key.rsplit('/').next() else { continue };
1192        if refs.contains(addr) || spared.contains(&o.key) {
1193            continue;
1194        }
1195        // A candidate the LISTING already shows inside the grace cannot be
1196        // past it on a HEAD taken now: a rewrite only makes an object
1197        // younger, so the listing's age bounds the HEAD's from above.
1198        // Skipping it keeps a chunk longer — a leak, never a deleted live
1199        // chunk — and every deletion still rests on the HEAD below. Without
1200        // this every commit HEADed every chunk superseded within the grace,
1201        // one at a time inside the fence: 700 ms of a 1 s hold after five
1202        // minutes of six writers (contention drill), growing for the hour.
1203        if let Some(listed) = o.last_modified_unix {
1204            if now.saturating_sub(listed) < cfg.orphan_grace_secs {
1205                continue;
1206            }
1207        }
1208        // The age must come from a HEAD taken NOW, never from the
1209        // listing above. That listing predates the fence, and rule 4
1210        // (adoption REWRITES what it adopts) works by refreshing a
1211        // chunk's age — so judging by the listing's timestamp makes
1212        // exactly the refresh that rule exists to produce invisible,
1213        // and the reaper deletes a chunk an in-flight publish is about
1214        // to name. The model computed this against the store's state at
1215        // DELETE time; reading it from a pre-fence snapshot was an
1216        // implementation that did not match the rule it was built to.
1217        //
1218        // One HEAD per CANDIDATE past the grace by the listing, not per
1219        // chunk: a busy workspace supersedes chunks on every publish.
1220        //
1221        // Gone already ⇒ nothing to do. Any other HEAD failure, or no
1222        // timestamp ⇒ leave it: a store that cannot date its objects
1223        // gets a leak, not a deleted live chunk, the same rule
1224        // `sweep_generations` follows and for the same reason.
1225        let fresh = match store.head(&o.key).await {
1226            Ok(m) => m.last_modified_unix,
1227            Err(StoreError::NotFound(_)) => continue,
1228            Err(e) => {
1229                eprintln!("flint-sync: could not date orphan chunk {} ({e}); leaving it", o.key);
1230                continue;
1231            }
1232        };
1233        // `>=`, not `>`: the config value is a DURATION the object must
1234        // have survived, so `orphan_grace_secs = 0` has to mean "no
1235        // grace". With `>` it would silently mean "collect nothing
1236        // written this second", which is most of them — a knob whose
1237        // zero disables the thing it configures is a trap.
1238        let age = fresh.map(|t| now.saturating_sub(t));
1239        if !age.is_some_and(|a| a >= cfg.orphan_grace_secs) {
1240            continue;
1241        }
1242        match store.delete(&o.key).await {
1243            Ok(()) => removed += 1,
1244            Err(e) => eprintln!("flint-sync: could not reap orphan chunk {}: {e}", o.key),
1245        }
1246    }
1247    Ok(removed)
1248}
1249
1250/// The three-way merge (the model's `inst`). Starts from THEIRS so
1251/// foreign entries survive by construction; applies my upserts; applies
1252/// my deletes only where theirs is unchanged since my merge base.
1253///
1254/// **`mine_upserts` and `mine_deletes` must be DISJOINT**, and that is
1255/// the caller's job, not this function's: which one wins depends on
1256/// which the tree saw LAST, and neither set carries that. The fused
1257/// barrier gets it free (both are computed from one scan). Resolving an
1258/// overlap here —
1259/// in either direction — cites a deleted file half the time and
1260/// amputates a live one the other half. TLC found both halves.
1261/// Returns (as `Merged`) the merged document, the foreign entries a consume must
1262/// integrate next (present in theirs, changed vs base, not mine), each
1263/// declared delete's outcome, and the entries MINE PUBLISHES OVER: theirs' version at a path this barrier
1264/// uploaded, changed since the merge base. The slot's 412 used to catch
1265/// those at the PUT; under handles the merge is the one place they show,
1266/// and the caller decides which it never integrated and must surface
1267/// (design 2026-09-19, R7).
1268pub fn merge(
1269    base: &BTreeMap<String, String>,
1270    theirs: &LeanManifest,
1271    mine_upserts: &BTreeMap<String, LeanEntry>,
1272    mine_deletes: &BTreeSet<String>,
1273    parked: &BTreeSet<String>,
1274) -> Merged {
1275    let mut merged = theirs.clone();
1276    merged.seq = theirs.seq + 1;
1277    // Same discipline, and for a sharper reason: inheriting it would
1278    // let a workspace that once published as a mirror keep refusing
1279    // adopts forever, and NOT inheriting it would let a mirror silently
1280    // stop being one the first time anything else wrote a manifest.
1281    // The installing pass sets it from its own config.
1282    merged.sole_writer = false;
1283    merged.boundary_source = None;
1284
1285    let mut foreign: Vec<(String, LeanEntry)> = vec![];
1286    let mut overridden: Vec<(String, LeanEntry)> = vec![];
1287    // The DELETES mine publishes over: a path mine uploads that the merge
1288    // base cited and theirs no longer does. Named by the version the
1289    // delete removed — theirs' tombstone, else the base's citation.
1290    let recreated: Vec<(String, String)> = mine_upserts
1291        .keys()
1292        .filter(|p| !theirs.entries.contains_key(*p))
1293        .filter_map(|p| {
1294            let base_etag = base.get(p)?;
1295            let deleted = theirs.tombstones.get(p).map(|t| t.etag.clone()).unwrap_or_else(|| base_etag.clone());
1296            Some((p.clone(), deleted))
1297        })
1298        .collect();
1299    for (p, e) in &theirs.entries {
1300        let changed = base.get(p).map(|b| b != &e.etag).unwrap_or(true);
1301        if changed && !mine_upserts.contains_key(p) && !parked.contains(p) {
1302            foreign.push((p.clone(), e.clone()));
1303        }
1304        if changed && mine_upserts.contains_key(p) {
1305            overridden.push((p.clone(), e.clone()));
1306        }
1307    }
1308
1309    for (p, e) in mine_upserts {
1310        merged.entries.insert(p.clone(), e.clone());
1311        // A path cited again has no delete to remember.
1312        merged.tombstones.remove(p);
1313    }
1314    let mut deletes = BTreeMap::new();
1315    let mut over_theirs: Vec<(String, LeanEntry)> = vec![];
1316    for p in mine_deletes {
1317        let outcome = delete_outcome(base, theirs, p, parked);
1318        deletes.insert(p.clone(), outcome);
1319        if outcome == DeleteOutcome::OverTheirs {
1320            if let Some(e) = theirs.entries.get(p) {
1321                over_theirs.push((p.clone(), e.clone()));
1322            }
1323        }
1324        if outcome != DeleteOutcome::Parked {
1325            if let Some(retired) = merged.entries.remove(p) {
1326                // H1e: the document says what this delete retired.
1327                merged.tombstones.insert(p.clone(), Tombstone { etag: retired.etag, seq: merged.seq });
1328            }
1329        }
1330    }
1331    merged.tombstones.retain(|_, t| merged.seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
1332    Merged { doc: merged, foreign, overridden, deletes, recreated, over_theirs }
1333}
1334
1335/// What `merge` returns: the document, and what it decided per path.
1336pub struct Merged {
1337    pub doc: LeanManifest,
1338    /// Entries a consume must integrate next: present in theirs, changed
1339    /// vs base, not mine.
1340    pub foreign: Vec<(String, LeanEntry)>,
1341    /// Theirs' version at a path this barrier uploaded, changed since the
1342    /// merge base (R7).
1343    pub overridden: Vec<(String, LeanEntry)>,
1344    /// Every declared delete, and what the merge did with it.
1345    pub deletes: BTreeMap<String, DeleteOutcome>,
1346    /// The deletes mine publishes over: `(path, the deleted version's
1347    /// etag)` for a path mine uploads that the merge base cited and theirs
1348    /// deleted. Mine wins; the caller records each (the user's rule: a
1349    /// delete over a dirty path lands, with a record).
1350    pub recreated: Vec<(String, String)>,
1351    /// The versions mine DELETES over: theirs at a path this barrier
1352    /// deletes, changed since the merge base (M3, P1-lite). Mine wins; the
1353    /// caller preserves and records each before the collector takes it.
1354    pub over_theirs: Vec<(String, LeanEntry)>,
1355}
1356
1357/// What the merge does with one of my deletes.
1358#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1359pub enum DeleteOutcome {
1360    /// Theirs is unchanged since my merge base, or already gone: the
1361    /// citation goes.
1362    Applied,
1363    /// Theirs moved off my merge base to a version this tree never had
1364    /// (M3, P1-lite): the delete still applies — mine wins — and theirs is
1365    /// preserved and recorded (`Merged::over_theirs`). Before P1-lite theirs
1366    /// stayed cited (outranked), and under a derived owed set that never
1367    /// settles: the tree lacks the path, the document cites theirs, the
1368    /// path is dirty so nothing is owed, and every barrier repeats.
1369    OverTheirs,
1370    /// Withheld this barrier.
1371    Parked,
1372}
1373
1374/// The ONE place a delete's outcome is decided. `merge` applies it, and
1375/// a caller that must know whether an install moves a path off its
1376/// citation asks this with the same base — never re-derives it. Re-deriving
1377/// it is how the repair rule came to read a different base from the merge
1378/// (M5, L-124).
1379pub fn delete_outcome(
1380    base: &BTreeMap<String, String>,
1381    theirs: &LeanManifest,
1382    p: &str,
1383    parked: &BTreeSet<String>,
1384) -> DeleteOutcome {
1385    if parked.contains(p) {
1386        return DeleteOutcome::Parked;
1387    }
1388    match (theirs.entries.get(p), base.get(p)) {
1389        (Some(e), Some(b)) if &e.etag == b => DeleteOutcome::Applied,
1390        // Already deleted by someone else: nothing is left to cite.
1391        (None, _) => DeleteOutcome::Applied,
1392        // Changed since base, or present in theirs and not in base.
1393        (Some(_), _) => DeleteOutcome::OverTheirs,
1394    }
1395}
1396
1397