fix(zset): a Rebuild is bound to the circuit its input came from (improvise-bw4p.12)

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

crates/zset/src/circuit.rs

@@ -73,12 +73,13 @@ impl<T: Clone> Stream<T> {
     }
 }
 
-/// Where a caller feeds deltas; they enter the circuit at the next `step`.
-pub struct Input<K>(Rc<RefCell<ZSet<K>>>);
+/// Where a caller feeds deltas; they enter the circuit at the next `step`. It remembers
+/// which circuit made it, so a `Rebuild` can refuse a circuit that is not its own.
+pub struct Input<K>(Rc<RefCell<ZSet<K>>>, Rc<()>);
 
 impl<K> Clone for Input<K> {
     fn clone(&self) -> Self {
-        Input(Rc::clone(&self.0))
+        Input(Rc::clone(&self.0), Rc::clone(&self.1))
     }
 }
 
@@ -186,6 +187,8 @@ impl<K: Ord + Clone> Node for Source<K> {
 pub struct Circuit {
     nodes: Vec<Box<dyn Node>>,
     ticks: u64,
+    /// This circuit's identity: every `Input` it makes holds a clone.
+    token: Rc<()>,
 }
 
 type Z<K> = ZSet<K>;
@@ -255,7 +258,7 @@ impl Circuit {
             pending: Rc::clone(&pending),
             out: out.clone(),
         }));
-        (Input(pending), out)
+        (Input(pending, Rc::clone(&self.token)), out)
     }
 
     pub fn output<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Output<K> {
@@ -426,7 +429,8 @@ pub enum Progress {
 }
 
 /// A snapshot load as a sliced job: one logical tick (`context/code/zset.md`). A pending
-/// job owns the clock: a `Circuit::step` before it is Done makes `step` panic.
+/// job owns the clock: a `Circuit::step` before it is Done makes `step` panic, and for good.
+/// Stepping it on a circuit other than its input's panics too.
 pub struct Rebuild<K> {
     input: Input<K>,
     rest: std::collections::btree_map::IntoIter<K, i64>,
@@ -446,8 +450,14 @@ impl<K: Ord> Rebuild<K> {
         }
     }
 
+    /// Whether `circuit` is the one that made this job's input.
+    pub fn belongs_to(&self, circuit: &Circuit) -> bool {
+        Rc::ptr_eq(&self.input.1, &circuit.token)
+    }
+
     /// Whether a `Circuit::step` since this job began has split its tick. A finished job
-    /// cannot be interrupted.
+    /// cannot be interrupted, and an interrupted one can never finish: drop it and start
+    /// another job from the snapshot.
     pub fn interrupted(&self, circuit: &Circuit) -> bool {
         self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks())
     }
@@ -455,6 +465,10 @@ impl<K: Ord> Rebuild<K> {
     /// Runs chunks until done or `expired()`, at least one per call. A predicate, not an
     /// `Instant` (wasm32). An empty snapshot takes its one tick; a finished job none.
     pub fn step(&mut self, circuit: &mut Circuit, mut expired: impl FnMut() -> bool) -> Progress {
+        assert!(
+            self.belongs_to(circuit),
+            "zset: the rebuild's input belongs to another circuit"
+        );
         assert!(
             !self.interrupted(circuit),
             "zset: the rebuild was interrupted by a Circuit::step before it finished"

crates/zset/tests/rebuild.rs

@@ -154,3 +154,42 @@ fn a_finished_rebuild_and_an_interrupted_check_do_not_trip_on_later_ticks() {
     assert!(!job.interrupted(&c));
     assert_eq!(job.step(&mut c, || true), Progress::Done);
 }
+
+#[test]
+#[should_panic(expected = "another circuit")]
+fn a_rebuild_stepped_on_a_foreign_circuit_is_refused_loudly() {
+    let (mut a, mut b) = (Circuit::new(), Circuit::new());
+    let (input, _s) = a.input::<u8>();
+    let mut job = Rebuild::new(input, [(1u8, 1)].into_iter().collect(), 1);
+    job.step(&mut b, || false);
+}
+
+#[test]
+fn a_rebuild_knows_the_circuit_its_input_came_from() {
+    let (mut a, b) = (Circuit::new(), Circuit::new());
+    let (input, _s) = a.input::<u8>();
+    let job = Rebuild::new(input.clone(), ZSet::new(), 1);
+    assert!(
+        job.belongs_to(&a),
+        "[owner] the circuit that made the input"
+    );
+    assert!(!job.belongs_to(&b), "[owner] another circuit");
+}
+
+#[test]
+fn an_interrupted_rebuild_stays_interrupted_and_a_new_job_loads_the_snapshot() {
+    let snap: Z<u8> = [(1, 1), (2, 1), (3, 1)].into_iter().collect();
+    let mut c = Circuit::new();
+    let (input, s) = c.input::<u8>();
+    let i = c.integrate(&s);
+    let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
+    assert_eq!(job.step(&mut c, || true), Progress::Pending);
+    c.step();
+    for _ in 0..3 {
+        assert!(job.interrupted(&c), "[forever] steps never un-interrupt it");
+        c.step();
+    }
+    let mut fresh = Rebuild::new(input, snap, 1);
+    assert_eq!(fresh.step(&mut c, || false), Progress::Done);
+    assert_eq!(i.current().len(), 3);
+}