From 9125098bce3f05f16e2423af934297cda182d5c9 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Tue, 11 Aug 2026 09:24:20 -0400
Subject: [PATCH 1/3] Fuse discovery and evaluation in the proxy reduce tactic
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The tactic enumerated each key's interesting times and then walked them a second
time to evaluate. The conventional reduce does neither: it runs one ascending
pass and evaluates as it discovers. It can because discovery never looks
backwards — every synthesized time is `next_time.join(t)` for a `t` not at or
below `next_time`, so it is strictly greater, and new work only ever lands ahead
of the sweep.

`Sweep` is that pass, cut where the conventional operator calls user logic. Each
`next_crossing` runs to the next in-interval time needing evaluation and
suspends with the buffers positioned to read the accumulations; the caller
evaluates and returns the corrections through `commit`; the next call resumes.
The tactic runs every live key to its next suspension and crosses them together,
which is what a batched backend wants, and no key enumerates its times first.

Written as a state machine rather than one body: `frontier` finds the position,
`absorb` steps the sources and decides whether the time is reached, `close`
joins forward, `settle` recomputes the meet. One tick reports `Passed`,
`Pended`, `Crossing(t)` or `Done`. That puts each clause of
`RoundCoverage.lean`'s `round_coverage` in one place — the novel witness is the
whole of `absorb`'s return value, and it is also why `close` is split, going
against the novel times always but against the prior times only for a time that
already carries a witness.

`discover_times` and the per-key moment lists go with it; so does the second
ordering of every key's records.

Measured against the two-phase schedule, on the reference backend as fixed by
the commits below it:

  churn        0.64x -> 0.58x of the cursor tactic
  multimoment  1.22x -> 1.09x
  propagate    1.20x -> 1.13x

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/src/operators/common.rs | 368 ++++++++++++++++++
 .../src/operators/int_proxy/reduce.rs         | 229 ++++-------
 2 files changed, 449 insertions(+), 148 deletions(-)

diff --git a/differential-dataflow/src/operators/common.rs b/differential-dataflow/src/operators/common.rs
index 069846f12..8c329e946 100644
--- a/differential-dataflow/src/operators/common.rs
+++ b/differential-dataflow/src/operators/common.rs
@@ -378,3 +378,371 @@ pub fn discover_times<T, RIn>(
     }
     sort_dedup(pended);
 }
