fix(incremental): a panicking turn buries its unit instead of hanging the pool; pool workers counted and stopped on drop; two untested more-clauses settled

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze

context/code/improvise-incremental.md

@@ -511,7 +511,9 @@ Multi-threading here is **scheduling**. Each sub-circuit stays synchronous and s
 (`zset.md`, "Threads"), the only thing that crosses threads is an edge (`zset.md`, "Edges"), and
 a small executor decides which thread runs which sub-circuit next. This is a ruling recorded here,
 not an ADR amendment. Native threads are assumed. `sched.rs` and `staged.rs` compile for wasm32
-(a `Mutex` and a `Condvar`), and `Pool`, the only thing that spawns a thread, is
+(a `Mutex` and a `Condvar`), but `improvise-command` depends on this crate, so the module is in
+the browser bundle: `Pool`, the only thing that spawns a thread, and the blocking calls
+(`claim_blocking`, `wait_idle`, which `Condvar::wait` would make panic there) are
 `cfg(not(target_arch = "wasm32"))`. Nothing here runs on wasm.
 
 1. **A unit is a sub-circuit with the edge ends around it** (`Unit`: `turn(budget)` and
@@ -530,8 +532,15 @@ not an ADR amendment. Native threads are assumed. `sched.rs` and `staged.rs` com
    queue and the slots, and a turn takes microseconds beside it.
 4. **Fairness is the budget and the back of the queue.** A turn handles at most `budget` frames
    (`EdgeSource::poll(max)`), then says `more`, and `finish` queues it behind the units already
-   waiting. A full edge (`Flow::Backlogged`) is also `more`: the unit retries, and coalesces
-   meanwhile. The resumable `step(deadline)` predicate of `Rebuild` is for a sliced snapshot load,
+   waiting. A full edge is **not** `more`: the sender would spin through the queue while its
+   consumer is not running (`a_full_edge_does_not_make_its_sender_spin`). The consumer that frees
+   room reports progress and kicks its producer back, so a backlogged sender runs when it can
+   send. For the same reason a middle unit (the aggregate) needs no `more` at its budget: its
+   consumer's progress kicks it. The tail (the projection) has no consumer to kick it back, so it
+   says `more` when it read a full budget (`a_unit_whose_kicks_coalesced_still_drains_its_edge`,
+   because kicks coalesce into one queue entry). Both clauses were first written untested; a
+   review injected their removal and everything passed, which is how the first was found to be a
+   spin and the second redundant. The resumable `step(deadline)` predicate of `Rebuild` is for a sliced snapshot load,
    and this example has none.
 5. **The executor spawns nothing.** `claim` and `finish` are what a worker calls. `Pool` is N
    threads looping on them; the tests' scheduler calls them in an order a seed chooses, with up to N
@@ -548,7 +557,21 @@ not an ADR amendment. Native threads are assumed. `sched.rs` and `staged.rs` com
    a total per row). The incremental backend (`Incremental`) keeps its groups in its own shapes
    and does not run on a `Circuit`, so the stage is the same computation built as zset nodes rather than a
    split of `Incremental` itself. Running `Incremental` on circuits is a later step.
-8. **Not done.** A work-stealing or priority scheme (one FIFO queue); a unit with a `Rebuild` job
-   sliced by the deadline predicate; a timeout that calls `connection_lost` on the edges (the
-   edges here never lose a frame); a panic in a turn (the executor's mutex poisons and the pool
-   reports it).
+8. **A panic in a turn kills that unit and nothing else.** A turn runs outside the lock, so
+   nothing is poisoned and, in the first version, the pool waited for ever on a unit that never
+   came back. Now the worker catches the unwind and calls `Executor::abandon`: ownership is
+   released, the unit is dead (never run or kicked again), the panic is recorded
+   (`Executor::panics`, `UnitPanic { unit, message }`), and neighbours go on. `wait_idle` and
+   `Pool::finish` return `Err(UnitPanic)` for the first one once the rest is quiet. A unit that
+   cannot name its neighbours is buried the same way. A kick of a unit that does not exist panics
+   before it takes the lock (`no such unit`), so a bad index cannot poison the executor; the
+   `lock()` message now names the only thing that can poison it, a panic in the executor itself.
+9. **Workers are counted and a dropped `Pool` stops them.** `Executor::workers()` counts live
+   worker threads, so a worker that leaves on a spurious wake is visible
+   (`every_worker_stays_through_wakes_with_nothing_queued`); `Drop for Pool` shuts down and joins
+   (`a_pool_that_is_dropped_stops_its_threads`), discarding what is queued. `claim_blocking`
+   returns `None` as soon as `shutdown` is set. The real-thread tests run behind a watchdog
+   (`tests/support/watchdog.rs`), so a lost `notify_all` fails by name and not by hanging.
+10. **Not done.** A work-stealing or priority scheme (one FIFO queue); a unit with a `Rebuild`
+    job sliced by the deadline predicate; a timeout that calls `connection_lost` on the edges (the
+    edges here never lose a frame); restarting a dead unit.

context/test-evidence/improvise-incremental.md

@@ -985,3 +985,65 @@ handed back. Rivals S1 and S2 fail it.
 
 Four kicking threads and four workers over one unit that asserts no two turns overlap. Rival S1
 fails it.
