Merge dbsp/zset-send into dbsp/zset-edge

# Conflicts:
#	context/test-evidence/zset.md

context/code/zset-ffi.md

@@ -75,7 +75,8 @@ link, so they call the `extern "C"` functions as plain functions.
    shared cell rather than panicking.
 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 calls `step` or `rebuild_step`, with that thread's thread-local state (Lisp dynamic
+    bindings, `errno`, `zs_last_error`), 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

context/code/zset.md

@@ -169,8 +169,8 @@ Decisions, where the ADR left a choice:
    host's.
 
    **`current()` is the whole tick.** A stream keeps the latest piece, which downstream
-   operators read, and the whole value of the tick, which `Stream::current` reads: pieces
-   within one tick are absorbed into it, so `integrate.current()` after a chunked load is the
+   operators read, and the whole value of the tick, which `Circuit::current(&s)` reads: pieces
+   within one tick are absorbed into it, so `current` of an integrate stream after a chunked load is the
    full integral and a continued `delay` still reads the previous tick's input. `Output`
    sums pieces as before.
 
@@ -231,8 +231,8 @@ clause's wording stands; this is how the crate reads it).
    its map (`a_panicking_value_clone_leaves_a_memtrace_unchanged`).
 5. **The edge's two copies.** An integrate stream's piece is also its output, so a second copy
    (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 `Circuit::current` reads the piece. An eval reads its input's piece in the arena by
+   reference rather than cloning it (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 shared ownership in the
    slot and was not done.
 6. **What a transaction does not undo.**
@@ -246,8 +246,8 @@ clause's wording stands; this is how the crate reads it).
      can overflow the pending delta before any step runs (`p1_a_chunk_whose_push_panics_in_a_txn_is_counted`). A later count makes `interrupted` true for good,
      finished or not (`rollback_rebuild.rs`: begun before the transaction and continued inside,
      finished inside, a first chunk that panicked, and a dead job revived by a later step).
-   - 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
+   - A `Circuit::take` of an output 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.** `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
@@ -284,25 +284,31 @@ threads are out of scope, and nothing in `zset` spawns one. `Send` and `mpsc` co
 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). It is not `Sync` by structure (boxed nodes, no lock); no test pins
-   that, because a negative needs a compile-fail test and the comment cap forbids a fence.
+   rolled back on another). Not `Sync` is pinned by `a_circuit_is_not_sync`, an ordinary test
+   using the ambiguous-impl trick: it fails to build (E0283) if `Sync` is ever implemented.
 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).
+   access is the price. Measured on an M-series Mac under load, a release build of a 1001-node
+   chain (map, filter, distinct, delay) over 3000 ticks, best of 5, alternating with main: main
+   127k-131k ticks/s, this 118k-122k, so 7-9% slower per tick (about 1 ns per node per tick).
+   The reviewer saw 10-25% on a more loaded run. Not reduced.
 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.
+   has no circuit; the old `b1_...` case does not compile, so it is gone. The handle types are
+   invariant in their element (`PhantomData<fn(T) -> T>`), as main's `Rc<RefCell<T>>` was, so a
+   stream of `for<'a> fn(&'a u8)` cannot be coerced into a stream of `fn(&'static u8)`, where
+   `current` would panic on a failed downcast. Nine call sites check the token, and
+   `every_call_that_takes_a_handle_refuses_a_foreign_one` has one case for each.
 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.
+   `Arena::read_write`, two disjoint entries (the read one is always allocated first).
+   `a_sink_adds_its_input_with_one_clone_per_key` pins it over `Counted` keys.
 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.md

@@ -607,14 +607,14 @@ so the file never runs; the same file's other tests move a circuit for real.
 
 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.
+node kept state outside the circuit (a thread-local, say), which would not move with it.
 
 ### 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
+### every_call_that_takes_a_handle_refuses_a_foreign_one
 
 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,
@@ -689,3 +689,20 @@ with "did not panic".
 Three hundred commands on a producer thread, a consumer on another, capacity 2, the producer
 yielding at uneven points and pumping on only two of three commands. The consumer's output equals
 the combined circuit's. Rival R1 fails it.
+
+Ids index an arena, so a foreign handle from a circuit of the same shape reads or wires the
+other circuit's slot. Ten cases (`current`, `take`, `push`, a unary operator, both operands of a
+binary one, `output`, `delay`, `integrate`, `distinct`) each need the panic text "handle from
+another circuit". The first draft pinned one site in nine. Rivals, each failing it: the check
+removed from `current`, `take`, `push`, `unary_with`, `binary_with` (each operand), `output`,
+`delay` or `integrate`.
+
+### a_circuit_is_not_sync
+
+The ambiguous-impl trick: the test does not build (E0283) if `Circuit` implements `Sync`.
+Rival: `unsafe impl Sync for Circuit {}` fails the build of `tests/threads.rs`.
+
+### a_sink_adds_its_input_with_one_clone_per_key
+
+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.

crates/zset-ffi/include/zset_ffi.h

@@ -20,7 +20,9 @@
  *  - 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
+ *    zs_rebuild_step, with THAT thread's thread-local state (Lisp dynamic bindings, errno,
+ *    zs_last_error), 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/src/circuit.rs

@@ -86,7 +86,7 @@ impl<T> Slot<T> {
 }
 
 /// The index of one value in a circuit's arena, typed so a read cannot name the wrong type.
