From 34c1e5a89154d9c5a4b245a8a8e1e14bf267ec90 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 04:51:10 -0700 Subject: [PATCH 1/3] Let an external stream subscription start at the tail or the newest N. The workflow thread cannot see where the store's tail is, so the transport records the request, the Worker resolves it against the store after the task that opened the subscription, and the marker records the boundary so replay reads it from History. --- .../external_workflow_streams/__init__.py | 2 + .../contrib/external_workflow_streams/_api.py | 14 ++- .../external_workflow_streams/_backend.py | 17 ++++ .../external_workflow_streams/_record.py | 22 +++++ .../external_workflow_streams/_runtime.py | 95 ++++++++++++++++--- temporalio/worker/_workflow.py | 5 + .../external_workflow_streams/test_api.py | 6 +- .../test_registry.py | 1 + 8 files changed, 149 insertions(+), 13 deletions(-) diff --git a/temporalio/contrib/external_workflow_streams/__init__.py b/temporalio/contrib/external_workflow_streams/__init__.py index 80acb17aea..a04387f214 100644 --- a/temporalio/contrib/external_workflow_streams/__init__.py +++ b/temporalio/contrib/external_workflow_streams/__init__.py @@ -100,6 +100,7 @@ Offset, OffsetComparator, RecordKind, + StartAtTail, StreamRecord, ) from temporalio.contrib.external_workflow_streams._wake import WakeRequest @@ -128,6 +129,7 @@ "ExternalStreamProducerTopic", "ExternalStreamSubscription", "ExternalStreamTopic", + "StartAtTail", "IdempotencyKey", "Offset", "OffsetComparator", diff --git a/temporalio/contrib/external_workflow_streams/_api.py b/temporalio/contrib/external_workflow_streams/_api.py index 3a80a12f21..e59b4ac225 100644 --- a/temporalio/contrib/external_workflow_streams/_api.py +++ b/temporalio/contrib/external_workflow_streams/_api.py @@ -37,6 +37,7 @@ from temporalio.contrib.external_workflow_streams._record import ( Cursor, Offset, + StartAtTail, StreamRecord, ) from temporalio.types import AnyType @@ -97,6 +98,7 @@ def register( stream_key: StreamKey, idle_timeout: timedelta, start_cursor: Cursor | None = None, + start_at_tail: StartAtTail | None = None, ) -> None: """Registers a wait with the Worker's subscription manager. @@ -278,7 +280,10 @@ class ExternalStreamTopic(Generic[AnyType]): options: ExternalStreamOptions def subscribe( - self, *, start_cursor: Cursor | None = None + self, + *, + start_cursor: Cursor | None = None, + start_at_tail: StartAtTail | None = None, ) -> ExternalStreamSubscription[AnyType]: """Starts a new subscription and returns its async iterator. @@ -297,6 +302,11 @@ def subscribe( ``BEGINNING`` on a first execution. A boundary the Workflow names is recorded in the marker's header like the restored one, so it must be derived deterministically: replay names it again. + start_at_tail: Start at the stream's tail instead, or at its newest + ``last`` records. The Worker resolves the boundary against the + store after this Workflow Task and records it with the + subscription, so replay starts where the live run did rather + than asking the store again. Exclusive with ``start_cursor``. """ state = _run_state() if state.runtime is None: @@ -313,6 +323,8 @@ def subscribe( start: dict[str, Any] = ( {} if start_cursor is None else {"start_cursor": start_cursor} ) + if start_at_tail is not None: + start["start_at_tail"] = start_at_tail state.runtime.register( wait_id=wait_id, stream_key=stream_key, diff --git a/temporalio/contrib/external_workflow_streams/_backend.py b/temporalio/contrib/external_workflow_streams/_backend.py index ca7e1ecdd3..1925bff06e 100644 --- a/temporalio/contrib/external_workflow_streams/_backend.py +++ b/temporalio/contrib/external_workflow_streams/_backend.py @@ -273,6 +273,23 @@ def compare_offsets(self, left: Offset, right: Offset) -> int: inside a validation loop. """ + async def tail_cursor(self, key: StreamKey, *, before_last: int = 0) -> Cursor: + """The boundary the newest ``before_last`` records begin after. + + With ``before_last=0`` it is the boundary after the newest record, so a + read from it sees only what is appended after this call. ``BEGINNING`` + when the stream holds fewer records than asked for. Resolved on the + Worker, never on the Workflow thread, and recorded with the + subscription so replay reads it from the marker instead of asking + again. A provider that cannot answer leaves this as it is, and a + subscription that asks for a tail start fails when the Worker resolves + it. + """ + raise NotImplementedError( + f"{type(self).__name__} cannot resolve a start {before_last} records " + f"before the tail of {key}; subscribe with a cursor instead" + ) + # --- parking (P2b) ------------------------------------------------------ supports_leased_claims: ClassVar[bool] = False diff --git a/temporalio/contrib/external_workflow_streams/_record.py b/temporalio/contrib/external_workflow_streams/_record.py index 549c5c0b33..8b4d96b46c 100644 --- a/temporalio/contrib/external_workflow_streams/_record.py +++ b/temporalio/contrib/external_workflow_streams/_record.py @@ -11,7 +11,29 @@ from dataclasses import dataclass, field from typing import Final, Protocol + +@dataclass(frozen=True) +class StartAtTail: + """A subscription start the Worker resolves against the store: the tail. + + ``last`` is how many of the newest records the read begins with; zero + means the boundary after the newest record, so the read sees only what + is appended after it was opened. The Workflow thread cannot ask the store + where the tail is, so the runtime records the request and the Worker + resolves it before the watcher starts, writing the boundary into the + marker beside the subscription as it does a cursor the Workflow named. + """ + + last: int = 0 + + def __post_init__(self) -> None: + """Refuse a negative count.""" + if self.last < 0: + raise ValueError(f"last must not be negative, got {self.last}") + + __all__ = [ + "StartAtTail", "AFTER", "BEGINNING", "Cursor", diff --git a/temporalio/contrib/external_workflow_streams/_runtime.py b/temporalio/contrib/external_workflow_streams/_runtime.py index e57b3aa911..29dbf6ad8c 100644 --- a/temporalio/contrib/external_workflow_streams/_runtime.py +++ b/temporalio/contrib/external_workflow_streams/_runtime.py @@ -87,6 +87,7 @@ BEGINNING, Cursor, RecordKind, + StartAtTail, StreamRecord, ) from temporalio.contrib.external_workflow_streams._replay import ReplayPlan @@ -199,6 +200,14 @@ class _SubscriptionState: generation: int = 0 """Increments each time this wait re-enters the blocked state.""" + start_pending: bool = False + """The start is a tail the Worker has yet to resolve against the store. + + Until it does, the cursors hold ``BEGINNING`` as a placeholder, no watcher + runs, and the binding stays out of every header and bindings frame, so the + marker never records the placeholder as where this wait began. + """ + announced: bool = False """Whether the *current* annotation has carried this subscription's binding. @@ -290,6 +299,9 @@ def __init__( #: would give replay whatever the stream holds now (ADR-022). self._continuation = continuation self._subscriptions: dict[int, _SubscriptionState] = {} + #: Tail starts registered in the activation under way, resolved by the + #: Worker once it returns. See :meth:`resolve_pending_starts`. + self._pending_starts: dict[int, StartAtTail] = {} #: Where the *current* annotation begins, per wait. Captured when the #: annotation begins rather than when its header is first needed -- #: lazily reading the delivery cursor would let a record delivered @@ -599,7 +611,7 @@ def _reserve_bytes(self) -> int: late = { wait_id: self._binding(state) for wait_id, state in sorted(self._subscriptions.items()) - if not state.announced + if not state.announced and not state.start_pending } if late and self._accumulator is not None: # Only with an accumulator: without one the header is about to carry @@ -1129,13 +1141,16 @@ def register( stream_key: StreamKey, idle_timeout: timedelta | None = None, start_cursor: Cursor | None = None, + start_at_tail: StartAtTail | None = None, ) -> None: """Registers a wait with the Worker's manager. Non-blocking, no I/O. The start cursor defaults to what the predecessor Run committed for this ``wait_id``, so a chain resumes where it left off without the Workflow code saying anything about it -- and a first execution gets ``BEGINNING`` - from the same path. + from the same path. A ``start_at_tail`` is resolved by the Worker after + this activation and recorded with the subscription; on replay the + recorded boundary is taken from the marker and the store is not asked. Raises: ExternalStreamCapacityError: This subscription set cannot be recorded @@ -1156,7 +1171,20 @@ def register( "external streams are not configured on this Worker; pass " "external_stream_backend=... to the Worker" ) - if start_cursor is None: + if start_at_tail is not None and start_cursor is not None: + raise ValueError( + "a subscription starts at a cursor or at the tail, not at both" + ) + pending = False + if start_at_tail is not None: + recorded = (self._replay_bindings or {}).get(wait_id) + if recorded is not None: + # The live run resolved it and the marker carries the answer. + start_cursor = recorded.start_cursor + else: + pending = True + start_cursor = BEGINNING + elif start_cursor is None: start_cursor = self.restored_start(wait_id, stream_key.stream_name) if self._replay_bindings is not None: # A subscription made while a marker is being replayed -- which is @@ -1174,6 +1202,7 @@ def register( delivery_cursor=start_cursor, consumption_cursor=start_cursor, idle_timeout=idle_timeout or self._default_idle_timeout, + start_pending=pending, ) self._subscriptions[wait_id] = state # A subscription created part-way through an annotation begins at its @@ -1189,16 +1218,57 @@ def register( del self._annotation_start[wait_id] raise self._update_reserve() - self._manager.register( - run_id=self._run_id, - wait_id=wait_id, - stream_key=stream_key, - start_cursor=start_cursor, - ) + if pending: + assert start_at_tail is not None + self._pending_starts[wait_id] = start_at_tail + else: + self._manager.register( + run_id=self._run_id, + wait_id=wait_id, + stream_key=stream_key, + start_cursor=start_cursor, + ) # A registration alone is replay-visible: it is what puts the stream in # the annotation header, without which replay cannot start. self._observed_this_activation = True + async def resolve_pending_starts(self) -> None: + """Resolves every tail start the activation just finished registered. + + On the Worker's loop, between the activation and its completion. The + Workflow thread can do no I/O, and the boundary has to be fixed before + anything is delivered on the wait, so the marker records it beside the + subscription and replay reads it there instead of asking the store, + which by then holds something else. + """ + if not self._pending_starts: + return + backend = self._backend + if backend is None: + raise RuntimeError( + "external streams are not configured on this Worker; pass " + "external_stream_backend=... to the Worker" + ) + pending, self._pending_starts = self._pending_starts, {} + for wait_id, tail in pending.items(): + state = self._subscriptions.get(wait_id) + if state is None or not state.start_pending: + continue + cursor = await backend.tail_cursor(state.stream_key, before_last=tail.last) + state.start_cursor = cursor + state.delivery_cursor = cursor + state.consumption_cursor = cursor + state.start_pending = False + self._annotation_start[wait_id] = cursor + self._check_annotation_capacity(wait_id) + self._update_reserve() + self._manager.register( + run_id=self._run_id, + wait_id=wait_id, + stream_key=state.stream_key, + start_cursor=cursor, + ) + def drain(self, wait_id: int, max_records: int | None = None) -> list[StreamRecord]: """Pops buffered records. Performs no I/O and never blocks. @@ -2001,7 +2071,7 @@ def _announce_late_subscriptions(self) -> None: late = { wait_id: state for wait_id, state in sorted(self._subscriptions.items()) - if not state.announced + if not state.announced and not state.start_pending } if not late: return @@ -2028,11 +2098,13 @@ def _header(self) -> AnnotationHeader: wrong store for all the others. """ for state in self._subscriptions.values(): - state.announced = True + if not state.start_pending: + state.announced = True return AnnotationHeader( streams={ wait_id: self._binding(state) for wait_id, state in sorted(self._subscriptions.items()) + if not state.start_pending }, ) @@ -2047,6 +2119,7 @@ def _header_preview(self) -> AnnotationHeader: streams={ wait_id: self._binding(state) for wait_id, state in sorted(self._subscriptions.items()) + if not state.start_pending }, ) diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 9030727bfe..8581f1fc99 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -650,6 +650,11 @@ async def _handle_activation( raise deadlock_exc from None output_runtime = self._external_stream_runtimes.get(act.run_id) + if output_runtime is not None and completion.HasField("successful"): + # A subscription opened at a stream's tail is positioned here, + # off the Workflow thread and before the completion goes out, so + # the marker records the boundary the watcher starts from. + await output_runtime.resolve_pending_starts() if ( output_runtime is not None and completion.HasField("successful") diff --git a/tests/contrib/external_workflow_streams/test_api.py b/tests/contrib/external_workflow_streams/test_api.py index 80bab01b3e..82f6873cd7 100644 --- a/tests/contrib/external_workflow_streams/test_api.py +++ b/tests/contrib/external_workflow_streams/test_api.py @@ -75,10 +75,14 @@ def register( stream_key: StreamKey, idle_timeout: timedelta, start_cursor: Cursor | None = None, + start_at_tail: Any = None, ) -> None: self.registrations.append((wait_id, stream_key)) self.idle_timeouts[wait_id] = idle_timeout - self.start_cursors[wait_id] = start_cursor + # A tail start stands in for the cursor the Worker would resolve. + self.start_cursors[wait_id] = ( + start_cursor if start_at_tail is None else start_at_tail + ) def drain(self, wait_id: int, max_records: int | None = None) -> list[StreamRecord]: buffered = self.buffers.get(wait_id, []) diff --git a/tests/contrib/external_workflow_streams/test_registry.py b/tests/contrib/external_workflow_streams/test_registry.py index c368be89cd..d18d658b4d 100644 --- a/tests/contrib/external_workflow_streams/test_registry.py +++ b/tests/contrib/external_workflow_streams/test_registry.py @@ -104,6 +104,7 @@ def test_supported_entry_points_are_public() -> None: "PrecedingWriteFailedError", "RecordKind", "RedisStreamBackend", + "StartAtTail", "StreamBackend", "StreamDecodeError", "StreamError", From 4f905470c7e0d1b96557c48d14b772a7ce4a16e4 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 04:51:10 -0700 Subject: [PATCH 2/3] Let a Redis workflow reader start at the tail or at the newest N records. stream_reader(after=END) and stream_reader(last=N) on the Redis provider now ask the transport for a tail start instead of being refused. --- temporalio/streams/providers/redis.py | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/temporalio/streams/providers/redis.py b/temporalio/streams/providers/redis.py index 6ab31a1dd4..a7ddf3eb0f 100644 --- a/temporalio/streams/providers/redis.py +++ b/temporalio/streams/providers/redis.py @@ -47,10 +47,13 @@ - Cursors are ``redis:-`` and name an entry of the topic's log, so a cursor a workflow reader returned seeds an outside read and the other way round. A reader opened without a cursor starts where the chain's - predecessor run committed, which is the transport's own rule. An outside - read positions ``END`` and ``last=N`` against the log on the first step of - the generator, since the call itself cannot reach the store. A workflow - reader starts at a cursor or at the beginning. + predecessor run committed, which is the transport's own rule. ``END`` and + ``last=N`` are positioned against the log when the read starts: outside, + on the first step of the generator, since the call itself cannot reach the + store; inside a workflow, by the worker right after the Workflow Task that + opened the subscription, which records the entry it resolved in the marker + beside the subscription, so replay and a cold start read it from History + and never ask the log again. - A workflow's streams are keyed by the chain's first run, so a handle follows continue-as-new by construction and ``run_id`` only decides whose close ends a read. @@ -134,6 +137,7 @@ ChainKeyMismatchError, ExternalStreamProducer, Offset, + StartAtTail, StreamDirection, WakeNotAcknowledgedError, WorkflowChainKey, @@ -867,14 +871,12 @@ def open_reader( ) -> ReadSource: _require_topic(topic) check_read_start(after, last) + subscribe = self._input.topic(topic, type=bytes).subscribe if last is not None or after == END: # The tail is where the log is when the worker looks, which the - # workflow thread cannot see. - raise StreamUnsupportedError( - "a workflow reader on the redis provider starts at a cursor or at " - "the beginning" - ) - subscribe = self._input.topic(topic, type=bytes).subscribe + # workflow thread cannot see: the transport has the worker resolve + # it after this task and record the entry with the subscription. + return _RedisReadSource(subscribe(start_at_tail=StartAtTail(last or 0))) position = _position(after) # Without a position the transport resumes where the chain's # predecessor run committed; with one, that is where the wait starts From c90c588d425647726b6100ee2e0e705fc567e376 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 04:51:10 -0700 Subject: [PATCH 3/3] Covered the Redis workflow reader's tail starts. A reader at the newest N records and one at END each start where the live run resolved, and replay starts there too. --- tests/streams/test_redis_replay.py | 60 ++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/tests/streams/test_redis_replay.py b/tests/streams/test_redis_replay.py index e977565839..f877d8e949 100644 --- a/tests/streams/test_redis_replay.py +++ b/tests/streams/test_redis_replay.py @@ -1144,3 +1144,63 @@ async def run(self) -> Any: reader.close() return value["n"] return None + + +async def test_a_workflow_reader_starts_at_the_last_n_records_and_replays_there( + live_client: Client, provider: RedisStreams +): + # The worker positions the subscription against the log after the task that + # opened it and records the entry with it; replay takes the entry from + # History, so what lands in the log later does not move the start. The + # producer needs the run to exist, so the reader opens on a signal. + workflow_id = f"streams-redis-last-{uuid.uuid4().hex}" + async with Worker( + live_client, + task_queue=f"tq-{workflow_id}", + workflows=[NewestTwoOnGo], + plugins=[provider], + ): + handle = await live_client.start_workflow( + NewestTwoOnGo.run, id=workflow_id, task_queue=f"tq-{workflow_id}" + ) + stream = provider.get_stream_handle(live_client, workflow_id) + producer = stream.producer(topic=INPUTS, producer_id="tool", attempt=1) + await producer.append({"n": 1}, {"n": 2}, {"n": 3}, {"n": 4}) + await handle.signal(NewestTwoOnGo.go) + assert await asyncio.wait_for(handle.result(), 60) == [3, 4] + history = await handle.fetch_history() + await producer.append({"n": 5}, {"n": 6}) + # Offline, with two more records in the log than the run ever saw. + await Replayer(workflows=[NewestTwoOnGo], plugins=[provider]).replay_workflow( + history + ) + + +async def test_a_workflow_reader_at_end_skips_what_was_there_and_replays( + live_client: Client, provider: RedisStreams +): + workflow_id = f"streams-redis-end-{uuid.uuid4().hex}" + async with Worker( + live_client, + task_queue=f"tq-{workflow_id}", + workflows=[FromNowOnGo], + plugins=[provider], + ): + handle = await live_client.start_workflow( + FromNowOnGo.run, id=workflow_id, task_queue=f"tq-{workflow_id}" + ) + stream = provider.get_stream_handle(live_client, workflow_id) + producer = stream.producer(topic=INPUTS, producer_id="tool", attempt=1) + await producer.append({"n": "old"}) + await handle.signal(FromNowOnGo.go) + result = asyncio.ensure_future(handle.result()) + # The subscription is positioned when the worker gets to it, which the + # test does not observe, so appends keep coming until the run takes one. + for _ in range(150): + await producer.append({"n": "new"}) + done, _ = await asyncio.wait({result}, timeout=0.2) + if done: + break + assert await asyncio.wait_for(result, 60) == "new" + history = await handle.fetch_history() + await Replayer(workflows=[FromNowOnGo], plugins=[provider]).replay_workflow(history)