From 5344d287a97a5cd18fc0f327439ebfb1777b1b64 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-threads)" Date: Sat, 10 Oct 2026 21:33:21 -0700 Subject: [PATCH] test(zset): the edge over a hostile network, and the cases the review found Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/delivery.md | 2 +- context/code/zset.md | 69 ++-- context/test-evidence/zset.md | 67 +++- crates/zset/src/edge.rs | 10 +- crates/zset/tests/edge.proptest-regressions | 2 + crates/zset/tests/edge.rs | 230 ++++++------ .../zset/tests/edge_net.proptest-regressions | 8 + crates/zset/tests/edge_net.rs | 333 ++++++++++++++++++ crates/zset/tests/support/mod.rs | 83 +++++ 9 files changed, 671 insertions(+), 133 deletions(-) create mode 100644 crates/zset/tests/edge_net.proptest-regressions create mode 100644 crates/zset/tests/edge_net.rs create mode 100644 crates/zset/tests/support/mod.rs 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.md b/context/code/zset.md index bd5fe56b..363a200c 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -335,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 b255abf9..59d68e07 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -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 diff --git a/crates/zset/src/edge.rs b/crates/zset/src/edge.rs index fe6816b1..72f35872 100644 --- a/crates/zset/src/edge.rs +++ b/crates/zset/src/edge.rs @@ -347,9 +347,8 @@ 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. Panics inside a - /// transaction: a rollback would drop what was pushed, and the stamps would not know. + /// 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(), @@ -381,9 +380,8 @@ impl, P: Pipe> EdgeSou self.marks += 1; } - /// A frame from an incarnation this receiver has not joined. It never moves the watermark, - /// and the first one seen for an epoch makes the receiver ask again (the request may have - /// been lost with the sender that read it). + /// 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) { diff --git a/crates/zset/tests/edge.proptest-regressions b/crates/zset/tests/edge.proptest-regressions index 6554d569..38dcb043 100644 --- a/crates/zset/tests/edge.proptest-regressions +++ b/crates/zset/tests/edge.proptest-regressions @@ -10,3 +10,5 @@ cc d717a61b42263ed743ce993d0dc7755a3effeddf3aeac9aede363a8c1e8af465 # shrinks to 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 +cc 2b94ff9ad9fcd2263bd37e6856318facb5a0033a8529da8848baff4cea972601 # shrinks to acts = [Commit([(0, 1)]), Poll(1), Pump, RestartSource, Commit([])], cap = 2 +cc 7b098037af1b4ccf063e1d9e9574f339f7fa7b99e7a770450d1afbed9617916c # shrinks to acts = [StepConsumer, Commit([(1, 1)]), RestartSink, Commit([])], cap = 1 diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs index ff01a53d..46995044 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,84 +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, 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()), - } - } -} - type Sink = EdgeSink>::Wire>>; type Source = EdgeSource>::Wire>>; @@ -560,6 +485,11 @@ fn answer(b: &mut ChannelPipe>, epoch: u64, through: u64, target: Z) - }) } +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_reply_from_a_dead_incarnation_changes_nothing() { let (a, mut b) = channel_pipes::>(8); @@ -572,9 +502,15 @@ fn a_late_reply_from_a_dead_incarnation_changes_nothing() { let late = answer(&mut b, 100, 1, z(&[(9, 1)])); src.connection_lost(); src.poll(&mut c, 10); - b.try_send(answer(&mut b, 1001, 1, z(&[(5, 1)]))).unwrap(); + reply(&mut b, 1001, 1, z(&[(5, 1)])); src.poll(&mut c, 10); - assert_eq!(src.stamp(), Some(Stamp { epoch: 1001, seq: 1 })); + assert_eq!( + src.stamp(), + Some(Stamp { + epoch: 1001, + seq: 1 + }) + ); b.try_send(late).unwrap(); b.try_send(Frame::Delta(Delta { epoch: 1001, @@ -587,10 +523,17 @@ fn a_late_reply_from_a_dead_incarnation_changes_nothing() { c.step(); assert_eq!( src.stamp(), - Some(Stamp { epoch: 1001, seq: 2 }), + 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"); + assert_eq!( + c.take(&o), + z(&[(5, 1), (6, 1)]), + "[late] and 1001 is not shut out" + ); } #[test] @@ -601,7 +544,7 @@ fn an_unsolicited_correction_from_another_incarnation_only_marks_stale() { let _o = c.output(&s); let mut src = EdgeSource::new(Direct, a, i); src.poll(&mut c, 10); - b.try_send(answer(&mut b, 1, 1, Z::new())).unwrap(); + reply(&mut b, 1, 1, Z::new()); src.poll(&mut c, 10); assert!(!src.stale()); let wrong = Frame::Correction(Correction { @@ -612,10 +555,68 @@ fn an_unsolicited_correction_from_another_incarnation_only_marks_stale() { }); b.try_send(wrong).unwrap(); src.poll(&mut c, 10); - assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 1 }), "[foreign] not moved"); + 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 a_new_epoch_frame_re_asks_once_after_a_lost_request() { let (a, mut b) = channel_pipes::>(64); @@ -624,7 +625,10 @@ fn a_new_epoch_frame_re_asks_once_after_a_lost_request() { 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"); + 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 { @@ -640,7 +644,10 @@ fn a_new_epoch_frame_re_asks_once_after_a_lost_request() { asks += 1; } } - assert_eq!(asks, 1, "[reask] once for the new epoch, not once per frame"); + assert_eq!( + asks, 1, + "[reask] once for the new epoch, not once per frame" + ); } #[test] @@ -651,13 +658,16 @@ fn an_answer_at_the_watermark_says_nothing_newer_and_clears_stale() { let _o = c.output(&s); let mut src = EdgeSource::new(Direct, a, i); src.poll(&mut c, 8); - b.try_send(answer(&mut b, 1, 1, Z::new())).unwrap(); + reply(&mut b, 1, 1, Z::new()); src.poll(&mut c, 8); src.connection_lost(); src.poll(&mut c, 8); - b.try_send(answer(&mut b, 1, 1, Z::new())).unwrap(); + reply(&mut b, 1, 1, Z::new()); src.poll(&mut c, 8); - assert!(!src.stale(), "[dup-answer] through == watermark clears stale"); + assert!( + !src.stale(), + "[dup-answer] through == watermark clears stale" + ); } #[test] @@ -668,7 +678,7 @@ fn trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale() { let _o = c.output(&s); let mut src = EdgeSource::new(Direct, a, i); src.poll(&mut c, 8); - b.try_send(answer(&mut b, 1, 1, Z::new())).unwrap(); + reply(&mut b, 1, 1, Z::new()); src.poll(&mut c, 8); src.connection_lost(); src.poll(&mut c, 8); @@ -694,17 +704,24 @@ fn a_poll_inside_a_transaction_is_refused() { let o = c.output(&s); let mut src = EdgeSource::new(Direct, a, i); src.poll(&mut c, 8); - b.try_send(answer(&mut b, 1, 1, z(&[(1, 1)]))).unwrap(); + 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"); + 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"); + assert!( + r.is_err(), + "[txn] a poll inside a transaction would be dropped by rollback" + ); } #[test] @@ -721,9 +738,17 @@ fn a_committed_that_overflows_leaves_the_sink_as_it_was() { 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"); + 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.held(), + &z(&[(1, i64::MAX)]), + "[overflow] and the key 2 did not slip in" + ); } #[test] @@ -763,9 +788,16 @@ fn a_rolled_back_command_does_not_cross() { 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 }); + 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"); + assert_eq!( + c.take(&out), + z(&[(2, 1)]), + "[retry] only the committed command crossed" + ); } #[test] diff --git a/crates/zset/tests/edge_net.proptest-regressions b/crates/zset/tests/edge_net.proptest-regressions new file mode 100644 index 00000000..6015424e --- /dev/null +++ b/crates/zset/tests/edge_net.proptest-regressions @@ -0,0 +1,8 @@ +# 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 9a5d5413ebacc680c1744f086331a406815ef5a10dcb70b95a157c4098474d18 # shrinks to acts = [], cap = 2 +cc 0ee260c6da9e8f1d424f277d372cb485a2028f1fea4c77402a331ac7c2c978e6 # shrinks to acts = [Commit([(0, 1)]), Commit([]), Commit([]), Pump, Commit([]), Drop(0, 0)], cap = 1 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()), + } + } +}