refactor(zset): a circuit is Send; edge values live in an owned arena and handles are ids

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

context/code/zset-ffi.md

@@ -59,7 +59,7 @@ link, so they call the `extern "C"` functions as plain functions.
    its return value (`flat_map`/`index_by` non-zero, `keep` and `expired` negative), and the
    step then fails with `ZS_ERR_CALLBACK`. A callback must return normally. A Rust panic across
    `extern "C"` aborts, and a Lisp `handler-case` outside `zs_circuit_step` that unwinds
-   through the Rust frames skips destructors and leaves a `RefCell` inside `zset` borrowed.
+   through the Rust frames skips destructors and leaves the circuit half way through a step.
    The ABI cannot stop the unwinding, so it detects it: `ZsCircuit::in_step` is set for the
    duration of `step` and `rebuild_step`, and cleared only when they return. Every entry
    point first reads it, so a circuit whose frames were skipped answers `ZS_ERR_POISONED`
@@ -73,9 +73,15 @@ link, so they call the `extern "C"` functions as plain functions.
    return the failure value (`tests/smoke.lisp` does). `index_by` needs exactly one emitted
    key and a keyed stream valid pairs, else `ZS_ERR_BAD_VALUE`; these faults are recorded in a
    shared cell rather than panicking.
-10. **Thread confinement.** A circuit and every rebuild job made from it share an `Rc`
-    (`Input`), so they are one group: keep the group on one thread at a time. Z-sets are
-    separate handles. The error message is per thread.
+10. **Threads** (`zset.md`, "Threads"). A `ZsCircuit` may be moved between threads and is used
+    by one at a time; it is `Send` and not `Sync`, with no lock. Callbacks run on whichever
+    thread calls `step` or `rebuild_step`, so the caller's `ctx` must be usable from every
+    thread that may step (the wrapper `Ctx` is `unsafe impl Send` on that contract). The
+    first-fault cell is the one lock, and it is the FFI's own. A `ZsRebuild` no longer shares
+    a reference count with its circuit (the handles are ids), so it is a separate object, and
+    a `zs_rebuild_step` call uses both. The caller still keeps each to one thread at a time.
+    Z-sets are separate handles. The error message is per thread
+    (`tests/threads_abi.rs`: callbacks ran on the thread that stepped).
 11. **The header is hand-written** (cbindgen is not a dependency of this repo). `tests/header.rs`
     keeps it honest. It maps each Rust parameter and return type of every exported function
     (including the `unary_op!` expansions) to its C spelling, through a table that panics on

context/code/zset.md

@@ -128,13 +128,14 @@ Clause 1's deferred scope, built without ADR 008. The ADR is not edited; this is
 
 Decisions, where the ADR left a choice:
 
-1. **Wiring is apart from logic.** `circuit.rs` is the only file with `Rc`, `RefCell` and
+1. **Wiring is apart from logic.** `circuit.rs` is the only file with the arena and
    `Box<dyn Node>`. `ops.rs` holds each stateful operator as a plain struct with a pure
    `step(delta) -> delta`, testable and usable without a circuit.
 2. **Scheduling is creation order.** A node is built from streams that already exist, so
    creation order is a topological order, and the circuit has no cycles: the ADR has no
    recursion operator, and `integrate` is its own stateful node, not a delayed feedback loop.
-   `delay` is a node with one held Z-set. Not `Send`; one thread per circuit (clause 1).
+   `delay` is a node with one held Z-set. `Send`, not `Sync`: one thread steps a circuit at a
+   time (clause 1; "Threads" below).
 3. **Incremental means the integral of the output is the operator of the integral of the
    input.** That is the property every operator is tested against, over random delta streams.
    `distinct` and `aggregate` are not linear, so they keep the integrated input and emit the
@@ -232,8 +233,8 @@ clause's wording stands; this is how the crate reads it).
    (the whole-tick value) only exists while a tick is split. `Slot::whole` is `None` otherwise
    and `current()` reads the piece. `piece()` lends a borrow rather than a clone
    (improvise-bw4p.11). An integrate stream now holds its integral twice (the node's `acc`
-   and the stream's piece), not three times; sharing those two needs an `Rc` in the slot and
-   was not done.
+   and the stream's piece), not three times; sharing those two needs shared ownership in the
+   slot and was not done.
 6. **What a transaction does not undo.**
    - A closure's captured state (a `map` function that counts, an FFI callback's side
      effects). Operators that keep state keep it in the crate's own types for this reason.
@@ -248,7 +249,7 @@ clause's wording stands; this is how the crate reads it).
    - An `Output::take` made inside the transaction: a rollback subtracts the step's output
      from the sink, which a `take` has emptied. Take after `commit`; `zset-ffi` refuses to take
      while a transaction is open.
-7. **Pushes.** `Input::push` is not logged (a handle has no circuit). `rollback` puts the
+7. **Pushes.** `Circuit::push` is not logged. `rollback` puts the
    pending delta back to what it was at `begin`, so a push made after `begin` is dropped, and
    one made before it is pending again. Operators built after `begin` start logging at once
    and go back to empty; they are not removed.
@@ -262,10 +263,45 @@ circuit unusable: drop it and build another.
 
 ## Rebuilds and circuits (improvise-bw4p.12)
 
-A `Rebuild` takes its circuit's identity from its `Input`: the circuit holds an `Rc<()>` and every
-`Input` it makes holds a clone, so `Rebuild::belongs_to(&circuit)` is a pointer comparison and no
-global counter is needed. `Rebuild::step` on another circuit panics, where it used to report
+A `Rebuild` takes its circuit's identity from its `Input`: every handle a circuit makes holds the
+circuit's token (a process-wide counter since the arena, `Rc::ptr_eq` before), so
+`Rebuild::belongs_to(&circuit)` is a comparison. `Rebuild::step` on another circuit panics, where it used to report
 `Done` and land its chunk in that circuit's next tick. An interrupted job (decision 8 above) can
 never finish: `interrupted` stays true however many steps follow, because a split tick cannot
 be mended and a rollback that took a chunk is counted forever (decision 6 above). Drop it and start another job from the same snapshot; the host keeps the
 snapshot.
+
+## Threads (user ruling 2026-10-11; ADR 009 clause 1 and section 9)
+
+The graph is network-agnostic and multi-threaded from the start, and **multi-threading is
+scheduling**: a circuit stays synchronous and single-threaded, and a host moves whole circuits
+between threads. Shared-nothing: one circuit per runnable unit, joined only by edges (the edge
+adapter, a later PR, is the one thing that crosses threads). Splitting one circuit's keys across
+threads (data parallelism) is not done and not precluded: its exchange step would be the same
+edge. This is a ruling recorded here, not an ADR amendment. Native threads are assumed; wasm
+threads are out of scope, and nothing in `zset` spawns one. `Send` and `mpsc` compile on wasm32.
+
+1. **`Circuit` is `Send` and not `Sync`, with no lock anywhere in it.** It moves between
+   threads, one at a time (`tests/threads.rs`: a circuit built on one thread, stepped on a new
+   thread each tick, equals a circuit that never moved; a transaction begun on one thread is
+   rolled back on another). Two `compile_fail` doctests on `Circuit` pin that it is not `Sync`
+   and refuses an `Rc` closure.
+2. **The `Rc<RefCell>` graph is an arena.** What made the circuit `!Send` was the sharing
+   between the circuit and the handles it hands out. A circuit now owns every edge's slot, every
+   input's pending delta and every sink's sum in one `Vec<Box<dyn Any + Send>>`; a node holds
+   typed ids, and `eval`, `begin`, `commit` and `rollback` take the arena. A downcast on every
+   access is the price (an `Any` type check; `tests/cost.rs` counts clones, which it leaves as they were,
+   and time was not measured).
+3. **Handles are plain ids, used through the circuit.** `Stream`, `Input` and `Output` are
+   `Copy`, `Send` and `Sync`, and carry their circuit's token. The calls moved onto the circuit:
+   `Circuit::current(&stream)`, `Circuit::push(&input, delta)`, `Circuit::take(&output)`. A
+   handle from another circuit panics ("handle from another circuit"), where before it was
+   quietly wrong. A consequence: a closure can no longer read a stream during a step, since it
+   has no circuit; the old `b1_...` case does not compile, so it is gone.
+4. **User closures are `Send + 'static`;** keys and values are `Send`. A custom `Trace` handed to
+   `join_indexed_in` or `aggregate_in` must be `Send`. `Piecewise` requires `Send + 'static`.
+5. **The sink does not clone.** Its node reads the input edge and writes the sum through
+   `Arena::read_write`, two disjoint entries, so `an_eval_borrows_its_input_instead_of_cloning_it`
+   still holds.
+6. **Handing a circuit to a thread is the host's business** (`std::thread` lives in host crates).
+   Whoever owns the circuit pushes, steps and takes; a handle is a number and needs no thread.

context/test-evidence/zset-ffi.md

