Fix premature triggering of eventLoop in case of activity cancellation (#1691)

Issue #1558

Author: Spikhalskiy

build.gradle

@@ -33,12 +33,17 @@ ext {
     grpcVersion = '1.53.0' // [1.38.0,) Needed for io.grpc.protobuf.services.HealthStatusManager
     jacksonVersion = '2.14.2' // [2.9.0,)
     // we don't upgrade to 1.10.x because it requires kotlin 1.6. Users may use 1.10.x in their environments though.
-    micrometerVersion = project.hasProperty("edgeDepsTest") ? '1.10.3' : '1.9.7' // [1.0.0,)
+    micrometerVersion = project.hasProperty("edgeDepsTest") ? '1.10.5' : '1.9.9' // [1.0.0,)
 
     // stay on 1.x for a while to don't use any APIs from 2.x which may break our users which still stay on 1.x
     // also slf4j 2.x is not compatible with spring boot 2.x
     slf4jVersion = project.hasProperty("edgeDepsTest") ? '2.0.6' : '1.7.36' // [1.4.0,)
-    protoVersion = '3.22.0' // [3.12.0,) 3.12 is brought by min gRPC 1.38
+    // [3.12.0,)
+    // 3.12 is brought by min gRPC 1.38.
+    // We can't move pass 3.22.0 because 3.22.2 deprecates some methods used by generated code produced by
+    // the old protoc we keep for compatibility of our generated code with old protobuf-java versions.
+    // Which leads to build failure because of -Werror.
+    protoVersion = '3.22.0'
     annotationApiVersion = '1.3.2'
     guavaVersion = '31.1-jre' // [10.0,)
     tallyVersion = '0.13.0' // [0.4.0,)
@@ -51,12 +56,12 @@ ext {
 
     // Spring Boot 3 requires Java 17, java-sdk builds against 2.x version because we support Java 8.
     // We do test compatibility with Spring Boot 3 in integration tests.
-    springBootVersion = project.hasProperty("edgeDepsTest") ? '3.0.2' : '2.7.8'// [2.4.0,)
+    springBootVersion = project.hasProperty("edgeDepsTest") ? '3.0.4' : '2.7.9'// [2.4.0,)
 
     // test scoped
     // we don't upgrade to 1.3 and 1.4 because they require slf4j 2.x
     logbackVersion = project.hasProperty("edgeDepsTest") ? '1.3.5' : '1.2.11'
-    mockitoVersion = '5.1.1'
+    mockitoVersion = '5.2.0'
     junitVersion = '4.13.2'
 }
 

temporal-sdk/src/main/java/io/temporal/internal/statemachines/ActivityStateMachine.java

@@ -37,6 +37,7 @@
 import io.temporal.api.history.v1.ActivityTaskTimedOutEventAttributes;
 import io.temporal.workflow.Functions;
 import java.util.Optional;
+import javax.annotation.Nonnull;
 
 final class ActivityStateMachine
     extends EntityStateMachineInitialCommand<
@@ -54,7 +55,7 @@ final class ActivityStateMachine
   private final ActivityType activityType;
   private final ActivityCancellationType cancellationType;
 
-  private final Functions.Proc2<Optional<Payloads>, Failure> completionCallback;
+  private final Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback;
 
   private ExecuteActivityParameters parameters;
 
@@ -107,7 +108,7 @@ enum State {
                   State.SCHEDULE_COMMAND_CREATED,
                   ExplicitEvent.CANCEL,
                   State.CANCELED,
-                  ActivityStateMachine::cancelCommandNotifyCanceled)
+                  ActivityStateMachine::cancelCommandNotifyCanceledImmediately)
               .add(
                   State.SCHEDULED_EVENT_RECORDED,
                   EventType.EVENT_TYPE_ACTIVITY_TASK_STARTED,
@@ -147,7 +148,7 @@ enum State {
                   State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
                   CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK,
                   State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
-                  ActivityStateMachine::notifyCanceledIfTryCancel)
+                  ActivityStateMachine::notifyCanceledIfTryCancelImmediately)
               .add(
                   State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
                   EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED,
@@ -175,7 +176,7 @@ applied to the state machine (as it is done for all command events before
                   State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
                   EventType.EVENT_TYPE_ACTIVITY_TASK_CANCELED,
                   State.CANCELED,
-                  ActivityStateMachine::notifyCanceled)
+                  ActivityStateMachine::notifyCanceledFromEvent)
               .add(
                   State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
                   EventType.EVENT_TYPE_ACTIVITY_TASK_STARTED,
@@ -193,7 +194,7 @@ applied to the state machine (as it is done for all command events before
                   State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
                   EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED,
                   State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
-                  ActivityStateMachine::notifyCanceledIfTryCancel)
+                  ActivityStateMachine::notifyCanceledIfTryCancelFromEvent)
               /*
               These state transitions are not possible.
               It looks like it is valid when an event, handling of which requests activity
@@ -247,15 +248,15 @@ applied to the state machine (as it is done for all command events before
    */
   public static ActivityStateMachine newInstance(
       ExecuteActivityParameters parameters,
-      Functions.Proc2<Optional<Payloads>, Failure> completionCallback,
+      Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback,
       Functions.Proc1<CancellableCommand> commandSink,
       Functions.Proc1<StateMachine> stateMachineSink) {
     return new ActivityStateMachine(parameters, completionCallback, commandSink, stateMachineSink);
   }
 
   private ActivityStateMachine(
       ExecuteActivityParameters parameters,
-      Functions.Proc2<Optional<Payloads>, Failure> completionCallback,
+      Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback,
       Functions.Proc1<CancellableCommand> commandSink,
       Functions.Proc1<StateMachine> stateMachineSink) {
     super(STATE_MACHINE_DEFINITION, commandSink, stateMachineSink);
@@ -276,32 +277,76 @@ public void createScheduleActivityTaskCommand() {
             .build());
   }
 
+  private void setStartedCommandEventId() {
+    startedCommandEventId = currentEvent.getEventId();
+  }
+
   public void cancel() {
     if (cancellationType == ActivityCancellationType.ABANDON) {
-      notifyCanceled();
+      notifyCanceled(false);
     } else if (!isFinalState()) {
       explicitEvent(ExplicitEvent.CANCEL);
     }
   }
 
-  private void setStartedCommandEventId() {
-    startedCommandEventId = currentEvent.getEventId();
-  }
+  // *Immediately versions don't wait for a matching command, underlying callback will trigger an
+  // event loop so the workflow can make progress, because promise gets filled.
 
-  private void cancelCommandNotifyCanceled() {
+  /**
+   * {@link CommandType#COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK} command is not yet left to the server.
+   * Cancel it in place and immediately notify the workflow code.
+   */
+  private void cancelCommandNotifyCanceledImmediately() {
     cancelCommand();
+    // TODO With {@link ActivityCancellationType#ABANDON} we shouldn't even get here as it gets
+    //  handled in #cancel.
+    //  It's a code path for TRY_CANCEL and WAIT_CANCELLATION_COMPLETED only.
+    //  Was the original design to cancel a not-yet-sent
+    //  {@link CommandType#COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK} in case of ABANDON too?
     if (cancellationType != ActivityCancellationType.ABANDON) {
-      notifyCanceled();
+      notifyCanceled(false);
+    }
+  }
+
+  /**
+   * Workflow code doesn't need to wait for the cancellation event if {@link
+   * ActivityCancellationType#TRY_CANCEL}, immediately notify the workflow code.
+   */
+  private void notifyCanceledIfTryCancelImmediately() {
+    if (cancellationType == ActivityCancellationType.TRY_CANCEL) {
+      notifyCanceled(false);
     }
   }
 
-  private void notifyCanceledIfTryCancel() {
+  // *FromEvent versions will not trigger event loop as they need to wait for all events to be
+  // applied before and there will be WorkflowTaskStarted to trigger the event loop.
+
+  /**
+   * if {@link EventType#EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED} is observed, notify the workflow
+   * code if {@link ActivityCancellationType#TRY_CANCEL}, this mode doesn't need a confirmation of
+   * cancellation.
+   */
+  private void notifyCanceledIfTryCancelFromEvent() {
     if (cancellationType == ActivityCancellationType.TRY_CANCEL) {
-      notifyCanceled();
+      notifyCanceled(true);
     }
   }
 
-  private void notifyCanceled() {
+  /**
+   * Notify workflow code of the cancellation from the {@link
+   * EventType#EVENT_TYPE_ACTIVITY_TASK_CANCELED} event.
+   *
+   * <p>There is no harm in notifying {@link ActivityCancellationType#TRY_CANCEL} again, but it
+   * should not be needed as it should be already done by {@link
+   * #notifyCanceledIfTryCancelFromEvent} as there should be no {@link
+   * EventType#EVENT_TYPE_ACTIVITY_TASK_CANCELED} without {@link
+   * EventType#EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED}.
+   */
+  private void notifyCanceledFromEvent() {
+    notifyCanceled(true);
+  }
+
+  private void notifyCanceled(boolean fromEvent) {
     Failure canceledFailure =
         Failure.newBuilder()
             .setSource(JAVA_SDK)
@@ -321,7 +366,7 @@ private void notifyCanceled() {
             .setCause(canceledFailure)
             .setMessage(ACTIVITY_CANCELED_MESSAGE)
             .build();
-    completionCallback.apply(Optional.empty(), failure);
+    completionCallback.apply(Optional.empty(), new FailureResult(failure, fromEvent));
   }
 
   private void notifyCompleted() {
@@ -349,7 +394,7 @@ private void notifyFailed() {
             .setCause(failed.getFailure())
             .setMessage(ACTIVITY_FAILED_MESSAGE)
             .build();
-    completionCallback.apply(Optional.empty(), failure);
+    completionCallback.apply(Optional.empty(), new FailureResult(failure, true));
   }
 
   private void notifyTimedOut() {
@@ -370,7 +415,7 @@ private void notifyTimedOut() {
             .setCause(timedOut.getFailure())
             .setMessage(ACTIVITY_TIMED_OUT_MESSAGE)
             .build();
-    completionCallback.apply(Optional.empty(), failure);
+    completionCallback.apply(Optional.empty(), new FailureResult(failure, true));
   }
 
   private void notifyCancellationFromEvent() {
@@ -398,7 +443,7 @@ private void notifyCancellationFromEvent() {
               .setMessage(ACTIVITY_CANCELED_MESSAGE)
               .build();
 
-      completionCallback.apply(Optional.empty(), failure);
+      completionCallback.apply(Optional.empty(), new FailureResult(failure, true));
     }
   }
 
@@ -412,4 +457,27 @@ private void createRequestCancelActivityTaskCommand() {
             .build());
     parameters = null; // avoid retaining large input for the duration of the activity
   }
+
+  public static class FailureResult {
+    private final @Nonnull Failure failure;
+    private final boolean fromEvent;
+
+    public FailureResult(@Nonnull Failure failure, boolean fromEvent) {
+      this.failure = failure;
+      this.fromEvent = fromEvent;
+    }
+
+    @Nonnull
+    public Failure getFailure() {
+      return failure;
+    }
+
+    /**
+     * @return true if this failure is created from the event during event <-> command matching
+     *     phase.
+     */
+    public boolean isFromEvent() {
+      return fromEvent;
+    }
+  }
 }

temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java

@@ -545,8 +545,16 @@ public Functions.Proc scheduleActivityTask(
         ActivityStateMachine.newInstance(
             attributes,
             (p, f) -> {
-              callback.apply(p, f);
-              if (f != null && f.hasCause() && f.getCause().hasCanceledFailureInfo()) {
+              Failure failure = f != null ? f.getFailure() : null;
+              callback.apply(p, failure);
+
+              if (f != null
+                  && !f.isFromEvent()
+                  && failure.hasCause()
+                  && failure.getCause().hasCanceledFailureInfo()) {
+                // If !f.isFromEvent(), we want to unblock the event loop as the promise got filled
+                // and the workflow may make progress. If f.isFromEvent(), we need to delay event
+                // loop triggering until WorkflowTaskStarted.
                 eventLoop();
               }
             },

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

@@ -663,7 +663,7 @@ public static Random newRandom() {
    *
    * <p>This method always returns false if called from a non workflow thread.
    *
-   * @deprecated use {{@link WorkflowUnsafe#isReplaying()}}
+   * @deprecated use {@link WorkflowUnsafe#isReplaying()}
    */
   @Deprecated
   public static boolean isReplaying() {