Progress

Author: Evanthx

temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java

@@ -139,7 +139,7 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) {
     // Server >=1.31 with EnableCHASMSignalBacklinks returns a backlink pointing at the signal
     // event; older servers leave it unset. Propagate when present.
     if (CurrentNexusOperationContext.isNexusContext() && response.hasLink()) {
-      CurrentNexusOperationContext.get().setSignalWorkflowResponseLink(response.getLink());
+      CurrentNexusOperationContext.get().addSignalWorkflowResponseLink(response.getLink());
     }
     return new WorkflowSignalOutput();
   }
@@ -178,7 +178,7 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu
     // Server >=1.31 with EnableCHASMSignalBacklinks returns a backlink pointing at the signal
     // event; older servers leave it unset. Propagate when present.
     if (CurrentNexusOperationContext.isNexusContext() && response.hasSignalLink()) {
-      CurrentNexusOperationContext.get().setSignalWorkflowResponseLink(response.getSignalLink());
+      CurrentNexusOperationContext.get().addSignalWorkflowResponseLink(response.getSignalLink());
     }
     // TODO currently SignalWithStartWorkflowExecutionResponse doesn't have eagerWorkflowTask.
     //  We should wire it when it's implemented server-side.

temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java

@@ -6,6 +6,7 @@
 import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor;
 import io.temporal.nexus.NexusOperationContext;
 import io.temporal.nexus.NexusOperationInfo;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
