From 71677cd92783ec45d8e6f6cd2740828b6e56d330 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 20 Sep 2026 14:08:14 -0400 Subject: [PATCH 1/4] Include the SCC example instead of copying it into tests `explain`'s `SCC` fixture and `lower`'s `lowers_scc_depth_two` source were byte-identical copies of `examples/programs/scc.ddp`, maintained by hand. Both now `include_str!` the example, so the fixture cannot drift from the program the rest of the tree runs. `tests/explain.rs`'s `SCC_ROW` stays inline: it is a different program (its export is the edge set, not the aggregate), as its comment says. Co-Authored-By: Claude Opus 5 (1M context) --- interactive/src/explain/mod.rs | 32 +++----------------------------- interactive/src/lower.rs | 30 +----------------------------- 2 files changed, 4 insertions(+), 58 deletions(-) diff --git a/interactive/src/explain/mod.rs b/interactive/src/explain/mod.rs index 8b6385d61..d31054d05 100644 --- a/interactive/src/explain/mod.rs +++ b/interactive/src/explain/mod.rs @@ -1050,35 +1050,9 @@ mod tests { fn parse(src: &str) -> Vec { crate::parse::pipe::parse(src) } - const SCC: &str = r#" - let edges = input 0 | key($0[0] ; $0[1]); - let trans = edges | key($1 ; $0); - outer: { - let scc = edges + trim; - fwd: { - let nodes = edges | key($1 ; $1) | enter_at($1[0]); - let labels = proposals + nodes | min; - var proposals = labels | join(scc, ($2 ; $1)); - } - let trim_fwd = edges - | join(fwd::labels, ($1 ; $0, $2)) - | join(fwd::labels, ($0 ; $1, $2)) - | filter($1[1] == $1[2]) - | key($0 ; $1[0]); - bwd: { - let nodes = trans | key($1 ; $1) | enter_at($1[0]); - let labels = proposals + nodes | min; - var proposals = labels | join(trim_fwd, ($2 ; $1)); - } - let trim_bwd = trans - | join(bwd::labels, ($1 ; $0, $2)) - | join(bwd::labels, ($0 ; $1, $2)) - | filter($1[1] == $1[2]) - | key($0 ; $1[0]); - var trim = trim_bwd - edges; - } - export "result" = outer::scc | map(;) | arrange | inspect(total); - "#; + /// The checked-in example, so the fixture cannot drift from the program + /// the rest of the tree runs. + const SCC: &str = include_str!("../../examples/programs/scc.ddp"); fn vars_total(s: &Scope) -> usize { s.vars.len() + s.items.iter().map(|i| match i { Item::Sub(c) => vars_total(c), _ => 0 }).sum::() diff --git a/interactive/src/lower.rs b/interactive/src/lower.rs index 93641d522..4469befa1 100644 --- a/interactive/src/lower.rs +++ b/interactive/src/lower.rs @@ -391,35 +391,7 @@ mod tree_tests { #[test] fn lowers_scc_depth_two() { - let src = r#" - let edges = input 0 | key($0[0] ; $0[1]); - let trans = edges | key($1 ; $0); - outer: { - let scc = edges + trim; - fwd: { - let nodes = edges | key($1 ; $1) | enter_at($1[0]); - let labels = proposals + nodes | min; - var proposals = labels | join(scc, ($2 ; $1)); - } - let trim_fwd = edges - | join(fwd::labels, ($1 ; $0, $2)) - | join(fwd::labels, ($0 ; $1, $2)) - | filter($1[1] == $1[2]) - | key($0 ; $1[0]); - bwd: { - let nodes = trans | key($1 ; $1) | enter_at($1[0]); - let labels = proposals + nodes | min; - var proposals = labels | join(trim_fwd, ($2 ; $1)); - } - let trim_bwd = trans - | join(bwd::labels, ($1 ; $0, $2)) - | join(bwd::labels, ($0 ; $1, $2)) - | filter($1[1] == $1[2]) - | key($0 ; $1[0]); - var trim = trim_bwd - edges; - } - export "result" = outer::scc | map(;) | arrange | inspect(total); - "#; + let src = include_str!("../examples/programs/scc.ddp"); let prog = lower_tree(parse(src)); assert_eq!(subs(&prog.root).len(), 1); // outer let outer = subs(&prog.root)[0]; From 9d58ada93bc7082833f7cc0b82fa359254f3687b Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 20 Sep 2026 14:08:14 -0400 Subject: [PATCH 2/4] Share one in-memory Dataflow between the contract modules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `value_contract` and `backstop` each defined the same `Mem` — a `Dataflow` over `Vec<(Value, Value)>` running projections and predicates through `ir::eval` with a nested-loop join — once formatted long and once short. They now share a single definition. `nested_contract` keeps its own: that one is a genuinely different model, with its own `Proj`/`Pred`. The module header also claimed the flat `[i64]` model "lives in `explain.rs` and reuses `folded`". There is no `explain.rs`, and `folded` is retired; the header now says `Val` is the only model the crate evaluates. Co-Authored-By: Claude Opus 5 (1M context) --- interactive/src/explain/decouple.rs | 111 +++++++++++++--------------- 1 file changed, 50 insertions(+), 61 deletions(-) diff --git a/interactive/src/explain/decouple.rs b/interactive/src/explain/decouple.rs index 0afdbd96c..217ffd103 100644 --- a/interactive/src/explain/decouple.rs +++ b/interactive/src/explain/decouple.rs @@ -10,8 +10,9 @@ //! the scope builder `Sb` for real explain. //! //! Changing the data model means reimplementing [`RowModel`]; the rules and the -//! orchestration are untouched. The flat `[i64]` impl lives in `explain.rs` and -//! reuses `folded` for the time-filter/strip algebra. +//! orchestration are untouched. [`crate::explain::Val`] is the only model the +//! crate evaluates; the flat `[i64]` model this was factored out for, and the +//! `folded` algebra it used, are both retired. //! //! ## The demand envelope //! @@ -247,6 +248,49 @@ where M: RowModel, D: Dataflow { // against a nested `Value`-shaped `RowModel` — the shape the real `explain::Val` // model uses — so it remains the runnable spec for the reverse rules. +/// In-memory [`Dataflow`] over `Vec<(Value, Value)>`, shared by the contract +/// modules below: projections and predicates run through the `Term` +/// interpreter (`ir::eval`), and `join` is a nested-loop equi-join on the key. +#[cfg(test)] +mod mem { + use super::Dataflow; + use crate::ir::{eval, Projection, Term, Value}; + + pub type Coll = Vec<(Value, Value)>; + pub struct Mem; + + impl Dataflow for Mem { + type Handle = Coll; + type Proj = Projection; + type Pred = Term; + fn project(&mut self, c: &Coll, p: Projection) -> Coll { + c.iter().map(|(k, v)| { + let mut e = vec![k.clone(), v.clone()]; + (eval(&p.key, &mut e), eval(&p.val, &mut e)) + }).collect() + } + fn filter(&mut self, c: &Coll, p: Term) -> Coll { + c.iter().filter(|(k, v)| { + let mut e = vec![k.clone(), v.clone()]; + eval(&p, &mut e).truthy() + }).cloned().collect() + } + fn join(&mut self, l: &Coll, r: &Coll, p: Projection) -> Coll { + let mut out = Vec::new(); + for (lk, lv) in l { + for (rk, rv) in r { + if lk == rk { + let mut e = vec![lk.clone(), lv.clone(), rv.clone()]; + out.push((eval(&p.key, &mut e), eval(&p.val, &mut e))); + } + } + } + out + } + fn concat(&mut self, cs: Vec) -> Coll { cs.into_iter().flatten().collect() } + } +} + #[cfg(test)] mod nested_contract { //! Proof that the trait is model-agnostic: a second `RowModel` over a @@ -502,45 +546,11 @@ mod value_contract { use super::*; use crate::explain::Val; - use crate::ir::{eval, Value}; + use super::mem::{Coll, Mem}; + use crate::ir::Value; use crate::ir::{Projection, Term}; type Row = Value; - type Coll = Vec<(Row, Row)>; - - /// In-memory dataflow: applies projections/predicates with the `Term` - /// interpreter (`ir::eval`); `join` is a nested-loop equi-join on the key. - struct Mem; - impl Dataflow for Mem { - type Handle = Coll; - type Proj = Projection; - type Pred = Term; - fn project(&mut self, c: &Coll, p: Projection) -> Coll { - c.iter().map(|(k, v)| { - let mut e = vec![k.clone(), v.clone()]; - (eval(&p.key, &mut e), eval(&p.val, &mut e)) - }).collect() - } - fn filter(&mut self, c: &Coll, p: Term) -> Coll { - c.iter().filter(|(k, v)| { - let mut e = vec![k.clone(), v.clone()]; - eval(&p, &mut e).truthy() - }).cloned().collect() - } - fn join(&mut self, l: &Coll, r: &Coll, p: Projection) -> Coll { - let mut out = Vec::new(); - for (lk, lv) in l { - for (rk, rv) in r { - if lk == rk { - let mut e = vec![lk.clone(), lv.clone(), rv.clone()]; - out.push((eval(&p.key, &mut e), eval(&p.val, &mut e))); - } - } - } - out - } - fn concat(&mut self, cs: Vec) -> Coll { cs.into_iter().flatten().collect() } - } /// A 1-field key `(n)`. fn key(n: i64) -> Row { Value::Tuple(vec![Value::Int(n)]) } @@ -626,32 +636,11 @@ mod backstop { //! endpoint, so `RESIDUAL` is the whole input (here, the list). This pins //! "the gap is closable" before the real rule + wiring are built. + use super::mem::{Coll, Mem}; use super::*; - use crate::ir::{eval, Value}; + use crate::ir::Value; use crate::ir::{Projection, Term}; - type Coll = Vec<(Value, Value)>; - struct Mem; - impl Dataflow for Mem { - type Handle = Coll; - type Proj = Projection; - type Pred = Term; - fn project(&mut self, c: &Coll, p: Projection) -> Coll { - c.iter().map(|(k, v)| { let mut e = vec![k.clone(), v.clone()]; (eval(&p.key, &mut e), eval(&p.val, &mut e)) }).collect() - } - fn filter(&mut self, c: &Coll, p: Term) -> Coll { - c.iter().filter(|(k, v)| { let mut e = vec![k.clone(), v.clone()]; eval(&p, &mut e).truthy() }).cloned().collect() - } - fn join(&mut self, l: &Coll, r: &Coll, p: Projection) -> Coll { - let mut out = Vec::new(); - for (lk, lv) in l { for (rk, rv) in r { - if lk == rk { let mut e = vec![lk.clone(), lv.clone(), rv.clone()]; out.push((eval(&p.key, &mut e), eval(&p.val, &mut e))); } - }} - out - } - fn concat(&mut self, cs: Vec) -> Coll { cs.into_iter().flatten().collect() } - } - fn int(n: i64) -> Value { Value::Int(n) } fn list(xs: &[i64]) -> Value { Value::List(xs.iter().map(|&n| int(n)).collect()) } fn tup(xs: Vec) -> Value { Value::Tuple(xs) } From 9cbf3eb4683e5dac61afd7ae896d55b4fce8fac0 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 20 Sep 2026 14:08:14 -0400 Subject: [PATCH 3/4] Give the two session loops one dispatch step MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stdin/TCP reader and the WebSocket reader each built the same prepare-and-send sequence: check for `exit`, `prepare`, build a `Request`, hand it to the control channel. They now call one `dispatch` returning `Continue`/`Exit`/`Disconnected`, and each loop decides what that means — the stdin/TCP reader stops, the WebSocket reader sets `should_exit` and keeps draining the current message. Behaviour is unchanged on both paths. Co-Authored-By: Claude Opus 5 (1M context) --- interactive/server/src/main.rs | 75 +++++++++++++++++++++------------- 1 file changed, 46 insertions(+), 29 deletions(-) diff --git a/interactive/server/src/main.rs b/interactive/server/src/main.rs index ad10d7a6c..b2654b462 100644 --- a/interactive/server/src/main.rs +++ b/interactive/server/src/main.rs @@ -64,6 +64,40 @@ impl ControlHandle { } } +/// What a session should do after handing one parsed line to the workers. +enum Dispatch { + /// Keep reading this session. + Continue, + /// The client asked to exit. + Exit, + /// The worker group is gone; nothing more can be sent. + Disconnected, +} + +/// Prepare one parsed line and hand it to the worker group. The two session +/// loops (stdin/TCP and WebSocket) differ only in what they do with the +/// answer, so the protocol step itself lives here once. +fn dispatch( + parsed: (cmd::ReqId, Result), + control: &ControlHandle, + resp_tx: &Sender, + connection_id: ConnectionId, +) -> Dispatch { + let (reqid, kind) = parsed; + let is_exit = matches!(kind, Ok(cmd::Cmd::Exit)); + let kind = kind.and_then(prepare); + let req = Request { + reqid, + kind, + resp: resp_tx.clone(), + connection_id, + }; + if control.send(ControlEvent::Request(req)).is_err() { + return Dispatch::Disconnected; + } + if is_exit { Dispatch::Exit } else { Dispatch::Continue } +} + /// Process-wide allocator for per-session ids. Stdin uses 0; /// subsequent connections get 1, 2, 3, .... static NEXT_CONNECTION_ID: AtomicU64 = AtomicU64::new(1); @@ -246,22 +280,12 @@ fn run_session( let Ok(line) = line else { break; }; - if let Some((reqid, kind)) = parser.feed(&line) { - let is_exit = matches!(kind, Ok(cmd::Cmd::Exit)); - let kind = kind.and_then(prepare); - let req = Request { - reqid, - kind, - resp: resp_tx.clone(), - connection_id, - }; - if control.send(ControlEvent::Request(req)).is_err() { - break; - } + if let Some(parsed) = parser.feed(&line) { // For stdin, `exit` terminates the whole server; for TCP, it // terminates only this session. Either way the reader stops. - if is_exit { - break; + match dispatch(parsed, &control, &resp_tx, connection_id) { + Dispatch::Continue => {} + Dispatch::Exit | Dispatch::Disconnected => break, } } } @@ -369,21 +393,14 @@ fn run_ws_session(stream: TcpStream, control: ControlHandle) -> Result<(), tungs if line.trim().is_empty() && !parser.awaiting_body() { continue; } - if let Some((reqid, kind)) = parser.feed(line) { - let is_exit = matches!(kind, Ok(cmd::Cmd::Exit)); - let kind = kind.and_then(prepare); - let req = Request { - reqid, - kind, - resp: resp_tx.clone(), - connection_id, - }; - if control.send(ControlEvent::Request(req)).is_err() { - should_exit = true; - break; - } - if is_exit { - should_exit = true; + if let Some(parsed) = parser.feed(line) { + match dispatch(parsed, &control, &resp_tx, connection_id) { + Dispatch::Continue => {} + Dispatch::Exit => should_exit = true, + Dispatch::Disconnected => { + should_exit = true; + break; + } } } } From 82c41946b7f9ebbf50842610cc289ef03c376311 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 20 Sep 2026 14:08:14 -0400 Subject: [PATCH 4/4] Delete Server::has_trace and a driver that does not exist `has_trace` has no callers: its identifier appears exactly once in the whole of `interactive/`, at its own definition. `backend`'s header still advertised "the example binaries, the server, and a wasm front-end" as drivers. There is no wasm anywhere in the tree and the example driver is retired (`server/src/main.rs:30` says so); the server is the one driver. Co-Authored-By: Claude Opus 5 (1M context) --- interactive/src/backend/mod.rs | 3 +-- interactive/src/server.rs | 5 ----- 2 files changed, 1 insertion(+), 7 deletions(-) diff --git a/interactive/src/backend/mod.rs b/interactive/src/backend/mod.rs index 897d0d075..622f11ff7 100644 --- a/interactive/src/backend/mod.rs +++ b/interactive/src/backend/mod.rs @@ -7,8 +7,7 @@ //! everything else (enter/leave/concat/feedback) is DD's own container-generic //! machinery. Time is always `ir::Time`; only the container varies. //! -//! The example binaries, the server, and a wasm front-end are thin drivers -//! that pick a backend and call [`render_tree`]. +//! The server is a thin driver: it picks a backend and calls [`render_tree`]. pub mod vec; pub mod corgi; diff --git a/interactive/src/server.rs b/interactive/src/server.rs index 1149189cf..0bd6473de 100644 --- a/interactive/src/server.rs +++ b/interactive/src/server.rs @@ -390,11 +390,6 @@ impl Server { self.epoch } - /// Whether a trace is registered under `name`. - pub fn has_trace(&self, name: &str) -> bool { - self.traces.contains_key(name) - } - /// Clone a trace reader for a transient peek or subscription dataflow. pub fn trace(&self, name: &str) -> Option { self.traces.get(&canonical_source_name(name)).cloned()