fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker

[aglinxinyuan]

What changes were proposed in this PR?

Fixes the ~1-in-10 LoopIntegrationSpec flake in amber-integration. Region termination logs:

java.lang.IllegalStateException: worker still has unprocessed messages
  at ...promisehandlers.EndHandler.endWorker(EndHandler.scala:51)
[WARN] Failed to terminate region N on attempt 1 of 150. Retrying in 200 ms.

Root cause. The "unprocessed" message is always a ReturnInvocation — the coordinator's ack for an RPC the worker itself sent fire-and-forget (workerExecutionCompleted / portCompleted). The race is inherent: that very RPC is what makes the coordinator decide to end the worker, so the ack and EndWorker are concurrently in flight:

worker ──workerExecutionCompleted──▶ coordinator
                                        │ decides region is done
worker ◀───────ack (ReturnInvocation)───┤   (ack still in flight)
worker ◀───────EndWorker────────────────┘
        EndWorker sees the queued ack → throws

The worker never awaits these acks, so an ack-only backlog carries no work. Loops amplify the race because every iteration terminates and re-executes regions — and the suite's single retry doesn't save runs where it strikes twice (same fail-fast previously seen in the DataProcessingSpec flake, #5614).

Fix (both sides of the engine, mirrored):

Side	Before	After
Scala EndHandler	throws on any queued message	ack-only backlog → warn + succeed; anything else (control invocation, data, ECM, actor command) still fails so the termination retry lets the worker drain it
Python EndWorkerHandler	consumed one message as a side effect of logging it, then bare assert empty (crashed on 2 queued acks; silently swallowed real work)	same ack-only leniency; non-ack messages are put back on the queue before failing

Scope note. Since #5737 (terminate-with-retry), this race usually self-heals on the retry — so its steady-state cost is termination latency, log noise, and the residual risk of repeated collisions across a loop's many terminations, rather than a guaranteed test failure. This PR removes the spurious failure mode at the source; if amber-integration flakes persist after it, the next suspect is a separate silent region-transition hang (a region's last worker completes but the region is never terminated, no EndHandler warning) that has been observed independently and is not addressed here.

Any related issues, documentation, discussions?

Closes #6891. Related: #5614 (the same fail-fast race in DataProcessingSpec) and #6522 (ports fail-and-retry semantics to the Python handler; the Python change here subsumes its non-destructive re-queue and additionally adds the ack-leniency on both sides).

How was this PR tested?

• EndHandlerSpec (Scala): new cases — ack-only backlog succeeds; ack + real work still fails. Existing cases (empty queue succeeds; control message / actor command fail) unchanged and passing.
• test_end_worker_handler.py (Python, new): ack-only backlog succeeds; empty queue succeeds; ack + control invocation fails and the control invocation survives in the queue (pins the old swallow-one bug).
• Full local battery: test-compile, scalafmtCheckAll, scalafixAll --check, ruff format+check, and the loop-adjacent Python suite (68 tests) all green (Java 17).
• The real-world validation is statistical: the EndHandler termination-retry warnings (and the failure mode they can escalate into) should disappear from amber-integration logs on subsequent runs.

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.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 95.65217% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 78.54%. Comparing base (f521750) to head (4b01b25).

Files with missing lines	Patch %	Lines
...chitecture/worker/promisehandlers/EndHandler.scala	87.50%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6892      +/-   ##
============================================
- Coverage     78.70%   78.54%   -0.17%     
+ Complexity     3740     3737       -3     
============================================
  Files          1161     1161              
  Lines         46084    46073      -11     
  Branches       5110     5106       -4     
============================================
- Hits          36269    36186      -83     
- Misses         8206     8268      +62     
- Partials       1609     1619      +10     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)		Carriedforward from eae4f21
agent-service	76.76% <ø> (ø)		Carriedforward from eae4f21
amber	71.48% <87.50%> (-0.05%)	⬇️
computing-unit-managing-service	20.49% <ø> (ø)		Carriedforward from eae4f21
config-service	66.66% <ø> (ø)		Carriedforward from eae4f21
file-service	67.21% <ø> (ø)		Carriedforward from eae4f21
frontend	82.57% <ø> (-0.34%)	⬇️	Carriedforward from eae4f21
notebook-migration-service	78.94% <ø> (ø)		Carriedforward from eae4f21
pyamber	95.39% <100.00%> (+0.01%)	⬆️
workflow-compiling-service	55.14% <ø> (ø)		Carriedforward from eae4f21

*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

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

Compared against main f521750 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	374	0.228	25,715/37,701/37,701 us	🔴 +22.1% / 🔴 +130.3%
⚪	bs=100 sw=10 sl=64	820	0.501	122,649/140,824/140,824 us	⚪ within ±5% / 🔴 +28.8%
⚪	bs=1000 sw=10 sl=64	920	0.562	1,087,967/1,126,454/1,126,454 us	⚪ within ±5% / 🔴 +8.0%

Baseline details

Latest main f521750 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	374 tuples/sec	449 tuples/sec	759.72 tuples/sec	-16.7%	-50.8%
bs=10 sw=10 sl=64	MB/s	0.228 MB/s	0.274 MB/s	0.464 MB/s	-16.8%	-50.8%
bs=10 sw=10 sl=64	p50	25,715 us	21,065 us	12,653 us	+22.1%	+103.2%
bs=10 sw=10 sl=64	p95	37,701 us	33,233 us	16,371 us	+13.4%	+130.3%
bs=10 sw=10 sl=64	p99	37,701 us	33,233 us	19,941 us	+13.4%	+89.1%
bs=100 sw=10 sl=64	throughput	820 tuples/sec	827 tuples/sec	966.78 tuples/sec	-0.8%	-15.2%
bs=100 sw=10 sl=64	MB/s	0.501 MB/s	0.505 MB/s	0.59 MB/s	-0.8%	-15.1%
bs=100 sw=10 sl=64	p50	122,649 us	119,407 us	103,654 us	+2.7%	+18.3%
bs=100 sw=10 sl=64	p95	140,824 us	134,875 us	109,308 us	+4.4%	+28.8%
bs=100 sw=10 sl=64	p99	140,824 us	134,875 us	116,369 us	+4.4%	+21.0%
bs=1000 sw=10 sl=64	throughput	920 tuples/sec	914 tuples/sec	997.95 tuples/sec	+0.7%	-7.8%
bs=1000 sw=10 sl=64	MB/s	0.562 MB/s	0.558 MB/s	0.609 MB/s	+0.7%	-7.7%
bs=1000 sw=10 sl=64	p50	1,087,967 us	1,093,249 us	1,007,348 us	-0.5%	+8.0%
bs=1000 sw=10 sl=64	p95	1,126,454 us	1,130,191 us	1,050,149 us	-0.3%	+7.3%
bs=1000 sw=10 sl=64	p99	1,126,454 us	1,130,191 us	1,079,497 us	-0.3%	+4.3%

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,534.45,200,128000,374,0.228,25715.14,37700.55,37700.55
1,100,10,64,20,2438.83,2000,1280000,820,0.501,122648.69,140824.20,140824.20
2,1000,10,64,20,21733.59,20000,12800000,920,0.562,1087967.06,1126454.32,1126454.32