fix(amber, v1.2): advance region executions in a later coordinator round, not inside portCompleted - #7096
Merged
xuang7 merged 1 commit intoJul 30, 2026
Conversation
Contributor
Author
|
The cherry-pick conflicted and was committed with conflict markers. Resolve the conflicts on this branch, then mark this PR ready for review. Conflicting files:
|
…ot inside portCompleted (#6960) ### 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? (backported from commit 226614b) Generated-by: Claude Code (Fable 5)
aglinxinyuan
force-pushed
the
backport/6960-advance-region-executions-in-a-later-coo-v1.2
branch
from
July 30, 2026 02:04
2c5abe2 to
a2e5fcf
Compare
Contributor
Author
Automated Reviewer SuggestionsBased on the
|
aglinxinyuan
marked this pull request as ready for review
July 30, 2026 02:05
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## release/v1.2 #7096 +/- ##
===============================================
Coverage 52.49% 52.50%
- Complexity 2498 2505 +7
===============================================
Files 1077 1078 +1
Lines 42296 42315 +19
Branches 4551 4556 +5
===============================================
+ Hits 22203 22217 +14
- Misses 18782 18787 +5
Partials 1311 1311
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
xuang7
approved these changes
Jul 30, 2026
xuang7
deleted the
backport/6960-advance-region-executions-in-a-later-coo-v1.2
branch
July 30, 2026 06:28
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this PR?
Automated backport of #6960 to
release/v1.2.Source: 226614b · automation run
Any related issues, documentation, discussions?
Backport of #6960.
How was this PR tested?
Release-branch CI runs on this branch once the conflicts are resolved and this PR is marked ready for review.
Was this PR authored or co-authored using generative AI tooling?
No.