src/warehouse/saga_events.rs
//! Iceberg writer + reader for **saga record/replay** (`saga_events`, EPIC SAGA
//! / task #25) — the production sibling of [`super::test_results`].
//!
//! The PURE model ([`SagaEvent`], [`EventKind`], [`ReplayPlan`], [`ReplayStep`],
//! the [`SagaSink`] trait) lives in the lean [`nornir_saga`] crate and is
//! **re-exported here**; this module keeps only the Iceberg-coupled write/read
//! and the [`IcebergSagaSink`] implementation over the warehouse.
//!
//! Write path: [`append_saga_events`] appends one Iceberg snapshot per drain.
//! Each event carries its own `session_id` + monotonic `seq`, so a batch is just
//! whatever the recorder drained.
//!
//! Read path: [`query_saga_events`] scans the table, scopes by `session_id` (or
//! all), and returns events sorted by `(session_id, seq)`. [`list_saga_sessions`]
//! rolls them up into one [`SagaSessionSummary`] per session (the viz Sessions
//! pane's row), and [`session_replay_plan`] hands back the [`ReplayPlan`] for one
//! session.
//!
//! Schema: [`super::iceberg_schema::saga_events`] (17 columns).
use std::collections::BTreeMap;
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{Array, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use sha2::{Digest, Sha256};
use super::iceberg::{IcebergWarehouse, TABLE_SAGA_EVENTS, append_batch};
// ─── re-export the pure model from the lean crate ──────────────────────────
pub use nornir_saga::analysis::{NavFunnel, SessionFrustration, frustration_by_session};
pub use nornir_saga::event::{EventKind, SagaEvent, SessionCtx};
pub use nornir_saga::replay::{ReplayPlan, ReplayStep, StepKind};
pub use nornir_saga::sink::SagaSink;
// The Sessions-pane roll-up + its summariser are the canonical ones from the lean
// crate (`analysis`); re-exported here (move=delete — the byte-identical local
// copy that used to live in this file was removed). The tape-store seam is
// re-exported too so hosts consume one home (REUSE).
pub use nornir_saga::store::{FileTapeStore, SagaTapeStore};
pub use nornir_saga::{SagaSessionSummary, summarize_sessions};
// The ergonomic producer emit-API (edda's lean crate): an app builds a recorder
// over a tape store (here the warehouse-backed `WarehouseTapeStore`) and records
// its own journey. Re-exported so a nornir-hosted producer + the saga tests
// consume the one shared client (REUSE — no twin recorder).
pub use nornir_saga::SagaRecorder;
// The process-global capture buffer the viz nav/click choke feeds (feature
// `saga`; no-ops otherwise). Re-exported so integration tests (which don't dep
// `nornir-saga` directly) can drain what the viz recorded.
pub use nornir_saga::{drain_saga_events, record};
use super::Warehouse;
// Column indices match `iceberg_schema::saga_events` field order.
const COL_SESSION_ID: usize = 0;
const COL_SEQ: usize = 1;
const COL_TS_MICROS: usize = 2;
const COL_KIND: usize = 3;
const COL_ROUTE: usize = 4;
const COL_COMPONENT: usize = 5;
const COL_ACTION: usize = 6;
const COL_DETAIL: usize = 7;
const COL_STATE_JSON: usize = 8;
const COL_LEVEL: usize = 9;
const COL_SPAN_ID: usize = 10;
const COL_PARENT: usize = 11;
const COL_APP_USER: usize = 12;
const COL_APP: usize = 13;
const COL_APP_VERSION: usize = 14;
const COL_GIT_SHA: usize = 15;
const COL_SHA: usize = 16;
/// The sha256 of an event's canonical payload (the replay-relevant fields) — the
/// `sha` column, used for integrity + dedup. Stable across runs of the same
/// recorded step.
pub fn event_sha(e: &SagaEvent) -> String {
let mut h = Sha256::new();
h.update(e.session_id.as_bytes());
h.update([0]);
h.update(e.seq.to_le_bytes());
h.update(e.kind.as_str().as_bytes());
h.update([0]);
h.update(e.route.as_bytes());
h.update([0]);
h.update(e.component.as_bytes());
h.update([0]);
h.update(e.action.as_bytes());
h.update([0]);
h.update(e.state_json.as_bytes());
h.finalize().iter().map(|b| format!("{b:02x}")).collect::<String>()
}
/// Append a batch of captured `saga_events`. An empty slice is a no-op success.
///
/// The table is partitioned by `session_id`, and skade appends one partition at
/// a time, so a process-wide drain that mixes sessions is split into one
/// per-session snapshot (each a single partition). Within a session the order is
/// preserved by the `seq` column the recorder stamps.
pub async fn append_saga_events(wh: &IcebergWarehouse, events: &[SagaEvent]) -> Result<()> {
if events.is_empty() {
return Ok(());
}
// Group by session so each append is a single `session_id` partition.
let mut by_session: BTreeMap<&str, Vec<&SagaEvent>> = BTreeMap::new();
for e in events {
by_session.entry(e.session_id.as_str()).or_default().push(e);
}
for (_session, batch) in by_session {
append_one_session(wh, &batch).await?;
}
Ok(())
}
/// Append one session's events (all sharing one `session_id` → one partition).
async fn append_one_session(wh: &IcebergWarehouse, events: &[&SagaEvent]) -> Result<()> {
if events.is_empty() {
return Ok(());
}
let ident = wh.table_ident(TABLE_SAGA_EVENTS);
let table = wh.catalog().load_table(&ident).await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let batch = RecordBatch::try_new(arrow_schema, saga_cols(events))?;
append_batch(wh.catalog(), table, batch).await?;
Ok(())
}
/// The 17 `saga_events` Arrow columns for a slice of events (column order matches
/// `iceberg_schema::saga_events`). Shared by the inherent [`append_one_session`]
/// (Iceberg, table schema from the loaded table) and the [`WarehouseTapeStore`]
/// (`dyn Warehouse`, canonical schema) — one encoder, no twin (REUSE).
fn saga_cols(events: &[&SagaEvent]) -> Vec<Arc<dyn Array>> {
let s = |f: fn(&SagaEvent) -> String| -> Arc<dyn Array> {
Arc::new(StringArray::from(
events.iter().map(|e| f(e)).collect::<Vec<_>>(),
))
};
vec![
s(|e| e.session_id.clone()),
Arc::new(Int64Array::from(
events.iter().map(|e| e.seq as i64).collect::<Vec<_>>(),
)),
Arc::new(
TimestampMicrosecondArray::from(
events
.iter()
.map(|e| e.ts_micros as i64)
.collect::<Vec<_>>(),
)
.with_timezone("+00:00"),
),
s(|e| e.kind.as_str().to_string()),
s(|e| e.route.clone()),
s(|e| e.component.clone()),
s(|e| e.action.clone()),
s(|e| e.detail.clone()),
s(|e| e.state_json.clone()),
s(|e| e.level.clone()),
s(|e| e.span_id.clone()),
s(|e| e.parent.clone()),
s(|e| e.user.clone()),
s(|e| e.app.clone()),
s(|e| e.app_version.clone()),
s(|e| e.git_sha.clone()),
s(event_sha),
]
}
/// The canonical Arrow schema for the `saga_events` table (from
/// `iceberg_schema::saga_events`) — used to build an append batch for a
/// [`Warehouse`] that has no locally-loaded table (e.g. the remote gRPC backend).
/// Identical to the loaded-table schema (both derive from the same canonical
/// Iceberg schema via `schema_to_arrow_schema`).
fn saga_arrow_schema() -> Result<Arc<arrow::datatypes::Schema>> {
Ok(Arc::new(schema_to_arrow_schema(
&super::iceberg_schema::saga_events()?,
)?))
}
/// Which sessions / events a [`query_saga_events`] read returns.
#[derive(Debug, Clone)]
pub enum SagaSelector {
/// One session's events.
Session(String),
/// Every event in the table.
All,
}
/// Read saga events, scoped by `sel`, returned sorted by `(session_id, seq)`.
pub async fn query_saga_events(
wh: &IcebergWarehouse,
sel: &SagaSelector,
) -> Result<Vec<SagaEvent>> {
// Blank/torn Iceberg metadata JSON → 0 rows, never a JSON-EOF crash (load+read).
let batches: Vec<RecordBatch> =
super::iceberg::load_and_read_all(wh, TABLE_SAGA_EVENTS).await?;
let mut out: Vec<SagaEvent> = Vec::new();
for b in &batches {
for row in batch_to_events(b)? {
let keep = match sel {
SagaSelector::Session(id) => &row.session_id == id,
SagaSelector::All => true,
};
if keep {
out.push(row);
}
}
}
out.sort_by(|a, b| (&a.session_id, a.seq).cmp(&(&b.session_id, b.seq)));
Ok(out)
}
/// Decode one Arrow [`RecordBatch`] of `saga_events` rows back into [`SagaEvent`]s
/// (column order matches `iceberg_schema::saga_events`). Shared by the inherent
/// [`query_saga_events`] and the [`WarehouseTapeStore`] read path — one decoder,
/// no twin (REUSE). Rows are returned batch-order; callers scope + sort.
fn batch_to_events(b: &RecordBatch) -> Result<Vec<SagaEvent>> {
let session_id = col_str(b, COL_SESSION_ID)?;
let seq = col_i64(b, COL_SEQ)?;
let ts = col_ts(b, COL_TS_MICROS)?;
let kind = col_str(b, COL_KIND)?;
let route = col_str(b, COL_ROUTE)?;
let component = col_str(b, COL_COMPONENT)?;
let action = col_str(b, COL_ACTION)?;
let detail = col_str(b, COL_DETAIL)?;
let state_json = col_str(b, COL_STATE_JSON)?;
let level = col_str(b, COL_LEVEL)?;
let span_id = col_str(b, COL_SPAN_ID)?;
let parent = col_str(b, COL_PARENT)?;
let app_user = col_str(b, COL_APP_USER)?;
let app = col_str(b, COL_APP)?;
let app_version = col_str(b, COL_APP_VERSION)?;
let git_sha = col_str(b, COL_GIT_SHA)?;
let _sha = col_str(b, COL_SHA)?; // integrity column; not part of the model.
let mut rows = Vec::with_capacity(b.num_rows());
for i in 0..b.num_rows() {
rows.push(SagaEvent {
seq: seq.value(i) as u64,
ts_micros: ts.value(i) as u64,
session_id: session_id.value(i).to_string(),
user: app_user.value(i).to_string(),
kind: EventKind::parse(kind.value(i)).unwrap_or(EventKind::Log),
route: route.value(i).to_string(),
component: component.value(i).to_string(),
action: action.value(i).to_string(),
detail: detail.value(i).to_string(),
state_json: state_json.value(i).to_string(),
level: level.value(i).to_string(),
span_id: span_id.value(i).to_string(),
parent: parent.value(i).to_string(),
app: app.value(i).to_string(),
app_version: app_version.value(i).to_string(),
git_sha: git_sha.value(i).to_string(),
});
}
Ok(rows)
}
// The `SagaSessionSummary` Sessions-pane roll-up and its `summarize_sessions`
// summariser (painful-first) are the canonical ones from `nornir_saga::analysis`,
// re-exported at the top of this module (move=delete — the byte-identical local
// copy that used to live right here was deleted; one home, REUSE).
/// List recorded sessions in the warehouse as viz-pane rows (newest first).
pub async fn list_saga_sessions(wh: &IcebergWarehouse) -> Result<Vec<SagaSessionSummary>> {
let events = query_saga_events(wh, &SagaSelector::All).await?;
Ok(summarize_sessions(&events))
}
/// The [`ReplayPlan`] for one session — read its events back, build the plan.
pub async fn session_replay_plan(wh: &IcebergWarehouse, session_id: &str) -> Result<ReplayPlan> {
let events = query_saga_events(wh, &SagaSelector::Session(session_id.to_string())).await?;
Ok(ReplayPlan::for_session(session_id, &events))
}
/// The **nav-path / funnel** for one session (the `nornir saga path <session>`
/// verb): read its `saga_events` back and fold the `nav` stream into an ordered
/// route [`NavFunnel`]. An unknown/empty session yields an empty funnel (never an
/// error) so the CLI prints a clear "no events" message instead of panicking.
pub async fn session_nav_funnel(wh: &IcebergWarehouse, session_id: &str) -> Result<NavFunnel> {
let events = query_saga_events(wh, &SagaSelector::Session(session_id.to_string())).await?;
Ok(NavFunnel::from_events(&events))
}
/// The warehouse-backed [`SagaSink`]: appends captured events to the Iceberg
/// `saga_events` table. Leaf apps with no warehouse use `nornir_saga`'s
/// `JsonFileSink` / `NullSink` instead.
pub struct IcebergSagaSink<'a> {
wh: &'a IcebergWarehouse,
}
impl<'a> IcebergSagaSink<'a> {
pub fn new(wh: &'a IcebergWarehouse) -> Self {
Self { wh }
}
}
impl SagaSink for IcebergSagaSink<'_> {
fn write(&self, events: &[SagaEvent]) -> Result<()> {
// The warehouse's block_on bridges the sync SagaSink trait to the async
// Iceberg writer (the rest of nornir's CLI is sync over block_on too).
self.wh.block_on(append_saga_events(self.wh, events))
}
}
/// Bridge a UI-movement recording ([`nornir_saga::ui::Saga`]) into the warehouse
/// `saga_events` model — the `Nav`/`Click` movements become replayable
/// [`SagaEvent`]s (see [`nornir_saga::ui::Saga::to_saga_events`]). The seam that
/// lets a web/egui `Saga` capture flow into the same `saga_events` table + replay
/// path the native recorder feeds.
pub fn saga_from_ui(saga: &nornir_saga::ui::Saga) -> Vec<SagaEvent> {
saga.to_saga_events()
}
/// The **live-Mimir** [`SagaTapeStore`]: lands recorded tapes into (and reads them
/// back out of) nornir's Iceberg `saga_events` table over the durable
/// [`Warehouse`] seam. This is the read/write half a **tape recorder** needs — the
/// store the dwarves tape-recorder pane drives over gRPC (nornir hosts it; dwarves
/// cannot depend on nornir without a cycle, so it implements the same trait on its
/// side of the wire).
///
/// Backed by `Arc<dyn Warehouse>`, so it rides either the local
/// [`IcebergWarehouse`] or a remote gRPC warehouse unchanged, and it reuses the
/// same `saga_cols`/`batch_to_events` encode/decode + [`summarize_sessions`]
/// (painful-first) the inherent `append_saga_events`/`query_saga_events` path uses
/// — one home, no twin (REUSE).
pub struct WarehouseTapeStore {
wh: Arc<dyn Warehouse>,
}
impl WarehouseTapeStore {
/// A tape store over the given warehouse (typically the live Mimir one).
pub fn new(wh: Arc<dyn Warehouse>) -> Self {
Self { wh }
}
/// Read every recorded event back, decoded and sorted by `(session_id, seq)`.
/// A blank/torn/absent table reads as 0 rows (the `scan_arrow` contract), never
/// a crash.
fn read_all(&self) -> Result<Vec<SagaEvent>> {
let mut out: Vec<SagaEvent> = Vec::new();
for b in &self.wh.scan_arrow(TABLE_SAGA_EVENTS)? {
out.extend(batch_to_events(b)?);
}
out.sort_by(|a, b| (&a.session_id, a.seq).cmp(&(&b.session_id, b.seq)));
Ok(out)
}
}
impl SagaTapeStore for WarehouseTapeStore {
fn append_events(&self, events: &[SagaEvent]) -> Result<()> {
if events.is_empty() {
return Ok(());
}
// Group by session so each append is a single `session_id` partition
// (mirrors the inherent `append_saga_events`).
let mut by_session: BTreeMap<&str, Vec<&SagaEvent>> = BTreeMap::new();
for e in events {
by_session.entry(e.session_id.as_str()).or_default().push(e);
}
let schema = saga_arrow_schema()?;
for (_session, batch) in by_session {
let rb = RecordBatch::try_new(schema.clone(), saga_cols(&batch))?;
self.wh.append_arrow(TABLE_SAGA_EVENTS, rb)?;
}
Ok(())
}
fn list_sessions(&self) -> Result<Vec<SagaSessionSummary>> {
Ok(summarize_sessions(&self.read_all()?))
}
fn read_session(&self, session_id: &str) -> Result<Vec<SagaEvent>> {
let mut evs: Vec<SagaEvent> = self
.read_all()?
.into_iter()
.filter(|e| e.session_id == session_id)
.collect();
// `read_all` already sorts by seq within a session; scope the guarantee
// here too so the `read_session` contract is self-standing.
evs.sort_by_key(|e| e.seq);
Ok(evs)
}
}
// ─── column helpers ──────────────────────────────────────────────────────
fn col_str(b: &RecordBatch, idx: usize) -> Result<&StringArray> {
b.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("saga_events col {idx} is not StringArray"))
}
fn col_i64(b: &RecordBatch, idx: usize) -> Result<&Int64Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| anyhow!("saga_events col {idx} is not Int64Array"))
}
fn col_ts(b: &RecordBatch, idx: usize) -> Result<&TimestampMicrosecondArray> {
b.column(idx)
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
.ok_or_else(|| anyhow!("saga_events col {idx} is not TimestampMicrosecondArray"))
}
#[cfg(test)]
mod tests {
use super::*;
fn ev(session: &str, seq: u64, kind: EventKind, comp: &str) -> SagaEvent {
let ctx = SessionCtx::new(session, "alice", "njord", "0.9.0", "abc123");
let mut e = match kind {
EventKind::Nav => SagaEvent::nav("/home", comp, "open", r#"{"v":1}"#),
EventKind::Click => SagaEvent::click("/home", comp, "submit", r#"{"v":2}"#),
EventKind::Error => SagaEvent::error(comp, "boom"),
EventKind::Log => SagaEvent::log("info", "noise"),
EventKind::State => SagaEvent::state(comp, r#"{"v":3}"#),
}
.in_session(&ctx);
e.seq = seq;
e.ts_micros = 1000 + seq;
e
}
#[test]
fn ui_saga_bridges_and_round_trips_through_the_warehouse() {
use nornir_saga::ui::{Saga, UiMsg};
// A UI-movement recording (web/egui) with 2 actionable + 2 noise moves.
let saga = Saga::new("checkout")
.push(UiMsg::nav("/checkout").at(0).with_state(r#"{"step":1}"#))
.push(UiMsg::pointer_move(1.0, 2.0).at(5))
.push(UiMsg::click(3.0, 4.0, "💳 Pay").at(10))
.push(UiMsg::key("Enter").at(15));
let mut events = saga_from_ui(&saga);
assert_eq!(events.len(), 2, "only Nav + Click bridge across");
// Stamp identity + seq so they group into one session, then persist +
// read back through the same warehouse path the native recorder uses.
let ctx = SessionCtx::new("ui-sess", "alice", "njord", "0.9.0", "abc123");
for (i, e) in events.iter_mut().enumerate() {
*e = e.clone().in_session(&ctx);
e.seq = i as u64;
e.ts_micros = 1000 + i as u64;
}
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
IcebergSagaSink::new(&wh).write(&events).unwrap();
// The session lists + its replay plan carries exactly the nav+click steps.
let plan = wh.block_on(session_replay_plan(&wh, "ui-sess")).unwrap();
assert_eq!(plan.steps.len(), 2, "nav + click → 2 replay steps");
assert_eq!(plan.steps[0].route, "/checkout");
assert_eq!(plan.steps[1].component, "💳 Pay");
assert!(plan.is_replayable());
}
#[test]
fn append_query_round_trips_and_scopes() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let sess_a = vec![
ev("sA", 0, EventKind::Nav, "Dashboard"),
ev("sA", 1, EventKind::Log, "log"),
ev("sA", 2, EventKind::Click, "Submit"),
];
let sess_b = vec![
ev("sB", 3, EventKind::Nav, "Map"),
ev("sB", 4, EventKind::Error, "Map"),
];
wh.block_on(append_saga_events(&wh, &sess_a)).unwrap();
wh.block_on(append_saga_events(&wh, &sess_b)).unwrap();
// Session scope returns just one journey, sorted by seq.
let a = wh
.block_on(query_saga_events(&wh, &SagaSelector::Session("sA".into())))
.unwrap();
assert_eq!(a.len(), 3);
assert!(a.windows(2).all(|w| w[0].seq < w[1].seq), "sorted by seq");
let click = a.iter().find(|e| e.component == "Submit").unwrap();
assert_eq!(click.kind, EventKind::Click);
assert_eq!(click.state_json, r#"{"v":2}"#, "state round-trips exactly");
assert_eq!(click.app, "njord");
// All scope sees both sessions.
let all = wh
.block_on(query_saga_events(&wh, &SagaSelector::All))
.unwrap();
assert_eq!(all.len(), 5);
}
#[test]
fn summarize_and_replay_plan_from_warehouse() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let events = vec![
ev("sA", 0, EventKind::Nav, "Dashboard"),
ev("sA", 1, EventKind::Log, "noise"),
ev("sA", 2, EventKind::Click, "Submit"),
ev("sB", 3, EventKind::Error, "Map"),
];
wh.block_on(append_saga_events(&wh, &events)).unwrap();
let sessions = wh.block_on(list_saga_sessions(&wh)).unwrap();
assert_eq!(sessions.len(), 2);
let a = sessions.iter().find(|s| s.session_id == "sA").unwrap();
assert_eq!(a.event_count, 3);
assert_eq!(a.step_count, 2, "nav + click are replayable steps");
assert!(a.replayable);
let b = sessions.iter().find(|s| s.session_id == "sB").unwrap();
assert_eq!(b.step_count, 0, "an error-only session has no steps");
assert_eq!(b.error_count, 1);
assert!(!b.replayable);
// The replay plan for sA is nav→click in capture order.
let plan = wh.block_on(session_replay_plan(&wh, "sA")).unwrap();
assert_eq!(plan.steps.len(), 2);
assert_eq!(plan.steps[0].kind, StepKind::Navigate);
assert_eq!(plan.steps[1].kind, StepKind::Activate);
}
#[test]
fn session_nav_funnel_from_warehouse() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let ctx = SessionCtx::new("sF", "alice", "njord", "0.9.0", "abc123");
let nav = |route: &str, seq: u64| {
let mut e = SagaEvent::nav(route, "Page", "open", "{}").in_session(&ctx);
e.seq = seq;
e.ts_micros = 1000 + seq;
e
};
// /a → /b → /c, deterministic order.
let events = vec![nav("/a", 0), nav("/b", 1), nav("/c", 2)];
wh.block_on(append_saga_events(&wh, &events)).unwrap();
let funnel = wh.block_on(session_nav_funnel(&wh, "sF")).unwrap();
assert_eq!(funnel.session_count, 1);
assert_eq!(
funnel
.steps
.iter()
.map(|s| s.route.as_str())
.collect::<Vec<_>>(),
vec!["/a", "/b", "/c"],
"ordered route funnel read back from the warehouse"
);
// An unknown session is an EMPTY funnel, not an error (CLI prints "no events").
let empty = wh.block_on(session_nav_funnel(&wh, "ghost")).unwrap();
assert_eq!(empty.session_count, 0);
assert!(empty.steps.is_empty());
}
#[test]
fn iceberg_sink_writes_via_trait() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let sink = IcebergSagaSink::new(&wh);
sink.write(&[ev("sS", 0, EventKind::Nav, "Home")]).unwrap();
let back = wh
.block_on(query_saga_events(&wh, &SagaSelector::Session("sS".into())))
.unwrap();
assert_eq!(back.len(), 1);
assert_eq!(back[0].route, "/home");
}
#[test]
fn stuck_sessions_float_to_the_top_of_the_summary() {
// A calm session (started LATER) and a rage-click session (started
// EARLIER). Recency alone would rank the calm one first; the frustration
// score must override that and float the stuck session to the top.
let calm = SessionCtx::new("calm", "u", "njord", "1.0", "sha");
let rage = SessionCtx::new("rage", "u", "njord", "1.0", "sha");
let mut events = Vec::new();
// rage: 3 consecutive clicks on one component → a rage cluster.
for s in 0..3u64 {
let mut e = SagaEvent::click("/x", "Retry", "retry", "{}").in_session(&rage);
e.seq = s;
e.ts_micros = 100 + s; // earliest
events.push(e);
}
// calm: a plain nav, later in wall-clock.
let mut e = SagaEvent::nav("/home", "Home", "open", "{}").in_session(&calm);
e.seq = 10;
e.ts_micros = 9000; // most recent
events.push(e);
let summaries = summarize_sessions(&events);
assert_eq!(summaries.len(), 2);
assert_eq!(
summaries[0].session_id, "rage",
"the stuck session floats above the more-recent calm one"
);
assert!(summaries[0].stuck, "rage session flagged stuck");
assert!(summaries[0].frustration_score > 0);
assert!(!summaries[1].stuck, "calm session not stuck");
assert_eq!(summaries[1].frustration_score, 0);
}
#[test]
fn warehouse_tape_store_round_trips_over_dyn_warehouse() {
// The live-Mimir `SagaTapeStore` (append → list → read) over the real
// Iceberg warehouse behind the `dyn Warehouse` seam. Red-when-broken:
// any break in the encode/decode round-trip or the painful-first ordering
// fails an assert.
let dir = tempfile::tempdir().unwrap();
let wh: Arc<dyn Warehouse> = Arc::new(IcebergWarehouse::open(dir.path()).unwrap());
let store = WarehouseTapeStore::new(wh);
// Empty store: no sessions, empty read (blank table → 0 rows, never crash).
assert!(
store.list_sessions().unwrap().is_empty(),
"empty store → no sessions"
);
assert!(store.read_session("nope").unwrap().is_empty());
store.append_events(&[]).unwrap(); // empty append is a no-op success
assert!(store.list_sessions().unwrap().is_empty());
// A rage-click session (started EARLIER) and a calm nav (started LATER):
// recency alone would rank calm first; the frustration score must float
// the stuck session to the top of `list_sessions`.
let rage_ctx = SessionCtx::new("rage", "u", "njord", "1.0", "sha");
let mut rage = Vec::new();
for s in 0..3u64 {
let mut e =
SagaEvent::click("/x", "Retry", "retry", r#"{"n":1}"#).in_session(&rage_ctx);
e.seq = s;
e.ts_micros = 100 + s; // earliest
rage.push(e);
}
let calm = ev("calm", 9, EventKind::Nav, "Home"); // ts_micros = 1009, most recent
store.append_events(&rage).unwrap();
store.append_events(&[calm]).unwrap();
// list_sessions: painful-first — the stuck session floats above the calm one.
let sessions = store.list_sessions().unwrap();
assert_eq!(sessions.len(), 2, "two distinct sessions rolled up");
assert_eq!(
sessions[0].session_id, "rage",
"the stuck session floats above the more-recent calm one"
);
assert!(sessions[0].stuck, "rage session flagged stuck");
assert!(sessions[0].frustration_score > 0);
assert_eq!(sessions[0].step_count, 3, "3 clicks are replayable steps");
assert!(!sessions[1].stuck && sessions[1].session_id == "calm");
// read_session: the rage tape back, scoped, seq-ordered, byte-for-byte.
let back = store.read_session("rage").unwrap();
assert_eq!(back.len(), 3, "only rage's events");
assert!(back.iter().all(|e| e.session_id == "rage"));
assert!(
back.windows(2).all(|w| w[0].seq <= w[1].seq),
"sorted by seq"
);
assert_eq!(back[0].component, "Retry");
assert_eq!(
back[0].state_json, r#"{"n":1}"#,
"state_json round-trips exactly"
);
assert_eq!(back[0].app, "njord");
// An unknown session reads back empty (never an error).
assert!(store.read_session("ghost").unwrap().is_empty());
}
#[test]
fn warehouse_tape_store_is_object_safe() {
// Hold it behind the trait object — the host swap seam (file ↔ warehouse).
let dir = tempfile::tempdir().unwrap();
let wh: Arc<dyn Warehouse> = Arc::new(IcebergWarehouse::open(dir.path()).unwrap());
let store: Box<dyn SagaTapeStore> = Box::new(WarehouseTapeStore::new(wh));
store
.append_events(&[ev("s", 0, EventKind::Nav, "Home")])
.unwrap();
assert_eq!(store.list_sessions().unwrap().len(), 1);
assert_eq!(store.read_session("s").unwrap().len(), 1);
}
#[test]
fn event_sha_is_stable_and_payload_sensitive() {
let a = ev("sA", 0, EventKind::Nav, "Dashboard");
let same = ev("sA", 0, EventKind::Nav, "Dashboard");
assert_eq!(event_sha(&a), event_sha(&same), "same payload → same sha");
let other = ev("sA", 0, EventKind::Click, "Dashboard");
assert_ne!(
event_sha(&a),
event_sha(&other),
"different kind → different sha"
);
}
/// **THE "usable" PROOF (producer → receiver, embedded)** — a live app's
/// [`SagaRecorder`] records a session and flushes it through the RECEIVER path
/// (a [`WarehouseTapeStore`] over an embedded Mimir `saga_events` table); the tape
/// then reads back painful-first + seq-ordered with `state_json` intact. This is
/// the fat (local-warehouse) recording path. Red-when-broken: a dropped/reordered
/// event, a mangled `state_json`, or a broken painful-first ranking fails an assert.
#[test]
fn recorder_flushes_a_session_into_mimir_and_reads_back_painful_first() {
let dir = tempfile::tempdir().unwrap();
// ONE embedded warehouse shared by every store — the receiver both recorders
// flush into and the reader reads back from.
let wh: Arc<dyn Warehouse> = Arc::new(IcebergWarehouse::open(dir.path()).unwrap());
// A RAGE-CLICK session: three hammered "Retry" clicks — stuck.
let rage_ctx = SessionCtx::new("rage", "alice", "njord", "0.9.0", "abc123");
let rage = SagaRecorder::new(rage_ctx, WarehouseTapeStore::new(wh.clone()));
rage.click("/checkout", "Retry", "retry", r#"{"n":1}"#);
rage.click("/checkout", "Retry", "retry", r#"{"n":2}"#);
rage.click("/checkout", "Retry", "retry", r#"{"n":3}"#);
assert_eq!(rage.pending(), 3, "three clicks buffered before flush");
assert_eq!(
rage.flush().unwrap(),
3,
"the rage session flushed into the warehouse"
);
assert_eq!(rage.pending(), 0, "buffer cleared on a durable flush");
// A CALM nav session, recorded LATER (more recent in wall-clock).
let calm_ctx = SessionCtx::new("calm", "alice", "njord", "0.9.0", "abc123");
let calm = SagaRecorder::new(calm_ctx, WarehouseTapeStore::new(wh.clone()));
calm.nav("/home", "Home", "open", r#"{"tab":"home"}"#);
calm.flush().unwrap();
// Read the tape back through the receiver (a fresh store, same warehouse).
let reader = WarehouseTapeStore::new(wh.clone());
let sessions = reader.list_sessions().unwrap();
assert_eq!(sessions.len(), 2, "both recorded sessions landed in Mimir");
assert_eq!(
sessions[0].session_id, "rage",
"the stuck session floats above the more-recent calm one (painful-first)"
);
assert!(sessions[0].stuck, "the rage-click cluster is flagged stuck");
assert_eq!(
sessions[0].step_count, 3,
"three clicks are replayable steps"
);
assert_eq!(
sessions[0].app, "njord",
"session carries the producer's app identity"
);
assert!(!sessions[1].stuck && sessions[1].session_id == "calm");
// read_session: the rage tape back, scoped, seq-ordered, state_json byte-exact.
let back = reader.read_session("rage").unwrap();
assert_eq!(back.len(), 3, "exactly the rage session's events");
assert!(
back.iter().all(|e| e.session_id == "rage"),
"scoped to the session"
);
assert!(
back.windows(2).all(|w| w[0].seq < w[1].seq),
"strictly seq-ordered"
);
assert_eq!(back[0].component, "Retry");
assert_eq!(back[0].state_json, r#"{"n":1}"#, "first state_json intact");
assert_eq!(
back[2].state_json, r#"{"n":3}"#,
"last state_json intact — no drop/reorder"
);
assert!(reader.read_session("ghost").unwrap().is_empty());
}
}