crates/nornir-saga/src/sink.rs
//! Where a session's captured events go. The [`SagaSink`] trait is the durable
//! seam — exactly like `nornir-testmatrix`'s `TestSink`:
//!
//! * **nornir** implements it over its Iceberg warehouse `saga_events` table
//! (`nornir::warehouse::saga_events::IcebergSagaSink`); saga itself never names
//! a backend (the crate stays serde-only — no iceberg/arrow).
//! * a leaf app with no warehouse uses the built-in [`JsonFileSink`] (append-only
//! JSONL history) or [`NullSink`] (drop, for dry-runs / tests).
//!
//! A capture driver calls [`crate::drain_saga_events`] on a cadence and hands the
//! batch to a `SagaSink::write`.
use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use crate::event::SagaEvent;
/// A destination for a batch of captured [`SagaEvent`]s. One `write` per drain
/// (the batch shares no run id — events carry their own `session_id` + `seq`).
///
/// Object-safe so a caller can hold a `Box<dyn SagaSink>`. Implementations
/// should treat an empty slice as a no-op success.
pub trait SagaSink {
/// Persist `events`. An empty slice is a no-op success.
fn write(&self, events: &[SagaEvent]) -> Result<()>;
}
/// Drops everything — for dry-runs and tests. Always `Ok`.
#[derive(Debug, Default, Clone, Copy)]
pub struct NullSink;
impl SagaSink for NullSink {
fn write(&self, _events: &[SagaEvent]) -> Result<()> {
Ok(())
}
}
/// Appends each drained batch to a JSON-lines history file (one JSON *array* per
/// drain, one line per drain) so a leaf app keeps a portable session history
/// without a warehouse. Re-readable (flattened) with [`JsonFileSink::read_all`].
#[derive(Debug, Clone)]
pub struct JsonFileSink {
path: PathBuf,
}
impl JsonFileSink {
pub fn new(path: impl Into<PathBuf>) -> Self {
Self { path: path.into() }
}
pub fn path(&self) -> &Path {
&self.path
}
/// Read every event ever written (flattening all drain-lines into one Vec),
/// re-sorted by capture `seq` so a reader gets them in journey order.
pub fn read_all(&self) -> Result<Vec<SagaEvent>> {
let text = match fs::read_to_string(&self.path) {
Ok(t) => t,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(e) => return Err(e).with_context(|| format!("read {}", self.path.display())),
};
let mut out = Vec::new();
// Collect the non-empty drain-lines up front so we can tell a torn TRAILING
// line (a `write` interrupted by a crash mid-append) apart from real
// corruption in the middle of the file.
let lines: Vec<&str> = text.lines().map(str::trim).filter(|l| !l.is_empty()).collect();
let n = lines.len();
for (i, line) in lines.iter().enumerate() {
match serde_json::from_str::<Vec<SagaEvent>>(line) {
Ok(batch) => out.extend(batch),
// A partial JSON tail on the LAST line is an append cut short by a
// crash: the sessions written before it are intact, so drop the
// torn batch rather than failing the whole read (append-only
// durability — see `write`). A corrupt NON-terminal line is real
// damage → surface it.
Err(_) if i + 1 == n => break,
Err(e) => {
return Err(e).with_context(|| {
format!("parse drain line {} in {}", i + 1, self.path.display())
})
}
}
}
out.sort_by_key(|e| e.seq);
Ok(out)
}
}
impl SagaSink for JsonFileSink {
fn write(&self, events: &[SagaEvent]) -> Result<()> {
if events.is_empty() {
return Ok(());
}
if let Some(parent) = self.path.parent() {
if !parent.as_os_str().is_empty() {
fs::create_dir_all(parent)
.with_context(|| format!("create {}", parent.display()))?;
}
}
// One compact JSON array per drain, APPENDED as its own line (JSONL of
// arrays). Append-only (`O_APPEND`) is the durability contract: a drain
// never rewrites the accumulated history, so a crash mid-write can at
// worst leave a torn final line (which `read_all` tolerates) — it can
// never truncate or lose the sessions already on disk. The previous
// read-whole-file / rewrite-whole-file approach risked losing the ENTIRE
// file to a crash and was O(n²) over the history; a single append is O(1)
// and each write() lands atomically at end-of-file even across processes.
let mut line = serde_json::to_string(events)?;
line.push('\n');
let mut f = fs::OpenOptions::new()
.create(true)
.append(true)
.open(&self.path)
.with_context(|| format!("open {}", self.path.display()))?;
f.write_all(line.as_bytes())
.with_context(|| format!("append {}", self.path.display()))?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::event::SessionCtx;
fn batch(session: &str, base_seq: u64) -> Vec<SagaEvent> {
let ctx = SessionCtx::new(session, "alice", "njord", "0.9.0", "abc123");
let mut a = SagaEvent::nav("/home", "Dashboard", "open", "{}").in_session(&ctx);
let mut b = SagaEvent::click("/home", "Btn", "submit", r#"{"ok":true}"#).in_session(&ctx);
a.seq = base_seq;
a.ts_micros = 100;
b.seq = base_seq + 1;
b.ts_micros = 100;
vec![a, b]
}
#[test]
fn null_sink_swallows_events() {
assert!(NullSink.write(&batch("s", 0)).is_ok());
assert!(NullSink.write(&[]).is_ok());
}
#[test]
fn json_file_sink_round_trips_two_drains() {
let dir = std::env::temp_dir().join(format!("saga-sink-{}", std::process::id()));
let file = dir.join("sessions.jsonl");
let _ = fs::remove_file(&file);
let sink = JsonFileSink::new(&file);
// Empty write is a no-op (file not created).
sink.write(&[]).unwrap();
assert!(sink.read_all().unwrap().is_empty(), "no events yet");
sink.write(&batch("sA", 0)).unwrap();
sink.write(&batch("sB", 10)).unwrap();
let back = sink.read_all().unwrap();
assert_eq!(back.len(), 4, "two drains × two events = 4 read back");
// Re-sorted by seq, both sessions present, payload exact.
assert!(back.windows(2).all(|w| w[0].seq <= w[1].seq), "sorted by seq");
assert!(back.iter().any(|e| e.session_id == "sA"));
assert!(back.iter().any(|e| e.session_id == "sB"));
let click = back.iter().find(|e| e.component == "Btn").unwrap();
assert_eq!(click.state_json, r#"{"ok":true}"#);
let _ = fs::remove_dir_all(&dir);
}
#[test]
fn write_appends_and_never_rewrites_prior_history() {
// Durability regression guard: a second drain must APPEND, leaving the
// bytes of the first drain-line byte-for-byte untouched (the old
// read-modify-write rewrote the whole file every time — a crash window
// that could lose ALL prior sessions).
let dir = std::env::temp_dir().join(format!("saga-append-{}", std::process::id()));
let file = dir.join("h.jsonl");
let _ = fs::remove_dir_all(&dir);
let sink = JsonFileSink::new(&file);
sink.write(&batch("sA", 0)).unwrap();
let after_first = fs::read(&file).unwrap();
sink.write(&batch("sB", 10)).unwrap();
let after_second = fs::read(&file).unwrap();
// The first line's bytes are a strict PREFIX of the file after the second
// write — proof the append did not touch them.
assert!(
after_second.starts_with(&after_first),
"second drain appended; the first drain-line was not rewritten"
);
assert_eq!(sink.read_all().unwrap().len(), 4, "both drains read back");
let _ = fs::remove_dir_all(&dir);
}
#[test]
fn read_all_tolerates_a_crash_torn_trailing_line() {
// Simulate a crash mid-append: a valid drain-line, then a truncated
// (torn) JSON tail with no newline. read_all must still return the intact
// sessions and simply drop the torn tail — never fail the whole read.
let dir = std::env::temp_dir().join(format!("saga-torn-{}", std::process::id()));
let file = dir.join("h.jsonl");
let _ = fs::remove_dir_all(&dir);
let sink = JsonFileSink::new(&file);
sink.write(&batch("good", 0)).unwrap();
// Append a partial line as a crashing writer would have left it.
{
let mut f = fs::OpenOptions::new().append(true).open(&file).unwrap();
f.write_all(br#"[{"seq":99,"ts_micros":1,"session_id":"torn""#).unwrap();
}
let back = sink.read_all().unwrap();
assert_eq!(back.len(), 2, "the intact first drain survives the torn tail");
assert!(back.iter().all(|e| e.session_id == "good"));
let _ = fs::remove_dir_all(&dir);
}
#[test]
fn read_all_still_errors_on_corruption_before_the_last_line() {
// A garbled line that is NOT the trailing one is real corruption, not a
// torn append — it must surface as an error, not be silently skipped.
let dir = std::env::temp_dir().join(format!("saga-corrupt-{}", std::process::id()));
let file = dir.join("h.jsonl");
let _ = fs::remove_dir_all(&dir);
fs::create_dir_all(&dir).unwrap();
// A bad first line, then a valid trailing line.
let good = serde_json::to_string(&batch("ok", 0)).unwrap();
fs::write(&file, format!("not json at all\n{good}\n")).unwrap();
let sink = JsonFileSink::new(&file);
assert!(sink.read_all().is_err(), "mid-file corruption is surfaced");
let _ = fs::remove_dir_all(&dir);
}
#[test]
fn sink_is_object_safe() {
// Hold it behind a trait object — the leaf-app seam.
let boxed: Box<dyn SagaSink> = Box::new(NullSink);
assert!(boxed.write(&batch("s", 0)).is_ok());
}
}