diff --git a/CHANGELOG.md b/CHANGELOG.md index 5474179818..00493c2f93 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,10 @@ to docs, or any other relevant information. Child workflow overrides and one-time routing require Temporal Server 1.32.0 or later. - `WorkerFactoryOptions.Builder.setLoggerTagPrefix` that can be used to customized structured logging tags (MDC keys) set by Temporal SDK in worker context. +- Added experimental `Workflow.getRandomStream(String name)`, which returns a deterministic pseudorandom stream private + to the given name. +- Added experimental `WorkflowUnsafe.isSubjectToReplay()`, which reports whether the calling code is re-executed on + replay. ### Changed - Release notes for all future releases are now in a single CHANGELOG.md file. `releases` directory with old release 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 de19d303f0..95f73cd29c 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 @@ -293,6 +293,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 f2396270ce..3c26e86c25 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 @@ -82,6 +82,11 @@ public Random newRandom() { return workflowStateMachines.newRandom(); } + @Override + public Random getRandomStream(String name) { + return workflowStateMachines.getRandomStream(name); + } + @Override public Scope getMetricsScope() { return replayAwareWorkflowMetricsScope; @@ -351,9 +356,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 893dc3fe5a..cfbfac50a4 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 99ffbcbd0c..791ab1d058 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 @@ -159,6 +159,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 @@ -174,6 +175,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 e0b28a77e0..e209434e66 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; @@ -1094,10 +1095,12 @@ public R sideEffect( () -> { try { readOnly = true; + subjectToReplay = false; R r = func.apply(); return dataConverterWithCurrentWorkflowContext.toPayloads(r); } finally { readOnly = false; + subjectToReplay = true; } }, userMetadata, @@ -1153,13 +1156,14 @@ private R mutableSideEffectImpl( id, metadata, (storedBinary) -> { - Optional stored = - storedBinary.map( - (b) -> - dataConverterWithCurrentWorkflowContext.fromPayloads( - 0, Optional.of(b), resultClass, resultType)); try { readOnly = true; + subjectToReplay = false; + Optional stored = + storedBinary.map( + (b) -> + dataConverterWithCurrentWorkflowContext.fromPayloads( + 0, Optional.of(b), resultClass, resultType)); R funcResult = Objects.requireNonNull( func.apply(), "mutableSideEffect function " + "returned null"); @@ -1170,6 +1174,7 @@ private R mutableSideEffectImpl( return Optional.empty(); // returned only when value doesn't need to be updated } finally { readOnly = false; + subjectToReplay = true; } }, (p) -> @@ -1312,6 +1317,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 0a1c982f07..7d27df3caa 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 @@ -735,6 +735,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()); } @@ -761,6 +772,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( @@ -946,6 +961,11 @@ static void assertNotReadOnly(String action) { } } + public static boolean isSubjectToReplay() { + Optional thread = DeterministicRunnerImpl.currentThreadInternalIfPresent(); + return thread.isPresent() && 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 40ecad495c..2de5a3b94a 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java +++ b/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java @@ -1003,6 +1003,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..0103531117 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,23 @@ 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. + * + * @return true if the calling context is re-executed on replay. This method always returns false + * if called from a non workflow thread. + */ + @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/MutableSideEffectDecodeSubjectToReplayTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/MutableSideEffectDecodeSubjectToReplayTest.java new file mode 100644 index 0000000000..3ed69efdaa --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/MutableSideEffectDecodeSubjectToReplayTest.java @@ -0,0 +1,62 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; + +import io.temporal.api.common.v1.Payload; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.JacksonJsonPayloadConverter; +import io.temporal.testing.CloudTestExclusion.NeedsCloudAdaptation; +import io.temporal.testing.CloudTestExclusionNote; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.shared.TestWorkflows.NoArgsWorkflow; +import io.temporal.workflow.unsafe.WorkflowUnsafe; +import java.lang.reflect.Type; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +@CloudTestExclusionNote("Cloud envconfig replaces the custom data converter.") +@Category(NeedsCloudAdaptation.class) +public class MutableSideEffectDecodeSubjectToReplayTest { + private static final AtomicReference subjectToReplayWhileDecoding = + new AtomicReference<>(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(TestWorkflowImpl.class) + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setDataConverter( + DefaultDataConverter.newDefaultInstance() + .withPayloadConverterOverrides(new StoredValueRecordingConverter())) + .build()) + .build(); + + /** The stored value is decoded only when an existing marker is updated live. */ + @Test + public void storedValueIsDecodedOutsideReplay() { + testWorkflowRule.newWorkflowStubTimeoutOptions(NoArgsWorkflow.class).execute(); + assertEquals(Boolean.FALSE, subjectToReplayWhileDecoding.get()); + } + + public static class TestWorkflowImpl implements NoArgsWorkflow { + @Override + public void execute() { + Workflow.mutableSideEffect("id", Integer.class, (stored, current) -> true, () -> 1); + Workflow.mutableSideEffect("id", Integer.class, (stored, current) -> true, () -> 2); + } + } + + private static class StoredValueRecordingConverter extends JacksonJsonPayloadConverter { + @Override + public T fromData(Payload content, Class valueClass, Type valueType) { + if (valueClass == Integer.class) { + subjectToReplayWhileDecoding.set(WorkflowUnsafe.isSubjectToReplay()); + } + return super.fromData(content, valueClass, valueType); + } + } +} 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..549d6e242c --- /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(client.getOptions().getNamespace()) + .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..473747f4d5 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/WorkflowUnsafeSubjectToReplayTest.java @@ -0,0 +1,196 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; + +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); + } + + @Test + public void isSubjectToReplayOffWorkflowThread() { + assertFalse(WorkflowUnsafe.isSubjectToReplay()); + } + + 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 8ca8ba9d1c..75a53e4e55 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 @@ -290,6 +290,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();