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

fix(zset): edge receivers join by token-matched answers, refuse poll in a transaction, and re-ask on a new epoch

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:20:01 -07:00
committed by Edward Langley
parent c2598f0c60
commit 491f47d33c
2 changed files with 280 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/zset/src/edge.rs
+96 -45
View File
@@ -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<K> {
pub struct Correction<K> {
pub epoch: u64,
pub through: u64,
/// The token of the request this answers.
pub token: u64,
pub target: ZSet<K>,
}
/// "Bring me up from here": 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<Stamp>,
pub token: u64,
}
/// Everything that crosses an edge, in both directions.
@@ -98,6 +101,7 @@ pub struct ChannelPipe<W> {
/// Two connected ends; each direction holds `cap` items.
pub fn channel_pipes<W: Send>(cap: usize) -> (ChannelPipe<W>, ChannelPipe<W>) {
assert!(cap > 0, "zset: a channel pipe holds at least one item");
let (atx, brx) = mpsc::sync_channel(cap);
let (btx, arx) = mpsc::sync_channel(cap);
(
@@ -142,7 +146,8 @@ pub struct EdgeSink<K, C, P> {
integral: ZSet<K>,
/// The commands `(sent, seq]`, summed; `sent == seq` means nothing waits.
waiting: Delta<K>,
owes_correction: bool,
/// The token of the latest request read and not yet answered.
owed: Option<u64>,
closed: bool,
}
@@ -167,7 +172,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
to: seq,
delta: ZSet::new(),
},
owes_correction: false,
owed: None,
closed: false,
}
}
@@ -181,7 +186,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
/// 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`.
@@ -191,6 +196,8 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
/// One committed command changed the output by `delta` (empty is still a command).
pub fn committed(&mut self, delta: ZSet<K>) -> Stamp {
preflight(&self.integral, &delta);
preflight(&self.waiting.delta, &delta);
self.seq += 1;
self.integral += &delta;
self.waiting.delta += delta;
@@ -217,8 +224,8 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
while !self.closed {
match self.pipe.try_recv() {
Ok(wire) => {
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,
@@ -228,10 +235,11 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
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 {
@@ -241,7 +249,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
};
match self.pipe.try_send(self.codec.encode(&frame)) {
Ok(()) => {
self.owes_correction = false;
self.owed = None;
self.waiting = Delta {
epoch: self.epoch,
from: self.seq,
@@ -259,8 +267,12 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
}
}
/// 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<K: Ord>(z: &ZSet<K>, d: &ZSet<K>) {
for (k, w) in d.iter() {
crate::plus(z.weight(k), w);
}
}
/// The receiving end, beside the consumer circuit: it applies deltas to the circuit's input in
/// order, and asks for a correction when the stamps show something missing.
@@ -271,9 +283,14 @@ pub struct EdgeSource<K, C, P> {
joined: Option<u64>,
watermark: u64,
held: ZSet<K>,
left: VecDeque<u64>,
stale: bool,
/// A request is out. `token` is its token, and `marks` counted the troubles seen so far.
asked: bool,
token: u64,
marks: u64,
ask_marks: u64,
/// The foreign incarnation a re-ask was already made for.
noted: Option<u64>,
closed: bool,
}
@@ -287,9 +304,12 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> 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,
}
}
@@ -309,7 +329,16 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
/// The same pipe and codec, after the consumer lost its state: it holds nothing again.
pub fn restart(self, input: Input<K>) -> 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<K> {
&self.held
}
/// The host's timeout or reconnect: ask again from the watermark held.
@@ -319,8 +348,13 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
}
/// 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.
/// 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.
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() {
@@ -330,7 +364,7 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> 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,
@@ -341,8 +375,21 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> 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 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).
fn foreign(&mut self, epoch: u64) {
self.mark();
if self.noted != Some(epoch) {
self.noted = Some(epoch);
self.asked = false;
}
}
fn apply(&mut self, circuit: &mut Circuit, delta: ZSet<K>) {
@@ -351,52 +398,56 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
}
fn on_delta(&mut self, circuit: &mut Circuit, d: Delta<K>) {
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<K>) {
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,
Ok(()) => {
self.asked = true;
self.token = token;
self.ask_marks = self.marks;
}
Err(Refused::Full(_)) => {}
Err(Refused::Closed) => self.closed = true,
}
crates/zset/tests/edge.rs
+184 -47
View File
@@ -129,10 +129,12 @@ impl EdgeCodec<u8> for Text {
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, zs(&c.target)),
Frame::Correction(c) => {
format!("C {} {} {} {}", c.epoch, c.through, c.token, zs(&c.target))
}
Frame::Request(r) => match r.have {
Some(s) => format!("R {} {}", s.epoch, s.seq),
None => "R".to_string(),
Some(s) => format!("R {} {} {}", r.token, s.epoch, s.seq),
None => format!("R {}", r.token),
},
}
}
@@ -149,15 +151,17 @@ impl EdgeCodec<u8> for Text {
delta: unzs(it.next().ok_or_else(bad)?)?,
})),
Some("C") => {
let mut it = w.splitn(4, ' ');
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 {
@@ -165,7 +169,7 @@ impl EdgeCodec<u8> for Text {
seq: num(it.next())?,
}),
};
Ok(Frame::Request(Request { have }))
Ok(Frame::Request(Request { have, token }))
}
_ => Err(bad()),
}
@@ -504,11 +508,8 @@ fn a_duplicate_delta_is_applied_once() {
let (i, s) = c.input::<u8>();
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 +544,186 @@ 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<Frame<u8>>, epoch: u64, through: u64, target: Z) -> Frame<u8> {
let mut token = None;
while let Ok(f) = b.try_recv() {
if let Frame::Request(r) = f {
token = Some(r.token);
}
}
Frame::Correction(Correction {
epoch,
through,
token: token.expect("a request was waiting"),
target,
})
}
#[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::<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);
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);
b.try_send(answer(&mut b, 1001, 1, z(&[(5, 1)]))).unwrap();
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!(src.stamp(), Some(Stamp { epoch: 2, seq: 3 }));
assert!(!src.stale(), "[left] and they do not ask for anything");
assert_eq!(c.take(&o), z(&[(5, 1), (6, 1)]), "[late] and 1001 is not shut out");
}
#[test]
fn an_unsolicited_correction_from_another_incarnation_only_marks_stale() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 10);
b.try_send(answer(&mut b, 1, 1, Z::new())).unwrap();
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 a_new_epoch_frame_re_asks_once_after_a_lost_request() {
let (a, mut b) = channel_pipes::<Frame<u8>>(64);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
assert!(matches!(b.try_recv(), Ok(Frame::Request(_))), "[first] the request");
// The sender read it and restarted before answering.
let mut asks = 0;
for n in 0..5u64 {
b.try_send(Frame::Delta(Delta {
epoch: 2,
from: n,
to: n + 1,
delta: Z::new(),
}))
.unwrap();
src.poll(&mut c, 8);
while let Ok(Frame::Request(_)) = b.try_recv() {
asks += 1;
}
}
assert_eq!(asks, 1, "[reask] once for the new epoch, not once per frame");
}
#[test]
fn an_answer_at_the_watermark_says_nothing_newer_and_clears_stale() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
b.try_send(answer(&mut b, 1, 1, Z::new())).unwrap();
src.poll(&mut c, 8);
src.connection_lost();
src.poll(&mut c, 8);
b.try_send(answer(&mut b, 1, 1, Z::new())).unwrap();
src.poll(&mut c, 8);
assert!(!src.stale(), "[dup-answer] through == watermark clears stale");
}
#[test]
fn trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
b.try_send(answer(&mut b, 1, 1, Z::new())).unwrap();
src.poll(&mut c, 8);
src.connection_lost();
src.poll(&mut c, 8);
let reply = answer(&mut b, 1, 1, Z::new());
// A gap seen after the request went out: the reply was cut before it.
b.try_send(Frame::Delta(Delta {
epoch: 1,
from: 4,
to: 5,
delta: Z::new(),
}))
.unwrap();
b.try_send(reply).unwrap();
src.poll(&mut c, 8);
assert!(src.stale(), "[marks] the reply predates the gap");
}
#[test]
fn a_poll_inside_a_transaction_is_refused() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
b.try_send(answer(&mut b, 1, 1, z(&[(1, 1)]))).unwrap();
// Outside a transaction a rollback of the consumer's step keeps what was polled.
src.poll(&mut c, 8);
c.begin();
c.step();
c.rollback();
c.step();
assert_eq!(c.take(&o), z(&[(1, 1)]), "[txn] a step rolled back after poll loses nothing");
c.begin();
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| src.poll(&mut c, 8)));
assert!(r.is_err(), "[txn] a poll inside a transaction would be dropped by rollback");
}
#[test]
#[should_panic(expected = "at least one item")]
fn a_zero_capacity_pipe_is_refused() {
let _ = channel_pipes::<u8>(0);
}
#[test]
fn a_committed_that_overflows_leaves_the_sink_as_it_was() {
let (mut sink, mut src, mut c, _, _) = mirror(8);
sink.committed(z(&[(1, i64::MAX)]));
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
sink.committed(z(&[(2, 1), (1, 1)]))
}));
assert!(r.is_err(), "the weight overflows");
assert_eq!(sink.stamp(), Stamp { epoch: 1, seq: 1 }, "[overflow] no stamp consumed");
drive(&mut sink, &mut src, &mut c);
assert_eq!(src.held(), &z(&[(1, i64::MAX)]), "[overflow] and the key 2 did not slip in");
}
#[test]
@@ -594,7 +739,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 +753,19 @@ 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 }
);
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]