fiddlerwoaroof/improvise
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #363 from fiddlerwoaroof/dbsp/zset-edge

Browse Source 
feat(zset): stamped coalescing delta edge between circuits
This commit is contained in:
Ed Langley
2026-10-10 22:11:48 -07:00
committed by GitHub
parent 08bacbe265 6d7cd481f9
commit 3d1dcfd5e4
10 changed files with 2056 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
context/code/delivery.md
+2
View File
@@ -259,3 +259,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`.** The `zset` crate's edge module, `crates/zset/src/edge.rs` (`context/code/zset.md`, "Edges") carries Z-set deltas between two circuits with the rules of this file (stamps, coalescing by summing, `target - held`, token-matched answers) and no log; this layer is unchanged and does not use it.
context/code/zset.md
+85
View File
@@ -311,3 +311,88 @@ threads are out of scope, and nothing in `zset` spawns one. `Send` and `mpsc` co
`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.
## 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, a frame that does not decode) is a *mark*: `stale` is set and a counter of
troubles moves. `stale` makes the next `poll` send a `Request`.
5. **A receiver joins an incarnation only by the answer to its latest request.** A `Request`
carries a token (the receiver's counter, which `EdgeSource::restart` keeps), and the sender
echoes the token of the last request it read in its `Correction { epoch, through: seq, token,
target: integral }`. The receiver pushes `target - held` (delivery's rule). This carries over
delivery's third fix, "the host that answers a request is live" (`delivery.md` section 4): a
frame from an incarnation the receiver has not joined only marks it stale and never moves the
watermark, unless it is the answer to the request just sent. A reply to an older request, a
duplicate, or a late reply from a dead sender is therefore never a join. A correction from the
joined incarnation applies when it is ahead of the watermark, whoever asked. There is no shared
log as in delivery, so every answer is a full correction. Two earlier designs were wrong, each
found by a property: joining by a first delta with `from == 0` (a restarted sender with a
non-empty integral at `seq 0` looks fresh), and a `left` list of incarnations (a late reply
from an incarnation never joined sent the receiver back and shut the live sender out:
`a_late_reply_from_a_dead_incarnation_changes_nothing`). There is no `left` list now.
6. **What clears `stale`.** Only an answer, and only if it is not behind the watermark
(`an_answer_older_than_the_watermark_does_not_clear_stale`) and no mark has been made since
its request went out (`trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale`);
an answer at the watermark says nothing newer exists
(`an_answer_at_the_watermark_says_nothing_newer_and_clears_stale`).
7. **A restart is a new epoch, and an epoch is never reused.** `EdgeSink::restart(epoch, seq,
integral)` keeps the pipe and codec. The first frame of an unjoined epoch makes the receiver
re-ask once for that epoch, because the request may have died with the sender that read it
(`a_new_epoch_frame_re_asks_once_after_a_lost_request`). A sender that reuses an epoch id at
a lower seq cannot be told from a reordered old frame, so the receiver ignores it as a
duplicate and is stranded until the sender restarts again: the caller must pass a fresh epoch.
A receiver that restarts (`EdgeSource::restart`) is stale again and keeps its token counter.
A brand-new receiver (`EdgeSource::new`, a process that lost its state) seeds the counter from
a hash of a process-wide counter under `RandomState`: random across processes natively and
unique per process on wasm32, which has no entropy. Seeding at 0 let a dead receiver's late
reply match a new receiver's request and join an older incarnation
(`review_token_collision_after_new_source`).
8. **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, and it must be longer than the
round trip, or every answer is one request behind; a host that times out repeatedly must back
off, not retry at a fixed short interval. Requests carry the receiver's stamp and
this sender does not read it.
9. **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`).
10. **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`). Two properties simulate arbitrary
interleavings: `tests/edge.rs` over FIFO pipes that lose frames, and `tests/edge_net.rs` over a
network that drops, duplicates, garbles and reorders in-flight frames both ways, restarts either
end at any seq and times out (capacity 1 to 4). It checks after every action that the
receiver's applied state is the sender's integral at the stamp it claims, and that it never
moves to an older incarnation.
`poll` and `committed_from` panic inside a transaction: a rollback drops what a poll pushed,
and the stamps would not know (`a_poll_inside_a_transaction_is_refused`).
`committed` checks overflow before it changes anything, so a caught panic leaves the sink as
it was. `pump` clones the waiting delta each attempt against a full pipe; that is left.
Native threads are assumed; `mpsc` and `Send` compile on wasm32 and nothing runs there.
11. **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
+146
View File
@@ -616,6 +616,137 @@ as at `begin`. The undo log lives in the nodes, which move with the circuit.
### 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,
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_reply_from_a_dead_incarnation_changes_nothing
The reviewer's case: epoch 100's reply to the first request is delayed, the sender restarts as
1001, the receiver asks again and joins it, and then the late reply arrives. The first design
(a `left` list) went back to epoch 100 and then dropped every 1001 frame. Now the late reply is
not the answer to the latest request, so it only marks stale. Rival N2 (a foreign correction
accepted) fails it.
### an_unsolicited_correction_from_another_incarnation_only_marks_stale
A correction with a token nobody asked for, from another epoch, leaves the stamp as it was and
makes the receiver ask. Rival N2 fails it.
### only_the_reply_to_the_latest_request_may_join_a_new_incarnation
Two requests are out (a timeout between them); the reply to the first, from another epoch, does
not join, and the reply to the second does. Rival N1 (any reply while asked is the answer) fails
it.
### an_answer_older_than_the_watermark_does_not_clear_stale
An answer through 1 to a receiver at 3 neither moves it back nor says it is current. Rival N6
(`<=` where `==` belongs) fails it.
### an_answer_at_the_watermark_says_nothing_newer_and_clears_stale
A repeat answer at the watermark clears `stale`; without the line the receiver asks for ever.
### trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale
A gap seen after the request went out, then an answer through the old watermark: still stale.
Rival N5 (no comparison of marks) fails it.
### a_new_epoch_frame_re_asks_once_after_a_lost_request
The sender read a request and restarted before answering. Five frames from the new epoch cause
exactly one more request. Rival N3 (no re-ask) fails the first half and rival N4 (a re-ask for
every foreign frame) the second.
### a_poll_inside_a_transaction_is_refused
A poll outside a transaction followed by a rolled-back consumer step loses nothing; a poll inside
one panics. The reviewer's sequence stranded the receiver at `(1, 1)` holding nothing. Rival N8
(the assertion removed) fails it.
### a_zero_capacity_pipe_is_refused
`channel_pipes(0)` made a rendezvous channel that `try_send` could never use. Rival N10 fails it.
### a_committed_that_overflows_leaves_the_sink_as_it_was
A weight overflow in `committed` no longer leaves a bumped `seq` or half an integral behind.
Rival N9 (mutating first) fails it.
### the_receiver_is_never_wrong_over_a_hostile_network
The reviewer's generator as a permanent property: dropped, duplicated, garbled and reordered
frames in both directions, restarts at any seq, timeouts and receiver restarts, capacity 1 to 4.
After every action the receiver's applied state equals the sender's integral at the stamp it
claims and its epoch never goes down; after the network is reliable it catches up. Run at 20,000
cases three times, all green. Rivals N4 and N7 (a foreign delta moves the receiver) fail 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.
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
@@ -632,3 +763,18 @@ Rival: `unsafe impl Sync for Circuit {}` fails the build of `tests/threads.rs`.
A sink over `Counted` keys clones each key once (the insert into its sum). Rival: a sink that
clones its input piece first, as the reviewer injected, makes `N` more.
### review_token_collision_after_new_source
The old receiver's request 2 is answered by E1 and the reply is delayed; E1 dies, E2 takes over;
a brand-new receiver joins E2, times out, asks its request 2, and the dead reply matched it and
took the receiver back to epoch 1. Tokens started at 0 for every `new`. Now they start at a
random base. Rival T1 (base 0) fails it. The deep property found nothing at 256 cases; this
history is the check.
### the_reviewers_late_reply_case_is_a_history
Now polls one frame at a time, so each late frame is judged alone (one `Poll(4)` hid it). Rival
N2 (a foreign correction accepted) fails it. N1 (any reply while asked) is caught by
`only_the_reply_to_the_latest_request_may_join_a_new_incarnation` and the collision history; the
random property needs a deep run (20,000 cases) to find N1, and finds N2 at the default 256.
crates/zset/src/circuit.rs
+5
View File
@@ -499,6 +499,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
+463
View File
@@ -0,0 +1,463 @@
//! 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::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,
/// The token of the request this answers.
pub token: u64,
pub target: ZSet<K>,
}
/// "Bring me up from here". `have` is carried and never read by this sender; `token` is echoed
/// by the answer, and only the answer to the latest request may move the receiver to a new epoch.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Request {
pub have: Option<Stamp>,
pub token: u64,
}
/// 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>) {
assert!(cap > 0, "zset: a channel pipe holds at least one item");
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>,
/// The token of the latest request read and not yet answered.
owed: Option<u64>,
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(),
},
owed: None,
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.owed.is_some()
}
/// 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 {
preflight(&self.integral, &delta);
preflight(&self.waiting.delta, &delta);
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(r)) = self.codec.decode(wire) {
self.owed = Some(r.token);
}
}
Err(Recv::Empty) => break,
Err(Recv::Closed) => self.closed = true,
}
}
if self.closed {
return Flow::Closed;
}
let frame = if let Some(token) = self.owed {
Frame::Correction(Correction {
epoch: self.epoch,
through: self.seq,
token,
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.owed = None;
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
}
}
}
}
/// A token base no other receiver incarnation shares (`context/code/zset.md`, "Edges", item 7):
/// a hash of a process-wide counter under `RandomState`.
fn fresh_token_base() -> u64 {
use std::hash::BuildHasher;
use std::sync::atomic::{AtomicU64, Ordering};
static NEXT: AtomicU64 = AtomicU64::new(1);
let n = NEXT.fetch_add(1, Ordering::Relaxed);
std::collections::hash_map::RandomState::new().hash_one(n) >> 1
}
/// Panics, before anything changes, if adding `d` to `z` would overflow a weight.
fn preflight<K: Ord>(z: &ZSet<K>, d: &ZSet<K>) {
for (k, w) in d.iter() {
crate::plus(z.weight(k), w);
}
}
/// 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>,
stale: bool,
/// A request is out. `token` is its token, and `marks` counted the troubles seen so far.
asked: bool,
token: u64,
marks: u64,
ask_marks: u64,
/// The foreign incarnation a re-ask was already made for.
noted: Option<u64>,
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(),
stale: true,
asked: false,
token: fresh_token_base(),
marks: 0,
ask_marks: 0,
noted: None,
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 {
let token = self.token;
EdgeSource {
token,
..Self::new(self.codec, self.pipe, input)
}
}
/// What has been applied to the circuit's input so far.
pub fn held(&self) -> &ZSet<K> {
&self.held
}
/// 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, then asks if wanted; returns the frames read. Panics inside
/// a transaction, where a rollback would drop what was pushed.
pub fn poll(&mut self, circuit: &mut Circuit, max: usize) -> usize {
assert!(
!circuit.in_transaction(),
"zset: an edge applies frames only outside a transaction"
);
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.mark(),
}
}
Err(Recv::Empty) => break,
Err(Recv::Closed) => self.closed = true,
}
}
self.ask();
read
}
/// Something is wrong or unknown: ask, and no answer to an earlier request may say otherwise.
fn mark(&mut self) {
self.stale = true;
self.marks += 1;
}
/// A frame from an unjoined epoch never moves the watermark; the first one per epoch
/// re-asks, since the request may have died with the sender that read it.
fn foreign(&mut self, epoch: u64) {
self.mark();
if self.noted != Some(epoch) {
self.noted = Some(epoch);
self.asked = false;
}
}
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.joined != Some(d.epoch) {
return self.foreign(d.epoch);
}
if d.to <= self.watermark {
return;
}
if d.from == self.watermark {
self.watermark = d.to;
self.apply(circuit, d.delta);
} else {
self.mark();
}
}
fn on_correction(&mut self, circuit: &mut Circuit, c: Correction<K>) {
let answer = self.asked && c.token == self.token;
let known = self.joined == Some(c.epoch);
if !known && !answer {
return self.foreign(c.epoch);
}
if answer {
self.asked = false;
self.noted = None;
}
if !known || c.through > self.watermark {
self.joined = Some(c.epoch);
self.watermark = c.through;
let delta = &c.target - &self.held;
self.apply(circuit, delta);
}
if answer && c.through == self.watermark && self.marks == self.ask_marks {
self.stale = false;
}
}
fn ask(&mut self) {
if !self.stale || self.asked || self.closed {
return;
}
let token = self.token + 1;
let wire = self.codec.encode(&Frame::Request(Request {
have: self.stamp(),
token,
}));
match self.pipe.try_send(wire) {
Ok(()) => {
self.asked = true;
self.token = token;
self.ask_marks = self.marks;
}
Err(Refused::Full(_)) => {}
Err(Refused::Closed) => self.closed = true,
}
}
}
crates/zset/src/lib.rs
+5
View File
@@ -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
+12
View File
@@ -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
+875
View File
@@ -0,0 +1,875 @@
//! 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`.
mod support;
use proptest::prelude::*;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
use support::Text;
use zset::{
ChannelPipe, Circuit, Correction, Delta, Direct, EdgeCodec, EdgeSink, EdgeSource, Flow, Frame,
Input, Output, Pipe, Recv, Refused, 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,
)
}
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);
src.poll(&mut c, 10);
let join = answer(&mut b, 1, 0, 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 }));
}
/// Answers the receiver's latest request in `b` with a correction that echoes its token.
fn answer(b: &mut ChannelPipe<Frame<u8>>, epoch: u64, through: u64, target: Z) -> Frame<u8> {
let mut token = None;
while let Ok(f) = b.try_recv() {
if let Frame::Request(r) = f {
token = Some(r.token);
}
}
Frame::Correction(Correction {
epoch,
through,
token: token.expect("a request was waiting"),
target,
})
}
fn reply(b: &mut ChannelPipe<Frame<u8>>, epoch: u64, through: u64, target: Z) {
let f = answer(b, epoch, through, target);
b.try_send(f).unwrap();
}
#[test]
fn a_late_reply_from_a_dead_incarnation_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);
src.poll(&mut c, 10);
// Epoch 100 reads the first request; its reply is delayed. The sender restarts as 1001.
let late = answer(&mut b, 100, 1, z(&[(9, 1)]));
src.connection_lost();
src.poll(&mut c, 10);
reply(&mut b, 1001, 1, z(&[(5, 1)]));
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp {
epoch: 1001,
seq: 1
})
);
b.try_send(late).unwrap();
b.try_send(Frame::Delta(Delta {
epoch: 1001,
from: 1,
to: 2,
delta: z(&[(6, 1)]),
}))
.unwrap();
src.poll(&mut c, 10);
c.step();
assert_eq!(
src.stamp(),
Some(Stamp {
epoch: 1001,
seq: 2
}),
"[late] the reply of an incarnation never joined does not move the receiver back"
);
assert_eq!(
c.take(&o),
z(&[(5, 1), (6, 1)]),
"[late] and 1001 is not shut out"
);
}
#[test]
fn an_unsolicited_correction_from_another_incarnation_only_marks_stale() {
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);
src.poll(&mut c, 10);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 10);
assert!(!src.stale());
let wrong = Frame::Correction(Correction {
epoch: 2,
through: 9,
token: 77,
target: z(&[(3, 1)]),
});
b.try_send(wrong).unwrap();
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp { epoch: 1, seq: 1 }),
"[foreign] not moved"
);
assert!(src.stale(), "[foreign] but it asks");
}
#[test]
fn only_the_reply_to_the_latest_request_may_join_a_new_incarnation() {
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);
src.poll(&mut c, 10);
let late = answer(&mut b, 100, 1, z(&[(9, 1)]));
src.connection_lost();
src.poll(&mut c, 10);
b.try_send(late).unwrap();
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
None,
"[token] the reply to the first request is not the answer"
);
reply(&mut b, 1001, 1, z(&[(5, 1)]));
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp {
epoch: 1001,
seq: 1
})
);
}
#[test]
fn an_answer_older_than_the_watermark_does_not_clear_stale() {
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);
src.poll(&mut c, 10);
reply(&mut b, 1, 3, Z::new());
src.poll(&mut c, 10);
src.connection_lost();
src.poll(&mut c, 10);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp { epoch: 1, seq: 3 }),
"[older] never moves back"
);
assert!(
src.stale(),
"[older] and it does not say the receiver is current"
);
}
#[test]
fn a_new_epoch_frame_re_asks_once_after_a_lost_request() {
let (a, mut b) = channel_pipes::<Frame<u8>>(64);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
assert!(
matches!(b.try_recv(), Ok(Frame::Request(_))),
"[first] the request"
);
// The sender read it and restarted before answering.
let mut asks = 0;
for n in 0..5u64 {
b.try_send(Frame::Delta(Delta {
epoch: 2,
from: n,
to: n + 1,
delta: Z::new(),
}))
.unwrap();
src.poll(&mut c, 8);
while let Ok(Frame::Request(_)) = b.try_recv() {
asks += 1;
}
}
assert_eq!(
asks, 1,
"[reask] once for the new epoch, not once per frame"
);
}
#[test]
fn an_answer_at_the_watermark_says_nothing_newer_and_clears_stale() {
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);
src.poll(&mut c, 8);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 8);
src.connection_lost();
src.poll(&mut c, 8);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 8);
assert!(
!src.stale(),
"[dup-answer] through == watermark clears stale"
);
}
#[test]
fn trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale() {
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);
src.poll(&mut c, 8);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 8);
src.connection_lost();
src.poll(&mut c, 8);
let reply = answer(&mut b, 1, 1, Z::new());
// A gap seen after the request went out: the reply was cut before it.
b.try_send(Frame::Delta(Delta {
epoch: 1,
from: 4,
to: 5,
delta: Z::new(),
}))
.unwrap();
b.try_send(reply).unwrap();
src.poll(&mut c, 8);
assert!(src.stale(), "[marks] the reply predates the gap");
}
#[test]
fn a_poll_inside_a_transaction_is_refused() {
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);
src.poll(&mut c, 8);
reply(&mut b, 1, 1, z(&[(1, 1)]));
// Outside a transaction a rollback of the consumer's step keeps what was polled.
src.poll(&mut c, 8);
c.begin();
c.step();
c.rollback();
c.step();
assert_eq!(
c.take(&o),
z(&[(1, 1)]),
"[txn] a step rolled back after poll loses nothing"
);
c.begin();
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| src.poll(&mut c, 8)));
assert!(
r.is_err(),
"[txn] a poll inside a transaction would be dropped by rollback"
);
}
#[test]
#[should_panic(expected = "at least one item")]
fn a_zero_capacity_pipe_is_refused() {
let _ = channel_pipes::<u8>(0);
}
#[test]
fn a_committed_that_overflows_leaves_the_sink_as_it_was() {
let (mut sink, mut src, mut c, _, _) = mirror(8);
sink.committed(z(&[(1, i64::MAX)]));
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
sink.committed(z(&[(2, 1), (1, 1)]))
}));
assert!(r.is_err(), "the weight overflows");
assert_eq!(
sink.stamp(),
Stamp { epoch: 1, seq: 1 },
"[overflow] no stamp consumed"
);
drive(&mut sink, &mut src, &mut c);
assert_eq!(
src.held(),
&z(&[(1, i64::MAX)]),
"[overflow] and the key 2 did not slip in"
);
}
#[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"
);
let request = b.try_recv().unwrap();
let token = request.split(' ').nth(1).unwrap();
b.try_send(format!("C 1 1 {token} 1:1")).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();
// The host's command fails and rolls back; the sink is not asked.
p.begin();
p.push(&p_in, z(&[(1, 1)]));
p.step();
p.rollback();
// The retry is a different command and commits.
p.begin();
p.push(&p_in, z(&[(2, 1)]));
p.step();
p.commit();
assert_eq!(
sink.committed_from(&mut p, &p_out),
Stamp { epoch: 1, seq: 1 }
);
drive(&mut sink, &mut src, &mut c);
assert_eq!(
c.take(&out),
z(&[(2, 1)]),
"[retry] only the committed command crossed"
);
}
#[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"
);
}
crates/zset/tests/edge_net.rs
+380
View File
@@ -0,0 +1,380 @@
//! An edge over a network that drops, duplicates, garbles and reorders frames.
//! Evidence: `context/test-evidence/zset.md`.
mod support;
use proptest::prelude::*;
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use support::Text;
use zset::{Circuit, EdgeSink, EdgeSource, Output, Pipe, Recv, Refused, Stamp, ZSet};
type Z = ZSet<u8>;
/// One direction of the network: frames in flight (which the test may reorder, drop, duplicate
/// or garble) and frames delivered (which `try_recv` takes in order).
#[derive(Default)]
struct Lane {
flight: Vec<String>,
ready: VecDeque<String>,
}
#[derive(Default)]
struct Net {
lanes: [Lane; 2],
cap: usize,
}
type Shared = Arc<Mutex<Net>>;
/// One end: sends into lane `out`, receives from the other one.
struct End {
net: Shared,
out: usize,
}
impl Pipe<String> for End {
fn try_send(&mut self, wire: String) -> Result<(), Refused<String>> {
let mut n = self.net.lock().unwrap();
let cap = n.cap;
let lane = &mut n.lanes[self.out];
if lane.flight.len() + lane.ready.len() >= cap {
return Err(Refused::Full(wire));
}
lane.flight.push(wire);
Ok(())
}
fn try_recv(&mut self) -> Result<String, Recv> {
let mut n = self.net.lock().unwrap();
n.lanes[1 - self.out].ready.pop_front().ok_or(Recv::Empty)
}
}
fn net(cap: usize) -> (Shared, End, End) {
let shared: Shared = Arc::new(Mutex::new(Net {
cap,
..Net::default()
}));
let sink_end = End {
net: shared.clone(),
out: 0,
};
let src_end = End {
net: shared.clone(),
out: 1,
};
(shared, sink_end, src_end)
}
struct World {
net: Shared,
sink: EdgeSink<u8, Text, End>,
src: EdgeSource<u8, Text, End>,
cons: Circuit,
out: Output<u8>,
got: Z,
total: Z,
epoch: u64,
seq: u64,
history: HashMap<(u64, u64), Z>,
joined_max: u64,
}
impl World {
fn new(cap: usize) -> Self {
let (shared, a, b) = net(cap);
let mut cons = Circuit::new();
let (i, s) = cons.input::<u8>();
let out = cons.output(&s);
let mut history = HashMap::new();
history.insert((1, 0), Z::new());
World {
net: shared,
sink: EdgeSink::new(Text, a, 1),
src: EdgeSource::new(Text, b, i),
cons,
out,
got: Z::new(),
total: Z::new(),
epoch: 1,
seq: 0,
history,
joined_max: 0,
}
}
fn commit(&mut self, d: Z) {
self.total += &d;
self.seq += 1;
self.sink.committed(d);
self.history
.insert((self.epoch, self.seq), self.total.clone());
}
fn restart_sink(&mut self, seq: u64) {
self.epoch += 1;
self.seq = seq;
let old = std::mem::replace(
&mut self.sink,
EdgeSink::new(
Text,
End {
net: self.net.clone(),
out: 0,
},
0,
),
);
self.sink = old.restart(self.epoch, seq, self.total.clone());
self.history.insert((self.epoch, seq), self.total.clone());
}
/// The consumer process lost its state: a brand-new receiver, with no token to carry over.
fn new_source(&mut self) {
let mut cons = Circuit::new();
let (i, s) = cons.input::<u8>();
self.out = cons.output(&s);
let end = End {
net: self.net.clone(),
out: 1,
};
self.src = EdgeSource::new(Text, end, i);
self.cons = cons;
self.got = Z::new();
self.joined_max = 0;
}
fn restart_source(&mut self) {
let mut cons = Circuit::new();
let (i, s) = cons.input::<u8>();
self.out = cons.output(&s);
let old = std::mem::replace(
&mut self.src,
EdgeSource::new(
Text,
End {
net: self.net.clone(),
out: 1,
},
i,
),
);
self.src = old.restart(i);
self.cons = cons;
self.got = Z::new();
self.joined_max = 0;
}
fn step(&mut self) {
self.cons.step();
self.got += self.cons.take(&self.out);
}
fn lane(&mut self, dir: usize, idx: usize, f: impl FnOnce(&mut Lane, usize)) {
let mut n = self.net.lock().unwrap();
let lane = &mut n.lanes[dir];
if !lane.flight.is_empty() {
let at = idx % lane.flight.len();
f(lane, at);
}
}
/// The two safety rules, after every action.
fn check(&mut self, when: &str) {
if let Some(Stamp { epoch, seq }) = self.src.stamp() {
assert!(
epoch >= self.joined_max,
"[older] {when}: the receiver went back from epoch {} to {epoch}",
self.joined_max
);
self.joined_max = epoch;
let want = self.history.get(&(epoch, seq));
assert_eq!(
Some(self.src.held()),
want,
"[claim] {when}: held is not the sender's integral at ({epoch}, {seq})"
);
}
}
fn settle(&mut self) {
for _ in 0..400 {
let silent = {
let mut n = self.net.lock().unwrap();
let silent = n
.lanes
.iter()
.all(|l| l.flight.is_empty() && l.ready.is_empty());
for lane in &mut n.lanes {
let flight = std::mem::take(&mut lane.flight);
lane.ready.extend(flight);
}
silent
};
self.sink.pump();
let read = self.src.poll(&mut self.cons, 8);
self.step();
self.check("settle");
let quiet = {
let n = self.net.lock().unwrap();
n.lanes
.iter()
.all(|l| l.flight.is_empty() && l.ready.is_empty())
};
if quiet
&& !self.sink.backlog()
&& !self.src.stale()
&& self.src.stamp() == Some(self.sink.stamp())
{
break;
}
if silent && read == 0 {
// Nothing was in flight: the reply is lost, not late. Only now does the timeout fire.
self.src.connection_lost();
}
}
assert_eq!(
self.src.stamp(),
Some(self.sink.stamp()),
"[settle] never caught up"
);
self.step();
assert_eq!(self.got, self.total, "[settle] the consumer's integral");
}
}
#[derive(Clone, Debug)]
enum Act {
Commit(Vec<(u8, i64)>),
Pump,
Poll(usize),
Step,
Timeout,
Deliver(usize, usize),
Drop(usize, usize),
Dup(usize, usize),
Garble(usize, usize),
RestartSink(u64),
RestartSource,
NewSource,
}
fn act() -> impl Strategy<Value = Act> {
let lane = 0usize..2;
let at = 0usize..8;
prop_oneof![
5 => prop::collection::vec((0u8..10, -2i64..=2), 0..3).prop_map(Act::Commit),
4 => Just(Act::Pump),
4 => (1usize..4).prop_map(Act::Poll),
2 => Just(Act::Step),
2 => Just(Act::Timeout),
8 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Deliver(l, i)),
2 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Drop(l, i)),
2 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Dup(l, i)),
1 => (lane, at).prop_map(|(l, i)| Act::Garble(l, i)),
1 => (0u64..8).prop_map(Act::RestartSink),
1 => Just(Act::RestartSource),
1 => Just(Act::NewSource),
]
}
fn run(cap: usize, acts: &[Act]) {
let mut w = World::new(cap);
for (n, a) in acts.iter().enumerate() {
match a {
Act::Commit(d) => w.commit(d.iter().copied().collect()),
Act::Pump => {
w.sink.pump();
}
Act::Poll(m) => {
w.src.poll(&mut w.cons, *m);
}
Act::Step => w.step(),
Act::Timeout => w.src.connection_lost(),
Act::Deliver(d, i) => w.lane(*d, *i, |l, at| {
let f = l.flight.remove(at);
l.ready.push_back(f);
}),
Act::Drop(d, i) => w.lane(*d, *i, |l, at| {
l.flight.remove(at);
}),
Act::Dup(d, i) => w.lane(*d, *i, |l, at| {
let f = l.flight[at].clone();
l.flight.push(f);
}),
Act::Garble(d, i) => w.lane(*d, *i, |l, at| l.flight[at] = "garbage".to_string()),
Act::RestartSink(seq) => w.restart_sink(*seq),
Act::RestartSource => w.restart_source(),
Act::NewSource => w.new_source(),
}
w.check(&format!("after action {n} ({a:?})"));
}
w.settle();
}
proptest! {
/// Any history over a lossy, duplicating, garbling, reordering network keeps both safety
/// rules at every step, and settles to the sender's integral once the network is reliable.
#[test]
fn the_receiver_is_never_wrong_over_a_hostile_network(
acts in prop::collection::vec(act(), 0..60), cap in 1usize..5,
) {
run(cap, &acts);
}
}
#[test]
fn the_reviewers_late_reply_case_is_a_history() {
// Epoch 1 reads a request, its reply is delayed, the sender restarts, and the late reply comes.
// One frame per poll, so each is judged on its own.
let acts = [
Act::Poll(1),
Act::Pump,
Act::Commit(vec![(1, 1)]),
Act::Deliver(1, 0),
Act::Pump,
Act::RestartSink(3),
Act::Timeout,
Act::Poll(1),
Act::Deliver(1, 0),
Act::Pump,
Act::Deliver(0, 1),
Act::Poll(1),
Act::Deliver(0, 0),
Act::Poll(1),
Act::Poll(1),
];
run(2, &acts);
}
#[test]
fn review_token_collision_after_new_source() {
// The old receiver's request 2 is answered by E1 and the reply is delayed. E1 dies, E2 takes
// over, a new receiver joins E2 and times out, and the dead reply arrives for its request 2.
let acts = [
Act::Poll(1),
Act::Deliver(1, 0),
Act::Pump,
Act::Deliver(0, 0),
Act::Poll(4),
Act::Commit(vec![(1, 1)]),
Act::Timeout,
Act::Poll(1),
Act::Deliver(1, 0),
Act::Pump,
Act::RestartSink(5),
Act::Commit(vec![(2, 1)]),
Act::NewSource,
Act::Poll(1),
Act::Deliver(1, 0),
Act::Pump,
Act::Deliver(0, 1),
Act::Poll(4),
Act::Timeout,
Act::Poll(1),
Act::Deliver(0, 0),
Act::Poll(4),
];
run(4, &acts);
}
crates/zset/tests/support/mod.rs
+83
View File
@@ -0,0 +1,83 @@
//! A text wire form of `u8`-keyed frames, shared by the edge tests.
use zset::{CodecError, Correction, Delta, EdgeCodec, Frame, Request, Stamp, ZSet};
type Z = ZSet<u8>;
/// A text wire form, to run the same histories through an encoding.
#[derive(Clone, Copy)]
pub struct Text;
pub fn zs(z: &Z) -> String {
z.iter()
.map(|(k, w)| format!("{k}:{w}"))
.collect::<Vec<_>>()
.join(",")
}
pub 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, c.token, zs(&c.target))
}
Frame::Request(r) => match r.have {
Some(s) => format!("R {} {} {}", r.token, s.epoch, s.seq),
None => format!("R {}", r.token),
},
}
}
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(5, ' ');
it.next();
Ok(Frame::Correction(Correction {
epoch: num(it.next())?,
through: num(it.next())?,
token: num(it.next())?,
target: unzs(it.next().ok_or_else(bad)?)?,
}))
}
Some("R") => {
let token = num(it.next())?;
let have = match it.next() {
None => None,
e => Some(Stamp {
epoch: num(e)?,
seq: num(it.next())?,
}),
};
Ok(Frame::Request(Request { have, token }))
}
_ => Err(bad()),
}
}
}