diff --git a/differential-dataflow/src/operators/cursor/history.rs b/differential-dataflow/src/operators/cursor/history.rs
index f4592b7de..b489858b5 100644
--- a/differential-dataflow/src/operators/cursor/history.rs
+++ b/differential-dataflow/src/operators/cursor/history.rs
@@ -3,6 +3,7 @@
 use crate::lattice::Lattice;
 use crate::operators::history::{EditList, HistoryReplay, ValueHistory};
 use crate::trace::Cursor;
+use crate::trace::implementations::containers::BatchContainer;
 
 /// Walks the cursor's values at the current key into `target`, advancing times by `meet` if supplied.
 fn load_values<'a, V, T, D, C>(
@@ -47,6 +48,44 @@ where
     load_values(history.edits_mut(), cursor, storage, meet);
 }
 
+/// Loads the cursor's values at its current key into `values`, and their edits into `history`
+/// keyed by each value's index in `values`.
+///
+/// The indices are what let the history outlive the borrow of `storage`: a history of
+/// `Cursor::Val<'a>` can only live as long as the storage it reads from, and so cannot be held
+/// across calls by an iterator that owns that storage. A history of indices can, at the cost of
+/// copying the key's values into a container the iterator owns too. Callers that replay a key in
+/// one uninterrupted pass should prefer `load_current`, which copies nothing.
+///
+/// The cursor is left with its values exhausted, as `load_current` leaves it.
+pub(super) fn load_current_indexed<T, D, C>(
+    values: &mut C::ValContainer,
+    history: &mut ValueHistory<usize, T, D>,
+    cursor: &mut C,
+    storage: &C::Storage,
+    meet: Option<&T>,
+)
+where
+    T: Ord + Clone + Lattice,
+    D: crate::difference::Semigroup,
+    C: Cursor<Time = T, Diff = D>,
+{
+    history.clear();
+    values.clear();
+    let edits = history.edits_mut();
+    while let Some(val) = cursor.get_val(storage) {
+        let index = values.len();
+        values.push_ref(val);
+        cursor.map_times(storage, |time, diff| {
+            let mut time = C::owned_time(time);
+            if let Some(meet) = meet { time.join_assign(meet); }
+            edits.push(time, C::owned_diff(diff));
+        });
+        edits.seal(index);
+        cursor.step_val(storage);
+    }
+}
+
 /// Loads and replays a specified key.
 ///
 /// If the key is absent, the replayed history will be empty.
diff --git a/differential-dataflow/src/operators/cursor/join.rs b/differential-dataflow/src/operators/cursor/join.rs
index cf4e41326..890c01ace 100644
--- a/differential-dataflow/src/operators/cursor/join.rs
+++ b/differential-dataflow/src/operators/cursor/join.rs
@@ -9,11 +9,11 @@ use timely::ContainerBuilder;
 use timely::dataflow::Stream;
 use timely::progress::Timestamp;
 
-use super::history::load_current;
+use super::history::{load_current, load_current_indexed};
 use crate::lattice::Lattice;
 use crate::operators::arrange::Arranged;
 use crate::operators::history::ValueHistory;
-use crate::operators::join::{Fresh, JoinTactic, join_with_tactic};
+use crate::operators::join::{Fresh, JoinTactic, KEY_WORK_LIMIT, join_with_tactic};
 use crate::trace::{BatchCursor, BatchDiff, BatchVal, Cursor, Navigable, TraceReader};
 use crate::trace::cursor::cursor_list;
 use crate::trace::implementations::containers::BatchContainer;
@@ -117,6 +117,11 @@ where
             logic: Rc::clone(&self.logic),
             builder: CB::default(),
             ready: VecDeque::new(),
+            vals1: BatchContainer::with_capacity(0),
+            vals2: BatchContainer::with_capacity(0),
+            history1: ValueHistory::new(),
+            history2: ValueHistory::new(),
+            resume: None,
             done: false,
         })
     }
@@ -130,6 +135,14 @@ where
 /// the merge-join forward until the builder yields a container (or the cursors run dry), matching the
 /// former per-unit `work` loop but suspending at container boundaries rather than under a fuel budget:
 /// the driver stops pulling once its budget is spent and resumes the same iterator next activation.
+///
+/// A container boundary can fall in the middle of a key, and a single key's cross product can be
+/// arbitrarily larger than the driver's whole budget, so suspending only between keys would let one
+/// key buffer unboundedly. Keys are therefore handled by one of two strategies, chosen once the key's
+/// edits are loaded and its work is known: a key whose cross product fits [`KEY_WORK_LIMIT`] runs to
+/// completion in a single call, over histories of borrowed values; a larger key is reloaded into the
+/// iterator's own `vals`/`history` fields, which are indexed rather than borrowed and so survive a
+/// suspension, and is then replayed by [`replay_suspending`] a container at a time.
 struct DeferredIter<T, C1, C2, L, CB>
 where
     T: Timestamp+Lattice,
@@ -152,9 +165,36 @@ where
     builder: CB,
     /// Completed containers awaiting a `next` call.
     ready: VecDeque<CB::Container>,
+    /// The suspending strategy's copy of the open key's values, which its histories index.
+    vals1: C1::ValContainer,
+    vals2: C2::ValContainer,
+    /// The suspending strategy's histories for the open key, over indices into `vals1`/`vals2`.
+    history1: ValueHistory<usize, T, C1::Diff>,
+    history2: ValueHistory<usize, T, C2::Diff>,
+    /// Where in the open key's replay this iterator suspended; `None` when no key is open.
+    resume: Option<Resume>,
     done: bool,
 }
 
+/// Progress through one key, for the strategy that suspends within a key.
+///
+/// The variants mirror the two strategies of [`JoinThinker::think`]: the nested cross product it
+/// uses for small histories, and the time-ordered wave it uses for large ones. A wave step advances
+/// one history's buffer before emitting against it, so the step it is part-way through is part of
+/// the state: resuming must not advance that buffer a second time, which would consolidate it and
+/// invalidate the offset resumed from.
+#[derive(Clone, Copy)]
+enum Resume {
+    /// The nested cross product, at edit indices `(index1, index2)` of the two histories.
+    Simple(usize, usize),
+    /// The wave, between steps; the next step follows from the histories' least times.
+    WaveChoose,
+    /// The wave, emitting `history2`'s buffer against `history1`'s next edit, from this offset.
+    WaveStep1(usize),
+    /// The wave, emitting `history1`'s buffer against `history2`'s next edit, from this offset.
+    WaveStep2(usize),
+}
+
 impl<T, C1, C2, L, CB> Iterator for DeferredIter<T, C1, C2, L, CB>
 where
     T: Timestamp+Lattice,