@@ -21,12 +22,17 @@ public class InternalNexusOperationContext {
   // workflow client (signal, signalWithStart) can attach them to outgoing requests via
   // SignalWorkflowExecutionRequest.links, matching the Go SDK's NexusOperationLinksKey ctx value.
   private List<Link> nexusOperationLinks = Collections.emptyList();
-  // Backlink returned by SignalWorkflowExecutionResponse.link /
-  // SignalWithStartWorkflowExecutionResponse.signal_link.
-  // Populated by the workflow client and consumed by the task handler when building
-  // StartOperationResponse, so the caller workflow gets a link pointing at the signal event on
-  // the callee.
-  private Link signalWorkflowResponseLink;
+  // Backlinks returned by SignalWorkflowExecutionResponse.link /
+  // SignalWithStartWorkflowExecutionResponse.signal_link. One entry per signal RPC issued from
+  // within the Nexus operation handler. Drained by the task handler when building
+  // StartOperationResponse so every signal the handler issues gets a corresponding link on the
+  // caller workflow's history event.
+  //
+  // NOTE: this context is only safe for use from the single thread that runs the operation
+  // handler (the Nexus task executor's thread). Handlers that spawn their own threads to issue
+  // signals will not see the thread-local context, so the links from those signals will not
+  // propagate.
+  private final List<Link> signalWorkflowResponseLinks = new ArrayList<>();
 
   public InternalNexusOperationContext(
       String namespace,
@@ -93,12 +99,20 @@ public List<Link> getNexusOperationLinks() {
     return nexusOperationLinks;
   }
 
-  public void setSignalWorkflowResponseLink(Link link) {
-    this.signalWorkflowResponseLink = link;
+  /**
+   * Append a backlink returned by a signal-class RPC (signal or signalWithStart). Each signal the
+   * operation handler issues should add one entry; the task handler drains the list when building
+   * the operation's StartOperationResponse.
+   */
+  public void addSignalWorkflowResponseLink(Link link) {
+    if (link != null) {
+      this.signalWorkflowResponseLinks.add(link);
+    }
   }
 
-  public Link getSignalWorkflowResponseLink() {
-    return signalWorkflowResponseLink;
+  /** Backlinks from every signal RPC issued by the handler. Never null; may be empty. */
+  public List<Link> getSignalWorkflowResponseLinks() {
+    return signalWorkflowResponseLinks;
   }
 
   private class NexusOperationContextImpl implements NexusOperationContext {

temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java

@@ -301,10 +301,20 @@ private StartOperationResponse handleStartOperation(
                     "Invalid link URL: " + link.getUrl(),
                     e);
               }
+              // LinkConverter only returns a WorkflowEvent-shaped common.v1.Link; nexus links of
+              // other shapes (e.g. non-temporal URLs) come back null and are intentionally not
+              // forwarded onto SignalWorkflowExecutionRequest.links, which requires the
+              // WorkflowEvent variant. Log so a debugging session can see what was dropped.
               io.temporal.api.common.v1.Link commonLink =
                   LinkConverter.nexusLinkToWorkflowEvent(link);
               if (commonLink != null) {
                 inboundCommonLinks.add(commonLink);
+              } else {
+                log.warn(
+                    "Dropping inbound Nexus link from outbound signal propagation: type='{}',"
+                        + " url='{}' (not a parseable temporal WorkflowEvent link)",
+                    link.getType(),
+                    link.getUrl());
               }
             });
     CurrentNexusOperationContext.get().setNexusOperationLinks(inboundCommonLinks);
@@ -318,27 +328,30 @@ private StartOperationResponse handleStartOperation(
       try {
         OperationStartResult<HandlerResultContent> result =
             startOperation(context, operationStartDetails.build(), input.build());
-        // If a signal RPC issued by the handler returned a backlink, propagate it to the caller
-        // so the caller workflow's history event links to the signal event on the callee. Same
-        // backlink applies to both sync and async response variants.
-        io.temporal.api.nexus.v1.Link signalBacklink = null;
-        io.temporal.api.common.v1.Link signalResponseLink =
-            CurrentNexusOperationContext.get().getSignalWorkflowResponseLink();
-        if (signalResponseLink != null && signalResponseLink.hasWorkflowEvent()) {
-          signalBacklink =
+        // If signal/signalWithStart RPCs issued by the handler returned backlinks, propagate
+        // them to the caller so the caller workflow's history event links to each signal event
+        // on the callee. Same set of backlinks applies to both sync and async response variants.
+        List<io.temporal.api.nexus.v1.Link> signalBacklinks = new ArrayList<>();
+        for (io.temporal.api.common.v1.Link signalResponseLink :
+            CurrentNexusOperationContext.get().getSignalWorkflowResponseLinks()) {
+          if (!signalResponseLink.hasWorkflowEvent()) {
+            continue;
+          }
+          io.temporal.api.nexus.v1.Link converted =
               LinkConverter.workflowEventToNexusLink(signalResponseLink.getWorkflowEvent());
+          if (converted != null) {
+            signalBacklinks.add(converted);
+          }
         }
 
         if (result.isSync()) {
-          StartOperationResponse.Sync.Builder syncBuilder =
+          startResponseBuilder.setSyncSuccess(
               StartOperationResponse.Sync.newBuilder()
-                  .setPayload(Payload.parseFrom(result.getSyncResult().getDataBytes()));
-          if (signalBacklink != null) {
-            syncBuilder.addLinks(signalBacklink);
-          }
-          startResponseBuilder.setSyncSuccess(syncBuilder.build());
+                  .setPayload(Payload.parseFrom(result.getSyncResult().getDataBytes()))
+                  .addAllLinks(signalBacklinks)
+                  .build());
         } else {
-          StartOperationResponse.Async.Builder asyncBuilder =
+          startResponseBuilder.setAsyncSuccess(
               StartOperationResponse.Async.newBuilder()
                   .setOperationId(result.getAsyncOperationToken())
                   .setOperationToken(result.getAsyncOperationToken())
@@ -350,11 +363,9 @@ private StartOperationResponse handleStartOperation(
                                       .setType(link.getType())
                                       .setUrl(link.getUri().toString())
                                       .build())
-                          .collect(Collectors.toList()));
-          if (signalBacklink != null) {
-            asyncBuilder.addLinks(signalBacklink);
-          }
-          startResponseBuilder.setAsyncSuccess(asyncBuilder.build());
+                          .collect(Collectors.toList()))
+                  .addAllLinks(signalBacklinks)
+                  .build());
         }
       } catch (OperationException e) {
         throw e;