feat(zset): a stamped, coalescing delta edge between two circuits

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

context/code/delivery.md

@@ -242,3 +242,5 @@ pushes `Host::take_notices()`.
   does not require it.
 - **A pushed filter** is the client's `window`, applied on every request.
 - **Hidden-item overlays, pruning queries and column widths** are ADR 008 and clause 10.
+
+**The same edge, generalised in `zset`.** `zset::edge` (`context/code/zset.md`, "Edges") carries Z-set deltas between two circuits with the rules of this file (stamps, coalescing by summing, `target - held`, left epochs) and no log; this layer is unchanged and does not use it.

context/code/zset.md

@@ -305,3 +305,61 @@ threads are out of scope, and nothing in `zset` spawns one. `Send` and `mpsc` co
    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.
+
+## Edges (user ruling 2026-10-11; ADR 009 section 9)
+
+`crates/zset/src/edge.rs`, std only. An edge joins two circuits that share nothing: the
+producer's `Output` feeds an `EdgeSink`, the consumer's `Input` is fed by an `EdgeSource`, and a
+`Pipe` carries `Frame`s between them. A channel and a network are the same kind of edge: the
+channel edge is `Direct` over `ChannelPipe`, a network edge is a codec over a socket pipe.
+Nothing here spawns a thread; whoever owns each circuit calls `pump` and `poll`.
+
+1. **The clock is a counter the sender increments per committed command.** `Stamp { epoch, seq }`:
+   `epoch` names the sender's incarnation and is passed in, never read from a clock (the host
+   uses a random id, as `Host::with_instance` does). `delivery`'s `(host instance,
+   document_generation, publication)` maps to `(epoch, seq)` with the first two folded into
+   `epoch`; a delta is a notice with its `(from, to]` range. Delivery is not changed.
+2. **Only committed deltas cross.** `EdgeSink::committed_from(circuit, &output)` takes the output
+   after `Circuit::commit` and panics inside a transaction (`Circuit::in_transaction`). A rolled
+   back command is never taken, so it consumes no stamp
+   (`a_rolled_back_command_does_not_cross`). An empty command is still a stamp, as in delivery.
+3. **Backpressure coalesces exactly.** The sink keeps one waiting delta `(sent, seq]`. A command
+   adds to it (Z-set sum) and moves its `to`; `pump` sends it and, if the pipe says `Full`, leaves
+   it to grow. Nothing is dropped and the range has no gap
+   (`a_full_pipe_coalesces_exactly_and_keeps_the_range`).
+4. **Order and gaps.** The receiver applies a delta only if it is in the epoch it joined and
+   `from == watermark`. `to <= watermark` is a duplicate and changes nothing. Anything else (a
+   gap, an overlap, an unjoined epoch, a frame that does not decode) sets `stale`, and `stale`
+   makes the next `poll` send a `Request`.
+5. **A receiver always joins by correction, never by a delta.** A new `EdgeSource` is stale. The
+   sender answers a request with `Correction { epoch, through: seq, target: integral }` and the
+   receiver pushes `target - held` (delivery's rule), so the sender keeps its integral and the
+   receiver keeps what it has applied: two copies of the edge's state, one at each end. There is
+   no shared log as in delivery, so every answer is a full correction. Joining by a first delta
+   with `from == 0` was tried first, and the differential property found it wrong: a sender that
+   restarted with a non-empty integral at `seq 0` looks like a fresh one
+   (`two_circuits_on_an_edge_equal_one_circuit`, shrunk to a `RestartSink` after a rollback-free
+   history).
+6. **A restart is a new epoch.** `EdgeSink::restart(epoch, seq, integral)` keeps the pipe and
+   codec. The receiver sees a delta from an epoch it has not joined, goes stale, and the answer
+   moves it onto the new epoch, adding the old one to `left` (the newest 16). Frames from a left
+   epoch change nothing, so a late reply from a dead sender cannot regress the receiver. A
+   receiver that restarts (`EdgeSource::restart`) is stale again.
+7. **A lost final message is invisible to the receiver.** `connection_lost()` is the host's
+   timeout: it marks the receiver stale and lets it ask again. The host calls it on its own local
+   monotonic timeout, never by comparing times across machines. It is also what retries a request
+   whose answer was lost. Requests carry the receiver's stamp and this sender does not read it.
+8. **The wire form is a hook.** `EdgeCodec<K>` turns a `Frame<K>` into `Wire` and back, with
+   `CodecError` for what does not decode; `Direct` is the identity. `improvise-protocol` is to
+   implement it for its own key types in the lossless text of ADR 005, and `zset` stays free of
+   serde. The tests run every history through a hand-written text codec
+   (`the_same_histories_through_a_text_codec`).
+9. **Threads.** `EdgeSink` and `EdgeSource` are `Send` when the pipe and codec are. The real
+   two-thread test steps a producer on one thread and a consumer on another with a capacity-2 channel
+   (`a_circuit_on_each_of_two_threads_equals_one_circuit`). The property test simulates arbitrary
+   interleavings, losses and restarts by choosing the order of `commit`, `pump`, `poll` and step.
+   Native threads are assumed; `mpsc` and `Send` compile on wasm32 and nothing runs there.
+10. **Not built.** A sink with a log to answer a watermark with only the moved keys (delivery's
+    `Moved` correction). A pipe that blocks or notifies, because `try_send` and `try_recv` are
+    polling. Key-sharded data parallelism inside a circuit: its exchange step would be this same
+    edge, and nothing here forbids it.

context/test-evidence/zset.md

@@ -619,3 +619,73 @@ as at `begin`. The undo log lives in the nodes, which move with the circuit.
 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.
+
+### two_circuits_on_an_edge_equal_one_circuit
+
+Random histories of commit, rolled-back command, pump, poll, consumer step, a lost frame in
+either direction, a sender restart, a receiver restart and a timeout, at capacity 1 to 3, settle
+to the output of one circuit running both graphs. It found a real bug: a receiver that joined on a
+delta with `from == 0` took a restarted sender's first delta for a fresh stream and ended with
+`(1, 1)` where the combined circuit had nothing; joining only by correction fixed it. Rivals, each
+failing it (and the text-codec twin): coalescing that keeps only the last delta (R1), accepting a
+delta across a gap (R2), a correction applied as `target` instead of `target - held` (R3), joining
+by a first delta (R7). R1 also fails the two-thread test.
+
+### the_same_histories_through_a_text_codec
+
+The same property with a string wire form, so a frame is encoded and decoded on every hop. The
+rivals above fail it as well; it would also fail if the codec lost a weight or a stamp.
+
+### a_full_pipe_coalesces_exactly_and_keeps_the_range
+
+Capacity 1: a first frame fills the pipe, four more commands pile up, and the next frame is one
+delta `(1, 5]` with their exact sum. Rival R1 (the waiting delta replaced, not summed) fails it on
+the sum.
+
+### an_empty_command_is_still_a_stamp
+
+Two commands that change nothing still move the receiver to `seq 2`. Rival R8 (stamp only a
+non-empty delta) fails it.
+
+### a_lost_delta_is_seen_in_the_stamps_and_corrected
+
+Capacity 8: the second delta is dropped by the pipe, so the third starts at 2 while the receiver
+is at 1. The receiver goes stale, asks, and `target - held` brings it to `{1, 2, 3}`. Rivals R2
+(apply across the gap) and R3 fail it.
+
+### a_duplicate_delta_is_applied_once
+
+The same delta twice after a join: applied once, and the receiver is not stale. Rival R5 (a
+duplicate treated as a gap) leaves the output right and fails only the `stale` assertion, which is
+why that assertion is there.
+
+### a_restarted_sender_is_corrected_not_trusted
+
+A new epoch holding `{1, 2}` plus one command `9`: the receiver applies only `{9}` and stamps
+`(2, 1)`. Rival R3 applies the whole target and fails.
+
+### a_late_frame_from_an_epoch_the_receiver_left_changes_nothing
+
+After epoch 2 is joined, a correction from epoch 1 with a larger `through` and a delta of epoch 1
+are ignored and ask for nothing. Rival R6 (no `left` check on corrections) fails it.
+
+### an_unreadable_frame_asks_for_a_correction
+
+A frame that does not decode sets `stale`. A delta before the join is ignored and a correction
+joins.
+
+### a_rolled_back_command_does_not_cross
+
+A producer step inside `begin`/`rollback` consumes no stamp and leaves nothing waiting; the
+retry that commits is stamp 1.
+
+### taking_a_delta_inside_a_transaction_is_refused
+
+`committed_from` panics while a transaction is open. Rival R9 (the assertion removed) fails it
+with "did not panic".
+
+### a_circuit_on_each_of_two_threads_equals_one_circuit
+
+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.

crates/zset/src/circuit.rs

@@ -507,6 +507,11 @@ impl Circuit {
         self.run(false);
     }
 
+    /// Whether a transaction is open (`begin` and not yet `commit` or `rollback`).
+    pub fn in_transaction(&self) -> bool {
+        self.open.is_some()
+    }
+
     /// How many ticks `step` has begun. A `Rebuild` uses it to see a tick split.
     pub fn ticks(&self) -> u64 {
         self.ticks

crates/zset/src/edge.rs

@@ -0,0 +1,404 @@
+//! The edge between two circuits: stamped, ordered, coalescing, and the same whether the pipe
+//! is a channel or a network. See `context/code/zset.md`, "Edges".
+
+use crate::{Circuit, Input, Output, ZSet};
+use std::collections::VecDeque;
+use std::sync::mpsc;
+
+/// A point on the sender's causal clock: its incarnation, and how many commands it has committed.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub struct Stamp {
+    pub epoch: u64,
+    pub seq: u64,
+}
+
+/// What the sender's commands `(from, to]` changed. Deltas sum exactly, so adjacent ones merge.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Delta<K> {
+    pub epoch: u64,
+    pub from: u64,
+    pub to: u64,
+    pub delta: ZSet<K>,
+}
+
+/// The sender's whole integral as of `through`; the receiver applies `target - held`.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Correction<K> {
+    pub epoch: u64,
+    pub through: u64,
+    pub target: ZSet<K>,
+}
+
+/// "Bring me up from here": carried with the receiver's watermark and never read by this sender.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Request {
+    pub have: Option<Stamp>,
+}
+
+/// Everything that crosses an edge, in both directions.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum Frame<K> {
+    Delta(Delta<K>),
+    Correction(Correction<K>),
+    Request(Request),
+}
+
+/// A frame that does not decode.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CodecError(pub String);
+
+/// The lossless wire form of an edge's frames. `improvise-protocol` supplies one for its key
+/// types; `zset` knows no serde. `Direct` is the identity, for an in-process edge.
+pub trait EdgeCodec<K> {
+    type Wire: Send + 'static;
+    fn encode(&self, frame: &Frame<K>) -> Self::Wire;
+    fn decode(&self, wire: Self::Wire) -> Result<Frame<K>, CodecError>;
+}
+
+/// No encoding: the frame itself crosses.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct Direct;
+
+impl<K: Clone + Send + 'static> EdgeCodec<K> for Direct {
+    type Wire = Frame<K>;
+
+    fn encode(&self, frame: &Frame<K>) -> Frame<K> {
+        frame.clone()
+    }
+
+    fn decode(&self, wire: Frame<K>) -> Result<Frame<K>, CodecError> {
+        Ok(wire)
+    }
+}
+
+/// Why a pipe took nothing; `Full` is backpressure, `Closed` is the end.
+#[derive(Debug, PartialEq, Eq)]
+pub enum Refused<W> {
+    Full(W),
+    Closed,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum Recv {
+    Empty,
+    Closed,
+}
+
+/// One end of a duplex pipe of wire items: a channel in-process, a socket across a network.
+pub trait Pipe<W>: Send {
+    fn try_send(&mut self, wire: W) -> Result<(), Refused<W>>;
+    fn try_recv(&mut self) -> Result<W, Recv>;
+}
+
+/// An in-process pipe end over bounded channels.
+pub struct ChannelPipe<W> {
+    tx: mpsc::SyncSender<W>,
+    rx: mpsc::Receiver<W>,
+}
+
+/// Two connected ends; each direction holds `cap` items.
+pub fn channel_pipes<W: Send>(cap: usize) -> (ChannelPipe<W>, ChannelPipe<W>) {
+    let (atx, brx) = mpsc::sync_channel(cap);
+    let (btx, arx) = mpsc::sync_channel(cap);
+    (
+        ChannelPipe { tx: atx, rx: arx },
+        ChannelPipe { tx: btx, rx: brx },
+    )
+}
+
+impl<W: Send> Pipe<W> for ChannelPipe<W> {
+    fn try_send(&mut self, wire: W) -> Result<(), Refused<W>> {
+        self.tx.try_send(wire).map_err(|e| match e {
+            mpsc::TrySendError::Full(w) => Refused::Full(w),
+            mpsc::TrySendError::Disconnected(_) => Refused::Closed,
+        })
+    }
+
+    fn try_recv(&mut self) -> Result<W, Recv> {
+        self.rx.try_recv().map_err(|e| match e {
+            mpsc::TryRecvError::Empty => Recv::Empty,
+            mpsc::TryRecvError::Disconnected => Recv::Closed,
+        })
+    }
+}
+
+/// How a `pump` left the edge.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Flow {
+    /// Everything committed has been sent.
+    Idle,
+    /// The pipe was full: what is unsent has been coalesced and waits.
+    Backlogged,
+    Closed,
+}
+
+/// The sending end, beside the producer circuit: it stamps each committed command, keeps the
+/// integral a correction needs, and coalesces what the pipe will not take.
+pub struct EdgeSink<K, C, P> {
+    codec: C,
+    pipe: P,
+    epoch: u64,
+    seq: u64,
+    integral: ZSet<K>,
+    /// The commands `(sent, seq]`, summed; `sent == seq` means nothing waits.
+    waiting: Delta<K>,
+    owes_correction: bool,
+    closed: bool,
+}
+
+impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
+    /// A sender that has committed nothing. `epoch` identifies this incarnation: a fresh one
+    /// per start, never a time.
+    pub fn new(codec: C, pipe: P, epoch: u64) -> Self {
+        Self::resume(codec, pipe, epoch, 0, ZSet::new())
+    }
+
+    /// A sender that starts at `seq` with `integral` already committed, as after a restart.
+    pub fn resume(codec: C, pipe: P, epoch: u64, seq: u64, integral: ZSet<K>) -> Self {
+        EdgeSink {
+            codec,
+            pipe,
+            epoch,
+            seq,
+            integral,
+            waiting: Delta {
+                epoch,
+                from: seq,
+                to: seq,
+                delta: ZSet::new(),
+            },
+            owes_correction: false,
+            closed: false,
+        }
+    }
+
+    pub fn stamp(&self) -> Stamp {
+        Stamp {
+            epoch: self.epoch,
+            seq: self.seq,
+        }
+    }
+
+    /// Whether any committed command has not been sent.
+    pub fn backlog(&self) -> bool {
+        self.waiting.to > self.waiting.from || self.owes_correction
+    }
+
+    /// The same pipe and codec, after a restart: a new incarnation at `seq` holding `integral`.
+    pub fn restart(self, epoch: u64, seq: u64, integral: ZSet<K>) -> Self {
+        Self::resume(self.codec, self.pipe, epoch, seq, integral)
+    }
+
+    /// One committed command changed the output by `delta` (empty is still a command).
+    pub fn committed(&mut self, delta: ZSet<K>) -> Stamp {
+        self.seq += 1;
+        self.integral += &delta;
+        self.waiting.delta += delta;
+        self.waiting.to = self.seq;
+        self.stamp()
+    }
+
+    /// Takes `out` after the circuit's command committed. Panics inside a transaction: a
+    /// rolled-back command must not cross, and `take` there is not a delta.
+    pub fn committed_from(&mut self, circuit: &mut Circuit, out: &Output<K>) -> Stamp
+    where
+        K: Send + 'static,
+    {
+        assert!(
+            !circuit.in_transaction(),
+            "zset: an edge takes a command's delta only after it commits"
+        );
+        let delta = circuit.take(out);
+        self.committed(delta)
+    }
+
+    /// Answers requests, then sends what waits. Call it as often as the host likes.
+    pub fn pump(&mut self) -> Flow {
+        while !self.closed {
+            match self.pipe.try_recv() {
+                Ok(wire) => {
+                    if let Ok(Frame::Request(_)) = self.codec.decode(wire) {
+                        self.owes_correction = true;
+                    }
+                }
+                Err(Recv::Empty) => break,
+                Err(Recv::Closed) => self.closed = true,
+            }
+        }
+        if self.closed {
+            return Flow::Closed;
+        }
+        let frame = if self.owes_correction {
+            Frame::Correction(Correction {
+                epoch: self.epoch,
+                through: self.seq,
+                target: self.integral.clone(),
+            })
+        } else if self.waiting.to > self.waiting.from {
+            Frame::Delta(self.waiting.clone())
+        } else {
+            return Flow::Idle;
+        };
+        match self.pipe.try_send(self.codec.encode(&frame)) {
+            Ok(()) => {
+                self.owes_correction = false;
+                self.waiting = Delta {
+                    epoch: self.epoch,
+                    from: self.seq,
+                    to: self.seq,
+                    delta: ZSet::new(),
+                };
+                Flow::Idle
+            }
+            Err(Refused::Full(_)) => Flow::Backlogged,
+            Err(Refused::Closed) => {
+                self.closed = true;
+                Flow::Closed
+            }
+        }
+    }
+}
+
+/// How many incarnations a receiver remembers having left.
+const LEFT: usize = 16;
+
+/// The receiving end, beside the consumer circuit: it applies deltas to the circuit's input in
+/// order, and asks for a correction when the stamps show something missing.
+pub struct EdgeSource<K, C, P> {
+    codec: C,
+    pipe: P,
+    input: Input<K>,
+    joined: Option<u64>,
+    watermark: u64,
+    held: ZSet<K>,
+    left: VecDeque<u64>,
+    stale: bool,
+    asked: bool,
+    closed: bool,
+}
+
+impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSource<K, C, P> {
+    /// A receiver that holds nothing and asks for a correction on its first `poll`.
+    pub fn new(codec: C, pipe: P, input: Input<K>) -> Self {
+        EdgeSource {
+            codec,
+            pipe,
+            input,
+            joined: None,
+            watermark: 0,
+            held: ZSet::new(),
+            left: VecDeque::new(),
+            stale: true,
+            asked: false,
+            closed: false,
+        }
+    }
+
+    /// The stamp applied through, once any sender has been heard.
+    pub fn stamp(&self) -> Option<Stamp> {
+        self.joined.map(|epoch| Stamp {
+            epoch,
+            seq: self.watermark,
+        })
+    }
+
+    /// Whether a correction is wanted and has not arrived.
+    pub fn stale(&self) -> bool {
+        self.stale
+    }
+
+    /// The same pipe and codec, after the consumer lost its state: it holds nothing again.
+    pub fn restart(self, input: Input<K>) -> Self {
+        Self::new(self.codec, self.pipe, input)
+    }
+
+    /// The host's timeout or reconnect: ask again from the watermark held.
+    pub fn connection_lost(&mut self) {
+        self.stale = true;
+        self.asked = false;
+    }
+
+    /// Applies up to `max` frames to the circuit's input, then asks for a correction if one is
+    /// wanted. Returns how many frames it read. The host steps the circuit.
+    pub fn poll(&mut self, circuit: &mut Circuit, max: usize) -> usize {
+        let mut read = 0;
+        while read < max && !self.closed {
+            match self.pipe.try_recv() {
+                Ok(wire) => {
+                    read += 1;
+                    match self.codec.decode(wire) {
+                        Ok(Frame::Delta(d)) => self.on_delta(circuit, d),
+                        Ok(Frame::Correction(c)) => self.on_correction(circuit, c),
+                        Ok(Frame::Request(_)) => {}
+                        Err(_) => self.stale = true,
+                    }
+                }
+                Err(Recv::Empty) => break,
+                Err(Recv::Closed) => self.closed = true,
+            }
+        }
+        self.ask();
+        read
+    }
+
+    fn has_left(&self, epoch: u64) -> bool {
+        self.left.contains(&epoch)
+    }
+
+    fn apply(&mut self, circuit: &mut Circuit, delta: ZSet<K>) {
+        self.held += &delta;
+        circuit.push(&self.input, delta);
+    }
+
+    fn on_delta(&mut self, circuit: &mut Circuit, d: Delta<K>) {
+        if self.has_left(d.epoch) {
+            return;
+        }
+        let known = self.joined == Some(d.epoch);
+        if known && d.to <= self.watermark {
+            return;
+        }
+        if known && d.from == self.watermark {
+            self.watermark = d.to;
+            self.apply(circuit, d.delta);
+        } else {
+            self.stale = true;
+        }
+    }
+
+    fn on_correction(&mut self, circuit: &mut Circuit, c: Correction<K>) {
+        if self.has_left(c.epoch) {
+            return;
+        }
+        self.asked = false;
+        let known = self.joined == Some(c.epoch);
+        if known && c.through <= self.watermark {
+            self.stale &= c.through != self.watermark;
+            return;
+        }
+        if !known && let Some(old) = self.joined.replace(c.epoch) {
+            self.left.push_back(old);
+            if self.left.len() > LEFT {
+                self.left.pop_front();
+            }
+        }
+        let delta = &c.target - &self.held;
+        self.watermark = c.through;
+        self.stale = false;
+        self.apply(circuit, delta);
+    }
+
+    fn ask(&mut self) {
+        if !self.stale || self.asked || self.closed {
+            return;
+        }
+        let wire = self
+            .codec
+            .encode(&Frame::Request(Request { have: self.stamp() }));
+        match self.pipe.try_send(wire) {
+            Ok(()) => self.asked = true,
+            Err(Refused::Full(_)) => {}
+            Err(Refused::Closed) => self.closed = true,
+        }
+    }
+}

crates/zset/src/lib.rs

@@ -2,10 +2,15 @@
 //! wherever no weight leaves `i64`; past that it panics. See `context/code/zset.md`.
 
 mod circuit;
+mod edge;
 mod ops;
 mod trace;
 
 pub use circuit::{Circuit, Input, Output, Progress, Rebuild, Stream};
+pub use edge::{
+    ChannelPipe, CodecError, Correction, Delta, Direct, EdgeCodec, EdgeSink, EdgeSource, Flow,
+    Frame, Pipe, Recv, Refused, Request, Stamp, channel_pipes,
+};
 pub use ops::{AggregateState, DistinctState, JoinState};
 pub use trace::{MemTrace, Trace};
 

crates/zset/tests/edge.proptest-regressions

@@ -0,0 +1,12 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 329f754122c8c0316dfbf1e5e2d2aa3f2e30ecc1f7269af166ea3a38da116fdf # shrinks to acts = [Poll(1), Pump, LoseForward(1), Commit([]), Pump], cap = 1
+cc bda846f36781898e12a640582cf8ea8280ef664757e35dca34f4b0882994e613 # shrinks to acts = [Commit([]), Commit([]), Commit([]), LoseForward(1), Commit([]), Commit([]), Commit([(0, -1)]), Commit([]), RestartSink, Commit([(0, 1)])], cap = 1
+cc d717a61b42263ed743ce993d0dc7755a3effeddf3aeac9aede363a8c1e8af465 # shrinks to acts = [Poll(1), Pump, Commit([(0, 1)]), Commit([])], cap = 1
+cc b1f14324a4b9eb1409330b833483f0869a1b8bc54d145245ef8d32b585c54e85 # shrinks to acts = [Poll(1), Pump, Commit([(4, 1)]), Commit([])], cap = 1
+cc 24b3d06547af18b65c1f93abd0294a1012e4b4e3557bc6eb77af3bfe445ffea8 # shrinks to acts = [Commit([]), Poll(1), Pump, LoseForward(1), Commit([(0, 1)]), Commit([]), Pump, Commit([])], cap = 1
+cc dbc0a788d2546c9fdf88d2dd5e2a2df09625873daca1ae59313f1340a0dbd2f3 # shrinks to acts = [Poll(1), Pump, LoseForward(1), Commit([(6, 1)]), Commit([]), Commit([(0, 0)]), Pump, Commit([])], cap = 1

crates/zset/tests/edge.rs

@@ -0,0 +1,704 @@
+//! The edge between two circuits: a producer and a consumer joined by stamped, coalescing
+//! deltas equal one circuit running both. Evidence: `context/test-evidence/zset.md`.
+
+use proptest::prelude::*;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::thread;
+use zset::{
+    ChannelPipe, Circuit, CodecError, Correction, Delta, Direct, EdgeCodec, EdgeSink, EdgeSource,
+    Flow, Frame, Input, Output, Pipe, Recv, Refused, Request, Stamp, ZSet, channel_pipes,
+};
+
+type Z = ZSet<u8>;
+type Counts = ZSet<(u8, i64)>;
+
+fn z(entries: &[(u8, i64)]) -> Z {
+    entries.iter().copied().collect()
+}
+
+fn mapped(d: &Z) -> Z {
+    d.map(|k| k % 7)
+}
+
+/// The producer's graph: one map.
+fn producer() -> (Circuit, Input<u8>, Output<u8>) {
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<u8>();
+    let m = c.map(&s, |k| k % 7);
+    let o = c.output(&m);
+    (c, i, o)
+}
+
+/// The consumer's graph: distinct, then a count per group.
+fn consumer_tail(c: &mut Circuit, s: &zset::Stream<Z>) -> Output<(u8, i64)> {
+    let d = c.distinct(s);
+    let g = c.index_by(&d, |k| k % 3);
+    let a = c.aggregate(&g, |_, vals| Some(vals.len() as i64));
+    c.output(&a)
+}
+
+fn consumer() -> (Circuit, Input<u8>, Output<(u8, i64)>) {
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<u8>();
+    let o = consumer_tail(&mut c, &s);
+    (c, i, o)
+}
+
+/// Both graphs in one circuit: what the edge must be equal to.
+fn single() -> (Circuit, Input<u8>, Output<(u8, i64)>) {
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<u8>();
+    let m = c.map(&s, |k| k % 7);
+    let o = consumer_tail(&mut c, &m);
+    (c, i, o)
+}
+
+/// A pipe end that can lose what is sent through it.
+struct FaultPipe<W> {
+    inner: ChannelPipe<W>,
+    lose: Arc<AtomicUsize>,
+}
+
+impl<W: Send> Pipe<W> for FaultPipe<W> {
+    fn try_send(&mut self, wire: W) -> Result<(), Refused<W>> {
+        if self.lose.load(Ordering::SeqCst) > 0 {
+            self.lose.fetch_sub(1, Ordering::SeqCst);
+            return Ok(());
+        }
+        self.inner.try_send(wire)
+    }
+
+    fn try_recv(&mut self) -> Result<W, Recv> {
+        self.inner.try_recv()
+    }
+}
+
+fn fault<W: Send>(
+    cap: usize,
+) -> (
+    FaultPipe<W>,
+    FaultPipe<W>,
+    Arc<AtomicUsize>,
+    Arc<AtomicUsize>,
+) {
+    let (a, b) = channel_pipes(cap);
+    let (la, lb) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
+    (
+        FaultPipe {
+            inner: a,
+            lose: la.clone(),
+        },
+        FaultPipe {
+            inner: b,
+            lose: lb.clone(),
+        },
+        la,
+        lb,
+    )
+}
+
+/// A text wire form, to run the same histories through an encoding.
+#[derive(Clone, Copy)]
+struct Text;
+
+fn zs(z: &Z) -> String {
+    z.iter()
+        .map(|(k, w)| format!("{k}:{w}"))
+        .collect::<Vec<_>>()
+        .join(",")
+}
+
+fn unzs(s: &str) -> Result<Z, CodecError> {
+    let bad = || CodecError(format!("bad entries {s:?}"));
+    s.split(',')
+        .filter(|p| !p.is_empty())
+        .map(|p| {
+            let (k, w) = p.split_once(':').ok_or_else(bad)?;
+            Ok((
+                k.parse::<u8>().map_err(|_| bad())?,
+                w.parse::<i64>().map_err(|_| bad())?,
+            ))
+        })
+        .collect()
+}
+
+impl EdgeCodec<u8> for Text {
+    type Wire = String;
+
+    fn encode(&self, f: &Frame<u8>) -> String {
+        match f {
+            Frame::Delta(d) => format!("D {} {} {} {}", d.epoch, d.from, d.to, zs(&d.delta)),
+            Frame::Correction(c) => format!("C {} {} {}", c.epoch, c.through, zs(&c.target)),
+            Frame::Request(r) => match r.have {
+                Some(s) => format!("R {} {}", s.epoch, s.seq),
+                None => "R".to_string(),
+            },
+        }
+    }
+
+    fn decode(&self, w: String) -> Result<Frame<u8>, CodecError> {
+        let bad = || CodecError(format!("bad frame {w:?}"));
+        let mut it = w.splitn(5, ' ');
+        let num = |s: Option<&str>| s.and_then(|s| s.parse::<u64>().ok()).ok_or_else(bad);
+        match it.next() {
+            Some("D") => Ok(Frame::Delta(Delta {
+                epoch: num(it.next())?,
+                from: num(it.next())?,
+                to: num(it.next())?,
+                delta: unzs(it.next().ok_or_else(bad)?)?,
+            })),
+            Some("C") => {
+                let mut it = w.splitn(4, ' ');
+                it.next();
+                Ok(Frame::Correction(Correction {
+                    epoch: num(it.next())?,
+                    through: num(it.next())?,
+                    target: unzs(it.next().ok_or_else(bad)?)?,
+                }))
+            }
+            Some("R") => {
+                let have = match it.next() {
+                    None => None,
+                    e => Some(Stamp {
+                        epoch: num(e)?,
+                        seq: num(it.next())?,
+                    }),
+                };
+                Ok(Frame::Request(Request { have }))
+            }
+            _ => Err(bad()),
+        }
+    }
+}
+
+type Sink<C> = EdgeSink<u8, C, FaultPipe<<C as EdgeCodec<u8>>::Wire>>;
+type Source<C> = EdgeSource<u8, C, FaultPipe<<C as EdgeCodec<u8>>::Wire>>;
+
+/// Two circuits joined by an edge, and the combined circuit they must equal.
+struct World<C: EdgeCodec<u8> + Clone> {
+    codec: C,
+    prod: Circuit,
+    p_in: Input<u8>,
+    p_out: Output<u8>,
+    sink: Sink<C>,
+    lose_fwd: Arc<AtomicUsize>,
+    cons: Circuit,
+    c_out: Output<(u8, i64)>,
+    src: Source<C>,
+    lose_back: Arc<AtomicUsize>,
+    got: Counts,
+    p_total: Z,
+    epoch: u64,
+    one: Circuit,
+    one_in: Input<u8>,
+    one_out: Output<(u8, i64)>,
+    want: Counts,
+}
+
+impl<C: EdgeCodec<u8> + Clone> World<C> {
+    fn new(codec: C, cap: usize) -> Self {
+        let (prod, p_in, p_out) = producer();
+        let (cons, c_in, c_out) = consumer();
+        let (one, one_in, one_out) = single();
+        let (a, b, lose_fwd, lose_back) = fault(cap);
+        World {
+            sink: EdgeSink::new(codec.clone(), a, 1),
+            src: EdgeSource::new(codec.clone(), b, c_in),
+            codec,
+            prod,
+            p_in,
+            p_out,
+            lose_fwd,
+            cons,
+            c_out,
+            lose_back,
+            got: Counts::new(),
+            p_total: Z::new(),
+            epoch: 1,
+            one,
+            one_in,
+            one_out,
+            want: Counts::new(),
+        }
+    }
+
+    fn commit(&mut self, d: &Z) {
+        self.prod.begin();
+        self.prod.push(&self.p_in, d.clone());
+        self.prod.step();
+        self.prod.commit();
+        self.sink.committed_from(&mut self.prod, &self.p_out);
+        self.p_total += &mapped(d);
+        self.one.push(&self.one_in, d.clone());
+        self.one.step();
+        self.want += self.one.take(&self.one_out);
+    }
+
+    fn abort(&mut self, d: &Z) {
+        let before = self.sink.stamp();
+        self.prod.begin();
+        self.prod.push(&self.p_in, d.clone());
+        self.prod.step();
+        self.prod.rollback();
+        assert_eq!(
+            self.sink.stamp(),
+            before,
+            "[abort] a rolled-back command was stamped"
+        );
+        assert_eq!(
+            self.prod.take(&self.p_out),
+            Z::new(),
+            "[abort] its output survived"
+        );
+    }
+
+    fn step_consumer(&mut self) {
+        self.cons.step();
+        self.got += self.cons.take(&self.c_out);
+    }
+
+    fn restart_sink(&mut self) {
+        self.epoch += 1;
+        let old = std::mem::replace(
+            &mut self.sink,
+            EdgeSink::new(self.codec.clone(), fault_placeholder(), 0),
+        );
+        self.sink = old.restart(self.epoch, 0, self.p_total.clone());
+    }
+
+    fn restart_source(&mut self) {
+        let (cons, c_in, c_out) = consumer();
+        let old = std::mem::replace(
+            &mut self.src,
+            EdgeSource::new(self.codec.clone(), fault_placeholder(), c_in),
+        );
+        self.src = old.restart(c_in);
+        self.cons = cons;
+        self.c_out = c_out;
+        self.got = Counts::new();
+    }
+
+    /// Runs the edge until it is idle, with nothing lost any more.
+    fn settle(&mut self) {
+        self.lose_fwd.store(0, Ordering::SeqCst);
+        self.lose_back.store(0, Ordering::SeqCst);
+        for _ in 0..200 {
+            self.sink.pump();
+            let read = self.src.poll(&mut self.cons, 8);
+            self.step_consumer();
+            if !self.sink.backlog()
+                && !self.src.stale()
+                && self.src.stamp() == Some(self.sink.stamp())
+            {
+                break;
+            }
+            if read == 0 {
+                // Nothing arrived: the host's timeout fires and the receiver asks again.
+                self.src.connection_lost();
+            }
+        }
+        assert_eq!(
+            self.src.stamp(),
+            Some(self.sink.stamp()),
+            "[settle] never caught up"
+        );
+        self.step_consumer();
+    }
+}
+
+/// A pipe end that is never used: only to fill a slot while the real one is moved.
+fn fault_placeholder<W: Send>() -> FaultPipe<W> {
+    fault(1).0
+}
+
+#[derive(Clone, Debug)]
+enum Act {
+    Commit(Vec<(u8, i64)>),
+    Abort(Vec<(u8, i64)>),
+    Pump,
+    Poll(usize),
+    StepConsumer,
+    LoseForward(usize),
+    LoseBack(usize),
+    RestartSink,
+    RestartSource,
+    ConnectionLost,
+}
+
+fn delta() -> impl Strategy<Value = Vec<(u8, i64)>> {
+    prop::collection::vec((0u8..14, -2i64..=2), 0..4)
+}
+
+fn act() -> impl Strategy<Value = Act> {
+    prop_oneof![
+        6 => delta().prop_map(Act::Commit),
+        1 => delta().prop_map(Act::Abort),
+        4 => Just(Act::Pump),
+        4 => (1usize..4).prop_map(Act::Poll),
+        3 => Just(Act::StepConsumer),
+        1 => (1usize..3).prop_map(Act::LoseForward),
+        1 => (1usize..3).prop_map(Act::LoseBack),
+        1 => Just(Act::RestartSink),
+        1 => Just(Act::RestartSource),
+        1 => Just(Act::ConnectionLost),
+    ]
+}
+
+fn run<C: EdgeCodec<u8> + Clone>(codec: C, cap: usize, acts: &[Act]) {
+    let mut w = World::new(codec, cap);
+    for a in acts {
+        match a {
+            Act::Commit(d) => w.commit(&d.iter().copied().collect()),
+            Act::Abort(d) => w.abort(&d.iter().copied().collect()),
+            Act::Pump => {
+                w.sink.pump();
+            }
+            Act::Poll(n) => {
+                w.src.poll(&mut w.cons, *n);
+            }
+            Act::StepConsumer => w.step_consumer(),
+            Act::LoseForward(n) => w.lose_fwd.store(*n, Ordering::SeqCst),
+            Act::LoseBack(n) => w.lose_back.store(*n, Ordering::SeqCst),
+            Act::RestartSink => w.restart_sink(),
+            Act::RestartSource => w.restart_source(),
+            Act::ConnectionLost => w.src.connection_lost(),
+        }
+    }
+    w.settle();
+    assert_eq!(
+        w.got, w.want,
+        "[edge] the consumer's output after the edge settles"
+    );
+}
+
+proptest! {
+    /// The differential: any history of commits, aborts, pumps, polls, losses and restarts,
+    /// once settled, leaves the consumer where one combined circuit would be.
+    #[test]
+    fn two_circuits_on_an_edge_equal_one_circuit(
+        acts in prop::collection::vec(act(), 0..40), cap in 1usize..4,
+    ) {
+        run(Direct, cap, &acts);
+    }
+
+    /// The same histories through a text wire form.
+    #[test]
+    fn the_same_histories_through_a_text_codec(
+        acts in prop::collection::vec(act(), 0..40), cap in 1usize..4,
+    ) {
+        run(Text, cap, &acts);
+    }
+}
+
+/// Runs the edge until quiet, stepping the consumer between polls.
+fn drive<C: EdgeCodec<u8>>(
+    sink: &mut EdgeSink<u8, C, impl Pipe<C::Wire>>,
+    src: &mut EdgeSource<u8, C, impl Pipe<C::Wire>>,
+    c: &mut Circuit,
+) {
+    for _ in 0..20 {
+        sink.pump();
+        src.poll(c, 100);
+        c.step();
+    }
+}
+
+/// A consumer circuit that only mirrors its input, and a sink and source on a fault pipe.
+fn mirror(
+    cap: usize,
+) -> (
+    Sink<Direct>,
+    Source<Direct>,
+    Circuit,
+    Output<u8>,
+    Arc<AtomicUsize>,
+) {
+    let (a, b, lose_fwd, _) = fault(cap);
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<u8>();
+    let o = c.output(&s);
+    (
+        EdgeSink::new(Direct, a, 1),
+        EdgeSource::new(Direct, b, i),
+        c,
+        o,
+        lose_fwd,
+    )
+}
+
+#[test]
+fn deltas_arrive_in_order_and_are_applied() {
+    let (mut sink, mut src, mut c, out, _) = mirror(8);
+    sink.committed(z(&[(1, 1)]));
+    sink.pump();
+    sink.committed(z(&[(2, 1)]));
+    sink.pump();
+    drive(&mut sink, &mut src, &mut c);
+    assert_eq!(c.take(&out), z(&[(1, 1), (2, 1)]));
+    assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 2 }));
+}
+
+#[test]
+fn a_full_pipe_coalesces_exactly_and_keeps_the_range() {
+    let (a, mut b) = channel_pipes::<Frame<u8>>(1);
+    let mut sink = EdgeSink::new(Direct, a, 1);
+    sink.committed(z(&[(1, 1)]));
+    assert_eq!(sink.pump(), Flow::Idle);
+    for k in 2..6 {
+        sink.committed(z(&[(k, 1), (1, -1)]));
+        assert_eq!(sink.pump(), Flow::Backlogged, "[full] command {k}");
+    }
+    assert!(matches!(b.try_recv(), Ok(Frame::Delta(d)) if (d.from, d.to) == (0, 1)));
+    assert_eq!(sink.pump(), Flow::Idle);
+    let Ok(Frame::Delta(d)) = b.try_recv() else {
+        panic!("a merged delta")
+    };
+    assert_eq!(
+        (d.from, d.to),
+        (1, 5),
+        "[range] the merged delta covers (1, 5]"
+    );
+    assert_eq!(
+        d.delta,
+        z(&[(2, 1), (3, 1), (4, 1), (5, 1), (1, -4)]),
+        "[sum] coalescing sums exactly"
+    );
+}
+
+#[test]
+fn an_empty_command_is_still_a_stamp() {
+    let (mut sink, mut src, mut c, _, _) = mirror(8);
+    sink.committed(Z::new());
+    sink.committed(Z::new());
+    drive(&mut sink, &mut src, &mut c);
+    assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 2 }));
+}
+
+#[test]
+fn a_lost_delta_is_seen_in_the_stamps_and_corrected() {
+    let (mut sink, mut src, mut c, out, lose) = mirror(8);
+    drive(&mut sink, &mut src, &mut c);
+    sink.committed(z(&[(1, 1)]));
+    sink.pump();
+    src.poll(&mut c, 10);
+    lose.store(1, Ordering::SeqCst);
+    sink.committed(z(&[(2, 1)]));
+    sink.pump();
+    sink.committed(z(&[(3, 1)]));
+    sink.pump();
+    src.poll(&mut c, 10);
+    assert!(
+        src.stale(),
+        "[gap] the third delta does not follow the first"
+    );
+    drive(&mut sink, &mut src, &mut c);
+    assert_eq!(c.take(&out), z(&[(1, 1), (2, 1), (3, 1)]), "[correction]");
+    assert!(!src.stale());
+}
+
+#[test]
+fn a_duplicate_delta_is_applied_once() {
+    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<u8>();
+    let o = c.output(&s);
+    let mut src = EdgeSource::new(Direct, a, i);
+    let join = Frame::Correction(Correction {
+        epoch: 1,
+        through: 0,
+        target: Z::new(),
+    });
+    let d = Frame::Delta(Delta {
+        epoch: 1,
+        from: 0,
+        to: 1,
+        delta: z(&[(1, 1)]),
+    });
+    b.try_send(join).unwrap();
+    b.try_send(d.clone()).unwrap();
+    b.try_send(d).unwrap();
+    src.poll(&mut c, 10);
+    assert!(!src.stale(), "[dup] a duplicate is not a gap");
+    c.step();
+    assert_eq!(c.take(&o), z(&[(1, 1)]));
+}
+
+#[test]
+fn a_restarted_sender_is_corrected_not_trusted() {
+    let (mut sink, mut src, mut c, out, _) = mirror(8);
+    sink.committed(z(&[(1, 1)]));
+    sink.committed(z(&[(2, 1)]));
+    drive(&mut sink, &mut src, &mut c);
+    let _ = c.take(&out);
+    // A new incarnation that has since committed one command on top of {1, 2, 9}.
+    let mut sink = sink.restart(2, 0, z(&[(1, 1), (2, 1)]));
+    sink.committed(z(&[(9, 1)]));
+    drive(&mut sink, &mut src, &mut c);
+    assert_eq!(
+        c.take(&out),
+        z(&[(9, 1)]),
+        "[restart] only the difference is applied"
+    );
+    assert_eq!(src.stamp(), Some(Stamp { epoch: 2, seq: 1 }));
+}
+
+#[test]
+fn a_late_frame_from_an_epoch_the_receiver_left_changes_nothing() {
+    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<u8>();
+    let o = c.output(&s);
+    let mut src = EdgeSource::new(Direct, a, i);
+    let correction = |epoch, through, target| {
+        Frame::Correction(Correction {
+            epoch,
+            through,
+            target,
+        })
+    };
+    b.try_send(correction(1, 1, z(&[(1, 1)]))).unwrap();
+    b.try_send(correction(2, 3, z(&[(1, 1), (5, 1)]))).unwrap();
+    src.poll(&mut c, 10);
+    assert_eq!(src.stamp(), Some(Stamp { epoch: 2, seq: 3 }));
+    // A reply from the dead host that outlived it, and a delta that follows it.
+    b.try_send(correction(1, 9, z(&[(1, 1), (7, 1)]))).unwrap();
+    b.try_send(Frame::Delta(Delta {
+        epoch: 1,
+        from: 9,
+        to: 10,
+        delta: z(&[(8, 1)]),
+    }))
+    .unwrap();
+    src.poll(&mut c, 10);
+    c.step();
+    assert_eq!(
+        c.take(&o),
+        z(&[(1, 1), (5, 1)]),
+        "[left] epoch 1's late frames are refused"
+    );
+    assert_eq!(src.stamp(), Some(Stamp { epoch: 2, seq: 3 }));
+    assert!(!src.stale(), "[left] and they do not ask for anything");
+}
+
+#[test]
+fn an_unreadable_frame_asks_for_a_correction() {
+    let (a, mut b) = channel_pipes::<String>(8);
+    let mut c = Circuit::new();
+    let (i, s) = c.input::<u8>();
+    let o = c.output(&s);
+    let mut src = EdgeSource::new(Text, a, i);
+    b.try_send("D 1 0 1 1:1".to_string()).unwrap();
+    src.poll(&mut c, 10);
+    assert!(
+        src.stale(),
+        "[initial] a new receiver asks for a correction"
+    );
+    b.try_send("C 1 1 1:1".to_string()).unwrap();
+    src.poll(&mut c, 10);
+    assert!(!src.stale());
+    b.try_send("garbage".to_string()).unwrap();
+    src.poll(&mut c, 10);
+    assert!(src.stale(), "[unreadable] a frame that does not decode");
+    c.step();
+    assert_eq!(c.take(&o), z(&[(1, 1)]));
+}
+
+#[test]
+fn a_rolled_back_command_does_not_cross() {
+    let (mut sink, mut src, mut c, out, _) = mirror(8);
+    let (mut p, p_in, p_out) = producer();
+    p.begin();
+    p.push(&p_in, z(&[(1, 1)]));
+    p.step();
+    p.rollback();
+    assert_eq!(
+        sink.stamp(),
+        Stamp { epoch: 1, seq: 0 },
+        "[stamp] nothing consumed a number"
+    );
+    assert!(!sink.backlog(), "[send] nothing waits");
+    drive(&mut sink, &mut src, &mut c);
+    assert_eq!(c.take(&out), Z::new());
+    // The committed retry does cross, and is the first stamp.
+    p.begin();
+    p.push(&p_in, z(&[(1, 1)]));
+    p.step();
+    p.commit();
+    assert_eq!(
+        sink.committed_from(&mut p, &p_out),
+        Stamp { epoch: 1, seq: 1 }
+    );
+}
+
+#[test]
+#[should_panic(expected = "only after it commits")]
+fn taking_a_delta_inside_a_transaction_is_refused() {
+    let (mut sink, ..) = mirror(8);
+    let (mut p, p_in, p_out) = producer();
+    p.begin();
+    p.push(&p_in, z(&[(1, 1)]));
+    p.step();
+    sink.committed_from(&mut p, &p_out);
+}
+
+#[test]
+fn a_circuit_on_each_of_two_threads_equals_one_circuit() {
+    const N: u64 = 300;
+    let deltas: Vec<Z> = (0..N)
+        .map(|n| z(&[((n * 7 % 14) as u8, 1 - (n % 3) as i64), ((n % 5) as u8, 1)]))
+        .collect();
+    let (mut one, one_in, one_out) = single();
+    let mut want = Counts::new();
+    for d in &deltas {
+        one.push(&one_in, d.clone());
+        one.step();
+        want += one.take(&one_out);
+    }
+    let (a, b) = channel_pipes::<Frame<u8>>(2);
+    let finished = Arc::new(AtomicUsize::new(0));
+    let fin = finished.clone();
+    let producer_thread = thread::spawn(move || {
+        let (mut p, p_in, p_out) = producer();
+        let mut sink = EdgeSink::new(Direct, a, 9);
+        for (n, d) in deltas.iter().enumerate() {
+            p.begin();
+            p.push(&p_in, d.clone());
+            p.step();
+            p.commit();
+            sink.committed_from(&mut p, &p_out);
+            if n % 3 != 0 {
+                sink.pump();
+            }
+            if n % 7 == 0 {
+                thread::yield_now();
+            }
+        }
+        while fin.load(Ordering::SeqCst) == 0 {
+            sink.pump();
+            thread::yield_now();
+        }
+    });
+    let (mut c, c_in, c_out) = consumer();
+    let mut src = EdgeSource::new(Direct, b, c_in);
+    let mut got = Counts::new();
+    let mut spins = 0u64;
+    while src.stamp() != Some(Stamp { epoch: 9, seq: N }) {
+        spins += 1;
+        assert!(spins < 50_000_000, "[threads] the consumer never caught up");
+        src.poll(&mut c, 3);
+        c.step();
+        got += c.take(&c_out);
+        if spins.is_multiple_of(5) {
+            thread::yield_now();
+        }
+    }
+    c.step();
+    got += c.take(&c_out);
+    finished.store(1, Ordering::SeqCst);
+    producer_thread.join().unwrap();
+    assert_eq!(
+        got, want,
+        "[threads] the consumer after the producer thread finishes"
+    );
+}