@@ -182,6 +222,11 @@ where
         let storage2 = &self.storage2;
         let cursor1 = &mut self.cursor1;
         let cursor2 = &mut self.cursor2;
+        let vals1 = &mut self.vals1;
+        let vals2 = &mut self.vals2;
+        let history1 = &mut self.history1;
+        let history2 = &mut self.history2;
+        let resume = &mut self.resume;
         let builder = &mut self.builder;
         let ready = &mut self.ready;
         let mut logic = self.logic.borrow_mut();
@@ -197,23 +242,76 @@ where
                     Ordering::Greater => cursor2.seek_key(storage2, key1),
                     Ordering::Equal => {
 
-                        load_current(&mut thinker.history1, cursor1, storage1, meet1);
-                        load_current(&mut thinker.history2, cursor2, storage2, meet2);
+                        // A key is opened once and then either retired in this iteration or suspended
+                        // part-way; the cursors are not stepped until it retires, so a suspended key is
+                        // still the key both cursors report on re-entry.
+                        if resume.is_none() {
+                            load_current(&mut thinker.history1, cursor1, storage1, meet1);
+                            load_current(&mut thinker.history2, cursor2, storage2, meet2);
 
-                        thinker.think(|v1,v2,t,r1,r2| {
-                            logic(key1, v1, v2, t, r1, r2, builder);
-                        });
+                            let len1 = thinker.history1.edit_len();
+                            let len2 = thinker.history2.edit_len();
+                            if len1.saturating_mul(len2) <= KEY_WORK_LIMIT {
+                                thinker.think(|v1,v2,t,r1,r2| {
+                                    logic(key1, v1, v2, t, r1, r2, builder);
+                                });
 
-                        cursor1.step_key(storage1);
-                        cursor2.step_key(storage2);
+                                cursor1.step_key(storage1);
+                                cursor2.step_key(storage2);
 
-                        thinker.history1.clear();
-                        thinker.history2.clear();
+                                thinker.history1.clear();
+                                thinker.history2.clear();
 
-                        // Move any completed containers aside; we yield them one at a time.
-                        while let Some(container) = builder.extract() {
-                            // Avoiding the mem::take would require a non-iterator trait.
-                            ready.push_back(std::mem::take(container));
+                                // Move any completed containers aside; we yield them one at a time.
+                                drain_builder(builder, ready);
+                                continue;
+                            }
+
+                            // The key can outrun the budget, so reload it into histories that survive a
+                            // suspension. The strategy is chosen from the same edit counts `think` uses,
+                            // so both paths visit a key's matches in the same order.
+                            let simple = len1 < 10 || len2 < 10;
+                            thinker.history1.clear();
+                            thinker.history2.clear();
+                            cursor1.rewind_vals(storage1);
+                            cursor2.rewind_vals(storage2);
+                            load_current_indexed(vals1, history1, cursor1, storage1, meet1);
+                            load_current_indexed(vals2, history2, cursor2, storage2, meet2);
+                            *resume = Some(if simple {
+                                Resume::Simple(0, 0)
+                            } else {
+                                history1.build();
+                                history2.build();
+                                Resume::WaveChoose
+                            });
+                        }
+
+                        let state = resume.as_mut().expect("key is open");
+                        let retired = {
+                            let (loaded1, loaded2) = (&*vals1, &*vals2);
+                            replay_suspending(history1, history2, state, |index1, index2, t, r1, r2| {
+                                // `logic` takes its key and its values at one lifetime, and the values
+                                // here are borrowed from `vals1`/`vals2` rather than from `storage1`.
+                                // Reborrowing the key narrows it to match, instead of holding the
+                                // values' borrow open for as long as the storage's.
+                                let key1 = <C1::KeyContainer as BatchContainer>::reborrow(key1);
+                                logic(key1, loaded1.index(index1), loaded2.index(index2), t, r1, r2, builder);
+                                // Move any completed containers aside; we yield them one at a time.
+                                drain_builder(builder, ready)
+                            })
+                        };
+
+                        if retired {
+                            cursor1.step_key(storage1);
+                            cursor2.step_key(storage2);
+
+                            history1.clear();
+                            history2.clear();
+                            vals1.clear();
+                            vals2.clear();
+                            *resume = None;
+
+                            drain_builder(builder, ready);
                         }
                     }
                 },
@@ -235,6 +333,110 @@ where
     }
 }
 
