diff --git a/crates/zset/src/edge.rs b/crates/zset/src/edge.rs index 2e580568..71d77a4c 100644 --- a/crates/zset/src/edge.rs +++ b/crates/zset/src/edge.rs @@ -198,7 +198,11 @@ impl, P: Pipe> EdgeSink { /// The same pipe and codec, after a restart: a new incarnation at `seq` holding `integral`. pub fn restart(self, epoch: u64, seq: u64, integral: ZSet) -> Self { - Self::resume(self.codec, self.pipe, epoch, seq, integral) + let traffic = self.traffic; + EdgeSink { + traffic, + ..Self::resume(self.codec, self.pipe, epoch, seq, integral) + } } /// One committed command changed the output by `delta` (empty is still a command). @@ -345,9 +349,10 @@ impl, P: Pipe> EdgeSou /// The same pipe and codec, after the consumer lost its state: it holds nothing again. pub fn restart(self, input: Input) -> Self { - let token = self.token; + let (token, traffic) = (self.token, self.traffic); EdgeSource { token, + traffic, ..Self::new(self.codec, self.pipe, input) } } diff --git a/crates/zset/tests/edge.rs b/crates/zset/tests/edge.rs index 62e4b16d..f6de5a09 100644 --- a/crates/zset/tests/edge.rs +++ b/crates/zset/tests/edge.rs @@ -892,11 +892,13 @@ fn a_poll_reads_at_most_the_frames_it_is_given() { let mut c = Circuit::new(); let (i, _) = c.input::(); let mut src = EdgeSource::new(Direct, a, i); + src.poll(&mut c, 0); for through in 0..3 { let target = Z::new(); let frame = Frame::Correction(Correction { epoch: 1, through, + token: 1, target, }); b.try_send(frame).unwrap();