(texera) branch gh-readonly-queue/main/pr-8002-e750074eb92f57a6a945b2a09d07b8e742c0b82f created (now e54ecd2970)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-8002-e750074eb92f57a6a945b2a09d07b8e742c0b82f
in repository https://gitbox.apache.org/repos/asf/texera.git


      at e54ecd2970 fix(amber): guard resends by destination (#8002)

This branch includes the following new commits:

     new e54ecd2970 fix(amber): guard resends by destination (#8002)

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(texera) 01/01: fix(amber): guard resends by destination (#8002)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8002-e750074eb92f57a6a945b2a09d07b8e742c0b82f
in repository https://gitbox.apache.org/repos/asf/texera.git

commit e54ecd2970ac8d2099c1e8f061cf053d32dd5e85
Author: anthonychengit <an...@gmail.com>
AuthorDate: Fri Aug 28 03:37:51 2026 +0000

    fix(amber): guard resends by destination (#8002)
    
    ### 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.
    
    ```text
    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:
    
    ```text
    sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.common.PekkoMessageTransferServiceSpec"
    ```
    
    Result: 5 tests passed.
    
    ```text
    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.
---
 .../common/PekkoMessageTransferService.scala       |  2 +-
 .../common/PekkoMessageTransferServiceSpec.scala   | 63 ++++++++++++++++++++++
 2 files changed, 64 insertions(+), 1 deletion(-)

diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferService.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferService.scala
index 16c9e394ca..6af9d6cdcf 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferService.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferService.scala
@@ -172,7 +172,7 @@ class PekkoMessageTransferService(
         if (msgsNeedResend.nonEmpty) {
           logger.debug(s"output for $channel: ${cc.getStatusReport}")
         }
-        if (refService.hasActorRef(channel.fromWorkerId)) {
+        if (refService.hasActorRef(channel.toWorkerId)) {
           msgsNeedResend.foreach { msg =>
             refService.forwardToActor(msg)
           }
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferServiceSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferServiceSpec.scala
index a3858c8a9a..0653cd6348 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferServiceSpec.scala
@@ -80,6 +80,13 @@ class PekkoMessageTransferServiceSpec
   private def net(id: Long, chan: ChannelIdentity, seq: Long): NetworkMessage =
     NetworkMessage(id, fifo(chan, seq))
 
+  private def backdateSentTime(cc: CongestionControl, id: Long): Unit = {
+    val field = classOf[CongestionControl].getDeclaredField("sentTime")
+    field.setAccessible(true)
+    val sentTime = field.get(cc).asInstanceOf[mutable.LongMap[Long]]
+    sentTime(id) = System.currentTimeMillis() - cc.resendTimeLimit - 1
+  }
+
   // Pin the assumed payload size so this test fails loudly if the size accounting
   // changes in a way that would invalidate the credit math below.
   assert(WorkflowMessage.getInMemSize(fifo(dataChannel(), 0L)) == 200L)
@@ -189,6 +196,46 @@ class PekkoMessageTransferServiceSpec
     service.stop() // cancels the (already-cancelled) captured handles; must not throw
     assert(service.channelToCC.contains(chan))
   }
+
+  "checkResend" should "resend a timed-out message when its destination is registered" in {
+    val actorService = new CapturingActorService(actorId, freshContext())
+    val sender = ActorVirtualIdentity("known-sender")
+    val destination = ActorVirtualIdentity("known-destination")
+    val refService = new RecordingRefService(actorService, Set(destination))
+    val service = new PekkoMessageTransferService(actorService, refService, _ => ())
+    val channel = ChannelIdentity(sender, destination, isControl = false)
+    val message = net(10L, channel, 0L)
+    val cc = new CongestionControl()
+    cc.markMessageInTransit(message)
+    backdateSentTime(cc, message.messageId)
+    service.channelToCC(channel) = cc
+
+    service.initialize()
+    actorService.capturedCallables.head()
+
+    assert(refService.queriedIds == Seq(destination))
+    assert(refService.forwardedMessages == Seq(message))
+  }
+
+  it should "not resend a timed-out message when only its sender is registered" in {
+    val actorService = new CapturingActorService(actorId, freshContext())
+    val sender = ActorVirtualIdentity("known-sender")
+    val destination = ActorVirtualIdentity("removed-destination")
+    val refService = new RecordingRefService(actorService, Set(sender))
+    val service = new PekkoMessageTransferService(actorService, refService, _ => ())
+    val channel = ChannelIdentity(sender, destination, isControl = false)
+    val message = net(11L, channel, 0L)
+    val cc = new CongestionControl()
+    cc.markMessageInTransit(message)
+    backdateSentTime(cc, message.messageId)
+    service.channelToCC(channel) = cc
+
+    service.initialize()
+    actorService.capturedCallables.head()
+
+    assert(refService.queriedIds == Seq(destination))
+    assert(refService.forwardedMessages.isEmpty)
+  }
 }
 
 /** Minimal actor used only to obtain a real `ActorContext` from Pekko TestKit. */
@@ -215,3 +262,19 @@ class CapturingActorService(vid: ActorVirtualIdentity, ac: ActorContext)
     Cancellable.alreadyCancelled
   }
 }
+
+class RecordingRefService(
+    actorService: PekkoActorService,
+    knownIds: Set[ActorVirtualIdentity]
+) extends PekkoActorRefMappingService(actorService) {
+
+  val queriedIds: mutable.ArrayBuffer[ActorVirtualIdentity] = mutable.ArrayBuffer()
+  val forwardedMessages: mutable.ArrayBuffer[NetworkMessage] = mutable.ArrayBuffer()
+
+  override def hasActorRef(id: ActorVirtualIdentity): Boolean = {
+    queriedIds += id
+    knownIds.contains(id)
+  }
+
+  override def forwardToActor(msg: NetworkMessage): Unit = forwardedMessages += msg
+}