join: do not drop in-flight batches overtaken by advance_upper (#801)

[frankmcsherry]

The bug

join_with_tactic maintains per-input acknowledged frontiers and skips arriving
stream batches unless acknowledged ≤ batch.lower(), on the theory that anything
below acknowledged is "pre-loaded data" already joined at operator start-up.

But acknowledged is also advanced by TraceReader::advance_upper, which walks the
shared trace's current batches — a merged, logically compacted view. A batch region
whose updates consolidate to zero under compaction (e.g. an add/remove pair whose
distinct times collapse to one once the compaction frontier passes both) reads as an
empty batch there, so advance_upper leaps it — while the raw, non-empty stream
batches for that region are still in flight. On arrival they fail the guard and were
silently dropped, and their cross-products against the other input were never computed.

Dropping them is sound only for consumers reading at or beyond the compaction
frontier. Iterative dataflows read finer times than that: in propagate_core's
rotated loop, the reduce accumulates proposals at (outer, round) coordinates that
remain open long after the join's inputs' frontiers (which drive compaction) have
moved past them. A dropped batch there leaves the loop's state inconsistent with its
own history; the self-correcting reduce then chases the discrepancy around the loop.
When the chase damps, the run terminates with output that disagrees with a sequential
oracle at a few (late) times; when it does not damp, it becomes a period-2 limit cycle
that never converges — pending "interesting times" pile up across every live outer
round, each retire gets more expensive, and the computation livelocks. These are the
"wrong-complete" and "hang" modes of #801, one bug.

The trigger requires only batch-boundary placement, not multi-worker races: a
single-worker deterministic simulation reproduces both modes. Under real execution it
surfaces as schedule-dependence (CPU contention perturbing delivery), hence the
apparent nondeterminism.

The fix

• Snapshot preload_upper1/2 after the start-up loading loops; arriving batches
  wholly at-or-before them are the replayed pre-loaded data, and are still skipped.
• Every other non-empty arriving batch is joined against the opposing trace through
  its current acknowledged, even when this input's acknowledged has overtaken it.
• acknowledged is never regressed: it advances via received uppers when they are
  beyond it, and via advance_upper as before.

Each batch is still joined exactly once: acknowledged can only overtake an
in-flight batch via advance_upper, which requires the region to have consolidated
to empty in the trace — so every opposing work item that consulted the trace saw zero
contribution from that region, regardless of interleaving, and processing the raw
batch on arrival contributes each cross-product exactly once.

Validation

• strongly_connected_at (rotated propagate_at) SCC tests, 3 workers, under heavy
  CPU contention: previously ~15-40% wrong-output/hang per run; now clean across
  40+ contended executions of the full test binary.
• Deterministic-simulation harness (single-thread multi-worker with schedule-controlled
  message delivery, atop the timely Simulation work): previously 100% of seeds
  livelocked at 1000 input rounds and diverged (updates at SCC iteration 100+,
  propagate round 1400+, on a 10-node graph); now all seeds complete and match the
  sequential oracle, including the chaos-free schedules that previously diverged.
• Full differential-dataflow and dogsdogsdogs test suites pass.

Notes

• The reduce driver (reduce_with_tactic) tolerates late batches by construction —
  its retirement upper joins in the input frontier, which in-flight messages hold
  back, so it cannot skip content the same way. There is a theoretical path to a
  transiently malformed retirement interval via late empty batches regressing
  upper_limit when merges misalign with advance_upper's exact-boundary check;
  not observed, flagged for a follow-up audit.
• count/threshold use advance_upper only to advance compaction, and process
  every received batch; they do not have the skip guard.

Fixes #801.

🤖 Generated with Claude Code