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