+
+### a_panicking_turn_does_not_hang_finish
+
+A unit whose turn panics, on a two-thread pool: `Pool::finish` hung for ever (a 3-second watchdog
+reproduced it before the fix). The unwind is now caught and the unit buried. Rival P1 (the dead
+unit still counted as running) fails it by the watchdog.
+
+### a_panicking_unit_is_reported_and_its_neighbours_still_run
+
+`finish` returns `UnitPanic { unit: 0, message }` containing the payload, and a live unit ran.
+Rival P1 fails it.
+
+### a_dead_unit_is_never_run_again
+
+After `abandon`, a kick does not queue the unit. Rival P2 (the `dead` check removed) fails it.
+
+### a_kick_of_a_unit_that_does_not_exist_panics_before_the_lock
+
+`kick(5)` names the missing unit. Rival P3 (no index check) fails it with an index error.
+
+### an_executor_survives_a_bad_kick
+
+After the bad `kick`, `idle()` works: the lock was not poisoned. Rival P3 fails it.
+
+### every_worker_stays_through_wakes_with_nothing_queued
+
+Four workers, 200 kicks, the worker count stays four and is zero after `finish`. Rival W1 (a
+worker returns on a wake with an empty queue) fails it. Rivals W3 and W4 (a lost `notify_all` in
+`kick` or `finish`) fail it by the watchdog.
+
+### a_pool_that_is_dropped_stops_its_threads
+
+Dropping a pool leaves no worker. Rival W2 (an empty `Drop`) fails it.
+
+### a_full_edge_does_not_make_its_sender_spin
+
+The aggregate and projection are held, the head's edge (capacity 1) fills, and the head must not
+be queued again. Rival M4 (`more` while backlogged) makes it ask for ever and fails it at 50
+turns.
+
+### a_unit_whose_kicks_coalesced_still_drains_its_edge
+
+Two commands cross before the projection's kicks are served, with a budget of 1. The projection
+must ask to run again; the second frame is otherwise never read. Rival M5 fails it (and
+`placed_circuits_equal_one_circuit_for_any_workers_and_order`). The same clause on the aggregate
+is redundant, because the projection's progress kicks it back; removing it passes everything, so
+it was deleted.
+
+### a_dag_on_real_threads_equals_one_circuit_and_idle_means_done
+
+The reviewer's seven-unit graph (fan-out, fan-in, fan-out) with random edge capacities and
+budgets, 1 to 8 workers, `wait_idle` checkpoints mid-stream that must equal the reference for the
+prefix submitted. Rivals W3 and W4 fail it by the watchdog.
+
+### submitter_threads_race_the_pool
+
+Four threads submit commands to the same pool; the result is the reference. Rivals W3 and W4 fail
+it by the watchdog.
+
+### a_dag_on_virtual_workers_in_any_order_equals_one_circuit
+
+The same graph under a random claim and hand-back order of 1 to 8 virtual workers.

context/test-evidence/zset.md

@@ -772,3 +772,9 @@ Rival: `unsafe impl Sync for Circuit {}` fails the build of `tests/threads.rs`.
 
 A sink over `Counted` keys clones each key once (the insert into its sum). Rival: a sink that
 clones its input piece first, as the reviewer injected, makes `N` more.
+
+### traffic_never_goes_down_across_a_restart_of_either_end
+
+`EdgeSink::restart` and `EdgeSource::restart` kept the frame counters at 0 (they rebuilt the
+struct), which could make a unit read "no progress" and lose a kick. Rival: either restart
+resetting its counter fails it.

crates/improvise-incremental/src/sched.rs

@@ -2,6 +2,7 @@
 //! Reasoning: `context/code/improvise-incremental.md`, "Scheduling".
 
 use std::collections::VecDeque;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Condvar, Mutex, MutexGuard};
 
 /// What one turn of a unit did.
@@ -9,7 +10,8 @@ use std::sync::{Condvar, Mutex, MutexGuard};
 pub struct Turn {
     /// A frame moved on one of its edges, so its neighbours may have work.
     pub progress: bool,
-    /// It stopped at its budget, or an edge was full: run it again.
+    /// It stopped at its budget with frames left: run it again. A full edge is not `more`; the
+    /// reader that frees room reports progress and kicks it.
     pub more: bool,
 }
 
@@ -28,6 +30,8 @@ struct Slot {
     running: bool,
     /// Kicked while running: queue it again when the owner hands it back.
     again: bool,
+    /// Its turn panicked: it is never run or kicked again.
+    dead: bool,
 }
 
 struct State {
@@ -35,6 +39,7 @@ struct State {
     queue: VecDeque<usize>,
     running: usize,
     shutdown: bool,
+    panics: Vec<UnitPanic>,
 }
 
 /// A unit a worker owns until it hands it back with `finish`.
@@ -53,17 +58,37 @@ impl Claimed {
     }
 }
 
+/// A unit whose turn panicked. It is dead: it is never run again, and its neighbours go on.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct UnitPanic {
+    pub unit: usize,
+    pub message: String,
+}
+
+/// The text of a panic payload.
+pub fn panic_text(payload: &(dyn std::any::Any + Send)) -> String {
+    payload
+        .downcast_ref::<String>()
+        .cloned()
+        .or_else(|| payload.downcast_ref::<&str>().map(|s| s.to_string()))
+        .unwrap_or_else(|| "a panic with a non-string payload".to_string())
+}
+
 /// The run queue and the ownership of every unit. It spawns nothing: workers (threads, or a test
-/// scheduler choosing an order) call `claim` and `finish`.
+/// scheduler choosing an order) call `claim` and `finish`, or `abandon` after a panic.
 pub struct Executor {
     state: Mutex<State>,
     wake: Condvar,
     budget: usize,
+    units: usize,
+    /// Worker threads a `Pool` has started and not yet seen leave.
+    workers: AtomicUsize,
 }
 
 impl Executor {
     /// `budget` is the frames a unit may handle per turn; zero is read as one.
     pub fn new(units: Vec<Box<dyn Unit>>, budget: usize) -> Self {
+        let count = units.len();
         Executor {
             state: Mutex::new(State {
                 slots: units
@@ -73,14 +98,18 @@ impl Executor {
                         queued: false,
                         running: false,
                         again: false,
+                        dead: false,
                     })
                     .collect(),
                 queue: VecDeque::new(),
                 running: 0,
                 shutdown: false,
+                panics: Vec::new(),
             }),
             wake: Condvar::new(),
             budget: budget.max(1),
+            units: count,
+            workers: AtomicUsize::new(0),
         }
     }
 
@@ -88,12 +117,29 @@ impl Executor {
         self.budget
     }
 