-struct Id<T>(usize, PhantomData<fn() -> T>);
+struct Id<T>(usize, PhantomData<fn(T) -> T>);
 
 impl<T> Clone for Id<T> {
     fn clone(&self) -> Self {
@@ -114,26 +114,18 @@ impl Arena {
         self.0[id.0].downcast_mut().expect("zset: arena type")
     }
 
-    /// One value to read and another to write, which are never the same entry.
+    /// One value to read and another, allocated after it, to write.
     fn read_write<R: 'static, W: 'static>(&mut self, r: Id<R>, w: Id<W>) -> (&R, &mut W) {
-        assert_ne!(
-            r.0, w.0,
-            "zset: an entry cannot be read and written at once"
+        let (lo, hi) = self.0.split_at_mut(w.0);
+        assert!(
+            r.0 < w.0,
+            "zset: the read entry comes before the written one"
         );
         let ty = "zset: arena type";
-        if r.0 < w.0 {
-            let (lo, hi) = self.0.split_at_mut(w.0);
-            (
-                lo[r.0].downcast_ref().expect(ty),
-                hi[0].downcast_mut().expect(ty),
-            )
-        } else {
-            let (lo, hi) = self.0.split_at_mut(r.0);
-            (
-                hi[0].downcast_ref().expect(ty),
-                lo[w.0].downcast_mut().expect(ty),
-            )
-        }
+        (
+            lo[r.0].downcast_ref().expect(ty),
+            hi[0].downcast_mut().expect(ty),
+        )
     }
 }
 

crates/zset/tests/cost.rs

@@ -82,6 +82,22 @@ fn an_integrate_stream_holds_its_integral_twice_when_the_tick_is_whole() {
     assert_eq!(live(), base, "[leak] every key is dropped with the circuit");
 }
 
+#[test]
+fn a_sink_adds_its_input_with_one_clone_per_key() {
+    let mut c = Circuit::new();
+    let (input, s) = c.input::<Counted>();
+    let sink = c.output(&s);
+    c.push(&input, keys(N));
+    let before = clones();
+    c.step();
+    assert_eq!(
+        clones() - before,
+        N as usize,
+        "[sink] one clone per key it sums"
+    );
+    assert_eq!(c.take(&sink).len(), N as usize);
+}
+
 type E = (Counted, u8);
 
 #[derive(Clone, Copy)]

crates/zset/tests/threads.rs

@@ -95,10 +95,77 @@ fn a_transaction_begun_on_one_thread_is_rolled_back_on_another() {
     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));
+trait AmbiguousIfSync<A> {
+    fn some_item() {}
+}
+impl<T: ?Sized> AmbiguousIfSync<()> for T {}
+struct IsSync;
+impl<T: ?Sized + Sync> AmbiguousIfSync<IsSync> for T {}
+
+// Fails to build (E0283) if `Circuit` becomes `Sync`.
+#[test]
+fn a_circuit_is_not_sync() {
+    let _ = <Circuit as AmbiguousIfSync<_>>::some_item;
+}
+
+type Site = fn(&mut Circuit, &Input<u8>, &Stream<Z<u8>>, &Output<u8>);
+
+/// Two circuits of one shape, so a foreign handle indexes a live slot of the same type.
+fn twin() -> (Circuit, Input<u8>, Stream<Z<u8>>, Output<u8>) {
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<u8>();
+    let o = c.output(&s);
+    (c, i, s, o)
+}
+
+fn panic_text(f: impl FnOnce() + std::panic::UnwindSafe) -> Option<String> {
+    let e = std::panic::catch_unwind(f).err()?;
+    e.downcast_ref::<String>()
+        .cloned()
+        .or_else(|| e.downcast_ref::<&str>().map(|s| s.to_string()))
+}
+
+#[test]
+fn every_call_that_takes_a_handle_refuses_a_foreign_one() {
+    let sites: [(&str, Site); 10] = [
+        ("current", |c, _, s, _| {
+            let _ = c.current(s);
+        }),
+        ("take", |c, _, _, o| drop(c.take(o))),
+        ("push", |c, i, _, _| c.push(i, Z::new())),
+        ("unary (map)", |c, _, s, _| {
+            let _ = c.map(s, |k| *k);
+        }),
+        ("binary, first operand", |c, _, s, _| {
+            let (_, own) = c.input::<u8>();
+            let _ = c.plus(s, &own);
+        }),
+        ("binary, second operand", |c, _, s, _| {
+            let (_, own) = c.input::<u8>();
+            let _ = c.plus(&own, s);
+        }),
+        ("output", |c, _, s, _| {
+            let _ = c.output(s);
+        }),
+        ("delay", |c, _, s, _| {
+            let _ = c.delay(s);
+        }),
+        ("integrate", |c, _, s, _| {
+            let _ = c.integrate(s);
+        }),
+        ("distinct", |c, _, s, _| {
+            let _ = c.distinct(s);
+        }),
+    ];
+    for (name, site) in sites {
+        let (mut mine, ..) = twin();
+        let (_, fi, fs, fo) = twin();
+        let text = panic_text(std::panic::AssertUnwindSafe(|| {
+            site(&mut mine, &fi, &fs, &fo)
+        }));
+        assert!(
+            text.is_some_and(|t| t.contains("handle from another circuit")),
+            "[foreign] {name} accepted a handle from another circuit"
+        );
+    }
 }