fix(amber): treat a queued coordinator reply as processed work in EndWorker

[aglinxinyuan]

What changes were proposed in this PR?

Root cause. On every normal region termination, the coordinator sends EndWorker from inside the portCompleted handler, so the ReturnInvocation replying to that same portCompleted is always emitted after EndWorker on the same FIFO control channel:

Coordinator — portCompleted(W) handler, one synchronous continuation:
  1. advanceRegionExecutions() ──> sends EndWorker to every worker   (seq N   on W's control channel)
  2. returns EmptyReturn()     ──> sends ReturnInvocation to W       (seq N+1, same channel)

Worker W — processes EndWorker while ReturnInvocation(N+1) may already sit in its arrival queue:
  old check: ANY queued element => IllegalStateException("worker still has unprocessed messages")
  => attempt 1 of 150 fails => 200 ms retry succeeds

Whether the trailing reply has landed when EndHandler runs is a thread-scheduling race, so healthy teardowns routinely fail the first termination attempt and log an alarming storm (2 ERRORs + 2 WARNs + 2 stack traces, then a wasted 200 ms retry). A queued ReturnInvocation is not work: processing it only fulfills a promise for a request the worker already issued, and every worker-to-coordinator call discards its future (DataProcessor.scala:179,187,224,312), so no continuation can be lost.

Fix. Both workers now exclude queued ReturnInvocations from the unprocessed-messages check. Everything that is work — ControlInvocations, data, ECMs, timer-based controls, actor commands — still blocks termination, and the 150 × 200 ms retry/give-up contract is untouched.

	Before	After
Normal teardown logs	2 ERROR + 2 WARN + 2 stack traces	none
First EndWorker attempt	fails on a benign race	succeeds
Teardown latency	+200 ms retry	no retry
Real queued work	blocks termination	still blocks termination
Python: queued coordinator reply	fails EndWorker (since #6522; before that it was silently dropped by a destructive get())	not counted as work; kept in the queue and the call succeeds

Changes:

• EndHandler.scala: scan the arrival queue for pending work instead of testing bare emptiness; the warning names the payload type and channel instead of dumping the element; scaladoc no longer claims EndWorker is the last message a worker receives.
• end_worker_handler.py / internal_queue.py: reply-aware has_unprocessed_work() (an AtomicInteger work counter maintained in put/get, since the multi-queue offers no iteration; it also sees work in paused/backpressured sub-queues that size()/get() hide) plus a non-destructive InternalQueue.peek() used for the failure diagnostic. This builds on fix(pyamber): stop EndWorker from consuming straggler messages #6522, which stopped the handler from consuming stragglers; after this PR the Python worker also stops failing on the benign trailing reply, keeping it at parity with the Scala side (same "worker still has unprocessed messages" failure string).
• RegionExecutionManager.scala: fixed the false doc ("This will be the last message each worker receives") and demoted the detail-free duplicate "Error when terminating region X." WARN to debug — terminateWorkersWithRetry logs the same failure with attempt context (WARN) or gives up loudly (ERROR), unchanged.

Any related issues, documentation, discussions?

Related: #6796 — this addresses the EndWorker retry-storm item behaviorally (the expected race no longer fails the first attempt) rather than by adjusting log levels. Follow-up to the CI log-verbosity work in #6797 and to #6522 (Python straggler-consumption fix). Adjacent teardown defects found during this investigation are filed separately: #6918, #6919, #6920, #6921, #6922, #6923, #6924, #6925.

How was this PR tested?

• EndHandlerSpec (the three pre-existing tests are untouched — control invocations and actor commands still fail endWorker): new regression tests for the exact CI payload (ReturnInvocation(2, EmptyReturn()) queued → success), multiple queued replies, a ControlInvocation queued behind a reply (the scan must not stop at the head), and queue-size preservation on both paths. Written test-first: the two new success tests fail without the EndHandler change.
• test_end_worker_handler.py: keeps the five tests introduced by fix(pyamber): stop EndWorker from consuming straggler messages #6522 verbatim (their straggler elements still count as work) and adds the reply-aware cases — a queued coordinator reply succeeds and stays queued (size() == 1 afterwards), a ControlInvocation behind a reply still fails, queued data still fails, and repeated calls are idempotent across coordinator retries.
• test_internal_queue.py: peek() (None on empty, non-destructive) and has_unprocessed_work() (ignores replies; counts invocations, data, ECMs, SYSTEM items; sees data in a disabled sub-queue; not counted for rejected elements).
• Untouched and green as evidence the retry contract is unchanged: RegionExecutionManagerSpec, WorkflowExecutionManagerSpec, DPThreadSpec, WorkerSpec, AsyncRPCClientSpec.
• Lint: scalafmtCheck, scalafixAll --check, ruff check + ruff format --check all pass.

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, @eugenegujing
  You can notify them by mentioning @Yicong-Huang, @eugenegujing in a comment.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 92.68293% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 78.95%. Comparing base (49f9e2c) to head (490f820).

Files with missing lines	Patch %	Lines
...chitecture/worker/promisehandlers/EndHandler.scala	85.00%	1 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff            @@
##               main    #6916   +/-   ##
=========================================
  Coverage     78.95%   78.95%           
- Complexity     3795     3801    +6     
=========================================
  Files          1161     1161           
  Lines         46084    46114   +30     
  Branches       5110     5117    +7     
=========================================
+ Hits          36384    36410   +26     
- Misses         8077     8079    +2     
- Partials       1623     1625    +2     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)		Carriedforward from 49f9e2c
agent-service	76.76% <ø> (ø)		Carriedforward from 49f9e2c
amber	72.21% <85.00%> (+<0.01%)	⬆️
computing-unit-managing-service	20.49% <ø> (ø)		Carriedforward from 49f9e2c
config-service	66.66% <ø> (ø)		Carriedforward from 49f9e2c
file-service	67.21% <ø> (ø)		Carriedforward from 49f9e2c
frontend	82.91% <ø> (ø)		Carriedforward from 49f9e2c
notebook-migration-service	78.94% <ø> (ø)		Carriedforward from 49f9e2c
pyamber	95.40% <100.00%> (+0.01%)	⬆️
workflow-compiling-service	55.14% <ø> (ø)		Carriedforward from 49f9e2c

*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.

[github-actions]

⚠️ Benchmark changes need a look

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

Compared against main 49f9e2c 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	402	0.245	23,911/34,743/34,743 us	🟢 -12.9% / 🔴 +112.2%
🔴	bs=100 sw=10 sl=64	792	0.483	125,006/149,286/149,286 us	🔴 +5.2% / 🔴 +36.6%
🟢	bs=1000 sw=10 sl=64	913	0.557	1,092,965/1,135,757/1,135,757 us	🟢 -5.1% / 🔴 -8.6%

Baseline details

Latest main 49f9e2c from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	402 tuples/sec	369 tuples/sec	759.72 tuples/sec	+8.9%	-47.1%
bs=10 sw=10 sl=64	MB/s	0.245 MB/s	0.225 MB/s	0.464 MB/s	+8.9%	-47.2%
bs=10 sw=10 sl=64	p50	23,911 us	27,465 us	12,653 us	-12.9%	+89.0%
bs=10 sw=10 sl=64	p95	34,743 us	39,396 us	16,371 us	-11.8%	+112.2%
bs=10 sw=10 sl=64	p99	34,743 us	39,396 us	19,941 us	-11.8%	+74.2%
bs=100 sw=10 sl=64	throughput	792 tuples/sec	826 tuples/sec	966.78 tuples/sec	-4.1%	-18.1%
bs=100 sw=10 sl=64	MB/s	0.483 MB/s	0.504 MB/s	0.59 MB/s	-4.2%	-18.1%
bs=100 sw=10 sl=64	p50	125,006 us	118,861 us	103,654 us	+5.2%	+20.6%
bs=100 sw=10 sl=64	p95	149,286 us	155,008 us	109,308 us	-3.7%	+36.6%
bs=100 sw=10 sl=64	p99	149,286 us	155,008 us	116,369 us	-3.7%	+28.3%
bs=1000 sw=10 sl=64	throughput	913 tuples/sec	914 tuples/sec	997.95 tuples/sec	-0.1%	-8.5%
bs=1000 sw=10 sl=64	MB/s	0.557 MB/s	0.558 MB/s	0.609 MB/s	-0.2%	-8.6%
bs=1000 sw=10 sl=64	p50	1,092,965 us	1,087,281 us	1,007,348 us	+0.5%	+8.5%
bs=1000 sw=10 sl=64	p95	1,135,757 us	1,196,827 us	1,050,149 us	-5.1%	+8.2%
bs=1000 sw=10 sl=64	p99	1,135,757 us	1,196,827 us	1,079,497 us	-5.1%	+5.2%

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,497.68,200,128000,402,0.245,23911.47,34743.20,34743.20
1,100,10,64,20,2526.29,2000,1280000,792,0.483,125006.10,149285.52,149285.52
2,1000,10,64,20,21912.34,20000,12800000,913,0.557,1092964.81,1135757.19,1135757.19