crates/nornir-saga/src/store.rs
//! **`SagaTapeStore`** — the warehouse-agnostic seam a recorded tape *lands in*
//! and is *read back from*.
//!
//! [`crate::sink::SagaSink`] is the write-only drain seam (a capture buffer →
//! `write`). A **tape recorder** needs the other half too: it must LIST the
//! recorded sessions (painful-first, the Sessions rail) and READ ONE session's
//! events back to drive the transport. [`SagaTapeStore`] is that fuller seam —
//! `append_events` / `list_sessions` / `read_session` — so a HOST can land tapes
//! into (and read them out of) *any* backend:
//!
//! * **file-backed** (the built-in [`FileTapeStore`]) — a leaf app with no
//! warehouse keeps a portable `.jsonl` tape; reuses [`JsonFileSink`] verbatim.
//! * **live Mimir** — nornir implements it over its Iceberg `saga_events` table
//! (`append_saga_events` / `list_saga_sessions` / `query_saga_events`), so the
//! tape-recorder uses the REAL warehouse instead of a flat file.
//! * **over gRPC** — dwarves (which cannot depend on nornir — that would cycle)
//! reads a nornir-hosted store over the wire, implementing this same trait.
//!
//! The rows [`list_sessions`](SagaTapeStore::list_sessions) returns are the exact
//! [`SagaSessionSummary`] roll-up ([`summarize_sessions`], painful-first) that
//! both the nornir viz Sessions pane and the dwarves tape-recorder pane render —
//! one home, consumed by both (REUSE).
use anyhow::Result;
use crate::analysis::{summarize_sessions, SagaSessionSummary};
use crate::event::SagaEvent;
use crate::sink::JsonFileSink;
/// Where recorded [`SagaEvent`] tapes land and are read back from — the seam that
/// lets the tape-recorder use the real Mimir warehouse instead of a flat file.
///
/// Object-safe (`Box<dyn SagaTapeStore>`) so a host swaps a file-backed store for
/// a warehouse-backed one without touching the pane. An empty batch is a no-op
/// success (same contract as [`crate::sink::SagaSink`]).
pub trait SagaTapeStore {
/// Append a drained batch of events to the store (one logical drain). Each
/// event carries its own `session_id` + `seq`, so a batch is whatever the
/// recorder drained. An empty slice is a no-op success.
fn append_events(&self, events: &[SagaEvent]) -> Result<()>;
/// List the recorded sessions, rolled up one [`SagaSessionSummary`] per
/// session, ordered **painful-first** (see [`summarize_sessions`]).
fn list_sessions(&self) -> Result<Vec<SagaSessionSummary>>;
/// Read back every event of `session_id`, in capture (`seq`) order — the
/// stream the tape transport folds. An unknown session yields an empty Vec.
fn read_session(&self, session_id: &str) -> Result<Vec<SagaEvent>>;
}
/// The built-in **file-backed** [`SagaTapeStore`]: a portable `.jsonl` tape for a
/// leaf app with no warehouse. A thin wrapper over [`JsonFileSink`] (REUSE — the
/// append-only durability + torn-tail tolerance live there); this adds the
/// read-back half (`list_sessions` / `read_session`).
#[derive(Debug, Clone)]
pub struct FileTapeStore {
sink: JsonFileSink,
}
impl FileTapeStore {
/// A file-backed store writing/reading the `.jsonl` tape at `path`.
pub fn new(path: impl Into<std::path::PathBuf>) -> Self {
Self { sink: JsonFileSink::new(path) }
}
/// Wrap an existing [`JsonFileSink`] (e.g. the one a capture drain already
/// writes to) so the recorder and the reader share one tape file.
pub fn from_sink(sink: JsonFileSink) -> Self {
Self { sink }
}
/// The underlying sink (its `.jsonl` path, for a `SagaSink`-typed drain).
pub fn sink(&self) -> &JsonFileSink {
&self.sink
}
}
impl SagaTapeStore for FileTapeStore {
fn append_events(&self, events: &[SagaEvent]) -> Result<()> {
use crate::sink::SagaSink;
self.sink.write(events)
}
fn list_sessions(&self) -> Result<Vec<SagaSessionSummary>> {
Ok(summarize_sessions(&self.sink.read_all()?))
}
fn read_session(&self, session_id: &str) -> Result<Vec<SagaEvent>> {
let mut evs: Vec<SagaEvent> = self
.sink
.read_all()?
.into_iter()
.filter(|e| e.session_id == session_id)
.collect();
// Defensive: `read_all` already sorts by seq, but scope the guarantee here
// too so a store's read_session contract is self-standing.
evs.sort_by_key(|e| e.seq);
Ok(evs)
}
}
/// An **in-memory** [`SagaTapeStore`]: holds the appended events in a `Vec` behind
/// a `Mutex`. The single-home fixture/test store the constellation reuses — a host
/// (or robot test) seeds sessions with [`append_events`](SagaTapeStore::append_events),
/// then lists/reads them back through the SAME trait a file- or warehouse-backed
/// store exposes, so a pane driven over `Box<dyn SagaTapeStore>` is exercised with
/// no file or warehouse. Cheap to clone the underlying events out; never persists.
#[derive(Default)]
pub struct MemTapeStore {
events: std::sync::Mutex<Vec<SagaEvent>>,
}
impl MemTapeStore {
/// An empty in-memory store (seed it with `append_events`).
pub fn new() -> Self {
Self::default()
}
/// An in-memory store pre-seeded with `events` (a fixture tape a host renders
/// through the trait — e.g. a never-blank sample).
pub fn from_events(events: impl IntoIterator<Item = SagaEvent>) -> Self {
Self { events: std::sync::Mutex::new(events.into_iter().collect()) }
}
}
impl SagaTapeStore for MemTapeStore {
fn append_events(&self, events: &[SagaEvent]) -> Result<()> {
if events.is_empty() {
return Ok(());
}
self.events.lock().expect("saga MemTapeStore mutex").extend_from_slice(events);
Ok(())
}
fn list_sessions(&self) -> Result<Vec<SagaSessionSummary>> {
Ok(summarize_sessions(&self.events.lock().expect("saga MemTapeStore mutex")))
}
fn read_session(&self, session_id: &str) -> Result<Vec<SagaEvent>> {
let mut evs: Vec<SagaEvent> = self
.events
.lock()
.expect("saga MemTapeStore mutex")
.iter()
.filter(|e| e.session_id == session_id)
.cloned()
.collect();
evs.sort_by_key(|e| e.seq);
Ok(evs)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::event::SessionCtx;
fn ev(session: &str, seq: u64, kind_click: bool) -> SagaEvent {
let ctx = SessionCtx::new(session, "alice", "njord", "0.9.0", "abc123");
let mut e = if kind_click {
SagaEvent::click("/kund", "OffertForm", "submit", r#"{"ok":true}"#)
} else {
SagaEvent::nav("/kund", "KundList", "open", r#"{"route":"/kund"}"#)
}
.in_session(&ctx);
e.seq = seq;
e.ts_micros = 1_000 + seq;
e
}
#[test]
fn file_store_appends_lists_and_reads_back_one_session() {
let dir = std::env::temp_dir().join(format!("saga-store-{}", std::process::id()));
let file = dir.join("tape.jsonl");
let _ = std::fs::remove_dir_all(&dir);
let store = FileTapeStore::new(&file);
// Empty store: no sessions, empty read.
assert!(store.list_sessions().unwrap().is_empty(), "empty tape → no sessions");
assert!(store.read_session("nope").unwrap().is_empty());
// Two drains, two sessions.
store.append_events(&[ev("sA", 0, false), ev("sA", 1, true)]).unwrap();
store.append_events(&[ev("sB", 10, false)]).unwrap();
let sessions = store.list_sessions().unwrap();
assert_eq!(sessions.len(), 2, "two distinct sessions rolled up");
assert!(sessions.iter().any(|s| s.session_id == "sA" && s.step_count == 2));
assert!(sessions.iter().any(|s| s.session_id == "sB" && s.step_count == 1));
// Read one session back, in seq order, scoped to that session only.
let sa = store.read_session("sA").unwrap();
assert_eq!(sa.len(), 2, "only sA's events");
assert!(sa.iter().all(|e| e.session_id == "sA"));
assert!(sa.windows(2).all(|w| w[0].seq <= w[1].seq), "sorted by seq");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn empty_append_is_a_noop() {
let dir = std::env::temp_dir().join(format!("saga-store-empty-{}", std::process::id()));
let file = dir.join("tape.jsonl");
let _ = std::fs::remove_dir_all(&dir);
let store = FileTapeStore::new(&file);
store.append_events(&[]).unwrap();
assert!(store.list_sessions().unwrap().is_empty(), "empty append writes nothing");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn mem_store_seeds_lists_painful_first_and_reads_back() {
// Seed a stuck session (2 errors) + a calm one; list must float painful first.
let store = MemTapeStore::new();
let mut stuck_err = SagaEvent::error("OffertForm", "boom");
stuck_err.route = "/offert/7".into();
let ctx = SessionCtx::new("stuck", "alice", "njord", "0.9.0", "abc");
stuck_err = stuck_err.in_session(&ctx);
stuck_err.seq = 2;
store.append_events(&[ev("stuck", 0, false), ev("stuck", 1, true), stuck_err]).unwrap();
store.append_events(&[ev("calm", 0, false)]).unwrap();
let sessions = store.list_sessions().unwrap();
assert_eq!(sessions.len(), 2, "two sessions rolled up");
assert_eq!(sessions[0].session_id, "stuck", "painful-first: the erroring session floats up");
assert!(sessions[0].frustration_score > 0);
// Read one session back through the trait, scoped + seq-ordered.
let read = store.read_session("stuck").unwrap();
assert_eq!(read.len(), 3);
assert!(read.iter().all(|e| e.session_id == "stuck"));
assert!(read.windows(2).all(|w| w[0].seq <= w[1].seq));
assert!(store.read_session("nope").unwrap().is_empty());
}
#[test]
fn mem_store_from_events_and_object_safe() {
let seed = vec![ev("s", 0, false), ev("s", 1, true)];
let store: Box<dyn SagaTapeStore> = Box::new(MemTapeStore::from_events(seed));
assert_eq!(store.list_sessions().unwrap().len(), 1);
assert_eq!(store.read_session("s").unwrap().len(), 2);
store.append_events(&[]).unwrap(); // empty append is a no-op
assert_eq!(store.read_session("s").unwrap().len(), 2);
}
#[test]
fn store_is_object_safe() {
// Hold it behind a trait object — the host swap seam (file ↔ warehouse).
let dir = std::env::temp_dir().join(format!("saga-store-obj-{}", std::process::id()));
let file = dir.join("tape.jsonl");
let _ = std::fs::remove_dir_all(&dir);
let store: Box<dyn SagaTapeStore> = Box::new(FileTapeStore::new(&file));
store.append_events(&[ev("s", 0, false)]).unwrap();
assert_eq!(store.list_sessions().unwrap().len(), 1);
assert_eq!(store.read_session("s").unwrap().len(), 1);
let _ = std::fs::remove_dir_all(&dir);
}
}