feat: isSubjectToReplay

Author: patbeqo

temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java

@@ -355,9 +355,11 @@ public Integer getVersion(
             : (min, max) ->
                 WorkflowInternal.readOnly(
                     () ->
-                        preferredVersionProvider.getPreferredVersion(
-                            new PreferredVersionProviderInput(
-                                WorkflowInternal.getWorkflowInfo(), changeId, min, max))),
+                        WorkflowInternal.notSubjectToReplay(
+                            () ->
+                                preferredVersionProvider.getPreferredVersion(
+                                    new PreferredVersionProviderInput(
+                                        WorkflowInternal.getWorkflowInfo(), changeId, min, max)))),
         callback);
   }
 

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

@@ -99,6 +99,7 @@ public Optional<Payloads> handleQuery(
     }
     try {
       replayContext.setReadOnly(true);
+      replayContext.setSubjectToReplay(false);
       queryHandlerWorkflowContext.set(replayContext);
       Object result =
           inboundCallsInterceptor
@@ -107,6 +108,7 @@ public Optional<Payloads> handleQuery(
       return dataConverterWithWorkflowContext.toPayloads(result);
     } finally {
       replayContext.setReadOnly(false);
+      replayContext.setSubjectToReplay(true);
       queryHandlerWorkflowContext.set(null);
     }
   }

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

