Add named Workflow random streams and read-only detection

Author: patbeqo

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;

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/WorkflowInternal.java

@@ -744,6 +744,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(
@@ -919,11 +923,11 @@ static SyncWorkflowContext getRootWorkflowContext() {
     return DeterministicRunnerImpl.currentThreadInternal().getWorkflowContext();
   }
 
-  static boolean isReadOnly() {
+  public static boolean isReadOnly() {
     return getRootWorkflowContext().isReadOnly();
   }
 
-  static void assertNotReadOnly(String action) {
+  public static void assertNotReadOnly(String action) {
     if (isReadOnly()) {
       throw new ReadOnlyException(action);
     }

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 do not draw in read-only code. Use {@link
+   * WorkflowUnsafe#isReadOnly()} 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,21 @@ public static boolean isReplaying() {
     return WorkflowInternal.isReplaying();
   }
 
+  /**
+   * Reports whether the current code is running where Workflow state cannot be mutated.
+   *
+   * <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>Must be called from Workflow code.
+   *
+   * @return true in a read-only Workflow context
+   */
+  @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

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;
+  }
+}