operators: document and assert the ReduceTactic / JoinTactic contracts

differential-dataflow/src/operators/join.rs

@@ -61,7 +61,13 @@ pub(crate) trait JoinTactic<B0: BatchReader, B1: BatchReader<Time = B0::Time>, C
     fn defer(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, capability: Capability<B0::Time>);
     /// 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<B0::Time, EffortBuilder<CB>>);
 }
 

differential-dataflow/src/operators/reduce.rs

@@ -36,6 +36,26 @@
     /// 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<B1>,
@@ -169,6 +189,26 @@
                     // 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 {
@@ -505,13 +545,13 @@
                 }
             }
             #[inline(never)]
             pub fn compute<'a, K, C1, C2, C3, L>(
                 &mut self,
                 key: K,
                 (source_cursor, source_storage): (&mut C1, &'a C1::Storage),
                 (output_cursor, output_storage): (&mut C2, &'a C2::Storage),
                 (batch_cursor, batch_storage): (&mut C3, &'a C3::Storage),
                 times: &Vec<T>,
                 logic: &mut L,
                 upper_limit: &Antichain<T>,
                 outputs: &mut [(T, Vec<(V, T, D2)>)],