+
+/// A resumable, fused determination-and-evaluation sweep over one key's times.
+///
+/// [`discover_times`] enumerates a key's interesting times up front, and the caller then walks them
+/// again to evaluate. The conventional reduce does not: it runs ONE ascending pass and evaluates as
+/// it discovers. It can, because discovery never looks backwards — every synthesized time is
+/// `next_time.join(t)` for some `t` NOT at or below `next_time`, so it is strictly greater, and new
+/// work only ever lands ahead of the sweep.
+///
+/// This is that pass, cut at the point where the conventional operator would call user logic. Each
+/// [`next_crossing`](Self::next_crossing) returns the next in-interval time that needs evaluating,
+/// with the buffers positioned to read the accumulations; the caller evaluates and hands the
+/// corrections back through [`commit`](Self::commit); the next call resumes. Many keys can be run
+/// to their next crossing and evaluated together, which is what a batched backend wants, without
+/// any of them enumerating their times first.
+///
+/// The schedule is `formal/Differential/RoundCoverage.lean`'s `round_coverage`: a time carrying an
+/// output change lies in the join-closure of `prior ∪ novel` AND is at or above some novel time.
+/// The two clauses appear here as the `interesting` test and the split synthesis — see the comments
+/// at each.
+pub struct Sweep<T, RIn, ROut> {
+    /// The novel run: the only source that SEEDS interest, replayed unadvanced.
+    novel: ValueHistory<u64, T, RIn>,
+    /// The accumulated input and output: join partners, and the accumulations to evaluate over.
+    input: ValueHistory<u64, T, RIn>,
+    output: ValueHistory<u64, T, ROut>,
+    /// The key's due interesting times, ascending, with their suffix meets; `due_pos` consumes them.
+    due: Vec<T>,
+    due_meets: Vec<T>,
+    due_pos: usize,
+    /// Synthesized times not yet visited, sorted DESCENDING so `last()` is the least.
+    synth: Vec<T>,
+    /// The seed times reached so far, compacted by the running meet. They are the witnesses the
+    /// absorption test looks for, and the partners a close joins against; keeping them collapsed is
+    /// what stops a key with many reached times rescanning all of them.
+    reached: Vec<T>,
+    /// Scratch for one step's synthesized times.
+    temporary: Vec<T>,
+    /// Corrections emitted so far this sweep, meet-collapsed; both a join partner and part of the
+    /// output accumulation.
+    produced: Vec<((u64, T), ROut)>,
+    /// The meet of every time still to come.
+    meet: Option<T>,
+    /// Whether the last `next_crossing` returned a time whose step is not yet settled.
+    suspended: bool,
+}
+
+/// What one [`tick`](Sweep::tick) decided about the time it visited.
+enum Tick<T> {
+    /// No seed reaches this time; the sweep moved past it.
+    Passed,
+    /// Reached, but at or beyond `upper`: carried to a later round rather than evaluated.
+    Pended,
+    /// Reached and in the interval. The caller must evaluate here before the sweep goes on.
+    Crossing(T),
+    /// Every source is drained.
+    Done,
+}
+
+impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Sweep<T, RIn, ROut> {
+    /// An empty sweep, to be `load`ed. Reuse one per key rather than allocating per key.
+    pub fn new() -> Self {
+        Sweep {
+            novel: ValueHistory::new(), input: ValueHistory::new(), output: ValueHistory::new(),
+            due: Vec::new(), due_meets: Vec::new(), due_pos: 0,
+            synth: Vec::new(), reached: Vec::new(), temporary: Vec::new(),
+            produced: Vec::new(), meet: None, suspended: false,
+        }
+    }
+
+    /// Position the sweep at the start of one key.
+    ///
+    /// `novel` must be the batch's own `(id, time, diff)` support for the key and `input` the
+    /// accumulated input WITHOUT it: the two are never merged, because consolidating them can cancel
+    /// a novel update against a compacted history record and lose its interesting time.
+    pub fn load(
+        &mut self,
+        novel: impl Iterator<Item = (u64, T, RIn)>,
+        input: impl Iterator<Item = (u64, T, RIn)>,
+        output: impl Iterator<Item = (u64, T, ROut)>,
+        due: &[T],
+    ) {
+        self.due.clear();
+        self.due.extend(due.iter().cloned());
+        self.due_meets.clear();
+        self.due_meets.extend(due.iter().cloned());
+        for i in (1..self.due_meets.len()).rev() {
+            let (init, tail) = self.due_meets.split_at_mut(i);
+            init[i - 1].meet_assign(&tail[0]);
+        }
+        self.due_pos = 0;
+        self.synth.clear();
+        self.reached.clear();
+        self.temporary.clear();
+        self.produced.clear();
+        self.suspended = false;
+
+        // The novel run is loaded UNADVANCED: it is the seed, and its own times are what the
+        // schedule is stated over. Only then is the meet known, and the join partners advanced by it.
+        self.novel.load_iter(novel, None);
+        let mut meet: Option<T> = None;
+        update_meet(&mut meet, self.due_meets.first());
+        update_meet(&mut meet, self.novel.meet());
+        self.input.load_iter(input, meet.as_ref());
+        self.output.load_iter(output, meet.as_ref());
+        self.meet = meet;
+    }
+
+    /// Advance to the next in-interval time that needs evaluating, or `None` once the key is spent.
+    ///
+    /// Times at or beyond `upper` that the schedule reaches are appended to `pended` for the caller
+    /// to carry into a later round.
+    pub fn next_crossing(&mut self, upper: &Antichain<T>, pended: &mut Vec<T>) -> Option<T> {
+        loop {
+            // A crossing leaves its step half-finished, because `settle` must see the corrections
+            // the caller commits. Finishing it is the first thing the next call does.
+            if self.suspended {
+                self.suspended = false;
+                self.settle();
+            }
+            match self.tick(upper, pended) {
+                Tick::Done => return None,
+                Tick::Crossing(at) => {
+                    self.suspended = true;
+                    return Some(at);
+                }
+                Tick::Passed | Tick::Pended => {}
+            }
+        }
+    }
+
+    /// Visit one time: find it, decide whether it is reached, close it forward, and report.
+    fn tick(&mut self, upper: &Antichain<T>, pended: &mut Vec<T>) -> Tick<T> {
+        let Some(at) = self.frontier() else { return Tick::Done };
+        let reached = self.absorb(&at);
+        if upper.less_equal(&at) {
+            // Out of the interval: nothing can be emitted here, so there is nothing to close
+            // against either — a join with `at` is at or beyond `at`, hence also out of interval,
+            // and will be rediscovered from `at` in the round that admits it.
+            self.settle();
+            if reached { pended.push(at); return Tick::Pended; }
+            return Tick::Passed;
+        }
+        self.close(&at, reached, upper, pended);
+        if reached { return Tick::Crossing(at); }
+        self.settle();
+        Tick::Passed
+    }
+
+    /// The sweep's position: the least time any source still offers.
+    ///
+    /// The TOTAL order, not the partial one. Every time `close` produces is strictly greater than
+    /// the position that produced it, so new work only ever lands ahead of here and the sweep never
+    /// revisits.
+    fn frontier(&self) -> Option<T> {
+        [
+            self.novel.time(), self.due.get(self.due_pos), self.input.time(),
+            self.output.time(), self.synth.last(),
+        ].into_iter().flatten().min().cloned()
+    }
+
+    /// Step every source sitting at `at`, and decide whether `at` is REACHED.
+    ///
+    /// Reached is clause two of `round_coverage` — `∃ nu ∈ novel, nu ≤ at` — evaluated
+    /// incrementally: either a seed lands exactly here, or one already stepped in lies below.
+    ///
+    /// Input and output are stepped whether or not `at` is reached, and that is forced rather than
+    /// eager: they are sources of the frontier, so leaving them would stall the sweep, and their
+    /// edits must reach the buffers or they are lost to every later accumulation. Stepping only
+    /// moves an edit across; it consolidates nothing. The expensive part — `advance_buffer_by`,
+    /// which joins every buffered time and re-consolidates — is deferred to `close`, and happens
+    /// only where the buffers are actually read.
+    fn absorb(&mut self, at: &T) -> bool {
+        self.input.step_while_time_is(at);
+        self.output.step_while_time_is(at);
+
+        // A novel update here is a seed.
+        let mut reached = self.novel.step_while_time_is(at);
+        if reached {
+            if let Some(meet) = self.meet.as_ref() { self.novel.advance_buffer_by(meet); }
+        }
+        // So is a synthetic join scheduled for here, or a due time carried in. Both move into the
+        // reached set, where they become witnesses and join partners for later times.
+        while self.synth.last() == Some(at) {
+            self.reached.push(self.synth.pop().expect("nonempty"));
+            reached = true;
+        }
+        while self.due.get(self.due_pos) == Some(at) {
+            self.reached.push(at.clone());
+            self.due_pos += 1;
+            reached = true;
+        }
+        // Absorption: a time at or above a seed already stepped in is itself reached, because
+        // joining that seed with it yields it back. Checked against the stepped prefixes only.
+        reached
+            || self.novel.buffer().iter().any(|((_, t), _)| t.less_equal(at))
+            || self.reached.iter().any(|t| t.less_equal(at))
+    }
+
+    /// Close `at` forward under joins — clause one of `round_coverage`, the join-closure.
+    ///
+    /// Against the NOVEL times always, reached or not: an unreached time joined with a novel time
+    /// lands at or above that novel time, so it carries a witness and is on the schedule.
+    ///
+    /// Against the PRIOR times only when `at` is itself reached, because the join then inherits
+    /// `at`'s witness. A join of two prior times carries none and is deliberately never produced —
+    /// that asymmetry is the whole of why an incremental operator does less work than the closure
+    /// of everything.
+    ///
+    /// `produced` counts as prior. A correction emitted at `p` changes the accumulated output at
+    /// every time at or above `p`, so `p ∨ at` has to be visited; nothing else covers it, since
+    /// this round's corrections are not in the output history and `at` was not yet stepped in when
+    /// the sweep passed `p`.
+    fn close(&mut self, at: &T, reached: bool, upper: &Antichain<T>, pended: &mut Vec<T>) {
+        self.temporary.extend(self.novel.buffer().iter().map(|((_, t), _)| t)
+            .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+        self.temporary.extend(self.reached.iter()
+            .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+        if reached {
+            if let Some(meet) = self.meet.as_ref() {
+                self.input.advance_buffer_by(meet);
+                self.output.advance_buffer_by(meet);
+            }
+            self.temporary.extend(self.input.buffer().iter().map(|((_, t), _)| t)
+                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+            self.temporary.extend(self.output.buffer().iter().map(|((_, t), _)| t)
+                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+            self.temporary.extend(self.produced.iter().map(|((_, t), _)| t)
+                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+        }
+        sort_dedup(&mut self.temporary);
+        let before = self.synth.len();
+        for time in self.temporary.drain(..) {
+            if upper.less_equal(&time) { pended.push(time); } else { self.synth.push(time); }
+        }
+        if self.synth.len() > before {
+            self.synth.sort_by(|x, y| y.cmp(x));
+            self.synth.dedup();
+        }
+    }
+
+    /// The input accumulation at the suspended time: both input runs, meeting only here.
+    pub fn input_at(&self, at: &T, into: &mut Vec<(u64, RIn)>) {
+        for ((id, time), diff) in self.input.buffer().iter().chain(self.novel.buffer().iter()) {
+            if time.less_equal(at) { into.push((*id, diff.clone())); }
+        }
+        crate::consolidation::consolidate(into);
+    }
+
+    /// The tentative output accumulation at the suspended time, including this sweep's corrections.
+    pub fn output_at(&self, at: &T, into: &mut Vec<(u64, ROut)>) {
+        for ((id, time), diff) in self.output.buffer().iter().chain(self.produced.iter()) {
+            if time.less_equal(at) { into.push((*id, diff.clone())); }
+        }
+        crate::consolidation::consolidate(into);
+    }
+
+    /// Record the corrections evaluated at the suspended time, and collapse them by the meet.
+    pub fn commit(&mut self, at: &T, corrections: impl Iterator<Item = (u64, ROut)>) {
+        let before = self.produced.len();
+        for (id, diff) in corrections { self.produced.push(((id, at.clone()), diff)); }
+        if self.produced.len() > before {
+            if let Some(meet) = self.meet.as_ref() {
+                for entry in self.produced.iter_mut() { (entry.0).1.join_assign(meet); }
+            }
+            crate::consolidation::consolidate(&mut self.produced);
+        }
+    }
+
+    /// Close a step: recompute the meet of everything still to come, and compact the reached set by
+    /// it. This is what keeps a key with a long history linear rather than quadratic.
+    fn settle(&mut self) {
+        let mut meet: Option<T> = None;
+        update_meet(&mut meet, self.novel.meet());
+        update_meet(&mut meet, self.input.meet());
+        update_meet(&mut meet, self.output.meet());
+        for time in self.synth.iter() { update_meet(&mut meet, Some(time)); }
+        update_meet(&mut meet, self.due_meets.get(self.due_pos));
+        if let Some(m) = meet.as_ref() {
+            for time in self.reached.iter_mut() { *time = time.join(m); }
+        }
+        sort_dedup(&mut self.reached);
+        self.meet = meet;
+    }
+}
+
+#[cfg(test)]
+mod sweep_tests {
+
+    use timely::order::Product;
+    use timely::progress::Antichain;
+
+    use super::{discover_times, DiscoverScratch, KeyView, Sweep};
+
+    type Time = Product<u64, u64>;
+
+    /// A deterministic LCG: enough variety without a dependency.
+    struct Rng(u64);
+    impl Rng {
+        fn next(&mut self, bound: u64) -> u64 {
+            self.0 = self.0.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
+            (self.0 >> 33) % bound
+        }
+    }
+
+    fn time(r: &mut Rng, span: u64) -> Time { Product::new(r.next(span), r.next(span)) }
+
+    /// `Sweep` must never reach a time `discover_times` does not: determination is the schedule,
+    /// and a crossing outside it would be work the coverage argument does not cover.
+    ///
+    /// The converse is NOT asserted, and must not be. The fused pass may legitimately explore FEWER
+    /// times, because it reaches each one carrying live accumulations: a time whose updates have
+    /// consolidated away by the time the sweep arrives needs no evaluation, where determination —
+    /// which reads times only — has no way to know that. Pinning equality here would forbid the
+    /// pruning that fusing the two passes exists to enable. Output equality against the conventional
+    /// reduce is the real check, and lives in the dataflow tests.
+    ///
+    /// As it happens the two coincide today, since both consolidate their buffers identically and
+    /// nothing is committed here; the containment is what is guaranteed.
+    #[test]
+    fn sweep_within_discover_times() {
+        for seed in 0..400u64 {
+            let mut rng = Rng(seed.wrapping_mul(2654435761).wrapping_add(12345));
+            let span = 2 + rng.next(4);
+
+            let novel: Vec<(u64, Time, i64)> =
+                (0..rng.next(6)).map(|_| (rng.next(3), time(&mut rng, span), 1i64)).collect();
+            let input: Vec<(u64, Time, i64)> =
+                (0..rng.next(6)).map(|_| (rng.next(3), time(&mut rng, span), 1i64)).collect();
+            let output: Vec<(u64, Time, i64)> =
+                (0..rng.next(4)).map(|_| (rng.next(3), time(&mut rng, span), 1i64)).collect();
+            let mut due: Vec<Time> = (0..rng.next(3)).map(|_| time(&mut rng, span)).collect();
+            due.sort();
+            due.dedup();
+            let upper = Antichain::from_elem(Product::new(span, span));
+
+            // Determination: `p_in` is the accumulated input, the seeds are the novel times.
+            let p_in: Vec<((u64, u64), Time, i64)> =
+                input.iter().map(|(v, t, d)| ((0u64, *v), *t, *d)).collect();
+            let mut scratch: DiscoverScratch<Time, i64> = DiscoverScratch::new();
+            let (mut moments, mut pended) = (Vec::new(), Vec::new());
+            discover_times(
+                KeyView { p_in: &p_in[..], i0: 0, i1: p_in.len(), pending: &due[..] },
+                novel.iter().map(|(_, t, _)| *t),
+                output.iter().map(|(_, t, _)| *t),
+                &upper,
+                &mut scratch,
+                &mut moments,
+                &mut pended,
+            );
+
+            // The fused sweep, committing nothing.
+            let mut sweep: Sweep<Time, i64, i64> = Sweep::new();
+            sweep.load(novel.iter().cloned(), input.iter().cloned(), output.iter().cloned(), &due[..]);
+            let (mut crossings, mut sweep_pended) = (Vec::new(), Vec::new());
+            while let Some(at) = sweep.next_crossing(&upper, &mut sweep_pended) {
+                crossings.push(at);
+            }
+            crate::operators::reduce::sort_dedup(&mut sweep_pended);
+
+            assert!(crossings.iter().all(|t| moments.contains(t)),
+                    "seed {seed}: sweep reached a time outside the schedule: {crossings:?} vs {moments:?}");
+            assert!(crossings.windows(2).all(|w| w[0] <= w[1]), "seed {seed}: crossings must ascend");
+            assert!(sweep_pended.iter().all(|t| pended.contains(t)),
+                    "seed {seed}: sweep pended a time outside the schedule");
+        }
+    }
+}
diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index 27668ae49..af29bb982 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -15,8 +15,7 @@ use crate::trace::{BatchReader, Description};
 use super::ProxyBridge;
 use crate::operators::reduce::ReduceTactic;
 
-use super::history::IdHistory;
-use crate::operators::common::{discover_times, tile_descriptions, DiscoverScratch, KeyView};
+use crate::operators::common::{tile_descriptions, Sweep};
 
 /// A unit of proxied reduce work, presented to the backend.
 pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>> {
@@ -206,10 +205,11 @@ where
         let mut from = Some(0u64);
         let mut window: ReduceWindow<B1::Time, Bk::RIn, Bk::ROut> = ReduceWindow::default();
 
-        // Retire-wide reusable scratch (cleared per window/round/moment, not reallocated). See the
-        // profiling note on `DiscoverScratch`: fresh per-key/per-round `Vec`s were the dominant cost.
-        let mut discover_scratch: DiscoverScratch<B1::Time, Bk::RIn> = DiscoverScratch::new();
-        let mut states: Vec<KeyState<B1::Time, Bk::RIn, Bk::ROut>> = Vec::new();
+        // Retire-wide reusable scratch: cleared per window or wave, never reallocated. Fresh
+        // per-key/per-wave `Vec`s were once the dominant cost here, which is why the slots and the
+        // staging buffers are held across the whole retire rather than built where they are used.
+        let mut slots: Vec<KeySweep<B1::Time, Bk::RIn, Bk::ROut>> = Vec::new();
+        let mut live: Vec<usize> = Vec::new();
         let mut tile_deltas: Vec<Vec<((u64, u64), B1::Time, Bk::ROut)>> = (0..held_elems.len()).map(|_| Vec::new()).collect();
         let mut batch_keys: Vec<u64> = Vec::new();
         let mut in_ends: Vec<usize> = Vec::new();
@@ -219,8 +219,6 @@ where
         let mut active: Vec<(usize, B1::Time)> = Vec::new();
         let mut in_accum: Vec<(u64, Bk::RIn)> = Vec::new();
         let mut cur_out: Vec<(u64, Bk::ROut)> = Vec::new();
-        let mut moments_scratch: Vec<B1::Time> = Vec::new();
-        let mut pended_scratch: Vec<B1::Time> = Vec::new();
 
         while from.is_some() {
             let before = from;
@@ -249,21 +247,22 @@ where
 
             for deltas in tile_deltas.iter_mut() { deltas.clear(); }
 
-            // Phase 1 (determination): for every key in the window, discover its interesting times
-            // (times only — no accumulation) and stand up its per-moment replays. Peak state is
-            // O(window presentation), bounded by the window `next_window` already materialized.
-            // `states` is a long-lived buffer reloaded slot-by-slot (not cleared/rebuilt): a slot's
-            // `Vec`s and replays are allocated once and reused, so keys cost no per-key alloc/free.
-            // `n_states` is the live prefix this window; higher slots persist (retaining capacity).
             // The window's keys are the hashes its presentations mention: the least of the three
             // heads, each iteration, until all three are drained. A `changed` key that appears in
             // none of them has no records at all, so its reduction has nothing to read and nothing
-            // to retract — the moment its due time would raise reaches the evaluation gate with an
+            // to retract — the time its due moment would raise reaches the evaluation gate with an
             // empty input and an empty output, and produces nothing. Skipping it is exactly what
             // visiting it would do. (`changed` is still the backend's instruction about which keys
             // to present; it is just not a source of keys here.)
-            let mut n_states = 0usize;
+            //
+            // Each key gets a `Sweep`, which discovers and evaluates in ONE ascending pass,
+            // suspending where the conventional reduce would call user logic. Slots are reused
+            // across windows, so a key costs no allocation of its own beyond the first window wide
+            // enough to need it. Peak state is O(window presentation), bounded by what
+            // `next_window` already materialized.
+            let mut n_slots = 0usize;
             let (mut is, mut ns, mut os) = (0usize, 0usize, 0usize);
+            live.clear();
             loop {
                 // Mapped to hashes before the min: the three runs differ in their diff type.
                 let Some(key) = [
@@ -281,139 +280,82 @@ where
                 while os < p_out.len() && p_out[os].0.0 == key { os += 1; }
                 let o1 = os;
 
-                moments_scratch.clear();
-                pended_scratch.clear();
-                {
-                    // Only the DUE times seed determination; the carried ones are already in
-                    // `new_pending` and must not be re-derived (nor re-pended) here.
-                    let pending = due.get(&key).map(|p| &p[..]).unwrap_or(&[]);
-                    // The seeds are the novel run's own time support for this key — the whole point
-                    // of presenting it apart from `history`.
-                    let seed_times = (n0..n1).map(|n| p_nv[n].1.clone());
-                    let out_times = (o0..o1).map(|o| p_out[o].1.clone());
-                    discover_times(
-                        KeyView { p_in: &p_in[..], i0, i1, pending },
-                        seed_times, out_times, upper,
-                        &mut discover_scratch,
-                        &mut moments_scratch, &mut pended_scratch,
-                    );
-                }
-                if !pended_scratch.is_empty() {
-                    // The key may already carry times beyond `upper` from the split above; union.
-                    let entry = new_pending.entry(key).or_default();
-                    entry.append(&mut pended_scratch);
-                    crate::operators::reduce::sort_dedup(entry);
-                }
-                if moments_scratch.is_empty() {
-                    continue;
-                }
-
-                // Reload slot `n_states` in place (grow the buffer by one only when a window is wider
-                // than any before). `drain` moves the discovered moments in without copy or realloc.
-                if n_states == states.len() {
-                    states.push(KeyState::empty());
+                if n_slots == slots.len() { slots.push(KeySweep::empty()); }
+                let slot = &mut slots[n_slots];
+                slot.key = key;
+                slot.pended.clear();
+                // Only the DUE times seed the sweep; the carried ones are already in `new_pending`.
+                let owed = due.get(&key).map(|p| &p[..]).unwrap_or(&[]);
+                slot.sweep.load(
+                    (n0..n1).map(|n| (p_nv[n].0.1, p_nv[n].1.clone(), p_nv[n].2.clone())),
+                    (i0..i1).map(|i| (p_in[i].0.1, p_in[i].1.clone(), p_in[i].2.clone())),
+                    (o0..o1).map(|o| (p_out[o].0.1, p_out[o].1.clone(), p_out[o].2.clone())),
+                    owed,
+                );
+                slot.at = slot.sweep.next_crossing(upper, &mut slot.pended);
+                if slot.at.is_some() { live.push(n_slots); }
+                else if !slot.pended.is_empty() {
+                    new_pending.entry(key).or_default().append(&mut slot.pended);
                 }
-                let st = &mut states[n_states];
-                st.key = key;
-                st.cursor = 0;
-                st.produced.clear();
-                st.moments.clear();
-                st.moments.append(&mut moments_scratch);
-                st.meets.clear();
-                st.meets.extend(st.moments.iter().cloned());
-                for i in (1..st.meets.len()).rev() {
-                    let m = st.meets[i].clone();
-                    st.meets[i - 1].meet_assign(&m);
-                }
-                st.in_replay.load_iter((i0..i1).map(|i| (p_in[i].0.1, p_in[i].1.clone(), p_in[i].2.clone())), st.meets.first());
-                st.nv_replay.load_iter((n0..n1).map(|n| (p_nv[n].0.1, p_nv[n].1.clone(), p_nv[n].2.clone())), st.meets.first());
-                st.out_replay.load_iter((o0..o1).map(|o| (p_out[o].0.1, p_out[o].1.clone(), p_out[o].2.clone())), st.meets.first());
-                n_states += 1;
+                n_slots += 1;
             }
 
-            // Phase 2 (application): walk all keys' moments in ROUNDS. Each round assembles every
-            // active key's one-moment-deep input and current-output accumulations and crosses them in
-            // a SINGLE `reduce_corrections` — batching across keys (a key's own moments stay
-            // sequential, each seeing its earlier corrections via `produced`). This caps the backend
-            // call count at O(max moments over keys), not O(sum of moments), with peak materialization
-            // one moment deep per key. `produced` is meet-collapsed each round, exactly like the
-            // reference — bounded, not the O(times × values) delta history.
-            loop {
+            // Each wave: read every suspended key's accumulations, cross the non-empty ones in one
+            // call, hand the corrections back, and step every live key on. A key retires when its
+            // sweep runs dry, at which point its pended times are carried forward.
+            while !live.is_empty() {
                 batch_keys.clear();
                 in_ends.clear();
                 in_all.clear();
                 out_ends.clear();
                 out_all.clear();
                 active.clear();
-                let mut advanced = false;
-                for (si, st) in states[..n_states].iter_mut().enumerate() {
-                    if st.cursor >= st.moments.len() {
-                        continue;
-                    }
-                    advanced = true;
-                    let j = st.cursor;
-                    st.cursor += 1;
-                    let t = st.moments[j].clone();
-                    st.in_replay.step_through(&t);
-                    st.nv_replay.step_through(&t);
-                    st.out_replay.step_through(&t);
-                    st.in_replay.advance_buffer_by(&st.meets[j]);
-                    st.nv_replay.advance_buffer_by(&st.meets[j]);
-                    st.out_replay.advance_buffer_by(&st.meets[j]);
-                    for ((_, et), _) in st.produced.iter_mut() {
-                        *et = et.join(&st.meets[j]);
-                    }
-                    crate::consolidation::consolidate(&mut st.produced);
 
-                    // The two input runs meet HERE, at the accumulation — never as presentations.
+                for &si in live.iter() {
+                    let at = slots[si].at.clone().expect("live slots are suspended at a time");
                     in_accum.clear();
-                    for ((vid, et), d) in st.in_replay.buffer().iter().chain(st.nv_replay.buffer().iter()) {
-                        if et.less_equal(&t) {
-                            in_accum.push((*vid, d.clone()));
-                        }
-                    }
-                    crate::consolidation::consolidate(&mut in_accum);
                     cur_out.clear();
-                    for ((vid, et), d) in st.out_replay.buffer().iter().chain(st.produced.iter()) {
-                        if et.less_equal(&t) {
-                            cur_out.push((*vid, d.clone()));
-                        }
-                    }
-                    crate::consolidation::consolidate(&mut cur_out);
-
-                    if in_accum.is_empty() && cur_out.is_empty() {
-                        continue;
-                    }
-                    batch_keys.push(st.key);
+                    slots[si].sweep.input_at(&at, &mut in_accum);
+                    slots[si].sweep.output_at(&at, &mut cur_out);
+                    // An interesting time can still reach the gate with nothing to read; the
+                    // conventional reduce skips user logic there and so do we.
+                    if in_accum.is_empty() && cur_out.is_empty() { continue; }
+                    batch_keys.push(slots[si].key);
                     in_all.append(&mut in_accum);
                     in_ends.push(in_all.len());
                     out_all.append(&mut cur_out);
                     out_ends.push(out_all.len());
-                    active.push((si, t));
-                }
-                // Terminate only when every key is EXHAUSTED — not merely when this round produced no
-                // crossing. A round can be empty because every key's current moment is empty-gated
-                // while keys still have later (non-empty) moments; breaking here would drop them.
-                if !advanced {
-                    break;
-                }
-                if batch_keys.is_empty() {
-                    continue;
+                    active.push((si, at));
                 }
 
-                let (corr, corr_ends) = self.backend.reduce_corrections(&batch_keys, &in_ends, &in_all, &out_ends, &out_all);
-                let mut cstart = 0usize;
-                for (bi, (si, t)) in active.iter().enumerate() {
-                    let cend = corr_ends[bi];
-                    if cstart != cend {
-                        let idx = held_elems.iter().rposition(|h| h.less_equal(t)).expect("no held capability <= active time");
-                        for (vid, d) in &corr[cstart..cend] {
-                            states[*si].produced.push(((*vid, t.clone()), d.clone()));
-                            tile_deltas[idx].push(((states[*si].key, *vid), t.clone(), d.clone()));
+                if !batch_keys.is_empty() {
+                    let (corr, corr_ends) = self.backend.reduce_corrections(&batch_keys, &in_ends, &in_all, &out_ends, &out_all);
+                    let mut cstart = 0usize;
+                    for (bi, (si, at)) in active.iter().enumerate() {
+                        let cend = corr_ends[bi];
+                        if cstart != cend {
+                            let idx = held_elems.iter().rposition(|h| h.less_equal(at)).expect("no held capability <= active time");
+                            for (vid, d) in &corr[cstart..cend] {
+                                tile_deltas[idx].push(((slots[*si].key, *vid), at.clone(), d.clone()));
+                            }
+                            slots[*si].sweep.commit(at, corr[cstart..cend].iter().cloned());
                         }
+                        cstart = cend;
                     }
-                    cstart = cend;
                 }
+
+                // Step every live key past the time it was suspended at, and retire the spent ones.
+                for si in 0..live.len() {
+                    let si = live[si];
+                    let slot = &mut slots[si];
+                    slot.at = slot.sweep.next_crossing(upper, &mut slot.pended);
+                    if slot.at.is_none() && !slot.pended.is_empty() {
+                        let entry = new_pending.entry(slot.key).or_default();
+                        entry.append(&mut slot.pended);
+                        crate::operators::reduce::sort_dedup(entry);
+                    }
+                }
+                live.retain(|&si| slots[si].at.is_some());
             }
 
             for (held_index, deltas) in tile_deltas.iter_mut().enumerate() {
@@ -439,29 +381,20 @@ where
     }
 }
 
-/// Per-key application state for [`ProxyReduceTactic`]'s round-batched walk: the key's ordered
-/// interesting `moments` and their suffix `meets`, its accumulated-input, novel-input, and output
-/// replays (meet-collapsed; the two input replays combine only in the per-moment accumulation),
-/// the corrections `produced` this round so far, and a `cursor` into `moments`. Held for all of a
-/// window's keys at once so each round's crossing batches across keys — a key's own moments stay
-/// sequential (each sees its earlier corrections via `produced`), but distinct keys are independent.
-struct KeyState<T, RIn, ROut> {
+/// One key's slot in a window: its [`Sweep`], the time it is suspended at, and the times it has
+/// pended so far. Slots are reused across windows, so a key costs no allocation of its own beyond
+/// the first window wide enough to need it.
+struct KeySweep<T, RIn, ROut> {
     key: u64,
-    moments: Vec<T>,
-    meets: Vec<T>,
-    in_replay: IdHistory<T, RIn>,
-    nv_replay: IdHistory<T, RIn>,
-    out_replay: IdHistory<T, ROut>,
-    produced: Vec<((u64, T), ROut)>,
-    cursor: usize,
+    sweep: Sweep<T, RIn, ROut>,
+    /// Times at or beyond `upper` the sweep has reached; carried forward when the slot retires.
+    pended: Vec<T>,
+    /// The time the sweep last suspended at, or `None` once it is spent.
+    at: Option<T>,
 }
 
-impl<T: Timestamp + Lattice, RIn: Semigroup, ROut: Semigroup> KeyState<T, RIn, ROut> {
-    /// An empty slot, to be filled by [`ProxyReduceTactic`]'s phase 1 (`reload`-style). The `states`
-    /// vector holds these across windows and reloads them in place, so a key's buffers are allocated
-    /// once (per slot) and reused — never dropped per key (which was ~18% of load in `free`).
+impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> KeySweep<T, RIn, ROut> {
     fn empty() -> Self {
-        KeyState { key: 0, moments: Vec::new(), meets: Vec::new(), in_replay: IdHistory::new(), nv_replay: IdHistory::new(), out_replay: IdHistory::new(), produced: Vec::new(), cursor: 0 }
+        KeySweep { key: 0, sweep: Sweep::new(), pended: Vec::new(), at: None }
     }
 }
-

From e7513478a1992f08a1bd6e32f450c53f06f9201e Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Tue, 11 Aug 2026 15:22:38 -0400
Subject: [PATCH 2/3] Delete the determination pass the sweep replaced, and
 operators::common

`discover_times` lost its last caller when the fused sweep landed.
`KeyView`, `DiscoverScratch` and `TimeHistory` existed only to serve it.
The containment test that checked the sweep against it goes too: it was scaffolding for building the sweep, and the ten `int_proxy` tests and the corgi gate check the same property end to end against the conventional reduce.

`common.rs` said "Types and methods generally useful for differential computation".
Every live item in it had exactly one caller, and all three callers were in `int_proxy`.
Naming it that way claimed a generality it did not have, hid the tactic's size from anyone reading the tactic, and is how four items died unnoticed: a module named for a role nobody audits accumulates things.

So `Sweep`, its `update_meet` helper and `tile_descriptions` move to the tail of `int_proxy/reduce.rs`, and `bilinear_wave` to `int_proxy/join.rs`, in each case beside their only caller.
None of them is `pub` any more; none was ever part of an interface.
The sweep and the tactic that drives it now sit in one file, which is where they belong: the boundary between them is meant to dissolve as more of the work fuses.

A warning for anyone who benchmarks this commit.
Under the workspace release profile it appears to cost 8% on churn and multimoment, and that is an artifact of `codegen-units`, which the profile leaves at the default 16 despite `lto = true`.
Rustc partitions codegen units by module, so folding two modules into one merges two partitions and the optimizer makes different choices; setting `codegen-units = 1` makes this commit and its module-per-item alternative indistinguishable, and both land at the faster number.
Layout was ruled out separately, by permuting function order within an untouched file and remeasuring.

The branch was +301 lines against master-next and is now -67.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 differential-dataflow/src/operators/common.rs | 748 ------------------
 .../src/operators/int_proxy/join.rs           |  57 +-
 .../src/operators/int_proxy/reduce.rs         | 332 +++++++-
 differential-dataflow/src/operators/mod.rs    |   5 +-
 4 files changed, 387 insertions(+), 755 deletions(-)
 delete mode 100644 differential-dataflow/src/operators/common.rs

diff --git a/differential-dataflow/src/operators/common.rs b/differential-dataflow/src/operators/common.rs
deleted file mode 100644
index 8c329e946..000000000
--- a/differential-dataflow/src/operators/common.rs
+++ /dev/null
@@ -1,748 +0,0 @@
-//! Types and methods generally useful for differential computation.
-//!
-//! The contents operate on `(id, time, diff)` update streams and lattice operations, with key
-//! and value representation left to the caller: replaying histories in time order with
-//! meet-advanced buffers ([`ValueHistory`], [`TimeHistory`]), determining the times at which a
-//! key's reduction must be re-evaluated ([`discover_times`]), producing the join of two
-//! histories in time order ([`bilinear_wave`]), and cutting an output interval into batch
-//! descriptions along held capabilities ([`tile_descriptions`]).
-//!
-//! Each item's correctness is a property of its own inputs and outputs, so they can be used
-//! and tested in isolation; sequencing obligations that span calls (for example, where
-//! interesting-time seeds must come from) are stated at the item that imposes them and are
-//! the caller's responsibility.
-
-use timely::progress::{Antichain, Timestamp};
-
-use crate::difference::{Multiply, Semigroup};
-use crate::lattice::Lattice;
-use crate::trace::Description;
-use crate::operators::reduce::sort_dedup;
-
-pub use crate::operators::ValueHistory;
-
-/// Replays a set of times in ascending order, maintaining the meet of the times not yet
-/// replayed and a deduplicated buffer of replayed times advanced by that meet. A cheaper
-/// [`ValueHistory`] for callers that need time structure only (no values, no cancellation).
-pub struct TimeHistory<T> {
-    /// Un-replayed `(time, meet)`, sorted descending by time so popping replays ascending;
-    /// `meet` is the meet of this time with all times later in the replay.
-    history: Vec<(T, T)>,
-    /// Stepped-in times, advanced and deduplicated, sorted ascending.
-    buffer: Vec<T>,
-}
-
-impl<T> Default for TimeHistory<T> {
-    fn default() -> Self { TimeHistory { history: Vec::new(), buffer: Vec::new() } }
-}
-
-impl<T: Lattice + Clone + Ord> TimeHistory<T> {
-    /// An empty history, to be `load`ed.
-    pub fn new() -> Self { TimeHistory { history: Vec::new(), buffer: Vec::new() } }
-
-    /// Load `times`, advancing each by `advance_by` if supplied, and organize the replay
-    /// (sort + suffix meets).
-    pub fn load(&mut self, times: impl Iterator<Item = T>, advance_by: Option<&T>) {
-        self.history.clear();
-        self.buffer.clear();
-        for mut time in times {
-            if let Some(m) = advance_by {
-                time = time.join(m);
-            }
-            self.history.push((time.clone(), time));
-        }
-        self.history.sort_by(|x, y| y.0.cmp(&x.0));
-        self.history.iter_mut().reduce(|prev, cur| {
-            cur.1.meet_assign(&prev.1);
-            cur
-        });
-    }
-
-    /// The next (least) un-replayed time.
-    pub fn time(&self) -> Option<&T> {
-        self.history.last().map(|x| &x.0)
-    }
-    /// The meet of all un-replayed times.
-    pub fn meet(&self) -> Option<&T> {
-        self.history.last().map(|x| &x.1)
-    }
-
-    /// Step times while the next equals `time`; true iff any did.
-    pub fn step_while_time_is(&mut self, time: &T) -> bool {
-        let mut found = false;
-        while self.time() == Some(time) {
-            found = true;
-            let (t, _) = self.history.pop().unwrap();
-            self.buffer.push(t);
-        }
-        found
-    }
-
-    /// Advance buffered times by `meet` and deduplicate — the collapse that keeps replay
-    /// linear.
-    pub fn advance_buffer_by(&mut self, meet: &T) {
-        for time in self.buffer.iter_mut() {
-            *time = time.join(meet);
-        }
-        self.buffer.sort();
-        self.buffer.dedup();
-    }
-
-    /// The buffered (stepped-in, advanced) times.
-    pub fn buffer(&self) -> &[T] {
-        &self.buffer
-    }
-}
-
-/// Produces the join of two histories: every pair of edits, diffs multiplied and times
-/// joined, visited in time order. Repeatedly steps the history with the earlier un-replayed
-/// edit and multiplies it against the other's buffer, which is consolidated under the meet of
-/// its remaining times as the wave advances — so work is bounded by the netted accumulation
-/// sizes rather than the raw history lengths.
-///
-/// `emit` receives every produced `(id0, id1, joined time, multiplied diff)`. Both histories
-/// must be pre-loaded (`load`/`load_iter`) and are fully drained. For small histories a plain
-/// cross product is cheaper; callers should gate on size.
-pub fn bilinear_wave<V, T, R0, R1, RO>(
-    h0: &mut ValueHistory<V, T, R0>,
-    h1: &mut ValueHistory<V, T, R1>,
-    mut emit: impl FnMut(V, V, T, RO),
-) where
-    V: Copy + Ord,
-    T: Ord + Clone + Lattice,
-    R0: Semigroup + Multiply<R1, Output = RO> + Clone,
-    R1: Semigroup + Clone,
-{
-    while h0.time().is_some() && h1.time().is_some() {
-        if h0.time().unwrap() < h1.time().unwrap() {
-            h1.advance_buffer_by(h0.meet().unwrap());
-            let (v0, t0, d0) = h0.edit().unwrap();
-            for ((v1, t1), d1) in h1.buffer() {
-                emit(v0, *v1, t0.join(t1), d0.clone().multiply(d1));
-            }
-            h0.step();
-        } else {
-            h0.advance_buffer_by(h1.meet().unwrap());
-            let (v1, t1, d1) = h1.edit().unwrap();
-            for ((v0, t0), d0) in h0.buffer() {
-                emit(*v0, v1, t0.join(t1), d0.clone().multiply(d1));
-            }
-            h1.step();
-        }
-    }
-    while h0.time().is_some() {
-        h1.advance_buffer_by(h0.meet().unwrap());
-        let (v0, t0, d0) = h0.edit().unwrap();
-        for ((v1, t1), d1) in h1.buffer() {
-            emit(v0, *v1, t0.join(t1), d0.clone().multiply(d1));
-        }
-        h0.step();
-    }
-    while h1.time().is_some() {
-        h0.advance_buffer_by(h1.meet().unwrap());
-        let (v1, t1, d1) = h1.edit().unwrap();
-        for ((v0, t0), d0) in h0.buffer() {
-            emit(*v0, v1, t0.join(t1), d0.clone().multiply(d1));
-        }
-        h1.step();
-    }
-}
-
-/// Cuts the interval `[lower, upper)` into consecutive batch descriptions along `held`, which
-/// must be sorted: the `i`-th cut point is the frontier formed by inserting `held[i+1..]` into
-/// `upper`, so description `i` covers the part of the interval not greater-or-equal any held
-/// time after `held[i]` (and not covered by an earlier description). Descriptions whose
-/// interval is empty are skipped. Returns the descriptions, the held time associated with
-/// each, and, per held index, the index of its description (`None` if skipped). A batch built
-/// to description `i` can be committed at the capability `held[i]`.
-pub fn tile_descriptions<T: Timestamp + Lattice>(
-    lower: &Antichain<T>,
-    upper: &Antichain<T>,
-    held: &[T],
-) -> (Vec<Description<T>>, Vec<T>, Vec<Option<usize>>) {
-    let mut tile_descs: Vec<Description<T>> = Vec::new();
-    let mut tile_held: Vec<T> = Vec::new();
-    let mut tile_of: Vec<Option<usize>> = vec![None; held.len()];
-    let mut out_lower = lower.clone();
-    for index in 0..held.len() {
-        let mut out_upper = upper.clone();
-        for t in &held[index + 1..] {
-            out_upper.insert(t.clone());
-        }
-        if out_upper != out_lower {
-            tile_of[index] = Some(tile_descs.len());
-            tile_descs.push(Description::new(out_lower.clone(), out_upper.clone(), Antichain::from_elem(T::minimum())));
-            tile_held.push(held[index].clone());
-            out_lower = out_upper;
-        }
-    }
-    (tile_descs, tile_held, tile_of)
-}
-
-/// A one-key view into the ACCUMULATED input presentation: the read-only arguments
-/// [`discover_times`] needs about a single key — its slice `[i0, i1)` of the `(id, time, diff)` run
-/// `p_in` and the carried `pending` times. The novel run is not here; it arrives as `seed_times`,
-/// and the two are deliberately never merged (see the note on `seed_times` below).
-pub struct KeyView<'a, T, RIn> {
-    /// The presented `((key_hash, value_id), time, diff)` run the key's records live in.
-    pub p_in: &'a [((u64, u64), T, RIn)],
-    /// The key's first record.
-    pub i0: usize,
-    /// One past the key's last record.
-    pub i1: usize,
-    /// Interesting times pended for this key by earlier retires.
-    pub pending: &'a [T],
-}
-
-/// Updates an optional meet by an optional time.
-fn update_meet<T: Lattice + Clone>(meet: &mut Option<T>, other: Option<&T>) {
-    if let Some(time) = other {
-        match meet.as_mut() {
-            Some(m) => m.meet_assign(time),
-            None => *meet = Some(time.clone()),
-        }
-    }
-}
-
-/// Reusable per-key scratch for [`discover_times`]: held once and threaded through every key,
-/// so replays and time buffers are cleared and refilled rather than reallocated per key.
-pub struct DiscoverScratch<T, RIn> {
-    batch_replay: TimeHistory<T>,
-    input_replay: ValueHistory<u64, T, RIn>,
-    output_replay: TimeHistory<T>,
-    synth: Vec<T>,
-    times_current: Vec<T>,
-    temporary: Vec<T>,
-    meets: Vec<T>,
-}
-
-impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> DiscoverScratch<T, RIn> {
-    /// Fresh scratch; hold one per retire and thread it through every key.
-    pub fn new() -> Self {
-        DiscoverScratch {
-            batch_replay: TimeHistory::new(),
-            input_replay: ValueHistory::new(),
-            output_replay: TimeHistory::new(),
-            synth: Vec::new(),
-            times_current: Vec::new(),
-            temporary: Vec::new(),
-            meets: Vec::new(),
-        }
-    }
-}
-
-impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> Default for DiscoverScratch<T, RIn> {
-    fn default() -> Self { Self::new() }
-}
-
-/// Determines the times in `[lower, upper)` at which a key's reduction must be re-evaluated
-/// (`moments`), and the times at or beyond `upper` to carry into the next invocation
-/// (`pended`). Replays the key's `seed_times` (the novel run) and `pending` times in ascending
-/// order, marking
-/// those that carry updates and closing the set under joins with the input and output
-/// histories' times and with each other. No input collection is materialized, so peak memory
-/// is O(times); buffers are advanced by the meet of the times still to come, keeping a key
-/// with many distinct times linear rather than quadratic.
-///
-/// `seed_times` must be the novel batch's own time support for this key, and `key.p_in` the
-/// accumulated input WITHOUT it. Seeding from a view of the two consolidated together is unsound:
-/// compaction may advance a history record onto a novel time, where consolidation cancels the novel
-/// update and its interesting time is missed. This is why the caller keeps the two runs apart and
-/// combines them only when accumulating.
-#[allow(clippy::too_many_arguments)]
-pub fn discover_times<T, RIn>(
-    key: KeyView<'_, T, RIn>,
-    seed_times: impl Iterator<Item = T>,
-    out_times: impl Iterator<Item = T>,
-    upper: &Antichain<T>,
-    scratch: &mut DiscoverScratch<T, RIn>,
-    moments: &mut Vec<T>,
-    pended: &mut Vec<T>,
-) where
-    T: Timestamp + Lattice,
-    RIn: Semigroup + Clone,
-{
-    // Reuse the retire's scratch: `load`/`load_iter` reset the replays (keeping capacity); the plain
-    // buffers are cleared here. `meets_slice` reborrows `meets` immutably; the rest stay disjoint.
-    let DiscoverScratch { batch_replay, input_replay, output_replay, synth, times_current, temporary, meets } = scratch;
-    synth.clear();
-    times_current.clear();
-    temporary.clear();
-
-    batch_replay.load(seed_times, None);
-
-    meets.clear();
-    meets.extend(key.pending.iter().cloned());
-    for i in (1..meets.len()).rev() {
-        let m = meets[i].clone();
-        meets[i - 1].meet_assign(&m);
-    }
-
-    let mut meet: Option<T> = None;
-    update_meet(&mut meet, meets.first());
-    update_meet(&mut meet, batch_replay.meet());
-
-    // The merged (history ⊎ novel) run — replayed for its TIMES only (join base), never
-    // accumulated. Output times likewise: base joins, never seeds.
-    input_replay.load_iter(
-        (key.i0..key.i1).map(|i| (key.p_in[i].0.1, key.p_in[i].1.clone(), key.p_in[i].2.clone())),
-        meet.as_ref(),
-    );
-    output_replay.load(out_times, meet.as_ref());
-
-    let mut times_slice = key.pending;
-    let mut meets_slice = &meets[..];
-
-    while let Some(next_time) = [batch_replay.time(), times_slice.first(), input_replay.time(), output_replay.time(), synth.last()]
-        .into_iter()
-        .flatten()
-        .min()
-        .cloned()
-    {
-        input_replay.step_while_time_is(&next_time);
-        output_replay.step_while_time_is(&next_time);
-        let mut interesting = batch_replay.step_while_time_is(&next_time);
-        if interesting {
-            if let Some(m) = meet.as_ref() {
-                batch_replay.advance_buffer_by(m);
-            }
-        }
-        while synth.last() == Some(&next_time) {
-            times_current.push(synth.pop().expect("nonempty"));
-            interesting = true;
-        }
-        while times_slice.first() == Some(&next_time) {
-            times_current.push(times_slice[0].clone());
-            times_slice = &times_slice[1..];
-            meets_slice = &meets_slice[1..];
-            interesting = true;
-        }
-        interesting = interesting || batch_replay.buffer().iter().any(|t| t.less_equal(&next_time));
-        interesting = interesting || times_current.iter().any(|t| t.less_equal(&next_time));
-
-        if !upper.less_equal(&next_time) {
-            if interesting {
-                // Synthesize joins against the input/output histories (times only — no
-                // accumulation), then record `next_time` as an interesting moment.
-                if let Some(m) = meet.as_ref() {
-                    input_replay.advance_buffer_by(m);
-                }
-                for ((_, t), _) in input_replay.buffer().iter() {
-                    if !t.less_equal(&next_time) {
-                        temporary.push(next_time.join(t));
-                    }
-                }
-                if let Some(m) = meet.as_ref() {
-                    output_replay.advance_buffer_by(m);
-                }
-                for t in output_replay.buffer().iter() {
-                    if !t.less_equal(&next_time) {
-                        temporary.push(next_time.join(t));
-                    }
-                }
-                moments.push(next_time.clone());
-            }
-            temporary.extend(batch_replay.buffer().iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
-            temporary.extend(times_current.iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
-            sort_dedup(temporary);
-            let synth_len = synth.len();
-            for time in temporary.drain(..) {
-                if upper.less_equal(&time) {
-                    pended.push(time);
-                } else {
-                    synth.push(time);
-                }
-            }
-            if synth.len() > synth_len {
-                synth.sort_by(|x, y| y.cmp(x));
-                synth.dedup();
-            }
-        } else if interesting {
-            pended.push(next_time.clone());
-        }
-
-        meet = None;
-        update_meet(&mut meet, batch_replay.meet());
-        update_meet(&mut meet, input_replay.meet());
-        update_meet(&mut meet, output_replay.meet());
-        for t in synth.iter() {
-            update_meet(&mut meet, Some(t));
-        }
-        update_meet(&mut meet, meets_slice.first());
-        if let Some(m) = meet.as_ref() {
-            for t in times_current.iter_mut() {
-                *t = t.join(m);
-            }
-        }
-        sort_dedup(times_current);
-    }
-    sort_dedup(pended);
-}
-
-/// A resumable, fused determination-and-evaluation sweep over one key's times.
-///
-/// [`discover_times`] enumerates a key's interesting times up front, and the caller then walks them
-/// again to evaluate. The conventional reduce does not: it runs ONE ascending pass and evaluates as
-/// it discovers. It can, because discovery never looks backwards — every synthesized time is
-/// `next_time.join(t)` for some `t` NOT at or below `next_time`, so it is strictly greater, and new
-/// work only ever lands ahead of the sweep.
-///
-/// This is that pass, cut at the point where the conventional operator would call user logic. Each
-/// [`next_crossing`](Self::next_crossing) returns the next in-interval time that needs evaluating,
-/// with the buffers positioned to read the accumulations; the caller evaluates and hands the
-/// corrections back through [`commit`](Self::commit); the next call resumes. Many keys can be run
-/// to their next crossing and evaluated together, which is what a batched backend wants, without
-/// any of them enumerating their times first.
-///
-/// The schedule is `formal/Differential/RoundCoverage.lean`'s `round_coverage`: a time carrying an
-/// output change lies in the join-closure of `prior ∪ novel` AND is at or above some novel time.
-/// The two clauses appear here as the `interesting` test and the split synthesis — see the comments
-/// at each.
-pub struct Sweep<T, RIn, ROut> {
-    /// The novel run: the only source that SEEDS interest, replayed unadvanced.
-    novel: ValueHistory<u64, T, RIn>,
-    /// The accumulated input and output: join partners, and the accumulations to evaluate over.
-    input: ValueHistory<u64, T, RIn>,
-    output: ValueHistory<u64, T, ROut>,
-    /// The key's due interesting times, ascending, with their suffix meets; `due_pos` consumes them.
-    due: Vec<T>,
-    due_meets: Vec<T>,
-    due_pos: usize,
-    /// Synthesized times not yet visited, sorted DESCENDING so `last()` is the least.
-    synth: Vec<T>,
-    /// The seed times reached so far, compacted by the running meet. They are the witnesses the
-    /// absorption test looks for, and the partners a close joins against; keeping them collapsed is
-    /// what stops a key with many reached times rescanning all of them.
-    reached: Vec<T>,
-    /// Scratch for one step's synthesized times.
-    temporary: Vec<T>,
-    /// Corrections emitted so far this sweep, meet-collapsed; both a join partner and part of the
-    /// output accumulation.
-    produced: Vec<((u64, T), ROut)>,
-    /// The meet of every time still to come.
-    meet: Option<T>,
-    /// Whether the last `next_crossing` returned a time whose step is not yet settled.
-    suspended: bool,
-}
-
-/// What one [`tick`](Sweep::tick) decided about the time it visited.
-enum Tick<T> {
-    /// No seed reaches this time; the sweep moved past it.
-    Passed,
-    /// Reached, but at or beyond `upper`: carried to a later round rather than evaluated.
-    Pended,
-    /// Reached and in the interval. The caller must evaluate here before the sweep goes on.
-    Crossing(T),
-    /// Every source is drained.
-    Done,
-}
-
-impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Sweep<T, RIn, ROut> {
-    /// An empty sweep, to be `load`ed. Reuse one per key rather than allocating per key.
-    pub fn new() -> Self {
-        Sweep {
-            novel: ValueHistory::new(), input: ValueHistory::new(), output: ValueHistory::new(),
-            due: Vec::new(), due_meets: Vec::new(), due_pos: 0,
-            synth: Vec::new(), reached: Vec::new(), temporary: Vec::new(),
-            produced: Vec::new(), meet: None, suspended: false,
-        }
-    }
-
-    /// Position the sweep at the start of one key.
-    ///
-    /// `novel` must be the batch's own `(id, time, diff)` support for the key and `input` the
-    /// accumulated input WITHOUT it: the two are never merged, because consolidating them can cancel
-    /// a novel update against a compacted history record and lose its interesting time.
-    pub fn load(
-        &mut self,
-        novel: impl Iterator<Item = (u64, T, RIn)>,
-        input: impl Iterator<Item = (u64, T, RIn)>,
-        output: impl Iterator<Item = (u64, T, ROut)>,
-        due: &[T],
-    ) {
-        self.due.clear();
-        self.due.extend(due.iter().cloned());
-        self.due_meets.clear();
-        self.due_meets.extend(due.iter().cloned());
-        for i in (1..self.due_meets.len()).rev() {
-            let (init, tail) = self.due_meets.split_at_mut(i);
-            init[i - 1].meet_assign(&tail[0]);
-        }
-        self.due_pos = 0;
-        self.synth.clear();
-        self.reached.clear();
-        self.temporary.clear();
-        self.produced.clear();
-        self.suspended = false;
-
-        // The novel run is loaded UNADVANCED: it is the seed, and its own times are what the
-        // schedule is stated over. Only then is the meet known, and the join partners advanced by it.
-        self.novel.load_iter(novel, None);
-        let mut meet: Option<T> = None;
-        update_meet(&mut meet, self.due_meets.first());
-        update_meet(&mut meet, self.novel.meet());
-        self.input.load_iter(input, meet.as_ref());
-        self.output.load_iter(output, meet.as_ref());
-        self.meet = meet;
-    }
-
-    /// Advance to the next in-interval time that needs evaluating, or `None` once the key is spent.
-    ///
-    /// Times at or beyond `upper` that the schedule reaches are appended to `pended` for the caller
-    /// to carry into a later round.
-    pub fn next_crossing(&mut self, upper: &Antichain<T>, pended: &mut Vec<T>) -> Option<T> {
-        loop {
-            // A crossing leaves its step half-finished, because `settle` must see the corrections
-            // the caller commits. Finishing it is the first thing the next call does.
-            if self.suspended {
-                self.suspended = false;
-                self.settle();
-            }
-            match self.tick(upper, pended) {
-                Tick::Done => return None,
-                Tick::Crossing(at) => {
-                    self.suspended = true;
-                    return Some(at);
-                }
-                Tick::Passed | Tick::Pended => {}
-            }
-        }
-    }
-
-    /// Visit one time: find it, decide whether it is reached, close it forward, and report.
-    fn tick(&mut self, upper: &Antichain<T>, pended: &mut Vec<T>) -> Tick<T> {
-        let Some(at) = self.frontier() else { return Tick::Done };
-        let reached = self.absorb(&at);
-        if upper.less_equal(&at) {
-            // Out of the interval: nothing can be emitted here, so there is nothing to close
-            // against either — a join with `at` is at or beyond `at`, hence also out of interval,
-            // and will be rediscovered from `at` in the round that admits it.
-            self.settle();
-            if reached { pended.push(at); return Tick::Pended; }
-            return Tick::Passed;
-        }
-        self.close(&at, reached, upper, pended);
-        if reached { return Tick::Crossing(at); }
-        self.settle();
-        Tick::Passed
-    }
-
-    /// The sweep's position: the least time any source still offers.
-    ///
-    /// The TOTAL order, not the partial one. Every time `close` produces is strictly greater than
-    /// the position that produced it, so new work only ever lands ahead of here and the sweep never
-    /// revisits.
-    fn frontier(&self) -> Option<T> {
-        [
-            self.novel.time(), self.due.get(self.due_pos), self.input.time(),
-            self.output.time(), self.synth.last(),
-        ].into_iter().flatten().min().cloned()
-    }
-
-    /// Step every source sitting at `at`, and decide whether `at` is REACHED.
-    ///
-    /// Reached is clause two of `round_coverage` — `∃ nu ∈ novel, nu ≤ at` — evaluated
-    /// incrementally: either a seed lands exactly here, or one already stepped in lies below.
-    ///
-    /// Input and output are stepped whether or not `at` is reached, and that is forced rather than
-    /// eager: they are sources of the frontier, so leaving them would stall the sweep, and their
-    /// edits must reach the buffers or they are lost to every later accumulation. Stepping only
-    /// moves an edit across; it consolidates nothing. The expensive part — `advance_buffer_by`,
-    /// which joins every buffered time and re-consolidates — is deferred to `close`, and happens
-    /// only where the buffers are actually read.
-    fn absorb(&mut self, at: &T) -> bool {
-        self.input.step_while_time_is(at);
-        self.output.step_while_time_is(at);
-
-        // A novel update here is a seed.
-        let mut reached = self.novel.step_while_time_is(at);
-        if reached {
-            if let Some(meet) = self.meet.as_ref() { self.novel.advance_buffer_by(meet); }
-        }
-        // So is a synthetic join scheduled for here, or a due time carried in. Both move into the
-        // reached set, where they become witnesses and join partners for later times.
-        while self.synth.last() == Some(at) {
-            self.reached.push(self.synth.pop().expect("nonempty"));
-            reached = true;
-        }
-        while self.due.get(self.due_pos) == Some(at) {
-            self.reached.push(at.clone());
-            self.due_pos += 1;
-            reached = true;
-        }
-        // Absorption: a time at or above a seed already stepped in is itself reached, because
-        // joining that seed with it yields it back. Checked against the stepped prefixes only.
-        reached
-            || self.novel.buffer().iter().any(|((_, t), _)| t.less_equal(at))
-            || self.reached.iter().any(|t| t.less_equal(at))
-    }
-
-    /// Close `at` forward under joins — clause one of `round_coverage`, the join-closure.
-    ///
-    /// Against the NOVEL times always, reached or not: an unreached time joined with a novel time
-    /// lands at or above that novel time, so it carries a witness and is on the schedule.
-    ///
-    /// Against the PRIOR times only when `at` is itself reached, because the join then inherits
-    /// `at`'s witness. A join of two prior times carries none and is deliberately never produced —
-    /// that asymmetry is the whole of why an incremental operator does less work than the closure
-    /// of everything.
-    ///
-    /// `produced` counts as prior. A correction emitted at `p` changes the accumulated output at
-    /// every time at or above `p`, so `p ∨ at` has to be visited; nothing else covers it, since
-    /// this round's corrections are not in the output history and `at` was not yet stepped in when
-    /// the sweep passed `p`.
-    fn close(&mut self, at: &T, reached: bool, upper: &Antichain<T>, pended: &mut Vec<T>) {
-        self.temporary.extend(self.novel.buffer().iter().map(|((_, t), _)| t)
-            .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
-        self.temporary.extend(self.reached.iter()
-            .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
-        if reached {
-            if let Some(meet) = self.meet.as_ref() {
-                self.input.advance_buffer_by(meet);
-                self.output.advance_buffer_by(meet);
-            }
-            self.temporary.extend(self.input.buffer().iter().map(|((_, t), _)| t)
-                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
-            self.temporary.extend(self.output.buffer().iter().map(|((_, t), _)| t)
-                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
-            self.temporary.extend(self.produced.iter().map(|((_, t), _)| t)
-                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
-        }
-        sort_dedup(&mut self.temporary);
-        let before = self.synth.len();
-        for time in self.temporary.drain(..) {
-            if upper.less_equal(&time) { pended.push(time); } else { self.synth.push(time); }
-        }
-        if self.synth.len() > before {
-            self.synth.sort_by(|x, y| y.cmp(x));
-            self.synth.dedup();
-        }
-    }
-
-    /// The input accumulation at the suspended time: both input runs, meeting only here.
-    pub fn input_at(&self, at: &T, into: &mut Vec<(u64, RIn)>) {
-        for ((id, time), diff) in self.input.buffer().iter().chain(self.novel.buffer().iter()) {
-            if time.less_equal(at) { into.push((*id, diff.clone())); }
-        }
-        crate::consolidation::consolidate(into);
-    }
-
-    /// The tentative output accumulation at the suspended time, including this sweep's corrections.
-    pub fn output_at(&self, at: &T, into: &mut Vec<(u64, ROut)>) {
-        for ((id, time), diff) in self.output.buffer().iter().chain(self.produced.iter()) {
-            if time.less_equal(at) { into.push((*id, diff.clone())); }
-        }
-        crate::consolidation::consolidate(into);
-    }
-
-    /// Record the corrections evaluated at the suspended time, and collapse them by the meet.
-    pub fn commit(&mut self, at: &T, corrections: impl Iterator<Item = (u64, ROut)>) {
-        let before = self.produced.len();
-        for (id, diff) in corrections { self.produced.push(((id, at.clone()), diff)); }
-        if self.produced.len() > before {
-            if let Some(meet) = self.meet.as_ref() {
-                for entry in self.produced.iter_mut() { (entry.0).1.join_assign(meet); }
-            }
-            crate::consolidation::consolidate(&mut self.produced);
-        }
-    }
-
-    /// Close a step: recompute the meet of everything still to come, and compact the reached set by
-    /// it. This is what keeps a key with a long history linear rather than quadratic.
-    fn settle(&mut self) {
-        let mut meet: Option<T> = None;
-        update_meet(&mut meet, self.novel.meet());
-        update_meet(&mut meet, self.input.meet());
-        update_meet(&mut meet, self.output.meet());
-        for time in self.synth.iter() { update_meet(&mut meet, Some(time)); }
-        update_meet(&mut meet, self.due_meets.get(self.due_pos));
-        if let Some(m) = meet.as_ref() {
-            for time in self.reached.iter_mut() { *time = time.join(m); }
-        }
-        sort_dedup(&mut self.reached);
-        self.meet = meet;
-    }
-}
-
-#[cfg(test)]
-mod sweep_tests {
-
-    use timely::order::Product;
-    use timely::progress::Antichain;
-
-    use super::{discover_times, DiscoverScratch, KeyView, Sweep};
-
-    type Time = Product<u64, u64>;
-
-    /// A deterministic LCG: enough variety without a dependency.
-    struct Rng(u64);
-    impl Rng {
-        fn next(&mut self, bound: u64) -> u64 {
-            self.0 = self.0.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
-            (self.0 >> 33) % bound
-        }
-    }
-
-    fn time(r: &mut Rng, span: u64) -> Time { Product::new(r.next(span), r.next(span)) }
-
-    /// `Sweep` must never reach a time `discover_times` does not: determination is the schedule,
-    /// and a crossing outside it would be work the coverage argument does not cover.
-    ///
-    /// The converse is NOT asserted, and must not be. The fused pass may legitimately explore FEWER
-    /// times, because it reaches each one carrying live accumulations: a time whose updates have
-    /// consolidated away by the time the sweep arrives needs no evaluation, where determination —
-    /// which reads times only — has no way to know that. Pinning equality here would forbid the
-    /// pruning that fusing the two passes exists to enable. Output equality against the conventional
-    /// reduce is the real check, and lives in the dataflow tests.
-    ///
-    /// As it happens the two coincide today, since both consolidate their buffers identically and
-    /// nothing is committed here; the containment is what is guaranteed.
-    #[test]
-    fn sweep_within_discover_times() {
-        for seed in 0..400u64 {
-            let mut rng = Rng(seed.wrapping_mul(2654435761).wrapping_add(12345));
-            let span = 2 + rng.next(4);
-
-            let novel: Vec<(u64, Time, i64)> =
-                (0..rng.next(6)).map(|_| (rng.next(3), time(&mut rng, span), 1i64)).collect();
-            let input: Vec<(u64, Time, i64)> =
-                (0..rng.next(6)).map(|_| (rng.next(3), time(&mut rng, span), 1i64)).collect();
-            let output: Vec<(u64, Time, i64)> =
-                (0..rng.next(4)).map(|_| (rng.next(3), time(&mut rng, span), 1i64)).collect();
-            let mut due: Vec<Time> = (0..rng.next(3)).map(|_| time(&mut rng, span)).collect();
-            due.sort();
-            due.dedup();
-            let upper = Antichain::from_elem(Product::new(span, span));
-
-            // Determination: `p_in` is the accumulated input, the seeds are the novel times.
-            let p_in: Vec<((u64, u64), Time, i64)> =
-                input.iter().map(|(v, t, d)| ((0u64, *v), *t, *d)).collect();
-            let mut scratch: DiscoverScratch<Time, i64> = DiscoverScratch::new();
-            let (mut moments, mut pended) = (Vec::new(), Vec::new());
-            discover_times(
-                KeyView { p_in: &p_in[..], i0: 0, i1: p_in.len(), pending: &due[..] },
-                novel.iter().map(|(_, t, _)| *t),
-                output.iter().map(|(_, t, _)| *t),
-                &upper,
-                &mut scratch,
-                &mut moments,
-                &mut pended,
-            );
-
-            // The fused sweep, committing nothing.
-            let mut sweep: Sweep<Time, i64, i64> = Sweep::new();
-            sweep.load(novel.iter().cloned(), input.iter().cloned(), output.iter().cloned(), &due[..]);
-            let (mut crossings, mut sweep_pended) = (Vec::new(), Vec::new());
-            while let Some(at) = sweep.next_crossing(&upper, &mut sweep_pended) {
-                crossings.push(at);
-            }
-            crate::operators::reduce::sort_dedup(&mut sweep_pended);
-
-            assert!(crossings.iter().all(|t| moments.contains(t)),
-                    "seed {seed}: sweep reached a time outside the schedule: {crossings:?} vs {moments:?}");
-            assert!(crossings.windows(2).all(|w| w[0] <= w[1]), "seed {seed}: crossings must ascend");
-            assert!(sweep_pended.iter().all(|t| pended.contains(t)),
-                    "seed {seed}: sweep pended a time outside the schedule");
-        }
-    }
-}
diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
index 7a024d596..401cad3fe 100644
--- a/differential-dataflow/src/operators/int_proxy/join.rs
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -13,6 +13,7 @@ use crate::lattice::Lattice;
 use crate::trace::BatchReader;
 use super::ProxyBridge;
 use crate::operators::join::{Fresh, JoinTactic};
+use crate::operators::ValueHistory;
 
 use super::history::IdHistory;
 
@@ -263,10 +264,64 @@ fn join_key<T, R0, R1, RO>(
     else {
         h0.load_iter(r0.map(|i| (p0[i].0.1, p0[i].1.clone(), p0[i].2.clone())), None);
         h1.load_iter(r1.map(|i| (p1[i].0.1, p1[i].1.clone(), p1[i].2.clone())), None);
-        crate::operators::common::bilinear_wave(h0, h1, |v0, v1, t, d| {
+        bilinear_wave(h0, h1, |v0, v1, t, d| {
             matches.ids.push((kh, (v0, v1)));
             matches.times.push(t);
             matches.diffs.push(d);
         });
     }
 }
+
+/// Produces the join of two histories: every pair of edits, diffs multiplied and times
+/// joined, visited in time order. Repeatedly steps the history with the earlier un-replayed
+/// edit and multiplies it against the other's buffer, which is consolidated under the meet of
+/// its remaining times as the wave advances — so work is bounded by the netted accumulation
+/// sizes rather than the raw history lengths.
+///
+/// `emit` receives every produced `(id0, id1, joined time, multiplied diff)`. Both histories
+/// must be pre-loaded (`load`/`load_iter`) and are fully drained. For small histories a plain
+/// cross product is cheaper; callers should gate on size.
+fn bilinear_wave<V, T, R0, R1, RO>(
+    h0: &mut ValueHistory<V, T, R0>,
+    h1: &mut ValueHistory<V, T, R1>,
+    mut emit: impl FnMut(V, V, T, RO),
+) where
+    V: Copy + Ord,
+    T: Ord + Clone + Lattice,
+    R0: Semigroup + Multiply<R1, Output = RO> + Clone,
+    R1: Semigroup + Clone,
+{
+    while h0.time().is_some() && h1.time().is_some() {
+        if h0.time().unwrap() < h1.time().unwrap() {
+            h1.advance_buffer_by(h0.meet().unwrap());
+            let (v0, t0, d0) = h0.edit().unwrap();
+            for ((v1, t1), d1) in h1.buffer() {
+                emit(v0, *v1, t0.join(t1), d0.clone().multiply(d1));
+            }
+            h0.step();
+        } else {
+            h0.advance_buffer_by(h1.meet().unwrap());
+            let (v1, t1, d1) = h1.edit().unwrap();
+            for ((v0, t0), d0) in h0.buffer() {
+                emit(*v0, v1, t0.join(t1), d0.clone().multiply(d1));
+            }
+            h1.step();
+        }
+    }
+    while h0.time().is_some() {
+        h1.advance_buffer_by(h0.meet().unwrap());
+        let (v0, t0, d0) = h0.edit().unwrap();
+        for ((v1, t1), d1) in h1.buffer() {
+            emit(v0, *v1, t0.join(t1), d0.clone().multiply(d1));
+        }
+        h0.step();
+    }
+    while h1.time().is_some() {
+        h0.advance_buffer_by(h1.meet().unwrap());
+        let (v1, t1, d1) = h1.edit().unwrap();
+        for ((v0, t0), d0) in h0.buffer() {
+            emit(*v0, v1, t0.join(t1), d0.clone().multiply(d1));
+        }
+        h1.step();
+    }
+}
diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index af29bb982..a7143fed4 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -13,9 +13,8 @@ use crate::difference::Semigroup;
 use crate::lattice::Lattice;
 use crate::trace::{BatchReader, Description};
 use super::ProxyBridge;
-use crate::operators::reduce::ReduceTactic;
-
-use crate::operators::common::{tile_descriptions, Sweep};
+use crate::operators::reduce::{sort_dedup, ReduceTactic};
+use crate::operators::ValueHistory;
 
 /// A unit of proxied reduce work, presented to the backend.
 pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>> {
@@ -398,3 +397,330 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Ke
         KeySweep { key: 0, sweep: Sweep::new(), pended: Vec::new(), at: None }
     }
 }
+
+/// Cuts the interval `[lower, upper)` into consecutive batch descriptions along `held`, which
+/// must be sorted: the `i`-th cut point is the frontier formed by inserting `held[i+1..]` into
+/// `upper`, so description `i` covers the part of the interval not greater-or-equal any held
+/// time after `held[i]` (and not covered by an earlier description). Descriptions whose
+/// interval is empty are skipped. Returns the descriptions, the held time associated with
+/// each, and, per held index, the index of its description (`None` if skipped). A batch built
+/// to description `i` can be committed at the capability `held[i]`.
+fn tile_descriptions<T: Timestamp + Lattice>(
+    lower: &Antichain<T>,
+    upper: &Antichain<T>,
+    held: &[T],
+) -> (Vec<Description<T>>, Vec<T>, Vec<Option<usize>>) {
+    let mut tile_descs: Vec<Description<T>> = Vec::new();
+    let mut tile_held: Vec<T> = Vec::new();
+    let mut tile_of: Vec<Option<usize>> = vec![None; held.len()];
+    let mut out_lower = lower.clone();
+    for index in 0..held.len() {
+        let mut out_upper = upper.clone();
+        for t in &held[index + 1..] {
+            out_upper.insert(t.clone());
+        }
+        if out_upper != out_lower {
+            tile_of[index] = Some(tile_descs.len());
+            tile_descs.push(Description::new(out_lower.clone(), out_upper.clone(), Antichain::from_elem(T::minimum())));
+            tile_held.push(held[index].clone());
+            out_lower = out_upper;
+        }
+    }
+    (tile_descs, tile_held, tile_of)
+}
+
+/// Updates an optional meet by an optional time.
+fn update_meet<T: Lattice + Clone>(meet: &mut Option<T>, other: Option<&T>) {
+    if let Some(time) = other {
+        match meet.as_mut() {
+            Some(m) => m.meet_assign(time),
+            None => *meet = Some(time.clone()),
+        }
+    }
+}
+
+/// A resumable, fused determination-and-evaluation sweep over one key's times.
+///
+/// A determination pass would enumerate a key's interesting times up front, and the caller would
+/// then walk them again to evaluate. The conventional reduce does not: it runs ONE ascending pass
+/// and evaluates as it discovers. It can, because discovery never looks backwards — every
+/// synthesized time is
+/// `next_time.join(t)` for some `t` NOT at or below `next_time`, so it is strictly greater, and new
+/// work only ever lands ahead of the sweep.
+///
+/// This is that pass, cut at the point where the conventional operator would call user logic. Each
+/// [`next_crossing`](Self::next_crossing) returns the next in-interval time that needs evaluating,
+/// with the buffers positioned to read the accumulations; the caller evaluates and hands the
+/// corrections back through [`commit`](Self::commit); the next call resumes. Many keys can be run
+/// to their next crossing and evaluated together, which is what a batched backend wants, without
+/// any of them enumerating their times first.
+///
+/// The schedule is `formal/Differential/RoundCoverage.lean`'s `round_coverage`: a time carrying an
+/// output change lies in the join-closure of `prior ∪ novel` AND is at or above some novel time.
+/// The two clauses appear here as the `interesting` test and the split synthesis — see the comments
+/// at each.
+struct Sweep<T, RIn, ROut> {
+    /// The novel run: the only source that SEEDS interest, replayed unadvanced.
+    novel: ValueHistory<u64, T, RIn>,
+    /// The accumulated input and output: join partners, and the accumulations to evaluate over.
+    input: ValueHistory<u64, T, RIn>,
+    output: ValueHistory<u64, T, ROut>,
+    /// The key's due interesting times, ascending, with their suffix meets; `due_pos` consumes them.
+    due: Vec<T>,
+    due_meets: Vec<T>,
+    due_pos: usize,
+    /// Synthesized times not yet visited, sorted DESCENDING so `last()` is the least.
+    synth: Vec<T>,
+    /// The seed times reached so far, compacted by the running meet. They are the witnesses the
+    /// absorption test looks for, and the partners a close joins against; keeping them collapsed is
+    /// what stops a key with many reached times rescanning all of them.
+    reached: Vec<T>,
+    /// Scratch for one step's synthesized times.
+    temporary: Vec<T>,
+    /// Corrections emitted so far this sweep, meet-collapsed; both a join partner and part of the
+    /// output accumulation.
+    produced: Vec<((u64, T), ROut)>,
+    /// The meet of every time still to come.
+    meet: Option<T>,
+    /// Whether the last `next_crossing` returned a time whose step is not yet settled.
+    suspended: bool,
+}
+
+/// What one [`tick`](Sweep::tick) decided about the time it visited.
+enum Tick<T> {
+    /// No seed reaches this time; the sweep moved past it.
+    Passed,
+    /// Reached, but at or beyond `upper`: carried to a later round rather than evaluated.
+    Pended,
+    /// Reached and in the interval. The caller must evaluate here before the sweep goes on.
+    Crossing(T),
+    /// Every source is drained.
+    Done,
+}
+
+impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> Sweep<T, RIn, ROut> {
+    /// An empty sweep, to be `load`ed. Reuse one per key rather than allocating per key.
+    fn new() -> Self {
+        Sweep {
+            novel: ValueHistory::new(), input: ValueHistory::new(), output: ValueHistory::new(),
+            due: Vec::new(), due_meets: Vec::new(), due_pos: 0,
+            synth: Vec::new(), reached: Vec::new(), temporary: Vec::new(),
+            produced: Vec::new(), meet: None, suspended: false,
+        }
+    }
+
+    /// Position the sweep at the start of one key.
+    ///
+    /// `novel` must be the batch's own `(id, time, diff)` support for the key and `input` the
+    /// accumulated input WITHOUT it: the two are never merged, because consolidating them can cancel
+    /// a novel update against a compacted history record and lose its interesting time.
+    fn load(
+        &mut self,
+        novel: impl Iterator<Item = (u64, T, RIn)>,
+        input: impl Iterator<Item = (u64, T, RIn)>,
+        output: impl Iterator<Item = (u64, T, ROut)>,
+        due: &[T],
+    ) {
+        self.due.clear();
+        self.due.extend(due.iter().cloned());
+        self.due_meets.clear();
+        self.due_meets.extend(due.iter().cloned());
+        for i in (1..self.due_meets.len()).rev() {
+            let (init, tail) = self.due_meets.split_at_mut(i);
+            init[i - 1].meet_assign(&tail[0]);
+        }
+        self.due_pos = 0;
+        self.synth.clear();
+        self.reached.clear();
+        self.temporary.clear();
+        self.produced.clear();
+        self.suspended = false;
+
+        // The novel run is loaded UNADVANCED: it is the seed, and its own times are what the
+        // schedule is stated over. Only then is the meet known, and the join partners advanced by it.
+        self.novel.load_iter(novel, None);
+        let mut meet: Option<T> = None;
+        update_meet(&mut meet, self.due_meets.first());
+        update_meet(&mut meet, self.novel.meet());
+        self.input.load_iter(input, meet.as_ref());
+        self.output.load_iter(output, meet.as_ref());
+        self.meet = meet;
+    }
+
+    /// Advance to the next in-interval time that needs evaluating, or `None` once the key is spent.
+    ///
+    /// Times at or beyond `upper` that the schedule reaches are appended to `pended` for the caller
+    /// to carry into a later round.
+    fn next_crossing(&mut self, upper: &Antichain<T>, pended: &mut Vec<T>) -> Option<T> {
+        loop {
+            // A crossing leaves its step half-finished, because `settle` must see the corrections
+            // the caller commits. Finishing it is the first thing the next call does.
+            if self.suspended {
+                self.suspended = false;
+                self.settle();
+            }
+            match self.tick(upper, pended) {
+                Tick::Done => return None,
+                Tick::Crossing(at) => {
+                    self.suspended = true;
+                    return Some(at);
+                }
+                Tick::Passed | Tick::Pended => {}
+            }
+        }
+    }
+
+    /// Visit one time: find it, decide whether it is reached, close it forward, and report.
+    fn tick(&mut self, upper: &Antichain<T>, pended: &mut Vec<T>) -> Tick<T> {
+        let Some(at) = self.frontier() else { return Tick::Done };
+        let reached = self.absorb(&at);
+        if upper.less_equal(&at) {
+            // Out of the interval: nothing can be emitted here, so there is nothing to close
+            // against either — a join with `at` is at or beyond `at`, hence also out of interval,
+            // and will be rediscovered from `at` in the round that admits it.
+            self.settle();
+            if reached { pended.push(at); return Tick::Pended; }
+            return Tick::Passed;
+        }
+        self.close(&at, reached, upper, pended);
+        if reached { return Tick::Crossing(at); }
+        self.settle();
+        Tick::Passed
+    }
+
+    /// The sweep's position: the least time any source still offers.
+    ///
+    /// The TOTAL order, not the partial one. Every time `close` produces is strictly greater than
+    /// the position that produced it, so new work only ever lands ahead of here and the sweep never
+    /// revisits.
+    fn frontier(&self) -> Option<T> {
+        [
+            self.novel.time(), self.due.get(self.due_pos), self.input.time(),
+            self.output.time(), self.synth.last(),
+        ].into_iter().flatten().min().cloned()
+    }
+
+    /// Step every source sitting at `at`, and decide whether `at` is REACHED.
+    ///
+    /// Reached is clause two of `round_coverage` — `∃ nu ∈ novel, nu ≤ at` — evaluated
+    /// incrementally: either a seed lands exactly here, or one already stepped in lies below.
+    ///
+    /// Input and output are stepped whether or not `at` is reached, and that is forced rather than
+    /// eager: they are sources of the frontier, so leaving them would stall the sweep, and their
+    /// edits must reach the buffers or they are lost to every later accumulation. Stepping only
+    /// moves an edit across; it consolidates nothing. The expensive part — `advance_buffer_by`,
+    /// which joins every buffered time and re-consolidates — is deferred to `close`, and happens
+    /// only where the buffers are actually read.
+    fn absorb(&mut self, at: &T) -> bool {
+        self.input.step_while_time_is(at);
+        self.output.step_while_time_is(at);
+
+        // A novel update here is a seed.
+        let mut reached = self.novel.step_while_time_is(at);
+        if reached {
+            if let Some(meet) = self.meet.as_ref() { self.novel.advance_buffer_by(meet); }
+        }
+        // So is a synthetic join scheduled for here, or a due time carried in. Both move into the
+        // reached set, where they become witnesses and join partners for later times.
+        while self.synth.last() == Some(at) {
+            self.reached.push(self.synth.pop().expect("nonempty"));
+            reached = true;
+        }
+        while self.due.get(self.due_pos) == Some(at) {
+            self.reached.push(at.clone());
+            self.due_pos += 1;
+            reached = true;
+        }
+        // Absorption: a time at or above a seed already stepped in is itself reached, because
+        // joining that seed with it yields it back. Checked against the stepped prefixes only.
+        reached
+            || self.novel.buffer().iter().any(|((_, t), _)| t.less_equal(at))
+            || self.reached.iter().any(|t| t.less_equal(at))
+    }
+
+    /// Close `at` forward under joins — clause one of `round_coverage`, the join-closure.
+    ///
+    /// Against the NOVEL times always, reached or not: an unreached time joined with a novel time
+    /// lands at or above that novel time, so it carries a witness and is on the schedule.
+    ///
+    /// Against the PRIOR times only when `at` is itself reached, because the join then inherits
+    /// `at`'s witness. A join of two prior times carries none and is deliberately never produced —
+    /// that asymmetry is the whole of why an incremental operator does less work than the closure
+    /// of everything.
+    ///
+    /// `produced` counts as prior. A correction emitted at `p` changes the accumulated output at
+    /// every time at or above `p`, so `p ∨ at` has to be visited; nothing else covers it, since
+    /// this round's corrections are not in the output history and `at` was not yet stepped in when
+    /// the sweep passed `p`.
+    fn close(&mut self, at: &T, reached: bool, upper: &Antichain<T>, pended: &mut Vec<T>) {
+        self.temporary.extend(self.novel.buffer().iter().map(|((_, t), _)| t)
+            .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+        self.temporary.extend(self.reached.iter()
+            .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+        if reached {
+            if let Some(meet) = self.meet.as_ref() {
+                self.input.advance_buffer_by(meet);
+                self.output.advance_buffer_by(meet);
+            }
+            self.temporary.extend(self.input.buffer().iter().map(|((_, t), _)| t)
+                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+            self.temporary.extend(self.output.buffer().iter().map(|((_, t), _)| t)
+                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+            self.temporary.extend(self.produced.iter().map(|((_, t), _)| t)
+                .filter(|t| !t.less_equal(at)).map(|t| t.join(at)));
+        }
+        sort_dedup(&mut self.temporary);
+        let before = self.synth.len();
+        for time in self.temporary.drain(..) {
+            if upper.less_equal(&time) { pended.push(time); } else { self.synth.push(time); }
+        }
+        if self.synth.len() > before {
+            self.synth.sort_by(|x, y| y.cmp(x));
+            self.synth.dedup();
+        }
+    }
+
+    /// The input accumulation at the suspended time: both input runs, meeting only here.
+    fn input_at(&self, at: &T, into: &mut Vec<(u64, RIn)>) {
+        for ((id, time), diff) in self.input.buffer().iter().chain(self.novel.buffer().iter()) {
+            if time.less_equal(at) { into.push((*id, diff.clone())); }
+        }
+        crate::consolidation::consolidate(into);
+    }
+
+    /// The tentative output accumulation at the suspended time, including this sweep's corrections.
+    fn output_at(&self, at: &T, into: &mut Vec<(u64, ROut)>) {
+        for ((id, time), diff) in self.output.buffer().iter().chain(self.produced.iter()) {
+            if time.less_equal(at) { into.push((*id, diff.clone())); }
+        }
+        crate::consolidation::consolidate(into);
+    }
+
+    /// Record the corrections evaluated at the suspended time, and collapse them by the meet.
+    fn commit(&mut self, at: &T, corrections: impl Iterator<Item = (u64, ROut)>) {
+        let before = self.produced.len();
+        for (id, diff) in corrections { self.produced.push(((id, at.clone()), diff)); }
+        if self.produced.len() > before {
+            if let Some(meet) = self.meet.as_ref() {
+                for entry in self.produced.iter_mut() { (entry.0).1.join_assign(meet); }
+            }
+            crate::consolidation::consolidate(&mut self.produced);
+        }
+    }
+
+    /// Close a step: recompute the meet of everything still to come, and compact the reached set by
+    /// it. This is what keeps a key with a long history linear rather than quadratic.
+    fn settle(&mut self) {
+        let mut meet: Option<T> = None;
+        update_meet(&mut meet, self.novel.meet());
+        update_meet(&mut meet, self.input.meet());
+        update_meet(&mut meet, self.output.meet());
+        for time in self.synth.iter() { update_meet(&mut meet, Some(time)); }
+        update_meet(&mut meet, self.due_meets.get(self.due_pos));
+        if let Some(m) = meet.as_ref() {
+            for time in self.reached.iter_mut() { *time = time.join(m); }
+        }
+        sort_dedup(&mut self.reached);
+        self.meet = meet;
+    }
+}
diff --git a/differential-dataflow/src/operators/mod.rs b/differential-dataflow/src/operators/mod.rs
index db7cd25cf..944bfe73f 100644
--- a/differential-dataflow/src/operators/mod.rs
+++ b/differential-dataflow/src/operators/mod.rs
@@ -9,7 +9,6 @@ pub use self::count::CountTotal;
 pub use self::threshold::ThresholdTotal;
 
 pub mod arrange;
-pub mod common;
 pub mod int_proxy;
 pub mod reduce;
 pub mod iterate;
@@ -90,8 +89,8 @@ impl<V: Copy, T: Ord + Lattice, D: crate::difference::Semigroup> EditList<V, T,
 }
 
 /// A loaded, time-ordered replay of one key's `(value, time, diff)` edits, with meet-advanced
-/// buffer collapse — the shared machinery under the cursor reduce, the `int_proxy` tactics, and
-/// the [`common`](crate::operators::common) helpers. Its local contract: after
+/// buffer collapse — the shared machinery under the cursor reduce and the `int_proxy` tactics.
+/// Its local contract: after
 /// `advance_buffer_by(meet)` with the meet of the un-replayed times, the buffer is consolidated
 /// and replay cost stays linear in edits rather than quadratic.
 pub struct ValueHistory<V, T, D> {

From b8e4f20d1ab998a1a4dfc14202be2aab7cadc9c2 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Tue, 11 Aug 2026 16:44:05 -0400
Subject: [PATCH 3/3] Tidy two clippy warnings in the fused sweep

The key-slot loop tested its terminating condition with a `let ... else { break }` where a `while let` says the same thing, and the step-live-keys loop ranged over indices into `live` only to index `live`, shadowing the loop variable with the value it read.

`join_key`'s `too_many_arguments` is left alone: it predates this branch.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../src/operators/int_proxy/reduce.rs            | 16 +++++++---------
 1 file changed, 7 insertions(+), 9 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index a7143fed4..736fcf817 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -262,13 +262,12 @@ where
             let mut n_slots = 0usize;
             let (mut is, mut ns, mut os) = (0usize, 0usize, 0usize);
             live.clear();
-            loop {
-                // Mapped to hashes before the min: the three runs differ in their diff type.
-                let Some(key) = [
-                    p_in.get(is).map(|record| record.0.0),
-                    p_nv.get(ns).map(|record| record.0.0),
-                    p_out.get(os).map(|record| record.0.0),
-                ].into_iter().flatten().min() else { break };
+            // Mapped to hashes before the min: the three runs differ in their diff type.
+            while let Some(key) = [
+                p_in.get(is).map(|record| record.0.0),
+                p_nv.get(ns).map(|record| record.0.0),
+                p_out.get(os).map(|record| record.0.0),
+            ].into_iter().flatten().min() {
                 let i0 = is;
                 while is < p_in.len() && p_in[is].0.0 == key { is += 1; }
                 let i1 = is;
@@ -344,8 +343,7 @@ where
                 }
 
                 // Step every live key past the time it was suspended at, and retire the spent ones.
-                for si in 0..live.len() {
-                    let si = live[si];
+                for &si in live.iter() {
                     let slot = &mut slots[si];
                     slot.at = slot.sweep.next_crossing(upper, &mut slot.pended);
                     if slot.at.is_none() && !slot.pended.is_empty() {