fix(amber): advance region executions in a later coordinator round, not inside portCompleted

[aglinxinyuan]

What changes were proposed in this PR?

Root cause of the LoopIntegrationSpec retry storm. portCompleted's handler advanced region executions inline in its own continuation. That advance terminates a completed region, which sends EndWorker to each of its workers — so EndWorker went out before the reply to the very portCompleted being handled, and both travel the same FIFO control channel to that worker:

Before — all inside one handler round:
  advanceRegionExecutions()  ──> EndWorker to W          (seq N   on W's control channel)
  return EmptyReturn()       ──> ReturnInvocation to W   (seq N+1, same channel)

W processes EndWorker while the reply is still queued behind it:
  EndHandler: queue not empty => IllegalStateException("worker still has unprocessed messages")
  => attempt 1 of 150 fails => 2 ERRORs + 2 WARNs + 2 stack traces => 200 ms retry succeeds

The check's premise — that EndWorker is the last message a worker receives — was being violated by the coordinator itself on every teardown. Whether the worker actually observed the trailing reply was a microsecond race, hence the ~1-in-10 flake.

Fix: advance in a later coordinator round instead of inline. A new coordinator-to-coordinator RPC carries the advance:

After:
  portCompleted round: bookkeeping; send CoordinatorInitiateAdvanceRegionExecutions to self
                       return EmptyReturn()                          ──> ReturnInvocation to W
  later round:         advance runs ──> EndWorker to W

A message the coordinator addresses to itself is transmitted and then received before it is handled, so it lands a whole message behind every reply the requesting round still owed — EndWorker can no longer overtake a reply, and the first termination attempt succeeds.

Second half: the worker tolerates a reply it cannot be ordered against. Deferring orders the replies the coordinator has already produced, but region completion keys only on port completion, so the advance fires as soon as the last portCompleted is booked — while the same worker's workerExecutionCompleted (emitted right after it) may still be queued at the coordinator. Since NetworkInputGateway.tryPickControlChannel selects channels out of a HashMap, there is no cross-channel order: the advance can run first and that reply be emitted afterwards, landing behind EndWorker. Ordering cannot close this, so EndWorker no longer counts a queued ReturnInvocation as unprocessed work — processing one only fulfills a promise for a request the worker already issued, and all four worker→coordinator calls discard their futures, so nothing is pending on it. Every other queued element still fails the request, so the coordinator retries and the worker drains it first. This PR is JVM-only: the Python worker keeps its current check, which fails EndWorker whenever anything is queued, and the coordinator's retry covers it exactly as it does today. Bringing it to parity needs a way to classify what is queued without inspecting the next message, which means changing InternalQueue — left for a follow-up rather than bundled here.

	Before	After
Coordinator send order	EndWorker, then the portCompleted reply	reply, then EndWorker
First EndWorker attempt	fails on a benign race (storm ~1 run in 10)	succeeds
Normal teardown logs	2 ERROR + 2 WARN + 2 stack traces	none
Worker-side check	any queued message fails	a queued ReturnInvocation no longer fails; all real work still does
Bookkeeping failure	ControlError to the reporting worker	unchanged
Advance failure	FatalError to the client	unchanged

Changes:

• coordinatorservice.proto: adds CoordinatorInitiateAdvanceRegionExecutions, alongside the existing self-directed CoordinatorInitiateQueryStatistics.
• AdvanceRegionExecutionsHandler (new): handles that self-addressed request by running the advance, and reports failures to the client exactly as the inline call did. advanceRegionExecutions's scaladoc now tells callers handling a worker-initiated request to go through it.
• PortCompletedHandler: keeps its bookkeeping and its reply position; only the advance moves.
• RegionExecutionManager: corrects the scaladoc that claimed EndWorker "will be the last message each worker receives".

StartWorkflowHandler still advances directly: its caller is the client, it awaits the future to answer with the workflow state, and no EndWorker is involved at startup.

Any related issues, documentation, discussions?

#6891.

How was this PR tested?

• PortCompletedHandlerSpec (new): drives a real CoordinatorProcessor through the real RPC server with a captured output gateway, over a single-region schedule already in flight. Pins that the advance leaves as a separate coordinator-addressed control message; that the only thing sent to the reporting worker while handling portCompleted is its reply (the direct regression guard — on main this list also contains EndWorker); that the port is recorded completed before the request goes out, since the advance now reads that state in a later round; that a port belonging to no executing region requests nothing; and that a failed continuation still returns a ControlError to the reporting worker.
• Baseline check: 3 of those tests fail against main's handler, while the ControlError test passes on both main and this branch — which is what verifies the worker-facing error contract is preserved rather than merely asserted.
• Untouched and green, as evidence termination semantics are unchanged: RegionExecutionManagerSpec, WorkflowExecutionManagerSpec, EndHandlerSpec, CoordinatorSpec, TrivialControlSpec, AsyncRPCClientSpec.
• Lint: scalafmtCheck, scalafixAll --check pass.
• Loop confirmation is CI-side and statistical: amber-integration teardown logs should contain zero Failed to terminate region ... on attempt 1 lines (runs without this fix show them repeatedly).

Was this PR authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Fable 5)

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• Contributors with relevant context: @Yicong-Huang
  You can notify them by mentioning @Yicong-Huang in a comment.

