diff --git a/CHANGELOG.md b/CHANGELOG.md index 4019816163..0e3941f15d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -69,6 +69,8 @@ to docs, or any other relevant information. - `TemporalDevServer`, `TemporalDevServerOptions`, `TestWorkflowEnvironment.startLocal`, and the `useDevServer` methods of `TestWorkflowRule` and `TestWorkflowExtension` are no longer marked experimental. `TemporalDevServerOptions` remain specific to the Temporal CLI dev server implementation and may no longer be valid if that implementation changes. +- With external storage, large activity arguments are no longer downloaded twice. Workflow workers now skip + downloading the inputs they sent to activities, child workflows, signals and Nexus operations. ### Fixed - `ActivityClient` now sends the `temporal-namespace` gRPC header like `WorkflowClient` and `ScheduleClient` do, diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageRunner.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageRunner.java index 295bb202b0..f265c24eb2 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageRunner.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageRunner.java @@ -3,6 +3,8 @@ import com.google.common.base.Throwables; import com.google.protobuf.Message; import io.temporal.api.common.v1.Payload; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.history.v1.HistoryEventOrBuilder; import io.temporal.api.sdk.v1.ExternalStorageReference; import io.temporal.common.CancellationToken; import io.temporal.internal.payload.visitor.MessageVisitor; @@ -11,6 +13,8 @@ import io.temporal.payload.storage.ExternalStorage; import io.temporal.payload.storage.StorageDriver; import io.temporal.payload.storage.StorageDriverTargetInfo; +import java.util.EnumSet; +import java.util.Set; import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; @@ -24,6 +28,26 @@ * configure external storage. */ public final class ExternalStorageRunner { + /** + * Events that echo the inputs of a command the workflow issued. Replay matches these events by + * identifiers only, so their payloads never need to be downloaded by the workflow worker. If the + * workflow worker ever needs to read one of these payloads, remove its event type from this set. + */ + private static final Set COMMAND_INPUT_EVENTS = + EnumSet.of( + EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED, + EventType.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED, + EventType.EVENT_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED, + EventType.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED); + + /** Marks payloads inside {@link #COMMAND_INPUT_EVENTS} so that retrieval leaves them alone. */ + private static final MessageVisitor SKIP_COMMAND_INPUTS = + (skip, message) -> + Boolean.TRUE.equals(skip) + || (message instanceof HistoryEventOrBuilder + && COMMAND_INPUT_EVENTS.contains( + ((HistoryEventOrBuilder) message).getEventType())); + private final ExternalStoragePayloadTransformer payloadTransformer; private final int payloadVisitConcurrency; @@ -92,7 +116,23 @@ public CompletableFuture retrieveAsync( T message, CancellationToken cancellationToken, @Nullable StorageOperationMetrics metrics) { - return PayloadVisitors.visit(message, retrieveOptions(cancellationToken, metrics)); + return PayloadVisitors.visit(message, retrieveOptions(null, cancellationToken, metrics)); + } + + /** + * Like {@link #retrieve}, for workflow history processed by the workflow worker. References in + * events that echo the workflow's own command inputs are left in place, because replay does not + * read them. The workflow's data converter does not download references, so code that reads these + * payloads would see the reference instead of the data. + */ + public T retrieveForReplay( + T message, + CancellationToken cancellationToken, + @Nullable StorageOperationMetrics metrics) { + return getOrThrowIfCancelled( + PayloadVisitors.visit( + message, retrieveOptions(SKIP_COMMAND_INPUTS, cancellationToken, metrics)), + cancellationToken); } /** @@ -157,12 +197,16 @@ private PayloadVisitorOptions storeOptions( .build(); } - private PayloadVisitorOptions retrieveOptions( + private PayloadVisitorOptions retrieveOptions( + @Nullable MessageVisitor skipVisitor, CancellationToken cancellationToken, @Nullable StorageOperationMetrics metrics) { - return PayloadVisitorOptions.newBuilder( - (context, payloads) -> - payloadTransformer.retrieve(payloads, cancellationToken, metrics)) + return PayloadVisitorOptions.newBuilder( + (skip, payloads) -> + Boolean.TRUE.equals(skip) + ? CompletableFuture.completedFuture(payloads) + : payloadTransformer.retrieve(payloads, cancellationToken, metrics)) + .setMessageVisitor(skipVisitor) .setConcurrency(payloadVisitConcurrency) .setSkipSearchAttributes(true) .build(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java index 34da175096..ae449a4f82 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java @@ -196,7 +196,7 @@ private PollWorkflowTaskQueueResponse.Builder retrieveStoredPayloads( return workflowTask; } return externalStorageRunner - .retrieve(workflowTask.build(), options.getStorageCancellation(), downloadMetrics) + .retrieveForReplay(workflowTask.build(), options.getStorageCancellation(), downloadMetrics) .toBuilder(); } @@ -435,7 +435,7 @@ private WorkflowRunTaskHandler createStatefulHandler( ExternalStorageRunner.throwIfContainsReference(getHistoryResponse); } else { getHistoryResponse = - externalStorageRunner.retrieve( + externalStorageRunner.retrieveForReplay( getHistoryResponse, options.getStorageCancellation(), downloadMetrics); } workflowTask diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ServiceWorkflowHistoryIterator.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ServiceWorkflowHistoryIterator.java index 733724db9a..6fe70693b4 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ServiceWorkflowHistoryIterator.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ServiceWorkflowHistoryIterator.java @@ -90,7 +90,8 @@ public boolean hasNext() { if (externalStorageRunner == null) { ExternalStorageRunner.throwIfContainsReference(history); } else { - history = externalStorageRunner.retrieve(history, storageCancellation, downloadMetrics); + history = + externalStorageRunner.retrieveForReplay(history, storageCancellation, downloadMetrics); } current = history.getEventsList().iterator(); nextPageToken = response.getNextPageToken(); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageRunnerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageRunnerTest.java index 323ef817fe..0a3d0229be 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageRunnerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageRunnerTest.java @@ -1,6 +1,7 @@ package io.temporal.internal.payload.storage; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; @@ -18,6 +19,15 @@ import io.temporal.api.common.v1.Payload; import io.temporal.api.common.v1.Payloads; import io.temporal.api.common.v1.SearchAttributes; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.history.v1.ActivityTaskCompletedEventAttributes; +import io.temporal.api.history.v1.ActivityTaskScheduledEventAttributes; +import io.temporal.api.history.v1.History; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.history.v1.NexusOperationScheduledEventAttributes; +import io.temporal.api.history.v1.SignalExternalWorkflowExecutionInitiatedEventAttributes; +import io.temporal.api.history.v1.StartChildWorkflowExecutionInitiatedEventAttributes; +import io.temporal.api.history.v1.WorkflowExecutionStartedEventAttributes; import io.temporal.api.sdk.v1.UserMetadata; import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest; import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest; @@ -117,6 +127,72 @@ public void walksNestedPayloads() throws Exception { assertEquals(command, transformer.retrieve(stored, CancellationToken.none())); } + @Test + public void retrieveForReplayLeavesCommandInputEventsAsReferences() throws Exception { + TestStorageDriver driver = TestStorageDriver.named("d1"); + ExternalStorageRunner transformer = transformer(driver, 0); + Payloads input = Payloads.newBuilder().addPayloads(payload("x")).build(); + History history = + History.newBuilder() + .addEvents( + HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) + .setWorkflowExecutionStartedEventAttributes( + WorkflowExecutionStartedEventAttributes.newBuilder().setInput(input))) + .addEvents( + HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED) + .setActivityTaskScheduledEventAttributes( + ActivityTaskScheduledEventAttributes.newBuilder().setInput(input))) + .addEvents( + HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED) + .setStartChildWorkflowExecutionInitiatedEventAttributes( + StartChildWorkflowExecutionInitiatedEventAttributes.newBuilder() + .setInput(input))) + .addEvents( + HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED) + .setSignalExternalWorkflowExecutionInitiatedEventAttributes( + SignalExternalWorkflowExecutionInitiatedEventAttributes.newBuilder() + .setInput(input))) + .addEvents( + HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED) + .setNexusOperationScheduledEventAttributes( + NexusOperationScheduledEventAttributes.newBuilder().setInput(payload("x")))) + .addEvents( + HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED) + .setActivityTaskCompletedEventAttributes( + ActivityTaskCompletedEventAttributes.newBuilder().setResult(input))) + .build(); + History.Builder builder = history.toBuilder(); + transformer.store(builder, null, null, CancellationToken.none()); + History stored = builder.build(); + + History replayed = transformer.retrieveForReplay(stored, CancellationToken.none(), null); + + assertEquals( + "the started event's input is needed by replay", + history.getEvents(0), + replayed.getEvents(0)); + for (int i = 1; i <= 4; i++) { + assertEquals( + "a command's echoed input must be left as stored", + stored.getEvents(i), + replayed.getEvents(i)); + assertTrue(containsReference(replayed.getEvents(i))); + } + assertEquals( + "the activity result is needed by replay", history.getEvents(5), replayed.getEvents(5)); + assertFalse(containsReference(replayed.getEvents(5))); + assertEquals( + "plain retrieval still resolves everything", + history, + transformer.retrieve(stored, CancellationToken.none())); + } + @Test public void payloadBelowThresholdLeavesMessageUnchanged() throws Exception { TestStorageDriver driver = TestStorageDriver.named("d1"); @@ -284,6 +360,15 @@ private static ExternalStorageRunner transformer(StorageDriver driver, int thres return new ExternalStorageRunner(payloadTransformer, 4); } + private static boolean containsReference(HistoryEvent event) { + try { + ExternalStorageRunner.throwIfContainsReference(event); + return false; + } catch (ExternalStorageNotConfiguredException e) { + return true; + } + } + private static Payload payload(String data) { return Payload.newBuilder().setData(ByteString.copyFromUtf8(data)).build(); } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerExternalStorageTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerExternalStorageTest.java index def7641728..dd1ea3a585 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerExternalStorageTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerExternalStorageTest.java @@ -1,5 +1,8 @@ package io.temporal.internal.worker; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityOptions; +import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowClientOptions; import io.temporal.client.WorkflowOptions; import io.temporal.internal.payload.storage.TestStorageDriver; @@ -7,10 +10,14 @@ import io.temporal.testing.CloudTestExclusion.NeedsCloudAdaptation; import io.temporal.testing.CloudTestExclusionNote; import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.QueryMethod; +import io.temporal.workflow.SignalMethod; import io.temporal.workflow.Workflow; import io.temporal.workflow.WorkflowInterface; import io.temporal.workflow.WorkflowMethod; import io.temporal.workflow.shared.TestWorkflows; +import java.time.Duration; +import java.util.Collections; import java.util.UUID; import org.junit.Assert; import org.junit.Before; @@ -26,6 +33,10 @@ public class WorkflowWorkerExternalStorageTest { private static final int THRESHOLD = 4096; private static final String CONTINUED_MARKER = "continued:"; + private static final String ACTIVITY_ARG_MARKER = "activity-arg:"; + private static final String ACTIVITY_RESULT_MARKER = "activity-result:"; + private static final String CHILD_INPUT_MARKER = "child-input:"; + private static final String SIGNAL_INPUT_MARKER = "signal-input:"; private static final TestStorageDriver driver = TestStorageDriver.named("wf-happy"); @@ -35,7 +46,14 @@ public class WorkflowWorkerExternalStorageTest { @Rule public SDKTestWorkflowRule testWorkflowRule = SDKTestWorkflowRule.newBuilder() - .setWorkflowTypes(ContinueAsNewWorkflowImpl.class, EchoWorkflowImpl.class) + .setWorkflowTypes( + ContinueAsNewWorkflowImpl.class, + EchoWorkflowImpl.class, + LargeActivityArgWorkflowImpl.class, + ParentWorkflowImpl.class, + LengthWorkflowImpl.class, + SignalTargetWorkflowImpl.class) + .setActivityImplementations(new TransformActivityImpl()) .setWorkflowClientOptions( WorkflowClientOptions.newBuilder().setExternalStorage(storage).build()) .build(); @@ -89,6 +107,87 @@ public void aResultUnderTheThresholdIsLeftInlineAndReadableByTheClient() { Assert.assertEquals("nothing under the threshold should be offloaded", 0, driver.stores.get()); } + @Test + public void aLargeActivityArgumentIsDownloadedOnlyByTheActivityWorker() { + LargeActivityArgWorkflow workflow = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + LargeActivityArgWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setWorkflowId("extstore-activity-arg-" + UUID.randomUUID()) + .build()); + + // The workflow returns the length of the large activity result, which it must download. + int expectedLength = ACTIVITY_RESULT_MARKER.length() + THRESHOLD * 2; + Assert.assertEquals("len:" + expectedLength, workflow.execute()); + // A query on the closed workflow replays its full history. + Assert.assertEquals("len:" + expectedLength, workflow.lastResult()); + + Assert.assertEquals( + "only the activity worker should download the activity argument", + 1, + downloads(ACTIVITY_ARG_MARKER)); + Assert.assertTrue( + "the workflow worker must still download the activity result", + downloads(ACTIVITY_RESULT_MARKER) > 0); + } + + @Test + public void largeChildAndSignalInputsAreDownloadedOnlyByTheirReceivers() { + WorkflowClient client = testWorkflowRule.getWorkflowClient(); + String targetId = "extstore-signal-target-" + UUID.randomUUID(); + SignalTargetWorkflow target = + client.newWorkflowStub( + SignalTargetWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setWorkflowId(targetId) + .build()); + WorkflowClient.start(target::waitForSignal); + ParentWorkflow parent = + client.newWorkflowStub( + ParentWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setWorkflowId("extstore-parent-" + UUID.randomUUID()) + .build()); + + String childResult = "len:" + large(CHILD_INPUT_MARKER).length(); + Assert.assertEquals(childResult, parent.execute(targetId)); + Assert.assertEquals( + "len:" + large(SIGNAL_INPUT_MARKER).length(), + client.newUntypedWorkflowStub(targetId).getResult(String.class)); + // A query on the closed workflow replays its full history. + Assert.assertEquals(childResult, parent.lastResult()); + + Assert.assertEquals( + "only the child workflow should download its input", 1, downloads(CHILD_INPUT_MARKER)); + Assert.assertEquals( + "only the signaled workflow should download the signal input", + 1, + downloads(SIGNAL_INPUT_MARKER)); + } + + private static String large(String marker) { + StringBuilder large = new StringBuilder(marker); + for (int i = 0; i < THRESHOLD * 2; i++) { + large.append('x'); + } + return large.toString(); + } + + /** How many times the stored payload containing {@code marker} was downloaded. */ + private static int downloads(String marker) { + for (int i = 0; i < driver.storedData.size(); i++) { + if (driver.storedData.get(i).contains(marker)) { + return Collections.frequency(driver.retrievedKeys, "wf-happy-" + i); + } + } + throw new AssertionError("no payload containing " + marker + " was offloaded"); + } + public static class ContinueAsNewWorkflowImpl implements TestWorkflows.TestWorkflow1 { @Override public String execute(String input) { @@ -116,4 +215,114 @@ public String echo(String input) { return "echo: " + input; } } + + @WorkflowInterface + public interface LargeActivityArgWorkflow { + @WorkflowMethod + String execute(); + + @QueryMethod + String lastResult(); + } + + public static class LargeActivityArgWorkflowImpl implements LargeActivityArgWorkflow { + private String lastResult; + + @Override + public String execute() { + StringBuilder large = new StringBuilder(ACTIVITY_ARG_MARKER); + for (int i = 0; i < THRESHOLD * 2; i++) { + large.append('x'); + } + TransformActivity activity = + Workflow.newActivityStub( + TransformActivity.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + String result = activity.transform(large.toString()); + lastResult = + result.startsWith(ACTIVITY_RESULT_MARKER) ? "len:" + result.length() : "wrong result"; + return lastResult; + } + + @Override + public String lastResult() { + return lastResult; + } + } + + @ActivityInterface + public interface TransformActivity { + String transform(String input); + } + + public static class TransformActivityImpl implements TransformActivity { + @Override + public String transform(String input) { + return input.replace(ACTIVITY_ARG_MARKER, ACTIVITY_RESULT_MARKER); + } + } + + @WorkflowInterface + public interface ParentWorkflow { + @WorkflowMethod + String execute(String signalTargetId); + + @QueryMethod + String lastResult(); + } + + public static class ParentWorkflowImpl implements ParentWorkflow { + private String lastResult; + + @Override + public String execute(String signalTargetId) { + lastResult = + Workflow.newChildWorkflowStub(LengthWorkflow.class).length(large(CHILD_INPUT_MARKER)); + Workflow.newExternalWorkflowStub(SignalTargetWorkflow.class, signalTargetId) + .signal(large(SIGNAL_INPUT_MARKER)); + return lastResult; + } + + @Override + public String lastResult() { + return lastResult; + } + } + + @WorkflowInterface + public interface LengthWorkflow { + @WorkflowMethod + String length(String input); + } + + public static class LengthWorkflowImpl implements LengthWorkflow { + @Override + public String length(String input) { + return "len:" + input.length(); + } + } + + @WorkflowInterface + public interface SignalTargetWorkflow { + @WorkflowMethod + String waitForSignal(); + + @SignalMethod + void signal(String input); + } + + public static class SignalTargetWorkflowImpl implements SignalTargetWorkflow { + private String input; + + @Override + public String waitForSignal() { + Workflow.await(() -> input != null); + return "len:" + input.length(); + } + + @Override + public void signal(String input) { + this.input = input; + } + } }