+/// Moves every container the builder has completed into `ready`, reporting whether any moved.
+fn drain_builder<CB>(builder: &mut CB, ready: &mut VecDeque<CB::Container>) -> bool
+where
+    CB: ContainerBuilder<Container: Default>,
+{
+    let mut extracted = false;
+    while let Some(container) = builder.extract() {
+        // Avoiding the mem::take would require a non-iterator trait.
+        ready.push_back(std::mem::take(container));
+        extracted = true;
+    }
+    extracted
+}
+
+/// Replays one key's loaded histories, emitting each match, until the key is done or `emit` stops it.
+///
+/// This produces the same matches in the same order as [`JoinThinker::think`], over histories of value
+/// indices rather than of values. It differs in that it can be interrupted: `emit` returns `true` to
+/// suspend, `replay_suspending` then returns `false` having recorded in `resume` where it stopped, and
+/// a later call resumes from there. It returns `true` once the key is fully replayed, at which point
+/// `resume` is spent and the caller may retire the key.
+fn replay_suspending<T, D1, D2, F>(
+    history1: &mut ValueHistory<usize, T, D1>,
+    history2: &mut ValueHistory<usize, T, D2>,
+    resume: &mut Resume,
+    mut emit: F,
+) -> bool
+where
+    T: Ord + Clone + Lattice,
+    D1: Clone + crate::difference::Semigroup,
+    D2: Clone + crate::difference::Semigroup,
+    F: FnMut(usize, usize, T, &D1, &D2) -> bool,
+{
+    loop {
+        match *resume {
+            Resume::Simple(mut index1, mut index2) => {
+                let len1 = history1.edit_len();
+                let len2 = history2.edit_len();
+                while index1 < len1 {
+                    while index2 < len2 {
+                        let (val1, time1, diff1) = history1.edit_at(index1);
+                        let (val2, time2, diff2) = history2.edit_at(index2);
+                        let suspend = emit(val1, val2, time1.join(time2), diff1, diff2);
+                        index2 += 1;
+                        if suspend {
+                            *resume = Resume::Simple(index1, index2);
+                            return false;
+                        }
+                    }
+                    index2 = 0;
+                    index1 += 1;
+                }
+                return true;
+            }
+            Resume::WaveChoose => {
+                // Step whichever history holds the earlier un-replayed time, and once one is spent,
+                // whichever remains. A tie steps the second, as `think`'s comparison does.
+                let step1 = match (history1.time(), history2.time()) {
+                    (Some(time1), Some(time2)) => time1.cmp(time2) == Ordering::Less,
+                    (Some(_), None) => true,
+                    (None, Some(_)) => false,
+                    (None, None) => return true,
+                };
+                if step1 {
+                    history2.advance_buffer_by(history1.meet().unwrap());
+                    *resume = Resume::WaveStep1(0);
+                }
+                else {
+                    history1.advance_buffer_by(history2.meet().unwrap());
+                    *resume = Resume::WaveStep2(0);
+                }
+            }
+            Resume::WaveStep1(mut offset) => {
+                while offset < history2.buffer().len() {
+                    let (val1, time1, diff1) = history1.edit().unwrap();
+                    let ((val2, time2), diff2) = &history2.buffer()[offset];
+                    let suspend = emit(val1, *val2, time1.join(time2), diff1, diff2);
+                    offset += 1;
+                    if suspend {
+                        *resume = Resume::WaveStep1(offset);
+                        return false;
+                    }
+                }
+                history1.step();
+                *resume = Resume::WaveChoose;
+            }
+            Resume::WaveStep2(mut offset) => {
+                while offset < history1.buffer().len() {
+                    let ((val1, time1), diff1) = &history1.buffer()[offset];
+                    let (val2, time2, diff2) = history2.edit().unwrap();
+                    let suspend = emit(*val1, val2, time1.join(time2), diff1, diff2);
+                    offset += 1;
+                    if suspend {
+                        *resume = Resume::WaveStep2(offset);
+                        return false;
+                    }
+                }
+                history2.step();
+                *resume = Resume::WaveChoose;
+            }
+        }
+    }
+}
+
 struct JoinThinker<V1, V2, T, D1, D2> {
     pub history1: ValueHistory<V1, T, D1>,
     pub history2: ValueHistory<V2, T, D2>,
@@ -315,3 +517,180 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+
+    use std::cell::Cell;
+    use std::rc::Rc;
+
+    use timely::container::CapacityContainerBuilder;
+    use timely::container::PushInto;
+
+    use super::{CursorTactic, KEY_WORK_LIMIT};
+    use crate::operators::join::{Fresh, JoinTactic};
+    use crate::trace::Builder;
+    use crate::trace::implementations::ValBuilder;
+    use crate::trace::implementations::ord_neu::OrdValBatch;
+    use crate::trace::implementations::Vector;
+
+    type Update = ((u64, u64), u64, isize);
+    type Batch = Rc<OrdValBatch<Vector<Update>>>;
+    type Output = CapacityContainerBuilder<Vec<Update>>;
+
+    /// A batch holding `vals` values under the single key `0`, each at time `0` with diff `1`.
+    fn one_key(vals: u64) -> Batch {
+        let mut updates: Vec<Update> = (0..vals).map(|val| ((0, val), 0, 1)).collect();
+        let mut builder = <ValBuilder<u64, u64, u64, isize>>::default();
+        builder.push(&mut updates);
+        Rc::new(builder.done().expect("non-empty batch"))
+    }
+
+    /// Joins two single-key batches, reporting the first container, the total matches produced by
+    /// the time it arrived, and the whole run's matches as `(pair count, pair checksum)`.
+    fn join_one_key(vals1: u64, vals2: u64) -> (usize, usize, (usize, u64)) {
+        let produced = Rc::new(Cell::new(0));
+        let counter = Rc::clone(&produced);
+        let mut tactic = CursorTactic::<_, _, _, Output>::new(
+            move |_key: &u64, val1: &u64, val2: &u64, time, diff1: &isize, diff2: &isize, out: &mut Output| {
+                counter.set(counter.get() + 1);
+                out.push_into(((*val1, *val2), time, diff1 * diff2));
+            },
+        );
+
+        let mut work = tactic.prep(vec![one_key(vals1)], vec![one_key(vals2)], Fresh::Input1, 0);
+        let first = work.next().expect("at least one container");
+        let (first_len, produced_by_first) = (first.len(), produced.get());
+
+        let mut count = 0;
+        let mut checksum = 0;
+        for container in std::iter::once(first).chain(work) {
+            for ((val1, val2), time, diff) in container {
+                assert_eq!((time, diff), (0, 1));
+                count += 1;
+                checksum += val1 * vals2 + val2;
+            }
+        }
+        (first_len, produced_by_first, (count, checksum))
+    }
+
+    /// The matches of a `vals1` by `vals2` cross product of distinct values: one per pair, and each
+    /// pair exactly once, which the sum over `val1 * vals2 + val2` pins down.
+    fn expected(vals1: u64, vals2: u64) -> (usize, u64) {
+        let pairs = vals1 * vals2;
+        (pairs as usize, pairs * (pairs - 1) / 2)
+    }
+
+    /// A key large enough to outrun the driver's budget suspends within itself, rather than
+    /// buffering its whole cross product before the first container is served. Both sides carry
+    /// enough edits for the time-ordered wave.
+    #[test]
+    fn wave_suspends_within_a_key() {
+        let vals = (KEY_WORK_LIMIT as f64).sqrt() as u64 + 1;
+        assert!((vals * vals) as usize > KEY_WORK_LIMIT);
+
+        let (first_len, produced_by_first, totals) = join_one_key(vals, vals);
+        assert!(produced_by_first <= first_len + 1, "produced {produced_by_first} for a container of {first_len}");
+        assert_eq!(totals, expected(vals, vals));
+    }
+
+    /// The same, for the nested cross product the join uses when one side is small: few edits on a
+    /// side does not bound the work, only the other side's edit count does.
+    #[test]
+    fn nested_suspends_within_a_key() {
+        let vals1 = 5;
+        let vals2 = KEY_WORK_LIMIT as u64 / vals1 + 1;
+        assert!((vals1 * vals2) as usize > KEY_WORK_LIMIT);
+
+        let (first_len, produced_by_first, totals) = join_one_key(vals1, vals2);
+        assert!(produced_by_first <= first_len + 1, "produced {produced_by_first} for a container of {first_len}");
+        assert_eq!(totals, expected(vals1, vals2));
+    }
+
+    /// A batch holding `vals` values under the single key `0`, each present at `times` distinct
+    /// times, with diffs that differ per value and time so a dropped or duplicated match shows up.
+    fn one_key_over_time(vals: u64, times: u64) -> Batch {
+        let mut updates: Vec<Update> = Vec::new();
+        for val in 0..vals {
+            for time in 0..times {
+                updates.push(((0, val), time, 1 + isize::try_from(val + time).unwrap() % 3));
+            }
+        }
+        let mut builder = <ValBuilder<u64, u64, u64, isize>>::default();
+        builder.push(&mut updates);
+        Rc::new(builder.done().expect("non-empty batch"))
+    }
+
+    /// The `(value, time, diff)` edits `one_key_over_time` produces, as the join sees them.
+    fn edits_over_time(vals: u64, times: u64) -> Vec<(u64, u64, isize)> {
+        (0..vals)
+            .flat_map(|val| (0..times).map(move |time| (val, time, 1 + isize::try_from(val + time).unwrap() % 3)))
+            .collect()
+    }
+
+    /// Accumulates updates into a sorted, consolidated list, dropping those that cancel.
+    fn consolidate(mut updates: Vec<Update>) -> Vec<Update> {
+        updates.sort();
+        let mut out: Vec<Update> = Vec::new();
+        for (data, time, diff) in updates {
+            match out.last_mut() {
+                Some(last) if (last.0, last.1) == (data, time) => last.2 += diff,
+                _ => out.push((data, time, diff)),
+            }
+        }
+        out.retain(|update| update.2 != 0);
+        out
+    }
+
+    /// A key with many distinct times exercises the wave's buffer advance and consolidation, which
+    /// a suspension must resume without repeating. The join's output must consolidate to the same
+    /// collection as the cross product of the two sides' edits, with times joined and diffs
+    /// multiplied.
+    #[test]
+    fn suspended_wave_matches_the_cross_product() {
+        let (vals, times) = (200, 8);
+        assert!(((vals * times) * (vals * times)) as usize > KEY_WORK_LIMIT);
+
+        let mut tactic = CursorTactic::<_, _, _, Output>::new(
+            |_key: &u64, val1: &u64, val2: &u64, time, diff1: &isize, diff2: &isize, out: &mut Output| {
+                out.push_into(((*val1, *val2), time, diff1 * diff2));
+            },
+        );
+        let work = tactic.prep(
+            vec![one_key_over_time(vals, times)],
+            vec![one_key_over_time(vals, times)],
+            Fresh::Input1,
+            0,
+        );
+        let produced: Vec<Update> = work.flatten().collect();
+
+        let mut expected = Vec::new();
+        for (val1, time1, diff1) in edits_over_time(vals, times) {
+            for (val2, time2, diff2) in edits_over_time(vals, times) {
+                expected.push(((val1, val2), time1.max(time2), diff1 * diff2));
+            }
+        }
+
+        // The collections run to millions of updates, so report the first difference rather than
+        // both of them in full.
+        let (produced, expected) = (consolidate(produced), consolidate(expected));
+        let differs = produced.iter().zip(expected.iter()).position(|(x, y)| x != y);
+        assert_eq!(
+            differs.map(|at| (at, produced[at], expected[at])),
+            None,
+            "consolidated join output differs from the cross product",
+        );
+        assert_eq!(produced.len(), expected.len());
+    }
+
+    /// A key whose cross product fits the budget runs to completion in one call, and still produces
+    /// exactly its cross product.
+    #[test]
+    fn small_key_runs_in_one_call() {
+        let vals = 64;
+        assert!((vals * vals) as usize <= KEY_WORK_LIMIT);
+
+        let (_, _, totals) = join_one_key(vals, vals);
+        assert_eq!(totals, expected(vals, vals));
+    }
+}
diff --git a/differential-dataflow/src/operators/history.rs b/differential-dataflow/src/operators/history.rs
index c21386dc4..f4120c071 100644
--- a/differential-dataflow/src/operators/history.rs
+++ b/differential-dataflow/src/operators/history.rs
@@ -39,6 +39,15 @@ impl<V: Copy, T: Ord + Lattice, D: crate::difference::Semigroup> EditList<V, T,
             self.values.push((value, self.edits.len()));
         }
     }
