From c2598f0c603822b36b0630c37faf586809563013 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-threads)" Date: Sat, 10 Oct 2026 20:50:07 -0700 Subject: [PATCH] feat(zset): a stamped, coalescing delta edge between two circuits Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/delivery.md | 2 + context/code/zset.md | 58 ++ context/test-evidence/zset.md | 70 ++ crates/zset/src/circuit.rs | 5 + crates/zset/src/edge.rs | 404 +++++++++++ crates/zset/src/lib.rs | 5 + crates/zset/tests/edge.proptest-regressions | 12 + crates/zset/tests/edge.rs | 704 ++++++++++++++++++++ 8 files changed, 1260 insertions(+) create mode 100644 crates/zset/src/edge.rs create mode 100644 crates/zset/tests/edge.proptest-regressions create mode 100644 crates/zset/tests/edge.rs diff --git a/context/code/delivery.md b/context/code/delivery.md index 17db04d6..96ab4cfe 100644 --- a/context/code/delivery.md +++ b/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. diff --git a/context/code/zset.md b/context/code/zset.md index 28e74b4c..246be461 100644 --- a/context/code/zset.md +++ b/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` turns a `Frame` 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. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index f8aaf7b4..29166d5d 100644 --- a/context/test-evidence/zset.md +++ b/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. diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs index 08e7f0d2..1ede0c08 100644 --- a/crates/zset/src/circuit.rs +++ b/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 diff --git a/crates/zset/src/edge.rs b/crates/zset/src/edge.rs new file mode 100644 index 00000000..2b1415d8 --- /dev/null +++ b/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 { + pub epoch: u64, + pub from: u64, + pub to: u64, + pub delta: ZSet, +} + +/// The sender's whole integral as of `through`; the receiver applies `target - held`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Correction { + pub epoch: u64, + pub through: u64, + pub target: ZSet, +} + +/// "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, +} + +/// Everything that crosses an edge, in both directions. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Frame { + Delta(Delta), + Correction(Correction), + 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 { + type Wire: Send + 'static; + fn encode(&self, frame: &Frame) -> Self::Wire; + fn decode(&self, wire: Self::Wire) -> Result, CodecError>; +} + +/// No encoding: the frame itself crosses. +#[derive(Clone, Copy, Debug, Default)] +pub struct Direct; + +impl EdgeCodec for Direct { + type Wire = Frame; + + fn encode(&self, frame: &Frame) -> Frame { + frame.clone() + } + + fn decode(&self, wire: Frame) -> Result, CodecError> { + Ok(wire) + } +} + +/// Why a pipe took nothing; `Full` is backpressure, `Closed` is the end. +#[derive(Debug, PartialEq, Eq)] +pub enum Refused { + 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: Send { + fn try_send(&mut self, wire: W) -> Result<(), Refused>; + fn try_recv(&mut self) -> Result; +} + +/// An in-process pipe end over bounded channels. +pub struct ChannelPipe { + tx: mpsc::SyncSender, + rx: mpsc::Receiver, +} + +/// Two connected ends; each direction holds `cap` items. +pub fn channel_pipes(cap: usize) -> (ChannelPipe, ChannelPipe) { + 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 Pipe for ChannelPipe { + fn try_send(&mut self, wire: W) -> Result<(), Refused> { + 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 { + 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 { + codec: C, + pipe: P, + epoch: u64, + seq: u64, + integral: ZSet, + /// The commands `(sent, seq]`, summed; `sent == seq` means nothing waits. + waiting: Delta, + owes_correction: bool, + closed: bool, +} + +impl, P: Pipe> EdgeSink { + /// 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) -> 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) -> 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) -> Stamp { + self.seq += 1; + self.integral += δ + 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) -> 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 { + codec: C, + pipe: P, + input: Input, + joined: Option, + watermark: u64, + held: ZSet, + left: VecDeque, + stale: bool, + asked: bool, + closed: bool, +} + +impl, P: Pipe> EdgeSource { + /// A receiver that holds nothing and asks for a correction on its first `poll`. + pub fn new(codec: C, pipe: P, input: Input) -> 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 { + 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) -> 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) { + self.held += δ + circuit.push(&self.input, delta); + } + + fn on_delta(&mut self, circuit: &mut Circuit, d: Delta) { + 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) { + 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, + } + } +} diff --git a/crates/zset/src/lib.rs b/crates/zset/src/lib.rs index e18e5ebf..3c59806b 100644 --- a/crates/zset/src/lib.rs +++ b/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}; diff --git a/crates/zset/tests/edge.proptest-regressions b/crates/zset/tests/edge.proptest-regressions new file mode 100644 index 00000000..6554d569 --- /dev/null +++ b/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 diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs new file mode 100644 index 00000000..c177ef0c --- /dev/null +++ b/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; +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, Output) { + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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) -> 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, Output<(u8, i64)>) { + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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, Output<(u8, i64)>) { + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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 { + inner: ChannelPipe, + lose: Arc, +} + +impl Pipe for FaultPipe { + fn try_send(&mut self, wire: W) -> Result<(), Refused> { + 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 { + self.inner.try_recv() + } +} + +fn fault( + cap: usize, +) -> ( + FaultPipe, + FaultPipe, + Arc, + Arc, +) { + 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::>() + .join(",") +} + +fn unzs(s: &str) -> Result { + 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::().map_err(|_| bad())?, + w.parse::().map_err(|_| bad())?, + )) + }) + .collect() +} + +impl EdgeCodec for Text { + type Wire = String; + + fn encode(&self, f: &Frame) -> 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, 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::().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 = EdgeSink>::Wire>>; +type Source = EdgeSource>::Wire>>; + +/// Two circuits joined by an edge, and the combined circuit they must equal. +struct World + Clone> { + codec: C, + prod: Circuit, + p_in: Input, + p_out: Output, + sink: Sink, + lose_fwd: Arc, + cons: Circuit, + c_out: Output<(u8, i64)>, + src: Source, + lose_back: Arc, + got: Counts, + p_total: Z, + epoch: u64, + one: Circuit, + one_in: Input, + one_out: Output<(u8, i64)>, + want: Counts, +} + +impl + Clone> World { + 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() -> FaultPipe { + 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> { + prop::collection::vec((0u8..14, -2i64..=2), 0..4) +} + +fn act() -> impl Strategy { + 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 + 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>( + sink: &mut EdgeSink>, + src: &mut EdgeSource>, + 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, + Source, + Circuit, + Output, + Arc, +) { + let (a, b, lose_fwd, _) = fault(cap); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::>(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::>(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::>(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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 = (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::>(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" + ); +}