diff --git a/crates/zset/src/edge.rs b/crates/zset/src/edge.rs
index 2b1415d8..fe6816b1 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<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,
         }
diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs
index c177ef0c..ff01a53d 100644
--- a/crates/zset/tests/edge.rs
+++ b/crates/zset/tests/edge.rs
@@ -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]