[github-actions]

⚠️ Benchmark changes need a look

🟢 0 better · 🔴 7 worse · ⚪ 8 noise (<±5%) · 0 without baseline

Compared against main ff82157 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🔴	bs=10 sw=10 sl=64	407	0.248	24,360/34,024/34,024 us	🔴 +9.3% / 🔴 +115.7%
🔴	bs=100 sw=10 sl=64	794	0.485	120,214/162,180/162,180 us	🔴 +28.0% / 🔴 +51.1%
⚪	bs=1000 sw=10 sl=64	916	0.559	1,088,995/1,122,727/1,122,727 us	⚪ within ±5% / 🔴 -11.2%

Baseline details

Latest main ff82157 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	407 tuples/sec	431 tuples/sec	786.12 tuples/sec	-5.6%	-48.2%
bs=10 sw=10 sl=64	MB/s	0.248 MB/s	0.263 MB/s	0.48 MB/s	-5.7%	-48.3%
bs=10 sw=10 sl=64	p50	24,360 us	22,297 us	12,305 us	+9.3%	+98.0%
bs=10 sw=10 sl=64	p95	34,024 us	33,075 us	15,774 us	+2.9%	+115.7%
bs=10 sw=10 sl=64	p99	34,024 us	33,075 us	18,978 us	+2.9%	+79.3%
bs=100 sw=10 sl=64	throughput	794 tuples/sec	851 tuples/sec	999.71 tuples/sec	-6.7%	-20.6%
bs=100 sw=10 sl=64	MB/s	0.485 MB/s	0.519 MB/s	0.61 MB/s	-6.6%	-20.5%
bs=100 sw=10 sl=64	p50	120,214 us	117,526 us	100,616 us	+2.3%	+19.5%
bs=100 sw=10 sl=64	p95	162,180 us	126,723 us	107,356 us	+28.0%	+51.1%
bs=100 sw=10 sl=64	p99	162,180 us	126,723 us	113,255 us	+28.0%	+43.2%
bs=1000 sw=10 sl=64	throughput	916 tuples/sec	934 tuples/sec	1,031 tuples/sec	-1.9%	-11.2%
bs=1000 sw=10 sl=64	MB/s	0.559 MB/s	0.57 MB/s	0.63 MB/s	-1.9%	-11.2%
bs=1000 sw=10 sl=64	p50	1,088,995 us	1,070,312 us	980,328 us	+1.7%	+11.1%
bs=1000 sw=10 sl=64	p95	1,122,727 us	1,118,659 us	1,027,528 us	+0.4%	+9.3%
bs=1000 sw=10 sl=64	p99	1,122,727 us	1,118,659 us	1,054,298 us	+0.4%	+6.5%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,491.37,200,128000,407,0.248,24360.21,34023.93,34023.93
1,100,10,64,20,2517.70,2000,1280000,794,0.485,120213.95,162179.54,162179.54
2,1000,10,64,20,21824.28,20000,12800000,916,0.559,1088995.08,1122727.08,1122727.08

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 79.27%. Comparing base (a61702f) to head (9055dfc).
⚠️ Report is 8 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6960      +/-   ##
============================================
+ Coverage     79.20%   79.27%   +0.06%     
- Complexity     3786     3796      +10     
============================================
  Files          1160     1161       +1     
  Lines         46118    46139      +21     
  Branches       5114     5119       +5     
============================================
+ Hits          36529    36575      +46     
+ Misses         7969     7944      -25     
  Partials       1620     1620              

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)		Carriedforward from 5092e3e
agent-service	76.76% <ø> (ø)		Carriedforward from 5092e3e
amber	72.62% <100.00%> (+0.18%)	⬆️
computing-unit-managing-service	20.49% <ø> (ø)		Carriedforward from 5092e3e
config-service	66.66% <ø> (ø)		Carriedforward from 5092e3e
file-service	67.21% <ø> (ø)		Carriedforward from 5092e3e
frontend	83.08% <ø> (ø)		Carriedforward from 5092e3e
notebook-migration-service	78.94% <ø> (ø)		Carriedforward from 5092e3e
pyamber	96.34% <ø> (-0.01%)	⬇️	Carriedforward from 5092e3e
workflow-compiling-service	26.31% <ø> (ø)		Carriedforward from 5092e3e

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Yicong-Huang]

