diff --git a/context/code/delivery.md b/context/code/delivery.md index 96ab4cfe..16897206 100644 --- a/context/code/delivery.md +++ b/context/code/delivery.md @@ -243,4 +243,4 @@ pushes `Host::take_notices()`. - **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. +**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. diff --git a/context/code/zset-ffi.md b/context/code/zset-ffi.md index d9020e95..37fc794e 100644 --- a/context/code/zset-ffi.md +++ b/context/code/zset-ffi.md @@ -75,7 +75,8 @@ link, so they call the `extern "C"` functions as plain functions. shared cell rather than panicking. 10. **Threads** (`zset.md`, "Threads"). A `ZsCircuit` may be moved between threads and is used by one at a time; it is `Send` and not `Sync`, with no lock. Callbacks run on whichever - thread calls `step` or `rebuild_step`, so the caller's `ctx` must be usable from every + thread calls `step` or `rebuild_step`, with that thread's thread-local state (Lisp dynamic + bindings, `errno`, `zs_last_error`), so the caller's `ctx` must be usable from every thread that may step (the wrapper `Ctx` is `unsafe impl Send` on that contract). The first-fault cell is the one lock, and it is the FFI's own. A `ZsRebuild` no longer shares a reference count with its circuit (the handles are ids), so it is a separate object, and diff --git a/context/code/zset.md b/context/code/zset.md index a050ac5d..4e3399b9 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -169,8 +169,8 @@ Decisions, where the ADR left a choice: host's. **`current()` is the whole tick.** A stream keeps the latest piece, which downstream - operators read, and the whole value of the tick, which `Stream::current` reads: pieces - within one tick are absorbed into it, so `integrate.current()` after a chunked load is the + operators read, and the whole value of the tick, which `Circuit::current(&s)` reads: pieces + within one tick are absorbed into it, so `current` of an integrate stream after a chunked load is the full integral and a continued `delay` still reads the previous tick's input. `Output` sums pieces as before. @@ -231,8 +231,8 @@ clause's wording stands; this is how the crate reads it). its map (`a_panicking_value_clone_leaves_a_memtrace_unchanged`). 5. **The edge's two copies.** An integrate stream's piece is also its output, so a second copy (the whole-tick value) only exists while a tick is split. `Slot::whole` is `None` otherwise - and `current()` reads the piece. `piece()` lends a borrow rather than a clone - (improvise-bw4p.11). An integrate stream now holds its integral twice (the node's `acc` + and `Circuit::current` reads the piece. An eval reads its input's piece in the arena by + reference rather than cloning it (improvise-bw4p.11). An integrate stream now holds its integral twice (the node's `acc` and the stream's piece), not three times; sharing those two needs shared ownership in the slot and was not done. 6. **What a transaction does not undo.** @@ -246,8 +246,8 @@ clause's wording stands; this is how the crate reads it). can overflow the pending delta before any step runs (`p1_a_chunk_whose_push_panics_in_a_txn_is_counted`). A later count makes `interrupted` true for good, finished or not (`rollback_rebuild.rs`: begun before the transaction and continued inside, finished inside, a first chunk that panicked, and a dead job revived by a later step). - - An `Output::take` made inside the transaction: a rollback subtracts the step's output - from the sink, which a `take` has emptied. Take after `commit`; `zset-ffi` refuses to take + - A `Circuit::take` of an output made inside the transaction: a rollback subtracts the step's + output from the sink, which a `take` has emptied. Take after `commit`; `zset-ffi` refuses to take while a transaction is open. 7. **Pushes.** `Circuit::push` is not logged. `rollback` puts the pending delta back to what it was at `begin`, so a push made after `begin` is dropped, and @@ -284,25 +284,31 @@ threads are out of scope, and nothing in `zset` spawns one. `Send` and `mpsc` co 1. **`Circuit` is `Send` and not `Sync`, with no lock anywhere in it.** It moves between threads, one at a time (`tests/threads.rs`: a circuit built on one thread, stepped on a new thread each tick, equals a circuit that never moved; a transaction begun on one thread is - rolled back on another). It is not `Sync` by structure (boxed nodes, no lock); no test pins - that, because a negative needs a compile-fail test and the comment cap forbids a fence. + rolled back on another). Not `Sync` is pinned by `a_circuit_is_not_sync`, an ordinary test + using the ambiguous-impl trick: it fails to build (E0283) if `Sync` is ever implemented. 2. **The `Rc` graph is an arena.** What made the circuit `!Send` was the sharing between the circuit and the handles it hands out. A circuit now owns every edge's slot, every input's pending delta and every sink's sum in one `Vec>`; a node holds typed ids, and `eval`, `begin`, `commit` and `rollback` take the arena. A downcast on every - access is the price (an `Any` type check; `tests/cost.rs` counts clones, which it leaves as they were, - and time was not measured). + access is the price. Measured on an M-series Mac under load, a release build of a 1001-node + chain (map, filter, distinct, delay) over 3000 ticks, best of 5, alternating with main: main + 127k-131k ticks/s, this 118k-122k, so 7-9% slower per tick (about 1 ns per node per tick). + The reviewer saw 10-25% on a more loaded run. Not reduced. 3. **Handles are plain ids, used through the circuit.** `Stream`, `Input` and `Output` are `Copy`, `Send` and `Sync`, and carry their circuit's token. The calls moved onto the circuit: `Circuit::current(&stream)`, `Circuit::push(&input, delta)`, `Circuit::take(&output)`. A handle from another circuit panics ("handle from another circuit"), where before it was quietly wrong. A consequence: a closure can no longer read a stream during a step, since it - has no circuit; the old `b1_...` case does not compile, so it is gone. + has no circuit; the old `b1_...` case does not compile, so it is gone. The handle types are + invariant in their element (`PhantomData T>`), as main's `Rc>` was, so a + stream of `for<'a> fn(&'a u8)` cannot be coerced into a stream of `fn(&'static u8)`, where + `current` would panic on a failed downcast. Nine call sites check the token, and + `every_call_that_takes_a_handle_refuses_a_foreign_one` has one case for each. 4. **User closures are `Send + 'static`;** keys and values are `Send`. A custom `Trace` handed to `join_indexed_in` or `aggregate_in` must be `Send`. `Piecewise` requires `Send + 'static`. 5. **The sink does not clone.** Its node reads the input edge and writes the sum through - `Arena::read_write`, two disjoint entries, so `an_eval_borrows_its_input_instead_of_cloning_it` - still holds. + `Arena::read_write`, two disjoint entries (the read one is always allocated first). + `a_sink_adds_its_input_with_one_clone_per_key` pins it over `Counted` keys. 6. **Handing a circuit to a thread is the host's business** (`std::thread` lives in host crates). Whoever owns the circuit pushes, steps and takes; a handle is a number and needs no thread. @@ -329,37 +335,58 @@ Nothing here spawns a thread; whoever owns each circuit calls `pump` and `poll`. (`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 + 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. +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. 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 + monotonic timeout, never by comparing times across machines, and it must be longer than the + round trip, or every answer is one request behind. Requests carry the receiver's stamp and + this sender does not read it. +9. **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 +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`). The property test simulates arbitrary - interleavings, losses and restarts by choosing the order of `commit`, `pump`, `poll` and step. + (`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. -10. **Not built.** A sink with a log to answer a watermark with only the moved keys (delivery's +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. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index 2fae93af..82bdfeda 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -607,14 +607,14 @@ so the file never runs; the same file's other tests move a circuit for real. A circuit moved to a new thread for each of nine ticks equals one that never moved, over a pipeline with `integrate`, `distinct`, `aggregate`, `filter` and `delay`. It would fail if a -node kept state outside the circuit's arena, which a thread-local would do. +node kept state outside the circuit (a thread-local, say), which would not move with it. ### a_transaction_begun_on_one_thread_is_rolled_back_on_another `begin` and a step on one thread, `rollback` on another: the stream, the sink and `ticks` are as at `begin`. The undo log lives in the nodes, which move with the circuit. -### a_handle_from_another_circuit_is_refused +### every_call_that_takes_a_handle_refuses_a_foreign_one Ids index an arena, so a foreign handle used to be an out-of-range or mistyped slot. It is now refused by name ("handle from another circuit"); `should_panic(expected)` pins the message, @@ -664,10 +664,71 @@ why that assertion is there. 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 +### a_late_reply_from_a_dead_incarnation_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. +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. + +### the_reviewers_late_reply_case_is_a_history + +The late-reply sequence as a fixed history through the same World. ### an_unreadable_frame_asks_for_a_correction @@ -694,3 +755,20 @@ the combined circuit's. Rival R1 fails it. Three corrections waiting, `poll(c, 2)` reads two and the next reads one, so a unit's turn is bounded by its budget. Rival: the bound removed from the loop fails the first assertion. + +Ids index an arena, so a foreign handle from a circuit of the same shape reads or wires the +other circuit's slot. Ten cases (`current`, `take`, `push`, a unary operator, both operands of a +binary one, `output`, `delay`, `integrate`, `distinct`) each need the panic text "handle from +another circuit". The first draft pinned one site in nine. Rivals, each failing it: the check +removed from `current`, `take`, `push`, `unary_with`, `binary_with` (each operand), `output`, +`delay` or `integrate`. + +### a_circuit_is_not_sync + +The ambiguous-impl trick: the test does not build (E0283) if `Circuit` implements `Sync`. +Rival: `unsafe impl Sync for Circuit {}` fails the build of `tests/threads.rs`. + +### a_sink_adds_its_input_with_one_clone_per_key + +A sink over `Counted` keys clones each key once (the insert into its sum). Rival: a sink that +clones its input piece first, as the reviewer injected, makes `N` more. diff --git a/crates/zset-ffi/include/zset_ffi.h b/crates/zset-ffi/include/zset_ffi.h index 8ae19684..37a1f9b7 100644 --- a/crates/zset-ffi/include/zset_ffi.h +++ b/crates/zset-ffi/include/zset_ffi.h @@ -20,7 +20,9 @@ * - Not thread-safe, but movable. A ZsCircuit, a ZsRebuild and a ZsZSet may each be handed to * another thread; use each from one thread at a time (a zs_rebuild_step call uses both its * job and its circuit). Callbacks run on whichever thread calls zs_circuit_step or - * zs_rebuild_step, so a ctx must be usable from every thread that may step. Streams, inputs and sinks are uint32_t ids local to their circuit; a sink id is + * zs_rebuild_step, with THAT thread's thread-local state (Lisp dynamic bindings, errno, + * zs_last_error), so a ctx must be usable from every thread that may step. + * - Streams, inputs and sinks are uint32_t ids local to their circuit; a sink id is * never a stream id (they are refused as ZS_ERR_BAD_HANDLE), but an id from another circuit * is not detected. A ZsRebuild is stepped only on the circuit that made it (else * ZS_ERR_BAD_HANDLE), even if another circuit now lives at the old address. diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs index 1ede0c08..d8e5abca 100644 --- a/crates/zset/src/circuit.rs +++ b/crates/zset/src/circuit.rs @@ -86,7 +86,7 @@ impl Slot { } /// The index of one value in a circuit's arena, typed so a read cannot name the wrong type. -struct Id(usize, PhantomData T>); +struct Id(usize, PhantomData T>); impl Clone for Id { fn clone(&self) -> Self { @@ -114,26 +114,18 @@ impl Arena { self.0[id.0].downcast_mut().expect("zset: arena type") } - /// One value to read and another to write, which are never the same entry. + /// One value to read and another, allocated after it, to write. fn read_write(&mut self, r: Id, w: Id) -> (&R, &mut W) { - assert_ne!( - r.0, w.0, - "zset: an entry cannot be read and written at once" + let (lo, hi) = self.0.split_at_mut(w.0); + assert!( + r.0 < w.0, + "zset: the read entry comes before the written one" ); let ty = "zset: arena type"; - if r.0 < w.0 { - let (lo, hi) = self.0.split_at_mut(w.0); - ( - lo[r.0].downcast_ref().expect(ty), - hi[0].downcast_mut().expect(ty), - ) - } else { - let (lo, hi) = self.0.split_at_mut(r.0); - ( - hi[0].downcast_ref().expect(ty), - lo[w.0].downcast_mut().expect(ty), - ) - } + ( + lo[r.0].downcast_ref().expect(ty), + hi[0].downcast_mut().expect(ty), + ) } } diff --git a/crates/zset/src/edge.rs b/crates/zset/src/edge.rs index c65a5ff1..2e580568 100644 --- a/crates/zset/src/edge.rs +++ b/crates/zset/src/edge.rs @@ -2,7 +2,6 @@ //! 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. @@ -26,13 +25,17 @@ pub struct Delta { pub struct Correction { pub epoch: u64, pub through: u64, + /// The token of the request this answers. + pub token: u64, pub target: ZSet, } -/// "Bring me up from here": carried with the receiver's watermark and never read by this sender. +/// "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, + pub token: u64, } /// Everything that crosses an edge, in both directions. @@ -98,6 +101,7 @@ pub struct ChannelPipe { /// Two connected ends; each direction holds `cap` items. pub fn channel_pipes(cap: usize) -> (ChannelPipe, ChannelPipe) { + 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); ( @@ -142,7 +146,8 @@ pub struct EdgeSink { integral: ZSet, /// The commands `(sent, seq]`, summed; `sent == seq` means nothing waits. waiting: Delta, - owes_correction: bool, + /// The token of the latest request read and not yet answered. + owed: Option, closed: bool, traffic: u64, } @@ -168,7 +173,7 @@ impl, P: Pipe> EdgeSink { to: seq, delta: ZSet::new(), }, - owes_correction: false, + owed: None, closed: false, traffic: 0, } @@ -188,7 +193,7 @@ impl, P: Pipe> EdgeSink { /// Whether any committed command has not been sent. pub fn backlog(&self) -> bool { - self.waiting.to > self.waiting.from || self.owes_correction + 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`. @@ -198,6 +203,8 @@ impl, P: Pipe> EdgeSink { /// One committed command changed the output by `delta` (empty is still a command). pub fn committed(&mut self, delta: ZSet) -> Stamp { + preflight(&self.integral, &delta); + preflight(&self.waiting.delta, &delta); self.seq += 1; self.integral += δ self.waiting.delta += delta; @@ -225,8 +232,8 @@ impl, P: Pipe> EdgeSink { match self.pipe.try_recv() { Ok(wire) => { self.traffic += 1; - if let Ok(Frame::Request(_)) = self.codec.decode(wire) { - self.owes_correction = true; + if let Ok(Frame::Request(r)) = self.codec.decode(wire) { + self.owed = Some(r.token); } } Err(Recv::Empty) => break, @@ -236,10 +243,11 @@ impl, P: Pipe> EdgeSink { if self.closed { return Flow::Closed; } - let frame = if self.owes_correction { + 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 { @@ -250,7 +258,7 @@ impl, P: Pipe> EdgeSink { match self.pipe.try_send(self.codec.encode(&frame)) { Ok(()) => { self.traffic += 1; - self.owes_correction = false; + self.owed = None; self.waiting = Delta { epoch: self.epoch, from: self.seq, @@ -268,8 +276,12 @@ impl, P: Pipe> EdgeSink { } } -/// How many incarnations a receiver remembers having left. -const LEFT: usize = 16; +/// Panics, before anything changes, if adding `d` to `z` would overflow a weight. +fn preflight(z: &ZSet, d: &ZSet) { + 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. @@ -280,9 +292,14 @@ pub struct EdgeSource { joined: Option, watermark: u64, held: ZSet, - left: VecDeque, 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, closed: bool, traffic: u64, } @@ -297,9 +314,12 @@ impl, P: Pipe> EdgeSou joined: None, watermark: 0, held: ZSet::new(), - left: VecDeque::new(), stale: true, asked: false, + token: 0, + marks: 0, + ask_marks: 0, + noted: None, closed: false, traffic: 0, } @@ -325,7 +345,16 @@ impl, P: Pipe> EdgeSou /// 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) + 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 { + &self.held } /// The host's timeout or reconnect: ask again from the watermark held. @@ -334,9 +363,13 @@ impl, P: Pipe> EdgeSou 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. + /// 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() { @@ -347,7 +380,7 @@ impl, P: Pipe> EdgeSou 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(_) => self.mark(), } } Err(Recv::Empty) => break, @@ -358,8 +391,20 @@ impl, P: Pipe> EdgeSou read } - fn has_left(&self, epoch: u64) -> bool { - self.left.contains(&epoch) + /// 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) { @@ -368,54 +413,56 @@ impl, P: Pipe> EdgeSou } fn on_delta(&mut self, circuit: &mut Circuit, d: Delta) { - if self.has_left(d.epoch) { + if self.joined != Some(d.epoch) { + return self.foreign(d.epoch); + } + if d.to <= self.watermark { return; } - let known = self.joined == Some(d.epoch); - if known && d.to <= self.watermark { - return; - } - if known && d.from == self.watermark { + if d.from == self.watermark { self.watermark = d.to; self.apply(circuit, d.delta); } else { - self.stale = true; + self.mark(); } } fn on_correction(&mut self, circuit: &mut Circuit, c: Correction) { - if self.has_left(c.epoch) { - return; - } - self.asked = false; + let answer = self.asked && c.token == self.token; let known = self.joined == Some(c.epoch); - if known && c.through <= self.watermark { - self.stale &= c.through != self.watermark; - return; + if !known && !answer { + return self.foreign(c.epoch); } - if !known && let Some(old) = self.joined.replace(c.epoch) { - self.left.push_back(old); - if self.left.len() > LEFT { - self.left.pop_front(); - } + 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; } - 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() })); + 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.traffic += 1; + self.token = token; + self.ask_marks = self.marks; } Err(Refused::Full(_)) => {} Err(Refused::Closed) => self.closed = true, diff --git a/crates/zset/tests/cost.rs b/crates/zset/tests/cost.rs index 17e0e630..0af9ad1f 100644 --- a/crates/zset/tests/cost.rs +++ b/crates/zset/tests/cost.rs @@ -82,6 +82,22 @@ fn an_integrate_stream_holds_its_integral_twice_when_the_tick_is_whole() { assert_eq!(live(), base, "[leak] every key is dropped with the circuit"); } +#[test] +fn a_sink_adds_its_input_with_one_clone_per_key() { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let sink = c.output(&s); + c.push(&input, keys(N)); + let before = clones(); + c.step(); + assert_eq!( + clones() - before, + N as usize, + "[sink] one clone per key it sums" + ); + assert_eq!(c.take(&sink).len(), N as usize); +} + type E = (Counted, u8); #[derive(Clone, Copy)] diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs index 5a3bafb9..62e4b16d 100644 --- a/crates/zset/tests/edge.rs +++ b/crates/zset/tests/edge.rs @@ -1,13 +1,16 @@ //! 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, CodecError, Correction, Delta, Direct, EdgeCodec, EdgeSink, EdgeSource, - Flow, Frame, Input, Output, Pipe, Recv, Refused, Request, Stamp, ZSet, channel_pipes, + ChannelPipe, Circuit, Correction, Delta, Direct, EdgeCodec, EdgeSink, EdgeSource, Flow, Frame, + Input, Output, Pipe, Recv, Refused, Stamp, ZSet, channel_pipes, }; type Z = ZSet; @@ -98,80 +101,6 @@ fn fault( ) } -/// 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>>; @@ -504,11 +433,8 @@ fn a_duplicate_delta_is_applied_once() { 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(), - }); + src.poll(&mut c, 10); + let join = answer(&mut b, 1, 0, Z::new()); let d = Frame::Delta(Delta { epoch: 1, from: 0, @@ -543,42 +469,300 @@ fn a_restarted_sender_is_corrected_not_trusted() { 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>, epoch: u64, through: u64, target: Z) -> Frame { + 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>, epoch: u64, through: u64, target: Z) { + let f = answer(b, epoch, through, target); + b.try_send(f).unwrap(); +} + #[test] -fn a_late_frame_from_an_epoch_the_receiver_left_changes_nothing() { +fn a_late_reply_from_a_dead_incarnation_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(); + // 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: 1, - from: 9, - to: 10, - delta: z(&[(8, 1)]), + epoch: 1001, + from: 1, + to: 2, + delta: z(&[(6, 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" + 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::>(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::>(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::>(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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 traffic_never_goes_down_across_a_restart_of_either_end() { + let (mut sink, mut src, mut c, _, _) = mirror(8); + sink.committed(z(&[(1, 1)])); + drive(&mut sink, &mut src, &mut c); + let (s, r) = (sink.traffic(), src.traffic()); + assert!(s > 0 && r > 0, "[setup] frames moved both ways"); + let sink = sink.restart(2, 0, Z::new()); + let mut c2 = Circuit::new(); + let (i2, _) = c2.input::(); + let src = src.restart(i2); + assert_eq!((sink.traffic(), src.traffic()), (s, r), "[restart] counters kept"); +} + +#[test] +fn a_new_epoch_frame_re_asks_once_after_a_lost_request() { + let (a, mut b) = channel_pipes::>(64); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::>(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::>(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::>(8); + let mut c = Circuit::new(); + let (i, s) = c.input::(); + 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::(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" ); - assert_eq!(src.stamp(), Some(Stamp { epoch: 2, seq: 3 })); - assert!(!src.stale(), "[left] and they do not ask for anything"); } #[test] @@ -594,7 +778,7 @@ fn an_unreadable_frame_asks_for_a_correction() { src.stale(), "[initial] a new receiver asks for a correction" ); - b.try_send("C 1 1 1:1".to_string()).unwrap(); + b.try_send("C 1 1 1 1:1".to_string()).unwrap(); src.poll(&mut c, 10); assert!(!src.stale()); b.try_send("garbage".to_string()).unwrap(); @@ -608,27 +792,26 @@ fn an_unreadable_frame_asks_for_a_correction() { 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(); - 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. + // The retry is a different command and commits. p.begin(); - p.push(&p_in, z(&[(1, 1)])); + 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] diff --git a/crates/zset/tests/edge_net.rs b/crates/zset/tests/edge_net.rs new file mode 100644 index 00000000..35ec8ef5 --- /dev/null +++ b/crates/zset/tests/edge_net.rs @@ -0,0 +1,333 @@ +//! 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; + +/// 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, + ready: VecDeque, +} + +#[derive(Default)] +struct Net { + lanes: [Lane; 2], + cap: usize, +} + +type Shared = Arc>; + +/// One end: sends into lane `out`, receives from the other one. +struct End { + net: Shared, + out: usize, +} + +impl Pipe for End { + fn try_send(&mut self, wire: String) -> Result<(), Refused> { + 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 { + 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) +} + +fn z(entries: &[(u8, i64)]) -> Z { + entries.iter().copied().collect() +} + +struct World { + net: Shared, + sink: EdgeSink, + src: EdgeSource, + cons: Circuit, + out: Output, + 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::(); + 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()); + } + + fn restart_source(&mut self) { + let mut cons = Circuit::new(); + let (i, s) = cons.input::(); + 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, +} + +fn act() -> impl Strategy { + 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), + ] +} + +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(), + } + 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. + 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::Deliver(0, 0), + Act::Poll(4), + ]; + run(2, &acts); + let _ = z(&[]); +} diff --git a/crates/zset/tests/support/mod.rs b/crates/zset/tests/support/mod.rs new file mode 100644 index 00000000..6d110a08 --- /dev/null +++ b/crates/zset/tests/support/mod.rs @@ -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; + +/// 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::>() + .join(",") +} + +pub 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, 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, 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(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()), + } + } +} diff --git a/crates/zset/tests/threads.rs b/crates/zset/tests/threads.rs index f165e4b5..a90bace4 100644 --- a/crates/zset/tests/threads.rs +++ b/crates/zset/tests/threads.rs @@ -95,10 +95,77 @@ fn a_transaction_begun_on_one_thread_is_rolled_back_on_another() { assert_eq!(c.ticks(), 1); } -#[test] -#[should_panic(expected = "handle from another circuit")] -fn a_handle_from_another_circuit_is_refused() { - let (mut a, mut b) = (Circuit::new(), Circuit::new()); - let (input, _) = a.input::(); - b.push(&input, Z::singleton(1, 1)); +trait AmbiguousIfSync { + fn some_item() {} +} +impl AmbiguousIfSync<()> for T {} +struct IsSync; +impl AmbiguousIfSync for T {} + +// Fails to build (E0283) if `Circuit` becomes `Sync`. +#[test] +fn a_circuit_is_not_sync() { + let _ = >::some_item; +} + +type Site = fn(&mut Circuit, &Input, &Stream>, &Output); + +/// Two circuits of one shape, so a foreign handle indexes a live slot of the same type. +fn twin() -> (Circuit, Input, Stream>, Output) { + let mut c = Circuit::new(); + let (i, s) = c.input::(); + let o = c.output(&s); + (c, i, s, o) +} + +fn panic_text(f: impl FnOnce() + std::panic::UnwindSafe) -> Option { + let e = std::panic::catch_unwind(f).err()?; + e.downcast_ref::() + .cloned() + .or_else(|| e.downcast_ref::<&str>().map(|s| s.to_string())) +} + +#[test] +fn every_call_that_takes_a_handle_refuses_a_foreign_one() { + let sites: [(&str, Site); 10] = [ + ("current", |c, _, s, _| { + let _ = c.current(s); + }), + ("take", |c, _, _, o| drop(c.take(o))), + ("push", |c, i, _, _| c.push(i, Z::new())), + ("unary (map)", |c, _, s, _| { + let _ = c.map(s, |k| *k); + }), + ("binary, first operand", |c, _, s, _| { + let (_, own) = c.input::(); + let _ = c.plus(s, &own); + }), + ("binary, second operand", |c, _, s, _| { + let (_, own) = c.input::(); + let _ = c.plus(&own, s); + }), + ("output", |c, _, s, _| { + let _ = c.output(s); + }), + ("delay", |c, _, s, _| { + let _ = c.delay(s); + }), + ("integrate", |c, _, s, _| { + let _ = c.integrate(s); + }), + ("distinct", |c, _, s, _| { + let _ = c.distinct(s); + }), + ]; + for (name, site) in sites { + let (mut mine, ..) = twin(); + let (_, fi, fs, fo) = twin(); + let text = panic_text(std::panic::AssertUnwindSafe(|| { + site(&mut mine, &fi, &fs, &fo) + })); + assert!( + text.is_some_and(|t| t.contains("handle from another circuit")), + "[foreign] {name} accepted a handle from another circuit" + ); + } }