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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
//! `.flint-sync/gauges.json` — the Phase-3 observability minimum
//! (§2.6; review ledger OF-6).
//!
//! The question this file exists to answer is "why is the manifest not
//! advancing", and it has to be answerable from inside the pod, with
//! `cat`, before Phase 6's `/metrics` exists. A healthy no-change tick
//! and a wedged loop look identical from the outside: both publish
//! nothing.
//!
//! **Every field here is computed from LOCAL state — no bucket request,
//! ever.** That is enforced by the signature rather than by convention:
//! `write_gauges` is not `async` and takes no store, so it *cannot*
//! issue one. This is the "instrument reports on itself" class the
//! runas campaign paid for five times; here it would also make the
//! zero-added-cost oracle (leg B8) intermittently red and blame the
//! syncer for its own instrument.
//!
//! Phase 6 renders `/metrics` from this same struct — one renderer over
//! one struct cannot drift; two computations would.
use serde::{Deserialize, Serialize};
use super::{now_unix, LeanResult, Syncer};
const GAUGES: &str = "gauges.json";
/// Which coherent point last installed a citation, and what it named.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct LastBoundary {
pub source: String,
pub seq: u64,
pub unix: u64,
}
/// Why visibility is currently withheld. `None` = nothing is withheld.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Withheld {
/// A foreign 412 parked at least one path (it is NOT ours to
/// publish, and the conflict record says so).
Parked412,
/// The barrier held at least one path the syncer may not read, or
/// whose name is not UTF-8 (`BarrierReport::held`); the rest of the
/// tree publishes.
Unreadable,
}
impl Withheld {
pub fn as_str(&self) -> &'static str {
match self {
Withheld::Parked412 => "parked-412",
Withheld::Unreadable => "unreadable",
}
}
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Gauges {
/// Since the last boundary was installed. Elapsed time, NOT
/// exposure: an idle healthy workspace has nothing at risk and a
/// growing `rpo_secs`. Pair it with `withheld_reason`, which carries
/// whether there is anything to lose. Alerting on this number alone
/// pages someone for a workspace that is simply quiet.
pub rpo_secs: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub withheld_reason: Option<String>,
pub sentinel_budget_remaining: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_boundary: Option<LastBoundary>,
/// Unix time of the first renewal the store refused with 401/403
/// (`StoreError::Auth`) since the last successful one — a
/// credential or broker fault, never contention and never a lease
/// conflict. `None` once a renewal succeeds.
///
/// This is the ONLY liveness fact a credential-paused holder can
/// still record. The renewal that would carry it into the lease
/// echo is precisely the request that is failing, so the store
/// cannot be told; local evidence is all there is. Without it an
/// operator sees a lease going stale and a pod that is plainly
/// Running, and nothing that connects the two — the diagnosis
/// `StoreError::Auth` was split out of `Other` to make possible in
/// the first place (`flint-store/src/lib.rs`), stopping one layer
/// short of the binary that needed it.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub auth_paused_since_unix: Option<u64>,
/// Refreshed on EVERY tick, news or not: an idle-but-healthy
/// workspace must be distinguishable from a dead one (the same
/// heartbeat rule `remote.seq` carries).
pub updated_unix: u64,
/// Carried across ticks so `rpo_secs` measures durability rather
/// than "time since this process started".
#[serde(default)]
pub last_durable_unix: u64,
}
impl Syncer {
fn gauges_path(&self) -> std::path::PathBuf {
self.cfg.state_dir().join(GAUGES)
}
/// Last written gauges, or defaults. Never fails on a corrupt file:
/// a diagnosis surface that panics is worse than a stale one.
pub fn load_gauges(&self) -> LeanResult<Gauges> {
let p = self.gauges_path();
if !p.exists() {
return Ok(Gauges::default());
}
Ok(std::fs::read(&p)
.ok()
.and_then(|b| serde_json::from_slice(&b).ok())
.unwrap_or_default())
}
/// Record that a boundary installed.
pub fn note_boundary(&self, source: &str, seq: u64) -> LeanResult<()> {
let mut g = self.load_gauges()?;
g.last_boundary = Some(LastBoundary { source: source.into(), seq, unix: now_unix() });
g.last_durable_unix = now_unix();
self.save_gauges(&g)
}
/// Record that the store refused our credentials on a renewal.
/// First refusal wins: the gauge answers "since when", so a second
/// consecutive failure must not keep resetting the clock to now.
pub fn note_auth_pause(&self) -> LeanResult<()> {
let mut g = self.load_gauges()?;
if g.auth_paused_since_unix.is_none() {
g.auth_paused_since_unix = Some(now_unix());
self.save_gauges(&g)?;
}
Ok(())
}
/// Clear the pause. Writes only on an actual transition: this runs
/// on every successful renewal and floor tick, and the healthy path
/// must not rewrite the file every tick for no change.
pub fn clear_auth_pause(&self) -> LeanResult<()> {
let mut g = self.load_gauges()?;
if g.auth_paused_since_unix.is_some() {
g.auth_paused_since_unix = None;
self.save_gauges(&g)?;
}
Ok(())
}
fn save_gauges(&self, g: &Gauges) -> LeanResult<()> {
let bytes = serde_json::to_vec_pretty(g)
.map_err(|e| super::LeanError::State(format!("gauges: {e}")))?;
super::control::write_atomic(&self.gauges_path(), &bytes)
}
/// Recompute and write the gauges. Deliberately synchronous and
/// store-free: a scrape, a tick and an exec all cost zero bucket
/// requests, and the type system is what says so.
pub fn write_gauges(&self, withheld: Option<Withheld>) -> LeanResult<Gauges> {
let now = now_unix();
let prev = self.load_gauges()?;
let budget = self.load_budget().unwrap_or_default();
let g = Gauges {
rpo_secs: if prev.last_durable_unix == 0 {
0
} else {
now.saturating_sub(prev.last_durable_unix)
},
withheld_reason: withheld.map(|w| w.as_str().to_string()),
sentinel_budget_remaining: budget.remaining(now, self.cfg.sentinel_hourly_budget),
last_boundary: prev.last_boundary.clone(),
updated_unix: now,
last_durable_unix: prev.last_durable_unix,
// Carried, not recomputed: `write_gauges` is store-free by
// construction, so it cannot observe the credential state
// that set this. Dropping it here would erase the pause on
// the very next tick — the gauge would exist and always
// read `None`.
auth_paused_since_unix: prev.auth_paused_since_unix,
};
self.save_gauges(&g)?;
Ok(g)
}
}
/// What `flint-sync status` renders. Read STRICTLY from files: the verb
/// exists to diagnose a workspace whose syncer is dead or deposed, so
/// it must neither claim the lease (which would depose the very syncer
/// under diagnosis) nor take the state-directory occupancy lock (which
/// a live syncer already holds).
#[derive(Debug, Clone, Serialize)]
pub struct StatusReport {
pub root: String,
pub prefix: String,
pub gauges: Option<Gauges>,
pub capabilities: Option<super::control::Capabilities>,
pub remote_seq: Option<super::control::RemoteSeq>,
pub baseline_seq: u64,
pub incarnation_epoch: Option<u64>,
pub incarnation_holder: Option<String>,
/// A pending sentinel still standing (verb name) — the "my agent is
/// blocked on an ack" question.
pub pending_sentinels: Vec<String>,
pub recent_conflicts: usize,
pub checkout_complete: bool,
}
fn read_json<T: serde::de::DeserializeOwned>(p: &std::path::Path) -> Option<T> {
serde_json::from_slice(&std::fs::read(p).ok()?).ok()
}
pub fn status_report(cfg: &super::LeanConfig) -> LeanResult<StatusReport> {
let sd = cfg.state_dir();
let cd = cfg.control_dir();
// The snapshot AND its journal (lever D): the snapshot alone is the
// baseline as of the last whole write.
let baseline = super::state::read_baseline_at(&sd).unwrap_or_default();
let inc: Option<super::state::Incarnation> = read_json(&sd.join("incarnation.json"));
let mut pending_sentinels = vec![];
// Named by the SAME function that writes them. The previous form
// built the name a second time and got it wrong, so this field —
// the "is my agent blocked on an ack?" answer — was permanently
// empty on a workspace that had a sentinel standing.
for verb in [super::sentinel::Verb::Publish, super::sentinel::Verb::Sync] {
if sd.join(verb.pending_name()).exists() {
pending_sentinels.push(verb.sentinel_name().to_string());
}
}
let conflicts = std::fs::read_to_string(sd.join("conflicts.jsonl"))
.map(|s| s.lines().filter(|l| !l.trim().is_empty()).count())
.unwrap_or(0);
Ok(StatusReport {
root: cfg.root.display().to_string(),
prefix: cfg.prefix.clone(),
gauges: read_json(&sd.join(GAUGES)),
capabilities: read_json(&cd.join(super::control::CAPABILITIES)),
remote_seq: read_json(&cd.join(super::control::REMOTE_SEQ)),
baseline_seq: baseline.seq,
incarnation_epoch: inc.as_ref().map(|i| i.epoch),
incarnation_holder: inc.map(|i| i.holder_id),
pending_sentinels,
recent_conflicts: conflicts,
checkout_complete: sd.join("checkout-complete").exists(),
})
}