Code review results

Author: Evanthx

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

@@ -122,8 +122,15 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) {
 
     // If this signal is being issued from inside a Nexus operation handler, forward the inbound
     // Nexus task links so the SignalWorkflowExecution history event links back to the caller.
-    if (CurrentNexusOperationContext.isNexusContext()) {
+    boolean inNexusContext = CurrentNexusOperationContext.isNexusContext();
+    if (inNexusContext) {
       request.addAllLinks(CurrentNexusOperationContext.get().getNexusOperationLinks());
+    } else {
+      // Most signal calls (from a regular client or workflow) won't be in a Nexus context — this
+      // is normal. The log helps a debugger when a Nexus operation handler "mysteriously" lacks
+      // link propagation because it spawned a thread to issue the signal (the thread-local
+      // CurrentNexusOperationContext is invisible from that thread).
+      log.debug("signal RPC issued outside a Nexus operation context; no link propagation");
     }
 
     DataConverter dataConverterWitSignalContext =
@@ -138,7 +145,7 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) {
     SignalWorkflowExecutionResponse response = genericClient.signal(request.build());
     // 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()) {
+    if (inNexusContext && response.hasLink()) {
       CurrentNexusOperationContext.get().addBacklink(response.getLink());
     }
     return new WorkflowSignalOutput();
@@ -165,8 +172,12 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu
     // If this signalWithStart is being issued from inside a Nexus operation handler, forward
     // the inbound Nexus task links so both the WorkflowExecutionStarted and
     // WorkflowExecutionSignaled events on the callee link back to the caller.
-    if (CurrentNexusOperationContext.isNexusContext()) {
+    boolean inNexusContext = CurrentNexusOperationContext.isNexusContext();
+    if (inNexusContext) {
       requestBuilder.addAllLinks(CurrentNexusOperationContext.get().getNexusOperationLinks());
+    } else {
+      log.debug(
+          "signalWithStart RPC issued outside a Nexus operation context; no link propagation");
     }
     SignalWithStartWorkflowExecutionRequest request = requestBuilder.build();
     SignalWithStartWorkflowExecutionResponse response = genericClient.signalWithStart(request);
@@ -177,7 +188,7 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu
             .build();
     // 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()) {
+    if (inNexusContext && response.hasSignalLink()) {
       CurrentNexusOperationContext.get().addBacklink(response.getSignalLink());
     }
     // TODO currently SignalWithStartWorkflowExecutionResponse doesn't have eagerWorkflowTask.

temporal-sdk/src/main/java/io/temporal/internal/common/LinkConverter.java

@@ -6,6 +6,7 @@
 import io.temporal.api.enums.v1.EventType;
 import java.io.UnsupportedEncodingException;
 import java.net.URI;
+import java.net.URISyntaxException;
 import java.net.URLDecoder;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
@@ -33,8 +34,7 @@ public class LinkConverter {
       Link.WorkflowEvent.RequestIdReference.getDescriptor().getName();
 
   // Fully-qualified proto descriptor names used as the `type` field on nexus.v1.Link. Match the
-  // canonical server implementation at temporalio/temporal/common/nexus/link_converter.go so links
-  // round-trip cleanly across SDKs.
+  // server's Nexus link converter so links round-trip cleanly across SDKs.
   private static final String workflowEventType = Link.WorkflowEvent.getDescriptor().getFullName();
   private static final String nexusOperationType =
       Link.NexusOperation.getDescriptor().getFullName();
@@ -187,7 +187,7 @@ public static io.temporal.api.nexus.v1.Link nexusOperationToNexusLink(Link.Nexus
           .setUrl(url)
           .setType(nexusOperationType)
           .build();
-    } catch (Exception e) {
+    } catch (UnsupportedEncodingException e) {
       log.error("Failed to encode NexusOperation Nexus link URL", e);
     }
     return null;
@@ -200,6 +200,9 @@ public static io.temporal.api.nexus.v1.Link nexusOperationToNexusLink(Link.Nexus
    */
   public static Link nexusLinkToNexusOperation(io.temporal.api.nexus.v1.Link nexusLink) {
     try {
+
+      //Lots of if statements, but this way we double-check the validity of the link
+      //passed in.
       URI uri = new URI(nexusLink.getUrl());
       if (!"temporal".equals(uri.getScheme())) {
         log.error(
@@ -213,19 +216,40 @@ public static Link nexusLinkToNexusOperation(io.temporal.api.nexus.v1.Link nexus
             "Failed to parse NexusOperation Nexus link URL: invalid path: {}", uri.getRawPath());
         return null;
       }
+      if (!st.hasMoreTokens()) {
+        log.error(
+            "Failed to parse NexusOperation Nexus link URL: invalid path: {}", uri.getRawPath());
+        return null;
+      }
       String namespace = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
       if (!st.hasMoreTokens() || !"nexus-operations".equals(st.nextToken())) {
         log.error(
             "Failed to parse NexusOperation Nexus link URL: invalid path: {}", uri.getRawPath());
         return null;
       }
+      if (!st.hasMoreTokens()) {
+        log.error(
+            "Failed to parse NexusOperation Nexus link URL: invalid path: {}", uri.getRawPath());
+        return null;
+      }
       String operationId = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
+      if (!st.hasMoreTokens()) {
+        log.error(
+            "Failed to parse NexusOperation Nexus link URL: invalid path: {}", uri.getRawPath());
+        return null;
+      }
       String runId = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
       if (!st.hasMoreTokens() || !"details".equals(st.nextToken())) {
         log.error(
             "Failed to parse NexusOperation Nexus link URL: invalid path: {}", uri.getRawPath());
         return null;
       }
+      if (st.hasMoreTokens()) {
+        log.error(
+            "Failed to parse NexusOperation Nexus link URL: extra tokens after 'details': {}",
+            uri.getRawPath());
+        return null;
+      }
 
       return Link.newBuilder()
           .setNexusOperation(
@@ -234,7 +258,7 @@ public static Link nexusLinkToNexusOperation(io.temporal.api.nexus.v1.Link nexus
                   .setOperationId(operationId)
                   .setRunId(runId))
           .build();
-    } catch (Exception e) {
+    } catch (URISyntaxException | UnsupportedEncodingException e) {
       log.error("Failed to parse NexusOperation Nexus link URL", e);
       return null;
     }

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

@@ -23,15 +23,18 @@ public class InternalNexusOperationContext {
   // SignalWorkflowExecutionRequest.links.
   private List<Link> nexusOperationLinks = Collections.emptyList();
   // Backlinks returned by outbound RPCs the operation handler issues (currently
-  // SignalWorkflowExecutionResponse.link and SignalWithStartWorkflowExecutionResponse.signal_link;
-  // future update/start variants attach the same way). One entry per outbound RPC that returned
-  // a link. Drained by the task handler when building StartOperationResponse so each RPC the
-  // handler issued gets a corresponding link on the caller workflow's history event.
+  // SignalWorkflowExecutionResponse.link and SignalWithStartWorkflowExecutionResponse.signal_link).
+  // One entry per outbound RPC that returned a link. Drained by the task handler when building
+  // StartOperationResponse so each RPC the handler issued 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
-  // RPCs will not see the thread-local context, so the links from those RPCs will not propagate.
+  // This context is only safe for use from the single thread that runs the operation handler (the
+  // Nexus task executor's thread). The mutators below assert this contract; a stray cross-thread
+  // call fails fast rather than silently corrupting the ArrayList.
   private final List<Link> responseBacklinks = new ArrayList<>();
+  // Captured at construction (on the Nexus task executor's thread) and used to fail fast on any
+  // cross-thread mutation. See note on responseBacklinks.
+  private final Thread ownerThread;
 
   public InternalNexusOperationContext(
       String namespace,
@@ -44,6 +47,18 @@ public InternalNexusOperationContext(
     this.endpoint = endpoint;
     this.metricScope = metricScope;
     this.client = client;
+    this.ownerThread = Thread.currentThread();
+  }
+
+  private void assertOwnerThread() {
+    if (Thread.currentThread() != ownerThread) {
+      throw new IllegalStateException(
+          "InternalNexusOperationContext mutated from thread '"
+              + Thread.currentThread().getName()
+              + "' but is owned by '"
+              + ownerThread.getName()
+              + "'. Operation handlers must not spawn threads to issue link-propagating RPCs.");
+    }
   }
 
   public Scope getMetricsScope() {
@@ -90,6 +105,7 @@ public Link getStartWorkflowResponseLink() {
    * to RPCs issued by the operation handler.
    */
   public void setNexusOperationLinks(List<Link> links) {
+    assertOwnerThread();
     this.nexusOperationLinks = links == null ? Collections.emptyList() : links;
   }
 
@@ -99,19 +115,23 @@ public List<Link> getNexusOperationLinks() {
   }
 
   /**
-   * Append a backlink returned by an outbound RPC the operation handler issued (e.g. signal,
-   * signalWithStart, and future update/start variants). The task handler drains the list when
-   * building the operation's StartOperationResponse.
+   * Append a backlink returned by an outbound RPC the operation handler issued (signal or
+   * signalWithStart). The task handler drains the list when building the operation's
+   * StartOperationResponse.
    */
   public void addBacklink(Link link) {
+    assertOwnerThread();
     if (link != null) {
       this.responseBacklinks.add(link);
     }
   }
 
-  /** Backlinks from every outbound RPC the handler issued. Never null; may be empty. */
+  /**
+   * Backlinks from every outbound RPC the handler issued. Never null; may be empty. Returned as an
+   * unmodifiable view; callers must not attempt to mutate.
+   */
   public List<Link> getBacklinks() {
-    return responseBacklinks;
+    return Collections.unmodifiableList(responseBacklinks);
   }
 
   private class NexusOperationContextImpl implements NexusOperationContext {

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

@@ -317,49 +317,51 @@ private StartOperationResponse handleStartOperation(
     StartOperationResponse.Builder startResponseBuilder = StartOperationResponse.newBuilder();
     OperationContext context = ctx.build();
     try {
-      try {
-        OperationStartResult<HandlerResultContent> result =
-            startOperation(context, operationStartDetails.build(), input.build());
-        // If outbound RPCs the handler issued (signal, signalWithStart, future update/start
-        // variants) returned backlinks, propagate them to the caller so the caller workflow's
-        // history event links to each event on the callee. Same set of backlinks applies to both
-        // sync and async response variants.
-        List<io.temporal.api.nexus.v1.Link> backlinks = new ArrayList<>();
-        for (io.temporal.api.common.v1.Link backlink :
-            CurrentNexusOperationContext.get().getBacklinks()) {
-          io.temporal.api.nexus.v1.Link converted = LinkConverter.commonLinkToNexusLink(backlink);
-          if (converted != null) {
-            backlinks.add(converted);
-          }
-        }
-
-        if (result.isSync()) {
-          startResponseBuilder.setSyncSuccess(
-              StartOperationResponse.Sync.newBuilder()
-                  .setPayload(Payload.parseFrom(result.getSyncResult().getDataBytes()))
-                  .addAllLinks(backlinks)
-                  .build());
+      OperationStartResult<HandlerResultContent> result =
+          startOperation(context, operationStartDetails.build(), input.build());
+      // If signal or signalWithStart RPCs the handler issued returned backlinks, propagate them
+      // to the caller so the caller workflow's history event links to each event on the callee.
+      // Same set of backlinks applies to both sync and async response variants.
+      List<io.temporal.api.nexus.v1.Link> backlinks = new ArrayList<>();
+      for (io.temporal.api.common.v1.Link backlink :
+          CurrentNexusOperationContext.get().getBacklinks()) {
+        io.temporal.api.nexus.v1.Link converted = LinkConverter.commonLinkToNexusLink(backlink);
+        if (converted != null) {
+          backlinks.add(converted);
         } else {
-          startResponseBuilder.setAsyncSuccess(
-              StartOperationResponse.Async.newBuilder()
-                  .setOperationId(result.getAsyncOperationToken())
-                  .setOperationToken(result.getAsyncOperationToken())
-                  .addAllLinks(
-                      context.getLinks().stream()
-                          .map(
-                              link ->
-                                  io.temporal.api.nexus.v1.Link.newBuilder()
-                                      .setType(link.getType())
-                                      .setUrl(link.getUri().toString())
-                                      .build())
-                          .collect(Collectors.toList()))
-                  .addAllLinks(backlinks)
-                  .build());
+          // The SDK stashed this backlink itself in RootWorkflowClientInvoker; failing to re-encode
+          // it now means a LinkConverter regression or a malformed link from the server. Either is
+          // an SDK invariant violation worth shouting about (warn is too quiet — the caller's
+          // history event will be missing a link with no other diagnostic).
+          log.error(
+              "SDK-stashed backlink failed to re-encode as nexus.v1.Link; caller history will be"
+                  + " missing a link. backlink={}",
+              backlink);
         }
-      } catch (OperationException e) {
-        throw e;
-      } catch (Throwable failure) {
-        convertKnownFailures(failure);
+      }
+
+      if (result.isSync()) {
+        startResponseBuilder.setSyncSuccess(
+            StartOperationResponse.Sync.newBuilder()
+                .setPayload(Payload.parseFrom(result.getSyncResult().getDataBytes()))
+                .addAllLinks(backlinks)
+                .build());
+      } else {
+        startResponseBuilder.setAsyncSuccess(
+            StartOperationResponse.Async.newBuilder()
+                .setOperationId(result.getAsyncOperationToken())
+                .setOperationToken(result.getAsyncOperationToken())
+                .addAllLinks(
+                    context.getLinks().stream()
+                        .map(
+                            link ->
+                                io.temporal.api.nexus.v1.Link.newBuilder()
+                                    .setType(link.getType())
+                                    .setUrl(link.getUri().toString())
+                                    .build())
+                        .collect(Collectors.toList()))
+                .addAllLinks(backlinks)
+                .build());
       }
     } catch (OperationException e) {
       TemporalFailure temporalFailure;
@@ -377,6 +379,8 @@ private StartOperationResponse handleStartOperation(
             new RuntimeException("Unknown operation state: " + e.getState()));
       }
       startResponseBuilder.setFailure(dataConverter.exceptionToFailure(temporalFailure));
+    } catch (Throwable failure) {
+      convertKnownFailures(failure);
     }
     return startResponseBuilder.build();
   }