fix(zset): edge counters survive a restart of either end (merge fixes)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze
This commit is contained in:
committed by
Edward Langley
parent
92c7875673
commit
203dc5f3f7
@@ -198,7 +198,11 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
|
|||||||
|
|
||||||
/// The same pipe and codec, after a restart: a new incarnation at `seq` holding `integral`.
|
/// 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<K>) -> Self {
|
pub fn restart(self, epoch: u64, seq: u64, integral: ZSet<K>) -> 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).
|
/// One committed command changed the output by `delta` (empty is still a command).
|
||||||
@@ -345,9 +349,10 @@ 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.
|
/// The same pipe and codec, after the consumer lost its state: it holds nothing again.
|
||||||
pub fn restart(self, input: Input<K>) -> Self {
|
pub fn restart(self, input: Input<K>) -> Self {
|
||||||
let token = self.token;
|
let (token, traffic) = (self.token, self.traffic);
|
||||||
EdgeSource {
|
EdgeSource {
|
||||||
token,
|
token,
|
||||||
|
traffic,
|
||||||
..Self::new(self.codec, self.pipe, input)
|
..Self::new(self.codec, self.pipe, input)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -892,11 +892,13 @@ fn a_poll_reads_at_most_the_frames_it_is_given() {
|
|||||||
let mut c = Circuit::new();
|
let mut c = Circuit::new();
|
||||||
let (i, _) = c.input::<u8>();
|
let (i, _) = c.input::<u8>();
|
||||||
let mut src = EdgeSource::new(Direct, a, i);
|
let mut src = EdgeSource::new(Direct, a, i);
|
||||||
|
src.poll(&mut c, 0);
|
||||||
for through in 0..3 {
|
for through in 0..3 {
|
||||||
let target = Z::new();
|
let target = Z::new();
|
||||||
let frame = Frame::Correction(Correction {
|
let frame = Frame::Correction(Correction {
|
||||||
epoch: 1,
|
epoch: 1,
|
||||||
through,
|
through,
|
||||||
|
token: 1,
|
||||||
target,
|
target,
|
||||||
});
|
});
|
||||||
b.try_send(frame).unwrap();
|
b.try_send(frame).unwrap();
|
||||||
|
|||||||
Reference in New Issue
Block a user