@@ -158,6 +158,7 @@ public void handleUpdate(
             if (!callbacks.isReplaying()) {
               try {
                 workflowContext.setReadOnly(true);
+                workflowContext.setSubjectToReplay(false);
                 workflowProc.handleValidateUpdate(updateName, updateId, input, eventId, header);
               } catch (ReadOnlyException r) {
                 // Rethrow instead on rejecting the update to fail the WFT
@@ -173,6 +174,7 @@ public void handleUpdate(
                 return;
               } finally {
                 workflowContext.setReadOnly(false);
+                workflowContext.setSubjectToReplay(true);
               }
             }
             callbacks.accept();

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

@@ -102,6 +102,7 @@ final class SyncWorkflowContext implements WorkflowContext, WorkflowOutboundCall
   private NexusServiceOptions defaultNexusServiceOptions = null;
   private Map<String, NexusServiceOptions> nexusServiceOptionsMap;
   private boolean readOnly = false;
+  private boolean subjectToReplay = true;
   private final WorkflowThreadLocal<UpdateInfo> currentUpdateInfo = new WorkflowThreadLocal<>();
   @Nullable private String currentDetails;
 
@@ -1091,10 +1092,12 @@ public <R> R sideEffect(
           () -> {
             try {
               readOnly = true;
+              subjectToReplay = false;
               R r = func.apply();
               return dataConverterWithCurrentWorkflowContext.toPayloads(r);
             } finally {
               readOnly = false;
+              subjectToReplay = true;
             }
           },
           userMetadata,
@@ -1157,6 +1160,7 @@ private <R> R mutableSideEffectImpl(
                           0, Optional.of(b), resultClass, resultType));
           try {
             readOnly = true;
+            subjectToReplay = false;
             R funcResult =
                 Objects.requireNonNull(
                     func.apply(), "mutableSideEffect function " + "returned null");
@@ -1167,6 +1171,7 @@ private <R> R mutableSideEffectImpl(
             return Optional.empty(); // returned only when value doesn't need to be updated
           } finally {
             readOnly = false;
+            subjectToReplay = true;
           }
         },
         (p) ->
@@ -1309,6 +1314,14 @@ void setReadOnly(boolean readOnly) {
     this.readOnly = readOnly;
   }
 
+  boolean isSubjectToReplay() {
+    return subjectToReplay;
+  }
+
+  void setSubjectToReplay(boolean subjectToReplay) {
+    this.subjectToReplay = subjectToReplay;
+  }
+
   @Override
   public Map<Long, SignalHandlerInfo> getRunningSignalHandlers() {
     return signalDispatcher.getRunningSignalHandlers();

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

@@ -735,6 +735,17 @@ public static <T> T readOnly(Functions.Func<T> func) {
     }
   }
 
+  public static <T> T notSubjectToReplay(Functions.Func<T> func) {
+    SyncWorkflowContext workflowContext = getRootWorkflowContext();
+    boolean previousSubjectToReplay = workflowContext.isSubjectToReplay();
+    workflowContext.setSubjectToReplay(false);
+    try {
+      return func.apply();
+    } finally {
+      workflowContext.setSubjectToReplay(previousSubjectToReplay);
+    }
+  }
+
   public static WorkflowInfo getWorkflowInfo() {
     return new WorkflowInfoImpl(getRootWorkflowContext().getReplayContext());
   }
@@ -940,16 +951,20 @@ static SyncWorkflowContext getRootWorkflowContext() {
     return DeterministicRunnerImpl.currentThreadInternal().getWorkflowContext();
   }
 
-  public static boolean isReadOnly() {
+  static boolean isReadOnly() {
     return getRootWorkflowContext().isReadOnly();
   }
 
-  public static void assertNotReadOnly(String action) {
+  static void assertNotReadOnly(String action) {
     if (isReadOnly()) {
       throw new ReadOnlyException(action);
     }
   }
 
+  public static boolean isSubjectToReplay() {
+    return getRootWorkflowContext().isSubjectToReplay();
+  }
+
   static void assertNotInUpdateHandler(String message) {
     if (getCurrentUpdateInfo().isPresent()) {
       throw new UnsupportedContinueAsNewRequest(message);

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

@@ -1010,8 +1010,8 @@ public static Random newRandom() {
    * draws left it. A Workflow Reset replays the same values up to the reset point, then reseeds the
    * stream for the new Run. Each Continue-As-New Run gets a new sequence.
    *
-   * <p>Draws are not recorded in Workflow History, so do not draw in read-only code. Use {@link
-   * WorkflowUnsafe#isReadOnly()} to gate draws.
+   * <p>Draws are not recorded in Workflow History, so only draw where the code is re-executed on
+   * replay. Use {@link WorkflowUnsafe#isSubjectToReplay()} to gate draws.
    *
    * <p>Use a stable package-style name. Stream names are retained for the life of the Workflow Run.
    * The stream is deterministic pseudorandomness and is not cryptographically secure.

temporal-sdk/src/main/java/io/temporal/workflow/unsafe/WorkflowUnsafe.java

@@ -48,18 +48,21 @@ public static boolean isReplaying() {
   }
 
   /**
-   * Reports whether the current code is running where Workflow state cannot be mutated.
+   * Reports whether the currently executing code is re-executed when the Workflow replays.
    *
-   * <p>Read-only code includes Query handlers, Update validators, Side Effect functions, Await
-   * conditions, and other SDK callbacks that must not mutate Workflow state.
+   * <p>Unlike {@link #isReplaying()}, this is a property of the calling context rather than of the
+   * Workflow's current state. The Workflow method and its constructor, signal and update handlers,
+   * and Await conditions are subject to replay. Query handlers, Update validators, and Side Effect
+   * functions run once against the current state and are never re-executed, so they are not subject
+   * to replay even while {@link #isReplaying()} reports true.
    *
    * <p>Must be called from Workflow code.
    *
-   * @return true in a read-only Workflow context
+   * @return true if the calling context is re-executed on replay
    */
   @Experimental
-  public static boolean isReadOnly() {
-    return WorkflowInternal.isReadOnly();
+  public static boolean isSubjectToReplay() {
+    return WorkflowInternal.isSubjectToReplay();
   }
 
   /**

…workflow/WorkflowUnsafeReadOnlyTest.java …w/WorkflowUnsafeSubjectToReplayTest.javatemporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java renamed to temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeSubjectToReplayTest.java temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java renamed to temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeSubjectToReplayTest.java

@@ -9,23 +9,32 @@
 import io.temporal.common.interceptors.WorkflowInboundCallsInterceptorBase;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
 import io.temporal.worker.WorkerFactoryOptions;
+import io.temporal.worker.WorkerOptions;
 import io.temporal.workflow.unsafe.WorkflowUnsafe;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 
-public class WorkflowUnsafeReadOnlyTest {
+public class WorkflowUnsafeSubjectToReplayTest {
   private static final Map<String, Boolean> calls = new ConcurrentHashMap<>();
 
   @Rule
   public SDKTestWorkflowRule testWorkflowRule =
       SDKTestWorkflowRule.newBuilder()
-          .setWorkflowTypes(ReadOnlyWorkflowImpl.class)
+          .setWorkflowTypes(SubjectToReplayWorkflowImpl.class)
           .setWorkerFactoryOptions(
               WorkerFactoryOptions.newBuilder()
-                  .setWorkerInterceptors(new ReadOnlyRecordingInterceptor())
+                  .setWorkerInterceptors(new SubjectToReplayRecordingInterceptor())
+                  .build())
+          .setWorkerOptions(
+              WorkerOptions.newBuilder()
+                  .setPreferredVersionProvider(
+                      input -> {
+                        record("versionProvider");
+                        return null;
+                      })
                   .build())
           .build();
 
@@ -34,10 +43,14 @@ public void setUp() {
     calls.clear();
   }
 
+  /**
+   * Subjection to replay is a property of the calling context rather than of the Workflow's current
+   * state, so running once live covers the whole contract.
+   */
   @Test
-  public void isReadOnly() {
-    ReadOnlyWorkflow workflow =
-        testWorkflowRule.newWorkflowStubTimeoutOptions(ReadOnlyWorkflow.class);
+  public void isSubjectToReplay() {
+    SubjectToReplayWorkflow workflow =
+        testWorkflowRule.newWorkflowStubTimeoutOptions(SubjectToReplayWorkflow.class);
     WorkflowClient.start(workflow::run);
 
     workflow.query();
@@ -46,26 +59,32 @@ public void isReadOnly() {
     WorkflowStub.fromTyped(workflow).getResult(Void.class);
 
     Map<String, Boolean> expected = new ConcurrentHashMap<>();
-    expected.put("ExecuteWorkflow", false);
-    expected.put("workflowTask", false);
-    expected.put("ExecuteUpdate", false);
-    expected.put("updateHandler", false);
-    expected.put("HandleSignal", false);
-    expected.put("sideEffect", true);
+    // The durable Workflow path re-executes on every replay.
+    expected.put("ExecuteWorkflow", true);
+    expected.put("workflowTask", true);
+    expected.put("ExecuteUpdate", true);
+    expected.put("updateHandler", true);
+    expected.put("HandleSignal", true);
+    // An Await condition is read-only yet still re-evaluated on replay. This is the one context
+    // where subjection to replay and read-only disagree, and the reason the two are separate.
     expected.put("await", true);
-    expected.put("HandleQuery", true);
-    expected.put("query", true);
-    expected.put("ValidateUpdate", true);
-    expected.put("validator", true);
+    // Live callbacks run once against current state and are never re-executed from history.
+    expected.put("sideEffect", false);
+    expected.put("mutableSideEffect", false);
+    expected.put("versionProvider", false);
+    expected.put("HandleQuery", false);
+    expected.put("query", false);
+    expected.put("ValidateUpdate", false);
+    expected.put("validator", false);
     assertEquals(expected, calls);
   }
 
   private static void record(String name) {
-    calls.put(name, WorkflowUnsafe.isReadOnly());
+    calls.put(name, WorkflowUnsafe.isSubjectToReplay());
   }
 
   @WorkflowInterface
-  public interface ReadOnlyWorkflow {
+  public interface SubjectToReplayWorkflow {
     @WorkflowMethod
     void run();
 
@@ -82,18 +101,27 @@ public interface ReadOnlyWorkflow {
     void finish();
   }
 
-  public static class ReadOnlyWorkflowImpl implements ReadOnlyWorkflow {
+  public static class SubjectToReplayWorkflowImpl implements SubjectToReplayWorkflow {
     private boolean finished;
 
     @Override
     public void run() {
       record("workflowTask");
+      Workflow.getVersion("change", Workflow.DEFAULT_VERSION, 1);
       Workflow.sideEffect(
           Void.class,
           () -> {
             record("sideEffect");
             return null;
           });
+      Workflow.mutableSideEffect(
+          "id",
+          Integer.class,
+          Integer::equals,
+          () -> {
+            record("mutableSideEffect");
+            return 1;
+          });
       Workflow.await(
           () -> {
             record("await");
@@ -123,7 +151,7 @@ public void finish() {
     }
   }
 
-  private static class ReadOnlyRecordingInterceptor extends WorkerInterceptorBase {
+  private static class SubjectToReplayRecordingInterceptor extends WorkerInterceptorBase {
     @Override
     public WorkflowInboundCallsInterceptor interceptWorkflow(WorkflowInboundCallsInterceptor next) {
       return new WorkflowInboundCallsInterceptorBase(next) {