Nexus caller timeouts

Author: bergundy

temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java

@@ -799,6 +799,10 @@ public <R> ExecuteNexusOperationOutput<R> executeNexusOperation(
     input.getHeaders().forEach((k, v) -> attributes.putNexusHeader(k.toLowerCase(), v));
     attributes.setScheduleToCloseTimeout(
         ProtobufTimeUtils.toProtoDuration(input.getOptions().getScheduleToCloseTimeout()));
+    attributes.setScheduleToStartTimeout(
+        ProtobufTimeUtils.toProtoDuration(input.getOptions().getScheduleToStartTimeout()));
+    attributes.setStartToCloseTimeout(
+        ProtobufTimeUtils.toProtoDuration(input.getOptions().getStartToCloseTimeout()));
 
     @Nullable
     UserMetadata userMetadata =

temporal-sdk/src/main/java/io/temporal/workflow/NexusOperationOptions.java

@@ -31,6 +31,8 @@ public static NexusOperationOptions getDefaultInstance() {
 
   public static final class Builder {
     private Duration scheduleToCloseTimeout;
+    private Duration scheduleToStartTimeout;
+    private Duration startToCloseTimeout;
     private NexusOperationCancellationType cancellationType;
     private String summary;
 
@@ -46,6 +48,45 @@ public NexusOperationOptions.Builder setScheduleToCloseTimeout(
       return this;
     }
 
+    /**
+     * Sets the schedule to start timeout for the Nexus operation.
+     *
+     * <p>Maximum time to wait for the operation to be started (or completed if synchronous) by the
+     * handler. If the operation is not started within this timeout, it will fail with
+     * TIMEOUT_TYPE_SCHEDULE_TO_START.
+     *
+     * <p>Requires Temporal Server 1.31.0 or later.
+     *
+     * @param scheduleToStartTimeout the schedule to start timeout for the Nexus operation
+     * @return this
+     */
+    @Experimental
+    public NexusOperationOptions.Builder setScheduleToStartTimeout(
+        Duration scheduleToStartTimeout) {
+      this.scheduleToStartTimeout = scheduleToStartTimeout;
+      return this;
+    }
+
+    /**
+     * Sets the start to close timeout for the Nexus operation.
+     *
+     * <p>Maximum time to wait for an asynchronous operation to complete after it has been started.
+     * If the operation does not complete within this timeout after starting, it will fail with
+     * TIMEOUT_TYPE_START_TO_CLOSE.
+     *
+     * <p>Only applies to asynchronous operations. Synchronous operations ignore this timeout.
+     *
+     * <p>Requires Temporal Server 1.31.0 or later.
+     *
+     * @param startToCloseTimeout the start to close timeout for the Nexus operation
+     * @return this
+     */
+    @Experimental
+    public NexusOperationOptions.Builder setStartToCloseTimeout(Duration startToCloseTimeout) {
+      this.startToCloseTimeout = startToCloseTimeout;
+      return this;
+    }
+
     /**
      * Sets the cancellation type for the Nexus operation. Defaults to WAIT_COMPLETED.
      *
@@ -78,12 +119,19 @@ private Builder(NexusOperationOptions options) {
         return;
       }
       this.scheduleToCloseTimeout = options.getScheduleToCloseTimeout();
+      this.scheduleToStartTimeout = options.getScheduleToStartTimeout();
+      this.startToCloseTimeout = options.getStartToCloseTimeout();
       this.cancellationType = options.getCancellationType();
       this.summary = options.getSummary();
     }
 
     public NexusOperationOptions build() {
-      return new NexusOperationOptions(scheduleToCloseTimeout, cancellationType, summary);
+      return new NexusOperationOptions(
+          scheduleToCloseTimeout,
+          scheduleToStartTimeout,
+          startToCloseTimeout,
+          cancellationType,
+          summary);
     }
 
     public NexusOperationOptions.Builder mergeNexusOperationOptions(
@@ -95,6 +143,14 @@ public NexusOperationOptions.Builder mergeNexusOperationOptions(
           (override.scheduleToCloseTimeout == null)
               ? this.scheduleToCloseTimeout
               : override.scheduleToCloseTimeout;
+      this.scheduleToStartTimeout =
+          (override.scheduleToStartTimeout == null)
+              ? this.scheduleToStartTimeout
+              : override.scheduleToStartTimeout;
+      this.startToCloseTimeout =
+          (override.startToCloseTimeout == null)
+              ? this.startToCloseTimeout
+              : override.startToCloseTimeout;
       this.cancellationType =
           (override.cancellationType == null) ? this.cancellationType : override.cancellationType;
       this.summary = (override.summary == null) ? this.summary : override.summary;
@@ -104,9 +160,13 @@ public NexusOperationOptions.Builder mergeNexusOperationOptions(
 
   private NexusOperationOptions(
       Duration scheduleToCloseTimeout,
+      Duration scheduleToStartTimeout,
+      Duration startToCloseTimeout,
       NexusOperationCancellationType cancellationType,
       String summary) {
     this.scheduleToCloseTimeout = scheduleToCloseTimeout;
+    this.scheduleToStartTimeout = scheduleToStartTimeout;
+    this.startToCloseTimeout = startToCloseTimeout;
     this.cancellationType = cancellationType;
     this.summary = summary;
   }
@@ -116,13 +176,25 @@ public NexusOperationOptions.Builder toBuilder() {
   }
 
   private final Duration scheduleToCloseTimeout;
+  private final Duration scheduleToStartTimeout;
+  private final Duration startToCloseTimeout;
   private final NexusOperationCancellationType cancellationType;
   private final String summary;
 
   public Duration getScheduleToCloseTimeout() {
     return scheduleToCloseTimeout;
   }
 
+  @Experimental
+  public Duration getScheduleToStartTimeout() {
+    return scheduleToStartTimeout;
+  }
+
+  @Experimental
+  public Duration getStartToCloseTimeout() {
+    return startToCloseTimeout;
+  }
+
   public NexusOperationCancellationType getCancellationType() {
     return cancellationType;
   }
@@ -138,20 +210,31 @@ public boolean equals(Object o) {
     if (o == null || getClass() != o.getClass()) return false;
     NexusOperationOptions that = (NexusOperationOptions) o;
     return Objects.equals(scheduleToCloseTimeout, that.scheduleToCloseTimeout)
+        && Objects.equals(scheduleToStartTimeout, that.scheduleToStartTimeout)
+        && Objects.equals(startToCloseTimeout, that.startToCloseTimeout)
         && Objects.equals(cancellationType, that.cancellationType)
         && Objects.equals(summary, that.summary);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(scheduleToCloseTimeout, cancellationType, summary);
+    return Objects.hash(
+        scheduleToCloseTimeout,
+        scheduleToStartTimeout,
+        startToCloseTimeout,
+        cancellationType,
+        summary);
   }
 
   @Override
   public String toString() {
     return "NexusOperationOptions{"
         + "scheduleToCloseTimeout="
         + scheduleToCloseTimeout
+        + ", scheduleToStartTimeout="
+        + scheduleToStartTimeout
+        + ", startToCloseTimeout="
+        + startToCloseTimeout
         + ", cancellationType="
         + cancellationType
         + ", summary='"

temporal-serviceclient/src/main/proto

@@ -668,6 +668,8 @@ private static void scheduleNexusOperation(
             .setOperation(attr.getOperation())
             .setInput(attr.getInput())
             .setScheduleToCloseTimeout(attr.getScheduleToCloseTimeout())
+            .setScheduleToStartTimeout(attr.getScheduleToStartTimeout())
+            .setStartToCloseTimeout(attr.getStartToCloseTimeout())
             .putAllNexusHeader(attr.getNexusHeaderMap())
             .setRequestId(UUID.randomUUID().toString())
             .setWorkflowTaskCompletedEventId(workflowTaskCompletedId);
@@ -778,11 +780,27 @@ private static void completeNexusOperation(
 
   private static void timeoutNexusOperation(
       RequestContext ctx, NexusOperationData data, TimeoutType timeoutType, long notUsed) {
-    if (timeoutType != TimeoutType.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE) {
+    if (timeoutType != TimeoutType.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE
+        && timeoutType != TimeoutType.TIMEOUT_TYPE_SCHEDULE_TO_START
+        && timeoutType != TimeoutType.TIMEOUT_TYPE_START_TO_CLOSE) {
       throw new IllegalArgumentException(
           "Timeout type not supported for Nexus operations: " + timeoutType);
     }
 
+    String timeoutMessage;
+    switch (timeoutType) {
+      case TIMEOUT_TYPE_SCHEDULE_TO_START:
+        timeoutMessage = "operation timed out before starting";
+        break;
+      case TIMEOUT_TYPE_START_TO_CLOSE:
+        timeoutMessage = "operation timed out after starting";
+        break;
+      case TIMEOUT_TYPE_SCHEDULE_TO_CLOSE:
+      default:
+        timeoutMessage = "operation timed out";
+        break;
+    }
+
     Failure failure =
         Failure.newBuilder()
             .setMessage("nexus operation completed unsuccessfully")
@@ -795,7 +813,7 @@ private static void timeoutNexusOperation(
                     .setScheduledEventId(data.scheduledEventId))
             .setCause(
                 Failure.newBuilder()
-                    .setMessage("operation timed out")
+                    .setMessage(timeoutMessage)
                     .setTimeoutFailureInfo(
                         TimeoutFailureInfo.newBuilder().setTimeoutType(timeoutType)))
             .build();

temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachines.java

@@ -852,6 +852,18 @@ private void processScheduleNexusOperation(
                   operation.getData().getAttempt()),
           "NexusOperation ScheduleToCloseTimeout");
     }
+    if (attr.hasScheduleToStartTimeout()
+        && Durations.toMillis(attr.getScheduleToStartTimeout()) > 0) {
+      // ScheduleToStartTimeout is the time from schedule to start (or completion if synchronous)
+      ctx.addTimer(
+          ProtobufTimeUtils.toJavaDuration(attr.getScheduleToStartTimeout()),
+          () ->
+              timeoutNexusOperation(
+                  scheduleEventId,
+                  TimeoutType.TIMEOUT_TYPE_SCHEDULE_TO_START,
+                  operation.getData().getAttempt()),
+          "NexusOperation ScheduleToStartTimeout");
+    }
     ctx.lockTimer("processScheduleNexusOperation");
   }
 
@@ -2309,6 +2321,23 @@ public void startNexusOperation(
           StateMachine<NexusOperationData> operation = getPendingNexusOperation(scheduledEventId);
           operation.action(StateMachines.Action.START, ctx, resp, 0);
           operation.getData().identity = clientIdentity;
+
+          // Add start-to-close timeout timer if configured
+          NexusOperationScheduledEventAttributes scheduledEvent =
+              operation.getData().scheduledEvent;
+          if (scheduledEvent.hasStartToCloseTimeout()
+              && Durations.toMillis(scheduledEvent.getStartToCloseTimeout()) > 0) {
+            // StartToCloseTimeout measures from when the operation started to when it completes
+            ctx.addTimer(
+                ProtobufTimeUtils.toJavaDuration(scheduledEvent.getStartToCloseTimeout()),
+                () ->
+                    timeoutNexusOperation(
+                        scheduledEventId,
+                        TimeoutType.TIMEOUT_TYPE_START_TO_CLOSE,
+                        operation.getData().getAttempt()),
+                "NexusOperation StartToCloseTimeout");
+          }
+
           scheduleWorkflowTask(ctx);
         });
   }

temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java

@@ -2,7 +2,10 @@
 
 import static io.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED;
 import static io.temporal.api.enums.v1.WorkflowExecutionStatus.*;
-import static io.temporal.api.enums.v1.WorkflowIdReusePolicy.*;
+import static io.temporal.api.enums.v1.WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE;
+import static io.temporal.api.enums.v1.WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE_FAILED_ONLY;
+import static io.temporal.api.enums.v1.WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE;
+import static io.temporal.api.enums.v1.WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_UNSPECIFIED;
 import static io.temporal.api.workflowservice.v1.ExecuteMultiOperationRequest.Operation.OperationCase.START_WORKFLOW;
 import static io.temporal.api.workflowservice.v1.ExecuteMultiOperationRequest.Operation.OperationCase.UPDATE_WORKFLOW;
 import static io.temporal.internal.testservice.CronUtils.getBackoffInterval;
@@ -21,6 +24,7 @@
 import io.temporal.api.common.v1.*;
 import io.temporal.api.common.v1.Link;
 import io.temporal.api.enums.v1.*;
+import io.temporal.api.enums.v1.WorkflowIdReusePolicy;
 import io.temporal.api.errordetails.v1.MultiOperationExecutionFailure;
 import io.temporal.api.errordetails.v1.WorkflowExecutionAlreadyStartedFailure;
 import io.temporal.api.failure.v1.*;
@@ -285,6 +289,8 @@ private TestWorkflowMutableState getMutableState(
     return getMutableState(executionId, failNotExists);
   }
 
+  @SuppressWarnings("deprecation") // Backwards compatibility for
+  // WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING
   @Override
   public void startWorkflowExecution(
       StartWorkflowExecutionRequest request,
@@ -310,6 +316,8 @@ public void startWorkflowExecution(
     }
   }
 
+  @SuppressWarnings("deprecation") // Backwards compatibility for
+  // WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING
   StartWorkflowExecutionResponse startWorkflowExecutionImpl(
       StartWorkflowExecutionRequest startRequest,
       Duration backoffStartInterval,
@@ -325,8 +333,9 @@ StartWorkflowExecutionResponse startWorkflowExecutionImpl(
     validateWorkflowIdReusePolicy(reusePolicy, conflictPolicy);
     validateOnConflictOptions(startRequest);
 
-    // Backwards compatibility: WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING is deprecated
-    if (reusePolicy == WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING) {
+    // Backwards compatibility: WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING
+    // is deprecated
+    if (reusePolicy == WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING) {
       conflictPolicy = WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING;
       reusePolicy = WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE;
     }
@@ -475,12 +484,14 @@ private StartWorkflowExecutionResponse throwDuplicatedWorkflow(
         WorkflowExecutionAlreadyStartedFailure.getDescriptor());
   }
 
+  @SuppressWarnings("deprecation") // Backwards compatibility for
+  // WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING
   private void validateWorkflowIdReusePolicy(
       WorkflowIdReusePolicy reusePolicy, WorkflowIdConflictPolicy conflictPolicy) {
     if (conflictPolicy != WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_UNSPECIFIED
-        && reusePolicy == WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING) {
+        && reusePolicy == WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING) {
       throw createInvalidArgument(
-          "Invalid WorkflowIDReusePolicy: WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING cannot be used together with a WorkflowIDConflictPolicy.");
+          "Invalid WorkflowIDReusePolicy: WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING cannot be used together with a WorkflowIDConflictPolicy.");
     }
     if (conflictPolicy == WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING
         && reusePolicy == WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE) {