Delay flagged version replay callback to marker match

Run a constrained experiment for the interleaved update replay bug by changing VersionStateMachine replay timing only for histories with SKIP_YIELD_ON_VERSION set. In that path, getVersion still returns synchronously, but the replay callback is no longer fired at fake RECORD_MARKER command creation and is instead delayed until the real MARKER_RECORDED event is matched.

The goal of the experiment was to verify that flagged histories do not depend on the current early replay callback or its extra eventLoop scheduling. The legacy interleaved update repro history does not have SKIP_YIELD_ON_VERSION, so it continues to fail unchanged and serves as the control case.

Verified with:
./gradlew --offline :temporal-sdk:test --tests "io.temporal.workflow.versionTests.GetVersionMultithreadingRemoveTest" --tests "io.temporal.workflow.versionTests.GetVersionRemovedInReplayTest" --tests "io.temporal.workflow.versionTests.GetVersionWithoutCommandEventTest" --tests "io.temporal.workflow.versionTests.GetVersionAndTimerTest" --tests "io.temporal.workflow.versionTests.GetVersionMultipleCallsTest" --tests "io.temporal.workflow.versionTests.GetVersionInSignalTest" --tests "io.temporal.workflow.versionTests.GetVersionMultithreadingTest" --tests "io.temporal.workflow.versionTests.GetVersionInterleavedUpdateReplayTest" --tests "io.temporal.internal.replay.GetVersionInterleavedUpdateReplayTaskHandlerTest"
./gradlew --offline :temporal-sdk:test --tests "io.temporal.workflow.versionTests.GetVersionRemovedInReplayTest" --tests "io.temporal.workflow.versionTests.GetVersionMultithreadingRemoveTest" --tests "io.temporal.workflow.versionTests.GetVersionMultipleCallsTest" --tests "io.temporal.workflow.versionTests.GetVersionMultithreadingTest"

Author: eamsden

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

@@ -24,6 +24,7 @@ final class VersionStateMachine {
 
   private final String changeId;
   private final Functions.Func<Boolean> replaying;
+  private final Functions.Func<Boolean> notifyOnMarkerRecordedReplaying;
   private final Functions.Proc1<CancellableCommand> commandSink;
   private final Functions.Proc1<StateMachine> stateMachineSink;
 
@@ -264,6 +265,11 @@ void notifySkippedExecuting() {
     }
 
     void notifyMarkerCreatedReplaying() {
+      if (notifyOnMarkerRecordedReplaying.apply()) {
+        // Flagged histories already get the version synchronously from getVersion(), so delay the
+        // replay callback until the real marker event is matched.
+        return;
+      }
       try {
         // it's a replay and the version to return from the getVersion call should be preloaded from
         // the history
@@ -295,6 +301,14 @@ void flushPreloadedVersionAndUpdateFromEventReplaying() {
       Preconditions.checkState(
           preloadedVersion != null, "preloadedVersion is expected to be initialized");
       flushPreloadedVersionAndUpdateFromEvent(currentEvent);
+      if (notifyOnMarkerRecordedReplaying.apply()) {
+        try {
+          validateVersionAndThrow(false);
+          notifyFromVersion(false);
+        } catch (RuntimeException ex) {
+          notifyFromException(ex);
+        }
+      }
     }
 
     void notifySkippedReplaying() {
@@ -366,18 +380,22 @@ void flushPreloadedVersionAndUpdateFromEvent(HistoryEvent event) {
   public static VersionStateMachine newInstance(
       String id,
       Functions.Func<Boolean> replaying,
+      Functions.Func<Boolean> notifyOnMarkerRecordedReplaying,
       Functions.Proc1<CancellableCommand> commandSink,
       Functions.Proc1<StateMachine> stateMachineSink) {
-    return new VersionStateMachine(id, replaying, commandSink, stateMachineSink);
+    return new VersionStateMachine(
+        id, replaying, notifyOnMarkerRecordedReplaying, commandSink, stateMachineSink);
   }
 
   private VersionStateMachine(
       String changeId,
       Functions.Func<Boolean> replaying,
+      Functions.Func<Boolean> notifyOnMarkerRecordedReplaying,
       Functions.Proc1<CancellableCommand> commandSink,
       Functions.Proc1<StateMachine> stateMachineSink) {
     this.changeId = Objects.requireNonNull(changeId);
     this.replaying = Objects.requireNonNull(replaying);
+    this.notifyOnMarkerRecordedReplaying = Objects.requireNonNull(notifyOnMarkerRecordedReplaying);
     this.commandSink = Objects.requireNonNull(commandSink);
     this.stateMachineSink = stateMachineSink;
   }

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

@@ -647,7 +647,11 @@ private void preloadVersionMarker(HistoryEvent event) {
               changeId,
               (idKey) ->
                   VersionStateMachine.newInstance(
-                      changeId, this::isReplaying, commandSink, stateMachineSink));
+                      changeId,
+                      this::isReplaying,
+                      () -> checkSdkFlag(SdkFlag.SKIP_YIELD_ON_VERSION),
+                      commandSink,
+                      stateMachineSink));
       Integer version = versionStateMachine.handleMarkersPreload(event);
       if (versionStateMachine.isWriteVersionChangeSA()) {
         changeVersions.put(changeId, version);
@@ -1235,7 +1239,11 @@ public Integer getVersion(
             changeId,
             (idKey) ->
                 VersionStateMachine.newInstance(
-                    changeId, this::isReplaying, commandSink, stateMachineSink));
+                    changeId,
+                    this::isReplaying,
+                    () -> checkSdkFlag(SdkFlag.SKIP_YIELD_ON_VERSION),
+                    commandSink,
+                    stateMachineSink));
     return stateMachine.getVersion(
         minSupported,
         maxSupported,