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 integrates other writers' changes and the inbox
5//! into its tree as a side effect of publishing. A reader has nothing to
6//! publish, so its tick is the pull alone: the inbox 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 (a writer's next merge must
17//! still see a manifest change an inbox overlay hid), so the baseline
18//! would report "moved" on every tick and turn an idle reader into a
19//! manifest download per floor.
20
21use serde::{Deserialize, Serialize};
22
23use flint_store::{GenerationStamps, StoreError};
24
25use super::sync::SyncReport;
26use super::{inbox, manifest, LeanError, LeanResult, Syncer};
27
28/// State-dir file: the documents the last completed pull synced against.
29const PULLED_FROM: &str = "reader.json";
30
31#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
32struct PulledFrom {
33    manifest_etag: Option<String>,
34    inbox_etag: Option<String>,
35}
36
37/// What one reader tick found.
38#[derive(Debug, Default)]
39pub struct Pull {
40    /// The manifest's seq and etag as the pointer read reported them —
41    /// the news ticker's input, at no added request.
42    pub observed_seq: Option<u64>,
43    pub observed_etag: Option<String>,
44    /// `None`: neither document moved since the last pull, and nothing
45    /// was synced.
46    pub sync: Option<SyncReport>,
47}
48
49impl Syncer {
50    /// The refusal every publishing or fencing path gives a reader.
51    /// `Refused` because it is configuration, not contention: retrying
52    /// cannot make a read-only syncer write.
53    pub fn refuse_if_read(&self, what: &str) -> LeanResult<()> {
54        if self.cfg.access.is_read() {
55            return Err(LeanError::Refused(format!(
56                "{what}: this syncer has read access (FLINT_SYNC_ACCESS=read) and never writes to \
57                 the bucket"
58            )));
59        }
60        Ok(())
61    }
62
63    /// One reader tick: two GETs, and a `sync` only when there is news.
64    pub async fn reader_pull(&mut self) -> LeanResult<Pull> {
65        let ib = inbox::load(self.store.as_ref(), &self.cfg).await?;
66        let (observed_seq, manifest_etag) =
67            match manifest::load_pointer(self.store.as_ref(), &self.cfg).await? {
68                Some(p) => (Some(p.pointer.seq), Some(p.etag)),
69                // A workspace from before the pointer layout: the barrier's
70                // own fallback, a HEAD of the legacy document.
71                None => match self.store.head(&self.cfg.manifest_key()).await {
72                    Ok(meta) => {
73                        (GenerationStamps::from_meta(&meta.meta).map(|s| s.generation), Some(meta.etag))
74                    }
75                    Err(StoreError::NotFound(_)) => (None, None),
76                    Err(e) => return Err(e.into()),
77                },
78            };
79        let now = PulledFrom { manifest_etag: manifest_etag.clone(), inbox_etag: ib.etag };
80        let mut out = Pull { observed_seq, observed_etag: manifest_etag, sync: None };
81        if self.load_pulled_from().as_ref() == Some(&now) {
82            return Ok(out);
83        }
84        // `sync` reads both documents again. If either moved in between,
85        // what is remembered below is the OLDER etag, and the next tick
86        // syncs once more — the safe direction; the other would skip news.
87        out.sync = Some(self.sync().await?);
88        self.save_pulled_from(&now)?;
89        Ok(out)
90    }
91
92    fn load_pulled_from(&self) -> Option<PulledFrom> {
93        let bytes = std::fs::read(self.cfg.state_dir().join(PULLED_FROM)).ok()?;
94        serde_json::from_slice(&bytes).ok()
95    }
96
97    fn save_pulled_from(&self, p: &PulledFrom) -> LeanResult<()> {
98        let bytes =
99            serde_json::to_vec(p).map_err(|e| LeanError::State(format!("reader memo: {e}")))?;
100        // The state dir sits in the app-writable tree: the same checked
101        // write every other state file takes.
102        let path = self.cfg.state_dir().join(PULLED_FROM);
103        super::safefs::check_parent(&path)?;
104        super::safefs::write_via_tmp(&path, &path.with_extension("tmp"), &bytes, None).map(|_| ())
105    }
106}
flint_lean/reader.rs

flint_lean/
reader.rs
