Fix time skipping for dotnet SDK

Author: Quinn-With-Two-Ns

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

@@ -107,6 +107,8 @@ void startNexusOperation(
 
   void cancelNexusOperationRequestAcknowledge(NexusOperationRef ref);
 
+  void failNexusOperationCancelRequest(NexusOperationRef ref, Failure failure);
+
   void completeNexusOperation(NexusOperationRef ref, Payload result);
 
   void completeAsyncNexusOperation(

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

@@ -105,6 +105,10 @@ private interface UpdateProcedure {
   private final Map<String, Long> activityById = new HashMap<>();
   private final Map<Long, StateMachine<ChildWorkflowData>> childWorkflows = new HashMap<>();
   private final Map<Long, StateMachine<NexusOperationData>> nexusOperations = new HashMap<>();
+  // Tracks cancelRequestedEventId by scheduledEventId, persists after operation removal.
+  private final Map<Long, Long> nexusCancelRequestedEventIds = new HashMap<>();
+  // Tracks scheduledEventIds of nexus cancel requests that have not yet received a response.
+  private final Set<Long> unresolvedNexusCancelRequests = new HashSet<>();
   private final Map<String, StateMachine<TimerData>> timers = new HashMap<>();
   private final Map<String, StateMachine<SignalExternalData>> externalSignals = new HashMap<>();
   private final Map<String, StateMachine<CancelExternalData>> externalCancellations =
@@ -486,10 +490,13 @@ public void completeWorkflowTask(
                 .asRuntimeException();
           }
 
-          if (unhandledCommand(request) || unhandledMessages(request)) {
+          if (unhandledCommand(request)
+              || unhandledMessages(request)
+              || hasUnresolvedNexusCancelWithCompletion(request)) {
             // Fail the workflow task if there are new events or messages and a command tries to
-            // complete the workflow. Record the failure in history, then throw an error to the
-            // caller (matching real server behavior).
+            // complete the workflow, or if there are unresolved nexus cancel requests. Record the
+            // failure in history, then throw an error to the caller (matching real server
+            // behavior).
             failWorkflowTaskWithAReason(
                 WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_UNHANDLED_COMMAND,
                 null,
@@ -673,6 +680,12 @@ private boolean unhandledMessages(RespondWorkflowTaskCompletedRequest request) {
         && hasCompletionCommand(request.getCommandsList()));
   }
 
+  private boolean hasUnresolvedNexusCancelWithCompletion(
+      RespondWorkflowTaskCompletedRequest request) {
+    return !unresolvedNexusCancelRequests.isEmpty()
+        && hasCompletionCommand(request.getCommandsList());
+  }
+
   private boolean hasCompletionCommand(List<Command> commands) {
     for (Command command : commands) {
       if (WorkflowExecutionUtils.isWorkflowExecutionCompleteCommand(command)) {
@@ -899,6 +912,12 @@ private void processRequestCancelNexusOperation(
       ctx.setNeedWorkflowTask(true);
     } else {
       operation.action(Action.REQUEST_CANCELLATION, ctx, null, workflowTaskCompletedId);
+      ctx.onCommit(
+          historySize -> {
+            nexusCancelRequestedEventIds.put(
+                scheduleEventId, operation.getData().cancelRequestedEventId);
+            unresolvedNexusCancelRequests.add(scheduleEventId);
+          });
       ctx.addTimer(
           ProtobufTimeUtils.toJavaDuration(operation.getData().requestTimeout),
           () ->
@@ -2339,6 +2358,10 @@ public void startNexusOperation(
     update(
         ctx -> {
           StateMachine<NexusOperationData> operation = getPendingNexusOperation(scheduledEventId);
+          if (operation.getState() == State.STARTED) {
+            // Operation was already started (e.g. from a previous attempt before retry).
+            return;
+          }
           operation.action(StateMachines.Action.START, ctx, resp, 0);
           operation.getData().identity = clientIdentity;
 
@@ -2378,13 +2401,30 @@ public void cancelNexusOperation(NexusOperationRef ref, Failure failure) {
         });
   }
 
+  /**
+   * Resolves the cancelRequestedEventId for a nexus operation, checking both the active operations
+   * map and the persisted cancel request IDs (for operations that have already completed/removed).
+   */
+  private long resolveCancelRequestedEventId(long scheduledEventId) {
+    StateMachine<NexusOperationData> operation = nexusOperations.get(scheduledEventId);
+    if (operation != null) {
+      return operation.getData().cancelRequestedEventId;
+    }
+    Long stored = nexusCancelRequestedEventIds.get(scheduledEventId);
+    return stored != null ? stored : 0;
+  }
+
   @Override
   public void cancelNexusOperationRequestAcknowledge(NexusOperationRef ref) {
     update(
         ctx -> {
           StateMachine<NexusOperationData> operation =
-              getPendingNexusOperation(ref.getScheduledEventId());
-          if (!operationInFlight(operation.getState())) {
+              nexusOperations.get(ref.getScheduledEventId());
+          if (operation != null && !operationInFlight(operation.getState())) {
+            return;
+          }
+          long cancelRequestedEventId = resolveCancelRequestedEventId(ref.getScheduledEventId());
+          if (cancelRequestedEventId == 0) {
             return;
           }
           ctx.addEvent(
@@ -2393,12 +2433,39 @@ public void cancelNexusOperationRequestAcknowledge(NexusOperationRef ref) {
                   .setNexusOperationCancelRequestCompletedEventAttributes(
                       NexusOperationCancelRequestCompletedEventAttributes.newBuilder()
                           .setScheduledEventId(ref.getScheduledEventId())
-                          .setRequestedEventId(operation.getData().cancelRequestedEventId))
+                          .setRequestedEventId(cancelRequestedEventId))
                   .build());
+          ctx.onCommit(
+              historySize -> unresolvedNexusCancelRequests.remove(ref.getScheduledEventId()));
+          scheduleWorkflowTask(ctx);
           ctx.unlockTimer("cancelNexusOperationRequestAcknowledge");
         });
   }
 
+  @Override
+  public void failNexusOperationCancelRequest(NexusOperationRef ref, Failure failure) {
+    update(
+        ctx -> {
+          long cancelRequestedEventId = resolveCancelRequestedEventId(ref.getScheduledEventId());
+          if (cancelRequestedEventId == 0) {
+            return;
+          }
+          ctx.addEvent(
+              HistoryEvent.newBuilder()
+                  .setEventType(EventType.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_FAILED)
+                  .setNexusOperationCancelRequestFailedEventAttributes(
+                      NexusOperationCancelRequestFailedEventAttributes.newBuilder()
+                          .setScheduledEventId(ref.getScheduledEventId())
+                          .setRequestedEventId(cancelRequestedEventId)
+                          .setFailure(failure))
+                  .build());
+          ctx.onCommit(
+              historySize -> unresolvedNexusCancelRequests.remove(ref.getScheduledEventId()));
+          scheduleWorkflowTask(ctx);
+          ctx.unlockTimer("failNexusOperationCancelRequest");
+        });
+  }
+
   @Override
   public void completeNexusOperation(NexusOperationRef ref, Payload result) {
     update(
@@ -2471,6 +2538,8 @@ private void timeoutNexusOperation(
             }
             operation.action(StateMachines.Action.TIME_OUT, ctx, timeoutType, 0);
             nexusOperations.remove(scheduledEventId);
+            // The cancel response won't matter after a timeout, so clear the unresolved cancel.
+            unresolvedNexusCancelRequests.remove(scheduledEventId);
             scheduleWorkflowTask(ctx);
           });
     } catch (StatusRuntimeException e) {
@@ -2496,7 +2565,8 @@ private void timeoutNexusRequest(long scheduledEventId, String requestMethod, in
           ctx -> {
             StateMachine<NexusOperationData> operation = getPendingNexusOperation(scheduledEventId);
             if (attempt != operation.getData().getAttempt()
-                || isTerminalState(operation.getState())) {
+                || isTerminalState(operation.getState())
+                || operation.getState() == State.STARTED) {
               throw Status.NOT_FOUND.withDescription("Timer fired earlier").asRuntimeException();
             }
 
@@ -2510,6 +2580,8 @@ private void timeoutNexusRequest(long scheduledEventId, String requestMethod, in
 
             if (isTerminalState(operation.getState())) {
               nexusOperations.remove(scheduledEventId);
+              // Cancel response won't arrive after terminal state, unblock workflow completion.
+              unresolvedNexusCancelRequests.remove(scheduledEventId);
               scheduleWorkflowTask(ctx);
             } else {
               retryNexusTask(ctx, operation);

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

@@ -1048,9 +1048,30 @@ public void respondNexusTaskCompleted(
           Failure.Builder b = Failure.newBuilder().setMessage(opError.getFailure().getMessage());
 
           if (startResp.getOperationError().getOperationState().equals("canceled")) {
-            b.setCanceledFailureInfo(
-                CanceledFailureInfo.newBuilder()
-                    .setDetails(nexusFailureMetadataToPayloads(opError.getFailure())));
+            b.setCanceledFailureInfo(CanceledFailureInfo.getDefaultInstance());
+            // Parse the nexus failure details as a Failure proto for the cause chain,
+            // without overwriting the message (unlike nexusFailureToAPIFailure).
+            io.temporal.api.nexus.v1.Failure nexusFailure = opError.getFailure();
+            if (nexusFailure.getMetadataMap().containsKey("type")
+                && nexusFailure.getMetadataMap().get("type").equals(FAILURE_TYPE_STRING)) {
+              try {
+                Failure.Builder causeBuilder = Failure.newBuilder();
+                JSON_PARSER.merge(nexusFailure.getDetails().toString(UTF_8), causeBuilder);
+                Failure parsedCause = causeBuilder.build();
+                // If the parsed failure itself has CanceledFailureInfo, unwrap it
+                // since the outer failure already has CanceledFailureInfo set.
+                if (parsedCause.hasCanceledFailureInfo()) {
+                  b.setCanceledFailureInfo(parsedCause.getCanceledFailureInfo());
+                  if (parsedCause.hasCause()) {
+                    b.setCause(parsedCause.getCause());
+                  }
+                } else {
+                  b.setCause(parsedCause);
+                }
+              } catch (InvalidProtocolBufferException e) {
+                throw new RuntimeException(e);
+              }
+            }
             mutableState.cancelNexusOperation(tt.getOperationRef(), b.build());
           } else {
             mutableState.failNexusOperation(
@@ -1096,8 +1117,12 @@ public void respondNexusTaskFailed(
       NexusTaskToken tt = NexusTaskToken.fromBytes(request.getTaskToken());
       TestWorkflowMutableState mutableState =
           getMutableState(tt.getOperationRef().getExecutionId());
-      if (mutableState.validateOperationTaskToken(tt)) {
-        Failure failure = handlerErrorToFailure(request.getError());
+      Failure failure = handlerErrorToFailure(request.getError());
+      if (tt.isCancel()) {
+        // For cancel failures, the operation may already be completed/removed,
+        // so skip token validation and record the event directly.
+        mutableState.failNexusOperationCancelRequest(tt.getOperationRef(), failure);
+      } else if (mutableState.validateOperationTaskToken(tt)) {
         mutableState.failNexusOperation(tt.getOperationRef(), failure);
       }
       responseObserver.onNext(RespondNexusTaskFailedResponse.getDefaultInstance());
@@ -1171,6 +1196,7 @@ public void completeNexusOperation(
 
   private static Failure handlerErrorToFailure(HandlerError err) {
     return Failure.newBuilder()
+        .setMessage(err.getFailure().getMessage())
         .setNexusHandlerFailureInfo(
             NexusHandlerFailureInfo.newBuilder()
                 .setType(err.getErrorType())