From 203dc5f3f74ef166aee1fba9733a5aab60bb07fa Mon Sep 17 00:00:00 2001
From: "Claude (dbsp-threads)" <el-github@elangley.org>
Date: Sat, 10 Oct 2026 21:35:15 -0700
Subject: [PATCH] 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
---
 crates/zset/src/edge.rs   | 9 +++++++--
 crates/zset/tests/edge.rs | 2 ++
 2 files changed, 9 insertions(+), 2 deletions(-)

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<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`.
     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).
@@ -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.
     pub fn restart(self, input: Input<K>) -> 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::<u8>();
     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();