From 256b63ffcf75247ec759e86dd2f0d091d66d398b Mon Sep 17 00:00:00 2001 From: Patrik Beqo Date: Tue, 15 Sep 2026 19:00:22 -0400 Subject: [PATCH 1/2] Add named Workflow random streams and read-only detection --- .../replay/ReplayWorkflowContext.java | 3 + .../replay/ReplayWorkflowContextImpl.java | 5 + .../statemachines/WorkflowRandomStreams.java | 31 +++ .../statemachines/WorkflowStateMachines.java | 8 + .../internal/sync/WorkflowInternal.java | 8 +- .../java/io/temporal/workflow/Workflow.java | 18 ++ .../workflow/unsafe/WorkflowUnsafe.java | 16 ++ .../WorkflowRandomStreamsTest.java | 102 +++++++ .../workflow/WorkflowRandomStreamTest.java | 258 ++++++++++++++++++ .../workflow/WorkflowUnsafeReadOnlyTest.java | 162 +++++++++++ .../sync/DummySyncWorkflowContext.java | 5 + 11 files changed, 614 insertions(+), 2 deletions(-) create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowRandomStreams.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/statemachines/WorkflowRandomStreamsTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/WorkflowRandomStreamTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContext.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContext.java index 982737dbee..41854b9229 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContext.java +++ b/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. */ diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java index 2f600b20aa..97c7e20ff0 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java +++ b/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; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowRandomStreams.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowRandomStreams.java new file mode 100644 index 0000000000..dd2815a8aa --- /dev/null +++ b/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 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))); + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java index 2de6b6ea15..8f79cbbc10 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java +++ b/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> func, UserMetadata userMetadata, @@ -1548,6 +1555,7 @@ public void workflowTaskStarted( @Override public void updateRunId(String currentRunId) { WorkflowStateMachines.this.currentRunId = currentRunId; + randomStreams.reseed(currentRunId); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java index 84b1e91fd3..07b71769f2 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java +++ b/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); } diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java b/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java index d04a617d5d..561c2221e2 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java +++ b/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}. + * + *

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. + * + *

