flint_lean/
reader.rs

1//! Read access (per-user access design §4.4): the floor tick of a syncer
2//! that follows the workspace and never writes to the bucket.
3//!
4//! A writer's boundary takes other writers' and the gateway's commits into
5//! its tree in its first step (the consume). A reader has nothing to
6//! publish, so its tick is the pull alone: the request cell and the
7//! manifest pointer — the same two GETs an idle writer's tick makes —
8//! and a whole-tree `sync` when either has moved since the last pull.
9//! `sync` makes no store writes (`sync.rs`), applies only onto paths the
10//! scan finds clean, and records every change it declines, so a reader's
11//! tree converges the way a writer's does and never loses a local byte
12//! silently. It never publishes one either.
13//!
14//! What moved is judged by the two documents' etags, remembered here
15//! rather than read off the baseline: `sync` deliberately leaves
16//! `baseline.manifest_etag` where it was (`sync.rs`), so the baseline
17//! would report "moved" on every tick and turn an idle reader into a
18//! manifest download per floor.
19
20use serde::{Deserialize, Serialize};
21
22use flint_store::{GenerationStamps, StoreError};
23
24use super::sync::SyncReport;
25use super::{inbox, manifest, LeanError, LeanResult, Syncer};
26
27/// State-dir file: the documents the last completed pull synced against.
28const PULLED_FROM: &str = "reader.json";
29
30#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
31struct PulledFrom {
32    manifest_etag: Option<String>,
33    inbox_etag: Option<String>,
34}
35
36/// What one reader tick found.
37#[derive(Debug, Default)]
38pub struct Pull {
39    /// The manifest's seq and etag as the pointer read reported them —
40    /// the news ticker's input, at no added request.
41    pub observed_seq: Option<u64>,
42    pub observed_etag: Option<String>,
43    /// `None`: neither document moved since the last pull, and nothing
44    /// was synced.
45    pub sync: Option<SyncReport>,
46}
47
48impl Syncer {
49    /// The refusal every publishing or fencing path gives a reader.
50    /// `Refused` because it is configuration, not contention: retrying
51    /// cannot make a read-only syncer write.
52    pub fn refuse_if_read(&self, what: &str) -> LeanResult<()> {
53        if self.cfg.access.is_read() {
54            return Err(LeanError::Refused(format!(
55                "{what}: this syncer has read access (FLINT_SYNC_ACCESS=read) and never writes to \
56                 the bucket"
57            )));
58        }
59        Ok(())
60    }
61
62    /// One reader tick: two GETs, and a `sync` only when there is news.
63    pub async fn reader_pull(&mut self) -> LeanResult<Pull> {
64        let ib = inbox::load(self.store.as_ref(), &self.cfg).await?;
65        let (observed_seq, manifest_etag) =
66            match manifest::load_pointer(self.store.as_ref(), &self.cfg).await? {
67                Some(p) => (Some(p.pointer.seq), Some(p.etag)),
68                // A workspace from before the pointer layout: the barrier's
69                // own fallback, a HEAD of the legacy document.
70                None => match self.store.head(&self.cfg.manifest_key()).await {
71                    Ok(meta) => {
72                        (GenerationStamps::from_meta(&meta.meta).map(|s| s.generation), Some(meta.etag))
73                    }
74                    Err(StoreError::NotFound(_)) => (None, None),
75                    Err(e) => return Err(e.into()),
76                },
77            };
78        let now = PulledFrom { manifest_etag: manifest_etag.clone(), inbox_etag: ib.etag };
79        let mut out = Pull { observed_seq, observed_etag: manifest_etag, sync: None };
80        let skip = self.load_pulled_from().as_ref() == Some(&now) && !self.still_owed()?;
81        self.trace("pull", serde_json::json!({"seq": observed_seq, "skipped": skip}));
82        if skip {
83            return Ok(out);
84        }
85        // `sync` reads both documents again. If either moved in between,
86        // what is remembered below is the OLDER etag, and the next tick
87        // syncs once more — the safe direction; the other would skip news.
88        out.sync = Some(self.sync().await?);
89        self.save_pulled_from(&now)?;
90        Ok(out)
91    }
92
93    /// Did the last sync leave something owed? The consume's cheap-path
94    /// test on the baseline the sync wrote: nothing recorded as derived (a
95    /// fetch or a write it could not complete), or a path it skipped as
96    /// dirty that is clean again. Without it a reader judged by the two
97    /// etags alone, and a path its sync could not take waited for the
98    /// next commit (review 2026-09-29, S4).
99    fn still_owed(&self) -> LeanResult<bool> {
100        self.state.with_baseline(|b| {
101            b.derived_etag.is_none()
102                || !b.skipped.iter().all(|path| {
103                    super::barrier::check_contained(&self.cfg.root, path).is_ok()
104                        && super::barrier::local_dirty(&self.cfg.root.join(path), b.entries.get(path))
105                })
106        })
107    }
108
109    fn load_pulled_from(&self) -> Option<PulledFrom> {
110        let bytes = std::fs::read(self.cfg.state_dir().join(PULLED_FROM)).ok()?;
111        serde_json::from_slice(&bytes).ok()
112    }
113
114    fn save_pulled_from(&self, p: &PulledFrom) -> LeanResult<()> {
115        let bytes =
116            serde_json::to_vec(p).map_err(|e| LeanError::State(format!("reader memo: {e}")))?;
117        // The state dir sits in the app-writable tree: the same checked
118        // write every other state file takes.
119        let path = self.cfg.state_dir().join(PULLED_FROM);
120        super::safefs::check_parent(&path)?;
121        super::safefs::write_via_tmp(&path, &path.with_extension("tmp"), &bytes, None).map(|_| ())
122    }
123}
flint_lean/reader.rs

flint_lean/
reader.rs