+    /// The `index`th edit, as `(value, time, diff)`, in the order `map` visits them.
+    ///
+    /// The value is found by binary search over the per-value edit bounds, which `seal` leaves
+    /// ascending.
+    fn index(&self, index: usize) -> (V, &T, &D) {
+        let value_index = self.values.partition_point(|&(_, upper)| upper <= index);
+        let (time, diff) = &self.edits[index];
+        (self.values[value_index].0, time, diff)
+    }
     fn map<F: FnMut(V, &T, &D)>(&self, mut logic: F) {
         for index in 0 .. self.values.len() {
             let lower = if index == 0 { 0 } else { self.values[index-1].1 };
@@ -89,8 +98,16 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
         self.edits.map(logic)
     }
 
+    /// The `index`th edit, as `(value, time, diff)`, in the order `map_edits` visits them.
+    ///
+    /// Indexed access lets a reader replay the edits a few at a time, rather than in one
+    /// closure-driven sweep it cannot interrupt.
+    pub(in crate::operators) fn edit_at(&self, index: usize) -> (V, &T, &D) {
+        self.edits.index(index)
+    }
+
     /// Organizes history based on current contents of edits (sort + suffix meets).
-    fn build(&mut self) {
+    pub(in crate::operators) fn build(&mut self) {
         self.buffer.clear();
         self.history.clear();
         for value_index in 0 .. self.edits.values.len() {
diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
index 31986ca00..8b6715338 100644
--- a/differential-dataflow/src/operators/int_proxy/join.rs
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -11,7 +11,7 @@ use timely::progress::Timestamp;
 use crate::difference::{Multiply, Semigroup};
 use crate::lattice::Lattice;
 use super::ProxyBridge;
-use crate::operators::join::{Fresh, JoinTactic};
+use crate::operators::join::{Fresh, JoinTactic, KEY_WORK_LIMIT};
 use crate::operators::history::ValueHistory;
 
 use super::history::IdHistory;
@@ -44,10 +44,11 @@ pub trait ProxyJoinBackend<T, B0, B1> {
         bridge1: &mut ProxyBridge<T, Self::R1>,
     );
 
-    /// Interpret matches derived from the immediately preceding [`Self::advance`] call and place
-    /// them in `output`. The iterator calls `cross` before another `advance`, so a backend may keep
-    /// block-local interpretation state between the two calls. `cross` may be skipped when the
-    /// block produced no matches, in which case the next `advance` may overwrite that state.
+    /// Interpret matches derived from the preceding [`Self::advance`] call and place them in
+    /// `output`. A block is matched in bounded steps, so `cross` may be called several times for
+    /// one `advance`, and is skipped for a step that produced no matches. The iterator always makes
+    /// those calls before the next `advance`, so a backend may keep block-local interpretation state
+    /// across them, and the next `advance` may overwrite it.
     fn cross(
         &mut self,
         instance: &JoinInstance<T, B0, B1>,
@@ -114,6 +115,9 @@ where
             p1: Vec::new(),
             h0: IdHistory::new(),
             h1: IdHistory::new(),
+            at0: 0,
+            at1: 0,
+            resume: None,
             matches: JoinMatches::default(),
             ready: Vec::new(),
         })
@@ -125,6 +129,13 @@ where
 /// The iterator draws the proxy collection from the back-end a block at a time (`Bk::advance`).
 /// Each block is then translated to output updates with joined times and multiplied differences,
 /// which are provided to the back-end to translate into output containers, which are then returned.
+///
+/// A block is not matched in one go. `advance` reports a key entirely within the block that first
+/// mentions it, so a block is at least one whole key, and one key's cross product is bounded by
+/// nothing; matching a whole block would buffer it all before the back-end saw any of it. Matching
+/// therefore stops once the buffered matches reach [`KEY_WORK_LIMIT`], even part-way through a key,
+/// and `at0`/`at1`/`resume` record where to pick up. The state is all identifiers, times and diffs
+/// the iterator owns, so unlike the cursor tactic it needs nothing reloaded to survive a suspension.
 struct ProxyJoinIter<T, B0, B1, Bk>
 where
     Bk: ProxyJoinBackend<T, B0, B1>,
@@ -142,6 +153,11 @@ where
     /// Per-key replay histories, held across the iterator and reloaded per key when needed.
     h0: IdHistory<T, Bk::R0>,
     h1: IdHistory<T, Bk::R1>,
+    /// How far into the current block's two runs the matching has come.
+    at0: usize,
+    at1: usize,
+    /// Where in the open key's replay the matching suspended; `None` when no key is open.
+    resume: Option<Resume>,
     /// The block's matched records, held across blocks to keep their allocations.
     matches: JoinMatches<T, Bk::ROut>,
     /// The last block's containers, in reverse, served from the back one `next` at a time.
@@ -155,10 +171,13 @@ where
 {
     type Item = Bk::Output;
 
-    /// Serve a ready container, else draw and cross blocks until one yields any.
+    /// Serve a ready container, else match and cross until one yields any.
     fn next(&mut self) -> Option<Bk::Output> {
-        while self.ready.is_empty() && self.from.is_some() {
-            self.refill();
+        while self.ready.is_empty() {
+            if self.block_done() {
+                if self.from.is_none() { break; }
+                self.refill();
+            }
             self.work();
             if !self.matches.ids.is_empty() { self.cross(); }
         }
@@ -171,10 +190,18 @@ where
     T: Timestamp + Lattice,
     Bk: ProxyJoinBackend<T, B0, B1>,
 {
+    /// Whether the current block has been matched in full.
+    fn block_done(&self) -> bool {
+        self.resume.is_none() && (self.at0 >= self.p0.len() || self.at1 >= self.p1.len())
+    }
+
     /// Draw the next block from the backend.
     fn refill(&mut self) {
+        debug_assert!(self.block_done(), "a block is drawn only once the last one is matched");
         self.p0.clear();
         self.p1.clear();
+        self.at0 = 0;
+        self.at1 = 0;
         let before = self.from;
         self.backend.borrow_mut().advance(&self.instance, &mut self.from, &mut self.p0, &mut self.p1);
         // Without progress the iterator would never retire, so this guards liveness as well as contract.
@@ -196,25 +223,27 @@ where
         );
     }
 
-    /// Match the whole of the current block into the match buffers.
+    /// Match the current block into the match buffers, until it is drained or the buffers are full.
     fn work(&mut self) {
         // Disjoint field borrows, as `join_key` holds the bridges and the buffers at once.
         let (p0, p1) = (&self.p0, &self.p1);
         let (h0, h1) = (&mut self.h0, &mut self.h1);
 
-        let (mut i, mut j) = (0usize, 0usize);
-        while i < p0.len() && j < p1.len() {
-            let ki = p0[i].0.0;
-            debug_assert_eq!(ki, p1[j].0.0, "advance must report common keys");
-            let mut e0 = i;
+        while self.at0 < p0.len() && self.at1 < p1.len() {
+            let ki = p0[self.at0].0.0;
+            debug_assert_eq!(ki, p1[self.at1].0.0, "advance must report common keys");
+            let mut e0 = self.at0;
             while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
-            let mut e1 = j;
+            let mut e1 = self.at1;
             while e1 < p1.len() && p1[e1].0.0 == ki { e1 += 1; }
-            join_key(ki, p0, i..e0, p1, j..e1, h0, h1, &mut self.matches);
-            i = e0;
-            j = e1;
+            let retired = join_key(ki, p0, self.at0..e0, p1, self.at1..e1, h0, h1, &mut self.matches, &mut self.resume);
+            // A key that suspended stays open, so its runs must still be the ones `at0`/`at1` name.
+            if !retired { return; }
+            self.at0 = e0;
+            self.at1 = e1;
+            if self.matches.ids.len() >= KEY_WORK_LIMIT { return; }
         }
-        debug_assert!(i == p0.len() && j == p1.len(), "both bridges must drain together");
+        debug_assert!(self.at0 == p0.len() && self.at1 == p1.len(), "both bridges must drain together");
     }
 
     /// Turn the block's matches into containers, ready to be served one at a time.
@@ -232,11 +261,35 @@ where
     }
 }
 
-/// Match one key's records across the two presented runs.
+/// Progress through one key, for the matching that suspends within a key.
+///
+/// The variants mirror the two strategies [`join_key`] chooses between: the direct cross product it
+/// uses when a run is short, and the time-ordered wave it uses when both are long. A wave step
+/// advances one history's buffer before emitting against it, so the step it is part-way through is
+/// part of the state: resuming must not advance that buffer a second time, which would consolidate
+/// it and invalidate the offset resumed from.
+#[derive(Clone, Copy)]
+enum Resume {
+    /// The direct cross product, at offsets `(offset0, offset1)` into the key's two runs.
+    Nested(usize, usize),
+    /// The wave, between steps; the next step follows from the histories' least times.
+    WaveChoose,
+    /// The wave, emitting `h1`'s buffer against `h0`'s next edit, from this offset.
+    WaveStep0(usize),
+    /// The wave, emitting `h0`'s buffer against `h1`'s next edit, from this offset.
+    WaveStep1(usize),
+}
+
+/// Match one key's records across the two presented runs, reporting whether the key is retired.
 ///
 /// If either history is small, this performs a direct cross product.
 /// If both histories are large, this replays the histories compacting as it goes in
 /// order to (potentially) avoid quadratic blow-up.
+///
+/// A key whose cross product fits [`KEY_WORK_LIMIT`] is matched in one uninterrupted pass. A larger
+/// one stops once `matches` reaches the limit, records where it stopped in `resume`, and returns
+/// `false`; a later call with the same runs and the same `resume` picks up from there. Both
+/// strategies address their inputs by offset, so nothing has to be reloaded across a suspension.
 fn join_key<T, R0, R1, RO>(
     kh: u64,
     p0: &ProxyBridge<T, R0>,
@@ -246,29 +299,124 @@ fn join_key<T, R0, R1, RO>(
     h0: &mut IdHistory<T, R0>,
     h1: &mut IdHistory<T, R1>,
     matches: &mut JoinMatches<T, RO>,
-) where
+    resume: &mut Option<Resume>,
+) -> bool
+where
     T: Lattice + Timestamp,
     R0: Semigroup + Multiply<R1, Output = RO> + Clone,
     R1: Semigroup + Clone,
 {
-    if r0.len() < 16 || r1.len() < 16 {
-        for a in r0 {
-            for b in r1.clone() {
-                matches.ids.push((kh, (p0[a].0.1, p1[b].0.1)));
-                matches.times.push(p0[a].1.join(&p1[b].1));
-                matches.diffs.push(p0[a].2.clone().multiply(&p1[b].2));
+    let nested = r0.len() < 16 || r1.len() < 16;
+
+    if resume.is_none() {
+        if r0.len().saturating_mul(r1.len()) <= KEY_WORK_LIMIT {
+            if nested {
+                for a in r0 {
+                    for b in r1.clone() {
+                        matches.ids.push((kh, (p0[a].0.1, p1[b].0.1)));
+                        matches.times.push(p0[a].1.join(&p1[b].1));
+                        matches.diffs.push(p0[a].2.clone().multiply(&p1[b].2));
+                    }
+                }
+            }
+            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);
+                bilinear_wave(h0, h1, |v0, v1, t, d| {
+                    matches.ids.push((kh, (v0, v1)));
+                    matches.times.push(t);
+                    matches.diffs.push(d);
+                });
             }
+            return true;
         }
-    }
-    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);
-        bilinear_wave(h0, h1, |v0, v1, t, d| {
-            matches.ids.push((kh, (v0, v1)));
-            matches.times.push(t);
-            matches.diffs.push(d);
+
+        // The key can outrun the budget, so open it for the matching that suspends within it. The
+        // strategy is the one the uninterrupted pass would have taken, so both visit a key's
+        // matches in the same order.
+        *resume = Some(if nested {
+            Resume::Nested(r0.start, r1.start)
+        }
+        else {
+            h0.load_iter(r0.clone().map(|i| (p0[i].0.1, p0[i].1.clone(), p0[i].2.clone())), None);
+            h1.load_iter(r1.clone().map(|i| (p1[i].0.1, p1[i].1.clone(), p1[i].2.clone())), None);
+            Resume::WaveChoose
         });
     }
+
+    let mut state = resume.expect("key is open");
+    loop {
+        match state {
+            Resume::Nested(mut offset0, mut offset1) => {
+                while offset0 < r0.end {
+                    while offset1 < r1.end {
+                        matches.ids.push((kh, (p0[offset0].0.1, p1[offset1].0.1)));
+                        matches.times.push(p0[offset0].1.join(&p1[offset1].1));
+                        matches.diffs.push(p0[offset0].2.clone().multiply(&p1[offset1].2));
+                        offset1 += 1;
+                        if matches.ids.len() >= KEY_WORK_LIMIT {
+                            *resume = Some(Resume::Nested(offset0, offset1));
+                            return false;
+                        }
+                    }
+                    offset1 = r1.start;
+                    offset0 += 1;
+                }
+                *resume = None;
+                return true;
+            }
+            Resume::WaveChoose => {
+                // Step whichever history holds the earlier un-replayed time, and once one is spent,
+                // whichever remains. A tie steps the second, as `bilinear_wave`'s comparison does.
+                let step0 = match (h0.time(), h1.time()) {
+                    (Some(time0), Some(time1)) => time0 < time1,
+                    (Some(_), None) => true,
+                    (None, Some(_)) => false,
+                    (None, None) => { *resume = None; return true; }
+                };
+                if step0 {
+                    h1.advance_buffer_by(h0.meet().unwrap());
+                    state = Resume::WaveStep0(0);
+                }
+                else {
+                    h0.advance_buffer_by(h1.meet().unwrap());
+                    state = Resume::WaveStep1(0);
+                }
+            }
+            Resume::WaveStep0(mut offset) => {
+                while offset < h1.buffer().len() {
+                    let (val0, time0, diff0) = h0.edit().unwrap();
+                    let ((val1, time1), diff1) = &h1.buffer()[offset];
+                    matches.ids.push((kh, (val0, *val1)));
+                    matches.times.push(time0.join(time1));
+                    matches.diffs.push(diff0.clone().multiply(diff1));
+                    offset += 1;
+                    if matches.ids.len() >= KEY_WORK_LIMIT {
+                        *resume = Some(Resume::WaveStep0(offset));
+                        return false;
+                    }
+                }
+                h0.step();
+                state = Resume::WaveChoose;
+            }
+            Resume::WaveStep1(mut offset) => {
+                while offset < h0.buffer().len() {
+                    let ((val0, time0), diff0) = &h0.buffer()[offset];
+                    let (val1, time1, diff1) = h1.edit().unwrap();
+                    matches.ids.push((kh, (*val0, val1)));
+                    matches.times.push(time0.join(time1));
+                    matches.diffs.push(diff0.clone().multiply(diff1));
+                    offset += 1;
+                    if matches.ids.len() >= KEY_WORK_LIMIT {
+                        *resume = Some(Resume::WaveStep1(offset));
+                        return false;
+                    }
+                }
+                h1.step();
+                state = Resume::WaveChoose;
+            }
+        }
+    }
 }
 
 /// Produces the join of two histories: every pair of edits, diffs multiplied and times
@@ -324,3 +472,151 @@ fn bilinear_wave<V, T, R0, R1, RO>(
         h1.step();
     }
 }
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+
+    /// One `(key hash, id0, id1)` match, at a time and with a diff.
+    type Match = ((u64, u64, u64), u64, isize);
+
+    /// A backend that presents its two sides as a single block, and hands matches back verbatim.
+    ///
+    /// The single block is the shape the suspension has to cope with: `advance` must report a key
+    /// entirely within one block, so a hot key cannot be split across blocks however a backend
+    /// draws them.
+    struct OneBlock {
+        side0: ProxyBridge<u64, isize>,
+        side1: ProxyBridge<u64, isize>,
+    }
+
+    impl ProxyJoinBackend<u64, (), ()> for OneBlock {
+        type R0 = isize;
+        type R1 = isize;
+        type ROut = isize;
+        type Output = Vec<Match>;
+
+        fn advance(
+            &mut self,
+            _instance: &JoinInstance<u64, (), ()>,
+            from: &mut Option<u64>,
+            bridge0: &mut ProxyBridge<u64, Self::R0>,
+            bridge1: &mut ProxyBridge<u64, Self::R1>,
+        ) {
+            if from.is_some() {
+                bridge0.clone_from(&self.side0);
+                bridge1.clone_from(&self.side1);
+                *from = None;
+            }
+        }
+
+        fn cross(
+            &mut self,
+            _instance: &JoinInstance<u64, (), ()>,
+            matches: &mut JoinMatches<u64, Self::ROut>,
+            output: &mut Vec<Self::Output>,
+        ) {
+            let crossed = (0..matches.ids.len())
+                .map(|at| {
+                    let (key, (id0, id1)) = matches.ids[at];
+                    ((key, id0, id1), matches.times[at], matches.diffs[at])
+                })
+                .collect();
+            output.push(crossed);
+        }
+    }
+
+    /// One side of a join: `vals` values under the single key hash `0`, each at `times` distinct
+    /// times, with diffs that differ per value and time so a dropped or duplicated match shows up.
+    fn side(vals: u64, times: u64) -> ProxyBridge<u64, isize> {
+        let mut bridge = ProxyBridge::new();
+        for val in 0..vals {
+            for time in 0..times {
+                bridge.push(((0, val), time, 1 + isize::try_from(val + time).unwrap() % 3));
+            }
+        }
+        bridge
+    }
+
+    /// Joins two single-key sides, returning the containers the tactic yields.
+    fn join(side0: ProxyBridge<u64, isize>, side1: ProxyBridge<u64, isize>) -> Vec<Vec<Match>> {
+        let mut tactic = ProxyJoinTactic::new(OneBlock { side0, side1 });
+        tactic.prep(vec![], vec![], Fresh::Input1, 0).collect()
+    }
+
+    /// Accumulates matches into a sorted, consolidated list, dropping those that cancel.
+    fn consolidate(mut matches: Vec<Match>) -> Vec<Match> {
+        matches.sort();
+        let mut out: Vec<Match> = Vec::new();
+        for (data, time, diff) in matches {
+            match out.last_mut() {
+                Some(last) if (last.0, last.1) == (data, time) => last.2 += diff,
+                _ => out.push((data, time, diff)),
+            }
+        }
+        out.retain(|update| update.2 != 0);
+        out
+    }
+
+    /// The cross product of two sides, as the join should produce it.
+    fn cross(side0: &ProxyBridge<u64, isize>, side1: &ProxyBridge<u64, isize>) -> Vec<Match> {
+        let mut out = Vec::new();
+        for ((_, id0), time0, diff0) in side0 {
+            for ((_, id1), time1, diff1) in side1 {
+                out.push(((0, *id0, *id1), *time0.max(time1), diff0 * diff1));
+            }
+        }
+        out
+    }
+
+    /// A key large enough to outrun the driver's budget suspends within itself, rather than
+    /// matching its whole cross product before the backend is handed any of it. Both sides carry
+    /// enough records for the time-ordered wave.
+    #[test]
+    fn wave_suspends_within_a_key() {
+        let vals = 1_100;
+        assert!((vals * vals) as usize > KEY_WORK_LIMIT);
+
+        let containers = join(side(vals, 1), side(vals, 1));
+        assert!(containers.len() > 1, "the key was matched in one go");
+        assert!(containers[0].len() <= KEY_WORK_LIMIT);
+        assert_eq!(containers.iter().map(Vec::len).sum::<usize>(), (vals * vals) as usize);
+    }
+
+    /// The same, for the direct cross product used when one side's run is short: few records on a
+    /// side does not bound the work, only the other side's run length does.
+    #[test]
+    fn nested_suspends_within_a_key() {
+        let (vals0, vals1) = (5, 250_000);
+        assert!((vals0 * vals1) as usize > KEY_WORK_LIMIT);
+
+        let containers = join(side(vals0, 1), side(vals1, 1));
+        assert!(containers.len() > 1, "the key was matched in one go");
+        assert!(containers[0].len() <= KEY_WORK_LIMIT);
+        assert_eq!(containers.iter().map(Vec::len).sum::<usize>(), (vals0 * vals1) as usize);
+    }
+
+    /// A key with many distinct times exercises the wave's buffer advance and consolidation, which
+    /// a suspension must resume without repeating. The matches must consolidate to the same
+    /// collection as the cross product of the two sides, with times joined and diffs multiplied.
+    #[test]
+    fn suspended_wave_matches_the_cross_product() {
+        let (vals, times) = (200, 8);
+        assert!(((vals * times) * (vals * times)) as usize > KEY_WORK_LIMIT);
+
+        let (side0, side1) = (side(vals, times), side(vals, times));
+        let expected = consolidate(cross(&side0, &side1));
+        let produced = consolidate(join(side0, side1).concat());
+
+        // The collections run to millions of matches, so report the first difference rather than
+        // both of them in full.
+        let differs = produced.iter().zip(expected.iter()).position(|(x, y)| x != y);
+        assert_eq!(
+            differs.map(|at| (at, produced[at], expected[at])),
+            None,
+            "consolidated matches differ from the cross product",
+        );
+        assert_eq!(produced.len(), expected.len());
+    }
+}
diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs
index 761a500be..82bd25bb7 100644
--- a/differential-dataflow/src/operators/join.rs
+++ b/differential-dataflow/src/operators/join.rs
@@ -32,6 +32,23 @@ pub trait JoinTactic<T, B0, B1, C> {
     fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: T) -> Box<dyn Iterator<Item = C>>;
 }
 
+/// Records a join activation may ship before it yields to the timely runtime.
+///
+/// The budget is split evenly between the two input queues, preserving the historical
+/// `1_000_000` of progress per input each activation. Tactics consult it to decide how far
+/// they may run before suspending.
+pub(crate) const FUEL: usize = 2_000_000;
+
+/// Matches a tactic may produce for one key before it must be able to suspend within that key.
+///
+/// A key replayed in one uninterrupted pass buffers its whole cross product, so the bound is what a
+/// join is willing to buffer: the driver's per-queue budget, which is the most one activation ships
+/// from a single iterator anyway. Tactics measure a key by the product of its two sides' edit
+/// counts, an upper bound on its matches. The bound is exact for a nested cross product and
+/// conservative for a replay that nets its inputs down as it goes, so a key a replay would have
+/// kept under the budget is only handled more carefully than it needed to be, never wrongly.
+pub(crate) const KEY_WORK_LIMIT: usize = FUEL / 2;
+
 /// Which input contributed the freshly-arrived batch of a deferred join unit.
 ///
 /// The fresh batch's times all lie at or beyond the capability, so its side is not advanced by the
@@ -293,8 +310,7 @@ where
             // Perform some amount of outstanding work by pulling the deferred iterators and shipping the
             // containers they yield. Each direction drains against its own half of the budget, so a burst
             // on one input cannot starve the other. We reschedule the operator whenever any work remains,
-            // which is observable directly: an iterator has yet to yield `None`. The budget is split from
-            // `2_000_000` to preserve the historical `1_000_000` of progress per input each activation.
+            // which is observable directly: an iterator has yet to yield `None`.
             // The driver only ships finished containers (`give_container`), never pushing records, so it
             // pins the operator output to `NoopBuilder<C>` — the builder for exactly this "containers ready
             // to go" case, which is a `ContainerBuilder` for any `C` without further bounds.
@@ -311,9 +327,8 @@ where
                     }
                 }
             };
