Add Workflow random streams and read-only detection

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

@@ -14,6 +14,7 @@
 import io.temporal.workflow.Functions;
 import io.temporal.workflow.Functions.Func;
 import io.temporal.workflow.Functions.Func1;
+import io.temporal.workflow.WorkflowRandomStream;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
@@ -292,6 +293,9 @@ Integer getVersion(
   /** Replay safe random. */
   Random newRandom();
 
+  /** Replay safe named random stream. */
+  WorkflowRandomStream getRandomStream(String name);
+
   /**
    * @return scope to be used for metrics reporting.
    */

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

@@ -22,6 +22,7 @@
 import io.temporal.workflow.Functions;
 import io.temporal.workflow.Functions.Func;
 import io.temporal.workflow.Functions.Func1;
+import io.temporal.workflow.WorkflowRandomStream;
 import java.time.Duration;
 import java.util.*;
 import javax.annotation.Nonnull;
@@ -81,6 +82,11 @@ public Random newRandom() {
     return workflowStateMachines.newRandom();
   }
 
+  @Override
+  public WorkflowRandomStream 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,105 @@
+package io.temporal.internal.statemachines;
+
+import io.temporal.workflow.WorkflowRandomStream;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+final class WorkflowRandomStreams {
+  private static final byte[] SEED_VERSION =
+      "temporal.sdk.random.v1".getBytes(StandardCharsets.UTF_8);
+
+  private final Map<String, Stream> streams = new HashMap<>();
+  private String runId;
+
+  WorkflowRandomStream get(String name) {
+    Objects.requireNonNull(name, "name");
+    if (runId == null) {
+      throw new IllegalStateException("Workflow Run ID is not initialized");
+    }
+    return streams.computeIfAbsent(name, key -> new Stream(deriveSeed(runId, key)));
+  }
+
+  void updateRunId(String runId) {
+    this.runId = Objects.requireNonNull(runId, "runId");
+    streams.forEach((name, stream) -> stream.reseed(deriveSeed(runId, name)));
+  }
+
+  static byte[] deriveSeed(String runId, String name) {
+    MessageDigest digest = newSha256();
+    digest.update(SEED_VERSION);
+    digest.update((byte) 0);
+    digest.update(runId.getBytes(StandardCharsets.UTF_8));
+    digest.update((byte) 0);
+    digest.update(name.getBytes(StandardCharsets.UTF_8));
+    return digest.digest();
+  }
+
+  private static MessageDigest newSha256() {
+    try {
+      return MessageDigest.getInstance("SHA-256");
+    } catch (NoSuchAlgorithmException e) {
+      throw new IllegalStateException("SHA-256 is unavailable", e);
+    }
+  }
+
+  private static final class Stream implements WorkflowRandomStream {
+    private final MessageDigest digest = newSha256();
+    private byte[] seed;
+    private byte[] block = new byte[0];
+    private int blockOffset;
+    private long counter;
+
+    private Stream(byte[] seed) {
+      reseed(seed);
+    }
+
+    @Override
+    public void nextBytes(byte[] bytes) {
+      Objects.requireNonNull(bytes, "bytes");
+      int outputOffset = 0;
+      while (outputOffset < bytes.length) {
+        if (blockOffset == block.length) {
+          refill();
+        }
+        int length = Math.min(bytes.length - outputOffset, block.length - blockOffset);
+        System.arraycopy(block, blockOffset, bytes, outputOffset, length);
+        blockOffset += length;
+        outputOffset += length;
+      }
+    }
+
+    @Override
+    public long nextLong() {
+      byte[] bytes = new byte[Long.BYTES];
+      nextBytes(bytes);
+      long value = 0;
+      for (byte current : bytes) {
+        value = (value << Byte.SIZE) | (current & 0xffL);
+      }
+      return value;
+    }
+
+    private void refill() {
+      digest.reset();
+      digest.update(seed);
+      for (int shift = Long.SIZE - Byte.SIZE; shift >= 0; shift -= Byte.SIZE) {
+        digest.update((byte) (counter >>> shift));
+      }
+      counter++;
+      block = digest.digest();
+      blockOffset = 0;
+    }
+
+    private void reseed(byte[] seed) {
+      this.seed = Arrays.copyOf(seed, seed.length);
+      block = new byte[0];
+      blockOffset = 0;
+      counter = 0;
+    }
+  }
+}

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

@@ -35,6 +35,7 @@
 import io.temporal.workflow.ChildWorkflowCancellationType;
 import io.temporal.workflow.Functions;
 import io.temporal.workflow.NexusOperationCancellationType;
+import io.temporal.workflow.WorkflowRandomStream;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
 import java.util.function.BiFunction;
@@ -115,6 +116,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;
 
@@ -845,6 +848,7 @@ private void handleNonStatefulEvent(HistoryEvent event, boolean hasNextEvent) {
       case EVENT_TYPE_WORKFLOW_EXECUTION_STARTED:
         this.currentRunId =
             event.getWorkflowExecutionStartedEventAttributes().getOriginalExecutionRunId();
+        randomStreams.updateRunId(currentRunId);
         callbacks.start(event);
         break;
       case EVENT_TYPE_WORKFLOW_TASK_SCHEDULED:
@@ -1195,6 +1199,11 @@ public Random newRandom() {
     return new Random(randomUUID().getLeastSignificantBits());
   }
 
+  public WorkflowRandomStream getRandomStream(String name) {
+    checkEventLoopExecuting();
+    return randomStreams.get(name);
+  }
+
   public void sideEffect(
       Functions.Func<Optional<Payloads>> func,
       UserMetadata userMetadata,
@@ -1548,6 +1557,7 @@ public void workflowTaskStarted(
     @Override
     public void updateRunId(String currentRunId) {
       WorkflowStateMachines.this.currentRunId = currentRunId;
+      randomStreams.updateRunId(currentRunId);
     }
   }
 

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

@@ -744,6 +744,11 @@ public static Random newRandom() {
     return getRootWorkflowContext().newRandom();
   }
 
+  public static WorkflowRandomStream getRandomStream(String name) {
+    assertNotReadOnly("random");
+    return getRootWorkflowContext().getReplayContext().getRandomStream(name);
+  }
+
   public static Logger getLogger(Class<?> clazz) {
     Logger logger = LoggerFactory.getLogger(clazz);
     return new ReplayAwareLogger(
@@ -919,8 +924,12 @@ static SyncWorkflowContext getRootWorkflowContext() {
     return DeterministicRunnerImpl.currentThreadInternal().getWorkflowContext();
   }
 
-  static boolean isReadOnly() {
-    return getRootWorkflowContext().isReadOnly();
+  public static boolean isReadOnly() {
+    if (QueryDispatcher.isQueryHandler()) {
+      return getRootWorkflowContext().isReadOnly();
+    }
+    Optional<WorkflowThread> thread = DeterministicRunnerImpl.currentThreadInternalIfPresent();
+    return thread.isPresent() && getRootWorkflowContext().isReadOnly();
   }
 
   static void assertNotReadOnly(String action) {

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

@@ -711,6 +711,22 @@ 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>Each draw advances Workflow state without recording an Event in Workflow History. Replay
+   * must make the same draws in the same order. Calling this method in read-only code fails; shared
+   * code can check {@link WorkflowUnsafe#isReadOnly()} first.
+   */
+  @Experimental
+  public static WorkflowRandomStream getRandomStream(String name) {
+    return WorkflowInternal.getRandomStream(name);
+  }
+
   /**
    * True if workflow code is being replayed.
    *

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

@@ -0,0 +1,26 @@
+package io.temporal.workflow;
+
+import io.temporal.common.Experimental;
+import io.temporal.workflow.unsafe.WorkflowUnsafe;
+
+/**
+ * A named deterministic pseudorandom stream for Workflow code.
+ *
+ * <p>Repeated calls to {@link Workflow#getRandomStream(String)} with the same name return the same
+ * logical stream at its current position. Different names do not affect each other's sequences.
+ *
+ * <p>Each draw advances Workflow state without recording an Event in Workflow History. Replay must
+ * make the same draws in the same order. Do not draw in read-only code; shared code can check
+ * {@link WorkflowUnsafe#isReadOnly()}.
+ *
+ * <p>Use a stable package-style name. Stream names are retained for the life of the Workflow Run.
+ */
+@Experimental
+public interface WorkflowRandomStream {
+
+  /** Fills {@code bytes} with the next bytes from this stream. */
+  void nextBytes(byte[] bytes);
+
+  /** Returns the next signed 64-bit value from this stream in big-endian byte order. */
+  long nextLong();
+}

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,19 @@ 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.
+   *
+   * @return true in a read-only Workflow context, or false outside Workflow execution
+   */
+  @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,89 @@
+package io.temporal.internal.statemachines;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertSame;
+
+import io.temporal.workflow.WorkflowRandomStream;
+import java.util.Base64;
+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 goldenSeedAndBytes() {
+    assertEquals(
+        "cYwA+k67hflWG1GVOzY897A19H2s16mOOzlic16UFtI=",
+        Base64.getEncoder().encodeToString(WorkflowRandomStreams.deriveSeed(RUN_ID, NAME)));
+
+    WorkflowRandomStreams randoms = new WorkflowRandomStreams();
+    randoms.updateRunId(RUN_ID);
+    byte[] bytes = new byte[32];
+    randoms.get(NAME).nextBytes(bytes);
+
+    assertEquals(
+        "wY4Mb9uhREeU08hmLgZRqar87inHCMHTkHCss0U8Wi0=", Base64.getEncoder().encodeToString(bytes));
+    assertEquals(-4499645303130864569L, randoms(RUN_ID).get(NAME).nextLong());
+  }
+
+  @Test
+  public void seedFramingSeparatesRunIdAndName() {
+    assertNotEquals(
+        Base64.getEncoder().encodeToString(WorkflowRandomStreams.deriveSeed("ab", "c")),
+        Base64.getEncoder().encodeToString(WorkflowRandomStreams.deriveSeed("a", "bc")));
+  }
+
+  @Test
+  public void sameNameContinuesAndNamesAreIndependent() {
+    WorkflowRandomStreams interleaved = randoms(RUN_ID);
+    WorkflowRandomStream first = interleaved.get(NAME);
+    long firstValue = first.nextLong();
+    long otherValue = interleaved.get("other").nextLong();
+    WorkflowRandomStream second = interleaved.get(NAME);
+    long secondValue = second.nextLong();
+
+    WorkflowRandomStreams isolated = randoms(RUN_ID);
+    assertSame(first, second);
+    assertEquals(firstValue, isolated.get(NAME).nextLong());
+    assertEquals(secondValue, isolated.get(NAME).nextLong());
+    assertEquals(otherValue, isolated.get("other").nextLong());
+    assertNotEquals(firstValue, otherValue);
+  }
+
+  @Test
+  public void runIdUpdateReseedsExistingStreamInPlace() {
+    WorkflowRandomStreams randoms = randoms(RUN_ID);
+    WorkflowRandomStream before = randoms.get(NAME);
+    before.nextLong();
+
+    randoms.updateRunId("new-run");
+    WorkflowRandomStream after = randoms.get(NAME);
+
+    WorkflowRandomStreams fresh = randoms("new-run");
+    assertSame(before, after);
+    assertEquals(fresh.get(NAME).nextLong(), after.nextLong());
+  }
+
+  @Test
+  public void streamCreatedAfterRunIdUpdateUsesNewRun() {
+    WorkflowRandomStreams randoms = randoms(RUN_ID);
+    randoms.get("before-reset").nextLong();
+    randoms.updateRunId("new-run");
+
+    byte[] actual = new byte[32];
+    randoms.get(NAME).nextBytes(actual);
+
+    byte[] expected = new byte[32];
+    randoms("new-run").get(NAME).nextBytes(expected);
+    assertArrayEquals(expected, actual);
+  }
+
+  private static WorkflowRandomStreams randoms(String runId) {
+    WorkflowRandomStreams randoms = new WorkflowRandomStreams();
+    randoms.updateRunId(runId);
+    return randoms;
+  }
+}