Add check for duplicate handler start

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClient.java

@@ -9,14 +9,14 @@
  * Nexus-aware client wrapping {@link WorkflowClient}. Provides methods for interacting with
  * Temporal from within a Nexus operation handler.
  *
- * <p>Obtained via the {@link TemporalOperationHandler.StartFunction} parameter.
+ * <p>Obtained via the {@link TemporalOperationHandler.StartHandler} parameter.
  *
  * <p>Example usage to start a workflow from an operation handler:
  *
  * <pre>{@code
  * @OperationImpl
  * public OperationHandler<TransferInput, TransferResult> startTransfer() {
- *   return TemporalOperationHandler.from((context, client, input) -> {
+ *   return TemporalOperationHandler.create((context, client, input) -> {
  *     return client.startWorkflow(
  *         TransferWorkflow.class,
  *         TransferWorkflow::transfer, input.getFromAccount(), input.getToAccount(),
@@ -27,13 +27,13 @@
  * }
  * }</pre>
  *
- * <p>For advanced use cases, the underlying {@link WorkflowClient} can be accessed via {@link
- * #getWorkflowClient()}. For example, to send a signal and return a synchronous result:
+ * <p>For synchronous operations, use {@link #getWorkflowClient()} directly and return a {@link
+ * TemporalOperationResult#sync} result. For example, to send a signal:
  *
  * <pre>{@code
  * @OperationImpl
  * public OperationHandler<CancelOrderInput, Void> cancelOrder() {
- *   return TemporalOperationHandler.from((context, client, input) -> {
+ *   return TemporalOperationHandler.create((context, client, input) -> {
  *     client.getWorkflowClient()
  *         .newUntypedWorkflowStub("order-" + input.getOrderId())
  *         .signal("requestCancellation", input);

temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java

@@ -1,5 +1,6 @@
 package io.temporal.nexus;
 
+import io.nexusrpc.handler.HandlerException;
 import io.nexusrpc.handler.OperationContext;
 import io.nexusrpc.handler.OperationStartDetails;
 import io.temporal.client.WorkflowClient;
@@ -18,6 +19,7 @@ final class TemporalNexusClientImpl implements TemporalNexusClient {
   private final WorkflowClient client;
   private final OperationContext operationContext;
   private final OperationStartDetails operationStartDetails;
+  private boolean asyncOperationStarted;
 
   TemporalNexusClientImpl(
       WorkflowClient client,
@@ -196,11 +198,21 @@ public <R> TemporalOperationResult<R> startWorkflow(
   }
 
   private <R> TemporalOperationResult<R> invokeAndReturn(WorkflowHandle<R> handle) {
+    if (asyncOperationStarted) {
+      throw new HandlerException(
+          HandlerException.ErrorType.BAD_REQUEST,
+          new IllegalStateException(
+              "Only one async operation can be started per operation handler invocation. "
+                  + "Use getWorkflowClient() for additional workflow interactions."));
+    }
     NexusStartWorkflowResponse response =
         NexusStartWorkflowHelper.startWorkflowAndAttachLinks(
             operationContext,
             operationStartDetails,
             request -> handle.getInvoker().invoke(request));
+    // Set after successful start so that if startWorkflowAndAttachLinks throws,
+    // the handler can retry without being blocked by the guard.
+    asyncOperationStarted = true;
     return TemporalOperationResult.async(response.getOperationToken());
   }
 }

temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java

@@ -19,7 +19,7 @@
  * <pre>{@code
  * @OperationImpl
  * public OperationHandler<TransferInput, TransferResult> startTransfer() {
- *   return TemporalOperationHandler.from((context, client, input) -> {
+ *   return TemporalOperationHandler.create((context, client, input) -> {
  *     return client.startWorkflow(
  *         TransferWorkflow.class,
  *         TransferWorkflow::transfer, input.getFromAccount(), input.getToAccount(),
@@ -41,32 +41,32 @@
 public class TemporalOperationHandler<T, R> implements OperationHandler<T, R> {
 
   /**
-   * Function invoked when a Nexus start operation request is received.
+   * Handler invoked when a Nexus start operation request is received.
    *
    * @param <T> the input type
    * @param <R> the result type
    */
   @FunctionalInterface
-  public interface StartFunction<T, R> {
+  public interface StartHandler<T, R> {
     TemporalOperationResult<R> apply(
         TemporalOperationStartContext context, TemporalNexusClient client, T input);
   }
 
-  private final StartFunction<T, R> startFunction;
+  private final StartHandler<T, R> startHandler;
 
-  protected TemporalOperationHandler(StartFunction<T, R> startFunction) {
-    this.startFunction = startFunction;
+  protected TemporalOperationHandler(StartHandler<T, R> startHandler) {
+    this.startHandler = startHandler;
   }
 
   /**
-   * Creates a {@link TemporalOperationHandler} from a start function. Subclass and override {@link
+   * Creates a {@link TemporalOperationHandler} from a start handler. Subclass and override {@link
    * #cancelWorkflowRun} to customize cancel behavior.
    *
-   * @param startFunction the function to invoke on start operation requests
-   * @return an operation handler backed by the given start function
+   * @param startHandler the handler to invoke on start operation requests
+   * @return an operation handler backed by the given start handler
    */
-  public static <T, R> TemporalOperationHandler<T, R> from(StartFunction<T, R> startFunction) {
-    return new TemporalOperationHandler<>(startFunction);
+  public static <T, R> TemporalOperationHandler<T, R> create(StartHandler<T, R> startHandler) {
+    return new TemporalOperationHandler<>(startHandler);
   }
 
   @Override
@@ -77,7 +77,7 @@ public final OperationStartResult<R> start(
         new TemporalNexusClientImpl(nexusCtx.getWorkflowClient(), ctx, details);
 
     TemporalOperationStartContext startContext = new TemporalOperationStartContext(ctx, details);
-    TemporalOperationResult<R> result = startFunction.apply(startContext, client, input);
+    TemporalOperationResult<R> result = startHandler.apply(startContext, client, input);
 
     if (result.isSync()) {
       return OperationStartResult.newSyncBuilder(result.getSyncResult()).build();

temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationStartContext.java

@@ -7,7 +7,7 @@
 
 /**
  * Context for a Nexus start operation request, passed to {@link
- * TemporalOperationHandler.StartFunction} alongside the {@link TemporalNexusClient} and input.
+ * TemporalOperationHandler.StartHandler} alongside the {@link TemporalNexusClient} and input.
  */
 @Experimental
 public final class TemporalOperationStartContext {

temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerCancelTest.java

@@ -123,7 +123,7 @@ public interface TestNexusCancelService {
   public class TestNexusServiceImpl {
     @OperationImpl
     public OperationHandler<String, Void> operation() {
-      return TemporalOperationHandler.from(
+      return TemporalOperationHandler.create(
           (context, client, input) ->
               client.startWorkflow(
                   WaitForCancelWorkflowInterface.class,

temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerDoubleStartTest.java

@@ -0,0 +1,106 @@
+package io.temporal.workflow.nexus;
+
+import io.nexusrpc.Operation;
+import io.nexusrpc.Service;
+import io.nexusrpc.handler.HandlerException;
+import io.nexusrpc.handler.OperationHandler;
+import io.nexusrpc.handler.OperationImpl;
+import io.nexusrpc.handler.ServiceImpl;
+import io.temporal.client.WorkflowFailedException;
+import io.temporal.client.WorkflowOptions;
+import io.temporal.failure.ApplicationFailure;
+import io.temporal.failure.NexusOperationFailure;
+import io.temporal.nexus.TemporalOperationHandler;
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.*;
+import io.temporal.workflow.shared.TestMultiArgWorkflowFunctions;
+import io.temporal.workflow.shared.TestWorkflows;
+import java.time.Duration;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class GenericHandlerDoubleStartTest {
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder()
+          .setWorkflowTypes(
+              TestNexus.class, TestMultiArgWorkflowFunctions.TestMultiArgWorkflowImpl.class)
+          .setNexusServiceImplementation(new TestNexusServiceImpl())
+          .build();
+
+  private static final String EXPECTED_MESSAGE =
+      "Only one async operation can be started per operation handler "
+          + "invocation. Use getWorkflowClient() for additional workflow interactions.";
+
+  @Test
+  public void doubleStartThrows() {
+    TestWorkflows.TestWorkflow1 workflowStub =
+        testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class);
+
+    WorkflowFailedException e =
+        Assert.assertThrows(
+            WorkflowFailedException.class,
+            () -> workflowStub.execute(testWorkflowRule.getTaskQueue()));
+
+    Assert.assertTrue(e.getCause() instanceof NexusOperationFailure);
+    NexusOperationFailure nexusFailure = (NexusOperationFailure) e.getCause();
+
+    Assert.assertTrue(nexusFailure.getCause() instanceof HandlerException);
+    HandlerException handlerException = (HandlerException) nexusFailure.getCause();
+    Assert.assertEquals("handler error: " + EXPECTED_MESSAGE, handlerException.getMessage());
+
+    Assert.assertTrue(handlerException.getCause() instanceof ApplicationFailure);
+    ApplicationFailure appFailure = (ApplicationFailure) handlerException.getCause();
+    Assert.assertEquals("java.lang.IllegalStateException", appFailure.getType());
+    Assert.assertEquals(EXPECTED_MESSAGE, appFailure.getOriginalMessage());
+  }
+
+  public static class TestNexus implements TestWorkflows.TestWorkflow1 {
+    @Override
+    public String execute(String input) {
+      NexusOperationOptions options =
+          NexusOperationOptions.newBuilder()
+              .setScheduleToCloseTimeout(Duration.ofSeconds(10))
+              .build();
+      NexusServiceOptions serviceOptions =
+          NexusServiceOptions.newBuilder().setOperationOptions(options).build();
+
+      TestNexusServiceDoubleStart serviceStub =
+          Workflow.newNexusServiceStub(TestNexusServiceDoubleStart.class, serviceOptions);
+      return serviceStub.operation("input");
+    }
+  }
+
+  @Service
+  public interface TestNexusServiceDoubleStart {
+    @Operation
+    String operation(String input);
+  }
+
+  @ServiceImpl(service = TestNexusServiceDoubleStart.class)
+  public class TestNexusServiceImpl {
+    @OperationImpl
+    public OperationHandler<String, String> operation() {
+      return TemporalOperationHandler.create(
+          (context, client, input) -> {
+            // First start should succeed
+            client.startWorkflow(
+                TestMultiArgWorkflowFunctions.Test1ArgWorkflowFunc.class,
+                TestMultiArgWorkflowFunctions.Test1ArgWorkflowFunc::func1,
+                input,
+                WorkflowOptions.newBuilder()
+                    .setWorkflowId("double-start-first-" + context.getService())
+                    .build());
+            // Second start should throw
+            return client.startWorkflow(
+                TestMultiArgWorkflowFunctions.Test1ArgWorkflowFunc.class,
+                TestMultiArgWorkflowFunctions.Test1ArgWorkflowFunc::func1,
+                input,
+                WorkflowOptions.newBuilder()
+                    .setWorkflowId("double-start-second-" + context.getService())
+                    .build());
+          });
+    }
+  }
+}

temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerSyncResultTest.java

@@ -57,7 +57,7 @@ public interface TestNexusSyncService {
   public class TestNexusServiceImpl {
     @OperationImpl
     public OperationHandler<String, String> operation() {
-      return TemporalOperationHandler.from(
+      return TemporalOperationHandler.create(
           (context, client, input) -> TemporalOperationResult.sync("sync-" + input));
     }
   }

temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerTypedProcTest.java

@@ -62,7 +62,7 @@ public interface TestNexusServiceProc {
   public class TestNexusServiceImpl {
     @OperationImpl
     public OperationHandler<Integer, Void> operation() {
-      return TemporalOperationHandler.from(
+      return TemporalOperationHandler.create(
           (context, client, input) -> {
             String prefix = "generic-handler-test-proc" + input + "-";
             String workflowId = prefix + context.getService() + "-" + context.getOperation();

temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerTypedStartWorkflowTest.java

@@ -63,7 +63,7 @@ public interface TestNexusServiceGeneric {
   public class TestNexusServiceImpl {
     @OperationImpl
     public OperationHandler<Integer, String> operation() {
-      return TemporalOperationHandler.from(
+      return TemporalOperationHandler.create(
           (context, client, input) -> {
             String prefix = "generic-handler-test-func" + input + "-";
             String workflowId = prefix + context.getService() + "-" + context.getOperation();

temporal-sdk/src/test/java/io/temporal/workflow/nexus/GenericHandlerUntypedStartWorkflowTest.java

@@ -59,7 +59,7 @@ public interface TestNexusServiceUntyped {
   public class TestNexusServiceImpl {
     @OperationImpl
     public OperationHandler<String, String> operation() {
-      return TemporalOperationHandler.from(
+      return TemporalOperationHandler.create(
           (context, client, input) ->
               client.startWorkflow(
                   "func1",