Add getRandomStream and isSubjectToReplay

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

@@ -292,6 +292,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

@@ -81,6 +81,11 @@ public Random newRandom() {
     return workflowStateMachines.newRandom();
   }
 
+  @Override
+  public Random getRandomStream(String name) {
+    return workflowStateMachines.getRandomStream(name);
+  }
+
   @Override
   public Scope getMetricsScope() {
     return replayAwareWorkflowMetricsScope;
@@ -350,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/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/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;
 
@@ -1066,10 +1067,12 @@ public <R> R sideEffect(
           () -> {
             try {
               readOnly = true;
+              subjectToReplay = false;
               R r = func.apply();
               return dataConverterWithCurrentWorkflowContext.toPayloads(r);
             } finally {
               readOnly = false;
+              subjectToReplay = true;
             }
           },
           userMetadata,
@@ -1132,6 +1135,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");
@@ -1142,6 +1146,7 @@ private <R> R mutableSideEffectImpl(
             return Optional.empty(); // returned only when value doesn't need to be updated
           } finally {
             readOnly = false;
+            subjectToReplay = true;
           }
         },
         (p) ->
@@ -1284,6 +1289,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

@@ -718,6 +718,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());
   }
@@ -744,6 +755,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(
@@ -929,6 +944,10 @@ static void assertNotReadOnly(String 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

@@ -711,6 +711,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 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.
+   */
+  @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 re-executed when the Workflow replays.
+   *
+   * <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 if the calling context is re-executed on replay
+   */
+  @Experimental
+  public static boolean isSubjectToReplay() {
+    return WorkflowInternal.isSubjectToReplay();
+  }
+
   /**
    * 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

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

@@ -0,0 +1,102 @@
+package io.temporal.internal.statemachines;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertSame;
+
+import com.google.common.io.BaseEncoding;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.junit.Test;
+
+public class WorkflowRandomStreamsTest {
+  private static final String RUN_ID = "runID";
+  private static final String NAME = "io.temporal.test";
+
+  @Test
+  public void deriveSeed() {
+    WorkflowRandomStreams randoms = new WorkflowRandomStreams();
+    long seed = randoms.deriveSeed(RUN_ID, NAME);
+    assertNotEquals(seed, randoms.deriveSeed("other", NAME));
+    assertNotEquals(seed, randoms.deriveSeed(RUN_ID, "other"));
+    assertNotEquals(seed, randoms.deriveSeed("other", "other"));
+  }
+
+  /** Pins the seed derivation and resulting byte stream. Changing either breaks replay. */
+  @Test
+  public void getRandomStreamGolden() {
+    WorkflowRandomStreams randoms = new WorkflowRandomStreams();
+    assertEquals(8181915698088084985L, randoms.deriveSeed(RUN_ID, NAME));
+
+    byte[] bytes = new byte[32];
+    randoms.get(RUN_ID, NAME).nextBytes(bytes);
+    assertEquals(
+        "1c1d4dd36999ff851d72aa41f660ecd3220d83499109ba3ba24e455a1b776c21",
+        BaseEncoding.base16().lowerCase().encode(bytes));
+  }
+
+  @Test
+  public void deriveSeedSeparators() {
+    WorkflowRandomStreams randoms = new WorkflowRandomStreams();
+    assertNotEquals(randoms.deriveSeed("ab", "c"), randoms.deriveSeed("a", "bc"));
+  }
+
+  /** A second lookup under the same name continues the sequence rather than restarting it. */
+  @Test
+  public void getRandomStreamMemoizes() {
+    WorkflowRandomStreams randoms = new WorkflowRandomStreams();
+
+    Random first = randoms.get(RUN_ID, NAME);
+    long firstDraw = first.nextLong();
+
+    Random second = randoms.get(RUN_ID, NAME);
+    long secondDraw = second.nextLong();
+
+    assertSame(first, second);
+    assertNotEquals(firstDraw, secondDraw);
+  }
+
+  /**
+   * Interleaving draws across two names yields the same sequence per name as drawing from each on
+   * its own, so how often a workflow draws from one name cannot shift another.
+   */
+  @Test
+  public void getRandomStreamNamesAreIndependent() {
+    WorkflowRandomStreams randoms = new WorkflowRandomStreams();
+    Random first = randoms.get(RUN_ID, NAME);
+    Random second = randoms.get(RUN_ID, "other");
+
+    List<Long> interleavedA = new ArrayList<>();
+    List<Long> interleavedB = new ArrayList<>();
+    for (int i = 0; i < 3; i++) {
+      interleavedA.add(first.nextLong());
+      interleavedB.add(second.nextLong());
+    }
+
+    assertEquals(solo(NAME, 3), interleavedA);
+    assertEquals(solo("other", 3), interleavedB);
+    assertNotEquals(interleavedA, interleavedB);
+  }
+
+  @Test
+  public void reseedRandomsInPlace() {
+    WorkflowRandomStreams randoms = new WorkflowRandomStreams();
+
+    Random first = randoms.get(RUN_ID, NAME);
+    randoms.reseed("other");
+    Random second = randoms.get("other", NAME);
+
+    assertSame(first, second);
+    assertEquals(new WorkflowRandomStreams().get("other", NAME).nextLong(), second.nextLong());
+  }
+
+  private static List<Long> solo(String name, int draws) {
+    Random random = new WorkflowRandomStreams().get(RUN_ID, name);
+    List<Long> result = new ArrayList<>();
+    for (int i = 0; i < draws; i++) {
+      result.add(random.nextLong());
+    }
+    return result;
+  }
+}