diff --git a/temporalio/contrib/external_workflow_streams/__init__.py b/temporalio/contrib/external_workflow_streams/__init__.py index 80acb17ae..a04387f21 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 38986819b..90bfa4d85 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.contrib.external_workflow_streams._wake import channel_for @@ -98,6 +99,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. @@ -279,7 +281,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. @@ -298,6 +303,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: @@ -314,6 +324,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 90f0be3e0..6c05adaa5 100644 --- a/temporalio/contrib/external_workflow_streams/_backend.py +++ b/temporalio/contrib/external_workflow_streams/_backend.py @@ -275,6 +275,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" + ) + # --- waking ------------------------------------------------------------- wake_transport: WakeTransport = "auto" diff --git a/temporalio/contrib/external_workflow_streams/_record.py b/temporalio/contrib/external_workflow_streams/_record.py index 549c5c0b3..8b4d96b46 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 9e7c5e53a..d7b631758 100644 --- a/temporalio/contrib/external_workflow_streams/_runtime.py +++ b/temporalio/contrib/external_workflow_streams/_runtime.py @@ -88,6 +88,7 @@ BEGINNING, Cursor, RecordKind, + StartAtTail, StreamRecord, ) from temporalio.contrib.external_workflow_streams._replay import ReplayPlan @@ -200,6 +201,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. @@ -291,6 +300,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 @@ -604,7 +616,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 @@ -1134,13 +1146,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 @@ -1161,7 +1176,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 @@ -1179,6 +1207,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 @@ -1194,16 +1223,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. @@ -2026,7 +2096,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 @@ -2053,11 +2123,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 }, ) @@ -2072,6 +2144,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/streams/providers/redis.py b/temporalio/streams/providers/redis.py index 335467179..d0089f720 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, @@ -907,14 +911,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 diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 2ea360f58..90ce6c205 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 4ab024e0a..cb257cb0f 100644 --- a/tests/contrib/external_workflow_streams/test_api.py +++ b/tests/contrib/external_workflow_streams/test_api.py @@ -76,10 +76,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 c368be89c..d18d658b4 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", diff --git a/tests/streams/test_redis_replay.py b/tests/streams/test_redis_replay.py index bd5b58e58..0493dbcf3 100644 --- a/tests/streams/test_redis_replay.py +++ b/tests/streams/test_redis_replay.py @@ -1148,6 +1148,66 @@ async def run(self) -> Any: 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) + + async def _channel_support(client: Client) -> Any: """What the server offers the readers: no channels, independent ones or linked ones.""" from tests.contrib.external_workflow_streams.conftest import server_channel_support