diff --git a/CHANGELOG.md b/CHANGELOG.md index dbc9d1e623..01ba0e8085 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -69,6 +69,7 @@ to docs, or any other relevant information. specific to the Temporal CLI dev server implementation and may no longer be valid if that implementation changes. ### Fixed +- Workflow Streams subscriptions now deliver retained items after workflow completion, failure, cancellation, termination, or timeout using a query pinned to the closed run. - `ActivityClient` now sends the `temporal-namespace` gRPC header like `WorkflowClient` and `ScheduleClient` do, so standalone activity calls are no longer rejected with `PERMISSION_DENIED` by namespace-scoped auth (e.g. Temporal Cloud API keys). diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md index ad630c89b7..4e49f5a20a 100644 --- a/contrib/temporal-workflowstreams/README.md +++ b/contrib/temporal-workflowstreams/README.md @@ -4,7 +4,8 @@ A durable publish/subscribe log hosted inside a Temporal Workflow. External code (activities, starters, other processes) publishes messages to named topics via **signals**; subscribers long-poll for new items via -**updates**; a **query** exposes the current offset. The stream is backed by +**updates**, then read retained items after closure via **queries**; a separate +**query** exposes the current offset. The stream is backed by Temporal's durable execution, giving ordered, durable, exactly-once delivery with client-side batching, publisher dedup, continue-as-new survival, truncation, and ~1 MB response paging. @@ -18,7 +19,7 @@ All APIs in this module are experimental and may change. ## Workflow side Construct a `WorkflowStream` once in a `@WorkflowInit` constructor. The factory -registers the publish signal, poll update, and offset query handlers, and a +registers the publish signal, poll update, poll query, and offset query handlers. A `@WorkflowInit` constructor runs before any handler dispatch, so polls and offset queries arriving with the first workflow task (e.g. from update-with-start) are accepted rather than rejected. @@ -114,6 +115,18 @@ when the buffer reaches the max batch size, on `forceFlush`, on an explicit ## Subscribing +Subscriptions keep polling through updates while the workflow runs. After a +confirmed terminal state, they query that run for its remaining retained items +and finish after delivering the final page. The poll query shares +`__temporal_workflow_stream_poll` with the update, returns immediately, and reads +from the retained base when the requested offset was truncated. Continue-as-new +still follows the successor run. + +Upgrade workers before using upgraded clients. Closed-stream reads require +retained history and an available compatible worker; configured query rejection +and query failures are surfaced to the subscriber. Query reads do not add history +events. + There are two subscriber APIs over one shared poll engine: a non-blocking listener and a blocking iterator. Neither occupies a thread while a poll is blocked on the server — polling runs on a small executor shared by all of a @@ -206,7 +219,7 @@ the per-item payload encoding (base64 of the serialized `temporal.api.common.v1.Payload`) match other languages' packages exactly, so a Java publisher or subscriber interoperates with a workflow written in any of them and vice versa. The data converter codec chain -(encryption, compression) runs once on the signal/update envelope — never per +(encryption, compression) runs once on the signal/update/query envelope — never per item — so payloads are not double-encoded. One Java-specific caveat: the protocol envelope types are serialized by the diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollInput.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollInput.java index 373133047f..9504537921 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollInput.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollInput.java @@ -6,7 +6,7 @@ import java.util.List; /** - * The poll update payload: a request to long-poll for new items. + * The poll update or query payload: a request for items at a global offset. * *