+    /// How many worker threads are alive.
+    pub fn workers(&self) -> usize {
+        self.workers.load(Ordering::SeqCst)
+    }
+
+    /// The units that have panicked, in order.
+    pub fn panics(&self) -> Vec<UnitPanic> {
+        self.lock().panics.clone()
+    }
+
+    /// Only a panic inside the executor's own code, under the lock, poisons it. A unit's turn
+    /// runs outside the lock, so a panic there is `abandon`ed instead.
     fn lock(&self) -> MutexGuard<'_, State> {
-        self.state.lock().expect("a unit panicked in a turn")
+        self.state
+            .lock()
+            .expect("the executor's lock was poisoned by a panic inside the executor")
     }
 
     fn kick_in(st: &mut State, id: usize) {
         let slot = &mut st.slots[id];
+        if slot.dead {
+            return;
+        }
         if slot.running {
             slot.again = true;
         } else if !slot.queued {
@@ -102,8 +148,14 @@ impl Executor {
         }
     }
 
+    #[track_caller]
+    fn exists(&self, id: usize) {
+        assert!(id < self.units, "zset sched: no such unit {id}");
+    }
+
     /// Marks `id` runnable: it has input, or the host wants it to look.
     pub fn kick(&self, id: usize) {
+        self.exists(id);
         Self::kick_in(&mut self.lock(), id);
         self.wake.notify_all();
     }
@@ -136,26 +188,41 @@ impl Executor {
         (n > 0).then(|| Self::take(&mut st, pick(n) % n))
     }
 
-    /// Like `claim` for a worker thread: waits for work, and is `None` after `shutdown`.
+    /// Like `claim` for a worker thread: waits for work, and is `None` once `shutdown` is called.
+    #[cfg(not(target_arch = "wasm32"))]
     pub fn claim_blocking(&self) -> Option<Claimed> {
         let mut st = self.lock();
         loop {
-            if !st.queue.is_empty() {
-                return Some(Self::take(&mut st, 0));
-            }
             if st.shutdown {
                 return None;
             }
-            st = self.wake.wait(st).expect("a unit panicked in a turn");
+            if !st.queue.is_empty() {
+                return Some(Self::take(&mut st, 0));
+            }
+            st = self
+                .wake
+                .wait(st)
+                .expect("the executor's lock was poisoned by a panic inside the executor");
         }
     }
 
-    /// Hands a unit back; queues it again if it asked for more or was kicked meanwhile, and
-    /// kicks its neighbours if the turn moved a frame.
+    /// Hands a unit back, queued again if it wants more or was kicked; progress kicks its
+    /// neighbours. A unit that cannot name them is buried.
     pub fn finish(&self, claimed: Claimed, turn: Turn) {
         let Claimed { id, unit } = claimed;
+        let named = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let n = unit.neighbours().to_vec();
+            assert!(
+                n.iter().all(|&u| u < self.units),
+                "zset sched: unit {id} names a neighbour that does not exist"
+            );
+            n
+        }));
+        let neighbours = match named {
+            Ok(n) => n,
+            Err(p) => return self.bury(id, unit, panic_text(&*p)),
+        };
         let mut st = self.lock();
-        let neighbours = unit.neighbours().to_vec();
         let slot = &mut st.slots[id];
         slot.unit = Some(unit);
         slot.running = false;
@@ -173,21 +240,46 @@ impl Executor {
         self.wake.notify_all();
     }
 
+    /// A unit's turn panicked: it is dead, ownership is released and the panic is recorded.
+    /// `wait_idle` and `Pool::finish` report it, and nothing waits for the dead unit.
+    pub fn abandon(&self, claimed: Claimed, message: String) {
+        let Claimed { id, unit } = claimed;
+        self.bury(id, unit, message);
+    }
+
+    fn bury(&self, id: usize, unit: Box<dyn Unit>, message: String) {
+        let mut st = self.lock();
+        let slot = &mut st.slots[id];
+        slot.running = false;
+        slot.dead = true;
+        slot.again = false;
+        st.running -= 1;
+        st.panics.push(UnitPanic { unit: id, message });
+        drop(st);
+        drop(unit);
+        self.wake.notify_all();
+    }
+
     /// Whether nothing is queued and nothing is running.
     pub fn idle(&self) -> bool {
         let st = self.lock();
         st.queue.is_empty() && st.running == 0
     }
 
