Port live DDIR server to current interpreter

Cargo.toml

@@ -5,6 +5,7 @@ members = [
     "dogsdogsdogs",
     "experiments",
     "interactive",
+    "interactive/server",
     #"tpchlike",
     #"doop",
     "mdbook",

interactive/examples/ddir_server.rs

@@ -125,6 +125,16 @@ fn parse_command(line: &str) -> Result<Command, String> {
             Ok(Command::Feed { prog, input, key, val, time, diff })
         }
         "tick" => Ok(Command::Tick),
+        "bind" | "unbind" if toks.len() == 4 => {
+            let trace = toks[1].to_string();
+            let prog = toks[2].to_string();
+            let input: usize = toks[3].parse().map_err(|_| format!("{}: <in#> must be a number, got {:?}", toks[0], toks[3]))?;
+            if toks[0] == "bind" {
+                Ok(Command::Bind { trace, prog, input })
+            } else {
+                Ok(Command::Unbind { trace, prog, input })
+            }
+        }
         "drop" if toks.len() == 2 => Ok(Command::Drop { name: toks[1].to_string() }),
         "peek" if toks.len() == 2 || toks.len() == 3 => {
             let trace = toks[1].to_string();
@@ -146,6 +156,8 @@ fn print_help() {
     println!("  install <name> <file>");
     println!("  feed <prog> <in#> <value> [val=<value>] [time=<t>] [diff=<int>]");
     println!("  tick");
+    println!("  bind <trace> <prog> <in#>    (feed the trace's changes back in, each tick)");
+    println!("  unbind <trace> <prog> <in#>");
     println!("  drop <name>");
     println!("  peek <trace> [key]");
     println!("  list");
@@ -177,6 +189,16 @@ fn dispatch(cmd: &Command, server: &mut Server, worker: &mut Worker) -> bool {
             Ok(()) => if w0 { println!("dropped {:?}", name); },
             Err(e) => if w0 { println!("error: {}", e); },
         },
+        // Collective: the tap dataflow is built on every worker (each sees
+        // its shard, so the union delivers the delta exactly once).
+        Command::Bind { trace, prog, input } => match server.bind(worker, trace, prog, *input) {
+            Ok(()) => if w0 { println!("bound {:?} -> {:?} input {}", trace, prog, input); },
+            Err(e) => if w0 { println!("error: {}", e); },
+        },
+        Command::Unbind { trace, prog, input } => match server.unbind(worker, trace, prog, *input) {
+            Ok(()) => if w0 { println!("unbound {:?} -> {:?} input {}", trace, prog, input); },
+            Err(e) => if w0 { println!("error: {}", e); },
+        },
         // Collective: every worker imports its shard; `peek` gathers to worker 0
         // (which prints) and reports an error there if the trace is unknown.
         Command::Peek { trace, key } => {

interactive/server/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "ddir-server"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+description = "Interactive differential dataflow server: hold named arrangements live across DDIR program installs and drops."
+publish = false
+
+[[bin]]
+name = "ddir_server"
+path = "src/main.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+differential-dataflow = { workspace = true }
+timely = { workspace = true }
+interactive = { path = ".." }
+diagnostics = { path = "../../diagnostics" }
+tungstenite = "0.26"

interactive/server/README.md

@@ -0,0 +1,105 @@
+# Live DDIR server
+
+One long-running timely worker hosts interpreted DDIR dataflows through a
+load-run-drop lifecycle. Programs share results by name — each may import
+collections that others export — and clients follow along over TCP,
+WebSocket, or stdin.
+
+Run `cargo run -p ddir-server`, then open `interactive/server/console.html` or
+connect a line-oriented client to TCP port 7777. The same protocol is available
+over WebSocket on port 7778. Set `DDIR_BIND`, `DDIR_WS_BIND`, or
+`DDIR_TICK_MS` to change those defaults; `DDIR_TICK_MS=0` disables automatic
+progress while subscriptions are active. The current `diagnostics` crate is
+connected on `DDIR_DIAG_PORT` (default 51371).
+
+Every request can begin with an arbitrary request id. If omitted, the server
+generates one. Responses are `<id> data ...`, followed by `<id> ok ...` or
+`<id> err ...`. A `tail` remains active after its `ok` and ends when stopped.
+Between commands, blank lines and `#` comment lines are skipped, so command
+scripts can be piped to stdin (see `demo/`).
+
+The useful commands are `load`, `drop`, `list`, `feed`, `bind`, `unbind`,
+`peek`, `tail`, `stop`, `tick`, and `exit`. `load` accepts an inline
+pipe-syntax program:
+
+    load graph begin
+    let edges = import "random:nodes=8,edges=12,seed=1,churn=1";
+    export "graph.edges" = edges;
+    graph end-load
+    tail graph.edges
+
+A binding may also be spelled as a call, so
+`edges=random(seed=1,arity=2,range=8,count=12,churn=1)` redirects the local
+import named `edges` to the same content-addressed source as the
+`random:...` form. Such a source is deterministic: it begins with a
+fixed-size window into an infinite hash-derived sequence and replaces
+`churn` rows on every tick.
+
+Automatic ticking happens only while at least one tail is active. This makes a
+live demonstration move without assigning input durability semantics to DDIR.
+Explicit `tick [n]` remains available for reproducible sessions. Treat
+auto-tick as demo furniture rather than a design commitment: as specified,
+observation advances time (an observer effect), and the alternative — that a
+watcher must be present to move things along, by ticking or by running a
+metronome client whose ticks are ordinary logged commands — may be the better
+design once the server has real tenants.
+
+## Writes: `feed`
+
+    feed <prog> <in#> <key> [val=<v>] [time=<t>] [diff=<d>]
+
+pushes one update into a loaded program's positional input, exactly as in the
+`ddir_server` example (`1,2` → a tuple; `_` → unit; a closed scalar term such
+as `inject(2,tuple(3,4))` for ADT-shaped rows).
+
+The stance on contention: **writes are open; policy lives in the dataflow**.
+The server does not decide who may write what. Cooperating clients follow a
+simple protocol — include your id and an ordering epoch in the data — and
+programs resolve races over those facts (first-claim-wins is a `min` over
+`(epoch, id)`, see `demo/claims.txt`; full optimistic transactions are a
+recursive view, see `demo/txn.txt`). Racing writes settle identically on
+every replay. Identity is convention, not enforcement: we are not defending
+against adversarial clients yet, and server-side attribution is deliberately
+deferred until a deployment needs it.
+
+## Feedback: `bind`
+
+    bind <trace> <prog> <in#>        unbind <trace> <prog> <in#>
+
+From then on, every `tick` delivers the trace's *changes* into that input at
+the next epoch, so the input mirrors the trace one epoch delayed. This is the
+write path for *programs*: an installed dataflow can act on the world — or on
+itself — with no client in the loop, one well-founded recursion step per tick.
+
+The state-machine idiom (see `demo/counter.txt` and the `server_bind` tests):
+give the program a seed input and a dedicated feedback input,
+
+    let state = seed + feedback;
+
+and bind the export `f(state) + (seed | negate)` to the feedback input; then
+`state(t) = f(state(t-1))`, while later seed feeds still inject as
+perturbations. A bound source cannot be dropped (it holds an importer), nor
+can the bound target (unbind first).
+
+## One gate
+
+Loads are cheap to request and costly to render, so intake is bounded:
+`DDIR_MAX_PROGRAM_BYTES` (default 65536) — a larger `load` body is swallowed
+and rejected with one error, before parsing. This is transport self-defense,
+not semantics. There are no ownership or quota gates: sessions are trusted,
+and admission policy (auth, quotas, rate limits) belongs in a fronting proxy
+if a deployment ever needs one.
+
+## Demos
+
+    cargo run -p ddir-server --release
+    # then, or piped straight to stdin:
+    ./target/release/ddir_server < interactive/server/demo/counter.txt
+    ./target/release/ddir_server < interactive/server/demo/claims.txt
+    ./target/release/ddir_server < interactive/server/demo/txn.txt
+    python3 interactive/server/demo/two_sessions.py   # races + size gate over TCP
+
+`load --explain` and `query` are reserved but unimplemented: explanation
+support belongs on the scope-tree explanation machinery, and until that lands
+the server reports an error rather than giving those commands an improvised
+meaning.