[Jetty] Added Workflow reads of a Stream on both backends.

temporalio/contrib/streams/_access.py

@@ -128,8 +128,9 @@ class ReadSpec:
 
 def read_spec(
     handle: StreamHandle[Any],
+    # Any is a Topic of Workflow code, which this module can't import.
     topics: Sequence[
-        str | TopicDefinition[Any] | _AsyncTopic[Any] | SyncTopicBase[Any]
+        str | TopicDefinition[Any] | _AsyncTopic[Any] | SyncTopicBase[Any] | Any
     ],
     *,
     prefixes: Sequence[str],
@@ -141,6 +142,8 @@ def read_spec(
 ) -> ReadSpec:
     """The Topic names of a read, the decode type of each known Topic, and its options.
 
+    A Topic object of any context, such as a ``WorkflowTopic``, gives its name and type.
+
     Raises:
         ValueError: ``last`` is not above zero, or it comes with ``from_offset``.
     """
@@ -149,14 +152,14 @@ def read_spec(
     types = handle._known_types()
     names = []
     for topic in topics:
-        if isinstance(topic, (_AsyncTopic, SyncTopicBase)):
-            name, topic_type = topic.name, topic.type
-        elif isinstance(topic, TopicDefinition):
+        if isinstance(topic, TopicDefinition):
             name, topic_type = handle._topic_spec(topic, None)
-        else:
+        elif isinstance(topic, str):
             # A read names a Topic outside the definition with no type, and its values
             # decode with the read's default type.
             name, topic_type = topic, types.get(topic)
+        else:
+            name, topic_type = topic.name, topic.type
         names.append(name)
         if topic_type is not None:
             types[name] = topic_type

temporalio/contrib/streams/_errors.py

@@ -142,6 +142,31 @@ def from_call_error(err: Exception) -> StreamError:
     )
 
 
+_ERROR_CLASSES: list[type[StreamError]] = [StreamError, *_BY_REASON.values()]
+_BY_NAME: dict[str, type[StreamError]] = {cls.__name__: cls for cls in _ERROR_CLASSES}
+
+
+def from_application_error(
+    err: temporalio.exceptions.ApplicationError,
+) -> StreamError | None:
+    """The Stream error that an application error of a failed call carries, if any.
+
+    A Stream error's type is its class name, and its first detail is its reason. A server may
+    give the reason as the type, or at the start of the message, such as ``stream_closed: ...``.
+    """
+    token = err.message.split(":", 1)[0].strip()
+    reason = err.details[0] if err.details and isinstance(err.details[0], str) else ""
+    cls = _BY_NAME.get(err.type or "")
+    if cls is None:
+        reason = next(
+            (name for name in (err.type or "", token) if name in _BY_REASON), ""
+        )
+        cls = _BY_REASON.get(reason)
+    if cls is None:
+        return None
+    return cls(err.message, reason=reason)
+
+
 def from_reason(reason: str, message: str) -> StreamError:
     """The Stream error for a reason token."""
     return _BY_REASON.get(reason, StreamError)(message, reason=reason)

temporalio/contrib/streams/_workflow_reader.py

@@ -0,0 +1,191 @@
+"""Reads of a Stream in Workflow code.
+
+Each page is a call whose answer is in History, so a replay reads the same items. A Stream
+plugin's outbound interceptor reads a page with ``_temporal_stream_read_page``.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator, Mapping, Sequence
+from datetime import datetime, timedelta
+from typing import Any
+
+import temporalio.api.streamservice.v1 as stream_pb
+import temporalio.common
+import temporalio.converter
+import temporalio.exceptions
+import temporalio.nexus.system
+import temporalio.workflow
+from temporalio.contrib.streams import _convert, _errors
+from temporalio.contrib.streams._items import Closed, End, Record, Superseded
+from temporalio.contrib.streams.system import models
+
+Item = Record[Any] | Superseded | Closed
+
+# The service waits at most this long for a Stream that nobody created yet, per call.
+_MAX_CREATE_WAIT = timedelta(seconds=60)
+# A backend that can't park a read pauses on a durable timer after a page with no Record,
+# from the first pause up to the last.
+_FIRST_PAUSE = timedelta(seconds=1)
+_LAST_PAUSE = timedelta(seconds=60)
+
+
+def reader_of(backend: str) -> Any:
+    """The outbound interceptor of the Stream plugin with this backend name."""
+    outbound = getattr(temporalio.workflow._Runtime.current(), "_outbound", None)
+    while outbound is not None:
+        if (
+            getattr(outbound, "_temporal_stream_transport", False)
+            and getattr(outbound, "_temporal_stream_backend_name", None) == backend
+        ):
+            return outbound
+        outbound = getattr(outbound, "next", None)
+    raise RuntimeError(f"No Stream plugin of the Worker has the backend {backend}")
+
+
+async def read(
+    reader: Any,
+    ref: stream_pb.StreamRef,
+    *,
+    topics: Sequence[str] | None,
+    prefixes: Sequence[str],
+    types: Mapping[str, Any],
+    from_offset: int | End | None,
+    wait_new_records: bool,
+    create_wait: timedelta | None,
+) -> AsyncGenerator[Item, None]:
+    """Give the items of a Stream in order, until a close, or the newest Record when
+    ``wait_new_records`` is False.
+    """
+    request = stream_pb.ReadRecordsRequest(
+        stream=ref, topics=list(topics or []), topic_prefixes=list(prefixes)
+    )
+    if from_offset is End.END:
+        # END is only the first page's start. Each answer's next offset pins later pages.
+        request.start = stream_pb.ReadStart.READ_START_END
+    elif from_offset:
+        request.from_offset = from_offset
+    else:
+        request.start = stream_pb.ReadStart.READ_START_BEGINNING
+    create_until = temporalio.workflow.now() + create_wait if create_wait else None
+    pause = _FIRST_PAUSE
+    while True:
+        if temporalio.workflow.unsafe.is_read_only():
+            raise _errors.ReadOnlyContextError("Can't read a Stream in read-only code")
+        page_request = stream_pb.ReadRecordsRequest()
+        page_request.CopyFrom(request)
+        if wait_new_records:
+            page_request.wait.FromTimedelta(reader._temporal_stream_read_wait)
+        left = _left(create_until)
+        if left:
+            page_request.create_wait.FromTimedelta(min(left, _MAX_CREATE_WAIT))
+        try:
+            page = await reader._temporal_stream_read_page(page_request)
+        except _errors.StreamNotFoundError:
+            if not _left(create_until):
+                raise
+            await temporalio.workflow.sleep(pause)
+            pause = min(pause * 2, _LAST_PAUSE)
+            continue
+        for record in page.records:
+            item = workflow_item(record, types)
+            if item is None:
+                continue
+            yield item
+            if isinstance(item, Closed):
+                return
+        if page.done:
+            return
+        if page.next_offset:
+            request.from_offset = page.next_offset
+            request.ClearField("start")
+        if page.records:
+            pause = _FIRST_PAUSE
+            continue
+        if not wait_new_records:
+            return
+        if not reader._temporal_stream_parks_reads:
+            # The server wake comes later. Until then a quiet Stream costs one timer and one
+            # read per pause.
+            await temporalio.workflow.sleep(pause)
+            pause = min(pause * 2, _LAST_PAUSE)
+
+
+def _left(until: datetime | None) -> timedelta | None:
+    if until is None:
+        return None
+    left = until - temporalio.workflow.now()
+    return left if left > timedelta() else None
+
+
+def workflow_item(
+    record: stream_pb.StreamRecord, types: Mapping[str, Any]
+) -> Item | None:
+    """One read item, with the body decoded by the Workflow's converter and its Topic's type.
+
+    The codec already ran outside the sandbox, on the page.
+    """
+    converter = temporalio.workflow.payload_converter()
+    # The generated converter makes the producer and the supersession models, and the body
+    # stays a Payload that the Topic's type decodes.
+    with temporalio.nexus.system._user_converter_context(
+        temporalio.nexus.system._SystemNexusUserConverters(
+            payload_converter=_convert._RawPayloadConverter(),
+            failure_converter=_failure_converter(),
+        )
+    ):
+        model = models._StreamRecordTransferTypeConverter().from_transfer_type(
+            record, models.StreamRecord
+        )
+    offset = model.offset or 0
+    if model.kind == stream_pb.StreamRecordKind.STREAM_RECORD_KIND_DATA:
+        topic = model.topic or ""
+        return Record(
+            topic=topic,
+            data=_decode(converter, model.body, types.get(topic)),
+            offset=offset,
+            producer=model.producer,
+        )
+    if model.kind == stream_pb.StreamRecordKind.STREAM_RECORD_KIND_SUPERSEDED:
+        if model.supersession is None:
+            return None
+        return Superseded(offset=offset, supersession=model.supersession)
+    if model.kind == stream_pb.StreamRecordKind.STREAM_RECORD_KIND_CLOSE:
+        return Closed(
+            offset=offset,
+            result=_decode(converter, model.close_result, None),
+            failure=model.close_failure,
+        )
+    return None
+
+
+def _decode(
+    converter: temporalio.converter.PayloadConverter, body: Any, type_hint: Any
+) -> Any:
+    if not isinstance(body, temporalio.common.RawValue):
+        return body
+    return converter.from_payload(body.payload, type_hint)
+
+
+def _failure_converter() -> temporalio.converter.FailureConverter:
+    runtime = temporalio.workflow._Runtime.current()
+    return getattr(
+        runtime,
+        "_workflow_context_failure_converter",
+        temporalio.converter.DefaultFailureConverter(),
+    )
+
+
+def error_of(err: BaseException) -> BaseException:
+    """The Stream error that a failed page carries as its cause, or the error itself.
+
+    A page's failure is in History, so a replay raises the same error.
+    """
+    cause: BaseException | None = err
+    while cause is not None:
+        if isinstance(cause, temporalio.exceptions.ApplicationError):
+            stream_error = _errors.from_application_error(cause)
+            if stream_error is not None:
+                return stream_error
+        cause = cause.__cause__
+    return err

temporalio/contrib/streams/native.py

@@ -8,12 +8,26 @@
 
 from datetime import timedelta
 
+import temporalio.api.streamservice.v1 as stream_pb
 import temporalio.bridge.client
 import temporalio.bridge.streams
+import temporalio.exceptions
+import temporalio.nexus.system
 import temporalio.service
+import temporalio.worker
+import temporalio.workflow
+from temporalio.contrib.streams import _workflow_reader
 from temporalio.contrib.streams._backend import _ExtensionStreamBackend
 from temporalio.contrib.streams.proto import v1 as ext_pb
-from temporalio.contrib.streams.redis import _ExtensionStreamPlugin, _runtime_of
+from temporalio.contrib.streams.redis import (
+    _ExtensionStreamPlugin,
+    _runtime_of,
+    _WorkflowOutbound,
+)
+
+_SERVICE = "temporal.api.streamservice.v1.StreamService"
+
+_WAIT_COMPLETED = temporalio.workflow.NexusOperationCancellationType.WAIT_COMPLETED
 
 
 class _ConnectionRuntime:
@@ -55,6 +69,37 @@ async def _bridge_client(
     return await connect()
 
 
+class _NativeWorkflowOutbound(_WorkflowOutbound):
+    # The server parks a read at the end of the Stream and completes it on an append or a
+    # close, so a reader needs no timer.
+    _temporal_stream_parks_reads = True
+    _temporal_stream_read_wait = timedelta(seconds=60)
+
+    async def _temporal_stream_read_page(
+        self, request: stream_pb.ReadRecordsRequest
+    ) -> stream_pb.ReadRecordsResponse:
+        # The page is a System Nexus operation, so its answer is in History.
+        handle = await self.next.start_system_nexus_operation(
+            temporalio.worker.StartNexusOperationInput(
+                endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT,
+                service=_SERVICE,
+                operation="ReadRecords",
+                input=request,
+                output_type=stream_pb.ReadRecordsResponse,
+                schedule_to_close_timeout=None,
+                schedule_to_start_timeout=None,
+                start_to_close_timeout=None,
+                cancellation_type=_WAIT_COMPLETED,
+                headers=None,
+                summary=None,
+            )
+        )
+        try:
+            return await handle
+        except temporalio.exceptions.NexusOperationError as err:
+            raise _workflow_reader.error_of(err) from None
+
+
 class NativeStreamPlugin(_ExtensionStreamPlugin):
     """Carries Stream calls to the Temporal server, which keeps the Streams.
 
@@ -69,6 +114,7 @@ class NativeStreamPlugin(_ExtensionStreamPlugin):
         This API is experimental.
     """
 
+    _outbound_class = _NativeWorkflowOutbound
     _server_closes_with_run = True
 
     def _new_backend(