Draws are not recorded in Workflow History, so do not draw in read-only code. Use {@link + * WorkflowUnsafe#isReadOnly()} to gate draws. + * + *

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. * diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/unsafe/WorkflowUnsafe.java b/temporal-sdk/src/main/java/io/temporal/workflow/unsafe/WorkflowUnsafe.java index 1a67b4e8a6..da635d9727 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/unsafe/WorkflowUnsafe.java +++ b/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. + * + *

Read-only code includes Query handlers, Update validators, Side Effect functions, Await + * conditions, and other SDK callbacks that must not mutate Workflow state. + * + *

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 diff --git a/temporal-sdk/src/test/java/io/temporal/internal/statemachines/WorkflowRandomStreamsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/statemachines/WorkflowRandomStreamsTest.java new file mode 100644 index 0000000000..1752d17218 --- /dev/null +++ b/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 interleavedA = new ArrayList<>(); + List 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 solo(String name, int draws) { + Random random = new WorkflowRandomStreams().get(RUN_ID, name); + List result = new ArrayList<>(); + for (int i = 0; i < draws; i++) { + result.add(random.nextLong()); + } + return result; + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowRandomStreamTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowRandomStreamTest.java new file mode 100644 index 0000000000..b9463a4cd7 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowRandomStreamTest.java @@ -0,0 +1,258 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assume.assumeTrue; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.activity.ActivityOptions; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflowservice.v1.ResetWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.ResetWorkflowExecutionResponse; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowTargetOptions; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.WorkerOptions; +import java.time.Duration; +import java.util.Random; +import java.util.UUID; +import org.junit.Rule; +import org.junit.Test; + +public class WorkflowRandomStreamTest { + private static final String STREAM_NAME = "io.temporal.test"; + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes( + SimpleWorkflowImpl.class, + ReplayWorkflowImpl.class, + ResetWorkflowImpl.class, + ResetLateSourceWorkflowImpl.class, + ContinueAsNewWorkflowImpl.class, + ParentWorkflowImpl.class) + .setActivityImplementations(new SimpleActivityImpl()) + .setWorkerOptions( + WorkerOptions.newBuilder() + .setStickyQueueScheduleToStartTimeout(Duration.ZERO) + .build()) + .build(); + + @Test + public void noCollisionAcrossRuns() { + SimpleWorkflow first = testWorkflowRule.newWorkflowStubTimeoutOptions(SimpleWorkflow.class); + SimpleWorkflow second = testWorkflowRule.newWorkflowStubTimeoutOptions(SimpleWorkflow.class); + + assertNotEquals(first.run(), second.run()); + } + + @Test + public void deterministicReplay() { + ReplayWorkflow workflow = testWorkflowRule.newWorkflowStubTimeoutOptions(ReplayWorkflow.class); + + long result = workflow.run(); + + assertEquals(result, workflow.currentState()); + } + + @Test + public void resetReproducesValues() { + assumeTrue( + "Test Server doesn't support reset workflow", SDKTestWorkflowRule.useExternalService); + assertResetValues(ResetWorkflow.class); + } + + @Test + public void resetReseedsSourceCreatedAfterResetPoint() { + assumeTrue( + "Test Server doesn't support reset workflow", SDKTestWorkflowRule.useExternalService); + assertResetValues(ResetLateSourceWorkflow.class); + } + + @Test + public void continueAsNewDrawsNewValues() { + ContinueAsNewWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(ContinueAsNewWorkflow.class); + + long[] values = workflow.run(null); + + assertEquals(2, values.length); + assertNotEquals(values[0], values[1]); + } + + @Test + public void childContinueAsNewDrawsNewValues() { + ParentWorkflow workflow = testWorkflowRule.newWorkflowStubTimeoutOptions(ParentWorkflow.class); + + long[] values = workflow.run(); + + assertEquals(3, values.length); + assertNotEquals(values[0], values[1]); + assertNotEquals(values[0], values[2]); + assertNotEquals(values[1], values[2]); + } + + private void assertResetValues(Class workflowType) { + WorkflowClient client = testWorkflowRule.getWorkflowClient(); + WorkflowStub stub = + WorkflowStub.fromTyped(testWorkflowRule.newWorkflowStubTimeoutOptions(workflowType)); + WorkflowExecution execution = stub.start(); + long[] original = stub.getResult(long[].class); + assertEquals(2, original.length); + assertNotEquals(original[0], original[1]); + + // The reset targets the second Workflow Task (id=10), so the first draw is replayed and the + // second draw is redrawn + ResetWorkflowExecutionResponse response = + client + .getWorkflowServiceStubs() + .blockingStub() + .resetWorkflowExecution( + ResetWorkflowExecutionRequest.newBuilder() + .setNamespace(SDKTestWorkflowRule.NAMESPACE) + .setWorkflowExecution(execution) + .setWorkflowTaskFinishEventId(10) + .setReason("Integration test") + .setRequestId(UUID.randomUUID().toString()) + .build()); + + long[] afterReset = + client + .newUntypedWorkflowStub( + WorkflowTargetOptions.newBuilder() + .setWorkflowId(execution.getWorkflowId()) + .setRunId(response.getRunId()) + .build()) + .getResult(long[].class); + assertEquals(2, afterReset.length); + assertNotEquals(afterReset[0], afterReset[1]); + + assertEquals(original[0], afterReset[0]); + assertNotEquals(original[1], afterReset[1]); + } + + @WorkflowInterface + public interface SimpleWorkflow { + @WorkflowMethod + long run(); + } + + @WorkflowInterface + public interface ReplayWorkflow { + @WorkflowMethod + long run(); + + @QueryMethod + long currentState(); + } + + @WorkflowInterface + public interface ResetWorkflow { + @WorkflowMethod + long[] run(); + } + + @WorkflowInterface + public interface ResetLateSourceWorkflow { + @WorkflowMethod + long[] run(); + } + + @WorkflowInterface + public interface ContinueAsNewWorkflow { + @WorkflowMethod + long[] run(Long previous); + } + + @WorkflowInterface + public interface ParentWorkflow { + @WorkflowMethod + long[] run(); + } + + @ActivityInterface + public interface SimpleActivity { + @ActivityMethod + void run(); + } + + public static class SimpleWorkflowImpl implements SimpleWorkflow { + @Override + public long run() { + return Workflow.getRandomStream(STREAM_NAME).nextLong(); + } + } + + public static class ReplayWorkflowImpl implements ReplayWorkflow { + private long state; + + @Override + public long run() { + Random random = Workflow.getRandomStream(STREAM_NAME); + state = random.nextLong(); + newSimpleActivity().run(); + state = random.nextLong(); + return state; + } + + @Override + public long currentState() { + return state; + } + } + + public static class ResetWorkflowImpl implements ResetWorkflow { + @Override + public long[] run() { + Random random = Workflow.getRandomStream(STREAM_NAME); + long first = random.nextLong(); + newSimpleActivity().run(); + long second = random.nextLong(); + return new long[] {first, second}; + } + } + + public static class ResetLateSourceWorkflowImpl implements ResetLateSourceWorkflow { + @Override + public long[] run() { + long first = Workflow.getRandomStream("other").nextLong(); + newSimpleActivity().run(); + long second = Workflow.getRandomStream(STREAM_NAME).nextLong(); + return new long[] {first, second}; + } + } + + public static class ContinueAsNewWorkflowImpl implements ContinueAsNewWorkflow { + @Override + public long[] run(Long previous) { + long current = Workflow.getRandomStream(STREAM_NAME).nextLong(); + if (previous == null) { + Workflow.continueAsNew(current); + } + return new long[] {previous, current}; + } + } + + public static class ParentWorkflowImpl implements ParentWorkflow { + @Override + public long[] run() { + long parent = Workflow.getRandomStream(STREAM_NAME).nextLong(); + long[] child = Workflow.newChildWorkflowStub(ContinueAsNewWorkflow.class).run(null); + return new long[] {parent, child[0], child[1]}; + } + } + + public static class SimpleActivityImpl implements SimpleActivity { + @Override + public void run() {} + } + + private static SimpleActivity newSimpleActivity() { + return Workflow.newActivityStub( + SimpleActivity.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofMinutes(1)).build()); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java new file mode 100644 index 0000000000..5c12165d37 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java @@ -0,0 +1,162 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowStub; +import io.temporal.common.interceptors.WorkerInterceptorBase; +import io.temporal.common.interceptors.WorkflowInboundCallsInterceptor; +import io.temporal.common.interceptors.WorkflowInboundCallsInterceptorBase; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.WorkerFactoryOptions; +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 { + private static final Map calls = new ConcurrentHashMap<>(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(ReadOnlyWorkflowImpl.class) + .setWorkerFactoryOptions( + WorkerFactoryOptions.newBuilder() + .setWorkerInterceptors(new ReadOnlyRecordingInterceptor()) + .build()) + .build(); + + @Before + public void setUp() { + calls.clear(); + } + + @Test + public void isReadOnly() { + ReadOnlyWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(ReadOnlyWorkflow.class); + WorkflowClient.start(workflow::run); + + workflow.query(); + workflow.update(); + workflow.finish(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + Map 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); + expected.put("await", true); + expected.put("HandleQuery", true); + expected.put("query", true); + expected.put("ValidateUpdate", true); + expected.put("validator", true); + assertEquals(expected, calls); + } + + private static void record(String name) { + calls.put(name, WorkflowUnsafe.isReadOnly()); + } + + @WorkflowInterface + public interface ReadOnlyWorkflow { + @WorkflowMethod + void run(); + + @QueryMethod + boolean query(); + + @UpdateMethod + void update(); + + @UpdateValidatorMethod(updateName = "update") + void validateUpdate(); + + @SignalMethod + void finish(); + } + + public static class ReadOnlyWorkflowImpl implements ReadOnlyWorkflow { + private boolean finished; + + @Override + public void run() { + record("workflowTask"); + Workflow.sideEffect( + Void.class, + () -> { + record("sideEffect"); + return null; + }); + Workflow.await( + () -> { + record("await"); + return finished; + }); + } + + @Override + public boolean query() { + record("query"); + return true; + } + + @Override + public void update() { + record("updateHandler"); + } + + @Override + public void validateUpdate() { + record("validator"); + } + + @Override + public void finish() { + finished = true; + } + } + + private static class ReadOnlyRecordingInterceptor extends WorkerInterceptorBase { + @Override + public WorkflowInboundCallsInterceptor interceptWorkflow(WorkflowInboundCallsInterceptor next) { + return new WorkflowInboundCallsInterceptorBase(next) { + @Override + public WorkflowOutput execute(WorkflowInput input) { + record("ExecuteWorkflow"); + return super.execute(input); + } + + @Override + public void handleSignal(SignalInput input) { + record("HandleSignal"); + super.handleSignal(input); + } + + @Override + public QueryOutput handleQuery(QueryInput input) { + record("HandleQuery"); + return super.handleQuery(input); + } + + @Override + public void validateUpdate(UpdateInput input) { + record("ValidateUpdate"); + super.validateUpdate(input); + } + + @Override + public UpdateOutput executeUpdate(UpdateInput input) { + record("ExecuteUpdate"); + return super.executeUpdate(input); + } + }; + } + } +} diff --git a/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.java b/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.java index f89e61c64b..c62081a4a1 100644 --- a/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.java +++ b/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.java @@ -288,6 +288,11 @@ public Random newRandom() { throw new UnsupportedOperationException("not implemented"); } + @Override + public Random getRandomStream(String name) { + throw new UnsupportedOperationException("not implemented"); + } + @Override public Scope getMetricsScope() { return new NoopScope(); From d5bb22cfdfd3d687d2286633b005ee316bf623e8 Mon Sep 17 00:00:00 2001 From: Patrik Beqo Date: Fri, 18 Sep 2026 17:37:00 -0400 Subject: [PATCH 2/2] feat: isSubjectToReplay --- .../replay/ReplayWorkflowContextImpl.java | 8 ++- .../internal/sync/QueryDispatcher.java | 2 + .../temporal/internal/sync/SyncWorkflow.java | 2 + .../internal/sync/SyncWorkflowContext.java | 13 ++++ .../internal/sync/WorkflowInternal.java | 19 +++++- .../java/io/temporal/workflow/Workflow.java | 4 +- .../workflow/unsafe/WorkflowUnsafe.java | 15 ++-- ...=> WorkflowUnsafeSubjectToReplayTest.java} | 68 +++++++++++++------ 8 files changed, 98 insertions(+), 33 deletions(-) rename temporal-sdk/src/test/java/io/temporal/workflow/{WorkflowUnsafeReadOnlyTest.java => WorkflowUnsafeSubjectToReplayTest.java} (59%) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java index 97c7e20ff0..757d004d9a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowContextImpl.java +++ b/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); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/QueryDispatcher.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/QueryDispatcher.java index b92ac3b282..c7e3395e2d 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/QueryDispatcher.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/QueryDispatcher.java @@ -99,6 +99,7 @@ public Optional handleQuery( } try { replayContext.setReadOnly(true); + replayContext.setSubjectToReplay(false); queryHandlerWorkflowContext.set(replayContext); Object result = inboundCallsInterceptor @@ -107,6 +108,7 @@ public Optional handleQuery( return dataConverterWithWorkflowContext.toPayloads(result); } finally { replayContext.setReadOnly(false); + replayContext.setSubjectToReplay(true); queryHandlerWorkflowContext.set(null); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java index 9351f0f34e..66e6bc3af7 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java +++ b/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(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java index 742a663f8f..a208025ac0 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java +++ b/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 nexusServiceOptionsMap; private boolean readOnly = false; + private boolean subjectToReplay = true; private final WorkflowThreadLocal currentUpdateInfo = new WorkflowThreadLocal<>(); @Nullable private String currentDetails; @@ -1066,10 +1067,12 @@ public 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 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 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 getRunningSignalHandlers() { return signalDispatcher.getRunningSignalHandlers(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java index 07b71769f2..662d908483 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java @@ -718,6 +718,17 @@ public static T readOnly(Functions.Func func) { } } + public static T notSubjectToReplay(Functions.Func 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()); } @@ -923,16 +934,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); diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java b/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java index 561c2221e2..20e9c67e25 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java +++ b/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java @@ -718,8 +718,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. * - *

Draws are not recorded in Workflow History, so do not draw in read-only code. Use {@link - * WorkflowUnsafe#isReadOnly()} to gate draws. + *

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. * *

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. diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/unsafe/WorkflowUnsafe.java b/temporal-sdk/src/main/java/io/temporal/workflow/unsafe/WorkflowUnsafe.java index da635d9727..9894a02615 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/unsafe/WorkflowUnsafe.java +++ b/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. * - *

Read-only code includes Query handlers, Update validators, Side Effect functions, Await - * conditions, and other SDK callbacks that must not mutate Workflow state. + *

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. * *

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(); } /** diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeSubjectToReplayTest.java similarity index 59% rename from temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java rename to temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeSubjectToReplayTest.java index 5c12165d37..5a32f57b2c 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeReadOnlyTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeSubjectToReplayTest.java @@ -9,6 +9,7 @@ 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; @@ -16,16 +17,24 @@ import org.junit.Rule; import org.junit.Test; -public class WorkflowUnsafeReadOnlyTest { +public class WorkflowUnsafeSubjectToReplayTest { private static final Map 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 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) {