Field names are part of the cross-language wire protocol; this type must serialize to JSON * with exactly these names. diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollResult.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollResult.java index 5a1c0f0e56..a3137cdb03 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollResult.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollResult.java @@ -6,8 +6,8 @@ import java.util.List; /** - * The poll update response: items matching the poll request. When {@code more_ready} is true the - * response was truncated to stay within size limits and the subscriber should poll again + * The poll update or query response: items matching the poll request. When {@code more_ready} is + * true the response was truncated to stay within size limits and the subscriber should poll again * immediately rather than applying a cooldown. * *

Field names are part of the cross-language wire protocol; this type must serialize to JSON diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java index 32bfbc64bf..1624171fb3 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java @@ -20,10 +20,10 @@ /** * The workflow-side stream object: an append-only, multi-topic log served to external publishers - * (via signal), subscribers (via update), and offset queries (via query). + * (via signal) and subscribers (via update while running and query after closure). * *

Construct it once with {@link #newInstance}, preferably in a {@link - * io.temporal.workflow.WorkflowInit} constructor: the factory registers all three handlers on the + * io.temporal.workflow.WorkflowInit} constructor: the factory registers all stream handlers on the * current workflow, and a {@code @WorkflowInit} constructor runs before any handler dispatch, so * polls arriving with the first workflow task are accepted. Constructing it at the start of the * workflow method also works — signals received earlier are buffered by the SDK — but polls and @@ -274,6 +274,15 @@ public PollResult poll(PollInput input) { WorkflowStreamConstants.ERROR_TYPE_TRUNCATED_OFFSET); } + return collectBatch(input); + } + + @Override + public PollResult pollQuery(PollInput input) { + return collectBatch(input); + } + + private PollResult collectBatch(PollInput input) { long logOffset = Math.max(input.fromOffset - baseOffset, 0); Set topicSet = diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java index b5348879d4..d2caf187ef 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java @@ -142,8 +142,9 @@ public long getOffset() { * *

The consuming thread blocks waiting for items; polling itself runs on the client's poll * executor. Each item carries the raw {@link io.temporal.api.common.v1.Payload}; decode it with - * your data converter. The subscription ends cleanly when the workflow reaches a terminal state, - * automatically follows continue-as-new chains, and also ends when this client is closed. + * your data converter. The subscription delivers the retained tail before ending when the + * workflow reaches a terminal state, automatically follows continue-as-new chains, and also ends + * when this client is closed. */ public WorkflowStreamSubscription subscribe(SubscribeOptions options) { return new WorkflowStreamSubscription(listener -> newSubscriptionDriver(options, listener)); @@ -155,9 +156,10 @@ public WorkflowStreamSubscription subscribe(SubscribeOptions options) { * which is never held while a poll is blocked on the server, so many subscriptions share a small * pool. Delivery starts immediately. * - *

The stream ends cleanly with {@link WorkflowStreamListener#onCompleted} when the workflow - * reaches a terminal state, automatically follows continue-as-new chains, and reports - * unrecoverable failures to {@link WorkflowStreamListener#onError}. Stop it early with {@link + *

After delivering retained items, the stream ends with {@link + * WorkflowStreamListener#onCompleted} when the workflow reaches a terminal state, automatically + * follows continue-as-new chains, and reports unrecoverable failures to {@link + * WorkflowStreamListener#onError}. Stop it early with {@link * WorkflowStreamSubscriptionHandle#close}; closing this client also stops it. */ public WorkflowStreamSubscriptionHandle subscribe( diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamConstants.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamConstants.java index ae18cd4a49..5d0fdfa000 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamConstants.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamConstants.java @@ -17,6 +17,9 @@ public final class WorkflowStreamConstants { /** Update subscribers send to long-poll for new items. */ public static final String POLL_UPDATE_NAME = "__temporal_workflow_stream_poll"; + /** Query subscribers use to read retained items after the workflow closes. */ + public static final String POLL_QUERY_NAME = "__temporal_workflow_stream_poll"; + /** Query that returns the current global offset. */ public static final String OFFSET_QUERY_NAME = "__temporal_workflow_stream_offset"; diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamHandlers.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamHandlers.java index ca09bb082c..16d9ad1381 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamHandlers.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamHandlers.java @@ -22,6 +22,9 @@ public interface WorkflowStreamHandlers { @UpdateMethod(name = WorkflowStreamConstants.POLL_UPDATE_NAME) PollResult poll(PollInput input); + @QueryMethod(name = WorkflowStreamConstants.POLL_QUERY_NAME) + PollResult pollQuery(PollInput input); + @UpdateValidatorMethod(updateName = WorkflowStreamConstants.POLL_UPDATE_NAME) void validatePoll(PollInput input); diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java index fba824574f..b91e345e11 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java @@ -3,6 +3,7 @@ import io.temporal.api.enums.v1.WorkflowExecutionStatus; import io.temporal.client.UpdateOptions; import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowExecutionDescription; import io.temporal.client.WorkflowStub; import io.temporal.client.WorkflowTargetOptions; import io.temporal.client.WorkflowUpdateHandle; @@ -63,6 +64,7 @@ public final class SubscriptionDriver implements WorkflowStreamSubscriptionHandl // update's outcome so that, if that run continues-as-new mid-poll (failing the outcome), we // still know which run to inspect to tell a rollover apart from a terminal end. private String polledRunId = ""; + private WorkflowStub terminalRunStub; private volatile boolean closed; private final AtomicBoolean started = new AtomicBoolean(); @@ -114,6 +116,16 @@ private void poll() { finishSilent(); return; } + if (terminalRunStub != null) { + // Query failures must surface directly, without reclassifying a closed run. + deliver( + terminalRunStub.query( + WorkflowStreamConstants.POLL_QUERY_NAME, + PollResult.class, + new PollInput(topics, offset))); + return; + } + polledRunId = ""; WorkflowUpdateHandle handle; try { // Wait only for ACCEPTED so startUpdate returns the handle (and its run id) as soon @@ -170,6 +182,8 @@ private void deliverFrom(List items, int start, boolean more if (i == items.size()) { if (moreReady) { hop(this::poll); + } else if (terminalRunStub != null) { + finishCompleted(); } else { schedule(this::poll, pollCooldownMs); } @@ -232,14 +246,21 @@ private void handleError(Throwable e) { // The workflow may have continued-as-new or completed between polls. Follow the chain, // exit cleanly on a terminal state, otherwise surface the error. describe() is a blocking // call, which is why error handling always runs on the executor. - WorkflowExecutionStatus status = describePolledRun(); + WorkflowExecutionDescription description = describePolledRun(); + WorkflowExecutionStatus status = description.getStatus(); if (status == WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW) { // Subsequent polls use the latest-run stub, addressing the successor automatically. hop(this::poll); return; } - if (isTerminal(status)) { - finishCompleted(); + if (isTerminal(status) && !description.getExecution().getRunId().isEmpty()) { + terminalRunStub = + client.newUntypedWorkflowStub( + WorkflowTargetOptions.newBuilder() + .setWorkflowId(workflowId) + .setRunId(description.getExecution().getRunId()) + .build()); + hop(this::poll); return; } finishError(e); @@ -252,23 +273,19 @@ private void handleError(Throwable e) { * the latest run automatically. A blank run id (no poll has been admitted yet) falls back to * describing the latest run. */ - private WorkflowExecutionStatus describePolledRun() { - try { - WorkflowStub stub; - if (polledRunId == null || polledRunId.isEmpty()) { - stub = latestRunStub; - } else { - stub = - client.newUntypedWorkflowStub( - WorkflowTargetOptions.newBuilder() - .setWorkflowId(workflowId) - .setRunId(polledRunId) - .build()); - } - return stub.describe().getStatus(); - } catch (RuntimeException e) { - return WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED; + private WorkflowExecutionDescription describePolledRun() { + WorkflowStub stub; + if (polledRunId == null || polledRunId.isEmpty()) { + stub = latestRunStub; + } else { + stub = + client.newUntypedWorkflowStub( + WorkflowTargetOptions.newBuilder() + .setWorkflowId(workflowId) + .setRunId(polledRunId) + .build()); } + return stub.describe(); } private void finishSilent() { diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java index 97ca6e7b23..29700c62e0 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java @@ -2,6 +2,7 @@ import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; import io.temporal.client.WorkflowStub; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DefaultDataConverter; @@ -176,6 +177,66 @@ public void testTerminalCallsOnCompleted() throws Exception { } } + @Test + public void testClosedStreamCompletionWaitsForDelivery() throws Exception { + WorkflowStub stub = startHostWorkflow(); + stub.signal("publishLocal", "evt", "tail"); + stub.signal("finish"); + stub.getResult(Void.class); + RecordingListener listener = new RecordingListener(); + CompletableFuture gate = new CompletableFuture<>(); + listener.gates.add(gate); + try (WorkflowStreamClient streamClient = newStreamClient(stub); + WorkflowStreamSubscriptionHandle handle = streamClient.subscribe(FAST_POLL, listener)) { + awaitItems(listener, 1); + Assert.assertEquals("tail", decode(listener.items.get(0))); + Assert.assertEquals(1, listener.completed.getCount()); + Assert.assertFalse(handle.getDoneFuture().isDone()); + gate.complete(null); + handle.getDoneFuture().get(TIMEOUT_MS, TimeUnit.MILLISECONDS); + Assert.assertEquals(0, listener.completed.getCount()); + Assert.assertNull(listener.error.get()); + } + } + + @Test + public void testClosedStreamPagesStayPinnedAfterWorkflowIdReuse() throws Exception { + WorkflowStub stub = startHostWorkflow(); + String value = new String(new char[400_000]).replace('\0', 'x'); + for (int i = 0; i < 3; i++) { + stub.signal("publishLocal", "evt", value); + } + stub.signal("finish"); + stub.getResult(Void.class); + RecordingListener listener = new RecordingListener(); + CompletableFuture gate = new CompletableFuture<>(); + listener.gates.add(gate); + try (WorkflowStreamClient streamClient = newStreamClient(stub); + WorkflowStreamSubscriptionHandle handle = streamClient.subscribe(FAST_POLL, listener)) { + awaitItems(listener, 1); + SubscribeHostWorkflow replacement = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + SubscribeHostWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId(stub.getExecution().getWorkflowId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .build()); + WorkflowClient.start(replacement::execute, null); + replacement.publishLocal("evt", "replacement"); + gate.complete(null); + handle.getDoneFuture().get(TIMEOUT_MS, TimeUnit.MILLISECONDS); + Assert.assertEquals(3, listener.items.size()); + for (int i = 0; i < 3; i++) { + Assert.assertEquals(value, decode(listener.items.get(i))); + Assert.assertEquals(i, listener.items.get(i).getOffset()); + } + replacement.finish(); + WorkflowStub.fromTyped(replacement).getResult(Void.class); + } + } + @Test public void testListenerFollowsContinueAsNew() throws Exception { WorkflowStub stub = startHostWorkflow(); diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/PayloadWireTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/PayloadWireTest.java index 8115a21bc8..4a86b0c928 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/PayloadWireTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/PayloadWireTest.java @@ -1,5 +1,6 @@ package io.temporal.workflowstreams; +import com.fasterxml.jackson.databind.ObjectMapper; import com.google.protobuf.ByteString; import io.temporal.api.common.v1.Payload; import io.temporal.common.converter.DataConverter; @@ -7,6 +8,7 @@ import io.temporal.workflowstreams.internal.PayloadWire; import java.nio.charset.StandardCharsets; import java.util.Base64; +import java.util.Collections; import org.junit.Assert; import org.junit.Test; @@ -65,6 +67,32 @@ public void testBinaryPayloadRoundTrip() { Assert.assertArrayEquals(original, b); } + @Test + public void testSharedClosedStreamPollVector() throws Exception { + String wire = "ChgKCGVuY29kaW5nEgxiaW5hcnkvcGxhaW4SA2FiYw=="; + Payload payload = + Payload.newBuilder() + .putMetadata("encoding", ByteString.copyFromUtf8("binary/plain")) + .setData(ByteString.copyFromUtf8("abc")) + .build(); + Assert.assertEquals(wire, PayloadWire.encode(payload)); + Assert.assertEquals(payload, PayloadWire.decode(wire)); + ObjectMapper mapper = new ObjectMapper(); + Assert.assertEquals( + mapper.readTree("{\"topics\":[\"events\"],\"from_offset\":3}"), + mapper.readTree( + mapper.writeValueAsString(new PollInput(Collections.singletonList("events"), 3)))); + Assert.assertEquals( + mapper.readTree( + "{\"items\":[{\"topic\":\"events\",\"data\":\"" + + wire + + "\",\"offset\":3}],\"next_offset\":4,\"more_ready\":false}"), + mapper.readTree( + mapper.writeValueAsString( + new PollResult( + Collections.singletonList(new WireItem("events", wire, 3)), 4, false)))); + } + @Test public void testWireSize() { Payload payload = diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java index 88c9270a7a..d3f74dec78 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java @@ -110,6 +110,69 @@ public void testSubscribeEndsCleanlyOnTerminal() { } } + @Test + public void testSubscribeAfterCompletionDrainsRetainedPages() { + WorkflowStub stub = startHostWorkflow(); + String value = new String(new char[400_000]).replace('\0', 'x'); + stub.signal("publishLocal", "evt", value); + stub.signal("publishLocal", "other", "filtered"); + stub.signal("publishLocal", "evt", value); + stub.signal("publishLocal", "evt", value); + stub.signal("finish"); + stub.getResult(Void.class); + int historySize = + testWorkflowRule + .getExecutionHistory(stub.getExecution().getWorkflowId()) + .getHistory() + .getEventsCount(); + testWorkflowRule.invalidateWorkflowCache(); + + try (WorkflowStreamClient streamClient = newStreamClient(stub); + WorkflowStreamSubscription subscription = + streamClient.subscribe(SubscribeOptions.newBuilder().setTopics("evt").build())) { + for (long expected : new long[] {0, 2, 3}) { + WorkflowStreamItem item = subscription.next(); + Assert.assertEquals(expected, item.getOffset()); + Assert.assertEquals(value, decode(item)); + } + Assert.assertFalse(subscription.hasNext()); + } + Assert.assertEquals( + historySize, + testWorkflowRule + .getExecutionHistory(stub.getExecution().getWorkflowId()) + .getHistory() + .getEventsCount()); + } + + @Test + public void testSubscribeAfterCompletionClampsTruncatedOffsetsAndFinishesEmptyReads() { + WorkflowStub stub = startHostWorkflow(); + stub.signal("publishLocal", "evt", "a"); + stub.signal("publishLocal", "evt", "b"); + stub.signal("publishLocal", "evt", "c"); + stub.update("truncate", Void.class, 2L); + stub.signal("finish"); + stub.getResult(Void.class); + + try (WorkflowStreamClient streamClient = newStreamClient(stub)) { + try (WorkflowStreamSubscription subscription = + streamClient.subscribe(SubscribeOptions.newBuilder().setFromOffset(1).build())) { + Assert.assertEquals("c", decode(subscription.next())); + Assert.assertFalse(subscription.hasNext()); + } + for (SubscribeOptions options : + new SubscribeOptions[] { + SubscribeOptions.newBuilder().setFromOffset(3).build(), + SubscribeOptions.newBuilder().setTopics("missing").build() + }) { + try (WorkflowStreamSubscription subscription = streamClient.subscribe(options)) { + Assert.assertFalse(subscription.hasNext()); + } + } + } + } + @Test public void testSubscribeFollowsContinueAsNew() { WorkflowStub stub = startHostWorkflow(); diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscriptionDriverTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscriptionDriverTest.java new file mode 100644 index 0000000000..a459762a8b --- /dev/null +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscriptionDriverTest.java @@ -0,0 +1,212 @@ +package io.temporal.workflowstreams; + +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.enums.v1.WorkflowExecutionStatus; +import io.temporal.api.workflow.v1.WorkflowExecutionInfo; +import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionResponse; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowExecutionDescription; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowTargetOptions; +import io.temporal.client.WorkflowUpdateHandle; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.workflowstreams.internal.PayloadWire; +import io.temporal.workflowstreams.internal.SubscriptionDriver; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.junit.Assert; +import org.junit.Test; + +public class SubscriptionDriverTest { + @Test + public void testEveryTerminalStatusDrainsPinnedPages() throws Exception { + for (WorkflowExecutionStatus status : + new WorkflowExecutionStatus[] { + WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED, + WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_FAILED, + WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CANCELED, + WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TERMINATED, + WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TIMED_OUT + }) { + Fixture fixture = new Fixture(status); + fixture.run(); + Assert.assertEquals(2, fixture.items.size()); + Assert.assertEquals(3, fixture.items.get(0).getOffset()); + Assert.assertEquals(4, fixture.items.get(1).getOffset()); + Assert.assertEquals(1, fixture.completed); + Assert.assertEquals(java.util.Arrays.asList("closed-run", "closed-run"), fixture.queryRuns); + Assert.assertEquals(java.util.Arrays.asList(3L, 4L), fixture.queryOffsets); + } + } + + @Test + public void testAcceptedPollDescribesItsOwnRun() throws Exception { + Fixture fixture = new Fixture(WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED); + fixture.acceptPoll = true; + fixture.run(); + Assert.assertEquals("accepted-run", fixture.describedRun); + Assert.assertEquals(java.util.Arrays.asList("closed-run", "closed-run"), fixture.queryRuns); + } + + @Test + public void testTerminalDescriptionRequiresRunId() throws Exception { + Fixture fixture = new Fixture(WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED); + fixture.terminalRunId = ""; + try { + fixture.run(); + Assert.fail("an unpinned description must preserve the poll failure"); + } catch (ExecutionException e) { + Assert.assertEquals("closed", e.getCause().getMessage()); + } + Assert.assertTrue(fixture.queryRuns.isEmpty()); + Assert.assertEquals(0, fixture.completed); + } + + @Test + public void testQueryAndDescribeFailuresPropagate() throws Exception { + for (boolean queryFails : new boolean[] {true, false}) { + Fixture fixture = new Fixture(WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED); + fixture.queryError = queryFails ? new IllegalStateException("query failed") : null; + fixture.describeError = queryFails ? null : new IllegalStateException("describe failed"); + try { + fixture.run(); + Assert.fail("failure must reach the subscriber"); + } catch (ExecutionException e) { + Assert.assertSame(queryFails ? fixture.queryError : fixture.describeError, e.getCause()); + } + Assert.assertEquals(0, fixture.completed); + } + } + + private static class Fixture { + final WorkflowExecutionStatus status; + final List items = new ArrayList<>(); + final List queryRuns = new ArrayList<>(); + final List queryOffsets = new ArrayList<>(); + int completed; + boolean acceptPoll; + String describedRun; + String terminalRunId = "closed-run"; + RuntimeException queryError; + RuntimeException describeError; + + Fixture(WorkflowExecutionStatus status) { + this.status = status; + } + + void run() throws Exception { + WorkflowClient client = + (WorkflowClient) + Proxy.newProxyInstance( + WorkflowClient.class.getClassLoader(), + new Class[] {WorkflowClient.class}, + (proxy, method, args) -> { + if (method.getName().equals("newUntypedWorkflowStub")) { + String runId = + args[0] instanceof WorkflowTargetOptions + ? ((WorkflowTargetOptions) args[0]).getRunId() + : ""; + return stub(runId); + } + throw new UnsupportedOperationException(method.getName()); + }); + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + try { + SubscriptionDriver driver = + new SubscriptionDriver( + client, + "workflow", + SubscribeOptions.newBuilder().setFromOffset(3).build(), + executor, + new WorkflowStreamListener() { + @Override + public java.util.concurrent.CompletionStage onNext( + WorkflowStreamItem item) { + items.add(item); + return null; + } + + @Override + public void onCompleted() { + completed++; + } + + @Override + public void onError(Throwable error) {} + }, + ignored -> {}); + driver.start(); + driver.getDoneFuture().get(5, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + } + } + + WorkflowStub stub(String runId) { + return (WorkflowStub) + Proxy.newProxyInstance( + WorkflowStub.class.getClassLoader(), + new Class[] {WorkflowStub.class}, + (proxy, method, args) -> { + switch (method.getName()) { + case "startUpdate": + if (!acceptPoll) { + throw new IllegalStateException("closed"); + } + return Proxy.newProxyInstance( + WorkflowUpdateHandle.class.getClassLoader(), + new Class[] {WorkflowUpdateHandle.class}, + (handle, handleMethod, handleArgs) -> { + if (handleMethod.getName().equals("getExecution")) { + return WorkflowExecution.newBuilder().setRunId("accepted-run").build(); + } + CompletableFuture result = new CompletableFuture<>(); + result.completeExceptionally( + new IllegalStateException("closed during poll")); + return result; + }); + case "describe": + describedRun = runId; + if (describeError != null) { + throw describeError; + } + return new WorkflowExecutionDescription( + DescribeWorkflowExecutionResponse.newBuilder() + .setWorkflowExecutionInfo( + WorkflowExecutionInfo.newBuilder() + .setExecution( + WorkflowExecution.newBuilder() + .setWorkflowId("workflow") + .setRunId(terminalRunId)) + .setStatus(status)) + .build(), + DefaultDataConverter.STANDARD_INSTANCE); + case "query": + if (queryError != null) { + throw queryError; + } + Assert.assertEquals(WorkflowStreamConstants.POLL_QUERY_NAME, args[0]); + PollInput input = (PollInput) ((Object[]) args[2])[0]; + queryRuns.add(runId); + queryOffsets.add(input.fromOffset); + String data = + PayloadWire.encode( + DefaultDataConverter.STANDARD_INSTANCE.toPayload("tail").get()); + return new PollResult( + Collections.singletonList(new WireItem("events", data, input.fromOffset)), + input.fromOffset + 1, + input.fromOffset == 3); + default: + throw new UnsupportedOperationException(method.getName()); + } + }); + } + } +} diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamTest.java index 27a6069f1c..114afda4fc 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamTest.java @@ -74,6 +74,28 @@ public void testExternalPublishAndOffsetQuery() { stub.getResult(Void.class); } + @Test + public void testPollQueryReturnsImmediatelyForEmptyAndExhaustedStreams() { + WorkflowStub stub = startHostWorkflow(); + PollResult empty = + stub.query( + WorkflowStreamConstants.POLL_QUERY_NAME, + PollResult.class, + new PollInput(Collections.emptyList(), 0)); + Assert.assertTrue(empty.items.isEmpty()); + Assert.assertFalse(empty.moreReady); + stub.signal("finish"); + stub.getResult(Void.class); + PollResult exhausted = + stub.query( + WorkflowStreamConstants.POLL_QUERY_NAME, + PollResult.class, + new PollInput(Collections.singletonList("missing"), 100)); + Assert.assertTrue(exhausted.items.isEmpty()); + Assert.assertEquals(0, exhausted.nextOffset); + Assert.assertFalse(exhausted.moreReady); + } + @Test public void testPublisherDedup() { WorkflowStub stub = startHostWorkflow();