fix(pyamber): stop EndWorker from consuming straggler messages

[aglinxinyuan]

What changes were proposed in this PR?

Root cause. The Python worker's EndWorkerHandler guard logs its "unprocessed messages" warning through input_queue.get() — a destructive, blocking read that removes a pending message — and then assert input_queue.is_empty(). With exactly one straggler message the straggler is silently dropped and EndWorker is acknowledged as success; with two or more, one message is still destroyed and the RPC fails with a bare AssertionError — and because the coordinator retries EndWorker, every retry that hits the guard eats another queued message until the last one is dropped under a success ack.

queue state at EndWorker	before	after
empty	ack	ack (unchanged)
1 straggler	straggler dropped, then ack	RPC fails, straggler kept
≥ 2 stragglers	1 dropped, then AssertionError	RPC fails, all kept

Changes:

• end_worker_handler.py: log via a non-destructive peek() and raise RuntimeError("worker still has unprocessed messages") instead of consuming a message and asserting. The raise rides the existing failure path — AsyncRPCServer.receive converts a handler exception into a ControlError reply, AsyncRPCClient.fulfillPromise on the coordinator turns it into a failed future, and RegionExecutionManager.terminateWorkersWithRetry re-sends EndWorker after the queue drains — the exact Python analogue of the Scala EndHandler's Future.exception:

coordinator                        worker input queue at EndWorker arrival
    |                              [ReturnInvocation, ..., EndWorker]
    |-- EndWorker ---------------->|
    |<- ControlError --------------|   peek() for the log; nothing consumed
    |   (retry after delay)
    |-- EndWorker ---------------->|   queue drained by the main loop
    |<- EmptyReturn ---------------|   safe to gracefulStop

• internal_queue.py: expose InternalQueue.peek(), delegating to the existing non-destructive LinkedBlockingMultiQueue.peek() (mirrors what the Scala EndHandler reads via inputMessageQueue.peek()).

Any related issues, documentation, discussions?

Closes #6521

How was this PR tested?

TDD — the tests were written first and fail against the unfixed handler (with 1 straggler: no exception is raised and the message vanishes; with ≥ 2: AssertionError instead of a clean RPC failure):

• New src/test/python/core/architecture/handlers/control/test_end_worker_handler.py: acks on an empty queue; fails the RPC with one straggler; does not consume the straggler; keeps all messages with two stragglers; acks again once the queue drains (the retry protocol end-to-end). This is the Python analogue of the Scala EndHandlerSpec.
• src/test/python/core/models/test_internal_queue.py: peek() on an empty queue returns None; peek() is non-destructive under repeated calls; peek() always surfaces exactly the element the next get() returns across sub-queues (deliberately not pinning control-before-data ordering — the registration-order quirk is already documented by this file's xfail).

Ran locally: cd amber && pytest -m "not integration" on the touched test files plus the full unit suite, and ruff check src/main/python src/test/python && ruff format --check src/main/python src/test/python.

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.

[Copilot]

Pull request overview

This PR fixes a teardown race in the Python worker where EndWorkerHandler could destructively consume (“drop”) pending input-queue messages while checking for stragglers, causing incorrect successful acks and/or misleading failures during coordinator retries.

Changes:

• Make EndWorkerHandler log pending work via a non-destructive peek() and fail the RPC with a clear RuntimeError instead of consuming via get() and asserting.
• Add InternalQueue.peek() (delegating to LinkedBlockingMultiQueue.peek()) to support non-destructive inspection.
• Add focused unit tests covering peek() semantics and the EndWorker retry protocol behavior (fail while stragglers exist; succeed once drained).

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

File	Description
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py	Switch EndWorker guard from destructive get()+assert to peek()+RuntimeError to preserve straggler messages and participate in retry semantics.
amber/src/main/python/core/models/internal_queue.py	Add InternalQueue.peek() API to expose non-destructive “next element” inspection.
amber/src/test/python/core/architecture/handlers/control/test_end_worker_handler.py	New tests validating EndWorker rejects stragglers without consuming them and succeeds after the queue drains.
amber/src/test/python/core/models/test_internal_queue.py	New tests validating peek() returns None on empty, is non-destructive, and matches the next get() across sub-queues.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 70.58%. Comparing base (39db110) to head (87d618b).
⚠️ Report is 3 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6522      +/-   ##
============================================
+ Coverage     70.57%   70.58%   +0.01%     
  Complexity     3425     3425              
============================================
  Files          1146     1146              
  Lines         45062    45062              
  Branches       4960     4960              
============================================
+ Hits          31802    31807       +5     
+ Misses        11613    11608       -5     
  Partials       1647     1647              

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)		Carriedforward from b352eb1
agent-service	76.76% <ø> (ø)		Carriedforward from b352eb1
amber	66.72% <ø> (ø)		Carriedforward from b352eb1
computing-unit-managing-service	17.72% <ø> (ø)		Carriedforward from b352eb1
config-service	52.30% <ø> (ø)		Carriedforward from b352eb1
file-service	66.80% <ø> (ø)		Carriedforward from b352eb1
frontend	69.36% <ø> (ø)		Carriedforward from b352eb1
notebook-migration-service	78.94% <ø> (ø)		Carriedforward from b352eb1
pyamber	91.85% <100.00%> (+0.11%)	⬆️
workflow-compiling-service	55.14% <ø> (ø)		Carriedforward from b352eb1

