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 recognized ETag
19/// is the If-Match guard for the next publish and the HEAD-guard for GC.
20#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
21pub struct BaselineEntry {
22    pub etag: String,
23    pub generation: u64,
24    pub size: u64,
25    pub mtime_unix: i64,
26    /// Sub-second mtime, when this entry was written by a syncer that
27    /// stamps it; `None` compares as "unknown" (`scan::stat_changed`).
28    #[serde(default)]
29    pub mtime_nanos: Option<u32>,
30    /// CRC-64/NVME (wire form) of the object bytes this syncer
31    /// integrated at this path — computed from the bytes it wrote or
32    /// uploaded, never from a HEAD. A citation repair cites it, which
33    /// is what lets the manifest carry a CRC for an object whose
34    /// backend returns none (Ozone) or whose HITL uploader sent none.
35    ///
36    /// `None` only on the consume-dirty sentinel (`size == u64::MAX`):
37    /// the bytes on disk are the LOCAL edit, about to publish under a
38    /// CRC of its own, and the sentinel never repairs.
39    pub crc64_b64: Option<String>,
40}
41
42/// The persisted baseline snapshot: what this syncer believes the
43/// bucket holds AND has integrated locally. Distinct from `inst_base`
44/// (the manifest view at our last install — the three-way merge base):
45/// consuming a HITL entry advances the baseline for that path but not
46/// the merge base. The formal model carries the same split
47/// (baseline vs instBase in LeanSubtree.tla) — collapsing them made a
48/// syncer mistake its own consumed adoption for a foreign entry.
49#[derive(Debug, Clone, Default, Serialize, Deserialize)]
50pub struct Baseline {
51    /// Manifest seq at our last install/checkout.
52    pub seq: u64,
53    /// Manifest document ETag we expect at the next CAS.
54    pub manifest_etag: Option<String>,
55    pub entries: BTreeMap<String, BaselineEntry>,
56    /// The merge base: path -> ETag as cited by the manifest we last
57    /// installed (or checked out).
58    pub inst_base: BTreeMap<String, String>,
59    /// Paths present at the PREVIOUS scan (the two-consecutive-scans
60    /// deletion rule: absence must survive two scans).
61    pub prev_scan: BTreeSet<String>,
62}
63
64/// A rescope in flight (scoped-read design §4.3). `target` is the scope
65/// the workspace is moving TO; `None` means the whole tree.
66#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
67pub struct ScopeIntent {
68    pub target: Option<Vec<String>>,
69    /// The paths this rescope is removing from the held set, recorded
70    /// BEFORE the first mutation.
71    ///
72    /// Load-bearing, and the mutation check that found it: replay
73    /// cannot RE-DERIVE this set. Once a path is out of
74    /// `baseline.entries` it is indistinguishable from a file the agent
75    /// just created — same scan row, same absence from the baseline —
76    /// and the two have opposite correct answers: unlink the leftover,
77    /// KEEP the agent's new file. Deriving the drop set from the
78    /// baseline made a crash between the uncite and the unlink leave
79    /// six uncited files on disk forever; deriving it from the tree
80    /// would delete the agent's work instead.
81    #[serde(default)]
82    pub drop: Vec<String>,
83}
84
85/// The pod-incarnation identity + lease bookkeeping ({last_token,
86/// quiet_polls} persist so container restarts RESUME the takeover
87/// observation instead of resetting it — plan §2.1).
88#[derive(Debug, Clone, Serialize, Deserialize)]
89pub struct Incarnation {
90    pub holder_id: String,
91    pub epoch: u64,
92    pub last_token: Option<String>,
93    pub quiet_polls: u32,
94}
95
96/// The intent journal written BEFORE uploads: which keys this barrier
97/// will touch and under which flush_uuid, so a restarted container can
98/// recognize its own crashed/torn PUT at the 412 (AdoptOwn) instead of
99/// mistaking it for a foreign write. `recent_uuids` keeps the last few
100/// barriers' uuids for the same reason.
101/// A change ANOTHER writer made that this workspace's merge carried into
102/// the manifest but not yet into the tree: the next consume fetches it
103/// (or, for a deletion, removes a clean local copy).
104///
105/// Writer-local on purpose. These rode the SHARED inbox as
106/// `merge-preserved` entries, which was right while one syncer wrote a
107/// workspace and wrong the moment two did: the other writer's consume
108/// found its own bytes there, called the entry integrated and dropped
109/// it, and the writer that needed it never fetched the change; a
110/// `sync` read it as remote truth after the manifest had moved past it.
111/// And a peer's DELETE had no carrier at all. A pod replacement loses
112/// this file with the baseline, which is correct: the replacement's
113/// checkout materializes the whole manifest.
114#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
115pub struct ForeignChange {
116    pub path: String,
117    /// The etag the manifest cites; `None` means the path was deleted.
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub etag: Option<String>,
120    #[serde(default, skip_serializing_if = "Option::is_none")]
121    pub crc64_b64: Option<String>,
122}
123
124#[derive(Debug, Clone, Default, Serialize, Deserialize)]
125pub struct IntentJournal {
126    pub flush_uuid: String,
127    pub keys: Vec<String>,
128    pub recent_uuids: Vec<String>,
129    /// The ETag of the manifest document THIS workspace last installed,
130    /// written immediately after the CAS.
131    ///
132    /// The merge base (`Baseline::inst_base`) and the baseline are both
133    /// rewritten at step 7, after the CAS and after the GC deletes. A
134    /// container restart in that window leaves the bucket holding a
135    /// document we wrote and our persisted merge base one generation
136    /// behind it — so at the next merge our own entries read as foreign
137    /// changes, delete/modify resolves conservatively against the
138    /// agent's own delete, and the path is queued into the inbox as a
139    /// conflict nobody else ever touched. Recording the installed ETag
140    /// costs one small local write and restores exactly what step 7 was
141    /// going to say: if the bucket is still at this document, the merge
142    /// base IS this document.
143    #[serde(default)]
144    pub installed_etag: Option<String>,
145    /// The other writers' changes the install at `installed_etag` carried
146    /// into the manifest, journalled with that etag in the same write.
147    ///
148    /// Step 7 moves them into the foreign queue before the merge base
149    /// passes them. A restart before it leaves the merge base at the
150    /// installed document (above), where they read as integrated: the
151    /// tree would never receive them and `remote.seq` would say there is
152    /// no news. The next consume re-queues whatever is still here.
153    #[serde(default, skip_serializing_if = "Vec::is_empty")]
154    pub installed_foreign: Vec<ForeignChange>,
155    /// Paths whose DECLARED removal this barrier unlinked and is about
156    /// to cite out (delete/rename design §4). Journalled before the
157    /// window commitment, so a crash after the cell has been told and
158    /// before the manifest CAS still produces ONE generation without
159    /// them, instead of handing the deletion to the two-scan path and
160    /// citing a file the tree no longer has for a barrier.
161    #[serde(default, skip_serializing_if = "Vec::is_empty")]
162    pub declared_deletes: Vec<String>,
163}
164
165/// One surfaced conflict: both versions stay recoverable (local bytes in
166/// the tree, foreign bytes preserved at `preserved_key` or still current
167/// under the data key). Never a silent winner.
168#[derive(Debug, Clone, Serialize, Deserialize)]
169pub struct ConflictRecord {
170    pub path: String,
171    pub foreign_etag: String,
172    /// Where the foreign bytes were preserved (server-side copy), when
173    /// the local version is about to overwrite them.
174    pub preserved_key: Option<String>,
175    pub kind: String, // "consume-dirty" | "upload-412-parked" | "sync-dirty" | "gc-skip"
176    pub at_unix: u64,
177}
178
179pub struct SyncerState {
180    dir: PathBuf,
181    /// The state-directory occupancy lock (flock, held for the process
182    /// lifetime). Self-recognition of the lease via the persisted
183    /// incarnation id is only sound because the PREVIOUS process is
184    /// gone — and this lock is what makes that true. Without it, a
185    /// second flint-sync on the same workspace self-recognizes,
186    /// deposes a LIVE sibling, and both write the tree concurrently
187    /// (observed on the 0b rig: a diagnostic re-run raced a live 1M
188    /// checkout into tmp-rename ENOENT collisions). The hub has the
189    /// identical gate (`state_backend::is_single_occupant`).
190    _lock: std::fs::File,
191}
192
193const MARKER: &str = "checkout-complete";
194/// Written by the SIGTERM drain only after it published: the node
195/// plugin's evidence that a worker that is now gone actually drained
196/// (audit 2026-09-03, finding 3). Absent ⇒ the tree is preserved.
197pub const DRAINED: &str = "drained.json";
198const LOCK: &str = "lock";
199const BASELINE: &str = "baseline.json";
200/// The admitted set of a SCOPED checkout. Deliberately its own
201/// document rather than a field on `Baseline`: the baseline is
202/// rewritten by every barrier, every sync and every consume, and a
203/// scope that rides along is one `Baseline::default()` away from
204/// silently widening a workspace to the whole tree. This file is
205/// written once, by checkout, and thereafter only read. Absent means
206/// UNSCOPED — the only encoding of "no restriction", since an empty
207/// admitted set is refused before it can be written.
208const SCOPE: &str = "scope.json";
209
210/// A rescope IN FLIGHT: the target scope, written before the first
211/// mutation and removed only after the last (scoped-read design §4.3).
212///
213/// Its own document and not a field on `scope.json`, because the two
214/// answer different questions — `scope.json` is what this workspace
215/// holds, this is what it is on its way to holding — and a crash must
216/// leave both answers readable.
217///
218/// `null` is a LEGAL target (the whole tree), so absence of the file
219/// means "no rescope in flight", never "unscoped". That is why the
220/// document wraps the target in a struct instead of being a bare
221/// `Option<Vec<String>>`: a bare `null` on disk and a missing file
222/// would deserialize to the same thing.
223const SCOPE_INTENT: &str = "scope-intent.json";
224const INCARNATION: &str = "incarnation.json";
225const INTENT: &str = "intent.json";
226const FOREIGN_QUEUE: &str = "foreign-queue.json";
227/// When this writer last swept for untracked uploads (`untracked.rs`).
228const UNTRACKED_SWEEP: &str = "untracked-sweep";
229const CONFLICTS: &str = "conflicts.jsonl";
230/// The rotated generation. `load_conflicts` reads it FIRST, so the
231/// sequence a reader sees is unbroken across a rotation — which matters
232/// because `honor_sync` takes the count before a sync and `skip`s it
233/// after, and a rotation that SHORTENED the list would make it skip past
234/// the very records the sync just produced.
235const CONFLICTS_PREV: &str = "conflicts.1.jsonl";
236/// How many records were dropped when a SECOND rotation overwrote the
237/// first. Truncation that nobody can count is truncation that reads as
238/// "there were no more".
239const CONFLICTS_DROPPED: &str = "conflicts.dropped";
240/// Rotate past this. The file is append-only and `load_conflicts` parses
241/// it WHOLE — twice per sync honor, again per status read and per
242/// scrape (U23) — so an unbounded file is an unbounded parse on a hot
243/// path, and a standing condition (a UI autosaving over a path the agent
244/// holds dirty) appends on every barrier. Two generations bound the
245/// resident set at ~2 MiB, which is thousands of records.
246const CONFLICTS_MAX_BYTES: u64 = 1 << 20;
247
248fn write_atomic(path: &Path, bytes: &[u8]) -> LeanResult<()> {
249    // The state dir lives in the same app-writable emptyDir.
250    super::safefs::check_parent(path)?;
251    let tmp = path.with_extension("tmp");
252    super::safefs::write_via_tmp(path, &tmp, bytes, None).map(|_| ())
253}
254
255impl SyncerState {
256    pub fn open(dir: PathBuf) -> LeanResult<SyncerState> {
257        fs::create_dir_all(&dir)?;
258        let lock = fs::OpenOptions::new()
259            .create(true)
260            .truncate(false)
261            .write(true)
262            .open(dir.join(LOCK))?;
263        #[cfg(unix)]
264        {
265            use std::os::fd::AsRawFd;
266            // SAFETY: flock on an owned, open fd.
267            let rc = unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
268            if rc != 0 {
269                return Err(LeanError::State(format!(
270                    "another flint-sync already holds this workspace ({}): \
271                     refusing to run two syncers over one tree",
272                    dir.display()
273                )));
274            }
275        }
276        Ok(SyncerState { dir, _lock: lock })
277    }
278
279    pub fn marker_present(&self) -> bool {
280        self.dir.join(MARKER).exists()
281    }
282
283    /// Flush the filesystem holding the tree (the state dir lives in
284    /// it) so everything materialised so far is on stable storage.
285    pub fn sync_tree(&self) -> LeanResult<()> {
286        super::safefs::sync_tree(&self.dir)
287    }
288
289    /// The drain's attestation. `seq` is the manifest the drain left
290    /// installed, when it knows it.
291    pub fn write_drained(&self, seq: Option<u64>, acks: usize) -> LeanResult<()> {
292        let doc = serde_json::json!({
293            "unix": super::now_unix(),
294            "seq": seq,
295            "acks": acks,
296        });
297        write_atomic(&self.dir.join(DRAINED), doc.to_string().as_bytes())
298    }
299
300    /// A fresh incarnation owes its own attestation; a stale one from
301    /// an earlier life of this tree must not vouch for it.
302    pub fn clear_drained(&self) -> LeanResult<()> {
303        match fs::remove_file(self.dir.join(DRAINED)) {
304            Ok(()) => Ok(()),
305            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
306            Err(e) => Err(LeanError::State(format!("clear {DRAINED}: {e}"))),
307        }
308    }
309
310    pub fn drained_path(&self) -> PathBuf {
311        self.dir.join(DRAINED)
312    }
313
314    /// Written LAST at checkout: the agent-start gate.
315    pub fn write_marker(&self) -> LeanResult<()> {
316        write_atomic(&self.dir.join(MARKER), b"ok\n")
317    }
318
319    pub fn load_baseline(&self) -> LeanResult<Baseline> {
320        let p = self.dir.join(BASELINE);
321        if !p.exists() {
322            return Ok(Baseline::default());
323        }
324        let bytes = fs::read(&p)?;
325        serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("baseline: {e}")))
326    }
327
328    /// COMPACT: the baseline is the other O(files) document, rewritten
329    /// on the cadence tick and re-parsed several times per barrier. The
330    /// agent-facing O(1) files (gauges, capabilities, acks) stay pretty
331    /// — those exist to be `cat`-ed.
332    pub fn save_baseline(&self, b: &Baseline) -> LeanResult<()> {
333        let bytes =
334            serde_json::to_vec(b).map_err(|e| LeanError::State(format!("baseline: {e}")))?;
335        write_atomic(&self.dir.join(BASELINE), &bytes)
336    }
337
338    /// What this workspace was admitted to hold, or `None` for an
339    /// unscoped (whole-manifest) checkout.
340    ///
341    /// `fs::read` directly, with NO `exists()` pre-check: `exists()`
342    /// answers false for EACCES and EIO alike, and an unreadable scope
343    /// answered as "unscoped" is the same class of bug as an unreadable
344    /// path answered as "deleted" (`barrier.rs`, 14b3637c). Only
345    /// NotFound means absent.
346    pub fn load_scope(&self) -> LeanResult<Option<Vec<String>>> {
347        let p = self.dir.join(SCOPE);
348        let bytes = match fs::read(&p) {
349            Ok(b) => b,
350            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
351            Err(e) => {
352                return Err(LeanError::State(format!(
353                    "cannot read the checkout scope at {}: {e} — refusing to \
354                     guess what this workspace is holding",
355                    p.display()
356                )))
357            }
358        };
359        serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("scope: {e}")))
360    }
361
362    /// `Some` writes the admitted set; `None` REMOVES the document.
363    /// Removing matters: a scoped checkout that crashed before its
364    /// marker leaves a scope behind, and the whole-tree checkout that
365    /// replaces it must not inherit a claim it did not make.
366    pub fn save_scope(&self, scope: Option<&[String]>) -> LeanResult<()> {
367        let p = self.dir.join(SCOPE);
368        match scope {
369            Some(entries) => {
370                let bytes = serde_json::to_vec(entries)
371                    .map_err(|e| LeanError::State(format!("scope: {e}")))?;
372                write_atomic(&p, &bytes)
373            }
374            None => match fs::remove_file(&p) {
375                Ok(()) => Ok(()),
376                Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
377                Err(e) => Err(LeanError::State(format!("clear {SCOPE}: {e}"))),
378            },
379        }
380    }
381
382    /// Read a rescope in flight. `None` = none in flight.
383    pub fn load_scope_intent(&self) -> LeanResult<Option<ScopeIntent>> {
384        let p = self.dir.join(SCOPE_INTENT);
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 rescope intent at {}: {e} — refusing to run a barrier \
391                     over a half-applied scope",
392                    p.display()
393                )))
394            }
395        };
396        serde_json::from_slice(&bytes)
397            .map(Some)
398            .map_err(|e| LeanError::State(format!("scope intent: {e}")))
399    }
400
401    pub fn save_scope_intent(&self, intent: &ScopeIntent) -> LeanResult<()> {
402        let bytes = serde_json::to_vec(intent)
403            .map_err(|e| LeanError::State(format!("scope intent: {e}")))?;
404        write_atomic(&self.dir.join(SCOPE_INTENT), &bytes)
405    }
406
407    /// Cleared LAST, after the new scope is durable. Clearing it first
408    /// would turn a crash into a workspace whose scope says one thing
409    /// and whose held set says another, with nothing left to replay.
410    pub fn clear_scope_intent(&self) -> LeanResult<()> {
411        match fs::remove_file(self.dir.join(SCOPE_INTENT)) {
412            Ok(()) => Ok(()),
413            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
414            Err(e) => Err(LeanError::State(format!("clear {SCOPE_INTENT}: {e}"))),
415        }
416    }
417
418    pub fn load_incarnation(&self) -> LeanResult<Option<Incarnation>> {
419        let p = self.dir.join(INCARNATION);
420        if !p.exists() {
421            return Ok(None);
422        }
423        let bytes = fs::read(&p)?;
424        serde_json::from_slice(&bytes)
425            .map(Some)
426            .map_err(|e| LeanError::State(format!("incarnation: {e}")))
427    }
428
429    pub fn save_incarnation(&self, i: &Incarnation) -> LeanResult<()> {
430        let bytes = serde_json::to_vec_pretty(i)
431            .map_err(|e| LeanError::State(format!("incarnation: {e}")))?;
432        write_atomic(&self.dir.join(INCARNATION), &bytes)
433    }
434
435    /// The writer-local foreign-change queue. Only NotFound means empty
436    /// (an unreadable queue answered as empty would strand every change
437    /// in it).
438    pub fn load_foreign_queue(&self) -> LeanResult<Vec<ForeignChange>> {
439        match fs::read(self.dir.join(FOREIGN_QUEUE)) {
440            Ok(bytes) => serde_json::from_slice(&bytes)
441                .map_err(|e| LeanError::State(format!("foreign queue: {e}"))),
442            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(vec![]),
443            Err(e) => Err(e.into()),
444        }
445    }
446
447    pub fn save_foreign_queue(&self, q: &[ForeignChange]) -> LeanResult<()> {
448        let bytes =
449            serde_json::to_vec(q).map_err(|e| LeanError::State(format!("foreign queue: {e}")))?;
450        write_atomic(&self.dir.join(FOREIGN_QUEUE), &bytes)
451    }
452
453    /// Upsert `changes` into the foreign queue by path — a later change to
454    /// a path replaces the queued one — and save it. Idempotent.
455    pub fn queue_foreign(&self, changes: &[ForeignChange]) -> LeanResult<()> {
456        let mut q = self.load_foreign_queue()?;
457        for c in changes {
458            q.retain(|x| x.path != c.path);
459            q.push(c.clone());
460        }
461        self.save_foreign_queue(&q)
462    }
463
464    /// Move an install's journalled foreign changes into the queue, if a
465    /// restart kept its step 7 from doing so (`installed_foreign`). The
466    /// queue is saved before the journal lets go of them.
467    pub fn requeue_installed_foreign(&self) -> LeanResult<usize> {
468        let mut j = self.load_intent()?;
469        if j.installed_foreign.is_empty() {
470            return Ok(0);
471        }
472        let n = j.installed_foreign.len();
473        self.queue_foreign(&j.installed_foreign)?;
474        j.installed_foreign.clear();
475        self.save_intent(&j)?;
476        Ok(n)
477    }
478
479    /// Unix seconds of this writer's last untracked-upload sweep; 0 when it
480    /// has never swept.
481    pub fn load_untracked_sweep_at(&self) -> LeanResult<u64> {
482        match fs::read_to_string(self.dir.join(UNTRACKED_SWEEP)) {
483            Ok(s) => Ok(s.trim().parse().unwrap_or(0)),
484            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
485            Err(e) => Err(e.into()),
486        }
487    }
488
489    pub fn save_untracked_sweep_at(&self, at: u64) -> LeanResult<()> {
490        write_atomic(&self.dir.join(UNTRACKED_SWEEP), at.to_string().as_bytes())
491    }
492
493    pub fn load_intent(&self) -> LeanResult<IntentJournal> {
494        let p = self.dir.join(INTENT);
495        if !p.exists() {
496            return Ok(IntentJournal::default());
497        }
498        let bytes = fs::read(&p)?;
499        serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("intent: {e}")))
500    }
501
502    pub fn save_intent(&self, j: &IntentJournal) -> LeanResult<()> {
503        let bytes =
504            serde_json::to_vec_pretty(j).map_err(|e| LeanError::State(format!("intent: {e}")))?;
505        write_atomic(&self.dir.join(INTENT), &bytes)
506    }
507
508    /// Clear the per-barrier key list but KEEP the uuid history (the
509    /// AdoptOwn recognizer needs uuids from completed barriers whose
510    /// baseline rewrite raced a crash).
511    pub fn clear_intent_keys(&self) -> LeanResult<()> {
512        let mut j = self.load_intent()?;
513        if !j.flush_uuid.is_empty() {
514            if !j.recent_uuids.contains(&j.flush_uuid) {
515                j.recent_uuids.push(j.flush_uuid.clone());
516            }
517            let excess = j.recent_uuids.len().saturating_sub(8);
518            if excess > 0 {
519                j.recent_uuids.drain(..excess);
520            }
521        }
522        j.flush_uuid = String::new();
523        j.keys.clear();
524        j.declared_deletes.clear();
525        j.installed_foreign.clear();
526        self.save_intent(&j)
527    }
528
529    pub fn append_conflict(&self, c: &ConflictRecord) -> LeanResult<()> {
530        use std::io::Write;
531        let line =
532            serde_json::to_string(c).map_err(|e| LeanError::State(format!("conflict: {e}")))?;
533        let live = self.dir.join(CONFLICTS);
534        let mut f = fs::OpenOptions::new().create(true).append(true).open(&live)?;
535        writeln!(f, "{line}")?;
536        // Size off the handle we just wrote through — no extra stat of
537        // the path, and no line count, which would make every append
538        // O(records).
539        let len = f.metadata()?.len();
540        drop(f);
541        if len > CONFLICTS_MAX_BYTES {
542            self.rotate_conflicts()?;
543        }
544        Ok(())
545    }
546
547    /// Rotate live -> `.1`, counting whatever the previous `.1` held so
548    /// the loss is a NUMBER rather than a silence. Best effort: a
549    /// rotation that fails leaves a large file, which is a performance
550    /// problem; losing the record we just appended would be a
551    /// correctness one.
552    fn rotate_conflicts(&self) -> LeanResult<()> {
553        let live = self.dir.join(CONFLICTS);
554        let prev = self.dir.join(CONFLICTS_PREV);
555        // Only read the outgoing generation at ROTATION time — rare by
556        // construction, since it takes CONFLICTS_MAX_BYTES to get here.
557        if let Ok(text) = fs::read_to_string(&prev) {
558            let dropped = text.lines().filter(|l| !l.trim().is_empty()).count() as u64;
559            let total = self.conflicts_dropped()? + dropped;
560            write_atomic(&self.dir.join(CONFLICTS_DROPPED), total.to_string().as_bytes())?;
561        }
562        fs::rename(&live, &prev)?;
563        Ok(())
564    }
565
566    /// Records lost to rotation. A caller rendering conflicts owes the
567    /// user this number: a truncated list that does not say it was
568    /// truncated reads as a complete one.
569    pub fn conflicts_dropped(&self) -> LeanResult<u64> {
570        match fs::read_to_string(self.dir.join(CONFLICTS_DROPPED)) {
571            Ok(s) => Ok(s.trim().parse().unwrap_or(0)),
572            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
573            Err(e) => Err(LeanError::State(format!("{CONFLICTS_DROPPED}: {e}"))),
574        }
575    }
576
577    /// The rotated generation first, then the live one: oldest to
578    /// newest, unbroken across a single rotation.
579    pub fn load_conflicts(&self) -> LeanResult<Vec<ConflictRecord>> {
580        let mut out = vec![];
581        for name in [CONFLICTS_PREV, CONFLICTS] {
582            let p = self.dir.join(name);
583            // `fs::read_to_string` directly: `exists()` answers false
584            // for EACCES and EIO alike, and a conflict log that reads as
585            // EMPTY because it is unreadable is the same shape of bug as
586            // an unreadable path reading as deleted.
587            let text = match fs::read_to_string(&p) {
588                Ok(t) => t,
589                Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
590                Err(e) => {
591                    return Err(LeanError::State(format!(
592                        "cannot read {name}: {e} — refusing to report an unreadable \
593                         conflict log as an empty one"
594                    )))
595                }
596            };
597            for line in text.lines() {
598                if line.trim().is_empty() {
599                    continue;
600                }
601                out.push(
602                    serde_json::from_str(line)
603                        .map_err(|e| LeanError::State(format!("conflict line: {e}")))?,
604                );
605            }
606        }
607        Ok(out)
608    }
609}