flint_lean/
trace.rs

1//! The protocol event trace (`FLINT_SYNC_EVENT_TRACE=1`, or
2//! `spec.eventTrace` on a deployed workspace).
3//!
4//! One JSON line per protocol step — each consume, upload outcome, claim
5//! poll, observed-citation re-read, merge, CAS, GC delete, queue write,
6//! release, fence, ack and sync — so that when several writers share a
7//! workspace and something goes wrong, the interleaving can be rebuilt
8//! from the evidence instead of re-run. A race that fails once on a live
9//! rig may not fail again. The line format is pinned in
10//! `lean/e2e/writers-live/README.md` §2: every line starts `{"ts_ms":`,
11//! then `mono_ms`, `holder`, `ev`, then the event's own fields.
12//!
13//! Off by default. On, it costs a formatted line per step on stderr and,
14//! for the holder id, one small local read per event. A writer that has
15//! not claimed yet has no lease to name it, so the trace mints (and
16//! persists) the pod's incarnation the way the first claim would: without
17//! that, a pod's first barrier — its consume, scan and uploads — carried
18//! `holder: null` and could not be joined to its own commit.
19
20use std::sync::{Arc, Mutex, OnceLock};
21use std::time::{Instant, SystemTime, UNIX_EPOCH};
22
23/// Where trace lines go.
24#[derive(Clone)]
25pub enum Sink {
26    /// One line per event on stderr, beside the prose log.
27    Stderr,
28    /// Collected in memory — for tests that read their own trace.
29    Memory(Arc<Mutex<Vec<String>>>),
30}
31
32static START: OnceLock<Instant> = OnceLock::new();
33
34/// Format one trace line. `fields` must be a JSON object; its keys follow
35/// the common prefix in the order serde_json emits them.
36pub fn line(holder: Option<&str>, ev: &str, fields: serde_json::Value) -> String {
37    let ts_ms = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
38    let mono_ms = START.get_or_init(Instant::now).elapsed().as_millis() as u64;
39    let head = format!(
40        "{{\"ts_ms\":{ts_ms},\"mono_ms\":{mono_ms},\"holder\":{},\"ev\":{}",
41        serde_json::to_string(&holder).unwrap_or_else(|_| "null".into()),
42        serde_json::to_string(ev).unwrap_or_else(|_| "\"?\"".into()),
43    );
44    let rest = match fields {
45        serde_json::Value::Object(m) if !m.is_empty() => {
46            let body = serde_json::to_string(&serde_json::Value::Object(m)).unwrap_or_default();
47            format!(",{}", &body[1..body.len() - 1])
48        }
49        _ => String::new(),
50    };
51    format!("{head}{rest}}}")
52}
53
54impl super::Syncer {
55    /// Emit one trace event, when the trace is on.
56    pub fn trace(&self, ev: &str, fields: serde_json::Value) {
57        let Some(sink) = &self.cfg.event_trace else { return };
58        let holder = self
59            .lease
60            .as_ref()
61            .map(|l| l.holder_id.clone())
62            .or_else(|| super::lease::incarnation(self).ok().map(|i| i.holder_id));
63        let l = line(holder.as_deref(), ev, fields);
64        match sink {
65            Sink::Stderr => eprintln!("{l}"),
66            Sink::Memory(buf) => buf.lock().unwrap().push(l),
67        }
68    }
69
70    /// The store's request counts, as a trace field.
71    pub fn trace_requests(&self) -> serde_json::Value {
72        match self.store.request_counts() {
73            Some(c) => serde_json::to_value(c).unwrap_or(serde_json::Value::Null),
74            None => serde_json::Value::Null,
75        }
76    }
77}
78
79#[cfg(test)]
80mod tests {
81    #[test]
82    fn a_line_starts_with_the_pinned_prefix_and_parses() {
83        let l = super::line(Some("h1"), "gc", serde_json::json!({"path": "a/b", "result": "deleted", "z": 1}));
84        assert!(l.starts_with("{\"ts_ms\":"), "{l}");
85        let v: serde_json::Value = serde_json::from_str(&l).expect("a trace line is JSON");
86        assert_eq!(v["holder"], "h1");
87        assert_eq!(v["ev"], "gc");
88        assert_eq!(v["path"], "a/b");
89        assert!(v["mono_ms"].is_u64());
90        let empty = super::line(None, "barrier_start", serde_json::json!({}));
91        let v: serde_json::Value = serde_json::from_str(&empty).expect("no fields still parses");
92        assert!(v["holder"].is_null());
93    }
94}
flint_lean/trace.rs

flint_lean/
trace.rs