-    /// Blocks until `idle`.
-    pub fn wait_idle(&self) {
+    /// Blocks until `idle`, then reports the first unit that panicked, if any.
+    #[cfg(not(target_arch = "wasm32"))]
+    pub fn wait_idle(&self) -> Result<(), UnitPanic> {
         let mut st = self.lock();
         while !(st.queue.is_empty() && st.running == 0) {
-            st = self.wake.wait(st).expect("a unit panicked in a turn");
+            st = self
+                .wake
+                .wait(st)
+                .expect("the executor's lock was poisoned by a panic inside the executor");
         }
+        st.panics.first().cloned().map_or(Ok(()), Err)
     }
 
-    /// Tells the workers to leave once the queue is empty.
+    /// Tells the workers to leave at once, whatever is queued.
     pub fn shutdown(&self) {
         self.lock().shutdown = true;
         self.wake.notify_all();
@@ -201,16 +293,34 @@ pub struct Pool {
     workers: Vec<std::thread::JoinHandle<()>>,
 }
 
+/// Counts a worker as gone however its thread ends.
+#[cfg(not(target_arch = "wasm32"))]
+struct Alive(std::sync::Arc<Executor>);
+
+#[cfg(not(target_arch = "wasm32"))]
+impl Drop for Alive {
+    fn drop(&mut self) {
+        self.0.workers.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
 #[cfg(not(target_arch = "wasm32"))]
 impl Pool {
     pub fn start(exec: std::sync::Arc<Executor>, workers: usize) -> Self {
         let workers = (0..workers.max(1))
             .map(|_| {
-                let exec = std::sync::Arc::clone(&exec);
+                exec.workers.fetch_add(1, Ordering::SeqCst);
+                let alive = Alive(std::sync::Arc::clone(&exec));
                 std::thread::spawn(move || {
+                    let exec = std::sync::Arc::clone(&alive.0);
                     while let Some(mut claimed) = exec.claim_blocking() {
-                        let turn = claimed.run(exec.budget());
-                        exec.finish(claimed, turn);
+                        let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                            claimed.run(exec.budget())
+                        }));
+                        match ran {
+                            Ok(turn) => exec.finish(claimed, turn),
+                            Err(p) => exec.abandon(claimed, panic_text(&*p)),
+                        }
                     }
                 })
             })
@@ -222,13 +332,26 @@ impl Pool {
         &self.exec
     }
 
-    /// Waits for every unit to go quiet, stops the workers and returns the executor.
-    pub fn finish(self) -> std::sync::Arc<Executor> {
-        self.exec.wait_idle();
+    fn join(&mut self) {
         self.exec.shutdown();
-        for w in self.workers {
-            w.join().expect("a worker panicked");
+        for w in std::mem::take(&mut self.workers) {
+            w.join().expect("a worker panicked outside a turn");
         }
-        self.exec
+    }
+
+    /// Waits for every unit to go quiet and stops the workers. A unit that panicked is the
+    /// error; the workers are stopped either way.
+    pub fn finish(mut self) -> Result<std::sync::Arc<Executor>, UnitPanic> {
+        let idle = self.exec.wait_idle();
+        self.join();
+        idle.map(|()| std::sync::Arc::clone(&self.exec))
+    }
+}
+
+/// A pool that is dropped stops its threads, discarding what is queued.
+#[cfg(not(target_arch = "wasm32"))]
+impl Drop for Pool {
+    fn drop(&mut self) {
+        self.join();
     }
 }

crates/improvise-incremental/src/staged.rs

@@ -4,7 +4,7 @@
 use crate::sched::{Executor, Turn, Unit};
 use std::sync::{Arc, Mutex};
 use zset::{
-    ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Flow, Frame, Input, Output, Stream, ZSet,
+    ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Frame, Input, Output, Stream, ZSet,
     channel_pipes,
 };
 
@@ -76,10 +76,10 @@ impl Unit for Head {
         for d in commands {
             self.sink.committed(d);
         }
-        let flow = self.sink.pump();
+        self.sink.pump();
         Turn {
             progress: self.sink.traffic() != before,
-            more: flow == Flow::Backlogged,
+            more: false,
         }
     }
 
@@ -106,10 +106,10 @@ impl Unit for Aggregate {
             self.circuit.commit();
             self.sink.committed_from(&mut self.circuit, &self.out);
         }
-        let flow = self.sink.pump();
+        self.sink.pump();
         Turn {
             progress: self.source.traffic() + self.sink.traffic() != before,
-            more: read == budget || flow == Flow::Backlogged,
+            more: false,
         }
     }
 

crates/improvise-incremental/tests/sched.rs

@@ -1,11 +1,14 @@
 //! The scheduler: sub-circuits on an executor equal one circuit, whatever the number of workers
 //! and the order they run in. Evidence: `context/test-evidence/improvise-incremental.md`.
 
+mod support;
+
 use improvise_incremental::sched::{Claimed, Executor, Pool, Turn, Unit};
 use improvise_incremental::staged::{Cell, Staged, single_circuit};
 use proptest::prelude::*;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use support::watchdog::within;
 use zset::ZSet;
 
 fn command() -> impl Strategy<Value = ZSet<Cell>> {
@@ -96,29 +99,31 @@ fn run_some(exec: &Executor, workers: usize, choices: &mut impl Iterator<Item =
 
 #[test]
 fn real_worker_threads_equal_one_circuit() {
-    let commands: Vec<ZSet<Cell>> = (0..400u32)
-        .map(|n| {
-            [(
-                ((n % 3) as u8, (n / 3 % 3) as u8, (n % 7) as i64 - 3),
-                1 - (n % 3) as i64,
-            )]
-            .into_iter()
-            .collect()
-        })
-        .collect();
-    for workers in 1..=4 {
-        let staged = Staged::new(2, 3);
-        let pool = Pool::start(staged.executor().clone(), workers);
-        for d in &commands {
-            staged.submit(d.clone());
+    within(10, "real worker threads", || {
+        let commands: Vec<ZSet<Cell>> = (0..400u32)
+            .map(|n| {
+                [(
+                    ((n % 3) as u8, (n / 3 % 3) as u8, (n % 7) as i64 - 3),
+                    1 - (n % 3) as i64,
+                )]
+                .into_iter()
+                .collect()
+            })
+            .collect();
+        for workers in 1..=4 {
+            let staged = Staged::new(2, 3);
+            let pool = Pool::start(staged.executor().clone(), workers);
+            for d in &commands {
+                staged.submit(d.clone());
+            }
+            pool.finish().unwrap();
+            assert_eq!(
+                staged.grid(),
+                single_circuit(&commands),
+                "[threads] {workers} workers"
+            );
         }
-        pool.finish();
-        assert_eq!(
-            staged.grid(),
-            single_circuit(&commands),
-            "[threads] {workers} workers"
-        );
-    }
+    });
 }
 
 /// A unit that records which turns ran, and asks for `more` a fixed number of times.
@@ -226,32 +231,34 @@ impl Unit for Exclusive {
 
 #[test]
 fn at_most_one_worker_runs_a_unit_however_often_it_is_kicked() {
-    let (inside, turns) = (
-        Arc::new(AtomicBool::new(false)),
-        Arc::new(AtomicUsize::new(0)),
-    );
-    let exec = Arc::new(Executor::new(
-        vec![Box::new(Exclusive {
-            inside: inside.clone(),
-            turns: turns.clone(),
-        })],
-        1,
-    ));
-    let pool = Pool::start(exec.clone(), 4);
-    let kickers: Vec<_> = (0..4)
-        .map(|_| {
-            let exec = exec.clone();
-            std::thread::spawn(move || {
-                for _ in 0..500 {
-                    exec.kick(0);
-                    std::thread::yield_now();
-                }
+    within(10, "kicking threads", || {
+        let (inside, turns) = (
+            Arc::new(AtomicBool::new(false)),
+            Arc::new(AtomicUsize::new(0)),
+        );
+        let exec = Arc::new(Executor::new(
+            vec![Box::new(Exclusive {
+                inside: inside.clone(),
+                turns: turns.clone(),
+            })],
+            1,
+        ));
+        let pool = Pool::start(exec.clone(), 4);
+        let kickers: Vec<_> = (0..4)
+            .map(|_| {
+                let exec = exec.clone();
+                std::thread::spawn(move || {
+                    for _ in 0..500 {
+                        exec.kick(0);
+                        std::thread::yield_now();
+                    }
+                })
             })
-        })
-        .collect();
-    for k in kickers {
-        k.join().unwrap();
-    }
-    pool.finish();
-    assert!(turns.load(Ordering::SeqCst) > 0, "[own] it ran");
+            .collect();
+        for k in kickers {
+            k.join().unwrap();
+        }
+        pool.finish().unwrap();
+        assert!(turns.load(Ordering::SeqCst) > 0, "[own] it ran");
+    });
 }

crates/improvise-incremental/tests/sched_dag.rs

@@ -0,0 +1,384 @@
+//! A 7-unit DAG with fan-out and fan-in on real threads and on virtual workers, against the same
+//! graph in one circuit (from the review of #364). Evidence: `context/test-evidence/improvise-incremental.md`.
+
+mod support;
+
+use improvise_incremental::sched::{Claimed, Executor, Pool, Turn, Unit};
+use std::sync::{Arc, Mutex};
+use support::watchdog::within;
+use zset::{
+    ChannelPipe, Circuit, Direct, EdgeSink, EdgeSource, Flow, Frame, Output, Stream, ZSet,
+    channel_pipes,
+};
+
+type P = ChannelPipe<Frame<i64>>;
+type Z = ZSet<i64>;
+
+struct Rng(u64);
+impl Rng {
+    fn next(&mut self) -> u64 {
+        self.0 ^= self.0 << 13;
+        self.0 ^= self.0 >> 7;
+        self.0 ^= self.0 << 17;
+        self.0
+    }
+    fn below(&mut self, n: u64) -> u64 {
+        self.next() % n
+    }
+}
+
+fn st_a(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
+    c.map(s, |x| x * 2)
+}
+fn st_b(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
+    c.filter(s, |x| x % 3 != 0)
+}
+fn st_c(c: &mut Circuit, a: &Stream<Z>, b: &Stream<Z>) -> Stream<Z> {
+    let u = c.plus(a, b);
+    let k = c.map(&u, |x| (x.rem_euclid(7), *x));
+    let g = c.aggregate(&k, |_, vs| Some(vs.iter().map(|(v, w)| v * w).sum::<i64>()));
+    c.map(&g, |(k, s)| k * 1_000_000 + s)
+}
+fn st_d(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
+    let m = c.map(s, |x| x.rem_euclid(1000));
+    c.distinct(&m)
+}
+fn st_e(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
+    c.filter(s, |x| x % 2 == 0)
+}
+fn st_f(c: &mut Circuit, s: &Stream<Z>) -> Stream<Z> {
+    c.map(s, |x| x / 2)
+}
+
+fn reference(cmds: &[Z]) -> (Z, Z) {
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<i64>();
+    let a = st_a(&mut c, &s);
+    let b = st_b(&mut c, &s);
+    let cc = st_c(&mut c, &a, &b);
+    let d = st_d(&mut c, &cc);
+    let e = st_e(&mut c, &cc);
+    let f = st_f(&mut c, &e);
+    let (od, of) = (c.output(&d), c.output(&f));
+    for d in cmds {
+        c.push(&i, d.clone());
+        c.step();
+    }
+    (c.take(&od), c.take(&of))
+}
+
+struct Head {
+    sinks: Vec<EdgeSink<i64, Direct, P>>,
+    inbox: Arc<Mutex<Vec<Z>>>,
+    around: Vec<usize>,
+}
+impl Unit for Head {
+    fn turn(&mut self, _: usize) -> Turn {
+        let before: u64 = self.sinks.iter().map(|s| s.traffic()).sum();
+        let cmds = std::mem::take(&mut *self.inbox.lock().unwrap());
+        for d in cmds {
+            for s in &mut self.sinks {
+                s.committed(d.clone());
+            }
+        }
+        let mut more = false;
+        for s in &mut self.sinks {
+            more |= s.pump() == Flow::Backlogged;
+        }
+        let after: u64 = self.sinks.iter().map(|s| s.traffic()).sum();
+        Turn {
+            progress: after != before,
+            more,
+        }
+    }
+    fn neighbours(&self) -> &[usize] {
+        &self.around
+    }
+}
+
+struct Node {
+    c: Circuit,
+    sources: Vec<EdgeSource<i64, Direct, P>>,
+    sinks: Vec<(EdgeSink<i64, Direct, P>, Output<i64>)>,
+    collect: Option<(Output<i64>, Arc<Mutex<Z>>)>,
+    around: Vec<usize>,
+}
+impl Node {
+    fn traffic(&self) -> u64 {
+        self.sources.iter().map(|s| s.traffic()).sum::<u64>()
+            + self.sinks.iter().map(|(s, _)| s.traffic()).sum::<u64>()
+    }
+}
+impl Unit for Node {
+    fn turn(&mut self, budget: usize) -> Turn {
+        let before = self.traffic();
+        let mut read = 0;
+        let mut more = false;
+        for s in &mut self.sources {
+            let r = s.poll(&mut self.c, budget);
+            more |= r == budget;
+            read += r;
+        }
+        if read > 0 {
+            self.c.begin();
+            self.c.step();
+            self.c.commit();
+            for (s, o) in &mut self.sinks {
+                s.committed_from(&mut self.c, o);
+            }
+            if let Some((o, z)) = &self.collect {
+                *z.lock().unwrap() += self.c.take(o);
+            }
+        }
+        for (s, _) in &mut self.sinks {
+            more |= s.pump() == Flow::Backlogged;
+        }
+        Turn {
+            progress: self.traffic() != before,
+            more,
+        }
+    }
+    fn neighbours(&self) -> &[usize] {
+        &self.around
+    }
+}
+
+struct Dag {
+    exec: Arc<Executor>,
+    inbox: Arc<Mutex<Vec<Z>>>,
+    d: Arc<Mutex<Z>>,
+    f: Arc<Mutex<Z>>,
+}
+
+fn dag(rng: &mut Rng, budget: usize) -> Dag {
+    let mut cap = || 1 + rng.below(3) as usize;
+    let (h_a, a_in) = channel_pipes::<Frame<i64>>(cap());
+    let (h_b, b_in) = channel_pipes::<Frame<i64>>(cap());
+    let (a_c, c_in_a) = channel_pipes::<Frame<i64>>(cap());
+    let (b_c, c_in_b) = channel_pipes::<Frame<i64>>(cap());
+    let (c_d, d_in) = channel_pipes::<Frame<i64>>(cap());
+    let (c_e, e_in) = channel_pipes::<Frame<i64>>(cap());
+    let (e_f, f_in) = channel_pipes::<Frame<i64>>(cap());
+    let inbox = Arc::new(Mutex::new(Vec::new()));
+    let (dz, fz) = (
+        Arc::new(Mutex::new(Z::new())),
+        Arc::new(Mutex::new(Z::new())),
+    );
+    let head = Head {
+        sinks: vec![EdgeSink::new(Direct, h_a, 1), EdgeSink::new(Direct, h_b, 1)],
+        inbox: inbox.clone(),
+        around: vec![1, 2],
+    };
+    let unary = |pipe_in: P,
+                 f: fn(&mut Circuit, &Stream<Z>) -> Stream<Z>,
+                 out: Option<P>,
+                 col: Option<Arc<Mutex<Z>>>,
+                 around: Vec<usize>| {
+        let mut c = Circuit::new();
+        let (i, s) = c.input::<i64>();
+        let o = f(&mut c, &s);
+        let o = c.output(&o);
+        let source = EdgeSource::new(Direct, pipe_in, i);
+        let (sinks, collect) = match (out, col) {
+            (Some(p), None) => (vec![(EdgeSink::new(Direct, p, 1), o)], None),
+            (None, Some(z)) => (vec![], Some((o, z))),
+            _ => unreachable!(),
+        };
+        Node {
+            c,
+            sources: vec![source],
+            sinks,
+            collect,
+            around,
+        }
+    };
+    let a = unary(a_in, st_a, Some(a_c), None, vec![0, 3]);
+    let b = unary(b_in, st_b, Some(b_c), None, vec![0, 3]);
+    let cnode = {
+        let mut c = Circuit::new();
+        let (ia, sa) = c.input::<i64>();
+        let (ib, sb) = c.input::<i64>();
+        let o = st_c(&mut c, &sa, &sb);
+        let o1 = c.output(&o);
+        let o2 = c.output(&o);
+        Node {
+            c,
+            sources: vec![
+                EdgeSource::new(Direct, c_in_a, ia),
+                EdgeSource::new(Direct, c_in_b, ib),
+            ],
+            sinks: vec![
+                (EdgeSink::new(Direct, c_d, 1), o1),
+                (EdgeSink::new(Direct, c_e, 1), o2),
+            ],
+            collect: None,
+            around: vec![1, 2, 4, 5],
+        }
+    };
+    let d = unary(d_in, st_d, None, Some(dz.clone()), vec![3]);
+    let e = unary(e_in, st_e, Some(e_f), None, vec![3, 6]);
+    let f = unary(f_in, st_f, None, Some(fz.clone()), vec![5]);
+    let units: Vec<Box<dyn Unit>> = vec![
+        Box::new(head),
+        Box::new(a),
+        Box::new(b),
+        Box::new(cnode),
+        Box::new(d),
+        Box::new(e),
+        Box::new(f),
+    ];
+    let exec = Arc::new(Executor::new(units, budget));
+    exec.kick_all();
+    Dag {
+        exec,
+        inbox,
+        d: dz,
+        f: fz,
+    }
+}
+
+fn commands(rng: &mut Rng, n: usize) -> Vec<Z> {
+    (0..n)
+        .map(|_| {
+            let k = rng.below(4);
+            (0..k)
+                .map(|_| (rng.below(40) as i64 - 20, rng.below(5) as i64 - 2))
+                .collect()
+        })
+        .collect()
+}
+
+fn submit(g: &Dag, d: Z) {
+    g.inbox.lock().unwrap().push(d);
+    g.exec.kick(0);
+}
+
+fn out(g: &Dag) -> (Z, Z) {
+    (g.d.lock().unwrap().clone(), g.f.lock().unwrap().clone())
+}
+
+#[test]
+fn a_dag_on_real_threads_equals_one_circuit_and_idle_means_done() {
+    within(30, "the DAG on a pool", || {
+        let mut rng = Rng(0x9e3779b97f4a7c15);
+        for round in 0..8 {
+            let workers = 1 + round % 8;
+            let budget = 1 + rng.below(4) as usize;
+            let g = dag(&mut rng, budget);
+            let cmds = commands(&mut rng, 300);
+            let pool = Pool::start(g.exec.clone(), workers);
+            // Concurrent submitter thread, with idle checkpoints from the host.
+            let mut at = 0;
+            while at < cmds.len() {
+                let chunk = 1 + rng.below(80) as usize;
+                let end = (at + chunk).min(cmds.len());
+                for d in &cmds[at..end] {
+                    submit(&g, d.clone());
+                    if rng.below(7) == 0 {
+                        std::thread::yield_now();
+                    }
+                }
+                at = end;
+                if rng.below(3) == 0 {
+                    pool.executor().wait_idle().unwrap();
+                    assert_eq!(
+                        out(&g),
+                        reference(&cmds[..at]),
+                        "[idle] round {round} w{workers} at {at}"
+                    );
+                }
+            }
+            pool.finish().unwrap();
+            assert_eq!(
+                out(&g),
+                reference(&cmds),
+                "[final] round {round} w{workers} b{budget}"
+            );
+        }
+    });
+}
+
+#[test]
+fn submitter_threads_race_the_pool() {
+    within(30, "racing submitters", || {
+        let mut rng = Rng(0x1234567);
+        for workers in 1..=4 {
+            let g = Arc::new(dag(&mut rng, 2));
+            let cmds = commands(&mut rng, 600);
+            let pool = Pool::start(g.exec.clone(), workers);
+            // Four submitters; order between them is free, the result is a sum so order-free.
+            let parts: Vec<Vec<Z>> = (0..4)
+                .map(|k| cmds.iter().skip(k).step_by(4).cloned().collect())
+                .collect();
+            let hs: Vec<_> = parts
+                .into_iter()
+                .map(|p| {
+                    let g = g.clone();
+                    std::thread::spawn(move || {
+                        for d in p {
+                            submit(&g, d);
+                        }
+                    })
+                })
+                .collect();
+            for h in hs {
+                h.join().unwrap();
+            }
+            pool.finish().unwrap();
+            // `distinct` and aggregates are functions of the integral, so order-free.
+            assert_eq!(out(&g), reference(&cmds), "[race] w{workers}");
+        }
+    });
+}
+
+fn run_virtual(exec: &Executor, workers: usize, rng: &mut Rng, steps: usize, drain: bool) {
+    let mut held: Vec<(Claimed, Turn)> = Vec::new();
+    let mut n = 0;
+    loop {
+        if !drain && n >= steps {
+            break;
+        }
+        n += 1;
+        assert!(n < 2_000_000, "never quiet");
+        let c = rng.next() as usize;
+        if held.len() < workers
+            && (held.is_empty() || !c.is_multiple_of(3))
+            && let Some(mut cl) = exec.claim(|k| c / 3 % k)
+        {
+            let t = cl.run(exec.budget());
+            held.push((cl, t));
+            continue;
+        }
+        if held.is_empty() {
+            if drain {
+                break;
+            }
+            continue;
+        }
+        let (cl, t) = held.remove(c / 3 % held.len());
+        exec.finish(cl, t);
+    }
+    for (cl, t) in held {
+        exec.finish(cl, t);
+    }
+}
+
+#[test]
+fn a_dag_on_virtual_workers_in_any_order_equals_one_circuit() {
+    let mut rng = Rng(0xdeadbeef);
+    for round in 0..60 {
+        let workers = 1 + round % 8;
+        let budget = 1 + rng.below(3) as usize;
+        let g = dag(&mut rng, budget);
+        let n = 1 + rng.below(40) as usize;
+        let cmds = commands(&mut rng, n);
+        for d in &cmds {
+            submit(&g, d.clone());
+            let s = rng.below(10) as usize;
+            run_virtual(&g.exec, workers, &mut rng, s, false);
+        }
+        run_virtual(&g.exec, workers, &mut rng, 0, true);
+        assert!(g.exec.idle());
+        assert_eq!(out(&g), reference(&cmds), "[virt] round {round}");
+    }
+}

crates/improvise-incremental/tests/sched_threads.rs

@@ -0,0 +1,215 @@
+//! The scheduler under panics, idle workers and a full edge, with real threads behind a
+//! watchdog. Evidence: `context/test-evidence/improvise-incremental.md`.
+
+mod support;
+
+use improvise_incremental::sched::{Executor, Pool, Turn, Unit};
+use improvise_incremental::staged::{Cell, Staged, single_circuit};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use support::watchdog::within;
+use zset::ZSet;
+
+struct Boom;
+
+impl Unit for Boom {
+    fn turn(&mut self, _: usize) -> Turn {
+        panic!("boom");
+    }
+
+    fn neighbours(&self) -> &[usize] {
+        &[]
+    }
+}
+
+/// A unit that counts its turns.
+struct Count(Arc<AtomicUsize>);
+
+impl Unit for Count {
+    fn turn(&mut self, _: usize) -> Turn {
+        self.0.fetch_add(1, Ordering::SeqCst);
+        Turn {
+            progress: false,
+            more: false,
+        }
+    }
+
+    fn neighbours(&self) -> &[usize] {
+        &[]
+    }
+}
+
+#[test]
+fn a_panicking_turn_does_not_hang_finish() {
+    within(5, "Pool::finish after a unit panicked", || {
+        let exec = Arc::new(Executor::new(vec![Box::new(Boom)], 1));
+        let pool = Pool::start(exec.clone(), 2);
+        exec.kick(0);
+        let _ = pool.finish();
+    });
+}
+
+#[test]
+fn a_panicking_unit_is_reported_and_its_neighbours_still_run() {
+    let turns = Arc::new(AtomicUsize::new(0));
+    let t = turns.clone();
+    let err = within(5, "finish with one dead unit", move || {
+        let units: Vec<Box<dyn Unit>> = vec![Box::new(Boom), Box::new(Count(t))];
+        let exec = Arc::new(Executor::new(units, 1));
+        let pool = Pool::start(exec.clone(), 2);
+        exec.kick_all();
+        let err = pool.finish().err();
+        exec.kick(0);
+        exec.kick(1);
+        err
+    });
+    let err = err.expect("[panic] finish reports the panic");
+    assert_eq!(err.unit, 0);
+    assert!(
+        err.message.contains("boom"),
+        "[panic] the payload: {}",
+        err.message
+    );
+    assert!(
+        turns.load(Ordering::SeqCst) >= 1,
+        "[panic] the live unit ran"
+    );
+}
+
+#[test]
+fn a_dead_unit_is_never_run_again() {
+    let exec = Executor::new(vec![Box::new(Boom)], 1);
+    exec.kick(0);
+    let mut c = exec.claim(|_| 0).unwrap();
+    let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| c.run(1)));
+    assert!(r.is_err());
+    exec.abandon(c, "boom".to_string());
+    exec.kick(0);
+    assert!(
+        exec.claim(|_| 0).is_none(),
+        "[dead] a kick does not revive it"
+    );
+    assert!(exec.idle());
+}
+
+#[test]
+#[should_panic(expected = "no such unit")]
+fn a_kick_of_a_unit_that_does_not_exist_panics_before_the_lock() {
+    Executor::new(vec![Box::new(Boom)], 1).kick(5);
+}
+
+#[test]
+fn an_executor_survives_a_bad_kick() {
+    let exec = Executor::new(vec![Box::new(Boom)], 1);
+    let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| exec.kick(5)));
+    assert!(r.is_err());
+    assert!(
+        exec.idle(),
+        "[lock] the lock was not poisoned by the bad index"
+    );
+}
+
+#[test]
+fn every_worker_stays_through_wakes_with_nothing_queued() {
+    within(10, "the pool with idle wakes", || {
+        let turns = Arc::new(AtomicUsize::new(0));
+        let units: Vec<Box<dyn Unit>> =
+            vec![Box::new(Count(turns.clone())), Box::new(Count(turns))];
+        let exec = Arc::new(Executor::new(units, 1));
+        let pool = Pool::start(exec.clone(), 4);
+        assert_eq!(exec.workers(), 4);
+        for round in 0..200 {
+            exec.kick(round % 2);
+            if round % 20 == 0 {
+                let _ = exec.wait_idle();
+            }
+        }
+        let _ = exec.wait_idle();
+        assert_eq!(
+            exec.workers(),
+            4,
+            "[workers] a wake with nothing queued sent a worker home"
+        );
+        let _ = pool.finish();
+        assert_eq!(exec.workers(), 0, "[workers] finish stops them all");
+    });
+}
+
+#[test]
+fn a_pool_that_is_dropped_stops_its_threads() {
+    within(10, "dropping a pool", || {
+        let exec = Arc::new(Executor::new(vec![Box::new(Count(Arc::default()))], 1));
+        {
+            let _pool = Pool::start(exec.clone(), 3);
+            assert_eq!(exec.workers(), 3);
+        }
+        assert_eq!(exec.workers(), 0, "[drop] no thread is left parked");
+    });
+}
+
+fn cmd(r: u8, c: u8, v: i64) -> ZSet<Cell> {
+    [((r, c, v), 1)].into_iter().collect()
+}
+
+#[test]
+fn a_full_edge_does_not_make_its_sender_spin() {
+    let staged = Staged::new(1, 1);
+    let exec = staged.executor();
+    // Hold the aggregate and the projection, so nothing drains the head's edge.
+    let agg = exec.claim(|_| 1).unwrap();
+    let proj = exec.claim(|_| 1).unwrap();
+    assert_eq!((agg.id(), proj.id()), (1, 2));
+    for n in 0..4 {
+        staged.submit(cmd(0, 0, n));
+        let mut head = exec.claim(|_| 0).unwrap();
+        let t = head.run(exec.budget());
+        exec.finish(head, t);
+    }
+    let mut turns = 0;
+    while let Some(mut head) = exec.claim(|_| 0) {
+        turns += 1;
+        assert!(
+            turns < 50,
+            "[spin] the head keeps asking to run with its edge full"
+        );
+        let t = head.run(exec.budget());
+        exec.finish(head, t);
+    }
+    let t = Turn {
+        progress: false,
+        more: false,
+    };
+    exec.finish(agg, t);
+    exec.finish(proj, t);
+}
+
+#[test]
+fn a_unit_whose_kicks_coalesced_still_drains_its_edge() {
+    let staged = Staged::new(4, 1);
+    let exec = staged.executor();
+    // Settle the start-up requests.
+    while let Some(mut u) = exec.claim(|_| 0) {
+        let t = u.run(exec.budget());
+        exec.finish(u, t);
+    }
+    // Two commands cross before the aggregate runs: its two kicks are one queue entry.
+    staged.submit(cmd(0, 0, 1));
+    let mut head = exec.claim(|_| 0).unwrap();
+    let t = head.run(1);
+    exec.finish(head, t);
+    staged.submit(cmd(1, 1, 2));
+    let mut head = exec.claim(|n| n - 1).unwrap();
+    assert_eq!(head.id(), 0);
+    let t = head.run(1);
+    exec.finish(head, t);
+    while let Some(mut u) = exec.claim(|_| 0) {
+        let t = u.run(1);
+        exec.finish(u, t);
+    }
+    let want = single_circuit(&[cmd(0, 0, 1), cmd(1, 1, 2)]);
+    assert_eq!(
+        staged.grid(),
+        want,
+        "[coalesced] the second frame was left unread"
+    );
+}

crates/improvise-incremental/tests/support/mod.rs

@@ -3,6 +3,7 @@
 
 pub mod live;
 pub mod remote;
+pub mod watchdog;
 
 use improvise_core::formula::ast::AggFunc;
 use improvise_core::formula::parse_formula;

crates/improvise-incremental/tests/support/watchdog.rs

@@ -0,0 +1,23 @@
+//! A hung real-thread test must fail by name, not hang the binary.
+
+use std::sync::mpsc;
+use std::thread;
+use std::time::Duration;
+
+/// Runs `f` on its own thread and fails with `what` if it has not returned in `secs`. A panic in
+/// `f` is re-raised here. A timed-out thread is left behind; the process exits without it.
+pub fn within<T: Send + 'static>(
+    secs: u64,
+    what: &str,
+    f: impl FnOnce() -> T + Send + 'static,
+) -> T {
+    let (tx, rx) = mpsc::channel();
+    thread::spawn(move || {
+        let _ = tx.send(std::panic::catch_unwind(std::panic::AssertUnwindSafe(f)));
+    });
+    match rx.recv_timeout(Duration::from_secs(secs)) {
+        Ok(Ok(v)) => v,
+        Ok(Err(p)) => std::panic::resume_unwind(p),
+        Err(_) => panic!("[watchdog] {what} did not finish in {secs}s"),
+    }
+}

crates/zset/tests/edge.rs

@@ -628,7 +628,11 @@ fn traffic_never_goes_down_across_a_restart_of_either_end() {
     let mut c2 = Circuit::new();
     let (i2, _) = c2.input::<u8>();
     let src = src.restart(i2);
-    assert_eq!((sink.traffic(), src.traffic()), (s, r), "[restart] counters kept");
+    assert_eq!(
+        (sink.traffic(), src.traffic()),
+        (s, r),
+        "[restart] counters kept"
+    );
 }
 
 #[test]