From d220153d6204c4a80619a62cfedcb2b7a3c1fb1d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 01:55:01 -0700 Subject: [PATCH 1/4] Added the stream accessors for activities and clients. activity.stream_handle and client.get_stream_handle open a stream by ref or by topic, and create_stream makes a standalone one, so every context reaches a stream the same way. --- temporalio/activity.py | 128 +++++++++++ temporalio/client/_client.py | 142 +++++++++++++ temporalio/streams/__init__.py | 15 +- temporalio/streams/_provider.py | 25 +++ temporalio/streams/_ref.py | 91 +++++++- temporalio/streams/providers/memory.py | 281 +++++++++++++++++++++---- temporalio/worker/_activity.py | 8 + temporalio/worker/_worker.py | 1 + 8 files changed, 643 insertions(+), 48 deletions(-) diff --git a/temporalio/activity.py b/temporalio/activity.py index 768c8d108..40821db39 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -20,6 +20,7 @@ from typing import ( TYPE_CHECKING, Any, + Literal, NoReturn, overload, ) @@ -29,9 +30,11 @@ import temporalio.bridge.proto.activity_task import temporalio.common import temporalio.converter +import temporalio.streams from temporalio.converter._payload_converter import ( _TemporalTransferTypePayloadConverter, ) +from temporalio.streams._ref import open_ref from .types import CallableType @@ -209,6 +212,10 @@ class _Context: runtime_metric_meter: temporalio.common.MetricMeter | None client: Client | None cancellation_details: _ActivityCancellationDetailsHolder + stream_provider: temporalio.streams.StreamProvider | None = None + # A ``def`` activity is handed no client and no stream provider, so what + # it is missing cannot be read off the fields that are absent. + sync: bool = False _logger_details: Mapping[str, Any] | None = None _payload_converter: temporalio.converter.PayloadConverter | None = None _metric_meter: temporalio.common.MetricMeter | None = None @@ -294,6 +301,127 @@ def client() -> Client: return client +def stream_handle( + workflow_id: str | temporalio.streams.StreamRef | None = None, + *, + run_id: str | None = None, + scope: Literal["workflow", "activity"] | None = None, +) -> temporalio.streams.StreamHandle: + """Return a stream handle from the provider the worker was given. + + A :py:class:`temporalio.streams.StreamRef` in place of ``workflow_id``, + such as one this activity received as an argument, opens the stream it + names, whatever owns it, and takes no other argument; the handle's calls + that name no topic then address the ref's topic. + + Which stream a call with no ``workflow_id`` reaches is decided by where + the activity runs, never by what exists: + + - In an activity a workflow scheduled, it is that workflow's stream, + pinned to the run the activity belongs to. + - In a standalone activity, it is the activity's own stream. + - ``scope="activity"`` gives an activity a workflow scheduled its own + streams instead, apart from the workflow's. ``scope="workflow"`` asks + for the workflow explicitly, and a standalone activity has none. + + The rule is static because a stream is created by its first write, so a + rule that looked for one would send attempt 1 to the workflow and a retry + to the stream attempt 1 created. An activity's own streams are one per + activity execution, not per attempt: a retry writes to the same stream, + under a new attempt, and they end when the activity reaches a terminal + status. + + Name a ``workflow_id`` to address another workflow; ``run_id`` then pins + the handle to one run and its absence follows the execution chain. A + ``read``, ``latest`` or ``producer`` that names no topic addresses the + owner's default topic, :py:data:`temporalio.streams.DEFAULT_TOPIC`, the + one :py:func:`temporalio.workflow.stream_reader` and + :py:func:`temporalio.workflow.stream_writer` use without a topic. A + ``read`` starts at :py:data:`temporalio.streams.BEGINNING`, at + :py:data:`temporalio.streams.END` or at the last ``N`` records with + ``last=N``. See :py:mod:`temporalio.streams`. + + Like :py:func:`client`, this is only available in ``async def`` + activities. + + Args: + workflow_id: Another workflow whose stream to address, or a + :py:class:`temporalio.streams.StreamRef` naming the stream. + run_id: The run of ``workflow_id`` to pin to. + scope: ``"activity"`` for this activity's own streams, + ``"workflow"`` for its workflow's. Without it the rule above + decides. + + Returns: + :py:class:`temporalio.streams.StreamHandle` for use in the current + activity. + + Raises: + RuntimeError: When the client is not available, which is what a + ``def`` activity gets, or when ``scope="workflow"`` is asked of + an activity that belongs to no workflow. + temporalio.streams.StreamUnsupportedError: The worker has no stream + provider, or its provider cannot hold a stream an activity owns. + Register one with ``Client.connect(plugins=[provider])`` or + ``Worker(plugins=[provider])``. + ValueError: ``run_id`` was given without ``workflow_id``, + ``scope="activity"`` with one, or a ref with either. + """ + context = _Context.current() + if context.sync: + # A sync activity is handed neither a client nor a provider. Saying + # the worker has none would send the reader to fix a registration + # that is not the problem. + raise RuntimeError( + "No stream handle available. Stream handles are only available in " + "`async def` activities; not in `def` activities, which are handed no " + "client to reach the store with." + ) + provider = context.stream_provider + if provider is None: + raise temporalio.streams.StreamUnsupportedError( + "no stream provider is configured on this worker; register one with " + "Client.connect(plugins=[provider]) or Worker(plugins=[provider])" + ) + if isinstance(workflow_id, temporalio.streams.StreamRef): + if run_id is not None or scope is not None: + raise ValueError( + "a StreamRef names the stream in full, so it takes no run_id or scope" + ) + return open_ref(provider, client(), workflow_id) + if workflow_id is not None: + if scope == "activity": + raise ValueError( + "scope='activity' addresses this activity's own streams, so it takes " + "no workflow_id" + ) + return provider.get_stream_handle(client(), workflow_id, run_id=run_id) + if run_id is not None: + raise ValueError("run_id needs a workflow_id") + info = context.info() + if scope is None: + scope = "workflow" if info.in_workflow else "activity" + if scope == "workflow": + if info.workflow_id is None: + raise RuntimeError( + "this activity belongs to no workflow, so name the workflow_id to " + "address, or leave scope unset for the activity's own streams" + ) + return provider.get_stream_handle( + client(), info.workflow_id, run_id=info.workflow_run_id + ) + if info.workflow_id is not None: + return provider.get_activity_stream_handle( + client(), + info.activity_id, + workflow_id=info.workflow_id, + run_id=info.workflow_run_id, + ) + return provider.get_activity_stream_handle( + client(), info.activity_id, run_id=info.activity_run_id + ) + + def in_activity() -> bool: """Whether the current code is inside an activity. diff --git a/temporalio/client/_client.py b/temporalio/client/_client.py index 6bdf71469..5330bf7af 100644 --- a/temporalio/client/_client.py +++ b/temporalio/client/_client.py @@ -40,6 +40,7 @@ ServiceClient, TLSConfig, ) +from temporalio.streams._ref import open_ref from ..common import HeaderCodecBehavior from ..types import ( @@ -166,6 +167,7 @@ async def connect( grpc_compression: GrpcCompression = GrpcCompression.GZIP, payload_limits: PayloadLimitsConfig = PayloadLimitsConfig(), header_codec_behavior: HeaderCodecBehavior = HeaderCodecBehavior.NO_CODEC, + stream_provider: temporalio.streams.StreamProvider | None = None, ) -> Self: """Connect to a Temporal server. @@ -231,6 +233,11 @@ async def connect( payload_limits: Warning thresholds for outbound payload/memo sizes. Over-threshold fields are logged but still sent. Set a threshold to 0 to disable it. header_codec_behavior: Encoding behavior for headers sent by the client. + stream_provider: Experimental. The stream provider + :py:meth:`get_stream_handle` opens handles from, see + :py:mod:`temporalio.streams`. A provider that is also a + :py:class:`Plugin` sets this itself when passed in ``plugins``, + and workers built from this client inherit it. """ connect_config = temporalio.service.ConnectConfig( target_host=target_host, @@ -267,6 +274,7 @@ def make_lambda( default_workflow_query_reject_condition=default_workflow_query_reject_condition, header_codec_behavior=header_codec_behavior, plugins=plugins, + stream_provider=stream_provider, ) def __init__( @@ -899,6 +907,139 @@ def get_workflow_handle( result_type=result_type, ) + def get_stream_handle( + self, + workflow_id: str | temporalio.streams.StreamRef | None = None, + *, + run_id: str | None = None, + activity_id: str | None = None, + stream_id: str | None = None, + ) -> temporalio.streams.StreamHandle: + """Get a handle on a stream from the provider registered on this client. + + Mirrors :py:meth:`get_workflow_handle`: without ``run_id`` the handle + follows the workflow's execution chain across continue-as-new, with + one it is pinned to that run. With ``activity_id`` the handle is on + the streams that activity owns: a standalone activity's when + ``workflow_id`` is left out, and ``run_id`` then pins the activity's + run, or an activity that ``workflow_id`` scheduled. With + ``stream_id`` it is on a standalone stream, one with an id of its own + and no owner, which :py:meth:`create_stream` made; it takes no other + argument. A :py:class:`temporalio.streams.StreamRef` in place of + ``workflow_id`` opens the stream the ref names, whatever owns it, and + takes no other argument either: the handle's calls that name no topic + then address the ref's topic. The provider is the one registered + with ``plugins=[provider]`` at :py:meth:`connect`, or passed as + ``stream_provider``. The handle's ``read``, ``latest`` and + ``producer`` take a topic, and without one address the owner's + default topic, :py:data:`temporalio.streams.DEFAULT_TOPIC`. A + ``read`` starts at :py:data:`temporalio.streams.BEGINNING`, at + :py:data:`temporalio.streams.END` or at the last ``N`` records with + ``last=N``, and resumes only after a cursor it was handed. See + :py:mod:`temporalio.streams`. + + Args: + workflow_id: Workflow ID whose stream to get a handle to, the + workflow that scheduled ``activity_id``, or a + :py:class:`temporalio.streams.StreamRef` naming the stream. + run_id: Run ID to pin the handle to. + activity_id: Activity ID whose own streams to get a handle to. + stream_id: ID of the standalone stream to get a handle to. + + Returns: + The stream handle. + + Raises: + ValueError: No owner was named, or a ref or ``stream_id`` was + given together with another argument. + temporalio.streams.StreamUnsupportedError: No stream provider is + registered on this client, or it cannot hold a stream an + activity owns or a stream without an owner. + """ + provider = self._config.get("stream_provider") + if provider is None: + raise temporalio.streams.StreamUnsupportedError( + "no stream provider is registered on this client; connect with " + "plugins=[provider]" + ) + if isinstance(workflow_id, temporalio.streams.StreamRef): + if run_id is not None or activity_id is not None or stream_id is not None: + raise ValueError( + "a StreamRef names the stream in full, so it takes no run_id, " + "activity_id or stream_id" + ) + return open_ref(provider, self, workflow_id) + if stream_id is not None: + if workflow_id is not None or run_id is not None or activity_id is not None: + raise ValueError( + "stream_id names a standalone stream, which has no workflow_id, " + "run_id or activity_id" + ) + return provider.get_standalone_stream_handle(self, stream_id) + if activity_id is not None: + return provider.get_activity_stream_handle( + self, activity_id, workflow_id=workflow_id, run_id=run_id + ) + if workflow_id is None: + raise ValueError( + "name the workflow_id, the activity_id or the stream_id to address" + ) + return provider.get_stream_handle(self, workflow_id, run_id=run_id) + + async def create_stream( + self, + stream_id: str, + *, + retention: timedelta | None = None, + max_records: int | None = None, + max_bytes: int | None = None, + ) -> temporalio.streams.StreamHandle: + """Create a standalone stream and get a handle on it. + + A standalone stream has an id of its own and no owner, so it is + created here on purpose rather than by its first write, and it is + sealed on purpose with the handle's ``close()``, after which appends + are refused and the retained records stay readable. The three policy + arguments bound what it retains: records older than ``retention``, + beyond the newest ``max_records`` or past ``max_bytes`` of stored + records are dropped, and ``None`` leaves a bound to the provider's + default. Creating a stream that exists with the same policy returns a + handle on it, so a retried create is harmless. Another process + reaches the stream with ``get_stream_handle(stream_id=...)`` or with + the handle's ``ref()``. + + Args: + stream_id: ID of the stream to create. + retention: How long a record is kept. + max_records: How many of the newest records are kept. + max_bytes: How many bytes of records are kept. + + Returns: + A handle on the new or existing stream. + + Raises: + ValueError: ``stream_id`` is empty, a bound is not positive, or + the stream exists with a different policy. + temporalio.streams.StreamUnsupportedError: No stream provider is + registered on this client, or it cannot hold a stream without + an owner. + """ + provider = self._config.get("stream_provider") + if provider is None: + raise temporalio.streams.StreamUnsupportedError( + "no stream provider is registered on this client; connect with " + "plugins=[provider]" + ) + if not stream_id: + raise ValueError("stream_id must not be empty") + return await provider.create_standalone_stream( + self, + stream_id, + retention=retention, + max_records=max_records, + max_bytes=max_bytes, + ) + def get_workflow_handle_for( self, workflow: ( @@ -3295,6 +3436,7 @@ class ClientConnectConfig(TypedDict, total=False): grpc_compression: GrpcCompression payload_limits: PayloadLimitsConfig header_codec_behavior: HeaderCodecBehavior + stream_provider: temporalio.streams.StreamProvider | None class ClientConfig(TypedDict, total=False): diff --git a/temporalio/streams/__init__.py b/temporalio/streams/__init__.py index 9b7cb2df2..8b1d35f1c 100644 --- a/temporalio/streams/__init__.py +++ b/temporalio/streams/__init__.py @@ -43,12 +43,15 @@ ``Client.connect(plugins=[provider])``; workers built from that client inherit it, and ``Worker(plugins=[provider])`` or ``Replayer(plugins=[provider])`` registers it on a worker alone. Each context then asks for its stream the -same way. Workflow code uses ``temporalio.workflow.stream_reader`` and -``temporalio.workflow.stream_writer``. An activity uses -``temporalio.activity.stream_handle``, which is its own workflow pinned -to its run unless told otherwise. Any process holding a client uses -``temporalio.client.Client.get_stream_handle``, which mirrors -``get_workflow_handle``. The explicit form, +same way. Workflow code uses :func:`temporalio.workflow.stream_reader` and +:func:`temporalio.workflow.stream_writer`. An activity uses +:func:`temporalio.activity.stream_handle`: an activity a workflow scheduled +reaches that workflow's stream pinned to its run, a standalone activity +reaches its own, and ``scope="activity"`` gives the first kind its own +streams too. Any process holding a client uses +:meth:`temporalio.client.Client.get_stream_handle`, which mirrors +``get_workflow_handle`` and takes an ``activity_id`` for an activity's +streams. The explicit form, ``provider.get_stream_handle(client, workflow_id)``, stays for a process that talks to two stores. This module keeps the shared types, the errors and the protocols a provider implements; nothing here that workflow code imports does diff --git a/temporalio/streams/_provider.py b/temporalio/streams/_provider.py index 98c4851e0..7ef3a6b6f 100644 --- a/temporalio/streams/_provider.py +++ b/temporalio/streams/_provider.py @@ -393,6 +393,31 @@ def get_stream_handle( """ ... + def get_activity_stream_handle( + self, + client: Client, + activity_id: str, + *, + workflow_id: str | None = None, + run_id: str | None = None, + ) -> StreamHandle: + """A handle on the streams an activity owns. + + Without ``workflow_id`` the activity is a standalone one, an execution + of its own, and ``run_id`` pins one run of it. With ``workflow_id`` it + is an activity that workflow scheduled, and ``run_id`` pins the + workflow's run. Either way these streams are apart from any + workflow's: a topic here and the same topic on the workflow's handle + are two streams. A retry of the activity writes to the same streams, + and a read ends when the activity reaches a terminal status, not when + an attempt fails. + + Raises: + StreamUnsupportedError: The provider's store cannot hold a stream + an activity owns. + """ + ... + async def create_standalone_stream( self, client: Client, diff --git a/temporalio/streams/_ref.py b/temporalio/streams/_ref.py index 3d0eb0a5d..b4dcd126c 100644 --- a/temporalio/streams/_ref.py +++ b/temporalio/streams/_ref.py @@ -12,12 +12,22 @@ from __future__ import annotations import dataclasses +from collections.abc import AsyncGenerator from dataclasses import dataclass -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal +from temporalio.streams._record import BEGINNING, Cursor, StreamRecord from temporalio.streams._topic import DEFAULT_TOPIC, StreamTopic, resolve_topic -__all__ = ["StreamOwnerKind", "StreamRef"] +if TYPE_CHECKING: + from temporalio.client import Client + from temporalio.streams._provider import ( + StreamHandle, + StreamProducer, + StreamProvider, + ) + +__all__ = ["StreamOwnerKind", "StreamRef", "open_ref"] StreamOwnerKind = Literal["workflow", "activity", "standalone"] """What owns a stream: a workflow, an activity, or the stream itself.""" @@ -140,3 +150,80 @@ def with_topic(self, topic: str | StreamTopic[Any] | None) -> StreamRef: """ name, _ = resolve_topic(topic) return dataclasses.replace(self, topic=name) + + +def open_ref(provider: StreamProvider, client: Client, ref: StreamRef) -> StreamHandle: + """The handle ``ref`` names, on ``provider``. + + The ref's kind picks the provider call that opens the owner, and its + topic becomes the handle's default, so a call that names none addresses + the stream the ref names. A provider that cannot host that owner kind + raises :class:`temporalio.streams.StreamUnsupportedError` from the call + that would have opened it. + """ + if ref.kind == "workflow": + assert ref.workflow_id is not None + handle = provider.get_stream_handle(client, ref.workflow_id, run_id=ref.run_id) + elif ref.kind == "activity": + assert ref.activity_id is not None + handle = provider.get_activity_stream_handle( + client, ref.activity_id, workflow_id=ref.workflow_id, run_id=ref.run_id + ) + else: + assert ref.stream_id is not None + handle = provider.get_standalone_stream_handle(client, ref.stream_id) + return _RefHandle(handle, ref) + + +class _RefHandle: + """A provider's handle whose default topic is the one a ref names. + + Every call passes through unchanged when it names a topic; one that + names none gets the ref's. The wrapper exists so a ref can address a + stream without every provider learning about refs. + """ + + def __init__(self, inner: StreamHandle, ref: StreamRef) -> None: + self._inner = inner + self._ref = ref + + def _topic(self, topic: str | StreamTopic[Any] | None) -> str | StreamTopic[Any]: + return self._ref.topic if topic is None else topic + + def read( + self, + *, + topic: str | StreamTopic[Any] | None = None, + after: Cursor = BEGINNING, + last: int | None = None, + result_type: type | None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: + # The protocol's overloads each take one shape of topic and + # result_type; a passthrough hands over whatever it was given. + inner: Any = self._inner + return inner.read( + topic=self._topic(topic), after=after, last=last, result_type=result_type + ) + + async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: + return await self._inner.latest(topic=self._topic(topic)) + + def producer( + self, + *, + topic: str | StreamTopic[Any] | None = None, + producer_id: str = "", + attempt: int = 0, + ) -> StreamProducer[Any]: + inner: Any = self._inner + return inner.producer( + topic=self._topic(topic), producer_id=producer_id, attempt=attempt + ) + + def ref(self, *, topic: str | StreamTopic[Any] | None = None) -> StreamRef: + if topic is None: + return self._ref + return self._inner.ref(topic=topic) + + async def close(self) -> None: + await self._inner.close() diff --git a/temporalio/streams/providers/memory.py b/temporalio/streams/providers/memory.py index 87f4d5116..ce8a4ae10 100644 --- a/temporalio/streams/providers/memory.py +++ b/temporalio/streams/providers/memory.py @@ -15,10 +15,21 @@ handle only decides which run's close ends a read. - It learns that a workflow closed by describing it, so a handle opened without a client reads until the caller closes it. +- An activity's own streams are keyed by the activity, not by its run. A + standalone activity's read ends when describing it shows a terminal + status. A workflow's activity is described through its workflow: the read + ends once the activity has been seen pending and is no longer, or the + workflow closed, so a read opened after the activity already finished + waits for the workflow. - It keeps every record until :meth:`MemoryStreams.truncate` drops the oldest ones, which stands in for a store's retention in tests. -- It does not host standalone streams; both standalone calls raise - :class:`temporalio.streams.StreamUnsupportedError`. +- A standalone stream lives here with its policy and a sealed flag. + ``retention``, ``max_records`` and ``max_bytes`` are applied when a record + is appended, so a stream nobody writes to keeps records past their + retention. A read on it ends when it is sealed and the tail delivered, and + a read, ``latest`` or ``producer`` on a stream id that does not exist + raises :class:`temporalio.streams.StreamNotFoundError` at the call rather + than waiting for the stream to be created. - The outside path encodes and decodes bodies through the client's data converter, codec and external storage included, and fingerprints a retry over the converted bytes first. The workflow half has no client, so a @@ -35,7 +46,9 @@ import asyncio import logging +import time from collections.abc import AsyncGenerator +from dataclasses import dataclass from datetime import timedelta from typing import Any, Generic, TypeVar @@ -43,13 +56,14 @@ import temporalio.converter from temporalio import workflow -from temporalio.client import Client, WorkflowExecutionStatus +from temporalio.client import ActivityExecutionStatus, Client, WorkflowExecutionStatus from temporalio.service import RPCError, RPCStatusCode from temporalio.streams._body import content_fingerprint, decode_body, encode_body from temporalio.streams._errors import ( + StreamClosedError, StreamCursorError, + StreamNotFoundError, StreamProducerError, - StreamUnsupportedError, ) from temporalio.streams._ids import topic_key from temporalio.streams._provider import ReadSource, WriteSink @@ -87,15 +101,45 @@ def _wake(future: asyncio.Future[None]) -> None: future.set_result(None) +@dataclass(frozen=True) +class _Policy: + """What a standalone stream retains, applied as records are appended.""" + + retention: timedelta | None = None + max_records: int | None = None + max_bytes: int | None = None + + def __post_init__(self) -> None: + if self.retention is not None and self.retention <= timedelta(0): + raise ValueError("retention must be positive") + if self.max_records is not None and self.max_records <= 0: + raise ValueError("max_records must be positive") + if self.max_bytes is not None and self.max_bytes <= 0: + raise ValueError("max_bytes must be positive") + + +class _Standalone: + """One standalone stream: its policy, its seal and its topics.""" + + def __init__(self, policy: _Policy) -> None: + self.policy = policy + self.sealed = False + self.topics: dict[str, _Topic] = {} + + class _Topic: """One topic's records, and the waiters parked on its tail.""" - def __init__(self) -> None: + def __init__(self, policy: _Policy | None = None, *, sealed: bool = False) -> None: + self.policy = policy + self.sealed = sealed # The retained records, the first of which sits at offset ``base``. # Offsets are never reused, so a cursor keeps naming the same record # after truncation drops the ones before it. self.base = 0 self.records: list[bytes] = [] + # When each retained record landed, for a retention policy. + self.stamps: list[float] = [] # Dedupe identity is (producer#attempt, first sequence of the append), # the same pair the storage providers use, mapped to where the batch # landed and a digest of what it held, so a repeat answers with the @@ -126,7 +170,10 @@ def append( Raises: StreamProducerError: ``(writer, sequence)`` is held with different content. + StreamClosedError: The stream was sealed. """ + if self.sealed: + raise StreamClosedError("the stream is closed and takes no more records") key = (writer or "", sequence) bodies = [wire.SerializeToString(deterministic=True) for wire in wires] if content is None: @@ -142,13 +189,43 @@ def append( ) return first, count first = self.head + now = time.time() self.records.extend(bodies) + self.stamps.extend([now] * len(bodies)) if writer is not None: self.seen[key] = (first, len(wires), content) + self._apply_policy(now) + self._wake_waiters() + return first, len(wires) + + def _wake_waiters(self) -> None: waiters, self._waiters = self._waiters, [] for loop, future in waiters: loop.call_soon_threadsafe(_wake, future) - return first, len(wires) + + def _apply_policy(self, now: float) -> None: + policy = self.policy + if policy is None: + return + drop = 0 + if policy.max_records is not None: + drop = max(drop, len(self.records) - policy.max_records) + if policy.max_bytes is not None: + held = sum(len(record) for record in self.records) + while drop < len(self.records) and held > policy.max_bytes: + held -= len(self.records[drop]) + drop += 1 + if policy.retention is not None: + floor = now - policy.retention.total_seconds() + while drop < len(self.records) and self.stamps[drop] < floor: + drop += 1 + if drop: + self._drop(drop) + + def seal(self) -> None: + """Take no more records, and let parked readers see the end.""" + self.sealed = True + self._wake_waiters() @property def head(self) -> int: @@ -161,9 +238,12 @@ def at(self, offset: int) -> bytes: def truncate(self, keep: int) -> None: """Drop all but the newest ``keep`` records.""" - drop = max(0, len(self.records) - keep) - self.base += drop - del self.records[:drop] + self._drop(max(0, len(self.records) - keep)) + + def _drop(self, count: int) -> None: + self.base += count + del self.records[:count] + del self.stamps[:count] async def wait_past(self, offset: int, timeout: float | None) -> None: """Wait until a record exists at ``offset``, or ``timeout`` passes.""" @@ -358,20 +438,31 @@ async def _write(self, wires: list[WireRecord]) -> Cursor: class MemoryStreamHandle: - """One workflow's stream from outside, with the shared reader rules.""" + """One owner's stream from outside, with the shared reader rules. + + The owner is ``workflow_id``'s workflow, or with ``activity_id`` an + activity: a standalone one without ``workflow_id``, or one that workflow + scheduled. With ``stream_id`` the handle is on a standalone stream, which + has no owner. + """ def __init__( self, streams: MemoryStreams, client: Client | None, - workflow_id: str, + workflow_id: str | None, run_id: str | None, + activity_id: str | None = None, + stream_id: str | None = None, ) -> None: - """Address ``workflow_id``'s topics in ``streams``.""" + """Address the owner's topics in ``streams``.""" self._streams = streams self._client = client self._workflow_id = workflow_id self._run_id = run_id + self._activity_id = activity_id + self._stream_id = stream_id + self._seen_pending = False self._converter = ( client.data_converter if client is not None @@ -386,14 +477,15 @@ def read( last: int | None = None, result_type: type | None = None, ) -> AsyncGenerator[StreamRecord[Any], None]: - """Yield records on ``topic`` from where the read starts until the workflow closes. + """Yield records on ``topic`` from where the read starts until the owner closes. ``END`` and ``last=`` are resolved by this call, against what the - topic holds when it is made. + topic holds when it is made. On a standalone stream the read ends + once the stream is sealed and the tail delivered. """ check_read_start(after, last) name, result_type = resolve_topic(topic, result_type) - store = self._streams._topic(self._workflow_id, name) + store = self._store(name) # Parsed here so a foreign cursor fails this call, not the first # iteration of the generator. start = self._streams._start(store, after, last) @@ -444,21 +536,50 @@ async def _read( else self._streams._poll.total_seconds(), ) + def _store(self, topic: str) -> _Topic: + if self._stream_id is not None: + return self._streams._standalone_topic(self._stream_id, topic) + if self._activity_id is not None: + return self._streams._activity_topic( + self._workflow_id, self._activity_id, topic + ) + assert self._workflow_id is not None + return self._streams._topic(self._workflow_id, topic) + async def _closed(self) -> bool: + if self._stream_id is not None: + return self._streams._standalone_stream(self._stream_id).sealed if self._client is None: return False - handle = self._client.get_workflow_handle( - self._workflow_id, run_id=self._run_id - ) try: - description = await handle.describe() + if self._activity_id is not None and self._workflow_id is None: + activity = await self._client.get_activity_handle( + self._activity_id, run_id=self._run_id + ).describe() + return activity.status != ActivityExecutionStatus.RUNNING + assert self._workflow_id is not None + description = await self._client.get_workflow_handle( + self._workflow_id, run_id=self._run_id + ).describe() except RPCError as error: if error.status == RPCStatusCode.NOT_FOUND: - # A producer may write before the workflow exists; there is + # A producer may write before the owner exists; there is # nothing to follow yet, so keep waiting. return False raise status = description.status + if self._activity_id is not None: + pending = any( + info.activity_id == self._activity_id + for info in description.raw_description.pending_activities + ) + if pending: + self._seen_pending = True + elif self._seen_pending: + return True + # The activity's streams are not the workflow's chain: a run that + # continued as new took its activities with it. + return status is not None and status != WorkflowExecutionStatus.RUNNING if status is None or status == WorkflowExecutionStatus.RUNNING: return False # Following the chain, a run that continued as new is not the end: @@ -470,7 +591,7 @@ async def _closed(self) -> bool: async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: """The cursor of the newest record on ``topic``, for following from now.""" name, _ = resolve_topic(topic) - head = self._streams._topic(self._workflow_id, name).head + head = self._store(name).head return mint_cursor(_PROVIDER, str(head - 1)) if head else BEGINNING def producer( @@ -482,22 +603,34 @@ def producer( ) -> MemoryProducer[Any]: """A producer on ``topic``; inside an activity its identity is the activity's.""" name, _ = resolve_topic(topic) - store = self._streams._topic(self._workflow_id, name) + store = self._store(name) producer_id, attempt = producer_identity(producer_id, attempt) return MemoryProducer(store, self._converter, name, producer_id, attempt) def ref(self, *, topic: str | StreamTopic[Any] | None = None) -> StreamRef: - """A ref to ``topic`` of this workflow's stream, pinned as this handle is.""" + """A ref to ``topic`` of this owner's stream, pinned as this handle is.""" + if self._stream_id is not None: + return StreamRef.for_standalone(self._stream_id, topic=topic) + if self._activity_id is not None: + return StreamRef.for_activity( + self._activity_id, + workflow_id=self._workflow_id, + run_id=self._run_id, + topic=topic, + ) + assert self._workflow_id is not None return StreamRef.for_workflow( self._workflow_id, run_id=self._run_id, topic=topic ) async def close(self) -> None: - """Refuse: a workflow's stream ends with the workflow, not by a caller.""" - raise ValueError( - "only a standalone stream can be closed; this handle is on a workflow's " - "stream, which ends when the workflow does" - ) + """Seal a standalone stream. An owned stream ends with its owner, not by a caller.""" + if self._stream_id is None: + raise ValueError( + "only a standalone stream can be closed; this handle is on an owned " + "stream, which ends when its workflow or activity does" + ) + self._streams._seal(self._stream_id) class MemoryStreams(ProviderPlugin): @@ -519,10 +652,14 @@ def __init__( """ self._poll = poll_interval self._topics: dict[str, _Topic] = {} + self._activity_topics: dict[tuple[str | None, str, str], _Topic] = {} + self._standalone: dict[str, _Standalone] = {} def reset(self) -> None: - """Drop every topic. For tests.""" + """Drop every topic and every standalone stream. For tests.""" self._topics.clear() + self._activity_topics.clear() + self._standalone.clear() def truncate(self, workflow_id: str, topic: str, *, keep: int) -> None: """Drop all but the newest ``keep`` records of a topic. For tests. @@ -548,6 +685,21 @@ def get_stream_handle( """ return MemoryStreamHandle(self, client, workflow_id, run_id) + def get_activity_stream_handle( + self, + client: Client | None, + activity_id: str, + *, + workflow_id: str | None = None, + run_id: str | None = None, + ) -> MemoryStreamHandle: + """A handle on the topics ``activity_id`` owns, apart from any workflow's. + + As on :meth:`get_stream_handle`, ``client`` may be ``None``, and then + a read waits until the caller closes it. + """ + return MemoryStreamHandle(self, client, workflow_id, run_id, activity_id) + async def create_standalone_stream( self, client: Client | None, @@ -557,26 +709,38 @@ async def create_standalone_stream( max_records: int | None = None, max_bytes: int | None = None, ) -> MemoryStreamHandle: - """Refuse: this provider keeps no stream without an owner. + """Create the standalone stream ``stream_id``, or find it with the same policy. + + The policy is applied on every append to any of the stream's topics. + ``client`` may be ``None``, as on the other handles. Raises: - StreamUnsupportedError: Always. + ValueError: ``stream_id`` is empty, a bound is not positive, or + the stream exists with a different policy. """ - raise StreamUnsupportedError( - "the memory provider does not host standalone streams" - ) + if not stream_id: + raise ValueError("stream_id must not be empty") + policy = _Policy(retention, max_records, max_bytes) + existing = self._standalone.get(stream_id) + if existing is None: + self._standalone[stream_id] = _Standalone(policy) + elif existing.policy != policy: + raise ValueError( + f"standalone stream {stream_id!r} exists with policy " + f"{existing.policy}, not {policy}" + ) + return MemoryStreamHandle(self, client, None, None, stream_id=stream_id) def get_standalone_stream_handle( self, client: Client | None, stream_id: str ) -> MemoryStreamHandle: - """Refuse: this provider keeps no stream without an owner. + """A handle on the standalone stream ``stream_id``. - Raises: - StreamUnsupportedError: Always. + Nothing is checked here: a ``read``, ``latest`` or ``producer`` on a + stream that was never created raises + :class:`temporalio.streams.StreamNotFoundError` at the call. """ - raise StreamUnsupportedError( - "the memory provider does not host standalone streams" - ) + return MemoryStreamHandle(self, client, None, None, stream_id=stream_id) async def close(self) -> None: """Nothing to release: the provider holds no connection.""" @@ -590,6 +754,43 @@ def _topic(self, workflow_id: str, topic: str) -> _Topic: found = self._topics[key] = _Topic() return found + def _activity_topic( + self, workflow_id: str | None, activity_id: str, topic: str + ) -> _Topic: + # Kept apart from the workflow topics, so no workflow id can name an + # activity's stream. + if not topic: + raise ValueError("topic must not be empty") + key = (workflow_id, activity_id, topic) + found = self._activity_topics.get(key) + if found is None: + found = self._activity_topics[key] = _Topic() + return found + + def _standalone_stream(self, stream_id: str) -> _Standalone: + stream = self._standalone.get(stream_id) + if stream is None: + raise StreamNotFoundError( + f"standalone stream {stream_id!r} does not exist; create it with " + "client.create_stream" + ) + return stream + + def _standalone_topic(self, stream_id: str, topic: str) -> _Topic: + stream = self._standalone_stream(stream_id) + if not topic: + raise ValueError("topic must not be empty") + found = stream.topics.get(topic) + if found is None: + found = stream.topics[topic] = _Topic(stream.policy, sealed=stream.sealed) + return found + + def _seal(self, stream_id: str) -> None: + stream = self._standalone_stream(stream_id) + stream.sealed = True + for store in stream.topics.values(): + store.seal() + def _start(self, store: _Topic, after: Cursor, last: int | None) -> int: """The offset a read starts at, resolved against what ``store`` holds now.""" if last is not None: diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index f72df9cf4..0f289931b 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -33,6 +33,7 @@ import temporalio.common import temporalio.converter import temporalio.exceptions +import temporalio.streams from temporalio.converter import ( StorageDriverActivityInfo, StorageDriverStoreContext, @@ -63,9 +64,11 @@ def __init__( metric_meter: temporalio.common.MetricMeter, client: temporalio.client.Client, encode_headers: bool, + stream_provider: temporalio.streams.StreamProvider | None = None, ) -> None: self._bridge_worker = bridge_worker self._task_queue = task_queue + self._stream_provider = stream_provider self._activity_executor = activity_executor self._shared_state_manager = shared_state_manager self._running_activities: dict[bytes, _RunningActivity] = {} @@ -666,6 +669,10 @@ async def _execute_activity( runtime_metric_meter=None if sync_non_threaded else self._metric_meter, client=self._client if not running_activity.sync else None, cancellation_details=running_activity.cancellation_details, + stream_provider=( + self._stream_provider if not running_activity.sync else None + ), + sync=running_activity.sync, ) ) temporalio.activity.logger.debug("Starting activity") @@ -942,6 +949,7 @@ def heartbeat_fn(*details: Any) -> None: runtime_metric_meter=runtime_metric_meter, client=None, cancellation_details=cancellation_details, + sync=True, ) ) if not cancel_thread_raiser: diff --git a/temporalio/worker/_worker.py b/temporalio/worker/_worker.py index f4b7eddad..fade69c4d 100644 --- a/temporalio/worker/_worker.py +++ b/temporalio/worker/_worker.py @@ -510,6 +510,7 @@ def _init_from_config(self, client: temporalio.client.Client, config: WorkerConf interceptors=interceptors, metric_meter=self._runtime.metric_meter, client=client, + stream_provider=stream_provider, encode_headers=( client_config["header_codec_behavior"] == HeaderCodecBehavior.CODEC ), From d917032c0b144f69b5236348710c20187cde4540 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 01:55:01 -0700 Subject: [PATCH 2/4] Covered the stream accessors. Activity and client handles on memory, and the conformance cases for standalone streams. --- tests/streams/conftest.py | 5 + tests/streams/test_activity_streams.py | 263 ++++++++++++++++++ tests/streams/test_stream_accessors.py | 324 ++++++++++++++++++++++ tests/streams/test_streams_conformance.py | 252 ++++++++++++++++- 4 files changed, 839 insertions(+), 5 deletions(-) create mode 100644 tests/streams/test_activity_streams.py create mode 100644 tests/streams/test_stream_accessors.py diff --git a/tests/streams/conftest.py b/tests/streams/conftest.py index dfc70b086..079cca975 100644 --- a/tests/streams/conftest.py +++ b/tests/streams/conftest.py @@ -17,6 +17,11 @@ def pytest_configure(config: pytest.Config) -> None: "markers", "truncates: the case needs a way to drop a topic's oldest records", ) + config.addinivalue_line( + "markers", + "hosts_standalone_streams: the case needs a stream with an id of its own and " + "no owner", + ) config.addinivalue_line( "markers", "needs_channel_server: the case needs a server that serves notification " diff --git a/tests/streams/test_activity_streams.py b/tests/streams/test_activity_streams.py new file mode 100644 index 000000000..bd0a675ac --- /dev/null +++ b/tests/streams/test_activity_streams.py @@ -0,0 +1,263 @@ +"""Streams owned by activities, and which stream an activity's accessor reaches. + +``activity.stream_handle()`` resolves by a static rule, never by what exists: +an activity a workflow scheduled reaches its workflow's stream, a standalone +activity reaches its own, and ``scope="activity"`` gives an activity a +workflow scheduled its own streams. An activity's own streams are one per +activity execution, so a retry writes to the same stream and a reader sees +the attempt change as ``SUPERSEDED``. + +The memory provider always runs. A storage provider adds itself to +``SETUPS`` behind its own ``STREAMS_LIVE`` gate: its setup receives the +environment's client and hands back the provider and a client with it +registered, which the workers and the reads in these cases share. +""" + +from __future__ import annotations + +import asyncio +import uuid +from collections.abc import AsyncIterator, Callable +from dataclasses import dataclass +from datetime import timedelta +from typing import Any + +import pytest + +from temporalio import activity, workflow +from temporalio.client import Client +from temporalio.common import RetryPolicy +from temporalio.streams import ( + BEGINNING, + RecordKind, + StreamProvider, + topic, +) +from temporalio.streams.providers.memory import MemoryStreams +from temporalio.testing import WorkflowEnvironment +from tests.helpers import new_worker + +TOKENS = topic("tokens", dict) + + +@dataclass +class ActivitySetup: + """A provider under test, registered on the client the cases use.""" + + name: str + provider: StreamProvider + client: Client + + +async def _memory_setup(client: Client) -> AsyncIterator[ActivitySetup]: + provider = MemoryStreams(poll_interval=timedelta(milliseconds=50)) + config = client.config() + config["plugins"] = [provider] + yield ActivitySetup("memory", provider, Client(**config)) + provider.reset() + + +SETUPS: dict[str, Callable[[Client], AsyncIterator[ActivitySetup]]] = { + "memory": _memory_setup +} + + +@pytest.fixture(params=sorted(SETUPS)) +async def setup( + request: pytest.FixtureRequest, client: Client, env: WorkflowEnvironment +) -> AsyncIterator[ActivitySetup]: + if env.supports_time_skipping: + pytest.skip("the time-skipping test server has no standalone activities") + async for found in SETUPS[request.param](client): + yield found + + +async def read_all(records: Any, timeout: float = 30.0) -> list: + """Read until the stream ends, which is when its owner does.""" + + async def _collect() -> list: + return [record async for record in records] + + return await asyncio.wait_for(_collect(), timeout) + + +def summary(records: list) -> list[tuple[Any, ...]]: + return [ + (r.kind, r.attempt, r.value["token"] if r.kind is RecordKind.DATA else None) + for r in records + ] + + +@activity.defn +async def write_by_default(label: str) -> str: + # No arguments: where this lands depends only on where the activity runs. + producer = activity.stream_handle().producer(topic=TOKENS) + await producer.append({"token": label}) + await producer.finish() + return activity.info().activity_id + + +@activity.defn +async def write_to_own_streams(label: str) -> str: + producer = activity.stream_handle(scope="activity").producer(topic=TOKENS) + await producer.append({"token": label}) + await producer.finish() + return activity.info().activity_id + + +@workflow.defn +class RunsOneActivity: + """Runs one activity by name and returns what it returned.""" + + @workflow.run + async def run(self, name: str, label: str) -> str: + return await workflow.execute_activity( + name, + label, + activity_id="streamer", + start_to_close_timeout=timedelta(seconds=30), + ) + + +async def test_workflow_activity_defaults_to_its_workflow(setup: ActivitySetup): + client = setup.client + workflow_id = f"streams-wfa-{uuid.uuid4().hex}" + async with new_worker( + client, RunsOneActivity, activities=[write_by_default] + ) as worker: + await client.execute_workflow( + RunsOneActivity.run, + args=["write_by_default", "to the workflow"], + id=workflow_id, + task_queue=worker.task_queue, + ) + records = await read_all( + client.get_stream_handle(workflow_id).read(topic=TOKENS) + ) + assert summary(records) == [ + (RecordKind.DATA, 1, "to the workflow"), + (RecordKind.FINISH, 1, None), + ] + assert all(r.producer_id == "streamer" for r in records) + own = client.get_stream_handle(workflow_id, activity_id="streamer") + assert await own.latest(topic=TOKENS) == BEGINNING + + +async def test_scope_activity_gives_a_workflow_activity_its_own_streams( + setup: ActivitySetup, +): + client = setup.client + workflow_id = f"streams-wfa-own-{uuid.uuid4().hex}" + async with new_worker( + client, RunsOneActivity, activities=[write_to_own_streams] + ) as worker: + await client.execute_workflow( + RunsOneActivity.run, + args=["write_to_own_streams", "to the activity"], + id=workflow_id, + task_queue=worker.task_queue, + ) + own = client.get_stream_handle(workflow_id, activity_id="streamer") + assert summary(await read_all(own.read(topic=TOKENS))) == [ + (RecordKind.DATA, 1, "to the activity"), + (RecordKind.FINISH, 1, None), + ] + # The workflow's topic of the same name is a different stream. + workflow_stream = client.get_stream_handle(workflow_id) + assert await workflow_stream.latest(topic=TOKENS) == BEGINNING + + +async def test_standalone_activity_defaults_to_its_own_stream(setup: ActivitySetup): + client = setup.client + activity_id = f"streams-saa-{uuid.uuid4().hex}" + async with new_worker(client, activities=[write_by_default]) as worker: + handle = await client.start_activity( + write_by_default, + "standalone", + id=activity_id, + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=30), + ) + assert await handle.result() == activity_id + stream = client.get_stream_handle(activity_id=activity_id) + # The read ends by itself: the activity reached a terminal status. + records = await read_all(stream.read(topic=TOKENS)) + assert summary(records) == [ + (RecordKind.DATA, 1, "standalone"), + (RecordKind.FINISH, 1, None), + ] + assert all(r.producer_id == activity_id for r in records) + + +@activity.defn +async def fail_once_after_writing() -> None: + attempt = activity.info().attempt + producer = activity.stream_handle().producer(topic=TOKENS) + await producer.append({"token": f"attempt {attempt}"}) + if attempt == 1: + raise RuntimeError("the first attempt fails after writing") + await producer.finish() + + +async def test_a_retry_inherits_the_stream_and_supersedes(setup: ActivitySetup): + client = setup.client + activity_id = f"streams-saa-retry-{uuid.uuid4().hex}" + async with new_worker(client, activities=[fail_once_after_writing]) as worker: + handle = await client.start_activity( + fail_once_after_writing, + id=activity_id, + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=30), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=200), maximum_attempts=2 + ), + ) + await handle.result() + records = await read_all( + client.get_stream_handle(activity_id=activity_id).read(topic=TOKENS) + ) + # One stream across both attempts, with the takeover reported between. + assert summary(records) == [ + (RecordKind.DATA, 1, "attempt 1"), + (RecordKind.SUPERSEDED, 2, None), + (RecordKind.DATA, 2, "attempt 2"), + (RecordKind.FINISH, 2, None), + ] + superseded = records[1].supersession + assert superseded is not None + assert (superseded.previous_attempt, superseded.attempt) == (1, 2) + + +@activity.defn +async def ask_for_misaddressed_handles() -> list[str]: + errors: list[str] = [] + try: + activity.stream_handle(scope="workflow") + except RuntimeError as error: + errors.append(f"RuntimeError: {error}") + try: + activity.stream_handle("some-workflow", scope="activity") + except ValueError as error: + errors.append(f"ValueError: {error}") + return errors + + +async def test_a_standalone_activity_has_no_workflow_to_address(setup: ActivitySetup): + client = setup.client + async with new_worker(client, activities=[ask_for_misaddressed_handles]) as worker: + errors = await client.execute_activity( + ask_for_misaddressed_handles, + id=f"streams-saa-errors-{uuid.uuid4().hex}", + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=30), + ) + assert len(errors) == 2 + assert errors[0].startswith("RuntimeError: this activity belongs to no workflow") + assert errors[1].startswith("ValueError: scope='activity'") + + +async def test_get_stream_handle_needs_an_owner(setup: ActivitySetup): + with pytest.raises( + ValueError, match="workflow_id, the activity_id or the stream_id" + ): + setup.client.get_stream_handle() diff --git a/tests/streams/test_stream_accessors.py b/tests/streams/test_stream_accessors.py new file mode 100644 index 000000000..4fd3f9b4e --- /dev/null +++ b/tests/streams/test_stream_accessors.py @@ -0,0 +1,324 @@ +"""The accessors each context asks for its stream, with one registration. + +The provider is registered on the client alone. A worker built from that +client inherits it, an activity on that worker reaches its own workflow's +stream through ``activity.stream_handle()``, and any code holding the client +reaches a stream through ``client.get_stream_handle()``. Without a +registration, both say so with the same error. +""" + +from __future__ import annotations + +import asyncio +import uuid +from concurrent.futures import ThreadPoolExecutor +from datetime import timedelta +from typing import Any + +import pytest + +from temporalio import activity, workflow +from temporalio.client import Client +from temporalio.streams import ( + BEGINNING, + DEFAULT_TOPIC, + END, + RecordKind, + StreamClosedError, + StreamRef, + StreamUnsupportedError, + topic, +) +from temporalio.streams.providers.memory import MemoryStreams +from temporalio.testing import ActivityEnvironment, WorkflowEnvironment +from tests.helpers import new_worker + +INPUTS = topic("inputs", dict) +DECISIONS = topic("decisions", dict) + + +@pytest.fixture +def provider(env: WorkflowEnvironment): # pyright: ignore[reportUnusedFunction] + if env.supports_time_skipping: + pytest.skip( + "the memory provider polls on a timer, which time skipping turns into a spin" + ) + streams = MemoryStreams() + yield streams + streams.reset() + + +def _with_provider(client: Client, provider: MemoryStreams) -> Client: + # The same connection, with the provider registered the way an + # application registers it: once, on the client. + config = client.config() + config["plugins"] = [provider] + return Client(**config) + + +@activity.defn +async def emit(count: int) -> str: + # No workflow id and no run id: the handle is this activity's own + # workflow, pinned to its run, and the producer's identity is the + # activity's. + model = activity.stream_handle().producer(topic=INPUTS) + for n in range(count): + await model.append({"n": n}) + await model.finish() + return activity.info().activity_id + + +@workflow.defn +class Echo: + """Runs the emitting activity and echoes what arrives on ``inputs``.""" + + @workflow.run + async def run(self, count: int) -> list[Any]: + inputs = workflow.stream_reader(INPUTS) + decisions = workflow.stream_writer(DECISIONS) + emitting = workflow.start_activity( + emit, count, start_to_close_timeout=timedelta(seconds=30) + ) + seen: list[Any] = [] + async for record in inputs: + if record.kind is RecordKind.FINISH: + seen.append(("finish", record.producer_id)) + break + assert record.value is not None + seen.append(record.value["n"]) + decisions.publish({"echo": record.value["n"]}) + decisions.finish() + producer_id = await emitting + return [*seen, ("activity", producer_id)] + + +async def test_one_registration_on_the_client_serves_every_context( + client: Client, provider: MemoryStreams +): + registered = _with_provider(client, provider) + workflow_id = f"streams-wf-{uuid.uuid4().hex}" + # No plugins on the worker: it inherits the client's provider. + async with new_worker(registered, Echo, activities=[emit]) as worker: + handle = await registered.start_workflow( + Echo.run, 2, id=workflow_id, task_queue=worker.task_queue + ) + result = await handle.result() + # The activity wrote as itself onto its own workflow's topic. + assert result[:2] == [0, 1] + assert result[2] == ["finish", result[3][1]] + + stream = registered.get_stream_handle(workflow_id) + + async def read_everything() -> list[Any]: + return [(r.kind, r.value) async for r in stream.read(topic=DECISIONS)] + + assert await asyncio.wait_for(read_everything(), 30) == [ + (RecordKind.DATA, {"echo": 0}), + (RecordKind.DATA, {"echo": 1}), + (RecordKind.FINISH, None), + ] + + +@activity.defn +async def emit_on_the_default_topic() -> None: + await activity.stream_handle().producer().append({"from": "activity"}) + + +@workflow.defn +class Answers: + """Waits for the activity's record on the default topic and answers there.""" + + @workflow.run + async def run(self) -> None: + reader = workflow.stream_reader(result_type=dict) + await workflow.execute_activity( + emit_on_the_default_topic, start_to_close_timeout=timedelta(seconds=30) + ) + async for value in reader.values(): + workflow.stream_writer().publish({"answer": value["from"]}) + reader.close() + + +async def test_every_accessor_defaults_to_the_same_topic( + client: Client, provider: MemoryStreams +): + registered = _with_provider(client, provider) + workflow_id = f"streams-wf-{uuid.uuid4().hex}" + async with new_worker( + registered, Answers, activities=[emit_on_the_default_topic] + ) as worker: + stream = registered.get_stream_handle(workflow_id) + assert await stream.latest() == BEGINNING + await registered.execute_workflow( + Answers.run, id=workflow_id, task_queue=worker.task_queue + ) + + async def read_everything() -> list[Any]: + return [(r.topic, r.value) async for r in stream.read()] + + assert await asyncio.wait_for(read_everything(), 30) == [ + (DEFAULT_TOPIC, {"from": "activity"}), + (DEFAULT_TOPIC, {"answer": "activity"}), + ] + + +async def test_get_stream_handle_needs_a_registered_provider(client: Client): + with pytest.raises(StreamUnsupportedError, match="plugins="): + client.get_stream_handle("wf") + + +async def test_stream_handle_needs_a_provider_on_the_worker(): + async def ask() -> None: + activity.stream_handle() + + with pytest.raises(StreamUnsupportedError, match="plugins="): + await ActivityEnvironment().run(ask) + + +@activity.defn +def ask_from_a_sync_activity() -> str: + try: + activity.stream_handle() + except RuntimeError as error: + return str(error) + return "opened" + + +@workflow.defn +class RunsASyncActivity: + """Runs the `def` activity that reaches for a handle.""" + + @workflow.run + async def run(self) -> str: + return await workflow.execute_activity( + ask_from_a_sync_activity, start_to_close_timeout=timedelta(seconds=30) + ) + + +async def test_a_sync_activity_is_told_it_cannot_have_a_handle( + client: Client, provider: MemoryStreams +): + # The worker has a provider. What a `def` activity is missing is the + # client, so that is what the error has to say, rather than sending the + # reader to register a provider that is already there. + registered = _with_provider(client, provider) + with ThreadPoolExecutor(max_workers=1) as executor: + async with new_worker( + registered, + RunsASyncActivity, + activities=[ask_from_a_sync_activity], + activity_executor=executor, + ) as worker: + result = await registered.execute_workflow( + RunsASyncActivity.run, + id=f"streams-wf-{uuid.uuid4().hex}", + task_queue=worker.task_queue, + ) + assert "only available in `async def` activities" in result + assert "plugins=" not in result + + +async def test_an_activity_owned_stream_takes_every_read_start( + client: Client, provider: MemoryStreams +): + registered = _with_provider(client, provider) + stream = registered.get_stream_handle(activity_id=f"act-{uuid.uuid4().hex}") + producer = stream.producer(topic=INPUTS, producer_id="tool", attempt=1) + await producer.append({"n": 1}, {"n": 2}, {"n": 3}) + + async def first(records: Any) -> Any: + async for record in records: + await records.aclose() + return record.value + return None + + assert await first(stream.read(topic=INPUTS, after=BEGINNING)) == {"n": 1} + assert await first(stream.read(topic=INPUTS, last=1)) == {"n": 3} + at_end = stream.read(topic=INPUTS, after=END) + await producer.append({"n": 4}) + assert await asyncio.wait_for(first(at_end), 10) == {"n": 4} + + +NOTES = topic("notes", dict) + + +@activity.defn +async def append_to_ref(ref: StreamRef) -> str: + # The ref arrived as an argument and names the stream in full, so it + # takes no scope; the handle it opens writes to the ref's topic. + try: + activity.stream_handle(ref, scope="activity") + except ValueError as error: + refused = str(error) + else: + refused = "accepted" + await activity.stream_handle(ref).producer().append({"via": "activity"}) + return refused + + +@workflow.defn +class PublishToRef: + """Hands a stream ref to an activity and returns it, both as plain data.""" + + @workflow.run + async def run(self, ref: StreamRef) -> tuple[StreamRef, str]: + refused = await workflow.execute_activity( + append_to_ref, ref, start_to_close_timeout=timedelta(seconds=30) + ) + return ref, refused + + +async def test_a_ref_travels_as_data_and_opens_from_every_context( + client: Client, provider: MemoryStreams +): + registered = _with_provider(client, provider) + shared = await registered.create_stream(f"shared-{uuid.uuid4().hex}") + ref = shared.ref(topic=NOTES) + async with new_worker( + registered, PublishToRef, activities=[append_to_ref] + ) as worker: + returned, refused = await registered.execute_workflow( + PublishToRef.run, + ref, + id=f"streams-wf-{uuid.uuid4().hex}", + task_queue=worker.task_queue, + ) + # The workflow argument and result went through the data converter. + assert returned == ref + assert "takes no run_id or scope" in refused + # The activity wrote to the referenced topic, and the client reads it + # from the ref without naming the topic. Sealing the stream is what ends + # a full read of it. + await shared.close() + records = [ + (r.topic, r.value) async for r in registered.get_stream_handle(ref).read(last=1) + ] + assert records == [("notes", {"via": "activity"})] + with pytest.raises(ValueError, match="takes no run_id"): + registered.get_stream_handle(ref, run_id="r") + + +async def test_the_client_creates_reaches_and_closes_a_standalone_stream( + client: Client, provider: MemoryStreams +): + registered = _with_provider(client, provider) + stream_id = f"shared-{uuid.uuid4().hex}" + created = await registered.create_stream(stream_id, max_records=3) + producer = created.producer(topic=NOTES, producer_id="writer", attempt=1) + await producer.append({"n": 1}) + + reached = registered.get_stream_handle(stream_id=stream_id) + assert await reached.latest(topic=NOTES) != BEGINNING + await reached.close() + with pytest.raises(StreamClosedError): + await producer.append({"n": 2}) + assert [r.value async for r in reached.read(topic=NOTES)] == [{"n": 1}] + + with pytest.raises(ValueError, match="no workflow_id"): + registered.get_stream_handle(stream_id=stream_id, workflow_id="wf") + with pytest.raises(ValueError, match="empty"): + await registered.create_stream("") + with pytest.raises(StreamUnsupportedError, match="plugins="): + await client.create_stream(stream_id) + with pytest.raises(StreamUnsupportedError, match="plugins="): + client.get_stream_handle(stream_id=stream_id) diff --git a/tests/streams/test_streams_conformance.py b/tests/streams/test_streams_conformance.py index fb819bcb4..f9c53bdcf 100644 --- a/tests/streams/test_streams_conformance.py +++ b/tests/streams/test_streams_conformance.py @@ -5,9 +5,11 @@ store. A storage provider adds itself to ``SETUPS``, behind its own ``STREAMS_LIVE`` gate when it needs a store the test environment does not start: its setup receives the environment's client and hands back a provider -instance and which capabilities it lacks, so the cases marked -``reports_positions`` are skipped with a reason on a provider whose -``append()`` learns positions at read time. +instance, the client the cases should use, a ``host`` that starts the +workflow owning a stream when the store lives inside a running workflow, and +which capabilities it lacks, so the cases marked ``reports_positions`` are +skipped with a reason on a provider whose ``append()`` learns positions at +read time. What this file pins down is what a provider owes: producer identity, retry deduplication, positions, supersession, topic addressing, cursor resumption, @@ -29,6 +31,7 @@ import uuid from collections.abc import AsyncIterator, Awaitable, Callable, Sequence from dataclasses import dataclass +from datetime import timedelta from typing import Any import pytest @@ -45,20 +48,26 @@ StorageDriverRetrieveContext, StorageDriverStoreContext, ) +from temporalio.service import RPCError from temporalio.streams import ( BEGINNING, DEFAULT_TOPIC, END, Cursor, RecordKind, + StreamClosedError, StreamCursorError, + StreamError, StreamHandle, + StreamNotFoundError, StreamProducerError, StreamProvider, StreamRef, + StreamUnsupportedError, Supersession, topic, ) +from temporalio.streams._ref import open_ref from temporalio.streams.providers.memory import MemoryStreams # Defined once and shared by every case, the way an application shares them @@ -76,13 +85,21 @@ class ProviderCase: name: str provider: StreamProvider + client: Client | None = None reports_positions: bool = True """``append()`` returns where the records landed.""" detects_divergent_retries: bool = True """``append()`` compares a repeat's content with what it already holds.""" + host: Callable[[str], Awaitable[None]] | None = None + """Starts the workflow that owns ``workflow_id``'s stream, when a store needs one.""" truncate: Callable[[str, str, int], Awaitable[None]] | None = None """Drops all but the newest records of a workflow's topic, standing in for retention, or ``None`` when the provider offers no way to.""" + hosts_standalone_streams: bool = True + """The store holds a stream with an id of its own and no owner.""" + waits_for_standalone_creation: bool = False + """A read on a standalone stream id that does not exist yet parks until + the first write instead of raising ``StreamNotFoundError``.""" bounds_standalone_bytes: bool = True """A standalone stream's policy can bound the bytes it keeps.""" refuses_appends_past_byte_cap: bool = False @@ -99,10 +116,16 @@ async def open( run_id: str | None = None, client: Client | None = None, ) -> StreamHandle: + if self.host is not None: + await self.host(workflow_id) if client is not None: # The explicit form, for a case that needs the handle to encode # bodies through this client's data converter. return self.provider.get_stream_handle(client, workflow_id, run_id=run_id) + if self.client is not None: + # A storage provider's setup registers the provider on the client, + # so the cases go through the accessor an application uses. + return self.client.get_stream_handle(workflow_id, run_id=run_id) # Only the memory provider gets here, and it takes no client. return self.provider.get_stream_handle( None, # type: ignore[arg-type] @@ -110,6 +133,42 @@ async def open( run_id=run_id, ) + async def open_ref(self, ref: StreamRef) -> StreamHandle: + if self.client is not None: + return self.client.get_stream_handle(ref) + return open_ref(self.provider, None, ref) # type: ignore[arg-type] + + async def create_stream( + self, + stream_id: str, + *, + retention: timedelta | None = None, + max_records: int | None = None, + max_bytes: int | None = None, + ) -> StreamHandle: + if self.client is not None: + return await self.client.create_stream( + stream_id, + retention=retention, + max_records=max_records, + max_bytes=max_bytes, + ) + return await self.provider.create_standalone_stream( + None, # type: ignore[arg-type] + stream_id, + retention=retention, + max_records=max_records, + max_bytes=max_bytes, + ) + + async def open_standalone(self, stream_id: str) -> StreamHandle: + if self.client is not None: + return self.client.get_stream_handle(stream_id=stream_id) + return self.provider.get_standalone_stream_handle( + None, # type: ignore[arg-type] + stream_id, + ) + class RecordingDriver(StorageDriver): """An in-memory external storage driver that counts what it was asked to hold.""" @@ -195,6 +254,7 @@ async def truncate(workflow_id: str, topic: str, keep: int) -> None: "reports_positions": lambda case: case.reports_positions, "detects_divergent_retries": lambda case: case.detects_divergent_retries, "truncates": lambda case: case.truncate is not None, + "hosts_standalone_streams": lambda case: case.hosts_standalone_streams, } @@ -228,6 +288,19 @@ async def _collect() -> None: return out +async def drain(records: Any, timeout: float = 5.0) -> list: + """Every record until the read ends on its own.""" + + async def _collect() -> list: + return [record async for record in records] + + return await asyncio.wait_for(_collect(), timeout) + + +def new_stream_id() -> str: + return f"stream-{uuid.uuid4().hex}" + + async def test_append_read_roundtrip(case: ProviderCase): workflow_id = new_workflow_id() stream = await case.open(workflow_id) @@ -630,7 +703,9 @@ async def test_a_body_above_the_threshold_is_offloaded_and_read_back( external_storage=ExternalStorage(drivers=[driver], payload_size_threshold=256), ) workflow_id = new_workflow_id() - stream = await case.open(workflow_id, client=_client_with(client, converter)) + stream = await case.open( + workflow_id, client=_client_with(case.client or client, converter) + ) producer = stream.producer(topic=OUT, producer_id="model", attempt=1) small = {"n": 1} large = {"blob": "x" * 1024} @@ -652,7 +727,9 @@ async def test_a_retry_through_a_nondeterministic_codec_still_deduplicates( codec = NonceCodec() converter = dataclasses.replace(DataConverter.default, payload_codec=codec) workflow_id = new_workflow_id() - stream = await case.open(workflow_id, client=_client_with(client, converter)) + stream = await case.open( + workflow_id, client=_client_with(case.client or client, converter) + ) first = stream.producer(topic=OUT, producer_id="model", attempt=1) landed = await first.append({"id": "r1"}) assert codec.encoded == 1 @@ -672,3 +749,168 @@ async def test_a_retry_through_a_nondeterministic_codec_still_deduplicates( await first.append({"id": "r2"}) records = await take(stream.read(topic=OUT), 2) assert [r.value for r in records] == [{"id": "r1"}, {"id": "r2"}] + + +async def test_a_ref_opens_the_stream_it_names(case: ProviderCase): + workflow_id = new_workflow_id() + stream = await case.open(workflow_id) + await stream.producer(topic=A, producer_id="model", attempt=1).append({"n": 1}) + ref = stream.ref(topic=A) + + # The receiver names no topic: the ref carried it, so every call on the + # handle it opened addresses topic ``a`` of that workflow. + opened = await case.open_ref(ref) + records = await take(opened.read(result_type=dict), 1) + assert [(r.topic, r.value) for r in records] == [("a", {"n": 1})] + assert await opened.latest() == records[0].cursor + assert opened.ref() == ref + await opened.producer(producer_id="tool", attempt=1).append({"n": 2}) + assert [r.value for r in await take(stream.read(topic=A), 2)] == [ + {"n": 1}, + {"n": 2}, + ] + # Naming a topic on the opened handle addresses that topic instead. + assert await opened.latest(topic=B) == BEGINNING + assert opened.ref(topic=B) == stream.ref(topic=B) + + +@pytest.mark.hosts_standalone_streams +async def test_a_standalone_stream_is_read_from_another_handle(case: ProviderCase): + stream_id = new_stream_id() + created = await case.create_stream(stream_id) + producer = created.producer(topic=OUT, producer_id="writer", attempt=1) + await producer.append({"n": 1}, {"n": 2}) + + # Any process reaches the stream by its id, or by a ref the creator + # handed out; nothing about the stream depends on who created it. + other = await case.open_standalone(stream_id) + records = await take(other.read(topic=OUT), 2) + assert [r.value for r in records] == [{"n": 1}, {"n": 2}] + assert await other.latest(topic=OUT) == records[1].cursor + assert other.ref(topic=OUT) == StreamRef.for_standalone(stream_id, topic="out") + via_ref = await case.open_ref(created.ref(topic=OUT)) + assert [r.value for r in await take(via_ref.read(), 2)] == [{"n": 1}, {"n": 2}] + + +@pytest.mark.hosts_standalone_streams +async def test_a_missing_standalone_stream_is_not_found(case: ProviderCase): + if case.waits_for_standalone_creation: + pytest.skip( + f"the {case.name} provider waits for a standalone stream to be created" + ) + # get_stream_handle(stream_id=) creates nothing: the stream has to have + # been created on purpose, and a use before that says so. + stream = await case.open_standalone(new_stream_id()) + with pytest.raises(StreamNotFoundError): + await stream.latest(topic=OUT) + with pytest.raises(StreamNotFoundError): + await take(stream.read(topic=OUT), 1) + with pytest.raises(StreamNotFoundError): + await stream.producer(topic=OUT, producer_id="writer", attempt=1).append( + {"n": 1} + ) + + +@pytest.mark.hosts_standalone_streams +async def test_creating_a_standalone_stream_is_idempotent_for_one_policy( + case: ProviderCase, +): + stream_id = new_stream_id() + first = await case.create_stream(stream_id, max_records=10) + # The same id and policy again is the same stream, not an error, so a + # retried create is harmless. + again = await case.create_stream(stream_id, max_records=10) + await first.producer(topic=OUT, producer_id="writer", attempt=1).append({"n": 1}) + assert [r.value for r in await take(again.read(topic=OUT), 1)] == [{"n": 1}] + # A different policy on an existing id is a mistake, not a change. + with pytest.raises(ValueError): + await case.create_stream(stream_id, max_records=5) + for bad in (dict(max_records=0), dict(max_bytes=-1), dict(retention=timedelta(0))): + with pytest.raises(ValueError): + await case.create_stream(new_stream_id(), **bad) # type: ignore[arg-type] + + +@pytest.mark.hosts_standalone_streams +async def test_closing_a_standalone_stream_ends_reads_and_refuses_appends( + case: ProviderCase, +): + stream_id = new_stream_id() + stream = await case.create_stream(stream_id) + producer = stream.producer(topic=OUT, producer_id="writer", attempt=1) + await producer.append({"n": 1}) + # A reader parked on the tail before the close has to learn of it. + other = await case.open_standalone(stream_id) + parked = asyncio.ensure_future(drain(other.read(topic=OUT), timeout=10)) + await asyncio.sleep(0.2) + await producer.append({"n": 2}) + + await stream.close() + assert [r.value for r in await parked] == [{"n": 1}, {"n": 2}] + # Sealed: the tail stays readable, and a read opened now ends by itself. + assert [r.value for r in await drain(stream.read(topic=OUT))] == [ + {"n": 1}, + {"n": 2}, + ] + with pytest.raises(StreamClosedError): + await producer.append({"n": 3}) + with pytest.raises(StreamClosedError): + await other.producer(topic=A, producer_id="late", attempt=1).append({"n": 3}) + # Closing again is not an error. + await stream.close() + + +@pytest.mark.hosts_standalone_streams +async def test_a_standalone_stream_honors_its_retention_policy(case: ProviderCase): + by_count = await case.create_stream(new_stream_id(), max_records=2) + producer = by_count.producer(topic=OUT, producer_id="writer", attempt=1) + await producer.append({"n": 1}, {"n": 2}, {"n": 3}, {"n": 4}) + # BEGINNING is the oldest record still held, which the policy decided. + kept = await take(by_count.read(topic=OUT), 2) + assert [r.value for r in kept] == [{"n": 3}, {"n": 4}] + + if case.bounds_standalone_bytes and case.refuses_appends_past_byte_cap: + # The cap counts the stored record, body and metadata included, so it + # holds one of these and not two. + by_bytes = await case.create_stream(new_stream_id(), max_bytes=1000) + producer = by_bytes.producer(topic=OUT, producer_id="writer", attempt=1) + await producer.append({"n": 0, "blob": "x" * 500}) + # The second record would take the stream past its cap, so the store + # refuses it and keeps what it holds. + with pytest.raises((RPCError, StreamError)): + await producer.append({"n": 1, "blob": "x" * 500}) + kept = await take(by_bytes.read(topic=OUT), 1) + assert kept[0].value is not None and kept[0].value["n"] == 0 + elif case.bounds_standalone_bytes: + by_bytes = await case.create_stream(new_stream_id(), max_bytes=700) + producer = by_bytes.producer(topic=OUT, producer_id="writer", attempt=1) + for n in range(3): + await producer.append({"n": n, "blob": "x" * 500}) + kept = await take(by_bytes.read(topic=OUT), 1) + assert kept[0].value is not None and kept[0].value["n"] == 2 + else: + with pytest.raises(StreamUnsupportedError): + await case.create_stream(new_stream_id(), max_bytes=700) + + if not case.trims_open_stream_by_age: + return + by_age = await case.create_stream( + new_stream_id(), retention=timedelta(milliseconds=200) + ) + producer = by_age.producer(topic=OUT, producer_id="writer", attempt=1) + await producer.append({"n": "old"}) + await asyncio.sleep(0.3) + await producer.append({"n": "new"}) + # A store reclaims by age on its own schedule, and a coarse one may have + # aged the newer record out as well by the time it is looked at. What the + # policy decides is that the older record is no longer where BEGINNING + # starts. + kept = [] + for _ in range(50): + try: + kept = await take(by_age.read(topic=OUT), 1, timeout=1) + except asyncio.TimeoutError: + kept = [] + if not kept or kept[0].value == {"n": "new"}: + break + await asyncio.sleep(0.2) + assert [r.value for r in kept] in ([{"n": "new"}], []) From 67efbda8c1430313a38eef4180ae2ba9563d3f4b Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 01:55:01 -0700 Subject: [PATCH 3/4] Added stream_channel to name the channel a stream notifies. The server derives the channel from the stream's identity, and the client derives the same one, so a listener needs no lookup. --- temporalio/client/__init__.py | 2 ++ temporalio/client/_channel.py | 44 ++++++++++++++++++++++++++++++++++ tests/streams/test_channels.py | 32 +++++++++++++++++++++++++ 3 files changed, 78 insertions(+) diff --git a/temporalio/client/__init__.py b/temporalio/client/__init__.py index d77cf9693..7ed7187be 100644 --- a/temporalio/client/__init__.py +++ b/temporalio/client/__init__.py @@ -70,6 +70,7 @@ ChannelKind, ChannelListener, ChannelSubscriptionInfo, + stream_channel, ) from ._client import ( Client, @@ -373,6 +374,7 @@ "ChannelListener", "ChannelAddress", "ChannelSubscriptionInfo", + "stream_channel", "_ClientImpl", "_apply_headers", "_decode_user_metadata", diff --git a/temporalio/client/_channel.py b/temporalio/client/_channel.py index 0decc2265..3bdfa3ef2 100644 --- a/temporalio/client/_channel.py +++ b/temporalio/client/_channel.py @@ -10,6 +10,7 @@ import temporalio.api.notification.v1 import temporalio.api.workflow.v1 import temporalio.common +from temporalio.streams._ref import StreamRef from temporalio.workflow import Notification from ._callback import Callback @@ -20,8 +21,12 @@ "ChannelKind", "ChannelListener", "ChannelSubscriptionInfo", + "stream_channel", ] +STREAM_CHANNEL_PREFIX = "stream/" +"""The first segment of the channel a native stream notifies.""" + class ChannelKind(IntEnum): """Where a channel lives, which decides how a call addresses it. @@ -226,3 +231,42 @@ def workflow_id(self) -> str | None: ): return self.execution.business_id return None + + +def stream_channel(ref: StreamRef) -> ChannelAddress: + """The channel a native stream notifies on every append and on its close. + + The server derives the name from the stream's identity, and this helper + derives the same one, so a client polls or registers a callback without + asking. A stream a workflow owns notifies ``stream/`` linked to the + owning workflow. A stream an activity owns notifies + ``stream//`` linked to the workflow that scheduled + the activity, or ``stream/`` linked to the activity execution + itself when the activity is a standalone one. A standalone stream notifies + the independent channel ``stream/``, whatever the topic, since + its topics share one stream on the server. The address names the owner + without a run, so it reaches the owner's current run. + + Each change arrives as one notification: the stream's change sequence as + the counter, the head after the change as the position, and ``closed`` + set in the metadata on the close. + """ + if ref.kind == "workflow": + assert ref.workflow_id is not None + return ChannelAddress( + STREAM_CHANNEL_PREFIX + ref.topic, + temporalio.common.Execution.workflow(ref.workflow_id), + ) + if ref.kind == "activity": + assert ref.activity_id is not None + if not ref.workflow_id: + return ChannelAddress( + STREAM_CHANNEL_PREFIX + ref.topic, + temporalio.common.Execution.activity(ref.activity_id), + ) + return ChannelAddress( + f"{STREAM_CHANNEL_PREFIX}{ref.activity_id}/{ref.topic}", + temporalio.common.Execution.workflow(ref.workflow_id), + ) + assert ref.stream_id is not None + return ChannelAddress(STREAM_CHANNEL_PREFIX + ref.stream_id, None) diff --git a/tests/streams/test_channels.py b/tests/streams/test_channels.py index ab290734e..ce7179278 100644 --- a/tests/streams/test_channels.py +++ b/tests/streams/test_channels.py @@ -32,12 +32,15 @@ from temporalio.api.enums.v1 import EventType from temporalio.client import ( Callback, + ChannelAddress, ChannelKind, ChannelSubscriptionInfo, Client, WorkflowExecutionDescription, + stream_channel, ) from temporalio.service import RPCError, RPCStatusCode +from temporalio.streams import DEFAULT_TOPIC, StreamRef from temporalio.worker._workflow_instance import ( UnsandboxedWorkflowRunner, WorkflowInstance, @@ -551,6 +554,35 @@ async def test_a_subscription_after_an_unsubscribe_is_a_new_one(): } +async def test_a_stream_names_the_channel_it_notifies(): + owner = temporalio.common.Execution.workflow("wf") + assert stream_channel(StreamRef.for_workflow("wf", topic="out")) == ChannelAddress( + "stream/out", owner + ) + # The address names the owner without a run, so it follows the chain. + assert stream_channel(StreamRef.for_workflow("wf", run_id="run")) == ChannelAddress( + "stream/" + DEFAULT_TOPIC, owner + ) + assert stream_channel( + StreamRef.for_activity("act", workflow_id="wf", topic="out") + ) == ChannelAddress("stream/act/out", owner) + # A standalone activity is an execution of its own, so its stream's + # channel is linked to it under the topic's name alone. + assert stream_channel(StreamRef.for_activity("act", topic="out")) == ChannelAddress( + "stream/out", temporalio.common.Execution.activity("act") + ) + # A standalone stream's topics share one stream on the server, so the + # topic is not part of the name. + assert stream_channel( + StreamRef.for_standalone("sid", topic="out") + ) == ChannelAddress("stream/sid", None) + # The workflow id stays readable for a caller that addresses by it, and + # only names a workflow. + assert stream_channel(StreamRef.for_workflow("wf")).workflow_id == "wf" + assert stream_channel(StreamRef.for_activity("act")).workflow_id is None + assert stream_channel(StreamRef.for_standalone("sid")).workflow_id is None + + async def test_the_description_maps_every_channel_subscription_field(): [topic] = temporalio.converter.PayloadConverter.default.to_payloads(["inputs"]) pending = Notification(channel="orders", position=b"4-0", counter=4) From 5460ca211707a73eaf439bfcb954ebcf04222bca Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 01:55:01 -0700 Subject: [PATCH 4/4] Added a changelog entry for the stream interface. --- CHANGELOG.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d589a3fb..5a234f1b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,6 +46,26 @@ to include examples, links to docs, or any other relevant information. `workflow.linked_channel(name)` and ends a subscription with `unsubscribe()`. Notifications arrive with the workflow's tasks. `WorkflowExecutionDescription.channel_subscriptions` lists the channels a run listens on. +- **Experimental**: `temporalio.streams` defines one stream interface a workflow + can read, decide on, and write. A provider is registered once as a plugin, + `Client.connect(plugins=[provider])`, and workers built from that client + inherit it; each context then asks for its stream the same way: + `workflow.stream_reader()` and `workflow.stream_writer()` in workflow code, + `activity.stream_handle()` in an activity, and `client.get_stream_handle()` + anywhere a client is held. A topic is a typed definition, + `streams.topic("inputs", Token)`, shared by workflow, activity and client + code; a plain string names a topic decided at runtime, and a call that names + no topic addresses the default topic, `streams.DEFAULT_TOPIC` (`"output"`, + the server's default stream name). The record on the wire + is `temporal.api.stream.v1.StreamRecord` on every provider. A stream is + handed to another process as a `streams.StreamRef`, plain data naming the + owner and the topic, which `client.get_stream_handle(ref)` and + `activity.stream_handle(ref)` open; `client.create_stream(stream_id, ...)` + creates a standalone stream with a retention policy, and its handle's + `close()` seals it. A provider runs record bodies through the client's data + converter, so a payload codec and external storage apply to them. + `temporalio.streams.providers.memory.MemoryStreams` is the in-memory + reference provider the conformance tests run against. ### Changed