fix(amber): release the reconfiguration service's engine callback on teardown

[aglinxinyuan]

What changes were proposed in this PR?

ExecutionReconfigurationService.registerWorkerCompletionCallback discarded the Disposable returned by client.registerCallback, so its engine callback outlived the service. It now hands it to addSubscription(...), matching registerCompletionDiffHandler directly below it.

It was the only registerCallback site in org.apache.texera.web.service not doing this:

File	sites	wrapped
ExecutionStatsService.scala	5	yes
ExecutionResultService.scala	3	yes
ExecutionConsoleService.scala	2	yes
ExecutionRuntimeService.scala	1	yes
ExecutionReconfigurationService.scala	1	no

Why this actually leaks, rather than being tidied up by something else

The teardown path is live: WorkflowExecutionService.unsubscribeAll (WorkflowExecutionService.scala:185) calls executionReconfigurationService.unsubscribeAll(), which released every other subscription but not this one.

And client.shutdown() is not an alternative release — it flips isActive and poison-pills the ClientActor, but never touches registeredObservables, which is written only in registerCallback and never cleared. The PublishSubject chain therefore keeps the subscriber closure, and through it the service, its ExecutionStateStore and its Workflow, for as long as the AmberClient is reachable — which WorkflowExecutionService.client guarantees.

The fix is pinned

New test: "the worker completion callback" should "release the engine subscription once the service is unsubscribed". Verified with the production file reverted and restored:

	production reverted	with fix
ExecutionReconfigurationServiceSpec	13 passed, 1 failed	14 passed

The before-state failure is the right one — ArrayBuffer() did not contain element class …UpdateExecutorCompleted, i.e. the disposable was never disposed because it was never registered.

Two additions to the existing TestAmberClient double were needed, and the reasons are worth stating: an explicit disposedCallbacks record, because disposal-removal from the callbacks map alone is indistinguishable from "never registered" or "cleared by reset"; and a fireIfRegistered helper, because the strict fire calls fail(...) when nothing is registered and so cannot express "a late event is inert".

Two comments were rewritten, not just tests added

Both would otherwise have contradicted the code:

1. TestAmberClient.registerCallback's scaladoc said production "currently DISCARDS it" and that "no test below fires an engine event after unsubscribeAll, so the suite neither depends on the leak nor breaks when it is fixed." Both clauses are now false.
2. The routing comment in "stop announcing completions once the service is unsubscribed" called the engine callback an "unrelated, currently broken, seam". The routing decision — drive that test through onWorkerReconfigured rather than the engine event — is still right and unchanged, but re-justified: the two are separate subscriptions, so routing through the engine event would leave an empty batch explainable by either one being released. That test's assertion is untouched, so it stays agnostic exactly as test(amber): cover the reconfiguration service past its own test seams #7692 intended.

Verification

