crates/nornir-saga/src/record.rs
//! The capture buffer — [`record`] / [`drain_saga_events`] — gated behind the
//! `saga` feature exactly like nornir-testmatrix's `functional_status`.
//!
//! * `saga` ON → [`record`] stamps the event with a monotonic `seq` + wall-clock
//! `ts_micros` and pushes it into a process-global buffer; [`drain_saga_events`]
//! takes + clears them (a [`crate::SagaSink`] writes them to the nornir warehouse).
//! * `saga` OFF → [`record`] is an `#[inline]` empty stub and
//! [`drain_saga_events`] is hard-wired empty — capture compiles to nothing.
use crate::event::{SagaEvent, SessionCtx};
/// The maximum number of events accepted from a single [`ingest`] batch.
///
/// [`ingest`] is the entry point for events captured on an **untrusted client**
/// (a browser WASM bundle POSTing its captured `nav`/`click` journey to a server
/// ingest endpoint) — unlike [`record`], whose events are built by trusted
/// in-process capture. This cap bounds how much an untrusted caller can push into
/// the process-global buffer in one call. Events beyond the cap are **not**
/// recorded; [`ingest`] returns the count actually taken so the caller can detect
/// and log a truncated batch rather than silently drop the overflow.
pub const MAX_INGEST_BATCH: usize = 1024;
// ─── feature ON: stamp + record into a process-global buffer ─────────────────
#[cfg(feature = "saga")]
mod recorder {
use crate::event::SagaEvent;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock};
use std::time::{SystemTime, UNIX_EPOCH};
fn buffer() -> &'static Mutex<Vec<SagaEvent>> {
static BUF: OnceLock<Mutex<Vec<SagaEvent>>> = OnceLock::new();
BUF.get_or_init(|| Mutex::new(Vec::new()))
}
fn next_seq() -> u64 {
static SEQ: AtomicU64 = AtomicU64::new(0);
SEQ.fetch_add(1, Ordering::Relaxed)
}
fn now_micros() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_micros() as u64)
.unwrap_or(0)
}
/// Stamp identity (`seq`, `ts_micros`), **PII-redact by default**, and buffer
/// the event. Lock poisoning is recovered from — a panic mid-record must never
/// wedge an app's capture. Masking here means a stored journey never carries a
/// raw user email or secret token; a host that needs a field raw must build the
/// masked view itself (the record path is mask-by-default on purpose).
pub fn record(event: SagaEvent) {
let mut event = event.redacted();
event.seq = next_seq();
if event.ts_micros == 0 {
event.ts_micros = now_micros();
}
let mut buf = buffer().lock().unwrap_or_else(|p| p.into_inner());
buf.push(event);
}
/// Re-stamp each event in an untrusted client batch with `ctx`'s
/// session/app identity and [`record`] it (which PII-redacts by default),
/// capping the batch at [`super::MAX_INGEST_BATCH`]. Returns the number
/// recorded so a caller can detect a truncated over-cap batch. See
/// [`super::ingest`].
pub fn ingest(events: Vec<SagaEvent>, ctx: &super::SessionCtx) -> usize {
let mut n = 0;
for ev in events.into_iter().take(super::MAX_INGEST_BATCH) {
record(ev.in_session(ctx));
n += 1;
}
n
}
/// Take + clear every captured event (feature ON).
pub fn drain() -> Vec<SagaEvent> {
let mut buf = buffer().lock().unwrap_or_else(|p| p.into_inner());
std::mem::take(&mut *buf)
}
/// Durably flush the buffer to `sink`, clearing it **only after** the write
/// succeeds. Unlike `drain()`-then-`write()` (which takes the events out of
/// the buffer before the write and so LOSES them if the write fails), this
/// holds the buffer across the write and retains it on error, so a failed
/// flush can be safely retried. Returns the number of events written.
pub fn flush(sink: &dyn crate::sink::SagaSink) -> anyhow::Result<usize> {
let mut buf = buffer().lock().unwrap_or_else(|p| p.into_inner());
if buf.is_empty() {
return Ok(0);
}
sink.write(&buf)?; // write FIRST, still holding the buffer
let n = buf.len();
buf.clear(); // clear ONLY once the batch is durable
Ok(n)
}
}
/// Capture one [`SagaEvent`] for the matrix to replay later.
///
/// With the `saga` feature ON the event is stamped (`seq` + `ts_micros`) and
/// pushed into a process-global buffer that [`drain_saga_events`] hands to a
/// [`crate::SagaSink`]. With the feature OFF this is an `#[inline]` no-op — the
/// event is dropped and nothing is compiled or written. A capture adapter
/// (egui / web — deferred) builds the event with [`SagaEvent::in_session`] and
/// calls this.
#[cfg(feature = "saga")]
pub fn record(event: SagaEvent) {
recorder::record(event);
}
/// No-op stub when the `saga` feature is OFF. Takes the event by value and drops
/// it; inlined so the call (and any event-building wrapped at the call site) can
/// be stripped.
#[cfg(not(feature = "saga"))]
#[inline]
pub fn record(_event: SagaEvent) {}
/// Ingest a batch of events captured on an **untrusted client** (e.g. a browser
/// WASM bundle POSTing its captured journey to a server `#[server]`/HTTP ingest
/// endpoint) into the process-global capture buffer.
///
/// This is the reusable, toolkit-neutral half of the client→server transport a
/// web consumer needs: the app supplies the `#[server]`/Axum wire (which is
/// framework-specific and stays app-side), and hands the decoded `Vec<SagaEvent>`
/// plus the server-derived [`SessionCtx`] here. Each event is re-stamped with the
/// real session/app identity via [`SagaEvent::in_session`] and [`record`]ed — so
/// it flows through the same PII-redaction and the same flush → warehouse pipe as
/// in-process capture, and a client can never forge another user's identity.
///
/// The batch is capped at [`MAX_INGEST_BATCH`] to bound the memory an untrusted
/// caller can push in one call; the return value is the number of events actually
/// recorded, so a caller can detect and log a truncated over-cap batch instead of
/// silently dropping the overflow. With the `saga` feature OFF this is an
/// `#[inline]` no-op that records nothing and returns `0`.
#[cfg(feature = "saga")]
pub fn ingest(events: Vec<SagaEvent>, ctx: &SessionCtx) -> usize {
recorder::ingest(events, ctx)
}
/// No-op ingest when the `saga` feature is OFF: records nothing, returns `0`.
#[cfg(not(feature = "saga"))]
#[inline]
pub fn ingest(_events: Vec<SagaEvent>, _ctx: &SessionCtx) -> usize {
0
}
/// Drain (take + clear) the captured events. A [`crate::SagaSink`] driver calls
/// this on a cadence and writes the batch to the nornir warehouse. With the
/// `saga` feature OFF this is an `#[inline]` empty `Vec` — no global compiled in.
#[cfg(feature = "saga")]
pub fn drain_saga_events() -> Vec<SagaEvent> {
recorder::drain()
}
/// No-op drain when the `saga` feature is OFF: always empty.
#[cfg(not(feature = "saga"))]
#[inline]
pub fn drain_saga_events() -> Vec<SagaEvent> {
Vec::new()
}
/// Durably flush the captured events to `sink`, clearing the buffer **only after**
/// the write succeeds. This is the crash-safe alternative to
/// [`drain_saga_events`]`+`[`SagaSink::write`](crate::SagaSink::write): a drain
/// removes the events from the buffer *before* the write, so a write failure
/// drops them on the floor; `flush_saga_events` retains them on failure so the
/// flush can be retried and no captured journey is lost to a transient sink error.
/// Returns the number of events written. `saga` OFF → a no-op that writes nothing.
#[cfg(feature = "saga")]
pub fn flush_saga_events(sink: &dyn crate::sink::SagaSink) -> anyhow::Result<usize> {
recorder::flush(sink)
}
/// No-op flush when the `saga` feature is OFF: nothing captured, nothing written.
#[cfg(not(feature = "saga"))]
#[inline]
pub fn flush_saga_events(_sink: &dyn crate::sink::SagaSink) -> anyhow::Result<usize> {
Ok(0)
}
/// Test-only serialization lock for the process-global capture buffer, so an
/// interleaving drain in another test never steals a test's events.
#[cfg(all(test, feature = "saga"))]
pub(crate) fn test_lock() -> std::sync::MutexGuard<'static, ()> {
use std::sync::{Mutex, OnceLock};
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| Mutex::new(()))
.lock()
.unwrap_or_else(|p| p.into_inner())
}
#[cfg(test)]
mod tests {
use super::*;
// ── feature ON: capture → drain preserves order, stamps seq, then clears ──
#[cfg(feature = "saga")]
#[test]
fn record_then_drain_preserves_order_and_clears() {
let _guard = test_lock();
let _ = drain_saga_events(); // start clean
record(SagaEvent::nav("/a", "A", "open", "{}"));
record(SagaEvent::click("/a", "Btn", "submit", "{}"));
record(SagaEvent::nav("/b", "B", "open", "{}"));
let events = drain_saga_events();
assert_eq!(events.len(), 3, "all three captured");
// capture order is preserved and seq is strictly increasing.
assert_eq!(events[0].route, "/a");
assert_eq!(events[1].component, "Btn");
assert_eq!(events[2].route, "/b");
assert!(
events[0].seq < events[1].seq && events[1].seq < events[2].seq,
"seq strictly increasing in capture order"
);
assert!(events[0].ts_micros > 0, "recorder stamps a wall-clock ts");
// Drained → buffer cleared.
assert!(
drain_saga_events().is_empty(),
"drain clears the buffer"
);
}
// ── ingest: untrusted client batch is re-stamped, redacted, and capped ──
#[cfg(feature = "saga")]
#[test]
fn ingest_restamps_identity_redacts_client_pii_and_caps_batch() {
use crate::event::SessionCtx;
let _guard = test_lock();
let _ = drain_saga_events(); // start clean
// The server derives the identity; a browser client can only supply the
// journey (blank identity + possibly PII in state_json).
let ctx = SessionCtx::new("s1", "u1", "njord", "9.9", "deadbeef");
let from_client = vec![
SagaEvent::nav("/a", "A", "open", r#"{"email":"bob@example.com"}"#),
SagaEvent::click("/a", "Btn", "submit", "{}"),
];
let n = ingest(from_client, &ctx);
assert_eq!(n, 2, "both client events recorded");
let got = drain_saga_events();
assert_eq!(got.len(), 2);
// Server-side identity is stamped — a client cannot forge session/user/app.
assert_eq!(got[0].session_id, "s1");
assert_eq!(got[0].user, "u1");
assert_eq!(got[0].app, "njord");
// Client-supplied PII is redacted on the ingest path (same as record()).
assert!(
got[0].state_json.contains(crate::event::redact::MASK),
"ingest redacts client PII by default, got {:?}",
got[0].state_json
);
// An over-cap batch records only MAX_INGEST_BATCH and REPORTS the count
// (no silent drop of the overflow).
let big: Vec<SagaEvent> = (0..MAX_INGEST_BATCH + 10)
.map(|_| SagaEvent::nav("/x", "X", "open", "{}"))
.collect();
let taken = ingest(big, &ctx);
assert_eq!(taken, MAX_INGEST_BATCH, "untrusted over-cap batch is bounded");
assert_eq!(
drain_saga_events().len(),
MAX_INGEST_BATCH,
"exactly the capped count reached the buffer"
);
}
// ── flush is crash-safe: a failed write RETAINS the batch (no loss) ──
#[cfg(feature = "saga")]
#[test]
fn flush_retains_events_when_the_sink_fails_then_clears_on_success() {
use crate::sink::{JsonFileSink, SagaSink};
// A sink that always fails — models a transient warehouse/disk error.
struct FailSink;
impl SagaSink for FailSink {
fn write(&self, _e: &[crate::event::SagaEvent]) -> anyhow::Result<()> {
Err(anyhow::anyhow!("sink down"))
}
}
let _guard = test_lock();
let _ = drain_saga_events(); // start clean
record(SagaEvent::nav("/a", "A", "open", "{}"));
record(SagaEvent::click("/a", "Btn", "submit", "{}"));
// A failed flush must NOT lose the captured events.
assert!(flush_saga_events(&FailSink).is_err(), "sink error surfaces");
// A succeeding flush writes them and only THEN clears the buffer.
let dir = std::env::temp_dir().join(format!("saga-flush-{}", std::process::id()));
let file = dir.join("f.jsonl");
let _ = std::fs::remove_dir_all(&dir);
let ok = JsonFileSink::new(&file);
let n = flush_saga_events(&ok).expect("flush ok");
assert_eq!(n, 2, "the events survived the earlier failure and were written");
assert_eq!(ok.read_all().unwrap().len(), 2, "both persisted");
// Buffer cleared only after the durable write.
assert!(drain_saga_events().is_empty(), "flush cleared the buffer on success");
let _ = std::fs::remove_dir_all(&dir);
}
// ── feature OFF: prove the release/disabled path captures NOTHING ──
#[cfg(not(feature = "saga"))]
#[test]
fn disabled_path_is_a_noop_drain_is_empty() {
record(SagaEvent::nav("/a", "A", "open", "{}"));
record(SagaEvent::error("A", "would be captured"));
assert!(
drain_saga_events().is_empty(),
"saga OFF captures + drains NOTHING — zero cost"
);
}
}