Ddir deduplication

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<cmd::Cmd, String>),
+    control: &ControlHandle,
+    resp_tx: &Sender<String>,
+    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<R: BufRead, W: Write + Send + 'static>(
         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;
+                            }
                         }
                     }
                 }

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;

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<Proj = M::Proj, Pred = M::Pred> {
 // 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>) -> 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>) -> 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>) -> 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 { Value::Tuple(xs) }

interactive/src/explain/mod.rs

@@ -1050,35 +1050,9 @@ mod tests {
 
     fn parse(src: &str) -> Vec<crate::parse::Stmt> { 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::<usize>()

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];

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<ServerTrace> {
         self.traces.get(&canonical_source_name(name)).cloned()