fix(amber): guard resends by destination

[anthonychengit]

What changes were proposed in this PR?

Use the destination worker, rather than the sender, when deciding whether an unacknowledged network message can be resent.

Before: timeout -> sender exists -> resend toward a removed destination
After:  timeout -> destination exists -> resend; otherwise keep the message pending

Regression tests cover both directions: a registered destination with an unknown sender resends successfully, while a registered sender with a missing destination does not resend.

Any related issues, documentation, discussions?

Closes #6921.

How was this PR tested?

The two regression tests were added first and both failed against the old sender guard. After the one-line fix:

sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.common.PekkoMessageTransferServiceSpec"

Result: 5 tests passed.

sbt "scalafixAll --check" "scalafmtCheckAll"
git diff --check

Result: all lint, formatting, and diff checks passed.

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

Generated-by: Codex was used for implementation and verification assistance. I reviewed the final code and test output before submission.

[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: @Ma77Ball, @aglinxinyuan
  You can notify them by mentioning @Ma77Ball, @aglinxinyuan in a comment.

[github-actions]

⚠️ Benchmark changes need a look

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

Compared against main 444fc58 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	558	0.34	17,065/31,252/31,252 us	🔴 +21.3% / 🔴 +125.8%
🟢	bs=100 sw=10 sl=64	1,251	0.764	79,291/94,682/94,682 us	🟢 -12.2% / 🟢 -9.4%
⚪	bs=1000 sw=10 sl=64	1,444	0.881	694,076/723,898/723,898 us	⚪ within ±5% / 🟢 -22.8%

Baseline details

Latest main 444fc58 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	558 tuples/sec	608 tuples/sec	894.86 tuples/sec	-8.2%	-37.6%
bs=10 sw=10 sl=64	MB/s	0.34 MB/s	0.371 MB/s	0.546 MB/s	-8.4%	-37.7%
bs=10 sw=10 sl=64	p50	17,065 us	15,819 us	11,139 us	+7.9%	+53.2%
bs=10 sw=10 sl=64	p95	31,252 us	25,771 us	13,843 us	+21.3%	+125.8%
bs=10 sw=10 sl=64	p99	31,252 us	25,771 us	16,839 us	+21.3%	+85.6%
bs=100 sw=10 sl=64	throughput	1,251 tuples/sec	1,280 tuples/sec	1,166 tuples/sec	-2.3%	+7.3%
bs=100 sw=10 sl=64	MB/s	0.764 MB/s	0.781 MB/s	0.711 MB/s	-2.2%	+7.4%
bs=100 sw=10 sl=64	p50	79,291 us	75,999 us	87,509 us	+4.3%	-9.4%
bs=100 sw=10 sl=64	p95	94,682 us	107,793 us	93,675 us	-12.2%	+1.1%
bs=100 sw=10 sl=64	p99	94,682 us	107,793 us	102,153 us	-12.2%	-7.3%
bs=1000 sw=10 sl=64	throughput	1,444 tuples/sec	1,435 tuples/sec	1,198 tuples/sec	+0.6%	+20.5%
bs=1000 sw=10 sl=64	MB/s	0.881 MB/s	0.876 MB/s	0.731 MB/s	+0.6%	+20.5%
bs=1000 sw=10 sl=64	p50	694,076 us	696,563 us	859,766 us	-0.4%	-19.3%
bs=1000 sw=10 sl=64	p95	723,898 us	758,165 us	902,897 us	-4.5%	-19.8%
bs=1000 sw=10 sl=64	p99	723,898 us	758,165 us	937,957 us	-4.5%	-22.8%

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,358.72,200,128000,558,0.340,17064.81,31251.65,31251.65
1,100,10,64,20,1598.32,2000,1280000,1251,0.764,79290.58,94682.40,94682.40
2,1000,10,64,20,13850.38,20000,12800000,1444,0.881,694075.85,723897.69,723897.69

[github-actions]

Backport PR opened: draft #8062 (#8062) to release/v1.2, assigned to @anthonychengit — needs manual work because the cherry-pick conflicts.