🔴 1 must-fix · 1 advisory — clean, well-tested fix for the dominant portCompleted-reply instance of the #6891 race; one completeness question.

Correctness (2)

• PortCompletedHandler.scala:86 — deferral closes only the portCompleted-reply race; the same worker's workerExecutionCompleted ack can still race EndWorker, so #6891 may not be fully closed (must-fix, see inline)
• PortCompletedHandlerSpec.scala:200 — the spec never interleaves a workerExecutionCompleted, so it doesn't guard the ack race #6891 describes (advisory, see inline)

Verification trace

Traced the send ordering on the coordinator→worker FIFO channel. The deferral robustly orders portCompleted's own reply before EndWorker (both are decided within one round; the self-advance is only handled in a later round) — verified, and the new spec pins it. But region completion (RegionExecution.getState) keys only on port completion, and OutputManager.finalizeOutput emits FinalizePort (→portCompleted) before FinalizeExecutor (→workerExecutionCompleted), so the region-completing worker's workerExecutionCompleted is always still pending when its portCompleted triggers the advance. The self-advance (SELF channel) and that workerExecutionCompleted (worker→coordinator channel) have no cross-channel FIFO order — NetworkInputGateway.tryPickControlChannel scans a mutable.HashMap — so a round that picks the self-advance first emits EndWorker ahead of the workerExecutionCompleted ack, which then queues behind it at the worker and trips EndHandler. The stats-query round-trips only add nondeterministic delay, not an ordering guarantee.

[Yicong-Huang]

Conditionally approved.

After discussion, this can only partially fix the issue. There are more events could happen after EndWorker. If this can reduce the flakiness of the tests, I agree to merge it first and keep the issue open for further investigation as follow up. Please handle my inline comments as much as possible, though.

[Yicong-Huang]

I found unrelated changes? Please take a look

[github-actions]

Backport auto-label report

This fix: PR was checked against each actively-supported release branch. release/* labels drive the post-merge backport, so add or remove one to change where this fix lands.

Release branch	Analysis
✅ release/v1.2	Already labeled — this fix is queued to backport here.

Auto-label run.

[Yicong-Huang]

🔴 2 resolved · 0 open · 2 new (2 new = 2 newly introduced · 0 late catches)

The ack race I flagged last round is genuinely closed and well-tested — both new specs fail on main. Two things remain: a doc sweep, and reconsidering Closes.

Correctness (1)

• EndHandler.scala:49 (+ 3 more) — scaladoc references removed/renamed methods: hasUnprocessedWork → findUnprocessedWork, and WorkflowExecutionManager.requestAdvanceRegionExecutions() (removed) → the RPC coordinatorInitiateAdvanceRegionExecutions at WorkflowExecutionManager.scala:101, RegionExecutionManager.scala:144, AdvanceRegionExecutionsHandler.scala:36; the PR description repeats the dead name too (must-fix, see inline)

Conventions (1)

• Closes #6891 closes only the JVM instance of the race (the observed flake was JVM-side); the Python worker is deliberately left unfixed, so parity is a follow-up. Recommend Related to #6891 and keeping the issue open — matching the plan we agreed last round (advisory)

Verification trace

Traced the EndHandler ack-tolerance fix: the only worker-to-coordinator calls — workerExecutionCompleted, portCompleted, workerStateUpdated, consoleMessageTriggered, all in DataProcessor — are fire-and-forget (future discarded), and no worker-side code awaits any RPC reply, so terminating with a queued ReturnInvocation strands nothing. The deferral orders portCompleted's own reply before EndWorker deterministically (self-addressed advance is handled a round later); the residual workerExecutionCompleted ack cannot be ordered (self-advance vs worker-to-coordinator channel have no cross-channel FIFO — tryPickControlChannel scans a HashMap), which is exactly what the ack-tolerance covers. Both halves close #6891 for JVM workers.

[Yicong-Huang]

🟡 1 resolved · 1 open · 0 new (0 new = 0 newly introduced · 0 late catches)

The prior doc-sweep must-fix is fully resolved — all four references now name CoordinatorInitiateAdvanceRegionExecutions / findUnprocessedWork, and a repo-wide grep for the dead names is empty. One advisory remains.

Conventions (1)

• Closes #6891 closes the issue on merge, but the PR is JVM-only and #6891's expected behavior covers the Python side too. Recommend Related to #6891 and keeping the issue open for the Python-parity follow-up (advisory)

[github-actions]

Backport PR opened: draft #7096 (#7096) to release/v1.2, assigned to @aglinxinyuan — needs manual work because the cherry-pick conflicts.