src/saga_grpc.rs
//! **Saga over gRPC** — the remote ingest + read-back endpoint that makes saga
//! recording usable from a THIN producer (Njord, a leaf app with no local
//! warehouse) and the client-side store that ships to it (EPIC SAGA / task #42).
//!
//! ```text
//! thin producer nornir-server (holds the warehouse lock)
//! SagaRecorder<GrpcSagaStore> ──RecordEvents──▶ SagaGrpcService ─┐
//! │ .flush() │ append_events
//! ▼ ▼
//! (buffer) WarehouseTapeStore → Mimir `saga_events`
//! ◀──ListSessions/ReadSession── (painful-first roll-up + seq-ordered)
//! ```
//!
//! Two halves, both riding the ONE [`SagaTapeStore`] seam (REUSE — no second codec,
//! no second event type):
//!
//! * [`SagaGrpcService`] — the **server**. A generic tonic `Saga` service backed by
//! a per-request [`SagaTapeStore`] (resolved from the call's metadata, so the
//! `nornir-server` picks the served workspace's [`WarehouseTapeStore`]). It lands
//! `RecordEvents` into the store and answers `ListSessions` / `ReadSession` from
//! it. The warehouse calls are sync (they `block_on` the iceberg writer), so each
//! handler bridges them off the async reactor with `spawn_blocking` (an I/O
//! bridge, not CPU fan-out).
//! * [`GrpcSagaStore`] — the **client**. A [`SagaTapeStore`] impl that forwards
//! `append_events` / `list_sessions` / `read_session` to a nornir server's `Saga`
//! service, so a thin app's [`SagaRecorder`](nornir_saga::SagaRecorder) flushes
//! over the wire exactly as a fat app's flushes to the local warehouse.
//!
//! [`WarehouseTapeStore`]: crate::warehouse::saga_events::WarehouseTapeStore
use std::sync::Arc;
use anyhow::Result;
use tonic::{Request, Response, Status};
use nornir_saga::event::EventKind;
use nornir_saga::{SagaEvent, SagaTapeStore};
/// The generated `nornir.v1` prost/tonic bindings for the saga surface, now
/// sourced from the shared `nornir-wire` edda leaf (`nornir_wire::pb`) and
/// re-exported so this module's `pb::…` paths are unchanged.
pub use nornir_wire::pb;
/// A store the service resolves per RPC and holds across the (blocking) call.
pub type DynStore = Arc<dyn SagaTapeStore + Send + Sync>;
// ─── codec: SagaEvent ⇄ SagaEventPb (one home) ─────────────────────────────
/// Encode a [`SagaEvent`] onto the wire message (all 16 replay-relevant fields).
pub fn event_to_pb(e: &SagaEvent) -> pb::SagaEventPb {
pb::SagaEventPb {
seq: e.seq,
ts_micros: e.ts_micros,
session_id: e.session_id.clone(),
user: e.user.clone(),
kind: e.kind.as_str().to_string(),
route: e.route.clone(),
component: e.component.clone(),
action: e.action.clone(),
detail: e.detail.clone(),
state_json: e.state_json.clone(),
level: e.level.clone(),
span_id: e.span_id.clone(),
parent: e.parent.clone(),
app: e.app.clone(),
app_version: e.app_version.clone(),
git_sha: e.git_sha.clone(),
}
}
/// Decode a wire message back into a [`SagaEvent`]. An unknown `kind` string folds
/// to [`EventKind::Log`] (forward-compatible — a newer producer's kind never
/// crashes an older receiver), matching the warehouse read path.
pub fn pb_to_event(p: pb::SagaEventPb) -> SagaEvent {
SagaEvent {
seq: p.seq,
ts_micros: p.ts_micros,
session_id: p.session_id,
user: p.user,
kind: EventKind::parse(&p.kind).unwrap_or(EventKind::Log),
route: p.route,
component: p.component,
action: p.action,
detail: p.detail,
state_json: p.state_json,
level: p.level,
span_id: p.span_id,
parent: p.parent,
app: p.app,
app_version: p.app_version,
git_sha: p.git_sha,
}
}
// ─── the server: SagaGrpcService ────────────────────────────────────────────
/// The tonic `Saga` service: land recorded tapes into — and read them back out of —
/// a [`SagaTapeStore`] resolved per request from the call's metadata.
///
/// Generic over the resolver `R` so the SAME service serves the `nornir-server`
/// (which resolves the served workspace's [`WarehouseTapeStore`] from the
/// `nornir-workspace` header) and an in-process test (which returns one fixed
/// store). Build the tonic server with [`SagaGrpcService::into_server`].
///
/// Gated on `server`: the `pb::saga_server` bindings this service impls are only
/// generated by `build.rs`'s `build_server(want_server)` — a `viz`/`mcp` build (no
/// `server`) gets the client bindings + [`GrpcSagaStore`] but not the service.
#[cfg(feature = "server")]
pub struct SagaGrpcService<R> {
resolve: R,
}
#[cfg(feature = "server")]
impl<R> SagaGrpcService<R>
where
R: Fn(&tonic::metadata::MetadataMap) -> Result<DynStore, Status> + Send + Sync + 'static,
{
/// A service that resolves the destination store from each call's metadata.
pub fn new(resolve: R) -> Self {
Self { resolve }
}
/// Wrap this service in the generated tonic `SagaServer` (ready for
/// `Routes::add_service` / `Server::add_service`).
pub fn into_server(self) -> pb::saga_server::SagaServer<Self> {
pb::saga_server::SagaServer::new(self)
}
}
#[cfg(feature = "server")]
#[tonic::async_trait]
impl<R> pb::saga_server::Saga for SagaGrpcService<R>
where
R: Fn(&tonic::metadata::MetadataMap) -> Result<DynStore, Status> + Send + Sync + 'static,
{
/// Land a producer's captured events into the store. Ack = count accepted.
async fn record_events(
&self,
request: Request<pb::RecordSagaEventsRequest>,
) -> Result<Response<pb::SubmitCountResponse>, Status> {
let store = (self.resolve)(request.metadata())?;
let events: Vec<SagaEvent> = request
.into_inner()
.events
.into_iter()
.map(pb_to_event)
.collect();
let n = events.len() as u32;
if !events.is_empty() {
// The warehouse append is a blocking I/O write (it `block_on`s the
// iceberg writer) — run it on the blocking pool so the async reactor
// stays free. NOT CPU fan-out: one batch, one store call.
tokio::task::spawn_blocking(move || store.append_events(&events))
.await
.map_err(|e| Status::internal(format!("saga append task: {e}")))?
.map_err(|e| Status::internal(format!("saga append: {e:#}")))?;
}
Ok(Response::new(pb::SubmitCountResponse { accepted: n }))
}
/// The painful-first Sessions roll-up as JSON (a `Vec<SagaSessionSummary>`).
async fn list_sessions(
&self,
request: Request<pb::Empty>,
) -> Result<Response<pb::JsonResponse>, Status> {
let store = (self.resolve)(request.metadata())?;
let sessions = tokio::task::spawn_blocking(move || store.list_sessions())
.await
.map_err(|e| Status::internal(format!("saga list task: {e}")))?
.map_err(|e| Status::internal(format!("saga list_sessions: {e:#}")))?;
let json = serde_json::to_string(&sessions)
.map_err(|e| Status::internal(format!("serialize sessions: {e}")))?;
Ok(Response::new(pb::JsonResponse { json }))
}
/// One session's events, scoped + seq-ordered, for replay.
async fn read_session(
&self,
request: Request<pb::SagaSessionQuery>,
) -> Result<Response<pb::SagaEventList>, Status> {
let store = (self.resolve)(request.metadata())?;
let session_id = request.into_inner().session_id;
let events = tokio::task::spawn_blocking(move || store.read_session(&session_id))
.await
.map_err(|e| Status::internal(format!("saga read task: {e}")))?
.map_err(|e| Status::internal(format!("saga read_session: {e:#}")))?;
Ok(Response::new(pb::SagaEventList {
events: events.iter().map(event_to_pb).collect(),
}))
}
}
// ─── the client: GrpcSagaStore ─────────────────────────────────────────────
/// A [`SagaTapeStore`] that forwards to a remote nornir server's `Saga` service —
/// the thin producer's flush destination. Holds a dedicated current-thread tokio
/// runtime so the **sync** trait methods `block_on` the async RPC without a
/// "runtime within a runtime" panic (the same shape [`GrpcSagaStore`] shares with
/// [`crate::warehouse::remote::RemoteWarehouse`]). The channel is dialed lazily per
/// call.
pub struct GrpcSagaStore {
endpoint: String,
token: String,
workspace: String,
rt: tokio::runtime::Runtime,
}
impl GrpcSagaStore {
/// Connect a thin saga store to `endpoint` (`http://host:port`, or a bare
/// `host:port` that is `http://`-prefixed). `token` is the bearer auth (empty ⇒
/// none); `workspace` selects the served workspace via the `nornir-workspace`
/// header (empty ⇒ server default).
pub fn connect(
endpoint: impl Into<String>,
token: impl Into<String>,
workspace: impl Into<String>,
) -> Result<Self> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| anyhow::anyhow!("build tokio runtime for GrpcSagaStore: {e}"))?;
let endpoint = endpoint.into();
let endpoint = if endpoint.starts_with("http") {
endpoint
} else {
format!("http://{endpoint}")
};
// PREFER a cached OIDC access token (`nornir login`, read with NO network) when the
// caller passed no explicit bearer — the client half of holger-style auth. An
// explicit `token` still wins; a signed-out / lean build leaves it empty (open),
// exactly as before.
let token = token.into();
let token = if token.is_empty() {
crate::cached_oidc_bearer().unwrap_or_default()
} else {
token
};
Ok(Self {
endpoint,
token,
workspace: workspace.into(),
rt,
})
}
async fn dial(
&self,
) -> Result<
pb::saga_client::SagaClient<
tonic::service::interceptor::InterceptedService<
tonic::transport::Channel,
impl FnMut(Request<()>) -> Result<Request<()>, Status>,
>,
>,
> {
let channel = tonic::transport::Channel::from_shared(self.endpoint.clone())
.map_err(|e| anyhow::anyhow!("invalid server url `{}`: {e}", self.endpoint))?
.connect()
.await
.map_err(|e| anyhow::anyhow!("connect {}: {e}", self.endpoint))?;
let token = self.token.clone();
let workspace = self.workspace.clone();
let bearer: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> =
(!token.is_empty()).then(|| format!("Bearer {token}").parse().unwrap());
let ws: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> =
(!workspace.is_empty()).then(|| workspace.parse().unwrap());
Ok(pb::saga_client::SagaClient::with_interceptor(
channel,
move |mut req: Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
if let Some(w) = &ws {
req.metadata_mut().insert("nornir-workspace", w.clone());
}
Ok(req)
},
))
}
}
impl SagaTapeStore for GrpcSagaStore {
fn append_events(&self, events: &[SagaEvent]) -> Result<()> {
if events.is_empty() {
return Ok(());
}
let wire: Vec<pb::SagaEventPb> = events.iter().map(event_to_pb).collect();
let session_id = events
.first()
.map(|e| e.session_id.clone())
.unwrap_or_default();
self.rt.block_on(async move {
let mut client = self.dial().await?;
client
.record_events(pb::RecordSagaEventsRequest {
session_id,
events: wire,
})
.await
.map_err(|s| anyhow::anyhow!("Saga.RecordEvents RPC: {s}"))?;
Ok(())
})
}
fn list_sessions(&self) -> Result<Vec<nornir_saga::SagaSessionSummary>> {
self.rt.block_on(async move {
let mut client = self.dial().await?;
let json = client
.list_sessions(pb::Empty {})
.await
.map_err(|s| anyhow::anyhow!("Saga.ListSessions RPC: {s}"))?
.into_inner()
.json;
Ok(serde_json::from_str(&json)?)
})
}
fn read_session(&self, session_id: &str) -> Result<Vec<SagaEvent>> {
let session_id = session_id.to_string();
self.rt.block_on(async move {
let mut client = self.dial().await?;
let events = client
.read_session(pb::SagaSessionQuery { session_id })
.await
.map_err(|s| anyhow::anyhow!("Saga.ReadSession RPC: {s}"))?
.into_inner()
.events;
Ok(events.into_iter().map(pb_to_event).collect())
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use nornir_saga::SessionCtx;
fn ev(session: &str, seq: u64) -> SagaEvent {
let ctx = SessionCtx::new(session, "alice", "njord", "0.9.0", "abc123");
let mut e =
SagaEvent::click("/kund", "OffertForm", "submit", r#"{"ok":true}"#).in_session(&ctx);
e.seq = seq;
e.ts_micros = 1000 + seq;
e
}
#[test]
fn codec_round_trips_every_field() {
let e = ev("s", 7);
let back = pb_to_event(event_to_pb(&e));
assert_eq!(back, e, "SagaEvent ⇄ SagaEventPb is lossless");
}
#[test]
fn unknown_kind_folds_to_log_not_a_crash() {
let mut p = event_to_pb(&ev("s", 0));
p.kind = "future_kind".into();
assert_eq!(
pb_to_event(p).kind,
EventKind::Log,
"forward-compatible kind decode"
);
}
/// **THE REMOTE-INGEST ROUND-TRIP** — a THIN producer's [`SagaRecorder`] whose
/// store is a [`GrpcSagaStore`] records a session, flushes it over the wire to an
/// in-process nornir `Saga` service (backed by a [`WarehouseTapeStore`] over an
/// embedded warehouse), then reads the session back over gRPC. Red-when-broken: a
/// dropped/mis-ordered event, a mangled `state_json`, or a broken ack fails here.
#[cfg(feature = "server")] // exercises SagaGrpcService (server bindings) + WarehouseTapeStore
#[test]
fn thin_producer_records_over_grpc_and_reads_the_session_back() {
use crate::warehouse::Warehouse;
use crate::warehouse::iceberg::IcebergWarehouse;
use crate::warehouse::saga_events::WarehouseTapeStore;
use nornir_saga::SagaRecorder;
use std::time::Duration;
// The runtime hosting the in-process server (I/O accept loop, NOT CPU work).
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap();
// Receiver: a WarehouseTapeStore over an embedded Mimir warehouse.
let dir = tempfile::tempdir().unwrap();
let wh: Arc<dyn Warehouse> = Arc::new(IcebergWarehouse::open(dir.path()).unwrap());
let store: DynStore = Arc::new(WarehouseTapeStore::new(wh));
let svc = SagaGrpcService::new(move |_md: &tonic::metadata::MetadataMap| Ok(store.clone()))
.into_server();
// Grab a free loopback port (the probe listener is dropped before we serve).
let addr: std::net::SocketAddr = std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap();
rt.spawn(async move {
tonic::transport::Server::builder()
.add_service(svc)
.serve(addr)
.await
.expect("saga server serve");
});
std::thread::sleep(Duration::from_millis(400)); // let the accept loop bind
// The THIN producer, driven from the MAIN thread (no entered runtime, so
// GrpcSagaStore's own block_on is safe).
let client = GrpcSagaStore::connect(format!("http://{addr}"), "", "").unwrap();
let ctx = SessionCtx::new("remote-rage", "alice", "njord", "0.9.0", "abc123");
let rec = SagaRecorder::new(ctx, client);
rec.nav("/checkout", "Checkout", "open", r#"{"step":1}"#);
rec.click("/checkout", "Retry", "retry", r#"{"n":1}"#);
rec.click("/checkout", "Retry", "retry", r#"{"n":2}"#);
rec.click("/checkout", "Retry", "retry", r#"{"n":3}"#);
// flush() = RecordEvents over the wire; a broken ack panics here.
let n = rec.flush().unwrap();
assert_eq!(
n, 4,
"the ack confirms all four events were accepted by the server"
);
// Read the session back THROUGH THE SAME gRPC client store.
let sessions = rec.store().list_sessions().unwrap();
assert_eq!(
sessions.len(),
1,
"the recorded session is listed over gRPC"
);
assert_eq!(sessions[0].session_id, "remote-rage");
assert_eq!(
sessions[0].app, "njord",
"producer identity survived the round-trip"
);
assert!(
sessions[0].stuck,
"the rage-click cluster is flagged stuck server-side"
);
assert_eq!(
sessions[0].step_count, 4,
"nav + 3 clicks are replayable steps"
);
let back = rec.store().read_session("remote-rage").unwrap();
assert_eq!(
back.len(),
4,
"every recorded event came back — none dropped"
);
assert!(
back.iter().all(|e| e.session_id == "remote-rage"),
"scoped to the session"
);
assert!(
back.windows(2).all(|w| w[0].seq < w[1].seq),
"strictly seq-ordered over the wire"
);
assert_eq!(back[0].component, "Checkout");
assert_eq!(back[0].state_json, r#"{"step":1}"#, "nav state_json intact");
assert_eq!(
back[3].state_json, r#"{"n":3}"#,
"last click state_json intact — no reorder"
);
// An unknown session reads back empty (never an error) over the wire too.
assert!(rec.store().read_session("ghost").unwrap().is_empty());
}
}