@@ -221,3 +221,15 @@ A job whose first chunk failed inside the transaction resumed without that chunk
 The same path through the ABI: `zs_rebuild_step` returns `ZS_ERR_OVERFLOW` from the chunk's
 push, `zs_circuit_rollback` succeeds, and the next `zs_rebuild_step` was a silent success; it is
 now `ZS_ERR_INTERRUPTED`. Rival: the mark before the push removed fails `[f5]`.
+
+### the_opaque_handles_are_send
+
+`ZsCircuit` and `ZsRebuild` are `Send` (the engine's fault cell is an `Arc<Mutex>` and the
+callback context an `unsafe impl Send` wrapper). Rival: restoring the `Rc<Cell>` fault cell, or
+dropping `unsafe impl Send for Ctx`, fails `engine.rs` to build (`*mut c_void cannot be sent`).
+
+### a_circuit_stepped_on_two_threads_runs_its_callbacks_on_the_thread_that_steps
+
+A filter callback records its thread; the circuit is stepped on the test thread, moved, and
+stepped on another. The ids differ and match the stepping threads, so the callback is not
+pinned to the thread that built the circuit. The second thread's push is seen by its own step.

context/test-evidence/zset.md

@@ -552,9 +552,9 @@ A first chunk that panicked left `began_at` unset, and the job resumed without i
 now recorded before the chunk runs. Rival: recording it never (the `get_or_insert` removed)
 fails all four; a rival recording it after the chunk was not run separately.
 
-### b1_a_closure_that_reads_a_stream_during_eval_does_not_panic
+b1 (a closure reading a stream during an eval; the test is removed)
 
-The borrowed `piece()` does not conflict with a closure that reads streams during an eval,
+Superseded by the arena (PR dbsp/zset-send): a stream is read through `Circuit::current`, so a closure cannot read one during a step, and the case no longer compiles. It used to check that the borrowed `piece()` did not conflict with a closure that read streams,
 including its own source, inside a transaction. It passed before and after.
 
 ### c1_a_clone_panic_anywhere_in_a_step_is_rolled_back
