Add getRandomStream and isReadOnly

CHANGELOG.md

@@ -49,6 +49,10 @@ to docs, or any other relevant information.
 - Workflow tasks taking longer than 5 seconds now log a `[TMPRL1104]` warning reporting the task duration and the
   external storage downloads and uploads that contributed to it. The threshold is configurable with the
   `TEMPORAL_WORKFLOW_TASK_DURATION_WARN_SECONDS` environment variable.
+- Added experimental `Workflow.getRandomStream(String name)`, which returns a deterministic pseudorandom stream private
+  to the given name.
+- Added experimental `WorkflowUnsafe.isReadOnly()`, which reports whether the calling code is read-only.
+  replay.
 
 ### Changed
 - Moved the Spring AI integration to [temporalio/ai-integrations](https://github.com/temporalio/ai-integrations/tree/main/java/spring-ai).

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

@@ -293,6 +293,9 @@ Integer getVersion(
   /** Replay safe random. */
   Random newRandom();
 
+  /** Replay safe named random stream. */
+  Random getRandomStream(String name);
+
   /**
    * @return scope to be used for metrics reporting.
    */

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

@@ -82,6 +82,11 @@ public Random newRandom() {
     return workflowStateMachines.newRandom();
   }
 
+  @Override
+  public Random getRandomStream(String name) {
+    return workflowStateMachines.getRandomStream(name);
+  }
+
   @Override
   public Scope getMetricsScope() {
     return replayAwareWorkflowMetricsScope;

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

@@ -0,0 +1,31 @@
+package io.temporal.internal.statemachines;
+
+import com.google.common.hash.HashCode;
+import com.google.common.hash.Hashing;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+import javax.annotation.Nonnull;
+
+final class WorkflowRandomStreams {
+  private static final String SEED_VERSION = "temporal.sdk.random.v1";
+
+  private final Map<String, Random> streams = new HashMap<>();
+
+  long deriveSeed(@Nonnull String runId, @Nonnull String name) {
+    // The separators keep ("ab", "c") from colliding with ("a", "bc")
+    String seed = String.join("\0", SEED_VERSION, runId, name);
+    HashCode hash = Hashing.sha256().hashString(seed, StandardCharsets.UTF_8);
+    return ByteBuffer.wrap(hash.asBytes()).getLong();
+  }
+
+  Random get(@Nonnull String runId, @Nonnull String name) {
+    return streams.computeIfAbsent(name, key -> new Random(deriveSeed(runId, key)));
+  }
+
+  void reseed(@Nonnull String runId) {
+    streams.forEach((name, stream) -> stream.setSeed(deriveSeed(runId, name)));
+  }
+}

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

@@ -115,6 +115,8 @@ enum HandleEventStatus {
   /** Used Workflow.newRandom and randomUUID together with currentRunId. */
   private long idCounter;
 
+  private final WorkflowRandomStreams randomStreams = new WorkflowRandomStreams();
+
   /** Current workflow time. */
   private long currentTimeMillis = -1;
 
@@ -1195,6 +1197,11 @@ public Random newRandom() {
     return new Random(randomUUID().getLeastSignificantBits());
   }
 
+  public Random getRandomStream(String name) {
+    checkEventLoopExecuting();
+    return randomStreams.get(currentRunId, name);
+  }
+
   public void sideEffect(
       Functions.Func<Optional<Payloads>> func,
       UserMetadata userMetadata,
@@ -1548,6 +1555,7 @@ public void workflowTaskStarted(
     @Override
     public void updateRunId(String currentRunId) {
       WorkflowStateMachines.this.currentRunId = currentRunId;
+      randomStreams.reseed(currentRunId);
     }
   }
 

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

@@ -310,6 +310,9 @@ private void close(boolean fromWorkflowThread) {
     }
 
     closeStarted = true;
+    // Destroying the remaining threads runs their finally blocks. An eviction can happen at any
+    // time, so when those blocks run is not deterministic. Make them read-only.
+    workflowContext.setReadOnly(true);
     // lock is taken here
     try {
       // in some circumstances when a workflow broke Deadline Detector,

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

@@ -37,6 +37,8 @@ class SyncWorkflow implements ReplayWorkflow {
 
   private static final Logger log = LoggerFactory.getLogger(SyncWorkflow.class);
 
+  static final String UPDATE_THREAD_NAME_PREFIX = "update ";
+
   private final WorkflowThreadExecutor workflowThreadExecutor;
   private final SyncWorkflowDefinition workflow;
   @Nonnull private final WorkflowImplementationOptions workflowImplementationOptions;
@@ -148,7 +150,7 @@ public void handleUpdate(
       UpdateProtocolCallback callbacks) {
     final UpdateInfo updateInfo = new UpdateInfoImpl(updateName, updateId);
     runner.executeInWorkflowThread(
-        "update " + updateName,
+        UPDATE_THREAD_NAME_PREFIX + updateName,
         () -> {
           try {
             workflowContext.setCurrentUpdateInfo(updateInfo);

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

@@ -1153,13 +1153,14 @@ private <R> R mutableSideEffectImpl(
         id,
         metadata,
         (storedBinary) -> {
-          Optional<R> stored =
-              storedBinary.map(
-                  (b) ->
-                      dataConverterWithCurrentWorkflowContext.fromPayloads(
-                          0, Optional.of(b), resultClass, resultType));
           try {
+            // This callback is only run during live execution.
             readOnly = true;
+            Optional<R> stored =
+                storedBinary.map(
+                    (b) ->
+                        dataConverterWithCurrentWorkflowContext.fromPayloads(
+                            0, Optional.of(b), resultClass, resultType));
             R funcResult =
                 Objects.requireNonNull(
                     func.apply(), "mutableSideEffect function " + "returned null");
@@ -1184,8 +1185,15 @@ private <R> R mutableSideEffectImpl(
     if (unserialized != null) {
       return unserialized;
     }
-    return dataConverterWithCurrentWorkflowContext.fromPayloads(
-        0, result.get(), resultClass, resultType);
+
+    // Live execution skips this decode when a new value is recorded.
+    readOnly = true;
+    try {
+      return dataConverterWithCurrentWorkflowContext.fromPayloads(
+          0, result.get(), resultClass, resultType);
+    } finally {
+      readOnly = false;
+    }
   }
 
   @Override
@@ -1385,12 +1393,13 @@ public boolean await(Duration timeout, String reason, Supplier<Boolean> unblockC
   @Override
   public boolean await(
       Duration timeout, TimerOptions options, String reason, Supplier<Boolean> unblockCondition) {
+    Supplier<Boolean> condition = readOnly(unblockCondition);
     boolean cancelTimerOnCondition =
         replayContext.tryUseSdkFlag(SdkFlag.CANCEL_AWAIT_TIMER_ON_CONDITION);
 
     if (cancelTimerOnCondition) {
       // If condition is already satisfied, skip creating timer
-      if (unblockCondition.get()) {
+      if (condition.get()) {
         return true;
       }
       // Create timer in a cancellation scope so we can cancel it when condition is satisfied
@@ -1399,7 +1408,7 @@ public boolean await(
           Workflow.newCancellationScope(() -> timer.completeFrom(newTimer(timeout, options)));
       timerScope.run();
 
-      WorkflowThread.await(reason, () -> (timer.isCompleted() || unblockCondition.get()));
+      WorkflowThread.await(reason, () -> (timer.isCompleted() || condition.get()));
 
       boolean conditionSatisfied = !timer.isCompleted();
       if (conditionSatisfied) {
@@ -1409,14 +1418,26 @@ public boolean await(
     } else {
       // Old behavior: timer is not cancelled when condition is satisfied
       Promise<Void> timer = newTimer(timeout, options);
-      WorkflowThread.await(reason, () -> (timer.isCompleted() || unblockCondition.get()));
+      WorkflowThread.await(reason, () -> (timer.isCompleted() || condition.get()));
       return !timer.isCompleted();
     }
   }
 
   @Override
   public void await(String reason, Supplier<Boolean> unblockCondition) {
-    WorkflowThread.await(reason, unblockCondition);
+    WorkflowThread.await(reason, readOnly(unblockCondition));
+  }
+
+  private Supplier<Boolean> readOnly(Supplier<Boolean> condition) {
+    return () -> {
+      boolean previousReadOnly = readOnly;
+      readOnly = true;
+      try {
+        return condition.get();
+      } finally {
+        readOnly = previousReadOnly;
+      }
+    };
   }
 
   @SuppressWarnings("deprecation")

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

@@ -540,53 +540,24 @@ public static void executeActivity(ActivityInvocationOptions options, Functions.
   public static void await(String reason, Supplier<Boolean> unblockCondition)
       throws DestroyWorkflowThreadError {
     assertNotReadOnly(reason);
-    getWorkflowOutboundInterceptor()
-        .await(
-            reason,
-            () -> {
-              getRootWorkflowContext().setReadOnly(true);
-              try {
-                return unblockCondition.get();
-              } finally {
-                getRootWorkflowContext().setReadOnly(false);
-              }
-            });
+    // Interceptors may evaluate the condition themselves before delegating, so it is read-only here
+    // too, not only where SyncWorkflowContext evaluates it.
+    getWorkflowOutboundInterceptor().await(reason, () -> readOnly(unblockCondition::get));
   }
 
   public static boolean await(Duration timeout, String reason, Supplier<Boolean> unblockCondition)
       throws DestroyWorkflowThreadError {
     assertNotReadOnly(reason);
     return getWorkflowOutboundInterceptor()
-        .await(
-            timeout,
-            reason,
-            () -> {
-              getRootWorkflowContext().setReadOnly(true);
-              try {
-                return unblockCondition.get();
-              } finally {
-                getRootWorkflowContext().setReadOnly(false);
-              }
-            });
+        .await(timeout, reason, () -> readOnly(unblockCondition::get));
   }
 
   public static boolean await(
       Duration timeout, TimerOptions options, String reason, Supplier<Boolean> unblockCondition)
       throws DestroyWorkflowThreadError {
     assertNotReadOnly(reason);
     return getWorkflowOutboundInterceptor()
-        .await(
-            timeout,
-            options,
-            reason,
-            () -> {
-              getRootWorkflowContext().setReadOnly(true);
-              try {
-                return unblockCondition.get();
-              } finally {
-                getRootWorkflowContext().setReadOnly(false);
-              }
-            });
+        .await(timeout, options, reason, () -> readOnly(unblockCondition::get));
   }
 
   public static <R> R sideEffect(Class<R> resultClass, Type resultType, Func<R> func) {
@@ -785,6 +756,10 @@ public static Random newRandom() {
     return getRootWorkflowContext().newRandom();
   }
 
+  public static Random getRandomStream(String name) {
+    return getRootWorkflowContext().getReplayContext().getRandomStream(name);
+  }
+
   public static Logger getLogger(Class<?> clazz) {
     Logger logger = LoggerFactory.getLogger(clazz);
     return new ReplayAwareLogger(
@@ -960,14 +935,22 @@ static SyncWorkflowContext getRootWorkflowContext() {
     return DeterministicRunnerImpl.currentThreadInternal().getWorkflowContext();
   }
 
-  static boolean isReadOnly() {
-    return getRootWorkflowContext().isReadOnly();
+  /** Returns false if not under workflow code. */
+  public static boolean isReadOnly() {
+    boolean inWorkflow = isWorkflowThread() || QueryDispatcher.isQueryHandler();
+    return inWorkflow && getRootWorkflowContext().isReadOnly();
   }
 
   static void assertNotReadOnly(String action) {
-    if (isReadOnly()) {
-      throw new ReadOnlyException(action);
+    if (!isReadOnly()) {
+      return;
+    }
+    // This thread is being destroyed, so throw the error that ends it. Workflow code could catch a
+    // ReadOnlyException and keep running.
+    if (isWorkflowThread() && getWorkflowThread().getWorkflowThreadContext().isDestroyRequested()) {
+      throw new DestroyWorkflowThreadError();
     }
+    throw new ReadOnlyException(action);
   }
 
   static void assertNotInUpdateHandler(String message) {

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

@@ -84,10 +84,19 @@ public void run() {
       mdc.put(LoggerTag.TASK_QUEUE, replayWorkflowContext.getTaskQueue());
       mdc.put(LoggerTag.NAMESPACE, replayWorkflowContext.getNamespace());
 
-      // Repopulate the context(s)
-      ContextThreadLocal.setContextPropagators(this.contextPropagators);
-      ContextThreadLocal.propagateContextToCurrentThread(this.propagatedContexts);
       try {
+        // Repopulate the context(s)
+        ContextThreadLocal.setContextPropagators(this.contextPropagators);
+        // A rejected Update's thread only runs during live execution, so its context propagation
+        // is read-only.
+        boolean updateThread = name.startsWith(SyncWorkflow.UPDATE_THREAD_NAME_PREFIX);
+        SyncWorkflowContext workflowContext = getWorkflowContext();
+        workflowContext.setReadOnly(updateThread);
+        try {
+          ContextThreadLocal.propagateContextToCurrentThread(this.propagatedContexts);
+        } finally {
+          workflowContext.setReadOnly(false);
+        }
         // initialYield blocks thread until the first runUntilBlocked is called.
         // Otherwise, r starts executing without control of the sync.
         threadContext.initialYield();

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

@@ -1039,6 +1039,24 @@ public static Random newRandom() {
     return WorkflowInternal.newRandom();
   }
 
+  /**
+   * Returns a deterministic pseudorandom stream private to {@code name}.
+   *
+   * <p>Calling this method again with the same name returns the same logical stream where earlier
+   * 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 never draw in read-only code. Use {@link
+   * WorkflowUnsafe#isReadOnly()} to guard 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.
+   */
+  @Experimental
+  public static Random getRandomStream(String name) {
+    return WorkflowInternal.getRandomStream(name);
+  }
+
   /**
    * True if workflow code is being replayed.
    *

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

@@ -1,5 +1,6 @@
 package io.temporal.workflow.unsafe;
 
+import io.temporal.common.Experimental;
 import io.temporal.internal.sync.WorkflowInternal;
 import io.temporal.workflow.Functions;
 
@@ -46,6 +47,24 @@ public static boolean isReplaying() {
     return WorkflowInternal.isReplaying();
   }
 
+  /**
+   * Reports whether the currently executing code is read-only. Read-only code cannot schedule
+   * commands or change Workflow state, for example a Query handler, an Update validator, or an
+   * Await condition. Read-only code can also run a different number of times during live execution
+   * and on replay, so guard draws from {@link
+   * io.temporal.workflow.Workflow#getRandomStream(String)} with this.
+   *
+   * <p>Code that can also run outside a Workflow, such as a payload converter, should check {@link
+   * #isWorkflowThread()} too, because this returns false there.
+   *
+   * @return true if the calling context is read-only. This method always returns false if called
+   *     from outside Workflow code.
+   */
+  @Experimental
+  public static boolean isReadOnly() {
+    return WorkflowInternal.isReadOnly();
+  }
+
   /**
    * Runs the supplied procedure in the calling thread with disabled deadlock detection if called
    * from the workflow thread. Does nothing except the procedure execution if called from a