fix(zset): a chunked rebuild is one tick, empty groups have no row, a last chunk is Done (improvise-bw4p.1)

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

context/code/zset.md

@@ -152,15 +152,26 @@ Decisions, where the ADR left a choice:
    owned vectors. `MemTrace` is a `BTreeMap<K, ZSet<V>>`; the `_in` constructors take any
    `Trace`, which is the seam for paging or compression. An empty or inverted range reads
    nothing (`BTreeMap::range` would panic).
-8. **The resumable step takes a predicate, not an `Instant`.** `Rebuild::step(circuit,
-   expired)` loads `chunk` keys per circuit step and stops when `expired()` is true, always
-   running at least one chunk first, so every call progresses. `Instant::now()` panics on
-   wasm32 (`web-time` is a dependency this crate refuses), so the host closes over its own
-   clock. Outputs of the chunks sum to what one delta would give, which the test states.
-   Buffering live deltas during a rebuild (clause 7) is the host's: push them to the same input
-   after the job is done, or to a second input merged with `plus`.
+8. **The resumable step takes a predicate, not an `Instant`, and a load is one logical tick.**
+   `Rebuild::step(circuit, expired)` loads `chunk` keys at a time and stops when `expired()` is
+   true, always running at least one chunk first, so every call progresses. It reports `Done`
+   as soon as the snapshot is exhausted, whatever `expired()` would say. `Instant::now()`
+   panics on wasm32 (`web-time` is a dependency this crate refuses), so the host closes over
+   its own clock. The first chunk is `Circuit::step`; the rest are `step_continue`, which does
+   not advance the clock. A circuit tick that arrives in pieces is exact for the incremental
+   operators because their outputs are deltas. For the temporal ones the pieces are shaped so
+   the outputs still sum to the one-step output: a continued `delay` emits nothing and adds to
+   what it holds, a continued `integrate` emits only its piece. An empty snapshot still takes
+   its one tick; a finished job takes none. Deltas pushed to the same input between calls
+   are absorbed into the next piece. Buffering live deltas during a rebuild (clause 7) is the
+   host's.
 9. **Not built:** `Circuit` does not own a job runner (ADR 008's), and nothing here is
    serialised (clause 9, `improvise-protocol`).
 
 The tests are `crates/zset/tests/circuit.rs`; the rivals and where they failed are in
 `context/test-evidence/zset.md`.
+
+A panic inside `step` (a weight overflow in an operator, say) leaves the circuit unusable:
+nodes ahead of the panic have updated their state and the rest have not, and there is no
+rollback. Drop the circuit and build another. ADR 009 clause 11's per-command snapshot and
+restore is the host's job, not this crate's; `zset-ffi` poisons the circuit for this reason.

context/test-evidence/zset.md

@@ -354,3 +354,44 @@ covered by the same guard, but the equal-bounds corner is rare there.
 
 A counting `Trace` handed to `join_indexed_in` is looked up, so the join's state really is
 behind the trace interface and not a private map beside it.
+
+## `crates/zset/tests/circuit.rs`: review of #345
+
+The Opus review of PR #345 at ba17f92b found four defects; each test below failed before its fix.
+**Injected afterwards**, one at a time: `Rebuild` calling `step` for every chunk fails
+`a_chunked_rebuild_is_one_logical_step_for_delay_integrate_and_differentiate` and
+`a_chunked_temporal_rebuild_equals_one_step`; a continued `integrate` re-emitting its integral
+fails the same two; a continued `delay` emitting what it holds fails the first; folding empty
+groups (before or after) fails the two aggregate tests below; asking the deadline before
+checking the job is done fails `the_last_chunk_is_done_even_when_the_deadline_has_expired`.
+
+### a_chunked_rebuild_is_one_logical_step_for_delay_integrate_and_differentiate
+
+Snapshot `{1, 2}` after one prior tick `{9}`, loaded in chunks of one with a deadline always
+expired, against chunks of five. The three sinks (delay, integrate, differentiate) must sum
+equal. Before the fix delay gave `{1:1}` where `{9:1}` was due, integrate `{1:2, 2:1}`.
+
+### a_chunked_temporal_rebuild_equals_one_step
+
+The same property over random prior ticks, snapshots, chunk sizes and budgets, through a
+pipeline that combines all three temporal operators and ends in `distinct`.
+
+### an_empty_snapshot_is_still_one_tick
+
+An empty snapshot ticks the clock once (a `delay` shows the prior input), and a finished job
+does not tick again.
+
+### the_last_chunk_is_done_even_when_the_deadline_has_expired
+
+One chunk exhausts the snapshot with an always-expired deadline: `Done`, not `Pending`.
+
+### a_group_that_empties_leaves_no_phantom_row_for_a_count
+
+A count closure, insert then delete then insert another: before the fix the delete step gave
+phantom `(7, 0)` rows for the empty group, e.g. `{(7,0):-1, (7,1):1}` where `{(7,1):1}` was due.
+
+### an_aggregate_that_returns_some_on_empty_input_matches_recompute
+
+The same closure through the circuit against recompute-from-scratch. The earlier aggregate
+properties used closures that return `None` on empty input, which could not see the defect.
+`aggregate_state_refolds_only_the_touched_keys` now asserts both steps' fold counts and outputs.

crates/zset/src/circuit.rs

@@ -47,7 +47,8 @@ impl<K: Ord> Output<K> {
 }
 
 trait Node {
-    fn eval(&mut self);
+    /// `cont` is true for a later piece of a tick that has already begun.
+    fn eval(&mut self, cont: bool);
 }
 
 struct Unary<I, O, F> {
@@ -57,7 +58,7 @@ struct Unary<I, O, F> {
 }
 
 impl<I, O, F: FnMut(&I) -> O> Node for Unary<I, O, F> {
-    fn eval(&mut self) {
+    fn eval(&mut self, _cont: bool) {
         let v = (self.f)(&self.src.0.borrow());
         *self.out.0.borrow_mut() = v;
     }
@@ -71,12 +72,47 @@ struct Binary<A, B, O, F> {
 }
 
 impl<A, B, O, F: FnMut(&A, &B) -> O> Node for Binary<A, B, O, F> {
-    fn eval(&mut self) {
+    fn eval(&mut self, _cont: bool) {
         let v = (self.f)(&self.a.0.borrow(), &self.b.0.borrow());
         *self.out.0.borrow_mut() = v;
     }
 }
 
+/// out(t) = in(t - 1). A continued piece adds to what is held and emits nothing.
+struct Delay<K> {
+    src: Stream<ZSet<K>>,
+    out: Stream<ZSet<K>>,
+    held: ZSet<K>,
+}
+
+impl<K: Ord + Clone> Node for Delay<K> {
+    fn eval(&mut self, cont: bool) {
+        let d = self.src.0.borrow();
+        let v = if cont {
+            self.held += &*d;
+            ZSet::new()
+        } else {
+            std::mem::replace(&mut self.held, d.clone())
+        };
+        *self.out.0.borrow_mut() = v;
+    }
+}
+
+/// out(t) = in(0) + .. + in(t). A continued piece emits only its own input.
+struct Integrate<K> {
+    src: Stream<ZSet<K>>,
+    out: Stream<ZSet<K>>,
+    acc: ZSet<K>,
+}
+
+impl<K: Ord + Clone> Node for Integrate<K> {
+    fn eval(&mut self, cont: bool) {
+        let d = self.src.0.borrow();
+        self.acc += &*d;
+        *self.out.0.borrow_mut() = if cont { d.clone() } else { self.acc.clone() };
+    }
+}
+
 /// Moves what was pushed into the circuit's stream.
 struct Source<K> {
     pending: Rc<RefCell<ZSet<K>>>,
@@ -84,7 +120,7 @@ struct Source<K> {
 }
 
 impl<K: Ord> Node for Source<K> {
-    fn eval(&mut self) {
+    fn eval(&mut self, _cont: bool) {
         *self.out.0.borrow_mut() = std::mem::take(&mut *self.pending.borrow_mut());
     }
 }
@@ -105,8 +141,18 @@ impl Circuit {
 
     /// Runs one clock tick: every node, once, in order.
     pub fn step(&mut self) {
+        self.run(false);
+    }
+
+    /// Runs another piece of the tick the last `step` began; the clock does not advance.
+    /// Pieces sum to one `step` of their combined input (`context/code/zset.md`).
+    pub fn step_continue(&mut self) {
+        self.run(true);
+    }
+
+    fn run(&mut self, cont: bool) {
         for n in &mut self.nodes {
-            n.eval();
+            n.eval(cont);
         }
     }
 
@@ -161,17 +207,24 @@ impl Circuit {
 
     /// out(t) = in(t - 1); out(0) is zero.
     pub fn delay<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
-        let mut held = ZSet::new();
-        self.unary(s, move |d: &Z<K>| std::mem::replace(&mut held, d.clone()))
+        let out = Stream(Rc::default());
+        self.nodes.push(Box::new(Delay {
+            src: s.clone(),
+            out: out.clone(),
+            held: ZSet::new(),
+        }));
+        out
     }
 
     /// out(t) = in(0) + .. + in(t).
     pub fn integrate<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
-        let mut acc = ZSet::new();
-        self.unary(s, move |d: &Z<K>| {
-            acc += d;
-            acc.clone()
-        })
+        let out = Stream(Rc::default());
+        self.nodes.push(Box::new(Integrate {
+            src: s.clone(),
+            out: out.clone(),
+            acc: ZSet::new(),
+        }));
+        out
     }
 
     /// out(t) = in(t) - in(t - 1).
@@ -306,12 +359,13 @@ pub enum Progress {
     Done,
 }
 
-/// Building from a snapshot as a sliced job: the snapshot enters an input a chunk at a
-/// time, one circuit step per chunk. Outputs sum to what one big delta would give.
+/// A snapshot load as a sliced job: one logical tick, a `step` then `step_continue`s.
+/// Outputs sum to one big delta's, temporal operators included (`context/code/zset.md`).
 pub struct Rebuild<K> {
     input: Input<K>,
     rest: std::collections::btree_map::IntoIter<K, i64>,
     chunk: usize,
+    started: bool,
 }
 
 impl<K: Ord> Rebuild<K> {
@@ -321,19 +375,28 @@ impl<K: Ord> Rebuild<K> {
             input,
             rest: snapshot.into_iter(),
             chunk: chunk.max(1),
+            started: false,
         }
     }
 
     /// Runs chunks until done or `expired()`, at least one per call. A predicate, not an
-    /// `Instant`, which panics on wasm32 (`context/code/zset.md`).
+    /// `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 {
         loop {
             let chunk: ZSet<K> = self.rest.by_ref().take(self.chunk).collect();
-            if chunk.is_zero() {
+            if chunk.is_zero() && self.started {
                 return Progress::Done;
             }
             self.input.push(chunk);
-            circuit.step();
+            if self.started {
+                circuit.step_continue();
+            } else {
+                circuit.step();
+                self.started = true;
+            }
+            if self.rest.len() == 0 {
+                return Progress::Done;
+            }
             if expired() {
                 return Progress::Pending;
             }

crates/zset/src/ops.rs

@@ -76,7 +76,7 @@ impl<TA, TB> JoinState<TA, TB> {
 }
 
 /// Incremental group-by: `f` folds one key's integrated values to at most one result.
-/// Only the touched keys are re-folded: out = -(k, f(old)) + (k, f(new)).
+/// An empty group has no row, and `f` is not asked about it. Only touched keys are re-folded: out = -(k, f(old)) + (k, f(new)).
 #[derive(Clone, Debug, Default)]
 pub struct AggregateState<T> {
     groups: T,
@@ -104,13 +104,18 @@ impl<T> AggregateState<T> {
         }
         let mut out = ZSet::new();
         for (k, changes) in touched {
-            if let Some(r) = f(k, &self.groups.lookup(k)) {
+            let before = self.groups.lookup(k);
+            if let Some(r) = Some(&before)
+                .filter(|g| !g.is_empty())
+                .and_then(|g| f(k, g))
+            {
                 out.insert((k.clone(), r), -1);
             }
             for (v, w) in changes {
                 self.groups.apply(k, v, w);
             }
-            if let Some(r) = f(k, &self.groups.lookup(k)) {
+            let after = self.groups.lookup(k);
+            if let Some(r) = Some(&after).filter(|g| !g.is_empty()).and_then(|g| f(k, g)) {
                 out.insert((k.clone(), r), 1);
             }
         }

crates/zset/tests/circuit.proptest-regressions

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc da6d39e3f2d54dc796842f642b0fc17d09a1491a9937c3d3f95d453d2d9487ed # shrinks to prior = [(3, -1)], snapshot = [(3, 2), (5, -2), (0, 1), (1, 1)], chunk = 0, budget = 1

crates/zset/tests/circuit.rs

@@ -377,4 +377,131 @@ fn aggregate_state_refolds_only_the_touched_keys() {
     let out = st.step(&Z::singleton((7, 1), 1), f);
     assert_eq!(folds.get(), 2, "one key, folded before and after");
     assert_eq!(out, Z::from_iter([((7, 1), -1), ((7, 2), 1)]));
+    folds.set(0);
+    let out = st.step(&Z::singleton((7, 1), -1), f);
+    assert_eq!(folds.get(), 2, "the second step also folds only its key");
+    assert_eq!(out, Z::from_iter([((7, 2), -1), ((7, 1), 1)]));
+}
+
+fn temporal(c: &mut Circuit, s: &Stream<Z<u8>>) -> Stream<Z<u8>> {
+    let d = c.delay(s);
+    let i = c.integrate(s);
+    let df = c.differentiate(s);
+    let a = c.plus(&d, &i);
+    let b = c.plus(&a, &df);
+    c.distinct(&b)
+}
+
+/// The sums of three sinks over a snapshot loaded after one prior tick.
+fn loaded(snapshot: &Z<u8>, chunk: usize, budget: usize) -> (Z<u8>, Z<u8>, Z<u8>) {
+    let mut c = Circuit::new();
+    let (input, s) = c.input::<u8>();
+    let (d, i, df) = (c.delay(&s), c.integrate(&s), c.differentiate(&s));
+    let sinks = (c.output(&d), c.output(&i), c.output(&df));
+    input.push(Z::singleton(9, 1));
+    c.step();
+    let _ = (sinks.0.take(), sinks.1.take(), sinks.2.take());
+    let mut job = Rebuild::new(input, snapshot.clone(), chunk);
+    for _ in 0..50 {
+        let left = Cell::new(budget);
+        let p = job.step(&mut c, || {
+            left.set(left.get() - 1);
+            left.get() == 0
+        });
+        if p == Progress::Done {
+            break;
+        }
+    }
+    (sinks.0.take(), sinks.1.take(), sinks.2.take())
+}
+
+#[test]
+fn a_chunked_rebuild_is_one_logical_step_for_delay_integrate_and_differentiate() {
+    let snap: Z<u8> = [(1, 1), (2, 1)].into_iter().collect();
+    let want = loaded(&snap, 5, 5);
+    assert_eq!(
+        want.0,
+        Z::singleton(9, 1),
+        "delay shows the previous tick once"
+    );
+    assert_eq!(loaded(&snap, 1, 1), want, "[temporal-chunk]");
+}
+
+#[test]
+fn an_empty_snapshot_is_still_one_tick() {
+    let mut c = Circuit::new();
+    let (input, s) = c.input::<u8>();
+    let d = c.delay(&s);
+    input.push(Z::singleton(1, 1));
+    c.step();
+    let mut job = Rebuild::new(input, Z::new(), 3);
+    assert_eq!(job.step(&mut c, || false), Progress::Done);
+    assert_eq!(d.current().len(), 1, "the empty load ticked the clock once");
+    assert_eq!(job.step(&mut c, || false), Progress::Done);
+    assert_eq!(d.current().len(), 1, "a finished job does not tick again");
+}
+
+#[test]
+fn the_last_chunk_is_done_even_when_the_deadline_has_expired() {
+    let mut c = Circuit::new();
+    let (input, _s) = c.input::<u8>();
+    let mut job = Rebuild::new(input, Z::singleton(1, 1), 4);
+    assert_eq!(job.step(&mut c, || true), Progress::Done);
+}
+
+proptest! {
+    #[test]
+    fn a_chunked_temporal_rebuild_equals_one_step(
+        prior in pairs(), snapshot in pairs(), chunk in 0usize..4, budget in 1usize..4,
+    ) {
+        let (prior, snapshot): (Z<u8>, Z<u8>) = (prior.into_iter().collect(), snapshot.into_iter().collect());
+        let run = |chunk: usize, budget: usize| {
+            let mut c = Circuit::new();
+            let (input, s) = c.input::<u8>();
+            let out = temporal(&mut c, &s);
+            let sink = c.output(&out);
+            input.push(prior.clone());
+            c.step();
+            let _ = sink.take();
+            let mut job = Rebuild::new(input, snapshot.clone(), chunk);
+            for _ in 0..50 {
+                let left = Cell::new(budget);
+                if job.step(&mut c, || { left.set(left.get() - 1); left.get() == 0 }) == Progress::Done { break; }
+            }
+            sink.take()
+        };
+        prop_assert_eq!(run(chunk, budget), run(100, 100), "[temporal-rebuild]");
+    }
+}
+
+#[test]
+fn a_group_that_empties_leaves_no_phantom_row_for_a_count() {
+    let mut st = AggregateState::new(MemTrace::<u8, u8>::new());
+    let count = |_: &u8, vals: &[(u8, i64)]| Some(vals.len() as i64);
+    assert_eq!(
+        st.step(&Z::singleton((7, 1), 1), count),
+        Z::singleton((7, 1), 1)
+    );
+    assert_eq!(
+        st.step(&Z::singleton((7, 1), -1), count),
+        Z::singleton((7, 1), -1)
+    );
+    assert_eq!(
+        st.step(&Z::singleton((7, 2), 1), count),
+        Z::singleton((7, 1), 1)
+    );
+}
+
+#[test]
+fn an_aggregate_that_returns_some_on_empty_input_matches_recompute() {
+    let steps = vec![
+        Z::singleton((7u8, 1u8), 1),
+        Z::singleton((7, 1), -1),
+        Z::singleton((7, 2), 1),
+    ];
+    matches_recompute(
+        &steps,
+        |c, s| c.aggregate(s, |_, vals| Some(vals.len() as i64)),
+        |z| group(z, |vals| Some(vals.len() as i64)),
+    );
 }