From 6ba8d4ae1e7ba9ca222311c2c85e3cdabd51f6f6 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 18 Oct 2026 00:00:38 -0700 Subject: [PATCH 1/2] Read an operation's stream from a caller Workflow. --- temporalio/contrib/streams/nexus/__init__.py | 24 + temporalio/contrib/streams/nexus/_reader.py | 961 +++++++++++++++++++ 2 files changed, 985 insertions(+) create mode 100644 temporalio/contrib/streams/nexus/_reader.py diff --git a/temporalio/contrib/streams/nexus/__init__.py b/temporalio/contrib/streams/nexus/__init__.py index 25f4e55f29..21fde742fa 100644 --- a/temporalio/contrib/streams/nexus/__init__.py +++ b/temporalio/contrib/streams/nexus/__init__.py @@ -40,6 +40,18 @@ :meth:`temporalio.contrib.streams.StreamStorePlugin.close_stream` completes the operation, or :func:`temporalio.contrib.streams.close_workflow_stream` from the owning Workflow. + +A caller Workflow reads such a stream with :class:`StreamReader`: it waits on +the operation's progress and reads through this service. By default each +read is a Nexus operation of the Workflow, so History records each batch and +a replay hands over the same batches with no store at hand. This works on +any server, Temporal Cloud included. ``replay_from_store=True`` is the +opt-in. It reads in a Local Activity through a +:class:`StreamDirectTransport` that :class:`StreamDirectReads` maps the +endpoint to, and History keeps only each read's range and digest. +:class:`ClientTransport` calls ``ExecuteSyncNexusOperation`` with the +caller's own client, and :class:`HttpTransport` the HTTP ingress of a +self-hosted server. """ from temporalio.contrib.streams import StreamRef @@ -64,6 +76,13 @@ StreamOperationHandler, stream_ref_from_token, ) +from temporalio.contrib.streams.nexus._reader import ( + ReadPosition, + ReadStaleRecord, + ReadSupersession, + StreamIncompleteError, + StreamReader, +) __all__ = [ "AppendInput", @@ -71,10 +90,15 @@ "HTTPStatusError", "ReadInput", "ReadOutput", + "ReadStaleRecord", + "ReadPosition", + "ReadSupersession", "RecordWire", "StreamAccess", "StreamCursor", + "StreamIncompleteError", "StreamOperationHandler", + "StreamReader", "StreamRef", "TemporalStreams", "TemporalStreamsHandler", diff --git a/temporalio/contrib/streams/nexus/_reader.py b/temporalio/contrib/streams/nexus/_reader.py new file mode 100644 index 0000000000..b35a7de79d --- /dev/null +++ b/temporalio/contrib/streams/nexus/_reader.py @@ -0,0 +1,961 @@ +"""Read the stream a Nexus operation returns, from a Workflow. + +A stream-returning operation stays open while its stream runs. Its handler +reports progress whenever the stream moves and completes the operation when +the stream closes; the operation token names the stream. The reader turns +that into batches of typed records: it reads from its cursor through the +stream service on a Nexus endpoint, and waits for progress once a read finds +nothing new. Every append notifies the operation's callers, and the server +folds the notifications of a busy stream into one per debounce window. + +By default every read is a Nexus operation of the Workflow, so History +records what each read answered, and a replay hands over the same batches at +the same points. With ``replay_from_store=True`` each read runs in a Local +Activity, History keeps only the range that crossed and its digest, and a +replay reads that range again from the owner's store. That needs a +direct-read transport for the endpoint. +""" + +from __future__ import annotations + +import asyncio +import base64 +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Any, Generic, TypeVar, cast + +from temporalio import workflow +from temporalio.bridge.proto.streams import ReadRecord +from temporalio.bridge.proto.streams.v1 import StreamLink as WireLink +from temporalio.common import RawValue +from temporalio.contrib.streams._errors import ( + StreamError, + StreamRecordError, + StreamUnsupportedError, +) +from temporalio.contrib.streams._follow import ( + FollowedItem, + FollowPosition, + LinkState, + LinkStateKind, +) +from temporalio.contrib.streams._record import ( + Cursor, + RecordKind, + StreamRecord, + Supersession, +) +from temporalio.contrib.streams._ref import StreamRef +from temporalio.contrib.streams._wire import WireRecord, from_wire_link, to_record +from temporalio.contrib.streams._workflow import TaskCutoff, cut_task +from temporalio.contrib.streams.nexus._generated import ( + FollowedWire, + FollowInput, + FollowOutput, + ReadInput, + ReadOutput, + RecordWire, + TemporalStreams, +) +from temporalio.contrib.streams.nexus._operation import stream_ref_from_token +from temporalio.converter import PayloadConverter +from temporalio.exceptions import ActivityError, ApplicationError + +__all__ = [ + "ReadPosition", + "ReadStaleRecord", + "ReadSupersession", + "StreamIncompleteError", + "StreamReader", +] + +T = TypeVar("T") + +_BATCH = 100 +# Once the operation completes, a read waits this long for a record, so a +# reader waiting for the stream's end does not spin. +_DRAIN_WAIT_MS = 2000 +# How long a reader waits for progress before it reads again anyway. A +# producer that dies between its append and its notify sends no progress, so +# only a read finds that record. The wait doubles while nothing arrives, up to +# the same 5 minutes as the server's recheck of a Workflow reader. +_FIRST_RECHECK = timedelta(minutes=1) +_MAX_RECHECK = timedelta(minutes=5) +# A direct read's Local Activity waits at most its read's wait, and this +# leaves room for the call around it. +_DIRECT_READ_SLACK = timedelta(seconds=10) +_READERS_ATTRIBUTE = "_temporal_stream_nexus_readers" +# When a reader that reads directly ends the Workflow Task it reads in. +_TASK_CUTOFF = TaskCutoff() + + +class _RecheckDue(Exception): + pass + + +class StreamIncompleteError(StreamError): + """The stream and its operation didn't both end within the drain limit. + + .. warning:: + This API is experimental and may change in future versions. + + Either the operation completed but the stream never said it ended, or the + stream ended but the operation never completed. The reader gives up + after its drain limit, because every read is a Nexus operation in the + caller's History, and a wait with no end holds the caller open. The + first happens when the owner's close didn't reach the store, or when the + Workflow id moved on to a new run chain whose topic is still open. The + second happens when the close never reached the caller, for example when + the server refused it for good. + """ + + def __init__(self, message: str, *, cursor: Cursor) -> None: + """The reader stopped at ``cursor``.""" + super().__init__(message) + self.cursor = cursor + """Where the reader stopped. A new reader can resume after it.""" + + +@dataclass(frozen=True) +class ReadSupersession: + """Where a reader saw a later attempt of a producer take over. + + .. warning:: + This API is experimental and may change in future versions. + """ + + index: int + """How many items the reader had handed over before it. The items from + there on that the producer wrote belong to the later attempt; the ones + before it from the same producer may be replaced by them.""" + + cursor: Cursor + """The stream position it reports, just before the later attempt's first + record.""" + + supersession: Supersession + """The producer and the attempts it names.""" + + +@dataclass(frozen=True) +class ReadStaleRecord: + """A record from a producer attempt behind one the reader already saw. + + .. warning:: + This API is experimental and may change in future versions. + + An attempt that timed out but kept running can write after a later + attempt started. Its records are not the current answer, so the reader + keeps them out of the batches it hands over and reports them here. + """ + + index: int + """How many items the reader had handed over before it.""" + + cursor: Cursor + """The stream position of the record.""" + + producer_id: str + """The producer that wrote it. Empty when the service couldn't hand the + record over.""" + + attempt: int + """The attempt that wrote it, lower than one the reader already saw. Zero + when the service couldn't hand the record over.""" + + value: Any + """The record's item, decoded into the reader's item type. ``None`` when + the record couldn't be handed over or decoded, since a stale record + raises nothing.""" + + +@dataclass(frozen=True) +class ReadPosition: + """Where a reader is, with what the stream service knows of the read. + + .. warning:: + This API is experimental and may change in future versions. + + Pass it to a new reader as ``after=`` to go on where an earlier one + stopped, such as across Continue-as-New. It holds the cursor and the + service's read state, so the new reader's superseded and stale marks + stay exact for every producer. A cursor alone tells the service only the + cursor record's producer. Both fields are text, so the position passes as + a Workflow argument. + """ + + cursor: Cursor + """Past every record the reader handed over.""" + + state: str = "" + """The stream service's read state, opaque and in base64. Empty before + the first read.""" + + +class StreamReader(Generic[T]): + """Reads the stream a Nexus operation returns, in a Workflow. + + .. warning:: + This API is experimental and may change in future versions. + + Make one from the operation's handle and call :meth:`next` until it + returns ``None``:: + + reader = StreamReader(handle, item_type=Token, endpoint="chat") + while (batch := await reader.next()) is not None: + ... + + A generated client's ``read__stream`` method builds one with + the item type and endpoint the contract and the client know. + + The reader holds one cursor, starts at the oldest record the stream + retains, and hands over each record once. Only data records are handed + over; a producer's ``FINISH`` record is not. When a later attempt of a + producer takes over (a retried Activity, for example), the earlier + attempt's records already handed over stay handed over, and + :attr:`supersessions` says where the later attempt began. A record an + earlier attempt writes after a later one started is not handed over, and + :attr:`stale_records` reports it. The stream service marks both. Each read + carries the state the service answered with last, so the marks hold + across reads and across the service's Workers. A reader built with + ``after=`` an earlier reader's :attr:`position` keeps them exact too. One + built with only a cursor tells the service only the cursor record's + producer, so another producer's stale record can pass as current data. A + record whose body the Workflow's payload converter cannot decode into + ``item_type`` raises :class:`temporalio.contrib.streams.StreamRecordError` + from :meth:`next` once the records before it are handed over; the call + after that goes on past it. + + Each read goes to the run chain the Workflow id has when the read runs, + not to the chain the operation token names: a stream reference pins a + run, not a chain. So if the owner's chain ends and the same Workflow id + starts a new chain while this reader still drains, a later read can hand + over the new chain's records. Give each stream's owner a Workflow id of + its own, or let the reader finish before the id is reused. + + A reader that waits for progress reads again once a minute passes with + none, then after 2 and 4 minutes, and then every 5 minutes while the + stream stays quiet. A producer that dies between its append and its + notify sends no progress, so this read is what finds its record. Each + such read costs a timer and a read in History, about 290 a day for an + idle stream. + + By default every read is a Nexus operation result in the caller's + History, about three events and up to one read answer (1 MiB of records) + each. With ``replay_from_store=True`` a read is one Local Activity marker + of a few hundred bytes, as below. To read a long stream, Continue-as-New + before History grows too large: pass :attr:`position` to the new run, + start the operation again there, and build the reader with + ``after=position``. The position is at most about 22 KiB. After a + successful completion, the reader reads until the stream service says + the stream is done, and raises + :class:`StreamIncompleteError` once its reads have found nothing for + ``drain_limit``. A stream that ended waits at most ``drain_limit`` for its + operation to complete, then raises the same error. A read that fails + because retention dropped the records after the cursor raises with + ``StreamExpiredError`` as its cause. A new reader with no ``after`` starts + again at the oldest record still kept. + + Each read before the completion waits up to ``idle_wait`` for a record. + A read with records answers at once. A read that waits the whole + ``idle_wait`` and finds nothing puts the reader to sleep on progress, + which the stream's next append sends. So a gap shorter than ``idle_wait`` + costs one read and no progress Workflow Task. + + Bodies reach Workflow code decoded: the stream service decodes each body + with its Worker's data converter, payload codec included, and the read's + result travels back as one Nexus operation result, which this Worker's + codec decodes. So the Workers on both sides need the same payload codec, + as Workers in one namespace normally have. + + With ``replay_from_store=True`` each read runs in a Local Activity, which + reaches the endpoint's stream service through the transport that + :class:`StreamDirectReads` on the Worker's client maps the endpoint to. + The read names a floor holder, ``//``, so the owner's store keeps the records it read. The marker + holds the range that crossed and the stream runtime's digest of its + records, not the records. The Worker turns it back into the batch before Workflow code + sees it, from a cache on the live Workflow Task, or else with a + ``read_range`` through the same transport, which checks the digest. A + replay that finds other records fails the Workflow Task as + nondeterminism. One that finds the range gone, or can't reach the + endpoint, fails the Workflow Task, which is retried. So the endpoint + must stay reachable, and its store must keep the records, while the run + can replay. The reader never releases its hold while its run is open. + Each read and replay renews it, and it lapses the store's floor lease + after the last one. A read failure reaches :meth:`next` as an + ``ActivityError`` in place of a ``NexusOperationError``. + """ + + def __init__( + self, + handle: workflow.NexusOperationHandle[Any], + *, + item_type: type[T], + endpoint: str, + after: Cursor | ReadPosition | None = None, + drain_limit: timedelta = timedelta(seconds=60), + idle_wait: timedelta = timedelta(seconds=2), + replay_from_store: bool = False, + follow_links: bool = False, + max_depth: int = 4, + max_open: int = 16, + ) -> None: + """Read the stream ``handle``'s operation returns. + + Args: + handle: The started stream-returning operation. Its token names + the stream. + item_type: What each record's body decodes into. + endpoint: The Nexus endpoint that serves the stream service for + this stream, usually the operation's own. + after: Resume where an earlier reader stopped. Pass its + :attr:`position` to keep every producer's marks exact. A + bare cursor tells the stream service only the cursor record's + producer, so another producer's stale record can pass as + current data. ``None`` starts at the oldest record kept. + drain_limit: How long reads after a successful completion may + find nothing, or a stream that ended may wait for its + operation to complete, before the reader raises + :class:`StreamIncompleteError`. + idle_wait: How long a read waits for a record before the reader + waits for progress. Zero sleeps on the first empty read. Keep + it well below the 10 second timeout of a Nexus request. + replay_from_store: Read in a Local Activity and record only the + range of each read and its digest, so the records stay out + of History. It needs :class:`StreamDirectReads` on the + Worker's client with a transport for ``endpoint``, and on any + Replayer. Without one, the read raises + :class:`temporalio.contrib.streams.StreamUnsupportedError`, + which names this option. ``False``, the default, reads + through Nexus operations and records each read's answer, + records included, so it needs no transport. + follow_links: Follow the stream's link records too, through the + stream service's ``follow``. Read the items with + :meth:`next_followed` in place of :meth:`next`. Each item + names its source stream, and link reports come as items. Links + to another namespace or endpoint come as ``SKIPPED`` with + reason ``endpoint``. A replay reads the follow's ranges again + with ``follow_range``. + While a linked stream is open, the reader follows again at + once after an empty answer, since the stream service waits on + every open stream in the call. Only the root's progress wakes + a reader whose links all ended. A reader resumed from a + position learns which links are open from the reports that + follow. + max_depth: How deep a follower opens links. + max_open: How many linked streams a follower holds open at once. + + Raises: + ValueError: ``drain_limit`` is not positive, ``idle_wait`` is + negative, or a follow bound is below one. + """ + if drain_limit <= timedelta(0): + raise ValueError("drain_limit must be positive") + if idle_wait < timedelta(0): + raise ValueError("idle_wait must not be negative") + if max_depth < 1 or max_open < 1: + raise ValueError("max_depth and max_open must be at least 1") + self._follow_links = follow_links + self._max_depth = max_depth + self._max_open = max_open + self._follow_answer: FollowOutput | None = None + self._follow_pending: list[FollowedWire] = [] + self._sources: dict[int, tuple[StreamRef, int]] = {} + # The linked streams this reader saw open and not yet end. + self._open_links: set[StreamRef] = set() + self._idle_wait_ms = idle_wait // timedelta(milliseconds=1) + self._handle = handle + self._item_type = item_type + self._endpoint = endpoint + self._drain_limit = drain_limit + self._drain_since: datetime | None = None + self._done_since: datetime | None = None + self._stream: StreamRef | None = None + # The stream runtime's read state from the last answer, which the next + # read carries. + self._state: bytes | None = None + if isinstance(after, ReadPosition) and after.state: + self._state = base64.b64decode(after.state) + self._answer: ReadOutput | None = None + self._pending: list[RecordWire] = [] + if isinstance(after, ReadPosition): + after = after.cursor + self._cursor = after.token if after is not None else "" + self._counter = 0 + # Whether the last read answered with no record. An answer short of + # the batch may still leave records behind (the service caps an + # answer's bytes too), so only an empty one means the tail. + self._caught_up = False + # The stream service said no record will follow. + self._stream_done = False + self._operation_resolved = False + self._operation_failed = False + self._ended = False + self._handed_over = 0 + self._supersessions: list[ReadSupersession] = [] + self._stale_records: list[ReadStaleRecord] = [] + self._undecodable: StreamRecordError | None = None + self._recheck = _FIRST_RECHECK + self._replay_from_store = replay_from_store + self._reader_id = str(_next_reader()) if replay_from_store else "" + self._direct_reads = 0 + # The run that made the last direct read. + self._chain_run = "" + + @property + def cursor(self) -> Cursor: + """Where the reader is: past every record it has handed over.""" + return Cursor(self._cursor) + + @property + def position(self) -> ReadPosition: + """Where the reader is, with the service's read state, for a new reader.""" + state = base64.b64encode(self._state).decode() if self._state else "" + return ReadPosition(cursor=self.cursor, state=state) + + @property + def supersessions(self) -> Sequence[ReadSupersession]: + """Every producer attempt change this reader has seen, in order.""" + return tuple(self._supersessions) + + @property + def stale_records(self) -> Sequence[ReadStaleRecord]: + """Every record from an attempt behind one already seen, in order.""" + return tuple(self._stale_records) + + async def next(self) -> list[T] | None: + """The next batch of records, or ``None`` once the stream has ended. + + Reads on while reads answer with records, and waits for the + operation to report progress only once a read answers with none. + Never answers with an empty batch. The stream closes in the store + before the operation completes, so once the operation completes the + reader reads until the stream service says the stream is done, then + answers ``None``. An operation fails when its owner closes the stream + with an error, or ends without closing it after it failed, was + canceled, was terminated or timed out. The store closes before the + operation fails, so the reader hands over what reads find until one + that waited a moment answers with none, which loses no record, then + raises the failure. A canceled owner raises a canceled operation. + + Raises: + temporalio.exceptions.NexusOperationError: A read failed, or the + operation failed. A failed operation raises after the records + it left are handed over. + temporalio.contrib.streams.StreamRecordError: A record could not + be decoded into + ``item_type``, or was too large to cross. The next call goes + on past it. + StreamIncompleteError: The operation completed, but reads found + nothing for ``drain_limit`` and the stream never ended. Or the + stream ended, but the operation didn't complete within + ``drain_limit``. + ValueError: The operation's token does not name a stream, or the + reader follows links, so :meth:`next_followed` reads it. + """ + if self._follow_links: + raise ValueError("a reader with follow_links=True reads with next_followed") + return cast("list[T] | None", await self._next()) + + async def next_followed(self) -> list[FollowedItem[T]] | None: + """The next batch of a reader with ``follow_links=True``, or ``None`` at the end. + + As :meth:`next`, but each item names its source stream and carries + a record, which may be any kind, or a link report. The root's + records decode into ``item_type``. A linked stream's records decode + with the default rules, since its topic may hold another type. The + last item of each batch carries the position after the batch. + + Raises: + ValueError: The reader does not follow links. + temporalio.exceptions.NexusOperationError: As :meth:`next`. + temporalio.contrib.streams.StreamRecordError: As :meth:`next`. + StreamIncompleteError: As :meth:`next`. + """ + if not self._follow_links: + raise ValueError("a reader without follow_links=True reads with next") + return cast("list[FollowedItem[T]] | None", await self._next()) + + async def _next(self) -> list[Any] | None: + if self._undecodable is not None: + error, self._undecodable = self._undecodable, None + raise error + while not self._ended: + # After a completion, only the stream's end ends the read. + reads_on = self._operation_resolved and not self._operation_failed + if not self._stream_done and (reads_on or not self._caught_up): + batch = await self._read() + if batch: + self._recheck = _FIRST_RECHECK + self._handed_over += len(batch) + if reads_on: + self._drain_since = workflow.now() + return batch + if self._undecodable is not None: + error, self._undecodable = self._undecodable, None + raise error + if ( + reads_on + and self._caught_up + and not self._stream_done + and self._drain_since is not None + and workflow.now() - self._drain_since >= self._drain_limit + ): + raise StreamIncompleteError( + f"the operation completed, but the stream didn't end within " + f"{self._drain_limit} of empty reads", + cursor=self.cursor, + ) + # An answer of records that carry no item, such as a FINISH, + # is not the tail. + continue + if self._operation_resolved: + self._ended = True + # Raises the operation's failure, if it failed. + await self._handle + return None + try: + progress = await self._next_progress() + except _RecheckDue: + self._caught_up = False + continue + self._recheck = _FIRST_RECHECK + if progress is None: + # Records may have landed after the last progress. + self._operation_resolved = True + self._drain_since = workflow.now() + try: + await self._handle + except Exception: + self._operation_failed = True + else: + self._counter = progress.counter + self._caught_up = False + return None + + async def _next_progress(self) -> workflow.NexusOperationProgress | None: + if not self._stream_done: + try: + return await asyncio.wait_for( + self._handle.progress(after_counter=self._counter), + self._recheck.total_seconds(), + ) + except asyncio.TimeoutError: + self._recheck = min(self._recheck * 2, _MAX_RECHECK) + raise _RecheckDue from None + # No record follows the stream's end, so only the completion is left, + # and a completion that never comes would hold the caller open. + if self._done_since is None: + self._done_since = workflow.now() + left = self._drain_limit - (workflow.now() - self._done_since) + try: + if left <= timedelta(0): + raise asyncio.TimeoutError + return await asyncio.wait_for( + self._handle.progress(after_counter=self._counter), + left.total_seconds(), + ) + except asyncio.TimeoutError: + raise StreamIncompleteError( + f"the stream ended, but the operation didn't complete within " + f"{self._drain_limit}", + cursor=self.cursor, + ) from None + + async def _read(self) -> list[Any]: + if self._follow_links: + return await self._read_followed() + if not self._pending: + await self._fetch() + batch: list[T] = [] + while self._pending: + stored = self._pending.pop(0) + cursor = Cursor(stored.token) + self._cursor = stored.token + if stored.superseded is not None: + self._supersessions.append( + ReadSupersession( + index=self._handed_over + len(batch), + cursor=cursor, + supersession=Supersession( + producer_id=stored.superseded.producer_id, + previous_attempt=stored.superseded.previous_attempt, + attempt=stored.superseded.attempt, + ), + ) + ) + continue + if stored.stale: + # Not the current answer, so one that can't be read is still + # reported, and raises nothing. + self._stale_records.append(self._stale(stored, cursor, len(batch))) + continue + try: + record = self._record(stored) + except Exception as error: + # The rest of the answer stays pending, so the next call hands + # it over without reading again. + self._undecodable = StreamRecordError( + f"stream record at {cursor} could not be read as " + f"{getattr(self._item_type, '__name__', self._item_type)!s}: " + f"{error}", + cursor, + ) + if self._pending: + self._caught_up = False + else: + # The answer ends here, so its state and token go with + # the cursor now, or the next read sends the old state. + self._take_answer_end() + return batch + if record is None: + continue + batch.append(cast(T, record[2])) + self._take_answer_end() + return batch + + def _take_answer_end(self) -> None: + answer = self._answer + assert answer is not None + if answer.next_token: + self._cursor = answer.next_token + # The stream runtime's state matches the answer's end, so it goes with + # the cursor. + self._state = answer.state + self._caught_up = not answer.records + self._stream_done = answer.done + + async def _fetch(self) -> None: + if self._stream is None: + token = self._handle.operation_token + if not token: + raise ValueError("the operation has no token, so it names no stream") + self._stream = stream_ref_from_token(token) + # A read covers every record the progress seen so far announced. + latest = self._handle.latest_progress + if latest is not None: + self._counter = max(self._counter, latest.counter) + input = ReadInput( + stream=self._stream, + after_token=self._cursor or None, + max_records=_BATCH, + wait_ms=_DRAIN_WAIT_MS if self._operation_resolved else self._idle_wait_ms, + state=self._state, + # An empty answer makes the stream service ask the notifier to wake + # this operation's idle callback once the stream moves on. After the + # completion no progress follows, so a drain read asks for nothing. + wake_token=None + if self._operation_resolved + else self._handle.operation_token or None, + ) + if self._replay_from_store: + self._answer = await self._direct_read(input) + else: + client = workflow.create_nexus_client( + service=TemporalStreams, endpoint=self._endpoint + ) + self._answer = await client.execute_operation(TemporalStreams.read, input) + self._pending = list(self._answer.records) + + async def _read_followed(self) -> list[FollowedItem[Any]]: + if not self._follow_pending: + await self._fetch_followed() + answer = self._follow_answer + assert answer is not None and self._stream is not None + items, self._cursor, undecodable, self._follow_pending = followed_items( + workflow.payload_converter(), + self._sources, + self._stream, + answer, + self._follow_pending, + self._item_type, + self._cursor, + ) + for item in items: + if item.link_state is None: + continue + if item.link_state.kind is LinkStateKind.OPENED: + self._open_links.add(item.source) + elif item.link_state.kind in ( + LinkStateKind.DONE, + LinkStateKind.UNAVAILABLE, + ): + self._open_links.discard(item.source) + if undecodable is not None: + self._undecodable = undecodable + self._caught_up = False + return items + self._state = answer.state + # The root's progress says nothing of a linked stream, so while one is + # open the reader follows again, and the call waits on every open stream. + self._caught_up = not answer.records and not self._open_links + self._stream_done = answer.done + return items + + async def _fetch_followed(self) -> None: + if self._stream is None: + token = self._handle.operation_token + if not token: + raise ValueError("the operation has no token, so it names no stream") + self._stream = stream_ref_from_token(token) + latest = self._handle.latest_progress + if latest is not None: + self._counter = max(self._counter, latest.counter) + input = FollowInput( + stream=self._stream, + after_token=(self._cursor or None) if self._state is None else None, + state=self._state, + max_records=_BATCH, + wait_ms=_DRAIN_WAIT_MS if self._operation_resolved else self._idle_wait_ms, + max_depth=self._max_depth, + max_open=self._max_open, + wake_token=None + if self._operation_resolved + else self._handle.operation_token or None, + ) + if self._replay_from_store: + self._follow_answer = await self._direct_follow(input) + else: + client = workflow.create_nexus_client( + service=TemporalStreams, endpoint=self._endpoint + ) + self._follow_answer = await client.execute_operation( + TemporalStreams.follow, input + ) + self._follow_pending = list(self._follow_answer.records) + + async def _direct_follow(self, input: FollowInput) -> FollowOutput: + # Imported here, since that module imports the follower, which imports this one. + from temporalio.contrib.streams.nexus._direct import ( + DIRECT_FOLLOW_ACTIVITY, + follow_answer_of, + ) + + # Reads and follow steps run one after another in Local Activities, + # which keep the Workflow Task open, so the reader ends the task as a + # Workflow reader does. + cut_task(_TASK_CUTOFF) + wait = timedelta(milliseconds=max(self._idle_wait_ms, _DRAIN_WAIT_MS)) + try: + raw = await workflow.execute_local_activity( + DIRECT_FOLLOW_ACTIVITY, + args=[self._endpoint, self._reader_id, input], + result_type=RawValue, + start_to_close_timeout=wait + _DIRECT_READ_SLACK, + summary=f"stream follow {input.stream.topic}", + ) + except ValueError as error: + if DIRECT_FOLLOW_ACTIVITY not in str(error): + raise + raise StreamUnsupportedError( + f"a stream follow with replay_from_store=True follows directly from " + f"endpoint {self._endpoint!r}, and the Worker has no " + "StreamDirectReads; register it on the Worker's client with a " + "transport for the endpoint, or pass replay_from_store=False to the " + "StreamReader" + ) from error + except (ActivityError, ApplicationError) as error: + unsupported = _unsupported(error) + if unsupported is None: + raise + raise StreamUnsupportedError(unsupported) from error + return follow_answer_of(raw.payload) + + async def _direct_read(self, input: ReadInput) -> ReadOutput: + # Imported here, since that module imports the follower, which imports this one. + from temporalio.contrib.streams.nexus._direct import ( + DIRECT_READ_ACTIVITY, + answer_of, + chain_run_of, + ) + + # Reads and follow steps run one after another in Local Activities, + # which keep the Workflow Task open, so the reader ends the task as a + # Workflow reader does. + cut_task(_TASK_CUTOFF) + wait = timedelta(milliseconds=max(self._idle_wait_ms, _DRAIN_WAIT_MS)) + try: + raw = await workflow.execute_local_activity( + DIRECT_READ_ACTIVITY, + args=[ + self._endpoint, + self._reader_id, + input, + self._direct_reads + 1, + # A reset run replays the old run's reads, whose markers + # chain under that run, so its first read keeps its state. + self._chain_run == workflow.info().run_id, + ], + result_type=RawValue, + start_to_close_timeout=wait + _DIRECT_READ_SLACK, + summary=f"stream read {input.stream.topic}", + ) + self._direct_reads += 1 + self._chain_run = chain_run_of(raw.payload) + except ValueError as error: + # The Worker refuses at the call a Local Activity it doesn't have, + # which is the case with no StreamDirectReads on its client. + if DIRECT_READ_ACTIVITY not in str(error): + raise + raise StreamUnsupportedError( + f"a stream read with replay_from_store=True reads " + f"directly from endpoint {self._endpoint!r}, and the Worker has no " + "StreamDirectReads; register it on the Worker's client with a " + "transport for the endpoint, or pass replay_from_store=False to the " + "StreamReader" + ) from error + except (ActivityError, ApplicationError) as error: + unsupported = _unsupported(error) + if unsupported is None: + raise + # A Worker without the transport is a deployment fault, so the + # Workflow Task fails and is retried rather than the Workflow. + raise StreamUnsupportedError(unsupported) from error + return answer_of(raw.payload) + + def _stale( + self, stored: RecordWire, cursor: Cursor, batched: int + ) -> ReadStaleRecord: + producer_id, attempt, value = "", 0, None + if not stored.error: + try: + wire = WireRecord.FromString(stored.record) + producer_id, attempt = wire.producer_id, wire.attempt + record = self._record(stored) + value = record[2] if record is not None else None + except Exception: + pass + return ReadStaleRecord( + index=self._handed_over + batched, + cursor=cursor, + producer_id=producer_id, + attempt=attempt, + value=value, + ) + + def _record(self, stored: RecordWire) -> tuple[str, int, Any] | None: + """The producer, attempt and item of a data record, or ``None``.""" + if stored.error: + # The service left out a record it couldn't hand over. + raise ValueError(stored.error) + wire = WireRecord.FromString(stored.record) + kind = RecordKind(wire.kind) if wire.kind else RecordKind.DATA + if kind is RecordKind.SUPERSEDED: + raise ValueError("a SUPERSEDED record is marked by readers, never stored") + if kind is not RecordKind.DATA: + return None + value: Any = None + if wire.HasField("body"): + value = workflow.payload_converter().from_payloads( + [wire.body], [self._item_type] + )[0] + return wire.producer_id, wire.attempt, value + + +def _next_reader() -> int: + runtime = workflow._Runtime.current() + number = getattr(runtime, _READERS_ATTRIBUTE, 0) + 1 + setattr(runtime, _READERS_ATTRIBUTE, number) + return number + + +def _unsupported(error: BaseException) -> str | None: + """The message of a ``StreamUnsupportedError`` the read Activity raised, if it did.""" + found: BaseException | None = error + while found is not None: + if ( + isinstance(found, ApplicationError) + and found.type == StreamUnsupportedError.__name__ + ): + return found.message + found = found.cause if isinstance(found, ActivityError) else None + return None + + +def followed_items( + payload_converter: PayloadConverter, + sources: dict[int, tuple[StreamRef, int]], + root: StreamRef, + answer: FollowOutput, + pending: list[FollowedWire], + item_type: type | None, + cursor: str, +) -> tuple[list[FollowedItem[Any]], str, StreamRecordError | None, list[FollowedWire]]: + """The items of ``pending``, which are the rest of ``answer``, in follow order. + + Returns the items, the root cursor after them, the error of a record that + did not decode, and the items left after it. The root's records decode + into ``item_type``. The last item of a whole answer carries the position + after it. + """ + for source in answer.sources: + sources[source.number] = (source.stream, source.depth) + items: list[FollowedItem[Any]] = [] + pending = list(pending) + while pending: + followed = pending.pop(0) + ref, depth = sources.get(followed.source, (root, 0)) + if followed.link_state is not None: + state = followed.link_state + kind = LinkStateKind[state.kind.upper()] + items.append( + FollowedItem( + ref, + depth, + None, + LinkState( + kind=kind, + link=from_wire_link(WireLink.FromString(state.link)), + error=StreamError(state.error) if state.error else None, + reason=state.reason or "", + ), + None, + ) + ) + continue + stored = followed.record + if stored is None: + continue + try: + record = _followed_record( + payload_converter, stored, item_type if followed.source == 0 else None + ) + except Exception as error: + at = Cursor(stored.token) + return ( + items, + cursor, + StreamRecordError( + f"stream record at {at} could not be read: {error}", at + ), + pending, + ) + if followed.source == 0: + cursor = stored.token + items.append(FollowedItem(ref, depth, record, None, None)) + if items: + last = items[-1] + position = FollowPosition( + cursor, base64.b64encode(answer.state or b"").decode() + ) + items[-1] = FollowedItem( + last.source, last.depth, last.record, last.link_state, position + ) + return items, cursor, None, pending + + +def _followed_record( + payload_converter: PayloadConverter, stored: Any, item_type: type | None +) -> StreamRecord[Any]: + if stored.error: + raise ValueError(stored.error) + read = ReadRecord(cursor=stored.token, stale=bool(stored.stale)) + if stored.superseded is not None: + read.superseded.producer_id = stored.superseded.producer_id + read.superseded.previous_attempt = stored.superseded.previous_attempt + read.superseded.attempt = stored.superseded.attempt + else: + read.stored.CopyFrom(WireRecord.FromString(stored.record)) + return to_record(payload_converter, read, item_type) From 83298c7a774b423aa31681f9755462360d02d684 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 18 Oct 2026 03:29:15 -0700 Subject: [PATCH 2/2] Made positions the Nexus reader's default again. Positions-Only is the one default on every stage. Recorded batches stay the opt-in for a caller with no direct-read transport. --- temporalio/contrib/streams/nexus/__init__.py | 14 +++---- temporalio/contrib/streams/nexus/_reader.py | 40 ++++++++++---------- 2 files changed, 27 insertions(+), 27 deletions(-) diff --git a/temporalio/contrib/streams/nexus/__init__.py b/temporalio/contrib/streams/nexus/__init__.py index 21fde742fa..bea08be752 100644 --- a/temporalio/contrib/streams/nexus/__init__.py +++ b/temporalio/contrib/streams/nexus/__init__.py @@ -42,13 +42,13 @@ from the owning Workflow. A caller Workflow reads such a stream with :class:`StreamReader`: it waits on -the operation's progress and reads through this service. By default each -read is a Nexus operation of the Workflow, so History records each batch and -a replay hands over the same batches with no store at hand. This works on -any server, Temporal Cloud included. ``replay_from_store=True`` is the -opt-in. It reads in a Local Activity through a -:class:`StreamDirectTransport` that :class:`StreamDirectReads` maps the -endpoint to, and History keeps only each read's range and digest. +the operation's progress and reads through this service. By default it reads +in a Local Activity through a :class:`StreamDirectTransport` that +:class:`StreamDirectReads` maps the endpoint to, and History keeps only each +read's range and digest. With ``replay_from_store=False`` each read is a +Nexus operation of the Workflow, so History records each batch and a replay +hands over the same batches with no store at hand. That works on any server, +Temporal Cloud included. :class:`ClientTransport` calls ``ExecuteSyncNexusOperation`` with the caller's own client, and :class:`HttpTransport` the HTTP ingress of a self-hosted server. diff --git a/temporalio/contrib/streams/nexus/_reader.py b/temporalio/contrib/streams/nexus/_reader.py index b35a7de79d..5ac75a6888 100644 --- a/temporalio/contrib/streams/nexus/_reader.py +++ b/temporalio/contrib/streams/nexus/_reader.py @@ -8,12 +8,12 @@ nothing new. Every append notifies the operation's callers, and the server folds the notifications of a busy stream into one per debounce window. -By default every read is a Nexus operation of the Workflow, so History -records what each read answered, and a replay hands over the same batches at -the same points. With ``replay_from_store=True`` each read runs in a Local -Activity, History keeps only the range that crossed and its digest, and a -replay reads that range again from the owner's store. That needs a -direct-read transport for the endpoint. +By default, ``replay_from_store=True``, each read runs in a Local Activity, +History keeps only the range that crossed and its digest, and a replay reads +that range again from the owner's store. It needs a direct-read transport +for the endpoint. With ``replay_from_store=False`` every read is a Nexus +operation of the Workflow, so History records what each read answered, and a +replay hands over the same batches at the same points. """ from __future__ import annotations @@ -241,10 +241,10 @@ class StreamReader(Generic[T]): such read costs a timer and a read in History, about 290 a day for an idle stream. - By default every read is a Nexus operation result in the caller's - History, about three events and up to one read answer (1 MiB of records) - each. With ``replay_from_store=True`` a read is one Local Activity marker - of a few hundred bytes, as below. To read a long stream, Continue-as-New + By default a read is one Local Activity marker of a few hundred bytes, as + below. With ``replay_from_store=False`` every read is a Nexus operation + result in the caller's History, about three events and up to one read + answer (1 MiB of records) each. To read a long stream, Continue-as-New before History grows too large: pass :attr:`position` to the new run, start the operation again there, and build the reader with ``after=position``. The position is at most about 22 KiB. After a @@ -297,7 +297,7 @@ def __init__( after: Cursor | ReadPosition | None = None, drain_limit: timedelta = timedelta(seconds=60), idle_wait: timedelta = timedelta(seconds=2), - replay_from_store: bool = False, + replay_from_store: bool = True, follow_links: bool = False, max_depth: int = 4, max_open: int = 16, @@ -324,13 +324,13 @@ def __init__( it well below the 10 second timeout of a Nexus request. replay_from_store: Read in a Local Activity and record only the range of each read and its digest, so the records stay out - of History. It needs :class:`StreamDirectReads` on the - Worker's client with a transport for ``endpoint``, and on any - Replayer. Without one, the read raises + of History. The default. It needs :class:`StreamDirectReads` + on the Worker's client with a transport for ``endpoint``, and + on any Replayer. Without one, the read raises :class:`temporalio.contrib.streams.StreamUnsupportedError`, - which names this option. ``False``, the default, reads - through Nexus operations and records each read's answer, - records included, so it needs no transport. + which names this option. ``False`` reads through Nexus + operations and records each read's answer, records included, + so it needs no transport. follow_links: Follow the stream's link records too, through the stream service's ``follow``. Read the items with :meth:`next_followed` in place of :meth:`next`. Each item @@ -750,8 +750,8 @@ async def _direct_follow(self, input: FollowInput) -> FollowOutput: if DIRECT_FOLLOW_ACTIVITY not in str(error): raise raise StreamUnsupportedError( - f"a stream follow with replay_from_store=True follows directly from " - f"endpoint {self._endpoint!r}, and the Worker has no " + "a stream follow with replay_from_store=True (the default) follows " + f"directly from endpoint {self._endpoint!r}, and the Worker has no " "StreamDirectReads; register it on the Worker's client with a " "transport for the endpoint, or pass replay_from_store=False to the " "StreamReader" @@ -800,7 +800,7 @@ async def _direct_read(self, input: ReadInput) -> ReadOutput: if DIRECT_READ_ACTIVITY not in str(error): raise raise StreamUnsupportedError( - f"a stream read with replay_from_store=True reads " + "a stream read with replay_from_store=True (the default) reads " f"directly from endpoint {self._endpoint!r}, and the Worker has no " "StreamDirectReads; register it on the Worker's client with a " "transport for the endpoint, or pass replay_from_store=False to the "