fix(zset): count the rollbacks that undid a step so a Rebuild cannot outlive its chunks (improvise-bw4p.5)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze

crates/zset/src/circuit.rs

@@ -407,6 +407,11 @@ pub struct Circuit {
     token: Rc<()>,
     /// `Some(ticks)` as of `begin`, while a transaction is open.
     open: Option<u64>,
+    /// A step or piece ran since `begin`.
+    stepped: bool,
+    /// Rollbacks that undid a step or piece. It never goes down, unlike `ticks`, so a
+    /// `Rebuild` can tell that a rollback took some of its chunks.
+    undone: u64,
 }
 
 type Z<K> = ZSet<K>;
@@ -441,6 +446,7 @@ impl Circuit {
             "zset: a transaction is already open on this circuit"
         );
         self.open = Some(self.ticks);
+        self.stepped = false;
         for n in &mut self.nodes {
             n.begin();
         }
@@ -468,9 +474,12 @@ impl Circuit {
             n.rollback();
         }
         self.ticks = ticks;
+        self.undone += u64::from(self.stepped);
+        self.stepped = false;
     }
 
     fn run(&mut self, cont: bool) {
+        self.stepped |= self.open.is_some();
         for n in &mut self.nodes {
             n.eval(cont);
         }
@@ -735,6 +744,8 @@ pub struct Rebuild<K> {
     chunk: usize,
     /// The circuit's tick count right after this job's first chunk, once it has run one.
     began_at: Option<u64>,
+    /// The circuit's `undone` count when this job first ran, even if that chunk panicked.
+    undone_at: Option<u64>,
 }
 
 impl<K: Ord> Rebuild<K> {
@@ -745,6 +756,7 @@ impl<K: Ord> Rebuild<K> {
             rest: snapshot.into_iter(),
             chunk: chunk.max(1),
             began_at: None,
+            undone_at: None,
         }
     }
 
@@ -753,10 +765,11 @@ impl<K: Ord> Rebuild<K> {
         Rc::ptr_eq(&self.input.1, &circuit.token)
     }
 
-    /// Whether a `Circuit::step` split this job's tick. A finished job is never interrupted;
+    /// Whether a `Circuit::step` split this job's tick, or a rollback undid a step after it
-    /// an interrupted one never finishes: start another from the snapshot.
+    /// first ran (finished or not). An interrupted job never finishes: start another.
     pub fn interrupted(&self, circuit: &Circuit) -> bool {
-        self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks())
+        let split = self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks());
+        split || self.undone_at.is_some_and(|u| u != circuit.undone)
     }
 
     /// Runs chunks until done or `expired()`, at least one per call. A predicate, not an
@@ -766,9 +779,10 @@ impl<K: Ord> Rebuild<K> {
             self.belongs_to(circuit),
             "zset: the rebuild's input belongs to another circuit"
         );
+        self.undone_at.get_or_insert(circuit.undone);
         assert!(
             !self.interrupted(circuit),
-            "zset: the rebuild was interrupted by a Circuit::step before it finished"
+            "zset: the rebuild was interrupted by a Circuit::step or a rollback before it finished"
         );
         loop {
             let chunk: ZSet<K> = self.rest.by_ref().take(self.chunk).collect();