flint_lean/untracked.rs
1//! Finding 10: an upload whose commit never came.
2//!
3//! Uploads hold no lease. A writer lost for good between an upload and its
4//! commit — its pod replaced, its node gone — leaves bytes at a key the
5//! manifest cites under a different etag, and nothing tracks them: no
6//! manifest cites them, no inbox entry names them, and the writer that
7//! would have cited them is gone. A fresh checkout reads them (the S3-wins
8//! arm) while a live writer keeps the cited version, so the two disagree
9//! for as long as nobody rewrites the path. A NEW writer heals it — its
10//! checkout adopts the bytes and its next commit's citation repair cites
11//! them (measured 2026-09-15) — but a workspace whose writers all outlive
12//! the loss never checks out again.
13//!
14//! The sweep tracks such an object the way the gateway tracks a UI write:
15//! an inbox entry, which every writer's consume integrates (adopted where
16//! the path is clean, preserved beside the agent's edit where it is not)
17//! and the next commit cites. The formal model checks the rule: tracked at
18//! ANY time, a live writer's upload still in flight included, every
19//! invariant holds (`LeanBarrierLeaseOrphanTracked`), so the grace below is
20//! churn control rather than a safety margin; without it, a world in which
21//! every syncer has gone quiet still holds such bytes
22//! (`LeanBarrierLeaseOrphanDiverges`).
23
24use std::collections::HashMap;
25
26use super::inbox::{self, InboxEntry};
27use super::{manifest, LeanError, LeanResult, Syncer};
28
29/// The author an inbox entry from the sweep carries.
30pub const UNTRACKED_AUTHOR: &str = "untracked-sweep";
31
32impl Syncer {
33 /// Track every object at a CITED key whose etag the manifest does not
34 /// cite, that no inbox entry names, and that has sat so for at least
35 /// `untracked_grace_secs`. Returns the paths tracked. An object at a
36 /// key no manifest cites is not this: no checkout serves it (a delete
37 /// whose GC never ran leaves one), and it is garbage, not a
38 /// disagreement.
39 ///
40 /// One manifest read, one inbox read and a LIST of the files prefix.
41 /// An open barrier window stops the sweep; the next one resumes it.
42 pub async fn track_untracked(&self, now: u64) -> LeanResult<Vec<String>> {
43 let Some(loaded) = manifest::load(self.store.as_ref(), &self.cfg).await? else {
44 return Ok(vec![]);
45 };
46 let cited: HashMap<&str, (&str, &str)> = loaded
47 .manifest
48 .entries
49 .iter()
50 .map(|(p, e)| (e.key.as_str(), (p.as_str(), e.etag.as_str())))
51 .collect();
52 let doc = inbox::load(self.store.as_ref(), &self.cfg).await?.doc;
53 let same = |a: &str, b: &str| a.trim_matches('"') == b.trim_matches('"');
54 let listed = self.store.list(&format!("{}/files/", self.cfg.prefix)).await?;
55 let mut tracked = vec![];
56 for obj in listed {
57 let Some(&(path, cited_etag)) = cited.get(obj.key.as_str()) else { continue };
58 if same(cited_etag, &obj.etag) {
59 continue;
60 }
61 if doc.entries.iter().any(|e| e.path == path && same(&e.etag, &obj.etag)) {
62 continue;
63 }
64 // No modification time, no judgement of age: leave it.
65 let Some(modified) = obj.last_modified_unix else { continue };
66 if now.saturating_sub(modified) < self.cfg.untracked_grace_secs {
67 continue;
68 }
69 let entry = InboxEntry {
70 path: path.to_string(),
71 etag: obj.etag.clone(),
72 author: UNTRACKED_AUTHOR.into(),
73 added_unix: now,
74 // Nobody vouches for these bytes but the backend: the
75 // consume checks them against its attestation, as it does
76 // any entry without a writer's CRC.
77 crc64_b64: None,
78 };
79 match inbox::gateway_append(self.store.as_ref(), &self.cfg, entry).await {
80 Ok(()) => {
81 self.trace("untracked", serde_json::json!({"path": path, "etag": obj.etag, "cited": cited_etag}));
82 tracked.push(path.to_string());
83 }
84 Err(LeanError::State(m)) if m.contains("window open") => break,
85 Err(e) => return Err(e),
86 }
87 }
88 Ok(tracked)
89 }
90
91 /// The floor's hook: sweep when `untracked_sweep_secs` have passed since
92 /// the last sweep (0 = never). A writer's first tick starts the clock
93 /// instead of sweeping — its own checkout has just read every citation.
94 /// Best-effort: a sweep that fails is retried on the next tick and never
95 /// fails the floor.
96 pub async fn sweep_untracked_if_due(&self, now: u64) {
97 if self.cfg.untracked_sweep_secs == 0 || self.cfg.access.is_read() {
98 return;
99 }
100 let last = match self.state.load_untracked_sweep_at() {
101 Ok(t) => t,
102 Err(e) => {
103 eprintln!("flint-sync: untracked sweep clock unreadable (skipping): {e}");
104 return;
105 }
106 };
107 if last == 0 {
108 let _ = self.state.save_untracked_sweep_at(now);
109 return;
110 }
111 if now.saturating_sub(last) < self.cfg.untracked_sweep_secs {
112 return;
113 }
114 match self.track_untracked(now).await {
115 Ok(paths) => {
116 if !paths.is_empty() {
117 eprintln!(
118 "flint-sync: tracked {} upload(s) no commit cited, left by a lost writer: {paths:?}",
119 paths.len()
120 );
121 }
122 let _ = self.state.save_untracked_sweep_at(now);
123 }
124 Err(e) => eprintln!("flint-sync: untracked sweep failed (retrying next tick): {e}"),
125 }
126 }
127}