• ExecutionReconfigurationServiceSpec 14/14.
• Dependents and neighbours: ExecutionRuntimeServiceSpec, WorkflowExecutionServiceSpec, ExecutionStatsServiceSpec, ExecutionConsoleServiceSpec, WorkflowWebsocketResourceSpec, TexeraWebSocketEventSpec — 59/59 across 6 suites, 0 aborted.
• scalafmtCheck, Test/scalafmtCheck, scalafixAll --check all pass. (The one scalafix warning is a pre-existing // scalafix:ok in OutputManagerSpec, unrelated.)

One thing deliberately left alone

The suite still carries an assertion recorded as observed, not endorsed: an N-worker operator fires N ModifyLogicCompletedEvents, which contradicts registerCompletionDiffHandler's own comment claiming the frontend is notified once all workers finish. That is a separate defect with its own decision to make, so this PR does not touch it.

Any related issues, documentation, discussions?

Closes #7822

How was this PR tested?

STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly org.apache.texera.web.service.ExecutionReconfigurationServiceSpec"

[info] Total number of tests run: 14
[info] Tests: succeeded 14, failed 0, canceled 0, ignored 0, pending 0

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

Generated-by: Claude Code (Opus 5)

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[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	Change detected on this branch — label added; this fix is queued to backport here. Requested review from @xuang7.

Auto-label run.

[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]

✅ No material benchmark regressions detected

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

Compared against main 2edbdf9 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.246	24,567/29,781/29,781 us	🟢 -16.2% / 🔴 +98.6%
🟢	bs=100 sw=10 sl=64	812	0.496	122,786/143,113/143,113 us	🟢 -20.0% / 🔴 +34.9%
⚪	bs=1000 sw=10 sl=64	928	0.567	1,079,662/1,137,546/1,137,546 us	⚪ within ±5% / 🔴 +12.7%

Baseline details

Latest main 2edbdf9 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	402 tuples/sec	407 tuples/sec	780.31 tuples/sec	-1.2%	-48.5%
bs=10 sw=10 sl=64	MB/s	0.246 MB/s	0.249 MB/s	0.476 MB/s	-1.2%	-48.3%
bs=10 sw=10 sl=64	p50	24,567 us	23,591 us	12,590 us	+4.1%	+95.1%
bs=10 sw=10 sl=64	p95	29,781 us	35,549 us	14,995 us	-16.2%	+98.6%
bs=10 sw=10 sl=64	p99	29,781 us	35,549 us	19,141 us	-16.2%	+55.6%
bs=100 sw=10 sl=64	throughput	812 tuples/sec	813 tuples/sec	1,006 tuples/sec	-0.1%	-19.3%
bs=100 sw=10 sl=64	MB/s	0.496 MB/s	0.496 MB/s	0.614 MB/s	0.0%	-19.2%
bs=100 sw=10 sl=64	p50	122,786 us	118,173 us	99,387 us	+3.9%	+23.5%
bs=100 sw=10 sl=64	p95	143,113 us	178,858 us	106,056 us	-20.0%	+34.9%
bs=100 sw=10 sl=64	p99	143,113 us	178,858 us	113,689 us	-20.0%	+25.9%
bs=1000 sw=10 sl=64	throughput	928 tuples/sec	926 tuples/sec	1,043 tuples/sec	+0.2%	-11.0%
bs=1000 sw=10 sl=64	MB/s	0.567 MB/s	0.565 MB/s	0.637 MB/s	+0.4%	-10.9%
bs=1000 sw=10 sl=64	p50	1,079,662 us	1,077,761 us	965,215 us	+0.2%	+11.9%
bs=1000 sw=10 sl=64	p95	1,137,546 us	1,148,055 us	1,009,610 us	-0.9%	+12.7%
bs=1000 sw=10 sl=64	p99	1,137,546 us	1,148,055 us	1,039,108 us	-0.9%	+9.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,496.96,200,128000,402,0.246,24566.69,29781.43,29781.43
1,100,10,64,20,2462.30,2000,1280000,812,0.496,122785.79,143112.99,143112.99
2,1000,10,64,20,21543.44,20000,12800000,928,0.567,1079662.22,1137545.72,1137545.72

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 91.50%. Comparing base (2edbdf9) to head (73dfb45).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #7823      +/-   ##
============================================
- Coverage     91.50%   91.50%   -0.01%     
+ Complexity     4501     4499       -2     
============================================
  Files          1177     1177              
  Lines         47489    47490       +1     
  Branches       5324     5324              
============================================
- Hits          43457    43455       -2     
- Misses         2366     2367       +1     
- Partials       1666     1668       +2     

Flag	Coverage Δ		*Carryforward flag
access-control-service	81.00% <ø> (ø)		Carriedforward from 2edbdf9
agent-service	98.62% <ø> (ø)		Carriedforward from 2edbdf9
amber	88.02% <100.00%> (-0.02%)	⬇️
computing-unit-managing-service	73.67% <ø> (ø)		Carriedforward from 2edbdf9
config-service	86.73% <ø> (ø)		Carriedforward from 2edbdf9
file-service	75.74% <ø> (ø)		Carriedforward from 2edbdf9
frontend	93.24% <ø> (ø)		Carriedforward from 2edbdf9
notebook-migration-service	79.13% <ø> (ø)		Carriedforward from 2edbdf9
pyamber	97.57% <ø> (ø)		Carriedforward from 2edbdf9
workflow-compiling-service	77.19% <ø> (ø)		Carriedforward from 2edbdf9

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