-            let fuel = 2_000_000;
-            drain(&mut todo0, fuel / 2);
-            drain(&mut todo1, fuel / 2);
+            drain(&mut todo0, (FUEL / 2) as isize);
+            drain(&mut todo1, (FUEL / 2) as isize);
             if !todo0.is_empty() || !todo1.is_empty() {
                 activator.activate();
             }
diff --git a/differential-dataflow/tests/join_bench.rs b/differential-dataflow/tests/join_bench.rs
new file mode 100644
index 000000000..3f4ba7bd3
--- /dev/null
+++ b/differential-dataflow/tests/join_bench.rs
@@ -0,0 +1,84 @@
+//! Benchmarks the cursor join tactic across the key shapes its two strategies cover.
+//!
+//! The tactic runs a key to completion in one call when the key's cross product is small, and
+//! otherwise reloads the key and replays it a container at a time so that it can suspend within the
+//! key. The workloads here bracket that choice:
+//! * `unique` — one value per key on each side: the per-key overhead, where a join spends its time
+//!   opening and retiring keys rather than crossing them.
+//! * `fanout` — a handful of values per key on each side: still a key per call, with real work in it.
+//! * `warm-key` — a single key just under the threshold: the largest key run in one call.
+//! * `hot-key` — a single key well over the threshold: the suspending path, and the shape that
+//!   buffered its whole cross product before the tactic could suspend within a key.
+//!
+//! Each workload loads its inputs, then measures the round that joins them. Run with:
+//! ```text
+//! cargo test --release --test join_bench -- --ignored --nocapture
+//! ```
+
+use std::time::{Duration, Instant};
+
+use timely::dataflow::operators::probe::Handle as ProbeHandle;
+use timely::dataflow::operators::Probe;
+
+use differential_dataflow::input::InputSession;
+
+/// The `(key, value)` pairs of one side of a join: `keys` keys, each carrying `vals` values.
+fn side(keys: u64, vals: u64) -> impl Iterator<Item = (u64, u64)> {
+    (0..keys).flat_map(move |key| (0..vals).map(move |val| (key, val)))
+}
+
+/// Joins `keys` keys of `vals1` by `vals2` values, returning how long the joining round took.
+///
+/// The inputs are loaded and flushed at time `0`, and the round timed is the one that advances to
+/// time `1`, so the measurement covers the join and not the arrangement of its inputs.
+fn join_round(keys: u64, vals1: u64, vals2: u64) -> Duration {
+    let elapsed = timely::execute_directly(move |worker| {
+        let probe = ProbeHandle::new();
+        let mut input1 = <InputSession<u64, (u64, u64), isize>>::new();
+        let mut input2 = <InputSession<u64, (u64, u64), isize>>::new();
+        worker.dataflow(|scope| {
+            let collection1 = input1.to_collection(scope);
+            let collection2 = input2.to_collection(scope);
+            collection1
+                .arrange_by_key()
+                .join_core(collection2.arrange_by_key(), |key, val1: &u64, val2: &u64| {
+                    Some((*key, val1 ^ val2))
+                })
+                .inner
+                .probe_with(&probe);
+        });
+
+        // Load both sides at time `0`, and let the join of an empty side against them settle.
+        for (key, val) in side(keys, vals1) { input1.insert((key, val)); }
+        for (key, val) in side(keys, vals2) { input2.insert((key, val)); }
+        input1.advance_to(1);
+        input2.advance_to(1);
+        input1.flush();
+        input2.flush();
+
+        let start = Instant::now();
+        worker.step_while(|| probe.less_than(input1.time()));
+        start.elapsed()
+    });
+    elapsed
+}
+
+/// Runs one workload and reports its shape, its match count, and the time the joining round took.
+fn report(name: &str, keys: u64, vals1: u64, vals2: u64) {
+    let elapsed = join_round(keys, vals1, vals2);
+    let matches = keys * vals1 * vals2;
+    println!(
+        "{name:<10} keys {keys:>8}  vals {vals1:>7} x {vals2:<7}  matches {matches:>10}  {elapsed:>12.3?}",
+    );
+}
+
+#[test]
+#[ignore]
+fn join_bench() {
+    // Sized against the tactic's threshold, which is half the join driver's fuel: 1,000,000 matches.
+    report("unique", 1_000_000, 1, 1);
+    report("fanout", 100_000, 8, 8);
+    report("warm-key", 1, 900, 900);
+    report("hot-key", 1, 1_500, 1_500);
+    report("hot-key", 1, 3_000, 3_000);
+}