From 42ee9bac9820242563407300025c1a9a96b50b7b Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-circuit)" Date: Sat, 10 Oct 2026 11:44:20 -0700 Subject: [PATCH 1/5] feat(zset): the circuit, incremental operators, traces and resumable rebuild (improvise-bw4p.1) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- crates/zset/src/circuit.rs | 343 +++++++++++++++++++++++++++++++ crates/zset/src/lib.rs | 23 +++ crates/zset/src/ops.rs | 119 +++++++++++ crates/zset/src/trace.rs | 78 +++++++ crates/zset/tests/circuit.rs | 380 +++++++++++++++++++++++++++++++++++ 5 files changed, 943 insertions(+) create mode 100644 crates/zset/src/circuit.rs create mode 100644 crates/zset/src/ops.rs create mode 100644 crates/zset/src/trace.rs create mode 100644 crates/zset/tests/circuit.rs diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs new file mode 100644 index 00000000..ab388313 --- /dev/null +++ b/crates/zset/src/circuit.rs @@ -0,0 +1,343 @@ +//! The synchronous, single-threaded circuit. This module is the wiring (`Rc`, `RefCell`, +//! boxed nodes); the operators' logic is in `ops.rs` and `lib.rs`. + +use crate::ops::{AggregateState, DistinctState, JoinState}; +use crate::{MemTrace, Trace, ZSet}; +use std::cell::RefCell; +use std::rc::Rc; + +/// The value an edge carries during the current step. +pub struct Stream(Rc>); + +impl Clone for Stream { + fn clone(&self) -> Self { + Stream(Rc::clone(&self.0)) + } +} + +impl Stream { + /// The value of the last completed step (the zero before the first). + pub fn current(&self) -> T { + self.0.borrow().clone() + } +} + +/// Where a caller feeds deltas; they enter the circuit at the next `step`. +pub struct Input(Rc>>); + +impl Clone for Input { + fn clone(&self) -> Self { + Input(Rc::clone(&self.0)) + } +} + +impl Input { + pub fn push(&self, delta: ZSet) { + *self.0.borrow_mut() += delta; + } +} + +/// A sink that sums every step's delta until it is taken. +pub struct Output(Rc>>); + +impl Output { + pub fn take(&self) -> ZSet { + std::mem::take(&mut *self.0.borrow_mut()) + } +} + +trait Node { + fn eval(&mut self); +} + +struct Unary { + src: Stream, + out: Stream, + f: F, +} + +impl O> Node for Unary { + fn eval(&mut self) { + let v = (self.f)(&self.src.0.borrow()); + *self.out.0.borrow_mut() = v; + } +} + +struct Binary { + a: Stream, + b: Stream, + out: Stream, + f: F, +} + +impl O> Node for Binary { + fn eval(&mut self) { + let v = (self.f)(&self.a.0.borrow(), &self.b.0.borrow()); + *self.out.0.borrow_mut() = v; + } +} + +/// Moves what was pushed into the circuit's stream. +struct Source { + pending: Rc>>, + out: Stream>, +} + +impl Node for Source { + fn eval(&mut self) { + *self.out.0.borrow_mut() = std::mem::take(&mut *self.pending.borrow_mut()); + } +} + +/// Nodes run in creation order, which is a topological order because a node can only +/// be built from streams that already exist. There are no cycles to schedule. +#[derive(Default)] +pub struct Circuit { + nodes: Vec>, +} + +type Z = ZSet; + +impl Circuit { + pub fn new() -> Self { + Self::default() + } + + /// Runs one clock tick: every node, once, in order. + pub fn step(&mut self) { + for n in &mut self.nodes { + n.eval(); + } + } + + fn unary( + &mut self, + src: &Stream, + f: impl FnMut(&I) -> O + 'static, + ) -> Stream { + let out = Stream(Rc::default()); + self.nodes.push(Box::new(Unary { + src: src.clone(), + out: out.clone(), + f, + })); + out + } + + fn binary( + &mut self, + a: &Stream, + b: &Stream, + f: impl FnMut(&A, &B) -> O + 'static, + ) -> Stream { + let out = Stream(Rc::default()); + self.nodes.push(Box::new(Binary { + a: a.clone(), + b: b.clone(), + out: out.clone(), + f, + })); + out + } + + pub fn input(&mut self) -> (Input, Stream>) { + let pending: Rc>> = Rc::default(); + let out = Stream(Rc::default()); + self.nodes.push(Box::new(Source { + pending: Rc::clone(&pending), + out: out.clone(), + })); + (Input(pending), out) + } + + pub fn output(&mut self, s: &Stream>) -> Output { + let sink = Output(Rc::default()); + let acc = Rc::clone(&sink.0); + self.unary(s, move |d: &Z| { + *acc.borrow_mut() += d; + }); + sink + } + + /// 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())) + } + + /// 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() + }) + } + + /// out(t) = in(t) - in(t - 1). + pub fn differentiate(&mut self, s: &Stream>) -> Stream> { + let prev = self.delay(s); + self.binary(s, &prev, |a: &Z, b: &Z| a - b) + } + + pub fn plus( + &mut self, + a: &Stream>, + b: &Stream>, + ) -> Stream> { + self.binary(a, b, |a: &Z, b: &Z| a + b) + } + + pub fn map( + &mut self, + s: &Stream>, + mut f: impl FnMut(&K) -> L + 'static, + ) -> Stream> { + self.unary(s, move |d: &Z| d.map(&mut f)) + } + + pub fn filter( + &mut self, + s: &Stream>, + mut keep: impl FnMut(&K) -> bool + 'static, + ) -> Stream> { + self.unary(s, move |d: &Z| d.filter(&mut keep)) + } + + /// Each key becomes any number of images, each with the key's weight. + pub fn flat_map>( + &mut self, + s: &Stream>, + mut f: impl FnMut(&K) -> I + 'static, + ) -> Stream> { + self.unary(s, move |d: &Z| d.flat_map(&mut f)) + } + + /// Keys each element by `key`, giving the `(key, element)` pairs a join reads. + pub fn index_by( + &mut self, + s: &Stream>, + mut key: impl FnMut(&K) -> G + 'static, + ) -> Stream> { + self.unary(s, move |d: &Z| d.map(|k| (key(k), k.clone()))) + } + + /// Incremental `distinct`. + pub fn distinct(&mut self, s: &Stream>) -> Stream> { + let mut st = DistinctState::new(); + self.unary(s, move |d: &Z| st.step(d)) + } + + /// The bilinear delta join over in-memory traces. + pub fn join_indexed( + &mut self, + a: &Stream>, + b: &Stream>, + f: impl FnMut(&K, &A, &B) -> O + 'static, + ) -> Stream> + where + K: Ord + Clone + 'static, + A: Ord + Clone + 'static, + B: Ord + Clone + 'static, + O: Ord + 'static, + { + self.join_indexed_in(a, b, MemTrace::new(), MemTrace::new(), f) + } + + /// `join_indexed` over traces the caller chose. + pub fn join_indexed_in( + &mut self, + a: &Stream>, + b: &Stream>, + left: TA, + right: TB, + mut f: impl FnMut(&K, &A, &B) -> O + 'static, + ) -> Stream> + where + K: Ord + Clone + 'static, + A: Ord + Clone + 'static, + B: Ord + Clone + 'static, + O: Ord + 'static, + TA: Trace + 'static, + TB: Trace + 'static, + { + let mut st = JoinState::new(left, right); + self.binary(a, b, move |da: &Z<(K, A)>, db: &Z<(K, B)>| { + st.step(da, db, &mut f) + }) + } + + /// Incremental group-by over in-memory state. + pub fn aggregate( + &mut self, + s: &Stream>, + f: impl FnMut(&K, &[(V, i64)]) -> Option + 'static, + ) -> Stream> + where + K: Ord + Clone + 'static, + V: Ord + Clone + 'static, + R: Ord + 'static, + { + self.aggregate_in(s, MemTrace::new(), f) + } + + /// `aggregate` over a trace the caller chose. + pub fn aggregate_in( + &mut self, + s: &Stream>, + groups: T, + mut f: impl FnMut(&K, &[(V, i64)]) -> Option + 'static, + ) -> Stream> + where + K: Ord + Clone + 'static, + V: Ord + Clone + 'static, + R: Ord + 'static, + T: Trace + 'static, + { + let mut st = AggregateState::new(groups); + self.unary(s, move |d: &Z<(K, V)>| st.step(d, &mut f)) + } +} + +/// Whether a resumable job finished. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Progress { + Pending, + 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. +pub struct Rebuild { + input: Input, + rest: std::collections::btree_map::IntoIter, + chunk: usize, +} + +impl Rebuild { + /// `chunk` keys per circuit step; zero is read as one. + pub fn new(input: Input, snapshot: ZSet, chunk: usize) -> Self { + Rebuild { + input, + rest: snapshot.into_iter(), + chunk: chunk.max(1), + } + } + + /// Runs chunks until the snapshot is consumed or `expired()` says stop. It always + /// runs at least one chunk when any remain, so each call makes progress. `expired` + /// is a predicate, not an `Instant`: `Instant::now()` panics on wasm32. + 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() { + return Progress::Done; + } + self.input.push(chunk); + circuit.step(); + if expired() { + return Progress::Pending; + } + } + } +} diff --git a/crates/zset/src/lib.rs b/crates/zset/src/lib.rs index 8d60c65a..eef8e806 100644 --- a/crates/zset/src/lib.rs +++ b/crates/zset/src/lib.rs @@ -1,6 +1,14 @@ //! Z-sets: maps from keys to non-zero `i64` weights (ADR 009 clause 1). An abelian group //! wherever no weight leaves `i64`; past that it panics. See `context/code/zset.md`. +mod circuit; +mod ops; +mod trace; + +pub use circuit::{Circuit, Input, Output, Progress, Rebuild, Stream}; +pub use ops::{AggregateState, DistinctState, JoinState}; +pub use trace::{MemTrace, Trace}; + use std::collections::BTreeMap; use std::collections::btree_map::Entry; use std::iter::FromIterator; @@ -116,6 +124,21 @@ impl ZSet { out } + /// Maps every key to any number of images, each carrying the key's weight; the + /// weights of equal images sum. + pub fn flat_map>( + &self, + mut f: impl FnMut(&K) -> I, + ) -> ZSet { + let mut out = ZSet::new(); + for (k, w) in &self.weights { + for image in f(k) { + out.insert(image, *w); + } + } + out + } + /// Keeps the keys `keep` accepts, with their weights. It looks at the key only, /// which is what makes it linear. pub fn filter(&self, mut keep: impl FnMut(&K) -> bool) -> Self diff --git a/crates/zset/src/ops.rs b/crates/zset/src/ops.rs new file mode 100644 index 00000000..7c2caec0 --- /dev/null +++ b/crates/zset/src/ops.rs @@ -0,0 +1,119 @@ +//! The incremental operators' state and step, with no wiring: each `step` takes the +//! delta of its inputs and returns the delta of its output. + +use crate::{Trace, ZSet, plus, times}; +use std::collections::BTreeMap; + +/// Incremental `distinct`: out = distinct(held + d) - distinct(held), per touched key. +#[derive(Clone, Debug, Default)] +pub struct DistinctState { + held: ZSet, +} + +impl DistinctState { + pub fn new() -> Self { + DistinctState { held: ZSet::new() } + } + + pub fn step(&mut self, d: &ZSet) -> ZSet { + let mut out = ZSet::new(); + for (k, w) in d.iter() { + let before = self.held.weight(k); + let after = plus(before, w); + out.insert(k.clone(), i64::from(after > 0) - i64::from(before > 0)); + self.held.insert(k.clone(), w); + } + out + } +} + +/// The bilinear delta join of two keyed streams: +/// out = da x b_prev + a_prev x db + da x db, never a rescan of either side. +#[derive(Clone, Debug, Default)] +pub struct JoinState { + left: TA, + right: TB, +} + +impl JoinState { + pub fn new(left: TA, right: TB) -> Self { + JoinState { left, right } + } + + pub fn step( + &mut self, + da: &ZSet<(K, A)>, + db: &ZSet<(K, B)>, + mut f: impl FnMut(&K, &A, &B) -> O, + ) -> ZSet + where + K: Ord + Clone, + A: Ord + Clone, + B: Ord + Clone, + O: Ord, + TA: Trace, + TB: Trace, + { + let mut out = ZSet::new(); + for ((k, a), wa) in da.iter() { + for (b, wb) in self.right.lookup(k) { + out.insert(f(k, a, &b), times(wa, wb)); + } + } + for ((k, a), wa) in da.iter() { + self.left.apply(k, a, wa); + } + for ((k, b), wb) in db.iter() { + for (a, wa) in self.left.lookup(k) { + out.insert(f(k, &a, b), times(wa, wb)); + } + } + for ((k, b), wb) in db.iter() { + self.right.apply(k, b, wb); + } + out + } +} + +/// 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)). +#[derive(Clone, Debug, Default)] +pub struct AggregateState { + groups: T, +} + +impl AggregateState { + pub fn new(groups: T) -> Self { + AggregateState { groups } + } + + pub fn step( + &mut self, + d: &ZSet<(K, V)>, + mut f: impl FnMut(&K, &[(V, i64)]) -> Option, + ) -> ZSet<(K, R)> + where + K: Ord + Clone, + V: Ord + Clone, + R: Ord, + T: Trace, + { + let mut touched: BTreeMap<&K, Vec<(&V, i64)>> = BTreeMap::new(); + for ((k, v), w) in d.iter() { + touched.entry(k).or_default().push((v, w)); + } + let mut out = ZSet::new(); + for (k, changes) in touched { + if let Some(r) = f(k, &self.groups.lookup(k)) { + 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)) { + out.insert((k.clone(), r), 1); + } + } + out + } +} diff --git a/crates/zset/src/trace.rs b/crates/zset/src/trace.rs new file mode 100644 index 00000000..c8c0a1a5 --- /dev/null +++ b/crates/zset/src/trace.rs @@ -0,0 +1,78 @@ +//! The trace interface: operator state behind look up, read a range, apply a delta. + +use crate::ZSet; +use std::collections::BTreeMap; +use std::ops::Bound; + +/// An integrated `ZSet<(K, V)>`, indexed by `K`. Reads return owned values, so an +/// implementation may page or compress what it holds. See `context/code/zset.md`. +pub trait Trace { + /// Adds `w` to the weight of `(k, v)`. + fn apply(&mut self, k: &K, v: &V, w: i64); + /// The values at `k` with their non-zero weights, ascending by value. + fn lookup(&self, k: &K) -> Vec<(V, i64)>; + /// Every `(key, value, weight)` with a key in the range, ascending. + fn range(&self, lo: Bound<&K>, hi: Bound<&K>) -> Vec<(K, V, i64)>; +} + +/// The in-memory trace: a sorted map from key to a Z-set of values. +#[derive(Clone, Debug)] +pub struct MemTrace { + groups: BTreeMap>, +} + +impl Default for MemTrace { + fn default() -> Self { + MemTrace { + groups: BTreeMap::new(), + } + } +} + +impl MemTrace { + pub fn new() -> Self { + Self::default() + } + + /// The number of keys with at least one value. + pub fn keys(&self) -> usize { + self.groups.len() + } +} + +fn is_empty_range(lo: Bound<&K>, hi: Bound<&K>) -> bool { + match (lo, hi) { + (Bound::Included(a), Bound::Included(b)) => a > b, + (Bound::Included(a) | Bound::Excluded(a), Bound::Included(b) | Bound::Excluded(b)) => { + a >= b + } + _ => false, + } +} + +impl Trace for MemTrace { + fn apply(&mut self, k: &K, v: &V, w: i64) { + let group = self.groups.entry(k.clone()).or_default(); + group.insert(v.clone(), w); + if group.is_zero() { + self.groups.remove(k); + } + } + + fn lookup(&self, k: &K) -> Vec<(V, i64)> { + self.groups + .get(k) + .map(|g| g.iter().map(|(v, w)| (v.clone(), w)).collect()) + .unwrap_or_default() + } + + fn range(&self, lo: Bound<&K>, hi: Bound<&K>) -> Vec<(K, V, i64)> { + if is_empty_range(lo, hi) { + return Vec::new(); + } + self.groups + .range::((lo, hi)) + .flat_map(|(k, g)| g.iter().map(move |(v, w)| (k.clone(), v.clone(), w))) + .collect() + } +} diff --git a/crates/zset/tests/circuit.rs b/crates/zset/tests/circuit.rs new file mode 100644 index 00000000..04eb0aa1 --- /dev/null +++ b/crates/zset/tests/circuit.rs @@ -0,0 +1,380 @@ +//! The circuit's operators against recomputing from scratch, and the DBSP identities. +//! Evidence: `context/test-evidence/zset.md`. + +use proptest::prelude::*; +use std::cell::Cell; +use std::ops::Bound; +use std::rc::Rc; +use zset::{ + AggregateState, Circuit, MemTrace, Progress, Rebuild, Stream, Trace, ZSet, differentiate, + integrate, +}; + +type Z = ZSet; + +fn pairs() -> impl Strategy> { + prop::collection::vec((0u8..6, -3i64..=3), 0..6) +} + +fn stream() -> impl Strategy>> { + prop::collection::vec(pairs().prop_map(|p| p.into_iter().collect::>()), 1..8) +} + +fn kv() -> impl Strategy>> { + let one = prop::collection::vec(((0u8..4, 0u8..3), -3i64..=3), 0..6) + .prop_map(|p| p.into_iter().collect::>()); + prop::collection::vec(one, 1..8) +} + +/// Feeds `steps` to one input, one circuit step each, and checks after every step that +/// the sum of the outputs so far is `reference` of the sum of the inputs so far. +fn matches_recompute( + steps: &[Z], + build: impl FnOnce(&mut Circuit, &Stream>) -> Stream>, + reference: impl Fn(&Z) -> Z, +) where + K: Ord + Clone + std::fmt::Debug + 'static, + L: Ord + Clone + std::fmt::Debug + 'static, +{ + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let out = build(&mut c, &s); + let sink = c.output(&out); + let (mut fed, mut got) = (Z::new(), Z::new()); + for (i, d) in steps.iter().enumerate() { + input.push(d.clone()); + c.step(); + fed += d; + got += sink.take(); + assert_eq!(got, reference(&fed), "[incremental] after step {i}"); + } +} + +fn sum_by_key(z: &Z<(u8, u8)>) -> Z<(u8, i64)> { + group(z, |vals| { + let s: i64 = vals.iter().map(|(v, w)| i64::from(*v) * w).sum(); + (s != 0).then_some(s) + }) +} + +/// The recompute-from-scratch aggregate: `f` over each key's integrated values. +fn group(z: &Z<(u8, u8)>, f: impl Fn(&[(u8, i64)]) -> Option) -> Z<(u8, i64)> { + let mut out = Z::new(); + for k in z + .iter() + .map(|((k, _), _)| *k) + .collect::>() + { + let vals: Vec<(u8, i64)> = z + .iter() + .filter(|((a, _), _)| *a == k) + .map(|((_, v), w)| (*v, w)) + .collect(); + if let Some(r) = f(&vals) { + out.insert((k, r), 1); + } + } + out +} + +fn max_present(vals: &[(u8, i64)]) -> Option { + vals.iter() + .filter(|(_, w)| *w > 0) + .map(|(v, _)| i64::from(*v)) + .max() +} + +fn join_ref(a: &Z<(u8, u8)>, b: &Z<(u8, u8)>) -> Z<(u8, u8, u8)> { + let mut out = Z::new(); + for ((k, x), wa) in a.iter() { + for ((k2, y), wb) in b.iter() { + if k == k2 { + out.insert((*k, *x, *y), wa * wb); + } + } + } + out +} + +fn join_circuit(da: &[Z<(u8, u8)>], db: &[Z<(u8, u8)>]) -> Vec> { + let mut c = Circuit::new(); + let (ia, a) = c.input::<(u8, u8)>(); + let (ib, b) = c.input::<(u8, u8)>(); + let j = c.join_indexed(&a, &b, |k, x, y| (*k, *x, *y)); + let sink = c.output(&j); + let mut deltas = Vec::new(); + for i in 0..da.len().max(db.len()) { + ia.push(da.get(i).cloned().unwrap_or_default()); + ib.push(db.get(i).cloned().unwrap_or_default()); + c.step(); + deltas.push(sink.take()); + } + deltas +} + +proptest! { + #[test] + fn map_matches_recompute(steps in stream()) { + matches_recompute(&steps, |c, s| c.map(s, |k| k % 3), |z| z.map(|k| k % 3)); + } + + #[test] + fn filter_matches_recompute(steps in stream()) { + matches_recompute(&steps, |c, s| c.filter(s, |k| k % 2 == 0), |z| z.filter(|k| k % 2 == 0)); + } + + #[test] + fn flat_map_matches_recompute(steps in stream()) { + let f = |k: &u8| vec![*k % 3, *k % 3 + 10, 20]; + matches_recompute(&steps, |c, s| c.flat_map(s, f), |z| z.flat_map(f)); + } + + #[test] + fn index_by_matches_recompute(steps in stream()) { + matches_recompute(&steps, |c, s| c.index_by(s, |k| k % 2), |z| z.map(|k| (k % 2, *k))); + } + + #[test] + fn distinct_matches_recompute(steps in stream()) { + matches_recompute(&steps, |c, s| c.distinct(s), |z| z.distinct()); + } + + #[test] + fn a_composed_pipeline_matches_recompute(steps in stream()) { + let pipeline = |z: &Z| z.filter(|k| *k != 3).map(|k| k % 3).distinct(); + matches_recompute( + &steps, + |c, s| { + let f = c.filter(s, |k| *k != 3); + let m = c.map(&f, |k| k % 3); + c.distinct(&m) + }, + pipeline, + ); + } + + #[test] + fn aggregate_sum_matches_recompute(steps in kv()) { + matches_recompute( + &steps, + |c, s| c.aggregate(s, |_, vals| { + let t: i64 = vals.iter().map(|(v, w)| i64::from(*v) * w).sum(); + (t != 0).then_some(t) + }), + sum_by_key, + ); + } + + #[test] + fn aggregate_max_matches_recompute(steps in kv()) { + matches_recompute( + &steps, + |c, s| c.aggregate(s, |_, vals| max_present(vals)), + |z| group(z, max_present), + ); + } + + #[test] + fn join_matches_recompute(da in kv(), db in kv()) { + let deltas = join_circuit(&da, &db); + let (mut a, mut b, mut got) = (Z::new(), Z::new(), Z::new()); + for (i, d) in deltas.iter().enumerate() { + a += da.get(i).cloned().unwrap_or_default(); + b += db.get(i).cloned().unwrap_or_default(); + got += d; + prop_assert_eq!(&got, &join_ref(&a, &b), "[join] after step {}", i); + } + } + + #[test] + fn join_is_bilinear(a1 in kv(), a2 in kv(), b1 in kv(), b2 in kv()) { + let total = |s: &[Z<(u8, u8)>]| s.iter().fold(Z::new(), |acc, d| acc + d.clone()); + let (a1, a2, b1, b2) = (total(&a1), total(&a2), total(&b1), total(&b2)); + prop_assert_eq!(join_ref(&(&a1 + &a2), &b1), &join_ref(&a1, &b1) + &join_ref(&a2, &b1), "[bilinear-left]"); + prop_assert_eq!(join_ref(&a1, &(&b1 + &b2)), &join_ref(&a1, &b1) + &join_ref(&a1, &b2), "[bilinear-right]"); + // and the circuit's output is bilinear too: one step of a+a' equals two of a, a'. + let once = join_circuit(&[&a1 + &a2], &[b1.clone()]); + let twice = join_circuit(&[a1.clone(), a2.clone()], &[b1.clone(), Z::new()]); + let sum_once = once.into_iter().fold(Z::new(), |x, d| x + d); + let sum_twice = twice.into_iter().fold(Z::new(), |x, d| x + d); + prop_assert_eq!(sum_once, sum_twice, "[bilinear-circuit]"); + } + + #[test] + fn integrate_after_differentiate_is_the_identity(steps in stream()) { + prop_assert_eq!(integrate(&differentiate(&steps)), steps.clone(), "[slice]"); + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let d = c.differentiate(&s); + let back = c.integrate(&d); + let (sink_s, sink_back) = (c.output(&s), c.output(&back)); + let (mut seen_s, mut seen_back) = (Vec::new(), Vec::new()); + for d in &steps { + input.push(d.clone()); + c.step(); + seen_s.push(s.current()); + seen_back.push(back.current()); + } + let _ = (sink_s, sink_back); + prop_assert_eq!(seen_back, seen_s, "[circuit-i-d]"); + } + + #[test] + fn differentiate_after_integrate_is_the_identity(steps in stream()) { + prop_assert_eq!(differentiate(&integrate(&steps)), steps.clone(), "[slice]"); + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let i = c.integrate(&s); + let back = c.differentiate(&i); + for (n, d) in steps.iter().enumerate() { + input.push(d.clone()); + c.step(); + prop_assert_eq!(back.current(), d.clone(), "[circuit-d-i] step {}", n); + } + } + + #[test] + fn delay_is_the_previous_step(steps in stream()) { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let d = c.delay(&s); + let mut prev = Z::new(); + for (n, step) in steps.iter().enumerate() { + input.push(step.clone()); + c.step(); + prop_assert_eq!(d.current(), prev.clone(), "[delay] step {}", n); + prev = step.clone(); + } + } + + /// Whatever the deadline does, the rebuilt output is what one delta would give. + #[test] + fn a_resumable_rebuild_equals_an_uninterrupted_step( + snapshot in pairs(), + chunk in 0usize..4, + budget in 1usize..4, + ) { + let snapshot: Z = snapshot.into_iter().collect(); + let build = |c: &mut Circuit, s: &Stream>| { + let m = c.map(s, |k| k % 3); + c.distinct(&m) + }; + let mut whole = Circuit::new(); + let (input, s) = whole.input::(); + let out = build(&mut whole, &s); + let sink = whole.output(&out); + input.push(snapshot.clone()); + whole.step(); + let want = sink.take(); + + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let out = build(&mut c, &s); + let sink = c.output(&out); + let mut job = Rebuild::new(input, snapshot, chunk); + let mut got = Z::new(); + let mut calls = 0; + loop { + calls += 1; + prop_assert!(calls < 100, "[rebuild] never finished"); + let left = Cell::new(budget); + let progress = job.step(&mut c, || { + left.set(left.get() - 1); + left.get() == 0 + }); + got += sink.take(); + if progress == Progress::Done { break; } + } + prop_assert_eq!(got, want, "[rebuild]"); + } + + #[test] + fn a_trace_reads_what_was_applied(ops in prop::collection::vec(((0u8..5, 0u8..3), -3i64..=3), 0..12), lo in 0u8..6, hi in 0u8..6) { + let mut t = MemTrace::new(); + let mut reference: Z<(u8, u8)> = Z::new(); + for ((k, v), w) in &ops { + t.apply(k, v, *w); + reference.insert((*k, *v), *w); + } + for k in 0..5u8 { + let want: Vec<(u8, i64)> = reference.iter().filter(|((a, _), _)| *a == k).map(|((_, v), w)| (*v, w)).collect(); + prop_assert_eq!(t.lookup(&k), want, "[lookup] key {}", k); + } + let want: Vec<(u8, u8, i64)> = reference.iter().filter(|((a, _), _)| *a >= lo && *a < hi).map(|((a, v), w)| (*a, *v, w)).collect(); + prop_assert_eq!(t.range(Bound::Included(&lo), Bound::Excluded(&hi)), want, "[range]"); + let all: Vec<(u8, u8, i64)> = reference.iter().map(|((a, v), w)| (*a, *v, w)).collect(); + prop_assert_eq!(t.range(Bound::Unbounded, Bound::Unbounded), all, "[range-all]"); + prop_assert_eq!(t.keys(), reference.iter().map(|((a, _), _)| *a).collect::>().len()); + } +} + +#[test] +fn a_rebuild_that_is_out_of_time_reports_pending_after_one_chunk() { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let sink = c.output(&s); + let mut job = Rebuild::new(input, (0..6u8).map(|k| (k, 1)).collect(), 2); + assert_eq!(job.step(&mut c, || true), Progress::Pending); + assert_eq!(sink.take().len(), 2, "exactly one chunk ran"); + assert_eq!(job.step(&mut c, || false), Progress::Done); + assert_eq!(sink.take().len(), 4); + assert_eq!(job.step(&mut c, || true), Progress::Done); +} + +#[test] +fn an_unbounded_range_that_is_empty_reads_nothing() { + let mut t = MemTrace::new(); + t.apply(&1u8, &1u8, 1); + assert!(t.range(Bound::Excluded(&1), Bound::Included(&1)).is_empty()); + assert!(t.range(Bound::Included(&2), Bound::Included(&1)).is_empty()); + assert_eq!(t.range(Bound::Included(&1), Bound::Included(&1)).len(), 1); +} + +#[test] +fn the_join_reads_state_through_the_trace_it_was_given() { + struct Counting(MemTrace, Rc>); + impl Trace for Counting { + fn apply(&mut self, k: &u8, v: &u8, w: i64) { + self.0.apply(k, v, w) + } + fn lookup(&self, k: &u8) -> Vec<(u8, i64)> { + self.1.set(self.1.get() + 1); + self.0.lookup(k) + } + fn range(&self, lo: Bound<&u8>, hi: Bound<&u8>) -> Vec<(u8, u8, i64)> { + self.0.range(lo, hi) + } + } + let looked = Rc::new(Cell::new(0)); + let mut c = Circuit::new(); + let (ia, a) = c.input::<(u8, u8)>(); + let (ib, b) = c.input::<(u8, u8)>(); + let j = c.join_indexed_in( + &a, + &b, + Counting(MemTrace::new(), looked.clone()), + MemTrace::new(), + |k, x, y| (*k, *x, *y), + ); + ia.push(Z::singleton((1, 1), 1)); + ib.push(Z::singleton((1, 2), 1)); + c.step(); + assert_eq!(j.current(), Z::singleton((1, 1, 2), 1)); + assert!(looked.get() > 0); +} + +#[test] +fn aggregate_state_refolds_only_the_touched_keys() { + let mut st = AggregateState::new(MemTrace::::new()); + let folds = Cell::new(0); + let f = |_: &u8, vals: &[(u8, i64)]| { + folds.set(folds.get() + 1); + Some(vals.len() as i64) + }; + let big: Z<(u8, u8)> = (0..50u8).map(|k| ((k, 0), 1)).collect(); + st.step(&big, f); + folds.set(0); + 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)])); +} From a4dadae36616f4803b8c8bff375f0b55f1bb0252 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-circuit)" Date: Sat, 10 Oct 2026 11:45:56 -0700 Subject: [PATCH 2/5] docs(zset): the circuit's reasoning and its test evidence (improvise-bw4p.1) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/zset.md | 58 +++++++++++++++++++++++++- context/test-evidence/zset.md | 76 +++++++++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 1 deletion(-) diff --git a/context/code/zset.md b/context/code/zset.md index b3b551da..18928337 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -38,7 +38,7 @@ and which a `HashMap` would give up (`context/code/core.md`, gotcha 18's argumen The ADR is one paragraph on the crate, and says more about what it is for than what it is. These are the readings this step takes. -1. **Clause 1 lists a `Circuit` and more operators; this step builds the pure part.** The +1. **Clause 1 lists a `Circuit` and more operators; step 1 built the pure part, and "The circuit" below built the rest (improvise-bw4p.1).** The ADR's list is `delay`, `integrate`, `differentiate`, `map`, `filter`, `flat_map`, `index_by`, `join_indexed`, `distinct` and `aggregate`, run by a synchronous `Circuit`, plus resumable rebuilds and a trace interface. Step 1 as ticketed is the group and the @@ -108,3 +108,59 @@ Weights come from `-3..=3`, an explicit zero, and `±2^40`, so sums of three can The overflow policy is not a property over the generators (they cannot reach it by design) and is tested by example in `src/lib.rs`, with `should_panic(expected = ..)`. + +## The circuit (improvise-bw4p.1) + +Clause 1's deferred scope, built without ADR 008. The ADR is not edited; this is how the crate reads it. + +| item | what it is | +|---|---| +| `Circuit` | creation-order list of boxed nodes; `step()` runs each once | +| `Input`, `Stream>`, `Output` | push a delta in; the value on an edge; a sink that sums every step's delta until `take()` | +| `delay`, `integrate`, `differentiate`, `plus` | the stream operators as nodes | +| `map`, `filter`, `flat_map`, `index_by` | linear, per-element | +| `distinct` | incremental, state is the integrated input (`DistinctState`) | +| `join_indexed`, `join_indexed_in` | bilinear delta join of two `ZSet<(K, _)>` (`JoinState`) | +| `aggregate`, `aggregate_in` | group-by, re-folds only touched keys (`AggregateState`) | +| `Trace`, `MemTrace` | operator state behind lookup, range, apply | +| `Rebuild`, `Progress` | a snapshot loaded as a resumable job | + +Decisions, where the ADR left a choice: + +1. **Wiring is apart from logic.** `circuit.rs` is the only file with `Rc`, `RefCell` and + `Box`. `ops.rs` holds each stateful operator as a plain struct with a pure + `step(delta) -> delta`, testable and usable without a circuit. +2. **Scheduling is creation order.** A node is built from streams that already exist, so + creation order is a topological order, and the circuit has no cycles: the ADR has no + recursion operator, and `integrate` is its own stateful node, not a delayed feedback loop. + `delay` is a node with one held Z-set. Not `Send`; one thread per circuit (clause 1). +3. **Incremental means the integral of the output is the operator of the integral of the + input.** That is the property every operator is tested against, over random delta streams. + `distinct` and `aggregate` are not linear, so they keep the integrated input and emit the + difference of the operator at the touched keys only. +4. **`index_by` is `map` to `(key, element)`.** A keyed stream is a `ZSet<(K, V)>`; there is no + separate indexed collection type, so the join and the aggregate read the same shape. +5. **The join** is `da x b_prev + a_prev x db + da x db`, computed as `da` against the right + trace, `da` applied to the left, then `db` against the updated left (which holds the cross + term), then `db` applied. Output weights are products and panic on `i64` overflow like + every weight (interpretation 5). `f(&K, &A, &B) -> O` builds the output key. +6. **`aggregate`'s fold gets the key and the group's `(value, weight)` pairs** (owned, because + a paged trace cannot lend references) and returns `Option`; `None` means the group has no + row (an empty group, or a `Min` of nothing). Sum, count, min and max are closures; text + unanimity and the ADR's `ExactSum` belong to the caller (clause 4, 8), not this crate. +7. **The trace interface** is `apply(k, v, w)`, `lookup(k)` and `range(lo, hi)`, returning + 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`. +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`. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index c2890257..d2e03476 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -278,3 +278,79 @@ the result must equal the exact value when it fits `i64`, and the operation must does not. **Injected**, one at a time, `wrapping_add`, `wrapping_sub`, `wrapping_mul` and `wrapping_neg` in place of the checked calls: this test failed under every one, and the example tests named above failed under the matching one. + +## `crates/zset/tests/circuit.rs`: the circuit + +Each operator is checked against recomputing from scratch: a random stream of deltas goes in, +one circuit step each, and after every step the sum of the outputs so far must equal the +operator applied to the sum of the inputs so far. The failure message carries `[incremental]`, +`[join]`, `[rebuild]` and so on. Keys come from a range of six (four by three for the keyed +ones) and weights from `-3..=3`, so cancelling deltas and re-insertions are the common case. + +**Injected in the source**, one at a time, each reverted and the suite rerun. The failing +tests, by name (a build that printed `Running` was confirmed each time; the first attempt at +the `delay` fault did not compile and was redone): +- `DistinctState` does not update its held set: `distinct_matches_recompute`, + `a_composed_pipeline_matches_recompute`, `a_resumable_rebuild_equals_an_uninterrupted_step`; +- the join applies `da` to the left trace after reading it for `db` (drops the `da x db` + term): `join_matches_recompute`, `join_is_bilinear`, + `the_join_reads_state_through_the_trace_it_was_given`; +- the join drops `da x b_prev`: `join_matches_recompute`, `join_is_bilinear`; +- `aggregate` does not retract the old result: `aggregate_sum_matches_recompute`, + `aggregate_max_matches_recompute`, `aggregate_state_refolds_only_the_touched_keys`; +- `delay` passes the current input: `delay_is_the_previous_step`, and both identities; +- `integrate` forgets history (`acc = d`): `integrate_after_differentiate_is_the_identity`, + `differentiate_after_integrate_is_the_identity`; +- `Rebuild` stops after the first chunk: `a_rebuild_that_is_out_of_time_reports_pending_after_one_chunk`, + `a_resumable_rebuild_equals_an_uninterrupted_step`; +- `Rebuild` checks the deadline before running any chunk (no progress is possible): + the same two; +- `MemTrace::range` ignores the upper bound, and `MemTrace` keeps an emptied key: + `a_trace_reads_what_was_applied`; +- `flat_map` gives every image weight 1: `flat_map_matches_recompute`. + +### map_matches_recompute + +`map`, `filter`, `index_by` and the pipeline test below are linear or compositions of +linear and `distinct`; they pin the circuit's wiring (the input node, the output sink, +creation-order scheduling) as much as the operators, which are `ZSet` methods tested in +`properties.rs`. No fault in them separates from the wiring faults above, and none was +needed. + +### join_is_bilinear + +Three tags. `[bilinear-left]` and `[bilinear-right]` check the reference join is bilinear, so +the oracle itself is a legitimate one; `[bilinear-circuit]` checks the incremental join: one +step of `a1 + a2` and two steps `a1` then `a2` (same `b`) sum to the same output. A join that +drops the `da x db` term fails the first-step, one-delta case, which is why the cross term is +exercised at all. + +### a_resumable_rebuild_equals_an_uninterrupted_step + +The snapshot enters a pipeline (`map` then `distinct`) in chunks of 1 to 3, and the deadline +predicate expires after 1 to 3 chunks per call, so the job is interrupted at varied points. +The sum of the outputs must equal one uninterrupted step over the whole snapshot. The loop is +bounded at 100 calls, so a job that never finishes is the named failure `[rebuild]`, not a hang. + +### a_rebuild_that_is_out_of_time_reports_pending_after_one_chunk + +A predicate that is always expired still lets exactly one chunk through (two of six keys), +then `Pending`; the rest completes on the next call; an exhausted job answers `Done` even +when out of time. This is the progress guarantee the property cannot see, because its budget +is never zero. + +### aggregate_state_refolds_only_the_touched_keys + +Fifty groups are loaded, then one delta touches one key: `f` is called twice (before and +after), not once per group. It is the O(delta) claim, in the one place a count can show it. + +### an_unbounded_range_that_is_empty_reads_nothing + +`range(Excluded(1), Included(1))` and an inverted range are empty. `BTreeMap::range` panics +on those; the trace guards it first. Not from the property, whose `lo > hi` draws are +covered by the same guard, but the equal-bounds corner is rare there. + +### the_join_reads_state_through_the_trace_it_was_given + +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. From ba17f92bc915e0e6b27a7aef8889053e37ba9d54 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-circuit)" Date: Sat, 10 Oct 2026 11:48:21 -0700 Subject: [PATCH 3/5] style(zset): satisfy clippy and the comment cap (improvise-bw4p.1) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- crates/zset/src/circuit.rs | 5 ++--- crates/zset/tests/circuit.rs | 2 +- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs index ab388313..e1c3812e 100644 --- a/crates/zset/src/circuit.rs +++ b/crates/zset/src/circuit.rs @@ -324,9 +324,8 @@ impl Rebuild { } } - /// Runs chunks until the snapshot is consumed or `expired()` says stop. It always - /// runs at least one chunk when any remain, so each call makes progress. `expired` - /// is a predicate, not an `Instant`: `Instant::now()` panics on wasm32. + /// Runs chunks until done or `expired()`, at least one per call. A predicate, not an + /// `Instant`, which panics on wasm32 (`context/code/zset.md`). 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(); diff --git a/crates/zset/tests/circuit.rs b/crates/zset/tests/circuit.rs index 04eb0aa1..e7cff63f 100644 --- a/crates/zset/tests/circuit.rs +++ b/crates/zset/tests/circuit.rs @@ -193,7 +193,7 @@ proptest! { prop_assert_eq!(join_ref(&(&a1 + &a2), &b1), &join_ref(&a1, &b1) + &join_ref(&a2, &b1), "[bilinear-left]"); prop_assert_eq!(join_ref(&a1, &(&b1 + &b2)), &join_ref(&a1, &b1) + &join_ref(&a1, &b2), "[bilinear-right]"); // and the circuit's output is bilinear too: one step of a+a' equals two of a, a'. - let once = join_circuit(&[&a1 + &a2], &[b1.clone()]); + let once = join_circuit(&[&a1 + &a2], std::slice::from_ref(&b1)); let twice = join_circuit(&[a1.clone(), a2.clone()], &[b1.clone(), Z::new()]); let sum_once = once.into_iter().fold(Z::new(), |x, d| x + d); let sum_twice = twice.into_iter().fold(Z::new(), |x, d| x + d); From 0849981ae1bb424bb2c425926542a4d2d5ebddc1 Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-circuit)" Date: Sat, 10 Oct 2026 12:06:46 -0700 Subject: [PATCH 4/5] 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)), + ); } From 66898d81e482bef5ae2f87e4e924d71a6a1b50cb Mon Sep 17 00:00:00 2001 From: "Claude (dbsp-circuit)" Date: Sat, 10 Oct 2026 12:31:06 -0700 Subject: [PATCH 5/5] fix(zset): current() is the whole tick, a split rebuild tick is refused, later ticks are tested (improvise-bw4p.1) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze --- context/code/zset.md | 14 ++ context/test-evidence/zset.md | 34 ++++ crates/zset/src/circuit.rs | 147 ++++++++++++----- crates/zset/tests/circuit.rs | 8 +- .../zset/tests/rebuild.proptest-regressions | 7 + crates/zset/tests/rebuild.rs | 156 ++++++++++++++++++ 6 files changed, 327 insertions(+), 39 deletions(-) create mode 100644 crates/zset/tests/rebuild.proptest-regressions create mode 100644 crates/zset/tests/rebuild.rs diff --git a/context/code/zset.md b/context/code/zset.md index 227106dd..4c1380cb 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -165,6 +165,20 @@ Decisions, where the ADR left a choice: 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. + + **`current()` is the whole tick.** A stream keeps the latest piece, which downstream + operators read, and the whole value of the tick, which `Stream::current` reads: pieces + within one tick are absorbed into it, so `integrate.current()` after a chunked load is the + full integral and a continued `delay` still reads the previous tick's input. `Output` + sums pieces as before. + + **A pending rebuild owns the clock.** `Circuit::step` counts ticks, and a `Rebuild` records + the count after its first chunk. If a `Circuit::step` runs before the job is Done (the host + stepped between calls, or two jobs interleaved), the next `Rebuild::step` panics with + "the rebuild was interrupted"; `Rebuild::interrupted` asks without panicking, so a host + can refuse first. `step_continue` calls made by another job are not counted and are not + detected. The state a load leaves for the next tick is tested by pushing another delta + after Done and stepping twice. 9. **Not built:** `Circuit` does not own a job runner (ADR 008's), and nothing here is serialised (clause 9, `improvise-protocol`). diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index bb07b4c2..0089a2f0 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -395,3 +395,37 @@ phantom `(7, 0)` rows for the empty group, e.g. `{(7,0):-1, (7,1):1}` where `{(7 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/tests/rebuild.rs`: delta review of #345 (0849981a) + +**Injected**, one at a time: a continued `delay` that replaces what it holds fails +`every_operator_chunked_equals_one_step_and_so_do_the_ticks_after`; a continued `integrate` +that does not accumulate fails that and `a_chunked_temporal_rebuild_equals_one_step` (which +now pushes another delta after Done and steps twice, so the state left for later ticks is +compared); a `current()` that reads the last piece fails +`current_after_a_chunked_rebuild_is_the_whole_ticks_value`; an undetected split tick, or a +`step` that does not count ticks, fails both should-panic tests below. Before these, the +first two faults passed the whole suite. + +### every_operator_chunked_equals_one_step_and_so_do_the_ticks_after + +One pipeline with every public operator (delay, integrate, differentiate, index_by, join, +aggregate, distinct, flat_map), a prior tick, a chunked load, then two more ticks. The +load tick `[rebuild-tick]`, the next tick `[next-tick]` and the one after `[tick-after]` must +equal the single-step run. From the reviewer's property. + +### current_after_a_chunked_rebuild_is_the_whole_ticks_value + +`integrate.current()` after a load in chunks of one equals the one-chunk load. + +### a_host_step_between_rebuild_calls_is_refused_loudly + +A `Circuit::step` between two `Rebuild::step` calls makes the second panic. + +### two_interleaved_rebuilds_are_refused_loudly + +A second job's first chunk is a tick, so the first job's next call panics. + +### a_finished_rebuild_and_an_interrupted_check_do_not_trip_on_later_ticks + +A finished job is never interrupted, and steps afterwards do not make it panic. diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs index 6c970ba8..39ee55fd 100644 --- a/crates/zset/src/circuit.rs +++ b/crates/zset/src/circuit.rs @@ -6,8 +6,55 @@ use crate::{MemTrace, Trace, ZSet}; use std::cell::RefCell; use std::rc::Rc; +/// What an edge carries: the latest piece, and the whole value of the current tick. +/// They differ only while a tick arrives in pieces (`Circuit::step_continue`). +pub(crate) struct Slot { + piece: T, + whole: T, +} + +impl Default for Slot { + fn default() -> Self { + Slot { + piece: T::default(), + whole: T::default(), + } + } +} + +/// Values whose pieces of one tick add up to the tick's value. +pub trait Piecewise: Default + Clone { + fn absorb(&mut self, piece: &Self); +} + +impl Piecewise for () { + fn absorb(&mut self, _: &Self) {} +} + +impl Piecewise for ZSet { + fn absorb(&mut self, piece: &Self) { + *self += piece; + } +} + /// The value an edge carries during the current step. -pub struct Stream(Rc>); +pub struct Stream(Rc>>); + +impl Stream { + fn piece(&self) -> T { + self.0.borrow().piece.clone() + } + + fn publish(&self, v: T, cont: bool) { + let mut slot = self.0.borrow_mut(); + if cont { + slot.whole.absorb(&v); + } else { + slot.whole = v.clone(); + } + slot.piece = v; + } +} impl Clone for Stream { fn clone(&self) -> Self { @@ -16,9 +63,9 @@ impl Clone for Stream { } impl Stream { - /// The value of the last completed step (the zero before the first). + /// The value of the last tick, all its pieces together (the zero before the first). pub fn current(&self) -> T { - self.0.borrow().clone() + self.0.borrow().whole.clone() } } @@ -57,10 +104,10 @@ struct Unary { f: F, } -impl O> Node for Unary { - fn eval(&mut self, _cont: bool) { - let v = (self.f)(&self.src.0.borrow()); - *self.out.0.borrow_mut() = v; +impl O> Node for Unary { + fn eval(&mut self, cont: bool) { + let v = (self.f)(&self.src.piece()); + self.out.publish(v, cont); } } @@ -71,10 +118,10 @@ struct Binary { f: F, } -impl O> Node for Binary { - fn eval(&mut self, _cont: bool) { - let v = (self.f)(&self.a.0.borrow(), &self.b.0.borrow()); - *self.out.0.borrow_mut() = v; +impl O> Node for Binary { + fn eval(&mut self, cont: bool) { + let v = (self.f)(&self.a.piece(), &self.b.piece()); + self.out.publish(v, cont); } } @@ -87,14 +134,14 @@ struct Delay { impl Node for Delay { fn eval(&mut self, cont: bool) { - let d = self.src.0.borrow(); + let d = self.src.piece(); let v = if cont { - self.held += &*d; + self.held += &d; ZSet::new() } else { - std::mem::replace(&mut self.held, d.clone()) + std::mem::replace(&mut self.held, d) }; - *self.out.0.borrow_mut() = v; + self.out.publish(v, cont); } } @@ -107,9 +154,10 @@ struct Integrate { 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() }; + let d = self.src.piece(); + self.acc += &d; + let v = if cont { d } else { self.acc.clone() }; + self.out.publish(v, cont); } } @@ -119,9 +167,10 @@ struct Source { out: Stream>, } -impl Node for Source { - fn eval(&mut self, _cont: bool) { - *self.out.0.borrow_mut() = std::mem::take(&mut *self.pending.borrow_mut()); +impl Node for Source { + fn eval(&mut self, cont: bool) { + let v = std::mem::take(&mut *self.pending.borrow_mut()); + self.out.publish(v, cont); } } @@ -130,6 +179,7 @@ impl Node for Source { #[derive(Default)] pub struct Circuit { nodes: Vec>, + ticks: u64, } type Z = ZSet; @@ -141,9 +191,15 @@ impl Circuit { /// Runs one clock tick: every node, once, in order. pub fn step(&mut self) { + self.ticks += 1; self.run(false); } + /// How many ticks `step` has begun. A `Rebuild` uses it to see a tick split. + pub fn ticks(&self) -> u64 { + self.ticks + } + /// 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) { @@ -156,7 +212,7 @@ impl Circuit { } } - fn unary( + fn unary( &mut self, src: &Stream, f: impl FnMut(&I) -> O + 'static, @@ -170,7 +226,7 @@ impl Circuit { out } - fn binary( + fn binary( &mut self, a: &Stream, b: &Stream, @@ -186,7 +242,7 @@ impl Circuit { out } - pub fn input(&mut self) -> (Input, Stream>) { + pub fn input(&mut self) -> (Input, Stream>) { let pending: Rc>> = Rc::default(); let out = Stream(Rc::default()); self.nodes.push(Box::new(Source { @@ -241,7 +297,7 @@ impl Circuit { self.binary(a, b, |a: &Z, b: &Z| a + b) } - pub fn map( + pub fn map( &mut self, s: &Stream>, mut f: impl FnMut(&K) -> L + 'static, @@ -258,7 +314,11 @@ impl Circuit { } /// Each key becomes any number of images, each with the key's weight. - pub fn flat_map>( + pub fn flat_map< + K: Ord + Clone + 'static, + L: Ord + Clone + 'static, + I: IntoIterator, + >( &mut self, s: &Stream>, mut f: impl FnMut(&K) -> I + 'static, @@ -267,7 +327,7 @@ impl Circuit { } /// Keys each element by `key`, giving the `(key, element)` pairs a join reads. - pub fn index_by( + pub fn index_by( &mut self, s: &Stream>, mut key: impl FnMut(&K) -> G + 'static, @@ -292,7 +352,7 @@ impl Circuit { K: Ord + Clone + 'static, A: Ord + Clone + 'static, B: Ord + Clone + 'static, - O: Ord + 'static, + O: Ord + Clone + 'static, { self.join_indexed_in(a, b, MemTrace::new(), MemTrace::new(), f) } @@ -310,7 +370,7 @@ impl Circuit { K: Ord + Clone + 'static, A: Ord + Clone + 'static, B: Ord + Clone + 'static, - O: Ord + 'static, + O: Ord + Clone + 'static, TA: Trace + 'static, TB: Trace + 'static, { @@ -329,7 +389,7 @@ impl Circuit { where K: Ord + Clone + 'static, V: Ord + Clone + 'static, - R: Ord + 'static, + R: Ord + Clone + 'static, { self.aggregate_in(s, MemTrace::new(), f) } @@ -344,7 +404,7 @@ impl Circuit { where K: Ord + Clone + 'static, V: Ord + Clone + 'static, - R: Ord + 'static, + R: Ord + Clone + 'static, T: Trace + 'static, { let mut st = AggregateState::new(groups); @@ -359,13 +419,14 @@ pub enum Progress { Done, } -/// 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`). +/// 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. pub struct Rebuild { input: Input, rest: std::collections::btree_map::IntoIter, chunk: usize, - started: bool, + /// The circuit's tick count right after this job's first chunk, once it has run one. + began_at: Option, } impl Rebuild { @@ -375,24 +436,34 @@ impl Rebuild { input, rest: snapshot.into_iter(), chunk: chunk.max(1), - started: false, + began_at: None, } } + /// Whether a `Circuit::step` since this job began has split its tick. A finished job + /// cannot be interrupted. + pub fn interrupted(&self, circuit: &Circuit) -> bool { + self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks()) + } + /// 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.interrupted(circuit), + "zset: the rebuild was interrupted by a Circuit::step before it finished" + ); loop { let chunk: ZSet = self.rest.by_ref().take(self.chunk).collect(); - if chunk.is_zero() && self.started { + if chunk.is_zero() && self.began_at.is_some() { return Progress::Done; } self.input.push(chunk); - if self.started { + if self.began_at.is_some() { circuit.step_continue(); } else { circuit.step(); - self.started = true; + self.began_at = Some(circuit.ticks()); } if self.rest.len() == 0 { return Progress::Done; diff --git a/crates/zset/tests/circuit.rs b/crates/zset/tests/circuit.rs index 57cf168b..7d38b1fd 100644 --- a/crates/zset/tests/circuit.rs +++ b/crates/zset/tests/circuit.rs @@ -463,12 +463,18 @@ proptest! { input.push(prior.clone()); c.step(); let _ = sink.take(); + let input_again = input.clone(); 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() + let at = sink.take(); + input_again.push(Z::singleton(5, 1)); + c.step(); + let next = sink.take(); + c.step(); + (at, next, sink.take()) }; prop_assert_eq!(run(chunk, budget), run(100, 100), "[temporal-rebuild]"); } diff --git a/crates/zset/tests/rebuild.proptest-regressions b/crates/zset/tests/rebuild.proptest-regressions new file mode 100644 index 00000000..8f50472b --- /dev/null +++ b/crates/zset/tests/rebuild.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 b276367684808122e5f3fedb8e87a6ab970895f3efce9e6b98a40ade2cdb7227 # shrinks to prior = ZSet { weights: {} }, snap = ZSet { weights: {0: -1, 1: -1} }, post = ZSet { weights: {} }, chunk = 0, budget = 1 diff --git a/crates/zset/tests/rebuild.rs b/crates/zset/tests/rebuild.rs new file mode 100644 index 00000000..cb549941 --- /dev/null +++ b/crates/zset/tests/rebuild.rs @@ -0,0 +1,156 @@ +//! A rebuild across every public operator: the tick it loads, the ticks after it, and `current()`. +//! Evidence: `context/test-evidence/zset.md`. + +use proptest::prelude::*; +use std::cell::Cell; +use zset::{Circuit, Input, Progress, Rebuild, ZSet}; + +type Z = ZSet; + +fn pairs() -> impl Strategy> { + prop::collection::vec((0u8..6, -3i64..=3), 0..6).prop_map(|p| p.into_iter().collect()) +} + +/// A pipeline using every operator; returns the input and a closure that drains all sinks. +fn build(c: &mut Circuit) -> (Input, Box String>) { + let (input, s) = c.input::(); + let d = c.delay(&s); + let i = c.integrate(&s); + let df = c.differentiate(&s); + let a = c.index_by(&s, |k| k % 2); + let b = c.index_by(&i, |k| k % 2); + let j = c.join_indexed(&a, &b, |k, x, y| (*k, *x, *y)); + let ij = c.integrate(&j); + let g = c.index_by(&df, |k| k % 3); + let agg = c.aggregate(&g, |_, vals| { + Some(vals.iter().map(|(_, w)| *w).sum::()) + }); + let dagg = c.delay(&agg); + let dist = c.distinct(&i); + let ddist = c.differentiate(&dist); + let fm = c.flat_map(&d, |k| vec![*k, k + 1]); + let sinks = ( + c.output(&d), + c.output(&i), + c.output(&df), + c.output(&j), + c.output(&ij), + c.output(&agg), + c.output(&dagg), + c.output(&ddist), + c.output(&fm), + ); + let drain = move || { + format!( + "d={:?} i={:?} df={:?} j={:?} ij={:?} agg={:?} dagg={:?} ddist={:?} fm={:?}", + sinks.0.take(), + sinks.1.take(), + sinks.2.take(), + sinks.3.take(), + sinks.4.take(), + sinks.5.take(), + sinks.6.take(), + sinks.7.take(), + sinks.8.take() + ) + }; + (input, Box::new(drain)) +} + +fn run( + prior: &Z, + snap: &Z, + post: &Z, + chunk: usize, + budget: usize, +) -> (String, String, String) { + let mut c = Circuit::new(); + let (input, drain) = build(&mut c); + input.push(prior.clone()); + c.step(); + let _ = drain(); + let mut job = Rebuild::new(input.clone(), snap.clone(), chunk); + let mut done = false; + for _ in 0..50 { + let left = Cell::new(budget); + if job.step(&mut c, || { + left.set(left.get() - 1); + left.get() == 0 + }) == Progress::Done + { + done = true; + break; + } + } + assert!(done, "[rebuild] never finished"); + let at = drain(); + input.push(post.clone()); + c.step(); + let next = drain(); + c.step(); + (at, next, drain()) +} + +proptest! { + #[test] + fn every_operator_chunked_equals_one_step_and_so_do_the_ticks_after( + prior in pairs(), snap in pairs(), post in pairs(), chunk in 0usize..4, budget in 1usize..4, + ) { + let want = run(&prior, &snap, &post, 100, 100); + let got = run(&prior, &snap, &post, chunk, budget); + prop_assert_eq!(&got.0, &want.0, "[rebuild-tick]"); + prop_assert_eq!(&got.1, &want.1, "[next-tick]"); + prop_assert_eq!(&got.2, &want.2, "[tick-after]"); + } +} + +#[test] +fn current_after_a_chunked_rebuild_is_the_whole_ticks_value() { + let snap: Z = [(1, 1), (2, 1)].into_iter().collect(); + let cur = |chunk| { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let i = c.integrate(&s); + let mut job = Rebuild::new(input, snap.clone(), chunk); + while job.step(&mut c, || false) == Progress::Pending {} + i.current() + }; + assert_eq!(cur(1), cur(10), "[current]"); +} + +#[test] +#[should_panic(expected = "rebuild was interrupted")] +fn a_host_step_between_rebuild_calls_is_refused_loudly() { + let snap: Z = [(1, 1), (2, 1)].into_iter().collect(); + let mut c = Circuit::new(); + let (input, _s) = c.input::(); + let mut job = Rebuild::new(input, snap, 1); + assert_eq!(job.step(&mut c, || true), Progress::Pending); + c.step(); + job.step(&mut c, || false); +} + +#[test] +#[should_panic(expected = "rebuild was interrupted")] +fn two_interleaved_rebuilds_are_refused_loudly() { + let snap: Z = [(1, 1), (2, 1)].into_iter().collect(); + let mut c = Circuit::new(); + let (i1, _s1) = c.input::(); + let (i2, _s2) = c.input::(); + let (mut a, mut b) = (Rebuild::new(i1, snap.clone(), 1), Rebuild::new(i2, snap, 1)); + assert_eq!(a.step(&mut c, || true), Progress::Pending); + assert_eq!(b.step(&mut c, || true), Progress::Pending); + a.step(&mut c, || false); +} + +#[test] +fn a_finished_rebuild_and_an_interrupted_check_do_not_trip_on_later_ticks() { + let snap: Z = [(1, 1)].into_iter().collect(); + let mut c = Circuit::new(); + let (input, _s) = c.input::(); + let mut job = Rebuild::new(input, snap, 1); + assert_eq!(job.step(&mut c, || true), Progress::Done); + c.step(); + assert!(!job.interrupted(&c)); + assert_eq!(job.step(&mut c, || true), Progress::Done); +}