From 9da4039a7fb124b0bedea7008f51983f52a1b33b Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 30 Oct 2026 14:34:17 -0700 Subject: [PATCH 1/4] Skip payload downloads the workflow worker never uses The workflow worker downloaded stored inputs of activities, child workflows, signals and Nexus operations that the workflow itself started, even though it never reads them. Each large activity argument was downloaded twice: once by the workflow worker and once by the activity worker. Leave those payloads as references when the workflow worker loads history. Everything else is downloaded as before. --- CHANGELOG.md | 2 + .../storage/ExternalStorageRunner.java | 52 +++++++++-- .../replay/ReplayWorkflowTaskHandler.java | 4 +- .../ServiceWorkflowHistoryIterator.java | 3 +- .../storage/ExternalStorageRunnerTest.java | 85 ++++++++++++++++++ .../WorkflowWorkerExternalStorageTest.java | 86 ++++++++++++++++++- 6 files changed, 223 insertions(+), 9 deletions(-) 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..23c376be7a 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,25 @@ * 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. + */ + 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 +115,22 @@ 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. If anything does decode one later, the data converter retrieves it on demand. + */ + public T retrieveForReplay( + T message, + CancellationToken cancellationToken, + @Nullable StorageOperationMetrics metrics) { + return getOrThrowIfCancelled( + PayloadVisitors.visit( + message, retrieveOptions(SKIP_COMMAND_INPUTS, cancellationToken, metrics)), + cancellationToken); } /** @@ -157,12 +195,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..26496f82ff 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,7 @@ package io.temporal.internal.worker; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityOptions; import io.temporal.client.WorkflowClientOptions; import io.temporal.client.WorkflowOptions; import io.temporal.internal.payload.storage.TestStorageDriver; @@ -7,10 +9,13 @@ 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.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 +31,7 @@ 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 TestStorageDriver driver = TestStorageDriver.named("wf-happy"); @@ -35,7 +41,11 @@ public class WorkflowWorkerExternalStorageTest { @Rule public SDKTestWorkflowRule testWorkflowRule = SDKTestWorkflowRule.newBuilder() - .setWorkflowTypes(ContinueAsNewWorkflowImpl.class, EchoWorkflowImpl.class) + .setWorkflowTypes( + ContinueAsNewWorkflowImpl.class, + EchoWorkflowImpl.class, + LargeActivityArgWorkflowImpl.class) + .setActivityImplementations(new LengthActivityImpl()) .setWorkflowClientOptions( WorkflowClientOptions.newBuilder().setExternalStorage(storage).build()) .build(); @@ -89,6 +99,36 @@ 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()); + + int expectedLength = ACTIVITY_ARG_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()); + + String argumentKey = null; + for (int i = 0; i < driver.storedData.size(); i++) { + if (driver.storedData.get(i).contains(ACTIVITY_ARG_MARKER)) { + argumentKey = "wf-happy-" + i; + } + } + Assert.assertNotNull("the activity argument must have been offloaded", argumentKey); + Assert.assertEquals( + "only the activity worker should download the activity argument", + 1, + Collections.frequency(driver.retrievedKeys, argumentKey)); + } + public static class ContinueAsNewWorkflowImpl implements TestWorkflows.TestWorkflow1 { @Override public String execute(String input) { @@ -116,4 +156,48 @@ 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'); + } + LengthActivity activity = + Workflow.newActivityStub( + LengthActivity.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + lastResult = activity.length(large.toString()); + return lastResult; + } + + @Override + public String lastResult() { + return lastResult; + } + } + + @ActivityInterface + public interface LengthActivity { + String length(String input); + } + + public static class LengthActivityImpl implements LengthActivity { + @Override + public String length(String input) { + return "len:" + input.length(); + } + } } From 87ef8aa7f4d0fc5bf909c2d1f9632154f39b0b8f Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 30 Oct 2026 14:51:05 -0700 Subject: [PATCH 2/4] Correct the retrieveForReplay Javadoc The workflow's data converter does not download references, so a skipped payload is not fetched on demand if something reads it later. Say so, and note where to change the list of skipped events. --- .../internal/payload/storage/ExternalStorageRunner.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 23c376be7a..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 @@ -30,7 +30,8 @@ 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. + * 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( @@ -121,7 +122,8 @@ public CompletableFuture retrieveAsync( /** * 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. If anything does decode one later, the data converter retrieves it on demand. + * 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, From ebae18485d93609ea4605d3cc3cf55d49f94a2c0 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 30 Oct 2026 14:51:56 -0700 Subject: [PATCH 3/4] Check that large activity results still reach the workflow Make the test activity return a large result, so the test fails if the workflow worker skips downloads it needs. --- .../WorkflowWorkerExternalStorageTest.java | 46 +++++++++++-------- 1 file changed, 28 insertions(+), 18 deletions(-) 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 26496f82ff..ee054e9da2 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 @@ -32,6 +32,7 @@ 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 TestStorageDriver driver = TestStorageDriver.named("wf-happy"); @@ -45,7 +46,7 @@ public class WorkflowWorkerExternalStorageTest { ContinueAsNewWorkflowImpl.class, EchoWorkflowImpl.class, LargeActivityArgWorkflowImpl.class) - .setActivityImplementations(new LengthActivityImpl()) + .setActivityImplementations(new TransformActivityImpl()) .setWorkflowClientOptions( WorkflowClientOptions.newBuilder().setExternalStorage(storage).build()) .build(); @@ -111,22 +112,29 @@ public void aLargeActivityArgumentIsDownloadedOnlyByTheActivityWorker() { .setWorkflowId("extstore-activity-arg-" + UUID.randomUUID()) .build()); - int expectedLength = ACTIVITY_ARG_MARKER.length() + THRESHOLD * 2; + // 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()); - String argumentKey = null; - for (int i = 0; i < driver.storedData.size(); i++) { - if (driver.storedData.get(i).contains(ACTIVITY_ARG_MARKER)) { - argumentKey = "wf-happy-" + i; - } - } - Assert.assertNotNull("the activity argument must have been offloaded", argumentKey); Assert.assertEquals( "only the activity worker should download the activity argument", 1, - Collections.frequency(driver.retrievedKeys, argumentKey)); + downloads(ACTIVITY_ARG_MARKER)); + Assert.assertTrue( + "the workflow worker must still download the activity result", + downloads(ACTIVITY_RESULT_MARKER) > 0); + } + + /** 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 { @@ -175,11 +183,13 @@ public String execute() { for (int i = 0; i < THRESHOLD * 2; i++) { large.append('x'); } - LengthActivity activity = + TransformActivity activity = Workflow.newActivityStub( - LengthActivity.class, + TransformActivity.class, ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); - lastResult = activity.length(large.toString()); + String result = activity.transform(large.toString()); + lastResult = + result.startsWith(ACTIVITY_RESULT_MARKER) ? "len:" + result.length() : "wrong result"; return lastResult; } @@ -190,14 +200,14 @@ public String lastResult() { } @ActivityInterface - public interface LengthActivity { - String length(String input); + public interface TransformActivity { + String transform(String input); } - public static class LengthActivityImpl implements LengthActivity { + public static class TransformActivityImpl implements TransformActivity { @Override - public String length(String input) { - return "len:" + input.length(); + public String transform(String input) { + return input.replace(ACTIVITY_ARG_MARKER, ACTIVITY_RESULT_MARKER); } } } From de42939564ec3465061f0cff3e25ff07e49c6620 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 30 Oct 2026 14:52:42 -0700 Subject: [PATCH 4/4] Test that child workflow and signal inputs are downloaded once A parent starts a child with a large input and signals another workflow with a large input. Each input must be downloaded only by the workflow that receives it, including when the parent's history is replayed. --- .../WorkflowWorkerExternalStorageTest.java | 117 +++++++++++++++++- 1 file changed, 116 insertions(+), 1 deletion(-) 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 ee054e9da2..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 @@ -2,6 +2,7 @@ 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; @@ -10,6 +11,7 @@ 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; @@ -33,6 +35,8 @@ public class WorkflowWorkerExternalStorageTest { 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"); @@ -45,7 +49,10 @@ public class WorkflowWorkerExternalStorageTest { .setWorkflowTypes( ContinueAsNewWorkflowImpl.class, EchoWorkflowImpl.class, - LargeActivityArgWorkflowImpl.class) + LargeActivityArgWorkflowImpl.class, + ParentWorkflowImpl.class, + LengthWorkflowImpl.class, + SignalTargetWorkflowImpl.class) .setActivityImplementations(new TransformActivityImpl()) .setWorkflowClientOptions( WorkflowClientOptions.newBuilder().setExternalStorage(storage).build()) @@ -127,6 +134,50 @@ public void aLargeActivityArgumentIsDownloadedOnlyByTheActivityWorker() { 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++) { @@ -210,4 +261,68 @@ 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; + } + } }