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..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 @@ -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); } 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/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 84b1e91fd3..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()); } @@ -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); 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..20e9c67e25 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 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. + */ + @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..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 @@ -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. + * + *

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 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 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/WorkflowUnsafeSubjectToReplayTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeSubjectToReplayTest.java new file mode 100644 index 0000000000..5a32f57b2c --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeSubjectToReplayTest.java @@ -0,0 +1,190 @@ +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.worker.WorkerOptions; +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 WorkflowUnsafeSubjectToReplayTest { + private static final Map calls = new ConcurrentHashMap<>(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(SubjectToReplayWorkflowImpl.class) + .setWorkerFactoryOptions( + WorkerFactoryOptions.newBuilder() + .setWorkerInterceptors(new SubjectToReplayRecordingInterceptor()) + .build()) + .setWorkerOptions( + WorkerOptions.newBuilder() + .setPreferredVersionProvider( + input -> { + record("versionProvider"); + return null; + }) + .build()) + .build(); + + @Before + 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 isSubjectToReplay() { + SubjectToReplayWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(SubjectToReplayWorkflow.class); + WorkflowClient.start(workflow::run); + + workflow.query(); + workflow.update(); + workflow.finish(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + Map expected = new ConcurrentHashMap<>(); + // 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); + // 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.isSubjectToReplay()); + } + + @WorkflowInterface + public interface SubjectToReplayWorkflow { + @WorkflowMethod + void run(); + + @QueryMethod + boolean query(); + + @UpdateMethod + void update(); + + @UpdateValidatorMethod(updateName = "update") + void validateUpdate(); + + @SignalMethod + void finish(); + } + + 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"); + 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 SubjectToReplayRecordingInterceptor 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();