From 4b8fd08f5839edfb63a2dfdcc6b911f1d53eb10a Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 5 Jul 2026 21:51:12 -0400 Subject: [PATCH] operators: document and assert the ReduceTactic / JoinTactic contracts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reduce and join tactic drivers rely on contracts their tactics were only implicitly honoring. Write them down where implementors will see them, and cheaply enforce the driver-checkable ones so a misbehaving tactic fails loudly instead of silently corrupting the output trace or dropping work. ReduceTactic::retire — document the contract and, since the output is produced in order, check the two cheap invariants in `reduce_with_tactic` with a linear scan: - output batches are ordered and tile `[lower, upper)` (first lower is `lower`, each upper meets the next lower, last upper is `upper`); - each batch ships at a time the operator holds a capability for. The "no work => empty frontier" rule (whose violation deadlocks recursive scopes) stays a documented, tactic-self-enforced invariant: the withheld set is tactic-internal, so it is not driver-checkable, but deriving the frontier from the actual pending set gives it for free. JoinTactic::work — document the fuel protocol (non-negative on return iff all work is exhausted; the driver reschedules iff negative). Not driver-checkable either; the in-tree tactic already makes it structural by setting `fuel` from whether its queues are empty. Contracts + debug asserts only; no behavior change. reduce, reduce_reference, join, bfs, scc green with the asserts active — the cursor and reference tactics already conform. Co-Authored-By: Claude Opus 4.8 (1M context) --- differential-dataflow/src/operators/join.rs | 8 +++- differential-dataflow/src/operators/reduce.rs | 40 +++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs index b2771432c..72b2085e9 100644 --- a/differential-dataflow/src/operators/join.rs +++ b/differential-dataflow/src/operators/join.rs @@ -61,7 +61,13 @@ pub(crate) trait JoinTactic, C fn defer(&mut self, input0: Vec, input1: Vec, fresh: Fresh, capability: Capability); /// Perform an amount of work that just barely exceeds `fuel`, which is decremented. /// - /// Returning with a non-negative fuel indicates that all work was exhausted. + /// **Fuel protocol.** On return, `fuel` is non-negative *iff* all outstanding work is exhausted, + /// and left negative exactly when work remains. The driver ([`join_with_tactic`]) reschedules the + /// operator iff `fuel < 0`, so returning non-negative with work still queued silently drops it + /// (and returning negative with nothing to do spins). The protocol is not driver-checkable — the + /// work-remaining state is tactic-internal — so derive the sign *from* that state rather than + /// tracking it separately, and it holds by construction (as the in-tree tactic does, setting + /// `fuel` from whether its queues are empty). fn work(&mut self, fuel: &mut isize, output: &mut OutputBuilderSession>); } diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs index 55210ace4..efb30af5a 100644 --- a/differential-dataflow/src/operators/reduce.rs +++ b/differential-dataflow/src/operators/reduce.rs @@ -36,6 +36,26 @@ pub(crate) trait ReduceTactic> /// the new input batches, and `held`: the times the operator currently holds capabilities for. It /// reasons only about times, returning the output batches to ship — each tagged with the time at /// which to ship it — and the new frontier of interesting times for the operator to hold. + /// + /// # Contract + /// + /// The driver ([`reduce_with_tactic`]) relies on the following; the first two are cheap to check + /// and are `debug_assert!`ed there. + /// + /// * **Ordered, tiling output.** The returned `(time, batch)` pairs are in ascending order and + /// their descriptions *tile* `[lower, upper)`: the first batch's lower is `lower`, each batch's + /// upper is the next batch's lower, and the last batch's upper is `upper` — no gaps, no overlaps. + /// Sub-intervals with no updates are skipped; the next batch's lower simply picks up where the + /// last left off. Producing *in order* is a requirement, not a convenience — it is what lets the + /// driver check the tiling with a single linear scan. + /// * **Shipped at a held time.** Each batch's `time` tag is an element of `held`; the driver mints + /// a capability at it, which is only valid for a held time. + /// * **Frontier bounds withheld work, and collapses to empty when there is none.** The returned + /// frontier must be at-or-below every time the tactic defers, so the driver knows what is safe to + /// release. In particular, with no work to defer it must be the *empty* antichain. Derive it from + /// the actual withheld set rather than constructing it and this holds for free; returning a + /// non-empty frontier with nothing pending holds capabilities forever and **deadlocks recursive + /// scopes**. (Not driver-checkable — the withheld set is tactic-internal — so tactics self-enforce.) fn retire( &mut self, source_batches: Vec, @@ -169,6 +189,26 @@ where // each tagged with the time to ship it at, and the new frontier of interesting times. let (produced, new_frontier) = tactic.retire(source_batches, output_batches, batch_storage, &lower_limit, &upper_limit, &held); + // Contract checks (see `ReduceTactic::retire`). Cheap, debug-only. + debug_assert!( + produced.iter().all(|(time, _)| held.elements().contains(time)), + "ReduceTactic::retire shipped a batch at a time not held as a capability", + ); + debug_assert!( + { + // Ordered output makes tiling a single linear scan: each description's lower + // must meet the previous upper (starting at `lower_limit`), ending at `upper_limit`. + let mut edge = lower_limit.clone(); + let abutting = produced.iter().all(|(_, batch)| { + let matches = batch.description().lower() == &edge; + edge.clone_from(batch.description().upper()); + matches + }); + abutting && (produced.is_empty() || edge == upper_limit) + }, + "ReduceTactic::retire output must be ordered and tile [lower, upper)", + ); + // Ship each batch at a capability minted from the set at its time, and commit it to the // output trace. The times are elements of `held`, so they stay valid until we downgrade. for (time, batch) in produced {