*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 39db110 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	375	0.229	25,188/34,364/34,364 us	🔴 +10.2% / 🔴 +123.7%
🔴	bs=100 sw=10 sl=64	796	0.486	123,099/146,256/146,256 us	🔴 +9.7% / 🔴 +38.0%
⚪	bs=1000 sw=10 sl=64	913	0.557	1,098,399/1,138,550/1,138,550 us	⚪ within ±5% / 🔴 +12.9%

Baseline details

Latest main 39db110 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	375 tuples/sec	407 tuples/sec	770.44 tuples/sec	-7.9%	-51.3%
bs=10 sw=10 sl=64	MB/s	0.229 MB/s	0.248 MB/s	0.47 MB/s	-7.7%	-51.3%
bs=10 sw=10 sl=64	p50	25,188 us	22,848 us	12,688 us	+10.2%	+98.5%
bs=10 sw=10 sl=64	p95	34,364 us	34,755 us	15,365 us	-1.1%	+123.7%
bs=10 sw=10 sl=64	p99	34,364 us	34,755 us	19,670 us	-1.1%	+74.7%
bs=100 sw=10 sl=64	throughput	796 tuples/sec	837 tuples/sec	1,004 tuples/sec	-4.9%	-20.7%
bs=100 sw=10 sl=64	MB/s	0.486 MB/s	0.511 MB/s	0.613 MB/s	-4.9%	-20.7%
bs=100 sw=10 sl=64	p50	123,099 us	118,606 us	99,406 us	+3.8%	+23.8%
bs=100 sw=10 sl=64	p95	146,256 us	133,284 us	105,989 us	+9.7%	+38.0%
bs=100 sw=10 sl=64	p99	146,256 us	133,284 us	119,372 us	+9.7%	+22.5%
bs=1000 sw=10 sl=64	throughput	913 tuples/sec	918 tuples/sec	1,035 tuples/sec	-0.5%	-11.8%
bs=1000 sw=10 sl=64	MB/s	0.557 MB/s	0.561 MB/s	0.631 MB/s	-0.7%	-11.8%
bs=1000 sw=10 sl=64	p50	1,098,399 us	1,089,807 us	973,037 us	+0.8%	+12.9%
bs=1000 sw=10 sl=64	p95	1,138,550 us	1,125,114 us	1,015,887 us	+1.2%	+12.1%
bs=1000 sw=10 sl=64	p99	1,138,550 us	1,125,114 us	1,047,206 us	+1.2%	+8.7%

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,533.24,200,128000,375,0.229,25187.79,34364.13,34364.13
1,100,10,64,20,2512.35,2000,1280000,796,0.486,123098.85,146256.22,146256.22
2,1000,10,64,20,21914.01,20000,12800000,913,0.557,1098398.99,1138550.32,1138550.32

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.