Fix skipped mutable side effect replay recovery

CHANGELOG.md

@@ -69,6 +69,9 @@ to docs, or any other relevant information.
   specific to the Temporal CLI dev server implementation and may no longer be valid if that implementation changes.
 
 ### Fixed
+- Skipped `Workflow.mutableSideEffect` calls no longer leave invalid marker commands after cache
+  eviction or a worker restart when the new recovery flag is enabled. The flag remains disabled
+  by default in this release to preserve replay compatibility during SDK rollout.
 - `ActivityClient` now sends the `temporal-namespace` gRPC header like `WorkflowClient` and `ScheduleClient` do,
   so standalone activity calls are no longer rejected with `PERMISSION_DENIED` by namespace-scoped auth (e.g.
   Temporal Cloud API keys).

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

@@ -50,6 +50,22 @@ public enum SdkFlag {
    * Bug: https://github.com/temporalio/sdk-java/issues/2989
    */
   SCHEDULE_ASYNC_STUB_OPERATIONS(6),
+  /*
+   * Resolves skipped mutable-side-effect invocations after all command events in their replayed
+   * workflow task have been applied. An invocation still waiting for a marker at that point
+   * cancels its replay placeholder and returns the cached value. This lets its callback run at
+   * that task's workflow time, before the next task advances time or delivers signals.
+   *
+   * Also resolves pending replay placeholders at a live-task boundary, preventing empty
+   * RecordMarker commands from being sent to the server when no command event triggered cleanup.
+   *
+   * Introduced: 1.41.0
+   *
+   * Enabled: (pending)
+   *
+   * Bug: https://github.com/temporalio/sdk-java/issues/3163
+   */
+  MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY(7),
   UNKNOWN(Integer.MAX_VALUE);
 
   private final int value;

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

@@ -27,6 +27,7 @@ final class MutableSideEffectStateMachine {
   private final String id;
   private UserMetadata metadata;
   private final Functions.Func<Boolean> replaying;
+  private final Functions.Func<Boolean> useReplayRecovery;
   private final Functions.Proc1<CancellableCommand> commandSink;
 
   private Optional<Payloads> result = Optional.empty();
@@ -130,6 +131,21 @@ State getExecutionState() {
       return replaying.apply() ? State.REPLAYING : State.EXECUTING;
     }
 
+    @Override
+    public void handleWorkflowTaskStarted() {
+      // Recover only when the flag is already enabled. Unflagged tasks retain the callback timing
+      // of the SDK that wrote them.
+      if (getState() == State.RESULT_NOTIFIED_REPLAYING && useReplayRecovery.apply()) {
+        notifySkippedReplaying();
+      }
+    }
+
+    void notifySkippedReplaying() {
+      if (getState() == State.RESULT_NOTIFIED_REPLAYING) {
+        explicitEvent(ExplicitEvent.NON_MATCHING_EVENT);
+      }
+    }
+
     @Override
     public WorkflowStateMachines.HandleEventStatus handleEvent(
         HistoryEvent event, boolean hasNextEvent) {
@@ -249,30 +265,34 @@ public static MutableSideEffectStateMachine newInstance(
       String id,
       UserMetadata metadata,
       Functions.Func<Boolean> replaying,
+      Functions.Func<Boolean> useReplayRecovery,
       Functions.Proc1<CancellableCommand> commandSink,
       Functions.Proc1<StateMachine> stateMachineSink) {
     return new MutableSideEffectStateMachine(
-        id, metadata, replaying, commandSink, stateMachineSink);
+        id, metadata, replaying, useReplayRecovery, commandSink, stateMachineSink);
   }
 
   private MutableSideEffectStateMachine(
       String id,
       UserMetadata metadata,
       Functions.Func<Boolean> replaying,
+      Functions.Func<Boolean> useReplayRecovery,
       Functions.Proc1<CancellableCommand> commandSink,
       Functions.Proc1<StateMachine> stateMachineSink) {
     this.id = Objects.requireNonNull(id);
     this.metadata = metadata;
     this.replaying = Objects.requireNonNull(replaying);
+    this.useReplayRecovery = Objects.requireNonNull(useReplayRecovery);
     this.commandSink = Objects.requireNonNull(commandSink);
   }
 
-  public void mutableSideEffect(
+  public InvocationStateMachine mutableSideEffect(
       Functions.Func1<Optional<Payloads>, Optional<Payloads>> func,
       Functions.Proc1<Optional<Payloads>> callback,
       Functions.Proc1<StateMachine> stateMachineSink) {
     InvocationStateMachine ism = new InvocationStateMachine(func, callback, stateMachineSink);
     ism.explicitEvent(ExplicitEvent.CHECK_EXECUTION_STATE);
     ism.explicitEvent(ExplicitEvent.SCHEDULE);
+    return ism;
   }
 }

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

@@ -163,6 +163,9 @@ enum HandleEventStatus {
   /** Key is mutable side effect id */
   private final Map<String, MutableSideEffectStateMachine> mutableSideEffects = new HashMap<>();
 
+  private final Queue<MutableSideEffectStateMachine.InvocationStateMachine>
+      replayingMutableSideEffects = new ArrayDeque<>();
+
   /** Key is changeId */
   private final Map<String, VersionStateMachine> versions = new HashMap<>();
 
@@ -370,6 +373,27 @@ private void handleEventsBatch(WFTBuffer.EventBatch eventBatch, boolean hasNextB
         handleSingleMessage(msg);
       }
     }
+
+    if (eventBatch.getWorkflowTaskCompletedEvent().isPresent()) {
+      completeMutableSideEffects();
+    }
+  }
+
+  private void completeMutableSideEffects() {
+    // Only mutable-side-effect invocations participate. A callback can create another invocation;
+    // process it in invocation order, at this task's time, before applying the next task.
+    while (!replayingMutableSideEffects.isEmpty()) {
+      MutableSideEffectStateMachine.InvocationStateMachine invocation =
+          replayingMutableSideEffects.peek();
+      if (!invocation.isFinalState()
+          && !flags.checkSdkFlag(SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY)) {
+        // Preserve the timing of an unflagged invocation until normal command matching or a live
+        // task recovers it and records the recovery flag.
+        return;
+      }
+      replayingMutableSideEffects.poll();
+      invocation.notifySkippedReplaying();
+    }
   }
 
   /** Handle an event when looking ahead at history during replay */
@@ -1225,20 +1249,35 @@ public void mutableSideEffect(
       Functions.Func1<Optional<Payloads>, Optional<Payloads>> func,
       Functions.Proc1<Optional<Payloads>> callback) {
     checkEventLoopExecuting();
+    // Opt-in only for this flag's introductory release. Register its use at the API call, rather
+    // than waiting for a skipped invocation to need cleanup.
+    if (initialFlags.contains(SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY)) {
+      flags.tryUseSdkFlag(SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY);
+    }
     MutableSideEffectStateMachine stateMachine =
         mutableSideEffects.computeIfAbsent(
             id,
             (idKey) ->
                 MutableSideEffectStateMachine.newInstance(
-                    idKey, userMetadata, this::isReplaying, commandSink, stateMachineSink));
-    stateMachine.mutableSideEffect(
-        func,
-        (r) -> {
-          callback.apply(r);
-          // callback unblocked mutableSideEffect call. Give workflow code chance to make progress.
-          eventLoop();
-        },
-        stateMachineSink);
+                    idKey,
+                    userMetadata,
+                    this::isReplaying,
+                    () -> flags.checkSdkFlag(SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY),
+                    commandSink,
+                    stateMachineSink));
+    MutableSideEffectStateMachine.InvocationStateMachine invocation =
+        stateMachine.mutableSideEffect(
+            func,
+            (r) -> {
+              callback.apply(r);
+              // callback unblocked mutableSideEffect call. Give workflow code chance to make
+              // progress.
+              eventLoop();
+            },
+            stateMachineSink);
+    if (isReplaying()) {
+      replayingMutableSideEffects.add(invocation);
+    }
   }
 
   public Integer getVersion(
@@ -1524,7 +1563,8 @@ public void workflowTaskStarted(
         List<SuggestContinueAsNewReason> suggestContinueAsNewReasons,
         boolean isTargetWorkerDeploymentVersionChanged) {
       setCurrentTimeMillis(currentTimeMillis);
-      for (CancellableCommand cancellableCommand : commands) {
+      // Notifying a skipped call can execute workflow code and append commands.
+      for (CancellableCommand cancellableCommand : new ArrayList<>(commands)) {
         cancellableCommand.handleWorkflowTaskStarted();
       }
       // Give local activities a chance to recreate their requests if they were lost due

temporal-sdk/src/test/java/io/temporal/internal/statemachines/MutableSideEffectStateMachineTest.java

@@ -6,6 +6,7 @@
 import static io.temporal.internal.statemachines.MutableSideEffectStateMachine.MUTABLE_SIDE_EFFECT_MARKER_NAME;
 import static io.temporal.internal.statemachines.TestHistoryBuilder.assertCommand;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -18,9 +19,18 @@
 import io.temporal.api.history.v1.HistoryEvent;
 import io.temporal.api.history.v1.MarkerRecordedEventAttributes;
 import io.temporal.api.history.v1.TimerFiredEventAttributes;
+import io.temporal.api.history.v1.WorkflowExecutionSignaledEventAttributes;
+import io.temporal.api.history.v1.WorkflowTaskCompletedEventAttributes;
+import io.temporal.api.sdk.v1.WorkflowTaskCompletedMetadata;
+import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.DefaultDataConverter;
+import io.temporal.internal.common.SdkFlag;
+import io.temporal.worker.WorkflowImplementationOptions;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Optional;
 import org.junit.AfterClass;
@@ -196,6 +206,179 @@ protected void buildWorkflow(AsyncWorkflowBuilder<Void> builder) {
     }
   }
 
+  @Test
+  public void testSkippedInCommandlessTaskWithRecordedRecoveryFlag() {
+    List<String> results = new ArrayList<>();
+    boolean[] live = {true};
+    class TestListener extends TestEntityManagerListenerBase {
+      @Override
+      protected void buildWorkflow(AsyncWorkflowBuilder<Void> builder) {
+        readValue(builder);
+      }
+
+      @Override
+      protected void signal(HistoryEvent event, AsyncWorkflowBuilder<Void> builder) {
+        if (event.getWorkflowExecutionSignaledEventAttributes().getSignalName().equals("skip")) {
+          readValue(builder);
+        } else {
+          builder.add(v -> stateMachines.completeWorkflow(Optional.empty()));
+        }
+      }
+
+      private void readValue(AsyncWorkflowBuilder<Void> builder) {
+        builder
+            .<Optional<Payloads>>add1(
+                (v, callback) ->
+                    stateMachines.mutableSideEffect(
+                        "id1",
+                        null,
+                        previous -> {
+                          assertTrue(
+                              "Replayed mutable side effect must not execute its function.",
+                              live[0]);
+                          return previous.isPresent()
+                              ? Optional.empty()
+                              : converter.toPayloads("value");
+                        },
+                        callback))
+            .add(
+                result ->
+                    results.add(converter.fromPayloads(0, result, String.class, String.class)));
+      }
+    }
+    TestHistoryBuilder history =
+        new TestHistoryBuilder()
+            .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED)
+            .addWorkflowTaskScheduledAndStarted()
+            .add(
+                EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED,
+                WorkflowTaskCompletedEventAttributes.newBuilder()
+                    .setScheduledEventId(2)
+                    .setStartedEventId(3)
+                    .setSdkMetadata(
+                        WorkflowTaskCompletedMetadata.newBuilder()
+                            .addLangUsedFlags(
+                                SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY.getValue()))
+                    .build())
+            .add(
+                EventType.EVENT_TYPE_MARKER_RECORDED,
+                MarkerRecordedEventAttributes.newBuilder()
+                    .setMarkerName(MUTABLE_SIDE_EFFECT_MARKER_NAME)
+                    .putDetails(MARKER_ID_KEY, converter.toPayloads("id1").get())
+                    .putDetails(MARKER_DATA_KEY, converter.toPayloads("value").get())
+                    .putDetails(MARKER_SKIP_COUNT_KEY, converter.toPayloads(0).get())
+                    .build())
+            .add(
+                EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
+                WorkflowExecutionSignaledEventAttributes.newBuilder().setSignalName("skip").build())
+            .addWorkflowTask()
+            .add(
+                EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
+                WorkflowExecutionSignaledEventAttributes.newBuilder()
+                    .setSignalName("finish")
+                    .build())
+            .addWorkflowTask()
+            .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED);
+
+    stateMachines = newStateMachines(new TestListener());
+    assertCommand(
+        CommandType.COMMAND_TYPE_RECORD_MARKER,
+        history.handleWorkflowTaskTakeCommands(stateMachines, 1));
+    assertEquals(Arrays.asList("value"), results);
+    assertTrue(history.handleWorkflowTaskTakeCommands(stateMachines, 2).isEmpty());
+    assertEquals(Arrays.asList("value", "value"), results);
+    assertTrue(
+        "Reading a recorded flag must not activate a new flag",
+        stateMachines.takeNewSdkFlags().isEmpty());
+
+    // A fresh worker replays the commandless task before executing the final signal.
+    live[0] = false;
+    results.clear();
+    stateMachines = newStateMachines(new TestListener());
+    assertCommand(
+        CommandType.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
+        history.handleWorkflowTaskTakeCommands(stateMachines, 3));
+    assertEquals(Arrays.asList("value", "value"), results);
+
+    // The same history must also replay to completion without emitting commands.
+    results.clear();
+    stateMachines = newStateMachines(new TestListener());
+    assertTrue(history.handleWorkflowTaskTakeCommands(stateMachines).isEmpty());
+    assertEquals(Arrays.asList("value", "value"), results);
+  }
+
+  @Test
+  public void testRecoveryFlagIsRecordedAtMutableSideEffectCall() {
+    class TestListener extends TestEntityManagerListenerBase {
+      @Override
+      protected void buildWorkflow(AsyncWorkflowBuilder<Void> builder) {}
+
+      @Override
+      protected void signal(HistoryEvent event, AsyncWorkflowBuilder<Void> builder) {
+        builder
+            .<Optional<Payloads>>add1(
+                (v, callback) ->
+                    stateMachines.mutableSideEffect(
+                        "id1", null, previous -> converter.toPayloads("value"), callback))
+            .add(stateMachines::completeWorkflow);
+      }
+    }
+    TestHistoryBuilder history =
+        new TestHistoryBuilder()
+            .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED)
+            .addWorkflowTask()
+            .add(
+                EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
+                WorkflowExecutionSignaledEventAttributes.newBuilder().setSignalName("use").build())
+            .addWorkflowTaskScheduledAndStarted()
+            .add(
+                EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED,
+                WorkflowTaskCompletedEventAttributes.newBuilder()
+                    .setScheduledEventId(6)
+                    .setStartedEventId(7)
+                    .setSdkMetadata(
+                        WorkflowTaskCompletedMetadata.newBuilder()
+                            .addLangUsedFlags(
+                                SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY.getValue()))
+                    .build())
+            .add(
+                EventType.EVENT_TYPE_MARKER_RECORDED,
+                MarkerRecordedEventAttributes.newBuilder()
+                    .setMarkerName(MUTABLE_SIDE_EFFECT_MARKER_NAME)
+                    .putDetails(MARKER_ID_KEY, converter.toPayloads("id1").get())
+                    .putDetails(MARKER_DATA_KEY, converter.toPayloads("value").get())
+                    .putDetails(MARKER_SKIP_COUNT_KEY, converter.toPayloads(0).get())
+                    .build())
+            .add(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED);
+    stateMachines =
+        new WorkflowStateMachines(
+            new TestListener(),
+            stateMachineList::add,
+            GetSystemInfoResponse.Capabilities.newBuilder().setSdkMetadata(true).build(),
+            WorkflowImplementationOptions.newBuilder().build());
+    assertTrue(history.handleWorkflowTaskTakeCommands(stateMachines, 1).isEmpty());
+    assertFalse(
+        stateMachines.takeNewSdkFlags().contains(SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY));
+
+    List<SdkFlag> savedInitialFlags = WorkflowStateMachines.initialFlags;
+    List<SdkFlag> enabledFlags = new ArrayList<>(savedInitialFlags);
+    enabledFlags.add(SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY);
+    WorkflowStateMachines.initialFlags = Collections.unmodifiableList(enabledFlags);
+    try {
+      List<Command> commands = history.handleWorkflowTaskTakeCommands(stateMachines, 2);
+      assertEquals(2, commands.size());
+      assertEquals(CommandType.COMMAND_TYPE_RECORD_MARKER, commands.get(0).getCommandType());
+      assertEquals(
+          CommandType.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, commands.get(1).getCommandType());
+      assertEquals(
+          EnumSet.of(SdkFlag.MUTABLE_SIDE_EFFECT_REPLAY_RECOVERY), stateMachines.takeNewSdkFlags());
+    } finally {
+      WorkflowStateMachines.initialFlags = savedInitialFlags;
+    }
+    stateMachines = newStateMachines(new TestListener());
+    assertTrue(history.handleWorkflowTaskTakeCommands(stateMachines).isEmpty());
+  }
+
   @Test
   public void testRecordAcrossMultipleWorkflowTasks() {
     class TestListener extends TestEntityManagerListenerBase {