src/funnel/store.rs
//! Iceberg-backed funnel store.
//!
//! Replaces the original ndjson append-only file with an Iceberg
//! `funnel_events` table under `<workspace>/.nornir/warehouse`. The
//! materialised in-memory [`Funnel`] is built by scanning every row
//! and folding via [`Funnel::apply`] — same model as before, durable
//! storage is just typed Parquet now with snapshot-per-write.
//!
//! Validation still lives in `apply`: bad events are rejected
//! before they ever touch the Iceberg writer.
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
use super::event::Event;
use super::state::Funnel;
use crate::warehouse::funnel::{append_event, append_events, load_all_events};
use crate::warehouse::iceberg::IcebergWarehouse;
pub struct Store {
wh: Arc<IcebergWarehouse>,
root: PathBuf,
/// True when opened via [`open_read_only`](Self::open_read_only) and the
/// catalog was a copied-aside snapshot — writes would land in a throwaway
/// temp dir, so [`record`](Self::record) refuses them instead of losing them.
read_only: bool,
pub funnel: Funnel,
}
impl Store {
/// Default location: the home-derived `<home>/.nornir/funnel`. The funnel
/// lives under the running user's nornir data root — ONE rule for everyone,
/// no path env-vars (the server, running as the `nornir` user with
/// `HOME=/home/nornir`, resolves `/home/nornir/.nornir/funnel`).
pub fn default_root() -> PathBuf {
crate::config::nornir_home().join("funnel")
}
/// **Single source of truth** for *where the funnel lives*, shared by the
/// CLI (`nornir funnel …`), the gRPC server (`nornir-server`), and the MCP
/// (`nornir-mcp`) so a funnel written by one is read by the others — the
/// whole point of plan item E4 (a CLI-created funnel must show up in the
/// viz Funnel tab, which reads through the server).
///
/// **Precedence (highest first):**
///
/// 1. `explicit` — an explicit `--funnel-root <dir>` flag (None when unset),
/// a *root as-given* (no `warehouse/` suffix);
/// 2. `[storage].local_path` (from `nornir.toml`) →
/// `<workspace_root>/<local_path>/warehouse` — the workspace-local home,
/// kept in lock-step with `config::Loaded::warehouse_root`;
/// 3. home-derived default → `<home>/.nornir/funnel`
/// (= [`default_root`](Self::default_root)). ONE rule for everyone: the
/// CLI (as you) and the server (as the `nornir` user) both resolve their
/// own home, so a CLI-created funnel and the server's funnel collide on
/// the same dir whenever they run as the same user — no path env-var.
pub fn resolve_root(
workspace_root: &Path,
local_path: &str,
explicit: Option<&Path>,
) -> PathBuf {
if let Some(p) = explicit {
return p.to_path_buf();
}
if local_path.is_empty() {
Self::default_root()
} else {
workspace_root.join(local_path).join("warehouse")
}
}
/// Open (or create) the warehouse at `root` and replay every
/// `funnel_events` row into a materialised [`Funnel`].
pub fn open(root: impl Into<PathBuf>) -> Result<Self> {
Self::open_inner(root, false)
}
/// Lock-tolerant **read-only** open. Behaves like [`open`](Self::open) but,
/// when the warehouse's `catalog.redb` is already held exclusively by
/// another process *or another workspace inside the same server* (the funnel
/// root is home-derived and process-global, so only one `WorkspaceCtx` can
/// hold the exclusive lock), it falls back to a copied-aside read-only
/// snapshot instead of failing. The returned store reflects the funnel as of
/// now and must not be mutated — perfect for the read-only `Funnel.Show`.
pub fn open_read_only(root: impl Into<PathBuf>) -> Result<Self> {
Self::open_inner(root, true)
}
fn open_inner(root: impl Into<PathBuf>, read_only: bool) -> Result<Self> {
let root = root.into();
// `IcebergWarehouse::open_read_only` REFUSES a root with no
// `catalog.redb`, because a read that creates the warehouse it then
// reports as empty is indistinguishable from a broken read (see its doc
// comment and the 2026-08-22 incident). The funnel is the one place that
// refusal would be wrong: `Funnel.Show` on a workspace whose funnel has
// never been written must render an EMPTY funnel, not an error — that
// red Funnel tab is a bug this store already fixed once. So the funnel,
// and only the funnel, initialises its own root on a read.
//
// Safe HERE in a way it is not in the generic path: an absent funnel has
// no rows to contradict, so an empty answer IS the truth, and
// `funnel::outcome` already refuses to report an empty funnel as a
// green. A warehouse table, by contrast, can advertise 3.6 M records in
// its own metadata while the reader sees none — that contradiction is
// exactly what the generic refusal exists to catch.
let uninitialised = read_only && !root.join("catalog.redb").is_file();
let wh = if read_only && !uninitialised {
IcebergWarehouse::open_read_only(&root)
} else {
IcebergWarehouse::open(&root)
};
let wh =
Arc::new(wh.with_context(|| format!("open iceberg warehouse at {}", root.display()))?);
let events = wh
.block_on(async { load_all_events(&wh).await })
.with_context(|| format!("load funnel_events from {}", root.display()))?;
let mut funnel = Funnel::default();
for (i, ev) in events.iter().enumerate() {
funnel
.apply(ev)
.with_context(|| format!("replay funnel event {} of {}", i + 1, events.len()))?;
}
// `read_only` may have been requested *and not contended* (got the real
// exclusive lock): then it's a normal writable open. Only a copied-aside
// snapshot is truly read-only.
let read_only = wh.is_snapshot();
Ok(Self {
wh,
root,
read_only,
funnel,
})
}
/// True when this store is a read-only snapshot (catalog was locked
/// elsewhere). Writes are refused; reads reflect the funnel as of open time.
pub fn is_read_only(&self) -> bool {
self.read_only
}
/// Validate, then append the event as a new row (= new Iceberg snapshot).
pub fn record(&mut self, event: Event) -> Result<()> {
if self.read_only {
anyhow::bail!(
"funnel store at {} is a read-only snapshot (catalog locked by \
another process/workspace); cannot record events",
self.root.display()
);
}
self.funnel
.apply(&event)
.context("validate event against current funnel state")?;
self.wh
.block_on(async { append_event(&self.wh, &event).await })
.context("append event to iceberg")?;
Ok(())
}
/// Persist a BATCH of events as ONE Iceberg snapshot (one commit/fsync)
/// instead of one snapshot per event — the fsync-amortizing bulk-write path.
///
/// PERSIST-ONLY: unlike [`record`](Self::record), this does NOT apply the
/// events to the in-memory funnel — the caller is responsible for having
/// already `funnel.apply`'d them (bulk writers like `store_subtasks` must
/// apply as they build, because id generation reads the advancing
/// `funnel.next_node`/`next_plan`). Readback is byte-identical to N single
/// `record` calls — only the snapshot count differs.
pub fn persist_events(&mut self, events: &[Event]) -> Result<()> {
if self.read_only {
anyhow::bail!(
"funnel store at {} is a read-only snapshot (catalog locked by \
another process/workspace); cannot record events",
self.root.display()
);
}
if events.is_empty() {
return Ok(());
}
self.wh
.block_on(async { append_events(&self.wh, events).await })
.context("append event batch to iceberg")?;
Ok(())
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn warehouse(&self) -> &Arc<IcebergWarehouse> {
&self.wh
}
/// Async wrapper for [`Store::open`] safe to call from inside a
/// tokio runtime. Runs the blocking open on the dedicated
/// blocking-thread pool so the warehouse's inner runtime can
/// drive the catalog without a nested-runtime panic.
pub async fn open_async(root: impl Into<PathBuf>) -> Result<Self> {
let root = root.into();
tokio::task::spawn_blocking(move || Self::open(root))
.await
.context("join blocking task for Store::open")?
}
/// Async wrapper for [`Store::open_read_only`] — the lock-tolerant read
/// open, safe to call from inside a tokio runtime.
pub async fn open_read_only_async(root: impl Into<PathBuf>) -> Result<Self> {
let root = root.into();
tokio::task::spawn_blocking(move || Self::open_read_only(root))
.await
.context("join blocking task for Store::open_read_only")?
}
/// Async wrapper for [`Store::record`] safe to call from inside a
/// tokio runtime. Validation happens inline; the iceberg append
/// is offloaded to the blocking pool.
pub async fn record_async(&mut self, event: Event) -> Result<()> {
if self.read_only {
anyhow::bail!(
"funnel store at {} is a read-only snapshot (catalog locked by \
another process/workspace); cannot record events",
self.root.display()
);
}
self.funnel
.apply(&event)
.context("validate event against current funnel state")?;
let wh = Arc::clone(&self.wh);
let ev = event;
tokio::task::spawn_blocking(move || wh.block_on(async { append_event(&wh, &ev).await }))
.await
.context("join blocking task for append_event")??;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::funnel::event::Event;
use crate::funnel::event::{ItemKind, NodeStatus, PlanStatus};
use crate::funnel::ids::{IdeaId, NodeId, PlanId};
use chrono::Utc;
use std::sync::Mutex;
use tempfile::tempdir;
/// `$HOME` is process-global; serialize the HOME-mutating test so it can't
/// observe another test's value under the parallel runner.
static HOME_LOCK: Mutex<()> = Mutex::new(());
/// Root-resolution precedence: explicit override beats `[storage].local_path`
/// beats the home-derived `<home>/.nornir/funnel` default — and it's
/// identical to what the server/MCP compute, so the CLI writes where they
/// read. NO path env-var: the default is the running user's home.
#[test]
fn resolve_root_precedence() {
let _g = HOME_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let ws = Path::new("/ws");
// 3. home-derived default (empty local_path). Pin HOME so the assertion
// is deterministic regardless of host.
// TODO: Audit that the environment access only happens in single-threaded code.
unsafe { std::env::set_var("HOME", "/home/dora") };
assert_eq!(
Store::resolve_root(ws, "", None),
PathBuf::from("/home/dora/.nornir/funnel"),
);
// 2. config local_path joins under the workspace root, +`warehouse/`.
assert_eq!(
Store::resolve_root(ws, "store", None),
ws.join("store").join("warehouse"),
);
// 1. explicit --funnel-root beats everything, as-given.
assert_eq!(
Store::resolve_root(ws, "store", Some(Path::new("/explicit"))),
PathBuf::from("/explicit"),
);
}
/// **Inject-and-assert (E4):** write a full idea→plan→nodes→edges funnel
/// through the *exact same `Store::resolve_root` + `Store::record` calls the
/// CLI's `run_funnel` makes* into an explicit temp root, then reopen through
/// the *same resolution the server/MCP use* and assert the plan, its nodes,
/// and its edges match byte-for-byte — the regression guard for "a
/// CLI-created funnel doesn't show up in the viz Funnel tab".
#[test]
fn cli_write_server_read_roundtrip() {
let dir = tempdir().unwrap();
let root = dir.path().join("explicit-funnel");
let ws = dir.path().join("workspace"); // NOT where the funnel lands
// --- CLI WRITE PATH: resolve via an explicit `--funnel-root` (the
// workspace-local `local_path` is deliberately set to prove explicit
// wins), then record the same event sequence `nornir funnel
// submit/plan/node/link` emits.
let cli_root = Store::resolve_root(&ws, "store", Some(&root));
assert_eq!(cli_root, root, "explicit root must win over local_path");
// Replay orders events by `ts` (Iceberg scan order isn't stable), so —
// exactly like the real CLI, which calls `Utc::now()` per verb — give
// each event a strictly increasing timestamp.
let base = Utc::now();
let at = |i: i64| base + chrono::Duration::microseconds(i);
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
let n1 = NodeId::seq(1);
let n2 = NodeId::seq(2);
{
let mut s = Store::open(&cli_root).unwrap();
s.record(Event::IdeaSubmitted {
id: idea.clone(),
source: "cli".into(),
text: "ship E4".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: at(0),
})
.unwrap();
s.record(Event::PlanCreated {
id: plan.clone(),
idea_id: idea.clone(),
summary: "funnel store split".into(),
planner: "cli".into(),
ts: at(1),
})
.unwrap();
s.record(Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: at(2),
})
.unwrap();
s.record(Event::NodeAdded {
plan_id: plan.clone(),
node_id: n1.clone(),
kind: "code:write".into(),
params: serde_json::Map::new(),
targets: vec!["src/funnel/store.rs".into()],
prompt_excerpt: None,
ts: at(3),
})
.unwrap();
s.record(Event::NodeAdded {
plan_id: plan.clone(),
node_id: n2.clone(),
kind: "test:write".into(),
params: serde_json::Map::new(),
targets: vec![],
prompt_excerpt: None,
ts: at(4),
})
.unwrap();
// n2 depends on n1.
s.record(Event::EdgeAdded {
plan_id: plan.clone(),
from_node: n1.clone(),
to_node: n2.clone(),
ts: at(5),
})
.unwrap();
} // writer drops → exclusive catalog lock released
// --- SERVER/VIZ READ PATH: resolve the SAME way the server does
// (explicit > local_path > home-derived default) and replay. Use the
// lock-tolerant read open, exactly like the server's `Funnel.Show`.
let server_root = Store::resolve_root(&ws, "store", Some(&root));
assert_eq!(server_root, cli_root, "server must resolve the CLI's root");
let mut s2 = Store::open_read_only(&server_root).unwrap();
// The idea is there.
assert!(s2.funnel.ideas.contains_key(&idea));
// The plan matches EXACTLY — id, idea linkage, status, summary.
let p = s2.funnel.plans.get(&plan).expect("plan must round-trip");
assert_eq!(p.idea_id, idea);
assert_eq!(p.summary, "funnel store split");
assert_eq!(p.status, PlanStatus::Active);
// Both nodes round-trip with the right kind/targets.
assert_eq!(p.nodes.len(), 2, "both nodes must persist");
let pn1 = p.nodes.get(&n1).expect("n1");
assert_eq!(pn1.kind, "code:write");
assert_eq!(pn1.targets, vec!["src/funnel/store.rs".to_string()]);
let pn2 = p.nodes.get(&n2).expect("n2");
assert_eq!(pn2.kind, "test:write");
// Replay leaves both nodes `Pending` (readiness is a derived, in-memory
// projection via `promote_ready`, never persisted) — assert that exact
// persisted shape.
assert_eq!(pn1.status, NodeStatus::Pending);
assert_eq!(pn2.status, NodeStatus::Pending);
// The dependency edge round-trips exactly.
assert_eq!(p.edges.len(), 1, "exactly one edge");
assert!(p.edges.contains(&(n1.clone(), n2.clone())), "n1→n2 edge");
// And the derived readiness the viz computes is correct from the
// round-tripped graph: n1 (no deps) → Ready, n2 (waits on n1) stays
// Pending until n1 is Done.
s2.funnel.promote_ready();
let p = s2.funnel.plans.get(&plan).unwrap();
assert_eq!(p.nodes.get(&n1).unwrap().status, NodeStatus::Ready);
assert_eq!(p.nodes.get(&n2).unwrap().status, NodeStatus::Pending);
}
#[test]
fn roundtrip_idea_then_replay() {
let dir = tempdir().unwrap();
let root = dir.path().join("wh");
{
let mut s = Store::open(&root).unwrap();
s.record(Event::IdeaSubmitted {
id: IdeaId::seq(1),
source: "test".into(),
text: "first idea".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: Utc::now(),
})
.unwrap();
assert_eq!(s.funnel.ideas.len(), 1);
}
let s2 = Store::open(&root).unwrap();
assert_eq!(
s2.funnel.ideas.len(),
1,
"ideas should persist across reopen"
);
assert!(s2.funnel.ideas.contains_key(&IdeaId::seq(1)));
}
/// An uncontended read-only open is a *normal writable* open (it got the
/// real lock) — only a copied-aside snapshot is flagged read-only. This is
/// the path `Funnel.Show` takes when no one else holds the funnel root.
#[test]
fn read_only_open_uncontended_is_writable() {
let dir = tempdir().unwrap();
let root = dir.path().join("wh");
let s = Store::open_read_only(&root).unwrap();
assert!(
!s.is_read_only(),
"uncontended open should not be a snapshot"
);
}
/// While one writer holds the exclusive catalog lock, a second
/// `open_read_only` must NOT fail (the bug behind the red Funnel tab) — it
/// falls back to a read snapshot that replays the same events. Writes to the
/// snapshot are refused rather than silently lost.
#[test]
fn read_only_open_while_locked_yields_snapshot() {
let dir = tempdir().unwrap();
let root = dir.path().join("wh");
// Writer holds the exclusive lock and records one idea.
let mut writer = Store::open(&root).unwrap();
writer
.record(Event::IdeaSubmitted {
id: IdeaId::seq(1),
source: "test".into(),
text: "locked idea".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: Utc::now(),
})
.unwrap();
// Second open while the writer still holds the lock: must succeed via
// the read-only snapshot and see the recorded idea.
let mut snap = Store::open_read_only(&root).unwrap();
assert!(
snap.is_read_only(),
"contended open must be a read-only snapshot"
);
assert!(snap.funnel.ideas.contains_key(&IdeaId::seq(1)));
// Writes to the snapshot are refused, never silently dropped.
let err = snap
.record(Event::IdeaSubmitted {
id: IdeaId::seq(2),
source: "test".into(),
text: "should be rejected".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: Utc::now(),
})
.unwrap_err();
assert!(format!("{err:#}").contains("read-only snapshot"));
}
}