Skip to main content

flint_lean/
state.rs

1//! Durable emptyDir bookkeeping (plan §2.1 "restart matrix").
2//!
3//! Everything the barrier needs to survive a CONTAINER restart lives
4//! here as small JSON files under `<root>/.flint-sync/`, each written
5//! temp+rename. A POD replacement gets a fresh emptyDir and therefore a
6//! fresh identity — that asymmetry is deliberate (the plan's P4: the
7//! incarnation id is emptyDir-scoped, so only the same pod's restarted
8//! container may self-supersede the lease).
9
10use std::collections::{BTreeMap, BTreeSet};
11use std::fs;
12use std::path::{Path, PathBuf};
13
14use serde::{Deserialize, Serialize};
15
16use super::{LeanError, LeanResult};
17
18/// One published path as this syncer last knew it: the handle it
19/// integrated (the citation repair re-cites it, the merge recognises it
20/// as this tree's own) and the ETag that attests its bytes.
21#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
22pub struct BaselineEntry {
23    pub etag: String,
24    /// The HANDLE this tree integrated at the path — its own upload, the
25    /// entry it consumed, the citation it checked out (design 2026-09-19,
26    /// R1-R2). The merge counts a document citing it as citing THIS tree's
27    /// version, not a foreign one. `None` only on a baseline written
28    /// before handles.
29    #[serde(default, skip_serializing_if = "Option::is_none")]
30    pub key: Option<String>,
31    pub generation: u64,
32    pub size: u64,
33    pub mtime_unix: i64,
34    /// Sub-second mtime, when this entry was written by a syncer that
35    /// stamps it; `None` compares as "unknown" (`scan::stat_changed`).
36    #[serde(default)]
37    pub mtime_nanos: Option<u32>,
38    /// CRC-64/NVME (wire form) of the object bytes this syncer
39    /// integrated at this path — computed from the bytes it wrote or
40    /// uploaded, never from a HEAD: content convergence compares it, and
41    /// it is what lets the manifest carry a CRC for an object whose
42    /// backend returns none (Ozone). Every writer sets it; `None` only on
43    /// a baseline from before it existed.
44    pub crc64_b64: Option<String>,
45}
46
47/// The persisted baseline snapshot: what this syncer believes the
48/// bucket holds AND has integrated locally — and, since P1-lite
49/// (2026-09-25), the three-way MERGE BASE too. A separate merge base
50/// existed because a tree could adopt a UI write the document did not
51/// cite yet; under P2 every UI edit commits first, so no tree holds a
52/// version the document never cited, and the split had nothing left to
53/// keep apart (the sandbox pair `LeanCoreP1`/`LeanCoreP2R`).
54#[derive(Debug, Clone, Default, Serialize, Deserialize)]
55pub struct Baseline {
56    /// Manifest seq at our last install/checkout.
57    pub seq: u64,
58    /// Manifest document ETag we expect at the next CAS.
59    pub manifest_etag: Option<String>,
60    pub entries: BTreeMap<String, BaselineEntry>,
61    /// THE CHEAP PATH'S RECORD (scan trigger). The pointer etag of the
62    /// document the last full consume derived against, with nothing left
63    /// but the agent's own work; `None` = derive at the next consume (a
64    /// failed fetch, a sync, a baseline from before this field). Only the
65    /// consume writes it — and the commit, when its CAS replaced exactly
66    /// this document, so the new one is it plus this tree's own changes.
67    #[serde(default)]
68    pub derived_etag: Option<String>,
69    /// The paths that derive left untaken because the agent is working on
70    /// them (dirty against the baseline where the document differs). The
71    /// cheap path holds only while every one is still dirty: an agent that
72    /// backs out makes the document's version owed.
73    #[serde(default)]
74    pub skipped: BTreeSet<String>,
75    /// Paths present at the PREVIOUS scan (the two-consecutive-scans
76    /// deletion rule: absence must survive two scans).
77    pub prev_scan: BTreeSet<String>,
78}
79
80/// A rescope in flight (scoped-read design §4.3). `target` is the scope
81/// the workspace is moving TO; `None` means the whole tree.
82#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
83pub struct ScopeIntent {
84    pub target: Option<Vec<String>>,
85    /// The paths this rescope is removing from the held set, recorded
86    /// BEFORE the first mutation.
87    ///
88    /// Load-bearing, and the mutation check that found it: replay
89    /// cannot RE-DERIVE this set. Once a path is out of
90    /// `baseline.entries` it is indistinguishable from a file the agent
91    /// just created — same scan row, same absence from the baseline —
92    /// and the two have opposite correct answers: unlink the leftover,
93    /// KEEP the agent's new file. Deriving the drop set from the
94    /// baseline made a crash between the uncite and the unlink leave
95    /// six uncited files on disk forever; deriving it from the tree
96    /// would delete the agent's work instead.
97    #[serde(default)]
98    pub drop: Vec<String>,
99    /// What the uncite dropped, per path, recorded BEFORE the uncite (and
100    /// kept across a replay, which can no longer read it off the
101    /// baseline). The unlink takes a file only while its bytes are still
102    /// these: the agent may write a dropped path between the uncite and
103    /// the unlink — a crash leaves that window open until the next
104    /// barrier — and once uncited, its write and the leftover read alike
105    /// (the model's `LeanP1UnlinkBlind`, 2026-10-02).
106    #[serde(default)]
107    pub held: BTreeMap<String, BaselineEntry>,
108}
109
110/// The pod-incarnation identity + lease bookkeeping ({last_token,
111/// quiet_polls} persist so container restarts RESUME the takeover
112/// observation instead of resetting it — plan §2.1).
113#[derive(Debug, Clone, Serialize, Deserialize)]
114pub struct Incarnation {
115    pub holder_id: String,
116    pub epoch: u64,
117    pub last_token: Option<String>,
118    pub quiet_polls: u32,
119}
120
121/// What the conditional-write probes saw this store do
122/// (`conformance.rs`). Cached because the probes cost ~10 requests and
123/// a store does not change its mind between two barriers — but keyed by
124/// `identity`, because a workspace re-pointed at a different endpoint or
125/// bucket is a different store and must be asked again.
126///
127/// `probe_version` is what makes the cache safe to extend: a build that
128/// asks a NEW question must not read an old verdict as the answer to
129/// it, so a bump invalidates every cached verdict rather than inheriting
130/// a silent pass.
131#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
132pub struct StoreConformance {
133    pub identity: String,
134    pub probe_version: u32,
135    /// Whether the probe got an ANSWER at all. False means it could not
136    /// write its own object — a credential, a policy, the network — and
137    /// the two flags below say nothing. Never cached false: an outage
138    /// must not become this workspace's standing verdict.
139    pub answered: bool,
140    /// `If-None-Match:*` and `If-Match` on PUT — the manifest CAS, the
141    /// upload gate, the lease cell. Nothing works without it.
142    pub conditional_put: bool,
143    /// `If-Match` on DELETE — the file collector, and only it.
144    pub conditional_delete: bool,
145    /// What the failing probe said, for the operator message.
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub detail: Option<String>,
148    pub at_unix: u64,
149}
150
151/// The intent journal: the flush id of the barrier in flight, and the H10
152/// carrier. (Its `keys`, `recent_uuids` and `declared_deletes` served the
153/// slot's AdoptOwn and the cell's declared removals; nothing read them
154/// once both were gone — M6 of the 2026-09-24 simplification analysis.)
155#[derive(Debug, Clone, Default, Serialize, Deserialize)]
156pub struct IntentJournal {
157    pub flush_uuid: String,
158    /// What the installs that CARRIED the standing publish declaration
159    /// published, journalled in the write that follows each CAS (review
160    /// 2026-09-18, H10). The ack reads the barrier that honors the
161    /// declaration, and that is not always the one that carried it: a
162    /// restart after step 7, an ack write that failed, or an honor that
163    /// failed before the floor's cadence barrier published the
164    /// declaration, each left the ack to a later barrier — and when a
165    /// peer had deleted a declared path in between, that barrier was a
166    /// pull of the peer's document, answered `ok` while the tree still
167    /// held the declared bytes. Carried forward across barriers; matched
168    /// on the pending record's `id`.
169    #[serde(default, skip_serializing_if = "Option::is_none")]
170    pub carrier: Option<Carrier>,
171}
172
173/// See `IntentJournal::carrier`.
174#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
175pub struct Carrier {
176    /// `PendingSentinel::id` of the declaration these installs carried.
177    pub pending_id: String,
178    /// The dirty paths they published — uploads and deletes.
179    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
180    pub paths: BTreeSet<String>,
181    /// Which of `paths` were DELETES: a document that does not cite one
182    /// of those is the declaration honoured, not dropped.
183    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
184    pub deletes: BTreeSet<String>,
185    /// Carried uploads the document the LAST barrier ended at does not
186    /// cite: a peer deleted them after this writer published them. The
187    /// ack's document is that one, so it says so (H10). Recomputed at
188    /// every barrier's end from the document in hand; no request.
189    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
190    pub uncited: BTreeSet<String>,
191    /// Declared paths they did not carry (parked, deferred, or a delete
192    /// the merge outranked) — the ack's `report.dropped` survives them.
193    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
194    pub dropped: BTreeSet<String>,
195}
196
197/// One surfaced conflict: both versions stay recoverable (local bytes in
198/// the tree, foreign bytes preserved at `preserved_key` or still current
199/// under the data key). Never a silent winner.
200#[derive(Debug, Clone, Serialize, Deserialize)]
201pub struct ConflictRecord {
202    pub path: String,
203    pub foreign_etag: String,
204    /// Where the foreign bytes were preserved (server-side copy), when
205    /// the local version is about to overwrite them.
206    pub preserved_key: Option<String>,
207    pub kind: String, // "consume-dirty" | "upload-412-parked" | "sync-dirty" | "gc-skip"
208    pub at_unix: u64,
209}
210
211pub struct SyncerState {
212    dir: PathBuf,
213    /// The state-directory occupancy lock (flock, held for the process
214    /// lifetime). Self-recognition of the lease via the persisted
215    /// incarnation id is only sound because the PREVIOUS process is
216    /// gone — and this lock is what makes that true. Without it, a
217    /// second flint-sync on the same workspace self-recognizes,
218    /// deposes a LIVE sibling, and both write the tree concurrently
219    /// (observed on the 0b rig: a diagnostic re-run raced a live 1M
220    /// checkout into tmp-rename ENOENT collisions). The hub has the
221    /// identical gate (`state_backend::is_single_occupant`).
222    _lock: std::fs::File,
223}
224
225const MARKER: &str = "checkout-complete";
226/// Written by the SIGTERM drain only after it published: the node
227/// plugin's evidence that a worker that is now gone actually drained
228/// (audit 2026-09-03, finding 3). Absent ⇒ the tree is preserved.
229pub const DRAINED: &str = "drained.json";
230const LOCK: &str = "lock";
231const BASELINE: &str = "baseline.json";
232/// The admitted set of a SCOPED checkout. Deliberately its own
233/// document rather than a field on `Baseline`: the baseline is
234/// rewritten by every barrier, every sync and every consume, and a
235/// scope that rides along is one `Baseline::default()` away from
236/// silently widening a workspace to the whole tree. This file is
237/// written once, by checkout, and thereafter only read. Absent means
238/// UNSCOPED — the only encoding of "no restriction", since an empty
239/// admitted set is refused before it can be written.
240const SCOPE: &str = "scope.json";
241
242/// A rescope IN FLIGHT: the target scope, written before the first
243/// mutation and removed only after the last (scoped-read design §4.3).
244///
245/// Its own document and not a field on `scope.json`, because the two
246/// answer different questions — `scope.json` is what this workspace
247/// holds, this is what it is on its way to holding — and a crash must
248/// leave both answers readable.
249///
250/// `null` is a LEGAL target (the whole tree), so absence of the file
251/// means "no rescope in flight", never "unscoped". That is why the
252/// document wraps the target in a struct instead of being a bare
253/// `Option<Vec<String>>`: a bare `null` on disk and a missing file
254/// would deserialize to the same thing.
255const SCOPE_INTENT: &str = "scope-intent.json";
256const INCARNATION: &str = "incarnation.json";
257const INTENT: &str = "intent.json";
258/// When this writer last swept for untracked uploads (`untracked.rs`).
259const UNTRACKED_SWEEP: &str = "untracked-sweep";
260const ORPHAN_SWEEP: &str = "orphan-sweep";
261/// The retire logs this writer has read (`untracked.rs`): immutable, so each
262/// is fetched once.
263const RETIRE_CACHE: &str = "retire-cache.json";
264/// The cached conditional-write verdict for this store (`conformance.rs`).
265const CONFORMANCE: &str = "conformance.json";
266const CONFLICTS: &str = "conflicts.jsonl";
267/// The rotated generation. `load_conflicts` reads it FIRST, so the
268/// sequence a reader sees is unbroken across a rotation — which matters
269/// because `honor_sync` takes the count before a sync and `skip`s it
270/// after, and a rotation that SHORTENED the list would make it skip past
271/// the very records the sync just produced.
272const CONFLICTS_PREV: &str = "conflicts.1.jsonl";
273/// How many records were dropped when a SECOND rotation overwrote the
274/// first. Truncation that nobody can count is truncation that reads as
275/// "there were no more".
276const CONFLICTS_DROPPED: &str = "conflicts.dropped";
277/// Rotate past this. The file is append-only and `load_conflicts` parses
278/// it WHOLE — twice per sync honor, again per status read and per
279/// scrape (U23) — so an unbounded file is an unbounded parse on a hot
280/// path, and a standing condition (a UI autosaving over a path the agent
281/// holds dirty) appends on every barrier. Two generations bound the
282/// resident set at ~2 MiB, which is thousands of records.
283const CONFLICTS_MAX_BYTES: u64 = 1 << 20;
284
285fn write_atomic(path: &Path, bytes: &[u8]) -> LeanResult<()> {
286    // The state dir lives in the same app-writable emptyDir.
287    super::safefs::check_parent(path)?;
288    let tmp = path.with_extension("tmp");
289    super::safefs::write_via_tmp(path, &tmp, bytes, None).map(|_| ())
290}
291
292impl SyncerState {
293    pub fn open(dir: PathBuf) -> LeanResult<SyncerState> {
294        fs::create_dir_all(&dir)?;
295        let lock = fs::OpenOptions::new()
296            .create(true)
297            .truncate(false)
298            .write(true)
299            .open(dir.join(LOCK))?;
300        #[cfg(unix)]
301        {
302            use std::os::fd::AsRawFd;
303            // SAFETY: flock on an owned, open fd.
304            let rc = unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
305            if rc != 0 {
306                return Err(LeanError::State(format!(
307                    "another flint-sync already holds this workspace ({}): \
308                     refusing to run two syncers over one tree",
309                    dir.display()
310                )));
311            }
312        }
313        Ok(SyncerState { dir, _lock: lock })
314    }
315
316    pub fn marker_present(&self) -> bool {
317        self.dir.join(MARKER).exists()
318    }
319
320    /// Flush the filesystem holding the tree (the state dir lives in
321    /// it) so everything materialised so far is on stable storage.
322    pub fn sync_tree(&self) -> LeanResult<()> {
323        super::safefs::sync_tree(&self.dir)
324    }
325
326    /// The drain's attestation. `seq` is the manifest the drain left
327    /// installed, when it knows it.
328    pub fn write_drained(&self, seq: Option<u64>, acks: usize) -> LeanResult<()> {
329        let doc = serde_json::json!({
330            "unix": super::now_unix(),
331            "seq": seq,
332            "acks": acks,
333        });
334        write_atomic(&self.dir.join(DRAINED), doc.to_string().as_bytes())
335    }
336
337    /// A fresh incarnation owes its own attestation; a stale one from
338    /// an earlier life of this tree must not vouch for it.
339    pub fn clear_drained(&self) -> LeanResult<()> {
340        match fs::remove_file(self.dir.join(DRAINED)) {
341            Ok(()) => Ok(()),
342            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
343            Err(e) => Err(LeanError::State(format!("clear {DRAINED}: {e}"))),
344        }
345    }
346
347    pub fn drained_path(&self) -> PathBuf {
348        self.dir.join(DRAINED)
349    }
350
351    /// Written LAST at checkout: the agent-start gate.
352    pub fn write_marker(&self) -> LeanResult<()> {
353        write_atomic(&self.dir.join(MARKER), b"ok\n")
354    }
355
356    pub fn load_baseline(&self) -> LeanResult<Baseline> {
357        let p = self.dir.join(BASELINE);
358        if !p.exists() {
359            return Ok(Baseline::default());
360        }
361        let bytes = fs::read(&p)?;
362        serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("baseline: {e}")))
363    }
364
365    /// COMPACT: the baseline is the other O(files) document, rewritten
366    /// on the cadence tick and re-parsed several times per barrier. The
367    /// agent-facing O(1) files (gauges, capabilities, acks) stay pretty
368    /// — those exist to be `cat`-ed.
369    pub fn save_baseline(&self, b: &Baseline) -> LeanResult<()> {
370        let bytes =
371            serde_json::to_vec(b).map_err(|e| LeanError::State(format!("baseline: {e}")))?;
372        write_atomic(&self.dir.join(BASELINE), &bytes)
373    }
374
375    /// What this workspace was admitted to hold, or `None` for an
376    /// unscoped (whole-manifest) checkout.
377    ///
378    /// `fs::read` directly, with NO `exists()` pre-check: `exists()`
379    /// answers false for EACCES and EIO alike, and an unreadable scope
380    /// answered as "unscoped" is the same class of bug as an unreadable
381    /// path answered as "deleted" (`barrier.rs`, 14b3637c). Only
382    /// NotFound means absent.
383    pub fn load_scope(&self) -> LeanResult<Option<Vec<String>>> {
384        let p = self.dir.join(SCOPE);
385        let bytes = match fs::read(&p) {
386            Ok(b) => b,
387            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
388            Err(e) => {
389                return Err(LeanError::State(format!(
390                    "cannot read the checkout scope at {}: {e} — refusing to \
391                     guess what this workspace is holding",
392                    p.display()
393                )))
394            }
395        };
396        serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("scope: {e}")))
397    }
398
399    /// `Some` writes the admitted set; `None` REMOVES the document.
400    /// Removing matters: a scoped checkout that crashed before its
401    /// marker leaves a scope behind, and the whole-tree checkout that
402    /// replaces it must not inherit a claim it did not make.
403    pub fn save_scope(&self, scope: Option<&[String]>) -> LeanResult<()> {
404        let p = self.dir.join(SCOPE);
405        match scope {
406            Some(entries) => {
407                let bytes = serde_json::to_vec(entries)
408                    .map_err(|e| LeanError::State(format!("scope: {e}")))?;
409                write_atomic(&p, &bytes)
410            }
411            None => match fs::remove_file(&p) {
412                Ok(()) => Ok(()),
413                Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
414                Err(e) => Err(LeanError::State(format!("clear {SCOPE}: {e}"))),
415            },
416        }
417    }
418
419    /// Read a rescope in flight. `None` = none in flight.
420    pub fn load_scope_intent(&self) -> LeanResult<Option<ScopeIntent>> {
421        let p = self.dir.join(SCOPE_INTENT);
422        let bytes = match fs::read(&p) {
423            Ok(b) => b,
424            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
425            Err(e) => {
426                return Err(LeanError::State(format!(
427                    "cannot read the rescope intent at {}: {e} — refusing to run a barrier \
428                     over a half-applied scope",
429                    p.display()
430                )))
431            }
432        };
433        serde_json::from_slice(&bytes)
434            .map(Some)
435            .map_err(|e| LeanError::State(format!("scope intent: {e}")))
436    }
437
438    pub fn save_scope_intent(&self, intent: &ScopeIntent) -> LeanResult<()> {
439        let bytes = serde_json::to_vec(intent)
440            .map_err(|e| LeanError::State(format!("scope intent: {e}")))?;
441        write_atomic(&self.dir.join(SCOPE_INTENT), &bytes)
442    }
443
444    /// Cleared LAST, after the new scope is durable. Clearing it first
445    /// would turn a crash into a workspace whose scope says one thing
446    /// and whose held set says another, with nothing left to replay.
447    pub fn clear_scope_intent(&self) -> LeanResult<()> {
448        match fs::remove_file(self.dir.join(SCOPE_INTENT)) {
449            Ok(()) => Ok(()),
450            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
451            Err(e) => Err(LeanError::State(format!("clear {SCOPE_INTENT}: {e}"))),
452        }
453    }
454
455    pub fn load_incarnation(&self) -> LeanResult<Option<Incarnation>> {
456        let p = self.dir.join(INCARNATION);
457        if !p.exists() {
458            return Ok(None);
459        }
460        let bytes = fs::read(&p)?;
461        serde_json::from_slice(&bytes)
462            .map(Some)
463            .map_err(|e| LeanError::State(format!("incarnation: {e}")))
464    }
465
466    pub fn save_incarnation(&self, i: &Incarnation) -> LeanResult<()> {
467        let bytes = serde_json::to_vec_pretty(i)
468            .map_err(|e| LeanError::State(format!("incarnation: {e}")))?;
469        write_atomic(&self.dir.join(INCARNATION), &bytes)
470    }
471
472    /// Unix seconds of this writer's last untracked-upload sweep; 0 when it
473    /// has never swept.
474    pub fn load_untracked_sweep_at(&self) -> LeanResult<u64> {
475        match fs::read_to_string(self.dir.join(UNTRACKED_SWEEP)) {
476            Ok(s) => Ok(s.trim().parse().unwrap_or(0)),
477            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
478            Err(e) => Err(e.into()),
479        }
480    }
481
482    pub fn save_untracked_sweep_at(&self, at: u64) -> LeanResult<()> {
483        write_atomic(&self.dir.join(UNTRACKED_SWEEP), at.to_string().as_bytes())
484    }
485
486    /// The orphan sweep's clock (`untracked.rs`, `sweep_orphans_if_due`):
487    /// 0 = never swept, and the first commit starts it.
488    pub fn load_orphan_sweep_at(&self) -> LeanResult<u64> {
489        match fs::read_to_string(self.dir.join(ORPHAN_SWEEP)) {
490            Ok(s) => Ok(s.trim().parse().unwrap_or(0)),
491            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
492            Err(e) => Err(e.into()),
493        }
494    }
495
496    pub fn save_orphan_sweep_at(&self, at: u64) -> LeanResult<()> {
497        write_atomic(&self.dir.join(ORPHAN_SWEEP), at.to_string().as_bytes())
498    }
499
500    /// The retire logs read so far, by key. A cache: unreadable reads as
501    /// empty (the logs are fetched again), never as an error.
502    pub fn load_retire_cache(&self) -> BTreeMap<String, super::manifest::RetireLog> {
503        fs::read(self.dir.join(RETIRE_CACHE)).ok().and_then(|b| serde_json::from_slice(&b).ok()).unwrap_or_default()
504    }
505
506    pub fn save_retire_cache(&self, c: &BTreeMap<String, super::manifest::RetireLog>) -> LeanResult<()> {
507        let bytes = serde_json::to_vec(c).map_err(|e| LeanError::State(format!("retire cache: {e}")))?;
508        write_atomic(&self.dir.join(RETIRE_CACHE), &bytes)
509    }
510
511    /// A verdict that cannot be parsed reads as ABSENT, never as a pass:
512    /// the probe runs again. The expensive outcome of a corrupt cache is
513    /// ten requests; the cheap one would be running the collector on a
514    /// store nobody asked.
515    pub fn load_conformance(&self) -> LeanResult<Option<StoreConformance>> {
516        match fs::read(self.dir.join(CONFORMANCE)) {
517            Ok(b) => Ok(serde_json::from_slice(&b).ok()),
518            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
519            Err(e) => Err(e.into()),
520        }
521    }
522
523    pub fn save_conformance(&self, c: &StoreConformance) -> LeanResult<()> {
524        let bytes = serde_json::to_vec_pretty(c)
525            .map_err(|e| LeanError::State(format!("conformance: {e}")))?;
526        write_atomic(&self.dir.join(CONFORMANCE), &bytes)
527    }
528
529    pub fn load_intent(&self) -> LeanResult<IntentJournal> {
530        let p = self.dir.join(INTENT);
531        if !p.exists() {
532            return Ok(IntentJournal::default());
533        }
534        let bytes = fs::read(&p)?;
535        serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("intent: {e}")))
536    }
537
538    pub fn save_intent(&self, j: &IntentJournal) -> LeanResult<()> {
539        let bytes =
540            serde_json::to_vec_pretty(j).map_err(|e| LeanError::State(format!("intent: {e}")))?;
541        write_atomic(&self.dir.join(INTENT), &bytes)
542    }
543
544    /// The barrier is over: no flush in flight. The carrier stays.
545    pub fn clear_intent_keys(&self) -> LeanResult<()> {
546        let mut j = self.load_intent()?;
547        j.flush_uuid = String::new();
548        self.save_intent(&j)
549    }
550
551    pub fn append_conflict(&self, c: &ConflictRecord) -> LeanResult<()> {
552        use std::io::Write;
553        let line =
554            serde_json::to_string(c).map_err(|e| LeanError::State(format!("conflict: {e}")))?;
555        let live = self.dir.join(CONFLICTS);
556        let mut f = fs::OpenOptions::new().create(true).append(true).open(&live)?;
557        writeln!(f, "{line}")?;
558        // Size off the handle we just wrote through — no extra stat of
559        // the path, and no line count, which would make every append
560        // O(records).
561        let len = f.metadata()?.len();
562        drop(f);
563        if len > CONFLICTS_MAX_BYTES {
564            self.rotate_conflicts()?;
565        }
566        Ok(())
567    }
568
569    /// Rotate live -> `.1`, counting whatever the previous `.1` held so
570    /// the loss is a NUMBER rather than a silence. Best effort: a
571    /// rotation that fails leaves a large file, which is a performance
572    /// problem; losing the record we just appended would be a
573    /// correctness one.
574    fn rotate_conflicts(&self) -> LeanResult<()> {
575        let live = self.dir.join(CONFLICTS);
576        let prev = self.dir.join(CONFLICTS_PREV);
577        // Only read the outgoing generation at ROTATION time — rare by
578        // construction, since it takes CONFLICTS_MAX_BYTES to get here.
579        if let Ok(text) = fs::read_to_string(&prev) {
580            let dropped = text.lines().filter(|l| !l.trim().is_empty()).count() as u64;
581            let total = self.conflicts_dropped()? + dropped;
582            write_atomic(&self.dir.join(CONFLICTS_DROPPED), total.to_string().as_bytes())?;
583        }
584        fs::rename(&live, &prev)?;
585        Ok(())
586    }
587
588    /// Records lost to rotation. A caller rendering conflicts owes the
589    /// user this number: a truncated list that does not say it was
590    /// truncated reads as a complete one.
591    pub fn conflicts_dropped(&self) -> LeanResult<u64> {
592        match fs::read_to_string(self.dir.join(CONFLICTS_DROPPED)) {
593            Ok(s) => Ok(s.trim().parse().unwrap_or(0)),
594            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
595            Err(e) => Err(LeanError::State(format!("{CONFLICTS_DROPPED}: {e}"))),
596        }
597    }
598
599    /// Every record ever appended: the ones rotation dropped plus the
600    /// ones held. Unlike the held count it never goes down, so a reader
601    /// that takes it before an operation knows exactly how many records
602    /// the operation produced, across any rotation (`conflicts_since`).
603    pub fn conflicts_appended(&self) -> LeanResult<u64> {
604        Ok(self.conflicts_dropped()? + self.load_conflicts()?.len() as u64)
605    }
606
607    /// The rotated generation first, then the live one: oldest to
608    /// newest, unbroken across a single rotation.
609    pub fn load_conflicts(&self) -> LeanResult<Vec<ConflictRecord>> {
610        let mut out = vec![];
611        for name in [CONFLICTS_PREV, CONFLICTS] {
612            let p = self.dir.join(name);
613            // `fs::read_to_string` directly: `exists()` answers false
614            // for EACCES and EIO alike, and a conflict log that reads as
615            // EMPTY because it is unreadable is the same shape of bug as
616            // an unreadable path reading as deleted.
617            let text = match fs::read_to_string(&p) {
618                Ok(t) => t,
619                Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
620                Err(e) => {
621                    return Err(LeanError::State(format!(
622                        "cannot read {name}: {e} — refusing to report an unreadable \
623                         conflict log as an empty one"
624                    )))
625                }
626            };
627            for line in text.lines() {
628                if line.trim().is_empty() {
629                    continue;
630                }
631                out.push(
632                    serde_json::from_str(line)
633                        .map_err(|e| LeanError::State(format!("conflict line: {e}")))?,
634                );
635            }
636        }
637        Ok(out)
638    }
639}