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

test(zset): the edge over a hostile network, and the cases the review found

Browse Source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze
This commit is contained in:
Claude (dbsp-threads)
2026-10-10 21:33:21 -07:00
committed by Edward Langley
parent 5c577cddb1
commit 5344d287a9
9 changed files with 671 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
context/code/delivery.md
+1 -1
View File
@@ -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.
context/code/zset.md
+45 -24
View File
@@ -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<K>` turns a `Frame<K>` 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<K>` turns a `Frame<K>` into `Wire` and back, with
`CodecError` for what does not decode; `Direct` is the identity. `improvise-protocol` is to
implement it for its own key types in the lossless text of ADR 005, and `zset` stays free of
serde. The tests run every history through a hand-written text codec
(`the_same_histories_through_a_text_codec`).
9. **Threads.** `EdgeSink` and `EdgeSource` are `Send` when the pipe and codec are. The real
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.
context/test-evidence/zset.md
+64 -3
View File
@@ -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
crates/zset/src/edge.rs
+4 -6
View File
@@ -347,9 +347,8 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> 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<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> 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) {
crates/zset/tests/edge.proptest-regressions
+2
View File
@@ -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
crates/zset/tests/edge.rs
+131 -99
View File
@@ -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<u8>;
@@ -98,84 +101,6 @@ fn fault<W: Send>(
)
}
/// A text wire form, to run the same histories through an encoding.
#[derive(Clone, Copy)]
struct Text;
fn zs(z: &Z) -> String {
z.iter()
.map(|(k, w)| format!("{k}:{w}"))
.collect::<Vec<_>>()
.join(",")
}
fn unzs(s: &str) -> Result<Z, CodecError> {
let bad = || CodecError(format!("bad entries {s:?}"));
s.split(',')
.filter(|p| !p.is_empty())
.map(|p| {
let (k, w) = p.split_once(':').ok_or_else(bad)?;
Ok((
k.parse::<u8>().map_err(|_| bad())?,
w.parse::<i64>().map_err(|_| bad())?,
))
})
.collect()
}
impl EdgeCodec<u8> for Text {
type Wire = String;
fn encode(&self, f: &Frame<u8>) -> String {
match f {
Frame::Delta(d) => format!("D {} {} {} {}", d.epoch, d.from, d.to, zs(&d.delta)),
Frame::Correction(c) => {
format!("C {} {} {} {}", c.epoch, c.through, c.token, zs(&c.target))
}
Frame::Request(r) => match r.have {
Some(s) => format!("R {} {} {}", r.token, s.epoch, s.seq),
None => format!("R {}", r.token),
},
}
}
fn decode(&self, w: String) -> Result<Frame<u8>, CodecError> {
let bad = || CodecError(format!("bad frame {w:?}"));
let mut it = w.splitn(5, ' ');
let num = |s: Option<&str>| s.and_then(|s| s.parse::<u64>().ok()).ok_or_else(bad);
match it.next() {
Some("D") => Ok(Frame::Delta(Delta {
epoch: num(it.next())?,
from: num(it.next())?,
to: num(it.next())?,
delta: unzs(it.next().ok_or_else(bad)?)?,
})),
Some("C") => {
let mut it = w.splitn(5, ' ');
it.next();
Ok(Frame::Correction(Correction {
epoch: num(it.next())?,
through: num(it.next())?,
token: num(it.next())?,
target: unzs(it.next().ok_or_else(bad)?)?,
}))
}
Some("R") => {
let token = num(it.next())?;
let have = match it.next() {
None => None,
e => Some(Stamp {
epoch: num(e)?,
seq: num(it.next())?,
}),
};
Ok(Frame::Request(Request { have, token }))
}
_ => Err(bad()),
}
}
}
type Sink<C> = EdgeSink<u8, C, FaultPipe<<C as EdgeCodec<u8>>::Wire>>;
type Source<C> = EdgeSource<u8, C, FaultPipe<<C as EdgeCodec<u8>>::Wire>>;
@@ -560,6 +485,11 @@ fn answer(b: &mut ChannelPipe<Frame<u8>>, epoch: u64, through: u64, target: Z) -
})
}
fn reply(b: &mut ChannelPipe<Frame<u8>>, epoch: u64, through: u64, target: Z) {
let f = answer(b, epoch, through, target);
b.try_send(f).unwrap();
}
#[test]
fn a_late_reply_from_a_dead_incarnation_changes_nothing() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
@@ -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::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 10);
let late = answer(&mut b, 100, 1, z(&[(9, 1)]));
src.connection_lost();
src.poll(&mut c, 10);
b.try_send(late).unwrap();
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
None,
"[token] the reply to the first request is not the answer"
);
reply(&mut b, 1001, 1, z(&[(5, 1)]));
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp {
epoch: 1001,
seq: 1
})
);
}
#[test]
fn an_answer_older_than_the_watermark_does_not_clear_stale() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 10);
reply(&mut b, 1, 3, Z::new());
src.poll(&mut c, 10);
src.connection_lost();
src.poll(&mut c, 10);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp { epoch: 1, seq: 3 }),
"[older] never moves back"
);
assert!(
src.stale(),
"[older] and it does not say the receiver is current"
);
}
#[test]
fn a_new_epoch_frame_re_asks_once_after_a_lost_request() {
let (a, mut b) = channel_pipes::<Frame<u8>>(64);
@@ -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]
crates/zset/tests/edge_net.proptest-regressions
+8
View File
@@ -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
crates/zset/tests/edge_net.rs
+333
View File
@@ -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<u8>;
/// One direction of the network: frames in flight (which the test may reorder, drop, duplicate
/// or garble) and frames delivered (which `try_recv` takes in order).
#[derive(Default)]
struct Lane {
flight: Vec<String>,
ready: VecDeque<String>,
}
#[derive(Default)]
struct Net {
lanes: [Lane; 2],
cap: usize,
}
type Shared = Arc<Mutex<Net>>;
/// One end: sends into lane `out`, receives from the other one.
struct End {
net: Shared,
out: usize,
}
impl Pipe<String> for End {
fn try_send(&mut self, wire: String) -> Result<(), Refused<String>> {
let mut n = self.net.lock().unwrap();
let cap = n.cap;
let lane = &mut n.lanes[self.out];
if lane.flight.len() + lane.ready.len() >= cap {
return Err(Refused::Full(wire));
}
lane.flight.push(wire);
Ok(())
}
fn try_recv(&mut self) -> Result<String, Recv> {
let mut n = self.net.lock().unwrap();
n.lanes[1 - self.out].ready.pop_front().ok_or(Recv::Empty)
}
}
fn net(cap: usize) -> (Shared, End, End) {
let shared: Shared = Arc::new(Mutex::new(Net {
cap,
..Net::default()
}));
let sink_end = End {
net: shared.clone(),
out: 0,
};
let src_end = End {
net: shared.clone(),
out: 1,
};
(shared, sink_end, src_end)
}
fn z(entries: &[(u8, i64)]) -> Z {
entries.iter().copied().collect()
}
struct World {
net: Shared,
sink: EdgeSink<u8, Text, End>,
src: EdgeSource<u8, Text, End>,
cons: Circuit,
out: Output<u8>,
got: Z,
total: Z,
epoch: u64,
seq: u64,
history: HashMap<(u64, u64), Z>,
joined_max: u64,
}
impl World {
fn new(cap: usize) -> Self {
let (shared, a, b) = net(cap);
let mut cons = Circuit::new();
let (i, s) = cons.input::<u8>();
let out = cons.output(&s);
let mut history = HashMap::new();
history.insert((1, 0), Z::new());
World {
net: shared,
sink: EdgeSink::new(Text, a, 1),
src: EdgeSource::new(Text, b, i),
cons,
out,
got: Z::new(),
total: Z::new(),
epoch: 1,
seq: 0,
history,
joined_max: 0,
}
}
fn commit(&mut self, d: Z) {
self.total += &d;
self.seq += 1;
self.sink.committed(d);
self.history
.insert((self.epoch, self.seq), self.total.clone());
}
fn restart_sink(&mut self, seq: u64) {
self.epoch += 1;
self.seq = seq;
let old = std::mem::replace(
&mut self.sink,
EdgeSink::new(
Text,
End {
net: self.net.clone(),
out: 0,
},
0,
),
);
self.sink = old.restart(self.epoch, seq, self.total.clone());
self.history.insert((self.epoch, seq), self.total.clone());
}
fn restart_source(&mut self) {
let mut cons = Circuit::new();
let (i, s) = cons.input::<u8>();
self.out = cons.output(&s);
let old = std::mem::replace(
&mut self.src,
EdgeSource::new(
Text,
End {
net: self.net.clone(),
out: 1,
},
i,
),
);
self.src = old.restart(i);
self.cons = cons;
self.got = Z::new();
self.joined_max = 0;
}
fn step(&mut self) {
self.cons.step();
self.got += self.cons.take(&self.out);
}
fn lane(&mut self, dir: usize, idx: usize, f: impl FnOnce(&mut Lane, usize)) {
let mut n = self.net.lock().unwrap();
let lane = &mut n.lanes[dir];
if !lane.flight.is_empty() {
let at = idx % lane.flight.len();
f(lane, at);
}
}
/// The two safety rules, after every action.
fn check(&mut self, when: &str) {
if let Some(Stamp { epoch, seq }) = self.src.stamp() {
assert!(
epoch >= self.joined_max,
"[older] {when}: the receiver went back from epoch {} to {epoch}",
self.joined_max
);
self.joined_max = epoch;
let want = self.history.get(&(epoch, seq));
assert_eq!(
Some(self.src.held()),
want,
"[claim] {when}: held is not the sender's integral at ({epoch}, {seq})"
);
}
}
fn settle(&mut self) {
for _ in 0..400 {
let silent = {
let mut n = self.net.lock().unwrap();
let silent = n
.lanes
.iter()
.all(|l| l.flight.is_empty() && l.ready.is_empty());
for lane in &mut n.lanes {
let flight = std::mem::take(&mut lane.flight);
lane.ready.extend(flight);
}
silent
};
self.sink.pump();
let read = self.src.poll(&mut self.cons, 8);
self.step();
self.check("settle");
let quiet = {
let n = self.net.lock().unwrap();
n.lanes
.iter()
.all(|l| l.flight.is_empty() && l.ready.is_empty())
};
if quiet
&& !self.sink.backlog()
&& !self.src.stale()
&& self.src.stamp() == Some(self.sink.stamp())
{
break;
}
if silent && read == 0 {
// Nothing was in flight: the reply is lost, not late. Only now does the timeout fire.
self.src.connection_lost();
}
}
assert_eq!(
self.src.stamp(),
Some(self.sink.stamp()),
"[settle] never caught up"
);
self.step();
assert_eq!(self.got, self.total, "[settle] the consumer's integral");
}
}
#[derive(Clone, Debug)]
enum Act {
Commit(Vec<(u8, i64)>),
Pump,
Poll(usize),
Step,
Timeout,
Deliver(usize, usize),
Drop(usize, usize),
Dup(usize, usize),
Garble(usize, usize),
RestartSink(u64),
RestartSource,
}
fn act() -> impl Strategy<Value = Act> {
let lane = 0usize..2;
let at = 0usize..8;
prop_oneof![
5 => prop::collection::vec((0u8..10, -2i64..=2), 0..3).prop_map(Act::Commit),
4 => Just(Act::Pump),
4 => (1usize..4).prop_map(Act::Poll),
2 => Just(Act::Step),
2 => Just(Act::Timeout),
8 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Deliver(l, i)),
2 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Drop(l, i)),
2 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Dup(l, i)),
1 => (lane, at).prop_map(|(l, i)| Act::Garble(l, i)),
1 => (0u64..8).prop_map(Act::RestartSink),
1 => Just(Act::RestartSource),
]
}
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(&[]);
}
crates/zset/tests/support/mod.rs
+83
View File
@@ -0,0 +1,83 @@
//! A text wire form of `u8`-keyed frames, shared by the edge tests.
use zset::{CodecError, Correction, Delta, EdgeCodec, Frame, Request, Stamp, ZSet};
type Z = ZSet<u8>;
/// A text wire form, to run the same histories through an encoding.
#[derive(Clone, Copy)]
pub struct Text;
pub fn zs(z: &Z) -> String {
z.iter()
.map(|(k, w)| format!("{k}:{w}"))
.collect::<Vec<_>>()
.join(",")
}
pub fn unzs(s: &str) -> Result<Z, CodecError> {
let bad = || CodecError(format!("bad entries {s:?}"));
s.split(',')
.filter(|p| !p.is_empty())
.map(|p| {
let (k, w) = p.split_once(':').ok_or_else(bad)?;
Ok((
k.parse::<u8>().map_err(|_| bad())?,
w.parse::<i64>().map_err(|_| bad())?,
))
})
.collect()
}
impl EdgeCodec<u8> for Text {
type Wire = String;
fn encode(&self, f: &Frame<u8>) -> String {
match f {
Frame::Delta(d) => format!("D {} {} {} {}", d.epoch, d.from, d.to, zs(&d.delta)),
Frame::Correction(c) => {
format!("C {} {} {} {}", c.epoch, c.through, c.token, zs(&c.target))
}
Frame::Request(r) => match r.have {
Some(s) => format!("R {} {} {}", r.token, s.epoch, s.seq),
None => format!("R {}", r.token),
},
}
}
fn decode(&self, w: String) -> Result<Frame<u8>, CodecError> {
let bad = || CodecError(format!("bad frame {w:?}"));
let mut it = w.splitn(5, ' ');
let num = |s: Option<&str>| s.and_then(|s| s.parse::<u64>().ok()).ok_or_else(bad);
match it.next() {
Some("D") => Ok(Frame::Delta(Delta {
epoch: num(it.next())?,
from: num(it.next())?,
to: num(it.next())?,
delta: unzs(it.next().ok_or_else(bad)?)?,
})),
Some("C") => {
let mut it = w.splitn(5, ' ');
it.next();
Ok(Frame::Correction(Correction {
epoch: num(it.next())?,
through: num(it.next())?,
token: num(it.next())?,
target: unzs(it.next().ok_or_else(bad)?)?,
}))
}
Some("R") => {
let token = num(it.next())?;
let have = match it.next() {
None => None,
e => Some(Stamp {
epoch: num(e)?,
seq: num(it.next())?,
}),
};
Ok(Frame::Request(Request { have, token }))
}
_ => Err(bad()),
}
}
}