From 0849981ae1bb424bb2c425926542a4d2d5ebddc1 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-circuit)" Date: Sat, 10 Oct 2026 12:06:46 -0700 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/zset.md | 25 +++- context/test-evidence/zset.md | 41 ++++++ crates/zset/src/circuit.rs | 97 ++++++++++--- crates/zset/src/ops.rs | 11 +- .../zset/tests/circuit.proptest-regressions | 7 + crates/zset/tests/circuit.rs | 127 ++++++++++++++++++ 6 files changed, 281 insertions(+), 27 deletions(-) create mode 100644 crates/zset/tests/circuit.proptest-regressions diff --git a/context/code/zset.md b/context/code/zset.md index 18928337..227106dd 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -152,15 +152,26 @@ Decisions, where the ADR left a choice: owned vectors. `MemTrace` is a `BTreeMap>`; 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. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index d2e03476..bb07b4c2 100644 --- a/context/test-evidence/zset.md +++ b/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. diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs index e1c3812e..6c970ba8 100644 --- a/crates/zset/src/circuit.rs +++ b/crates/zset/src/circuit.rs @@ -47,7 +47,8 @@ impl Output { } 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 { @@ -57,7 +58,7 @@ struct Unary { } impl O> Node for Unary { - 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 { } impl O> Node for Binary { - 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 { + src: Stream>, + out: Stream>, + held: ZSet, +} + +impl Node for Delay { + 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 { + src: Stream>, + out: Stream>, + acc: ZSet, +} + +impl Node for Integrate { + 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 { pending: Rc>>, @@ -84,7 +120,7 @@ struct Source { } impl Node for Source { - 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(&mut self, s: &Stream>) -> Stream> { - let mut held = ZSet::new(); - self.unary(s, move |d: &Z| 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(&mut self, s: &Stream>) -> Stream> { - let mut acc = ZSet::new(); - self.unary(s, move |d: &Z| { - 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 { input: Input, rest: std::collections::btree_map::IntoIter, chunk: usize, + started: bool, } impl Rebuild { @@ -321,19 +375,28 @@ impl Rebuild { 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 = 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; } diff --git a/crates/zset/src/ops.rs b/crates/zset/src/ops.rs index 7c2caec0..95a9701f 100644 --- a/crates/zset/src/ops.rs +++ b/crates/zset/src/ops.rs @@ -76,7 +76,7 @@ impl JoinState { } /// 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 { groups: T, @@ -104,13 +104,18 @@ impl AggregateState { } 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); } } diff --git a/crates/zset/tests/circuit.proptest-regressions b/crates/zset/tests/circuit.proptest-regressions new file mode 100644 index 00000000..6ece7c62 --- /dev/null +++ b/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 diff --git a/crates/zset/tests/circuit.rs b/crates/zset/tests/circuit.rs index e7cff63f..57cf168b 100644 --- a/crates/zset/tests/circuit.rs +++ b/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>) -> Stream> { + 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, chunk: usize, budget: usize) -> (Z, Z, Z) { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + 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 = [(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::(); + 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::(); + 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, Z) = (prior.into_iter().collect(), snapshot.into_iter().collect()); + let run = |chunk: usize, budget: usize| { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + 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::::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)), + ); }