@@ -596,3 +596,26 @@ Delta review of #355 at b2445fed. `Rebuild::step` takes the chunk from its itera
 it before stepping; a push that overflows the pending delta panicked with no step run, so
 `undone` was not bumped and the resumed job silently dropped the chunk (key 1's `+1`). The step
 is now marked before the push. Rival: that mark removed fails `[p1]`.
+
+### a_circuit_and_every_handle_it_hands_out_is_send
+
+The Rc/RefCell graph made `Circuit` `!Send`. Rival: an `Rc<()>` field on `Circuit` fails the
+build of `tests/threads.rs` with three `Rc<()> cannot be sent between threads safely` errors,
+so the file never runs; the same file's other tests move a circuit for real.
+
+### a_circuit_built_on_one_thread_steps_on_another_and_returns
+
+A circuit moved to a new thread for each of nine ticks equals one that never moved, over a
+pipeline with `integrate`, `distinct`, `aggregate`, `filter` and `delay`. It would fail if a
+node kept state outside the circuit's arena, which a thread-local would do.
+
+### a_transaction_begun_on_one_thread_is_rolled_back_on_another
+
+`begin` and a step on one thread, `rollback` on another: the stream, the sink and `ticks` are
+as at `begin`. The undo log lives in the nodes, which move with the circuit.
+
+### a_handle_from_another_circuit_is_refused
+
+Ids index an arena, so a foreign handle used to be an out-of-range or mistyped slot. It is now
+refused by name ("handle from another circuit"); `should_panic(expected)` pins the message,
+so a downcast panic does not satisfy it.

crates/zset-ffi/include/zset_ffi.h

@@ -17,9 +17,10 @@
  *  - The library never returns memory the caller must free, except handles.
  *  - Weights are int64_t. A weight that would leave int64_t is ZS_ERR_OVERFLOW, never a
  *    wrong value, and never an unwinding exception.
- *  - Not thread-safe. A ZsCircuit and every ZsRebuild made from it form ONE group (they share
- *    a reference count): keep the whole group on one thread at a time. Z-sets are separate
- *    handles. Streams, inputs and sinks are uint32_t ids local to their circuit; a sink id is
+ *  - Not thread-safe, but movable. A ZsCircuit, a ZsRebuild and a ZsZSet may each be handed to
+ *    another thread; use each from one thread at a time (a zs_rebuild_step call uses both its
+ *    job and its circuit). Callbacks run on whichever thread calls zs_circuit_step or
+ *    zs_rebuild_step, so a ctx must be usable from every thread that may step. Streams, inputs and sinks are uint32_t ids local to their circuit; a sink id is
  *    never a stream id (they are refused as ZS_ERR_BAD_HANDLE), but an id from another circuit
  *    is not detected. A ZsRebuild is stepped only on the circuit that made it (else
  *    ZS_ERR_BAD_HANDLE), even if another circuit now lives at the old address.

crates/zset-ffi/src/engine.rs

@@ -2,10 +2,9 @@
 //! `zset` are caught and classified here. See `context/code/zset-ffi.md`.
 
 use std::any::Any;
-use std::cell::Cell;
 use std::ffi::c_void;
 use std::panic::{AssertUnwindSafe, catch_unwind};
-use std::rc::Rc;
+use std::sync::{Arc, Mutex, MutexGuard};
 use zset::{Circuit, Input, Output, Progress, Rebuild, Stream, ZSet};
 
 pub type Bytes = Vec<u8>;
@@ -130,19 +129,31 @@ pub type KeepFn = extern "C" fn(*mut c_void, *const u8, usize) -> i32;
 /// Returns 0 (still time), 1 (expired), or negative to fail the step.
 pub type ExpiredFn = extern "C" fn(*mut c_void) -> i32;
 
-/// A caller's context pointer; the circuit never dereferences it.
+/// A caller's context pointer; the circuit never dereferences it. It is `Send` because the
+/// circuit may move to another thread and a callback then runs there with this pointer: the
+/// caller owes a context that is usable from whichever thread steps (`context/code/zset-ffi.md`).
 #[derive(Clone, Copy)]
 struct Ctx(*mut c_void);
 
-type Fault = Rc<Cell<Option<(Code, &'static str)>>>;
+// SAFETY: see the type's contract; the pointer is only ever handed back to the caller.
+unsafe impl Send for Ctx {}
+
+impl Ctx {
+    /// A method, so a closure captures the whole `Ctx` and not its (non-`Send`) field.
+    fn get(self) -> *mut c_void {
+        self.0
+    }
+}
+
+type Fault = Arc<Mutex<Option<(Code, &'static str)>>>;
+
+fn fault_lock(fault: &Fault) -> MutexGuard<'_, Option<(Code, &'static str)>> {
+    fault.lock().unwrap_or_else(|e| e.into_inner())
+}
 
 /// Records a fault unless one is already recorded: the first one is the cause.
 fn note(fault: &Fault, code: Code, msg: &'static str) {
-    if let Some(first) = fault.take() {
-        fault.set(Some(first));
-    } else {
-        fault.set(Some((code, msg)));
-    }
+    fault_lock(fault).get_or_insert((code, msg));
 }
 
 extern "C" fn collect(ctx: *mut c_void, p: *const u8, n: usize) {
@@ -161,7 +172,7 @@ extern "C" fn collect(ctx: *mut c_void, p: *const u8, n: usize) {
 fn images(f: FlatMapFn, ctx: Ctx, key: &Bytes, fault: &Fault) -> Vec<Bytes> {
     let mut out: Vec<Bytes> = Vec::new();
     let rc = f(
-        ctx.0,
+        ctx.get(),
         key.as_ptr(),
         key.len(),
         collect,
@@ -227,7 +238,7 @@ impl Engine {
             streams: Vec::new(),
             inputs: Vec::new(),
             sinks: Vec::new(),
-            fault: Rc::default(),
+            fault: Arc::default(),
             poisoned: false,
             txn: false,
         }
@@ -236,7 +247,7 @@ impl Engine {
     fn stream(&self, id: u32) -> Res<Stream<BZ>> {
         self.streams
             .get(id as usize)
-            .cloned()
+            .copied()
             .ok_or_else(|| Fail::new(Code::BadHandle, "no such stream"))
     }
 
@@ -307,9 +318,9 @@ impl Engine {
     pub fn filter(&mut self, s: u32, keep: KeepFn, ctx: *mut c_void) -> Res<u32> {
         self.alive()?;
         let s = self.stream(s)?;
-        let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault));
+        let (ctx, fault) = (Ctx(ctx), Arc::clone(&self.fault));
         let out = self.circuit.filter(&s, move |k: &Bytes| {
-            match keep(ctx.0, k.as_ptr(), k.len()) {
+            match keep(ctx.get(), k.as_ptr(), k.len()) {
                 0 => false,
                 n if n > 0 => true,
                 _ => {
@@ -324,7 +335,7 @@ impl Engine {
     pub fn flat_map(&mut self, s: u32, f: FlatMapFn, ctx: *mut c_void) -> Res<u32> {
         self.alive()?;
         let s = self.stream(s)?;
-        let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault));
+        let (ctx, fault) = (Ctx(ctx), Arc::clone(&self.fault));
         let out = self
             .circuit
             .flat_map(&s, move |k: &Bytes| images(f, ctx, k, &fault));
@@ -335,7 +346,7 @@ impl Engine {
     pub fn index_by(&mut self, s: u32, f: FlatMapFn, ctx: *mut c_void) -> Res<u32> {
         self.alive()?;
         let s = self.stream(s)?;
-        let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault));
+        let (ctx, fault) = (Ctx(ctx), Arc::clone(&self.fault));
         let out = self.circuit.flat_map(&s, move |k: &Bytes| {
             let mut gs = images(f, ctx, k, &fault);
             if gs.len() != 1 {
@@ -359,7 +370,7 @@ impl Engine {
 
     /// Splits `pair(k, v)` elements; a malformed one records a fault and is dropped.
     fn keyed(&mut self, s: &Stream<BZ>) -> Stream<ZSet<(Bytes, Bytes)>> {
-        let fault = Rc::clone(&self.fault);
+        let fault = Arc::clone(&self.fault);
         self.circuit.flat_map(s, move |b: &Bytes| match unpair(b) {
             Some((k, v)) => vec![(k.to_vec(), v.to_vec())],
             None => {
@@ -378,7 +389,7 @@ impl Engine {
         self.alive()?;
         let (a, b) = (self.stream(a)?, self.stream(b)?);
         let (ka, kb) = (self.keyed(&a), self.keyed(&b));
-        let fault = Rc::clone(&self.fault);
+        let fault = Arc::clone(&self.fault);
         let j = self
             .circuit
             .join_indexed(&ka, &kb, move |k: &Bytes, x: &Bytes, y: &Bytes| {
@@ -396,7 +407,7 @@ impl Engine {
         self.alive()?;
         let s = self.stream(s)?;
         let keyed = self.keyed(&s);
-        let fault = Rc::clone(&self.fault);
+        let fault = Arc::clone(&self.fault);
         let folded = self
             .circuit
             .aggregate(&keyed, move |_, vals: &[(Bytes, i64)]| {
@@ -427,7 +438,7 @@ impl Engine {
                     }
                 }
             });
-        let fault = Rc::clone(&self.fault);
+        let fault = Arc::clone(&self.fault);
         let out = self
             .circuit
             .flat_map(&folded, move |(k, r): &(Bytes, i64)| {
@@ -445,26 +456,27 @@ impl Engine {
     fn input_of(&self, id: u32) -> Res<Input<Bytes>> {
         self.inputs
             .get(id as usize)
-            .and_then(|i| i.clone())
+            .and_then(|i| *i)
             .ok_or_else(|| Fail::new(Code::BadHandle, "that stream is not an input"))
     }
 
     pub fn push(&mut self, input: u32, z: &BZ) -> Res<()> {
         self.alive()?;
-        self.input_of(input)?.push(z.clone());
+        let input = self.input_of(input)?;
+        self.circuit.push(&input, z.clone());
         Ok(())
     }
 
     /// Runs `f` on the circuit; a panic or a recorded fault poisons the engine.
     fn guarded<T>(&mut self, f: impl FnOnce(&mut Circuit) -> T) -> Res<T> {
         self.alive()?;
-        self.fault.set(None);
+        *fault_lock(&self.fault) = None;
         match catch_unwind(AssertUnwindSafe(|| f(&mut self.circuit))) {
             Err(payload) => {
                 self.poisoned = true;
                 Err(classify(payload))
             }
-            Ok(v) => match self.fault.take() {
+            Ok(v) => match fault_lock(&self.fault).take() {
                 Some((code, msg)) => {
                     self.poisoned = true;
                     Err(Fail::new(code, msg))
@@ -529,7 +541,8 @@ impl Engine {
         self.sinks
             .get((sink & !SINK_BIT) as usize)
             .filter(|_| sink & SINK_BIT != 0)
-            .map(|s| s.take())
+            .copied()
+            .map(|s| self.circuit.take(&s))
             .ok_or_else(|| Fail::new(Code::BadHandle, "no such output"))
     }
 
@@ -551,7 +564,7 @@ impl Engine {
                 "the rebuild was interrupted by a circuit step before it finished",
             ));
         }
-        let fault = Rc::clone(&self.fault);
+        let fault = Arc::clone(&self.fault);
         let progress = self.guarded(|c| {
             job.step(c, || match expired {
                 None => false,

crates/zset-ffi/tests/threads_abi.rs

@@ -0,0 +1,98 @@
+//! A circuit moves between threads, one at a time, and its callbacks run on whichever thread
+//! steps it (`context/code/zset-ffi.md`, decision 10). Evidence: `context/test-evidence/zset-ffi.md`.
+
+use std::ffi::c_void;
+use std::ptr::null_mut;
+use std::sync::Mutex;
+use std::thread::{self, ThreadId};
+use zset_ffi::*;
+
+const OK: i32 = 0;
+
+fn is_send<T: Send>() {}
+
+#[test]
+fn the_opaque_handles_are_send() {
+    is_send::<ZsCircuit>();
+    is_send::<ZsRebuild>();
+}
+
+/// The caller's context: which threads ran the callback.
+struct Seen(Mutex<Vec<ThreadId>>);
+
+extern "C" fn keep_all(ctx: *mut c_void, _: *const u8, _: usize) -> i32 {
+    // SAFETY: `ctx` is the `Seen` the test keeps alive across every step.
+    unsafe { &*(ctx as *const Seen) }
+        .0
+        .lock()
+        .unwrap()
+        .push(thread::current().id());
+    1
+}
+
+/// A raw handle moved to another thread; the test is the one-at-a-time owner.
+struct Moved(*mut ZsCircuit);
+unsafe impl Send for Moved {}
+
+fn push_key(c: *mut ZsCircuit, input: u32, key: &[u8]) {
+    let mut z = null_mut();
+    assert_eq!(unsafe { zs_zset_new(&mut z) }, OK);
+    assert_eq!(unsafe { zs_zset_insert(z, key.as_ptr(), key.len(), 1) }, OK);
+    assert_eq!(unsafe { zs_circuit_push(c, input, z) }, OK);
+    unsafe { zs_zset_free(z) };
+}
+
+fn step_and_count(c: *mut ZsCircuit, sink: u32) -> usize {
+    assert_eq!(unsafe { zs_circuit_step(c) }, OK);
+    let mut out = null_mut();
+    assert_eq!(unsafe { zs_circuit_take(c, sink, &mut out) }, OK);
+    let mut n = 0;
+    let mut buf = [0u8; 64];
+    assert_eq!(
+        unsafe { zs_zset_encode(out, buf.as_mut_ptr(), buf.len(), &mut n) },
+        OK
+    );
+    unsafe { zs_zset_free(out) };
+    n / 13
+}
+
+#[test]
+fn a_circuit_stepped_on_two_threads_runs_its_callbacks_on_the_thread_that_steps() {
+    let seen = Seen(Mutex::new(Vec::new()));
+    let ctx = &seen as *const Seen as *mut c_void;
+    let (mut c, mut input, mut kept, mut sink) = (null_mut(), 0, 0, 0);
+    assert_eq!(unsafe { zs_circuit_new(&mut c) }, OK);
+    assert_eq!(unsafe { zs_circuit_input(c, &mut input) }, OK);
+    assert_eq!(
+        unsafe { zs_circuit_filter(c, input, Some(keep_all), ctx, &mut kept) },
+        OK
+    );
+    assert_eq!(unsafe { zs_circuit_output(c, kept, &mut sink) }, OK);
+
+    push_key(c, input, b"a");
+    assert_eq!(step_and_count(c, sink), 1);
+    let here = thread::current().id();
+
+    let moved = Moved(c);
+    let there = thread::spawn(move || {
+        let moved = moved;
+        push_key(moved.0, input, b"b");
+        let n = step_and_count(moved.0, sink);
+        (n, thread::current().id(), moved)
+    })
+    .join()
+    .unwrap();
+    assert_eq!(
+        there.0, 1,
+        "[threads] the second thread's step saw its own push"
+    );
+    assert_ne!(there.1, here);
+
+    let ids = seen.0.lock().unwrap().clone();
+    assert_eq!(
+        ids,
+        vec![here, there.1],
+        "[threads] each callback ran where it was stepped"
+    );
+    unsafe { zs_circuit_free(there.2.0) };
+}

crates/zset/tests/circuit.rs

@@ -4,7 +4,8 @@
 use proptest::prelude::*;
 use std::cell::Cell;
 use std::ops::Bound;
-use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use zset::{
     AggregateState, Circuit, MemTrace, Progress, Rebuild, Stream, Trace, ZSet, differentiate,
     integrate,
@@ -33,8 +34,8 @@ fn matches_recompute<K, L>(
     build: impl FnOnce(&mut Circuit, &Stream<Z<K>>) -> Stream<Z<L>>,
     reference: impl Fn(&Z<K>) -> Z<L>,
 ) where
-    K: Ord + Clone + std::fmt::Debug + 'static,
-    L: Ord + Clone + std::fmt::Debug + 'static,
+    K: Ord + Clone + std::fmt::Debug + Send + 'static,
+    L: Ord + Clone + std::fmt::Debug + Send + 'static,
 {
     let mut c = Circuit::new();
     let (input, s) = c.input::<K>();
@@ -42,10 +43,10 @@ fn matches_recompute<K, L>(
     let sink = c.output(&out);
     let (mut fed, mut got) = (Z::new(), Z::new());
     for (i, d) in steps.iter().enumerate() {
-        input.push(d.clone());
+        c.push(&input, d.clone());
         c.step();
         fed += d;
-        got += sink.take();
+        got += c.take(&sink);
         assert_eq!(got, reference(&fed), "[incremental] after step {i}");
     }
 }
@@ -104,10 +105,10 @@ fn join_circuit(da: &[Z<(u8, u8)>], db: &[Z<(u8, u8)>]) -> Vec<Z<(u8, u8, u8)>>
     let sink = c.output(&j);
     let mut deltas = Vec::new();
     for i in 0..da.len().max(db.len()) {
-        ia.push(da.get(i).cloned().unwrap_or_default());
-        ib.push(db.get(i).cloned().unwrap_or_default());
+        c.push(&ia, da.get(i).cloned().unwrap_or_default());
+        c.push(&ib, db.get(i).cloned().unwrap_or_default());
         c.step();
-        deltas.push(sink.take());
+        deltas.push(c.take(&sink));
     }
     deltas
 }
@@ -210,10 +211,10 @@ proptest! {
         let (sink_s, sink_back) = (c.output(&s), c.output(&back));
         let (mut seen_s, mut seen_back) = (Vec::new(), Vec::new());
         for d in &steps {
-            input.push(d.clone());
+            c.push(&input, d.clone());
             c.step();
-            seen_s.push(s.current());
-            seen_back.push(back.current());
+            seen_s.push(c.current(&s));
+            seen_back.push(c.current(&back));
         }
         let _ = (sink_s, sink_back);
         prop_assert_eq!(seen_back, seen_s, "[circuit-i-d]");
@@ -227,9 +228,9 @@ proptest! {
         let i = c.integrate(&s);
         let back = c.differentiate(&i);
         for (n, d) in steps.iter().enumerate() {
-            input.push(d.clone());
+            c.push(&input, d.clone());
             c.step();
-            prop_assert_eq!(back.current(), d.clone(), "[circuit-d-i] step {}", n);
+            prop_assert_eq!(c.current(&back), d.clone(), "[circuit-d-i] step {}", n);
         }
     }
 
@@ -240,9 +241,9 @@ proptest! {
         let d = c.delay(&s);
         let mut prev = Z::new();
         for (n, step) in steps.iter().enumerate() {
-            input.push(step.clone());
+            c.push(&input, step.clone());
             c.step();
-            prop_assert_eq!(d.current(), prev.clone(), "[delay] step {}", n);
+            prop_assert_eq!(c.current(&d), prev.clone(), "[delay] step {}", n);
             prev = step.clone();
         }
     }
@@ -263,9 +264,9 @@ proptest! {
         let (input, s) = whole.input::<u8>();
         let out = build(&mut whole, &s);
         let sink = whole.output(&out);
-        input.push(snapshot.clone());
+        whole.push(&input, snapshot.clone());
         whole.step();
-        let want = sink.take();
+        let want = whole.take(&sink);
 
         let mut c = Circuit::new();
         let (input, s) = c.input::<u8>();
@@ -282,7 +283,7 @@ proptest! {
                 left.set(left.get() - 1);
                 left.get() == 0
             });
-            got += sink.take();
+            got += c.take(&sink);
             if progress == Progress::Done { break; }
         }
         prop_assert_eq!(got, want, "[rebuild]");
@@ -315,9 +316,9 @@ fn a_rebuild_that_is_out_of_time_reports_pending_after_one_chunk() {
     let sink = c.output(&s);
     let mut job = Rebuild::new(input, (0..6u8).map(|k| (k, 1)).collect(), 2);
     assert_eq!(job.step(&mut c, || true), Progress::Pending);
-    assert_eq!(sink.take().len(), 2, "exactly one chunk ran");
+    assert_eq!(c.take(&sink).len(), 2, "exactly one chunk ran");
     assert_eq!(job.step(&mut c, || false), Progress::Done);
-    assert_eq!(sink.take().len(), 4);
+    assert_eq!(c.take(&sink).len(), 4);
     assert_eq!(job.step(&mut c, || true), Progress::Done);
 }
 
@@ -332,20 +333,20 @@ fn an_unbounded_range_that_is_empty_reads_nothing() {
 
 #[test]
 fn the_join_reads_state_through_the_trace_it_was_given() {
-    struct Counting(MemTrace<u8, u8>, Rc<Cell<usize>>);
+    struct Counting(MemTrace<u8, u8>, Arc<AtomicUsize>);
     impl Trace<u8, u8> for Counting {
         fn apply(&mut self, k: &u8, v: &u8, w: i64) {
             self.0.apply(k, v, w)
         }
         fn lookup(&self, k: &u8) -> Vec<(u8, i64)> {
-            self.1.set(self.1.get() + 1);
+            self.1.fetch_add(1, Ordering::Relaxed);
             self.0.lookup(k)
         }
         fn range(&self, lo: Bound<&u8>, hi: Bound<&u8>) -> Vec<(u8, u8, i64)> {
             self.0.range(lo, hi)
         }
     }
-    let looked = Rc::new(Cell::new(0));
+    let looked = Arc::new(AtomicUsize::new(0));
     let mut c = Circuit::new();
     let (ia, a) = c.input::<(u8, u8)>();
     let (ib, b) = c.input::<(u8, u8)>();
@@ -356,11 +357,11 @@ fn the_join_reads_state_through_the_trace_it_was_given() {
         MemTrace::new(),
         |k, x, y| (*k, *x, *y),
     );
-    ia.push(Z::singleton((1, 1), 1));
-    ib.push(Z::singleton((1, 2), 1));
+    c.push(&ia, Z::singleton((1, 1), 1));
+    c.push(&ib, Z::singleton((1, 2), 1));
     c.step();
-    assert_eq!(j.current(), Z::singleton((1, 1, 2), 1));
-    assert!(looked.get() > 0);
+    assert_eq!(c.current(&j), Z::singleton((1, 1, 2), 1));
+    assert!(looked.load(Ordering::Relaxed) > 0);
 }
 
 #[test]
@@ -398,9 +399,9 @@ fn loaded(snapshot: &Z<u8>, chunk: usize, budget: usize) -> (Z<u8>, Z<u8>, Z<u8>
     let (input, s) = c.input::<u8>();
     let (d, i, df) = (c.delay(&s), c.integrate(&s), c.differentiate(&s));
     let sinks = (c.output(&d), c.output(&i), c.output(&df));
-    input.push(Z::singleton(9, 1));
+    c.push(&input, Z::singleton(9, 1));
     c.step();
-    let _ = (sinks.0.take(), sinks.1.take(), sinks.2.take());
+    let _ = (c.take(&sinks.0), c.take(&sinks.1), c.take(&sinks.2));
     let mut job = Rebuild::new(input, snapshot.clone(), chunk);
     for _ in 0..50 {
         let left = Cell::new(budget);
@@ -412,7 +413,7 @@ fn loaded(snapshot: &Z<u8>, chunk: usize, budget: usize) -> (Z<u8>, Z<u8>, Z<u8>
             break;
         }
     }
-    (sinks.0.take(), sinks.1.take(), sinks.2.take())
+    (c.take(&sinks.0), c.take(&sinks.1), c.take(&sinks.2))
 }
 
 #[test]
@@ -432,13 +433,17 @@ fn an_empty_snapshot_is_still_one_tick() {
     let mut c = Circuit::new();
     let (input, s) = c.input::<u8>();
     let d = c.delay(&s);
-    input.push(Z::singleton(1, 1));
+    c.push(&input, Z::singleton(1, 1));
     c.step();
     let mut job = Rebuild::new(input, Z::new(), 3);
     assert_eq!(job.step(&mut c, || false), Progress::Done);
-    assert_eq!(d.current().len(), 1, "the empty load ticked the clock once");
+    assert_eq!(
+        c.current(&d).len(),
+        1,
+        "the empty load ticked the clock once"
+    );
     assert_eq!(job.step(&mut c, || false), Progress::Done);
-    assert_eq!(d.current().len(), 1, "a finished job does not tick again");
+    assert_eq!(c.current(&d).len(), 1, "a finished job does not tick again");
 }
 
 #[test]
@@ -460,21 +465,21 @@ proptest! {
             let (input, s) = c.input::<u8>();
             let out = temporal(&mut c, &s);
             let sink = c.output(&out);
-            input.push(prior.clone());
+            c.push(&input, prior.clone());
             c.step();
-            let _ = sink.take();
-            let input_again = input.clone();
+            let _ = c.take(&sink);
+            let input_again = input;
             let mut job = Rebuild::new(input, snapshot.clone(), chunk);
             for _ in 0..50 {
                 let left = Cell::new(budget);
                 if job.step(&mut c, || { left.set(left.get() - 1); left.get() == 0 }) == Progress::Done { break; }
             }
-            let at = sink.take();
-            input_again.push(Z::singleton(5, 1));
+            let at = c.take(&sink);
+            c.push(&input_again, Z::singleton(5, 1));
             c.step();
-            let next = sink.take();
+            let next = c.take(&sink);
             c.step();
-            (at, next, sink.take())
+            (at, next, c.take(&sink))
         };
         prop_assert_eq!(run(chunk, budget), run(100, 100), "[temporal-rebuild]");
     }

crates/zset/tests/clone_panic.rs

@@ -55,7 +55,7 @@ fn rig(which: u8) -> Rig {
     let mut c = Circuit::new();
     let (a, sa) = c.input::<B>();
     let (b, sb) = c.input::<B>();
-    let mut streams = vec![sa.clone(), sb.clone()];
+    let mut streams = vec![sa, sb];
     if which & 1 != 0 {
         streams.push(c.integrate(&sa));
     }
@@ -84,14 +84,20 @@ impl Rig {
     fn view(&self) -> String {
         format!(
             "{:?} {:?}",
-            self.streams.iter().map(|s| s.current()).collect::<Vec<_>>(),
-            self.agg.current()
+            self.streams
+                .iter()
+                .map(|s| self.c.current(s))
+                .collect::<Vec<_>>(),
+            self.c.current(&self.agg)
         )
     }
-    fn drain(&self) -> String {
+    fn drain(&mut self) -> String {
         format!(
             "{:?}",
-            self.sinks.iter().map(|s| s.take()).collect::<Vec<_>>()
+            self.sinks
+                .iter()
+                .map(|s| self.c.take(s))
+                .collect::<Vec<_>>()
         )
     }
 }
@@ -107,9 +113,9 @@ fn clone_panic_sweep(which: u8) -> Vec<String> {
     let mut failures = vec![];
     for n in 0..400u32 {
         let (mut x, mut y) = (rig(which), rig(which));
-        for r in [&x, &y] {
-            r.a.push(bz(&[(1, 1, 1), (2, 2, 1)]));
-            r.b.push(bz(&[(1, 3, 1), (3, 3, 1)]));
+        for r in [&mut x, &mut y] {
+            r.c.push(&r.a, bz(&[(1, 1, 1), (2, 2, 1)]));
+            r.c.push(&r.b, bz(&[(1, 3, 1), (3, 3, 1)]));
         }
         x.c.step();
         y.c.step();
@@ -117,8 +123,8 @@ fn clone_panic_sweep(which: u8) -> Vec<String> {
         let (v0, d0) = (x.view(), x.drain());
         y.drain();
         x.c.begin();
-        x.a.push(bz(&[(1, 5, 1), (4, 4, 1), (2, 2, -1)]));
-        x.b.push(bz(&[(4, 6, 1), (1, 3, -1)]));
+        x.c.push(&x.a, bz(&[(1, 5, 1), (4, 4, 1), (2, 2, -1)]));
+        x.c.push(&x.b, bz(&[(4, 6, 1), (1, 3, -1)]));
         CLONE_TRIP.with(|t| t.set(Some(n)));
         let r = catch_unwind(AssertUnwindSafe(|| x.c.step()));
         let fired = CLONE_TRIP.with(|t| t.get()).is_none();
@@ -135,16 +141,13 @@ fn clone_panic_sweep(which: u8) -> Vec<String> {
             ));
             continue;
         }
-        for r in [&x, &y] {
+        for r in [&mut x, &mut y] {
             // The keys the failed step touched come back, so state leaked by one shows.
-            r.a.push(bz(&[
-                (1, 7, 1),
-                (4, 1, 1),
-                (1, 5, 1),
-                (4, 4, 1),
-                (2, 2, -1),
-            ]));
-            r.b.push(bz(&[(4, 2, 1), (1, 9, 1)]));
+            r.c.push(
+                &r.a,
+                bz(&[(1, 7, 1), (4, 1, 1), (1, 5, 1), (4, 4, 1), (2, 2, -1)]),
+            );
+            r.c.push(&r.b, bz(&[(4, 2, 1), (1, 9, 1)]));
         }
         x.c.step();
         y.c.step();

crates/zset/tests/cost.rs

@@ -53,7 +53,7 @@ fn an_eval_borrows_its_input_instead_of_cloning_it() {
     let (input, s) = c.input::<Counted>();
     let m = c.map(&s, |k| k.0);
     let sink = c.output(&m);
-    input.push(keys(N));
+    c.push(&input, keys(N));
     let before = clones();
     c.step();
     assert_eq!(
@@ -61,7 +61,7 @@ fn an_eval_borrows_its_input_instead_of_cloning_it() {
         0,
         "[piece] a step over {N} keys cloned keys"
     );
-    assert_eq!(sink.take().len(), N as usize);
+    assert_eq!(c.take(&sink).len(), N as usize);
 }
 
 #[test]
@@ -70,7 +70,7 @@ fn an_integrate_stream_holds_its_integral_twice_when_the_tick_is_whole() {
     let mut c = Circuit::new();
     let (input, s) = c.input::<Counted>();
     let i = c.integrate(&s);
-    input.push(keys(N));
+    c.push(&input, keys(N));
     c.step();
     // The input's piece (N), the integral (N) and the integrate stream's piece (N).
     assert_eq!(
@@ -105,25 +105,25 @@ fn clones_of_one_step(n: u32, mode: Mode) -> usize {
     let sinks = (c.output(&d), c.output(&dl), c.output(&j), c.output(&g));
     let rows =
         |lo: u32, hi: u32| -> ZSet<E> { (lo..hi).map(|i| ((Counted::new(i), 0), 1)).collect() };
-    a.push(rows(0, n));
-    b.push(rows(0, n));
+    c.push(&a, rows(0, n));
+    c.push(&b, rows(0, n));
     c.step();
     for z in [&sinks.0, &sinks.1] {
-        z.take();
+        c.take(z);
     }
-    sinks.2.take();
-    sinks.3.take();
+    c.take(&sinks.2);
+    c.take(&sinks.3);
     // A small step after the load: a `delay` holds the previous tick's input, so a
     // transaction opened right after a big load would copy that load once.
-    a.push(rows(n, n + 2));
-    b.push(rows(n, n + 2));
+    c.push(&a, rows(n, n + 2));
+    c.push(&b, rows(n, n + 2));
     c.step();
     let before = clones();
     if !matches!(mode, Mode::Plain) {
         c.begin();
     }
-    a.push(rows(n + 2, n + 4));
-    b.push(rows(n + 2, n + 4));
+    c.push(&a, rows(n + 2, n + 4));
+    c.push(&b, rows(n + 2, n + 4));
     c.step();
     match mode {
         Mode::Plain => {}
@@ -148,13 +148,13 @@ fn clones_of_one_continue(n: u32, txn: bool) -> usize {
     let mut c = Circuit::new();
     let (input, s) = c.input::<Counted>();
     let _i = c.integrate(&s);
-    input.push((0..n).map(|k| (Counted::new(k), 1)).collect());
+    c.push(&input, (0..n).map(|k| (Counted::new(k), 1)).collect());
     c.step();
     let before = clones();
     if txn {
         c.begin();
     }
-    input.push((n..n + 1).map(|k| (Counted::new(k), 1)).collect());
+    c.push(&input, (n..n + 1).map(|k| (Counted::new(k), 1)).collect());
     c.step_continue();
     if txn {
         c.commit();

crates/zset/tests/rebuild.rs

@@ -12,7 +12,9 @@ fn pairs() -> impl Strategy<Value = Z<u8>> {
 }
 
 /// A pipeline using every operator; returns the input and a closure that drains all sinks.
-fn build(c: &mut Circuit) -> (Input<u8>, Box<dyn Fn() -> String>) {
+type Drain = Box<dyn Fn(&mut Circuit) -> String>;
+
+fn build(c: &mut Circuit) -> (Input<u8>, Drain) {
     let (input, s) = c.input::<u8>();
     let d = c.delay(&s);
     let i = c.integrate(&s);
@@ -40,18 +42,18 @@ fn build(c: &mut Circuit) -> (Input<u8>, Box<dyn Fn() -> String>) {
         c.output(&ddist),
         c.output(&fm),
     );
-    let drain = move || {
+    let drain = move |c: &mut Circuit| {
         format!(
             "d={:?} i={:?} df={:?} j={:?} ij={:?} agg={:?} dagg={:?} ddist={:?} fm={:?}",
-            sinks.0.take(),
-            sinks.1.take(),
-            sinks.2.take(),
-            sinks.3.take(),
-            sinks.4.take(),
-            sinks.5.take(),
-            sinks.6.take(),
-            sinks.7.take(),
-            sinks.8.take()
+            c.take(&sinks.0),
+            c.take(&sinks.1),
+            c.take(&sinks.2),
+            c.take(&sinks.3),
+            c.take(&sinks.4),
+            c.take(&sinks.5),
+            c.take(&sinks.6),
+            c.take(&sinks.7),
+            c.take(&sinks.8)
         )
     };
     (input, Box::new(drain))
@@ -66,10 +68,10 @@ fn run(
 ) -> (String, String, String) {
     let mut c = Circuit::new();
     let (input, drain) = build(&mut c);
-    input.push(prior.clone());
+    c.push(&input, prior.clone());
     c.step();
-    let _ = drain();
-    let mut job = Rebuild::new(input.clone(), snap.clone(), chunk);
+    let _ = drain(&mut c);
+    let mut job = Rebuild::new(input, snap.clone(), chunk);
     let mut done = false;
     for _ in 0..50 {
         let left = Cell::new(budget);
@@ -83,12 +85,12 @@ fn run(
         }
     }
     assert!(done, "[rebuild] never finished");
-    let at = drain();
-    input.push(post.clone());
+    let at = drain(&mut c);
+    c.push(&input, post.clone());
     c.step();
-    let next = drain();
+    let next = drain(&mut c);
     c.step();
-    (at, next, drain())
+    (at, next, drain(&mut c))
 }
 
 proptest! {
@@ -113,7 +115,7 @@ fn current_after_a_chunked_rebuild_is_the_whole_ticks_value() {
         let i = c.integrate(&s);
         let mut job = Rebuild::new(input, snap.clone(), chunk);
         while job.step(&mut c, || false) == Progress::Pending {}
-        i.current()
+        c.current(&i)
     };
     assert_eq!(cur(1), cur(10), "[current]");
 }
@@ -168,7 +170,7 @@ fn a_rebuild_stepped_on_a_foreign_circuit_is_refused_loudly() {
 fn a_rebuild_knows_the_circuit_its_input_came_from() {
     let (mut a, b) = (Circuit::new(), Circuit::new());
     let (input, _s) = a.input::<u8>();
-    let job = Rebuild::new(input.clone(), ZSet::new(), 1);
+    let job = Rebuild::new(input, ZSet::new(), 1);
     assert!(
         job.belongs_to(&a),
         "[owner] the circuit that made the input"
@@ -182,7 +184,7 @@ fn an_interrupted_rebuild_stays_interrupted_and_a_new_job_loads_the_snapshot() {
     let mut c = Circuit::new();
     let (input, s) = c.input::<u8>();
     let i = c.integrate(&s);
-    let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
+    let mut job = Rebuild::new(input, snap.clone(), 1);
     assert_eq!(job.step(&mut c, || true), Progress::Pending);
     c.step();
     for _ in 0..3 {
@@ -191,5 +193,5 @@ fn an_interrupted_rebuild_stays_interrupted_and_a_new_job_loads_the_snapshot() {
     }
     let mut fresh = Rebuild::new(input, snap, 1);
     assert_eq!(fresh.step(&mut c, || false), Progress::Done);
-    assert_eq!(i.current().len(), 3);
+    assert_eq!(c.current(&i).len(), 3);
 }

crates/zset/tests/rollback.rs

@@ -14,8 +14,8 @@ struct Rig {
     c: Circuit,
     a: Input<P>,
     b: Input<P>,
-    view: Box<dyn Fn() -> String>,
-    drain: Box<dyn Fn() -> String>,
+    view: Box<dyn Fn(&Circuit) -> String>,
+    drain: Box<dyn Fn(&mut Circuit) -> String>,
 }
 
 fn rig() -> Rig {
@@ -44,34 +44,34 @@ fn rig() -> Rig {
         c.output(&ag),
         c.output(&pl),
     );
-    let (s1, s2) = (sa.clone(), sb.clone());
-    let view = move || {
+    let (s1, s2) = (sa, sb);
+    let view = move |c: &Circuit| {
         format!(
             "{:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?}",
-            s1.current(),
-            s2.current(),
-            ia.current(),
-            da.current(),
-            df.current(),
-            ds.current(),
-            j.current(),
-            ij.current(),
-            ag.current(),
-            pl.current(),
-            s1.current().len(),
+            c.current(&s1),
+            c.current(&s2),
+            c.current(&ia),
+            c.current(&da),
+            c.current(&df),
+            c.current(&ds),
+            c.current(&j),
+            c.current(&ij),
+            c.current(&ag),
+            c.current(&pl),
+            c.current(&s1).len(),
         )
     };
-    let drain = move || {
+    let drain = move |c: &mut Circuit| {
         format!(
             "{:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?}",
-            sinks.0.take(),
-            sinks.1.take(),
-            sinks.2.take(),
-            sinks.3.take(),
-            sinks.4.take(),
-            sinks.5.take(),
-            sinks.6.take(),
-            sinks.7.take(),
+            c.take(&sinks.0),
+            c.take(&sinks.1),
+            c.take(&sinks.2),
+            c.take(&sinks.3),
+            c.take(&sinks.4),
+            c.take(&sinks.5),
+            c.take(&sinks.6),
+            c.take(&sinks.7),
         )
     };
     Rig {
@@ -85,8 +85,8 @@ fn rig() -> Rig {
 
 impl Rig {
     fn tick(&mut self, a: &Z<P>, b: &Z<P>) {
-        self.a.push(a.clone());
-        self.b.push(b.clone());
+        self.c.push(&self.a, a.clone());
+        self.c.push(&self.b, b.clone());
         self.c.step();
     }
 }
@@ -123,31 +123,31 @@ proptest! {
         for (a, b) in pre.iter().chain([&(preload, Z::new())]) {
             x.tick(a, b);
             y.tick(a, b);
-            prop_assert_eq!((x.view)(), (y.view)(), "[pre] state");
-            prop_assert_eq!((x.drain)(), (y.drain)(), "[pre] output");
+            prop_assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[pre] state");
+            prop_assert_eq!((x.drain)(&mut x.c), (y.drain)(&mut y.c), "[pre] output");
         }
-        x.a.push(carry.clone());
-        y.a.push(carry);
-        let (view_before, drain_before) = ((x.view)(), (x.drain)());
+        x.c.push(&x.a, carry.clone());
+        y.c.push(&y.a, carry);
+        let (view_before, drain_before) = ((x.view)(&x.c), (x.drain)(&mut x.c));
 
         x.c.begin();
         let mut a = bad.0.clone();
         if fail {
             a += poison();
         }
-        x.a.push(a);
-        x.b.push(bad.1.clone());
+        x.c.push(&x.a, a);
+        x.c.push(&x.b, bad.1.clone());
         let outcome = catch_unwind(AssertUnwindSafe(|| x.c.step()));
         prop_assert_eq!(outcome.is_err(), fail, "[panic] the poison step panics, the others do not");
         x.c.rollback();
 
-        prop_assert_eq!((x.view)(), view_before, "[rollback] every stream's current()");
-        prop_assert_eq!((x.drain)(), drain_before, "[rollback] every sink");
+        prop_assert_eq!((x.view)(&x.c), view_before, "[rollback] every stream's current()");
+        prop_assert_eq!((x.drain)(&mut x.c), drain_before, "[rollback] every sink");
         for (i, (a, b)) in post.iter().enumerate() {
             x.tick(a, b);
             y.tick(a, b);
-            prop_assert_eq!((x.view)(), (y.view)(), "[post] state, tick {}", i);
-            prop_assert_eq!((x.drain)(), (y.drain)(), "[post] output, tick {}", i);
+            prop_assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[post] state, tick {}", i);
+            prop_assert_eq!((x.drain)(&mut x.c), (y.drain)(&mut y.c), "[post] output, tick {}", i);
         }
     }
 
@@ -168,8 +168,8 @@ proptest! {
                 x.c.commit();
             }
             y.tick(a, b);
-            prop_assert_eq!((x.view)(), (y.view)(), "[commit] state, tick {}", i);
-            prop_assert_eq!((x.drain)(), (y.drain)(), "[commit] output, tick {}", i);
+            prop_assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[commit] state, tick {}", i);
+            prop_assert_eq!((x.drain)(&mut x.c), (y.drain)(&mut y.c), "[commit] output, tick {}", i);
         }
     }
 }
@@ -180,14 +180,14 @@ fn an_operator_built_inside_a_transaction_goes_back_to_empty() {
     let (input, s) = c.input::<u8>();
     c.begin();
     let i = c.integrate(&s);
-    input.push([(1u8, 1)].into_iter().collect());
+    c.push(&input, [(1u8, 1)].into_iter().collect());
     c.step();
     c.rollback();
-    input.push([(2u8, 1)].into_iter().collect());
+    c.push(&input, [(2u8, 1)].into_iter().collect());
     c.step();
     let want: Z<u8> = [(2u8, 1)].into_iter().collect();
     assert_eq!(
-        i.current(),
+        c.current(&i),
         want,
         "[built] the integral forgot the rolled-back tick"
     );
@@ -202,24 +202,24 @@ fn rollback_undoes_newest_first_so_no_undo_leaves_i64() {
     let (i, d) = (c.integrate(&s), c.distinct(&s));
     let g = c.aggregate(&s, |_, vals| Some(vals.len() as i64));
     let one = |w: i64| -> Z<P> { [((1, 1), w)].into_iter().collect() };
-    input.push(one(i64::MAX));
+    c.push(&input, one(i64::MAX));
     c.step();
-    let before = (i.current(), d.current(), g.current());
+    let before = (c.current(&i), c.current(&d), c.current(&g));
     c.begin();
     for w in [-1, 1] {
-        input.push(one(w));
+        c.push(&input, one(w));
         c.step();
     }
     c.rollback();
     assert_eq!(
-        (i.current(), d.current(), g.current()),
+        (c.current(&i), c.current(&d), c.current(&g)),
         before,
         "[order] streams"
     );
-    input.push(one(-1));
+    c.push(&input, one(-1));
     c.step();
     assert_eq!(
-        i.current(),
+        c.current(&i),
         one(i64::MAX - 1),
         "[order] the integral is back at MAX first"
     );
@@ -237,10 +237,10 @@ fn a_minimum_weight_applied_to_every_kind_of_state_is_taken_back() {
     let right: Z<P> = [((1, 2), 1)].into_iter().collect();
     x.tick(&Z::new(), &right);
     y.tick(&Z::new(), &right);
-    assert_eq!((x.view)(), (y.view)(), "[min] state");
+    assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[min] state");
     assert_eq!(
-        (x.drain)(),
-        (y.drain)(),
+        (x.drain)(&mut x.c),
+        (y.drain)(&mut y.c),
         "[min] output: the join saw no left side"
     );
 }
@@ -251,7 +251,7 @@ fn rollback_undoes_the_clock() {
     let (input, _s) = c.input::<u8>();
     c.step();
     c.begin();
-    input.push([(1u8, 1)].into_iter().collect());
+    c.push(&input, [(1u8, 1)].into_iter().collect());
     c.step();
     assert_eq!(c.ticks(), 2);
     c.rollback();
@@ -287,23 +287,27 @@ fn a_rollback_across_a_split_tick_restores_the_whole_value() {
     y.tick(&first, &Z::new());
     // The tick is already split when the transaction opens, so every edge holds a `whole`.
     for r in [&mut x, &mut y] {
-        r.a.push([((3, 1), 1)].into_iter().collect());
+        r.c.push(&r.a, [((3, 1), 1)].into_iter().collect());
         r.c.step_continue();
     }
     x.c.begin();
     for k in [4u8, 5] {
-        x.a.push([((k, 1), 1)].into_iter().collect());
+        x.c.push(&x.a, [((k, 1), 1)].into_iter().collect());
         x.c.step_continue();
     }
     x.c.rollback();
-    assert_eq!((x.view)(), (y.view)(), "[split] state after the rollback");
     assert_eq!(
-        (x.drain)(),
-        (y.drain)(),
+        (x.view)(&x.c),
+        (y.view)(&y.c),
+        "[split] state after the rollback"
+    );
+    assert_eq!(
+        (x.drain)(&mut x.c),
+        (y.drain)(&mut y.c),
         "[split] output after the rollback"
     );
     let next: Z<P> = [((5, 1), 1)].into_iter().collect();
     x.tick(&next, &Z::new());
     y.tick(&next, &Z::new());
-    assert_eq!((x.view)(), (y.view)(), "[split] the tick after");
+    assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[split] the tick after");
 }

crates/zset/tests/rollback_model.rs

@@ -120,7 +120,7 @@ impl World {
         struct S<'a>(&'a [Stream<Z>]);
         impl S<'_> {
             fn s(&self, i: usize) -> Stream<Z> {
-                self.0[i % self.0.len()].clone()
+                self.0[i % self.0.len()]
             }
         }
         let this = S(st);
@@ -199,10 +199,10 @@ impl World {
         self.sinks.push(o);
     }
     fn view(&self) -> Vec<Z> {
-        self.streams.iter().map(|s| s.current()).collect()
+        self.streams.iter().map(|s| self.c.current(s)).collect()
     }
-    fn drain(&self) -> Vec<Z> {
-        self.sinks.iter().map(|s| s.take()).collect()
+    fn drain(&mut self) -> Vec<Z> {
+        self.sinks.iter().map(|s| self.c.take(s)).collect()
     }
 }
 
@@ -274,7 +274,7 @@ fn act() -> impl Strategy<Value = Act> {
 
 fn apply_plain(w: &mut World, a: &Act) {
     match a {
-        Act::Push(b, z) => w.inputs[*b as usize].push(z.clone()),
+        Act::Push(b, z) => w.c.push(&w.inputs[*b as usize], z.clone()),
         Act::Step(_) => w.c.step(),
         Act::Cont(_) => w.c.step_continue(),
         Act::Build(op) => w.build(op),
@@ -282,7 +282,7 @@ fn apply_plain(w: &mut World, a: &Act) {
     }
 }
 
-fn check(x: &World, y: &World, at: &str) -> Result<(), TestCaseError> {
+fn check(x: &mut World, y: &mut World, at: &str) -> Result<(), TestCaseError> {
     prop_assert_eq!(x.c.ticks(), y.c.ticks(), "ticks {}", at);
     prop_assert_eq!(x.view(), y.view(), "view {}", at);
     prop_assert_eq!(x.drain(), y.drain(), "sinks {}", at);
@@ -327,7 +327,7 @@ fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> {
                 if steps_in_txn > 1 {
                     MULTI_STEP_TXN.fetch_add(1, Relaxed);
                 }
-                check(&x, &y, &at)?;
+                check(&mut x, &mut y, &at)?;
             }
             Act::Commit | Act::Rollback if open => {
                 x.c.rollback();
@@ -348,7 +348,7 @@ fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> {
                         y.build(&op);
                     }
                 }
-                check(&x, &y, &at)?;
+                check(&mut x, &mut y, &at)?;
             }
             Act::Commit | Act::Rollback => {}
             _ if open && failed => {} // only rollback after a failed step
@@ -388,7 +388,7 @@ fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> {
                     return Ok(());
                 }
                 apply_plain(&mut y, a);
-                check(&x, &y, &at)?;
+                check(&mut x, &mut y, &at)?;
             }
         }
     }
@@ -404,23 +404,23 @@ fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> {
         } else {
             RB_CLEAN.fetch_add(1, Relaxed);
         }
-        check(&x, &y, "final rollback")?;
+        check(&mut x, &mut y, "final rollback")?;
     }
     // ticks after: both behave the same on a fresh tick
     for i in 0..3u8 {
         let z: Z = [((i, i), 1), ((i + 1, 0), -1)].into_iter().collect();
         let r = catch_unwind(AssertUnwindSafe(|| {
-            x.inputs[0].push(z.clone());
-            x.inputs[1].push(z.clone());
+            x.c.push(&x.inputs[0], z.clone());
+            x.c.push(&x.inputs[1], z.clone());
             x.c.step()
         }));
         if r.is_err() {
             return Ok(());
         }
-        y.inputs[0].push(z.clone());
-        y.inputs[1].push(z);
+        y.c.push(&y.inputs[0], z.clone());
+        y.c.push(&y.inputs[1], z);
         y.c.step();
-        check(&x, &y, &format!("post tick {i}"))?;
+        check(&mut x, &mut y, &format!("post tick {i}"))?;
     }
     Ok(())
 }

crates/zset/tests/rollback_rebuild.rs

@@ -30,7 +30,7 @@ fn r1_rebuild_begun_before_txn_continued_inside_then_rolled_back_reports_interru
     let (input, s) = c.input::<u8>();
     let i = c.integrate(&s);
     let snap = z(&[(1, 1), (2, 1), (3, 1), (4, 1)]);
-    let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
+    let mut job = Rebuild::new(input, snap.clone(), 1);
     assert_eq!(job.step(&mut c, || true), Progress::Pending);
     c.begin();
     assert_eq!(job.step(&mut c, || true), Progress::Pending);
@@ -40,7 +40,7 @@ fn r1_rebuild_begun_before_txn_continued_inside_then_rolled_back_reports_interru
     let mut finished = None;
     if !interrupted {
         while job.step(&mut c, || false) == Progress::Pending {}
-        finished = Some(i.current());
+        finished = Some(c.current(&i));
     }
     assert!(
         interrupted,
@@ -55,7 +55,7 @@ fn r2_rebuild_finished_inside_a_rolled_back_txn_is_reported_done_with_nothing_lo
     let (input, s) = c.input::<u8>();
     let i = c.integrate(&s);
     let snap = z(&[(1, 1), (2, 1), (3, 1)]);
-    let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
+    let mut job = Rebuild::new(input, snap.clone(), 1);
     c.begin();
     assert_eq!(job.step(&mut c, || false), Progress::Done);
     c.rollback();
@@ -63,7 +63,7 @@ fn r2_rebuild_finished_inside_a_rolled_back_txn_is_reported_done_with_nothing_lo
     assert!(
         interrupted,
         "[r2] after rollback: not interrupted, integral {:?}",
-        i.current()
+        c.current(&i)
     );
 }
 
@@ -74,7 +74,7 @@ fn r3_an_interrupted_rebuild_resurrects_after_rollback_and_a_later_step() {
     let i = c.integrate(&s);
     c.step(); // ticks 1
     let snap = z(&[(1, 1), (2, 1), (3, 1), (4, 1)]);
-    let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
+    let mut job = Rebuild::new(input, snap.clone(), 1);
     c.begin();
     assert_eq!(job.step(&mut c, || true), Progress::Pending); // began_at = 2
     c.rollback(); // ticks back to 1
@@ -84,7 +84,7 @@ fn r3_an_interrupted_rebuild_resurrects_after_rollback_and_a_later_step() {
     let mut got = None;
     if !dead_after_step {
         while job.step(&mut c, || false) == Progress::Pending {}
-        got = Some(i.current());
+        got = Some(c.current(&i));
     }
     assert!(
         dead_after_step,
@@ -98,10 +98,10 @@ fn r4_rebuild_whose_first_chunk_panics_in_a_txn_then_rolled_back() {
     let mut c = Circuit::new();
     let (input, s) = c.input::<u8>();
     let i = c.integrate(&s);
-    input.push(z(&[(1, i64::MAX)]));
+    c.push(&input, z(&[(1, i64::MAX)]));
     c.step();
     let snap = z(&[(1, 1), (2, 1), (3, 1)]);
-    let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
+    let mut job = Rebuild::new(input, snap.clone(), 1);
     c.begin();
     let r = catch_unwind(AssertUnwindSafe(|| job.step(&mut c, || false)));
     assert!(r.is_err());
@@ -109,9 +109,9 @@ fn r4_rebuild_whose_first_chunk_panics_in_a_txn_then_rolled_back() {
     let interrupted = job.interrupted(&c);
     let mut got = None;
     if !interrupted {
-        input.push(z(&[(1, -5)])); // make room so the retry does not overflow
+        c.push(&input, z(&[(1, -5)])); // make room so the retry does not overflow
         while job.step(&mut c, || false) == Progress::Pending {}
-        got = Some(i.current());
+        got = Some(c.current(&i));
     }
     assert!(
         interrupted,
@@ -120,36 +120,13 @@ fn r4_rebuild_whose_first_chunk_panics_in_a_txn_then_rolled_back() {
     );
 }
 
-#[test]
-fn b1_a_closure_that_reads_a_stream_during_eval_does_not_panic() {
-    let mut c = Circuit::new();
-    let (input, s) = c.input::<u8>();
-    let s2 = s.clone();
-    let i = c.integrate(&s);
-    let i2 = i.clone();
-    // map reads its own source and another node's output during eval
-    let m = c.map(&s, move |k| {
-        k.wrapping_add(s2.current().len() as u8 + i2.current().len() as u8)
-    });
-    let m2 = m.clone();
-    // a later node reads an earlier node's output inside its own eval
-    let _f = c.filter(&i, move |_| m2.current().len() < 1000);
-    input.push(z(&[(1, 1), (2, 1)]));
-    c.step();
-    c.begin();
-    input.push(z(&[(3, 1)]));
-    c.step_continue();
-    c.rollback();
-    c.step();
-}
-
 #[test]
 fn p1_a_chunk_whose_push_panics_in_a_txn_is_counted() {
     let mut c = Circuit::new();
     let (input, s) = c.input::<u8>();
     let i = c.integrate(&s);
-    input.push(z(&[(1, i64::MAX)]));
-    let mut job = Rebuild::new(input.clone(), z(&[(1, 1), (2, 1)]), 1);
+    c.push(&input, z(&[(1, i64::MAX)]));
+    let mut job = Rebuild::new(input, z(&[(1, 1), (2, 1)]), 1);
     c.begin();
     let r = catch_unwind(AssertUnwindSafe(|| job.step(&mut c, || false)));
     assert!(r.is_err(), "the push overflows");
@@ -158,7 +135,7 @@ fn p1_a_chunk_whose_push_panics_in_a_txn_is_counted() {
     let mut got = None;
     if !interrupted {
         while job.step(&mut c, || false) == Progress::Pending {}
-        got = Some(i.current());
+        got = Some(c.current(&i));
     }
     assert!(
         interrupted,

crates/zset/tests/threads.rs

@@ -0,0 +1,104 @@
+//! A circuit moves between threads, one at a time (`context/code/zset.md`, "Threads").
+//! Evidence: `context/test-evidence/zset.md`.
+
+use std::thread;
+use zset::{Circuit, Input, Output, Rebuild, Stream, ZSet};
+
+type Z<K> = ZSet<K>;
+
+fn is_send<T: Send>() {}
+
+#[test]
+fn a_circuit_and_every_handle_it_hands_out_is_send() {
+    is_send::<Circuit>();
+    is_send::<Stream<Z<u8>>>();
+    is_send::<Input<u8>>();
+    is_send::<Output<u8>>();
+    is_send::<Rebuild<u8>>();
+}
+
+type Outs = (Output<(u8, i64)>, Output<(u8, u8)>);
+
+fn delta(n: u8) -> Z<(u8, u8)> {
+    [((n % 5, n), 1), (((n + 1) % 5, n / 2), -1)]
+        .into_iter()
+        .collect()
+}
+
+/// The pipeline's outputs after `ticks` steps, collecting what `take` returns each time.
+fn collect(c: &mut Circuit, i: &Input<(u8, u8)>, outs: &Outs, n: u8) -> String {
+    c.push(i, delta(n));
+    c.step();
+    format!("{:?} {:?}", c.take(&outs.0), c.take(&outs.1))
+}
+
+#[test]
+fn a_circuit_built_on_one_thread_steps_on_another_and_returns() {
+    let build = || {
+        let mut c = Circuit::new();
+        let (input, s) = c.input::<(u8, u8)>();
+        let i = c.integrate(&s);
+        let d = c.distinct(&i);
+        let g = c.aggregate(&d, |_, vals| {
+            Some(vals.iter().map(|(v, w)| i64::from(*v) * w).sum::<i64>())
+        });
+        let f = c.filter(&s, |(k, _)| k % 2 == 0);
+        let m = c.delay(&f);
+        let outs = (c.output(&g), c.output(&m));
+        (c, input, outs)
+    };
+    let (mut one, i1, o1) = build();
+    let want: Vec<String> = (0..9).map(|n| collect(&mut one, &i1, &o1, n)).collect();
+
+    let (mut c, input, outs) = build();
+    let mut got = Vec::new();
+    for n in 0..9 {
+        // Each tick runs on a fresh thread: the circuit is moved in and moved out again.
+        let (back, line) = thread::spawn(move || {
+            let line = collect(&mut c, &input, &outs, n);
+            (c, line)
+        })
+        .join()
+        .unwrap();
+        c = back;
+        got.push(line);
+    }
+    assert_eq!(got, want, "[threads] each tick on its own thread");
+}
+
+#[test]
+fn a_transaction_begun_on_one_thread_is_rolled_back_on_another() {
+    let mut c = Circuit::new();
+    let (input, s) = c.input::<u8>();
+    let i = c.integrate(&s);
+    let out = c.output(&i);
+    c.push(&input, Z::singleton(1, 1));
+    c.step();
+    let before = (c.current(&i), c.take(&out));
+    c.begin();
+    c.push(&input, Z::singleton(2, 1));
+    c.step();
+    let c = thread::spawn(move || {
+        let mut c = c;
+        c.rollback();
+        c
+    })
+    .join()
+    .unwrap();
+    let mut c = c;
+    assert_eq!(c.current(&i), before.0, "[txn] the stream is as it was");
+    assert_eq!(
+        c.take(&out),
+        Z::new(),
+        "[txn] the sink holds nothing of the step"
+    );
+    assert_eq!(c.ticks(), 1);
+}
+
+#[test]
+#[should_panic(expected = "handle from another circuit")]
+fn a_handle_from_another_circuit_is_refused() {
+    let (mut a, mut b) = (Circuit::new(), Circuit::new());
+    let (input, _) = a.input::<u8>();
+    b.push(&input, Z::singleton(1, 1));
+}