propagate_at / strongly_connected_at: nondeterministic (wrong output + hangs) under multi-worker contention

[frankmcsherry]

Summary

propagate_at (and strongly_connected_at, which wires it into SCC) is nondeterministic under multi-worker execution with CPU contention — it violates differential dataflow's determinism guarantee. It fails two ways:

• wrong-complete (~10%): the computation terminates normally but produces output that disagrees with a sequential SCC oracle.
• hang (~30%): the computation never makes progress (frontier stalls), so the run deadlocks.

The plain propagation path (.iterate() + reduce) is deterministic under the same conditions; only the propagate_at -> propagate_core path (the rotated reduce_abelian loop) is affected. The bug requires the enter_at-delayed label introduction — with logic = |_| 0 it does not reproduce.

Reproduction

The bug surfaces through strongly_connected_at, a thin wrapper that routes strongly_connected through propagate_at. Apply this to differential-dataflow/src/algorithms/graphs/scc.rs:

// change the import
-use super::propagate::propagate;
+use super::propagate::propagate_at;

// strongly_connected delegates to a new prioritized variant
pub fn strongly_connected<...>(graph: ...) -> ... { strongly_connected_at(graph, |_| 0) }

pub fn strongly_connected_at<'scope, T, N, R, F>(graph: VecCollection<'scope, T, (N,N), R>, logic: F) -> VecCollection<'scope, T, (N,N), R>
where T: Timestamp + Lattice + Hash, N: ExchangeData + Hash, R: ExchangeData + Abelian + Multiply<R, Output=R> + From<i8>, F: Fn(&N)->u64+Clone+'static,
{ /* body identical to strongly_connected, threading `logic` into both trim_edges calls */ }

// trim_edges takes `logic: F` and uses:
-    let labels = propagate(cycle, nodes).arrange_by_key();
+    let labels = propagate_at(cycle, nodes, logic).arrange_by_key();

(The full diff is on frankmcsherry/differential-dataflow at commit cd67ea48^.)

Add to differential-dataflow/tests/scc.rs a prioritized variant that calls the library SCC (not the test's local reachability):

#[test] fn scc_at_10_20_1000()  { test_sizes(10, 20, 1000, Config::process(3), true); }
#[test] fn scc_at_100_200_10()  { test_sizes(100, 200, 10, Config::process(3), true); }
#[test] fn scc_at_100_2000_1()  { test_sizes(100, 2000, 1, Config::process(3), true); }

where test_sizes(..., prioritized)'s dataflow uses:

let result = if prioritized {
    differential_dataflow::algorithms::graphs::scc::strongly_connected_at(edges, |x| *x as u64)
} else {
    _strongly_connected(edges)   // the test's existing local reachability
};

Run it — the trigger is concurrent contention, so run the whole binary (all tests in parallel), not an isolated single test:

cargo build --tests -p differential-dataflow
BIN=$(ls -t target/debug/deps/scc-* | grep -v '\.d$' | head -1)
for i in $(seq 1 20); do
  timeout 60 "$BIN"; rc=$?
  case $rc in 0) echo ok;; 124) echo HANG;; *) echo "WRONG-COMPLETE (rc=$rc)";; esac
  pkill -9 -f "deps/scc-"; sleep 0.3
done

Typical result over 20 runs: ~12 ok / ~2 wrong-complete (rc=101) / ~6 hang (rc=124).

What isolates it (bisection)

condition	result
plain SCC (local .iterate()+reduce), 3 workers, under contention	deterministic
_at (propagate_at), Config::thread / process(1) / process(2), isolated	deterministic
_at, process(3)/(4), isolated single run	deterministic
_at, process(3), under contention (full binary, all tests in parallel)	~40% bad (wrong-complete + hang)
_at with logic = |_| 0 (no enter_at delay), under contention	deterministic

So all three are necessary: propagate_at/propagate_core, multiple workers plus CPU contention, and a non-trivial enter_at delay.

A confirmed wrong-complete instance

test result: FAILED. 5 passed; 1 failed; finished in 3.60s — exit code 101 (normal test-failure termination), not the 60s timeout (which would be 124) or SIGKILL (137). So it terminated on its own with wrong output. results2 (the differential output) is non-empty and genuinely differs from the sequential oracle: 26 tuples in diff-not-oracle, 13 in oracle-not-diff, clustered at late rounds (~820–957 of 1000). The wrong output contains transient spurious edges that appear and immediately retract (e.g. (4,1), (6,1), (9,5) asserted at t=911, retracted at t=912) — suggesting intermediate iteration state escaping consolidation near convergence, rather than a stable wrong fixpoint.

Suspected location

propagate_core (in differential-dataflow/src/algorithms/graphs/propagate.rs) uses a rotated iteration: the Variable feedback is the join output, labels = reduce_abelian(proposals ++ enter_at'd nodes), and that reduce arrangement is reused for both join_core and as_collection. The commented-out "morally equivalent" .iterate() + .reduce() form directly above it is the deterministic reference. The interaction of enter_at future-timestamped nodes with the Variable summary (Product::new(Default, 1)) and the rotation is the prime suspect — the hang mode in particular points at iteration progress tracking.

[frankmcsherry]

Closing summary, for future readers.

Root cause (fixed in #802): join_with_tactic dropped in-flight non-empty batches. Its acknowledged frontiers — used to skip "pre-loaded data" — are also advanced by advance_upper, which consults the shared trace's merged, logically-compacted batches. A region whose updates consolidate to zero under compaction (an add/remove pair whose times the compaction frontier has equated) reads as an empty batch there, so acknowledged could leap over batches still in flight on the input stream; on arrival they failed the acknowledged ≤ batch.lower() guard and were silently discarded, their cross-products never computed. Dropping is sound only for readers at-or-beyond the compaction frontier; propagate_core's rotated loop reads finer times (the enter_at seeds and pending work at old outer rounds), so the loop's state went inconsistent with its own history. The self-correction then either damped — terminating with output that disagrees with the oracle at a few late times (the wrong-complete mode) — or entered an undamped period-2 limit cycle whose interesting-times work grew without bound (the hang mode, a livelock rather than a deadlock). One bug, both modes.

Why it looked like multi-worker nondeterminism: it isn't. There is no message-transport delay in the mechanism — only the one-scheduling-quantum lag between the trace view (shared state, updated synchronously by the arrange, merges included) and the stream view of an entered arrangement (whose batches pass through the BatchEnter map operator, forwarded only when scheduled; mid-step activations take effect at the next Activations::advance). Under a deterministic simulation of message delivery (timely#810), the bug reproduced on a single worker with no schedule chaos at all, 100% of seeds. Contention's only role is to create the backlog regime (wide batch seals containing add/remove pairs, compaction ahead of unprocessed regions) that makes the coincidence likely — hence the ~10%/~30% native flake rates in the original report. Native confirmation: with the drop sites instrumented, every hanging contended run logged dropped non-empty batches (up to 46 updates in one batch), and the fixed build was clean in 100+ contended runs.

The suspicion in the original report — the rotated reduce_abelian loop — was the right neighborhood but the wrong operator: the reduce's driver joins the input frontier into its retirement upper, which in-flight messages hold back, so it cannot skip content this way; the reference tactic swap and a pre-rewrite bisect both reproduced the bug, exonerating the reduce internals.

Regression tests: the scc_at_* tests landed with #802; deterministic simulation-based tests (which reproduce this bug on every seed pre-fix) are staged in #805, blocked on timely#810 shipping.