From 002f31386e541c4958fc7fbbaaea4e5b3400ee30 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 15:00:25 -0700 Subject: [PATCH 1/7] Added the Activity and client stream handles. An Activity writes to its Workflow's stream as itself, so readers see its retry, and a lower attempt after a higher one is reported as stale. --- temporalio/contrib/streams/__init__.py | 14 ++ temporalio/contrib/streams/_handles.py | 169 ++++++++++++++ temporalio/contrib/streams/_plugin.py | 66 +++++- temporalio/contrib/streams/_policy.py | 20 +- temporalio/contrib/streams/_wire.py | 2 +- tests/contrib/streams/test_outside_handles.py | 206 ++++++++++++++++++ tests/contrib/streams/test_records_wire.py | 41 ++++ 7 files changed, 511 insertions(+), 7 deletions(-) create mode 100644 temporalio/contrib/streams/_handles.py create mode 100644 tests/contrib/streams/test_outside_handles.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 838b3a4ea9..3368546c97 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -35,6 +35,12 @@ context then reaches a stream the same way: - Workflow code publishes to its own stream with :func:`workflow_writer`. +- An Activity reaches the stream of the Workflow that scheduled it with + :func:`activity_handle`, and writes as itself: its Activity id and its + Temporal attempt. +- Any process holding a client reaches a Workflow's stream with + :func:`get_stream_handle`, and writes with a producer id and an attempt of + its own. The record on the wire is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.contrib.streams.proto.v1`, with the user's value in ``body`` @@ -59,6 +65,11 @@ StreamProducerError, StreamUnsupportedError, ) +from temporalio.contrib.streams._handles import ( + ActivityStreamHandle, + activity_handle, + get_stream_handle, +) from temporalio.contrib.streams._plugin import StreamProviderPlugin from temporalio.contrib.streams._provider import ( StreamHandle, @@ -81,6 +92,7 @@ from temporalio.contrib.streams._workflow import WorkflowStreamWriter, workflow_writer __all__ = [ + "ActivityStreamHandle", "BEGINNING", "CONTENT_HASH_KEY", "Cursor", @@ -105,10 +117,12 @@ "StreamUnsupportedError", "Supersession", "WorkflowStreamWriter", + "activity_handle", "content_fingerprint", "content_hash", "decode_body", "encode_body", + "get_stream_handle", "resolve_topic", "topic", "workflow_writer", diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py new file mode 100644 index 0000000000..21d13b9ea6 --- /dev/null +++ b/temporalio/contrib/streams/_handles.py @@ -0,0 +1,169 @@ +"""Reaching a stream from an Activity or from client code. + +An Activity reaches the stream of the Workflow that scheduled it, and writes +as itself: its Activity id is the producer id and its Temporal attempt is +the producer attempt, so a retry is reported to readers as ``SUPERSEDED``. +Any process holding a client reaches a Workflow's stream by Workflow id and +writes with a producer id and attempt of its own. +""" + +from __future__ import annotations + +from collections.abc import AsyncGenerator +from typing import Any, TypeVar, overload + +import temporalio.activity +from temporalio.client import Client +from temporalio.contrib.streams._cursor import BEGINNING +from temporalio.contrib.streams._plugin import ( + provider_for_activity, + provider_for_client, +) +from temporalio.contrib.streams._provider import StreamHandle, StreamProducer +from temporalio.contrib.streams._record import Cursor, StreamRecord +from temporalio.contrib.streams._ref import StreamRef +from temporalio.contrib.streams._topic import StreamTopic + +__all__ = ["ActivityStreamHandle", "activity_handle", "get_stream_handle"] + +T = TypeVar("T") + + +class ActivityStreamHandle: + """The scheduling Workflow's stream, as the running Activity sees it. + + Reads and positions as :class:`temporalio.contrib.streams.StreamHandle` + does. :meth:`producer` writes as this Activity, which is the only + identity that lets readers tell this Activity's retry from a new + producer. + """ + + def __init__(self, inner: StreamHandle, producer_id: str, attempt: int) -> None: + """Prefer :func:`temporalio.contrib.streams.activity_handle`.""" + self._inner = inner + self._producer_id = producer_id + self._attempt = attempt + + @property + def ref(self) -> StreamRef: + """The stream this handle is on, pinned to the scheduling run.""" + return self._inner.ref + + @overload + def read( + self, *, topic: StreamTopic[T], after: Cursor = ... + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: type[T], + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: ... + + def read( + self, + *, + topic: str | StreamTopic[Any] | None = None, + after: Cursor = BEGINNING, + result_type: type | None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: + """See :meth:`temporalio.contrib.streams.StreamHandle.read`.""" + inner: Any = self._inner + return inner.read(topic=topic, after=after, result_type=result_type) + + async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: + """See :meth:`temporalio.contrib.streams.StreamHandle.latest`.""" + return await self._inner.latest(topic=topic) + + @overload + def producer(self, *, topic: StreamTopic[T]) -> StreamProducer[T]: ... + + @overload + def producer(self, *, topic: str | None = None) -> StreamProducer[Any]: ... + + def producer( + self, *, topic: str | StreamTopic[Any] | None = None + ) -> StreamProducer[Any]: + """A producer on ``topic`` that writes as this Activity attempt. + + The producer id is the Activity id and the attempt is the Activity's + Temporal attempt, so the first record of a retry makes readers see + ``SUPERSEDED``. A retry starts a new sequence, and the store + deduplicates a repeated append within one attempt. + """ + inner: Any = self._inner + return inner.producer( + topic=topic, producer_id=self._producer_id, attempt=self._attempt + ) + + +def activity_handle() -> ActivityStreamHandle: + """The stream of the Workflow that scheduled the running Activity. + + The handle is pinned to the scheduling run, so a read on it ends when + that run closes. To write to another Workflow's stream, use + :func:`get_stream_handle` with ``temporalio.activity.client()`` and a + producer id of your own. + + Raises: + ValueError: The Activity was not scheduled by a Workflow, or no stream + provider is registered on the Worker or its client. + RuntimeError: Not called from inside an Activity. + """ + info = temporalio.activity.info() + if info.workflow_id is None: + raise ValueError( + "this Activity was not scheduled by a Workflow, so it has no Workflow " + "stream" + ) + provider = provider_for_activity() + ref = StreamRef.for_workflow(info.workflow_id, run_id=info.workflow_run_id) + inner = provider.get_stream_handle(temporalio.activity.client(), ref) + return ActivityStreamHandle(inner, info.activity_id, info.attempt) + + +def get_stream_handle( + client: Client, + workflow_id: str | StreamRef, + *, + run_id: str | None = None, + topic: str | StreamTopic[Any] | None = None, +) -> StreamHandle: + """A handle on a Workflow's stream, through the client's provider. + + Without ``run_id`` the handle follows the Workflow's run chain, so a + read continues across Continue-as-New and ends when the chain closes. + With one it is pinned to that run. ``topic`` is the topic a call that + names none addresses. + + Args: + client: A client that carries a stream provider plugin. + workflow_id: The Workflow that owns the stream, or a + :class:`temporalio.contrib.streams.StreamRef` that names it. + run_id: Pin the handle to one run. + topic: The handle's default topic. Without one it is + :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + + Raises: + ValueError: ``run_id`` or ``topic`` was given with a ref, or no + stream provider is registered on ``client``. + """ + if isinstance(workflow_id, StreamRef): + if run_id is not None or topic is not None: + raise ValueError("a StreamRef carries its own run_id and topic") + ref = workflow_id + else: + ref = StreamRef.for_workflow(workflow_id, run_id=run_id, topic=topic) + return provider_for_client(client).get_stream_handle(client, ref) diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py index 7e5d6e7ddf..716ee1d3d3 100644 --- a/temporalio/contrib/streams/_plugin.py +++ b/temporalio/contrib/streams/_plugin.py @@ -6,17 +6,20 @@ ``Replayer(plugins=[provider])`` register it on a Worker alone. On a Worker the plugin adds one interceptor and nothing else. The -interceptor makes the provider reachable from Workflow code through the -Worker's extern functions, which is how a passthrough object crosses into -the sandbox. No Worker internals change. +interceptor makes the provider reachable from an Activity through a context +variable, and from Workflow code through the Worker's extern functions, +which is how a passthrough object crosses into the sandbox. No Worker +internals change. """ from __future__ import annotations +import contextvars from abc import ABC, abstractmethod from collections.abc import Sequence from typing import TYPE_CHECKING, Any, Protocol +import temporalio.activity import temporalio.worker import temporalio.workflow from temporalio.contrib.streams._errors import StreamUnsupportedError @@ -33,6 +36,10 @@ _WORKFLOW_EXTERN = "__temporal_contrib_streams_provider" +_activity_provider: contextvars.ContextVar[StreamProviderPlugin | None] = ( + contextvars.ContextVar("__temporal_contrib_streams_provider", default=None) +) + class _WorkflowOutput(Protocol): """Where a Workflow's own publish goes, on the Workflow thread. @@ -137,11 +144,16 @@ def _two_providers(held: StreamProviderPlugin, new: StreamProviderPlugin) -> str class _StreamsInterceptor(temporalio.worker.Interceptor): - """Makes the provider reachable from Workflow code.""" + """Makes the provider reachable from Activities and Workflow code.""" def __init__(self, provider: StreamProviderPlugin) -> None: self.provider = provider + def intercept_activity( + self, next: temporalio.worker.ActivityInboundInterceptor + ) -> temporalio.worker.ActivityInboundInterceptor: + return _ActivityInbound(next, self.provider) + def workflow_interceptor_class( self, input: temporalio.worker.WorkflowInterceptorClassInput ) -> type[temporalio.worker.WorkflowInboundInterceptor] | None: @@ -150,6 +162,25 @@ def workflow_interceptor_class( return None +class _ActivityInbound(temporalio.worker.ActivityInboundInterceptor): + def __init__( + self, + next: temporalio.worker.ActivityInboundInterceptor, + provider: StreamProviderPlugin, + ) -> None: + super().__init__(next) + self._provider = provider + + async def execute_activity( + self, input: temporalio.worker.ExecuteActivityInput + ) -> Any: + token = _activity_provider.set(self._provider) + try: + return await self.next.execute_activity(input) + finally: + _activity_provider.reset(token) + + def _no_provider(where: str) -> ValueError: return ValueError( f"no stream provider is registered on {where}; pass one as a plugin, for " @@ -157,6 +188,33 @@ def _no_provider(where: str) -> ValueError: ) +def provider_for_client(client: Client) -> StreamProviderPlugin: + """The provider registered on ``client``. + + Raises: + ValueError: No provider is registered on the client. + """ + for plugin in client.config()["plugins"]: + if isinstance(plugin, StreamProviderPlugin): + return plugin + raise _no_provider("this client") + + +def provider_for_activity() -> StreamProviderPlugin: + """The provider registered on the Worker running the current Activity. + + Raises: + ValueError: No provider is registered on the Worker or its client. + """ + provider = _activity_provider.get() + if provider is not None: + return provider + try: + return provider_for_client(temporalio.activity.client()) + except (RuntimeError, ValueError): + raise _no_provider("this Activity's Worker") from None + + def provider_for_workflow() -> StreamProviderPlugin: """The provider registered on the Worker running the current Workflow. diff --git a/temporalio/contrib/streams/_policy.py b/temporalio/contrib/streams/_policy.py index 0fcf4db5da..2d8f18380c 100644 --- a/temporalio/contrib/streams/_policy.py +++ b/temporalio/contrib/streams/_policy.py @@ -11,6 +11,7 @@ from __future__ import annotations +from collections.abc import Callable from typing import Any from temporalio.contrib.streams._record import ( @@ -26,9 +27,10 @@ class AttemptTracker: """Watches producer attempts on one read.""" - def __init__(self) -> None: - """Start with no producer seen.""" + def __init__(self, warn: Callable[[str], None] | None = None) -> None: + """Start with no producer seen, reporting anything odd through ``warn``.""" self._attempts: dict[str, int] = {} + self._warn = warn def note( self, producer_id: str, attempt: int, *, topic: str, previous: Cursor @@ -42,10 +44,24 @@ def note( A producer that declares no attempt supersedes nothing, because there is no generation to compare. + + An attempt that goes backwards supersedes nothing either, and is + reported through ``warn`` rather than passed off as ordinary data. + A lower attempt after a higher one means an older attempt was still + writing after a newer one started, such as an Activity attempt that + timed out but kept running. A consumer that reads it as the current + answer would show a stale one. """ if not producer_id or attempt <= 0: return None seen = self._attempts.get(producer_id, 0) + if attempt < seen and self._warn is not None: + self._warn( + f"stream record on {topic!r} after {previous} is from attempt " + f"{attempt} of producer {producer_id!r}, behind attempt {seen}, " + "which this reader already delivered: an older attempt wrote " + "after a newer one started" + ) if attempt <= seen: return None self._attempts[producer_id] = attempt diff --git a/temporalio/contrib/streams/_wire.py b/temporalio/contrib/streams/_wire.py index 6741e4943d..39f3f9b0bc 100644 --- a/temporalio/contrib/streams/_wire.py +++ b/temporalio/contrib/streams/_wire.py @@ -109,7 +109,7 @@ def __init__( self._result_type = result_type self._previous = after self._warn = warn - self._attempts = AttemptTracker() + self._attempts = AttemptTracker(warn) def decode(self, cursor: Cursor, wire: WireRecord) -> list[StreamRecord[Any]]: """The records to yield for one stored record, in order.""" diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py new file mode 100644 index 0000000000..2d2f0ee115 --- /dev/null +++ b/tests/contrib/streams/test_outside_handles.py @@ -0,0 +1,206 @@ +"""Reaching a Workflow's stream from an Activity or a client. + +An Activity writes to the stream of the Workflow that scheduled it as +itself, so its retry is reported to readers as ``SUPERSEDED``. A client +reaches a Workflow's stream by Workflow id through its provider. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +import uuid +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.contrib.streams import ( + RecordKind, + StreamRef, + Supersession, + activity_handle, + get_stream_handle, + topic, +) +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.exceptions import ApplicationError +from temporalio.testing import ActivityEnvironment +from tests.contrib.streams.test_workflow_writer import EVENTS, Publisher +from tests.helpers import new_worker + +TOKENS = topic("tokens", str) + + +def client_with(client: Client, provider: MemoryStreams | None) -> Client: + config = client.config() + config["plugins"] = [provider] if provider is not None else [] + return Client(**config) + + +def new_workflow_id() -> str: + return f"streams-publish-{uuid.uuid4().hex}" + + +async def read_all(records: Any, timeout: float = 10.0) -> list: + """Read until the stream ends, which is when its owner closes.""" + out = [] + + async def pull() -> None: + async for record in records: + out.append(record) + + try: + await asyncio.wait_for(pull(), timeout) + finally: + await records.aclose() + return out + + +@activity.defn +async def stream_tokens(tokens: list[str]) -> bool: + stream = activity_handle() + producer = stream.producer(topic=TOKENS) + attempt = activity.info().attempt + # The first attempt writes part of its answer and fails; the retry + # writes the whole answer, which readers must see as superseding it. + await producer.append(*tokens[:attempt]) + if attempt == 1: + raise ApplicationError("retry me") + await producer.append(*tokens[attempt:]) + await producer.finish() + return stream.ref.run_id == activity.info().workflow_run_id + + +@workflow.defn +class RunsStreamingActivity: + @workflow.run + async def run(self, tokens: list[str]) -> bool: + return await workflow.execute_activity( + stream_tokens, + tokens, + activity_id="tokens", + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=50), maximum_attempts=2 + ), + ) + + +async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): + provider = MemoryStreams() + streams_client = client_with(client, provider) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, RunsStreamingActivity, activities=[stream_tokens] + ) as worker: + pinned = await streams_client.execute_workflow( + RunsStreamingActivity.run, + ["a", "b", "c"], + id=workflow_id, + task_queue=worker.task_queue, + ) + assert pinned + records = await read_all( + get_stream_handle(streams_client, workflow_id).read(topic=TOKENS) + ) + assert [r.kind for r in records] == [ + RecordKind.DATA, + RecordKind.SUPERSEDED, + RecordKind.DATA, + RecordKind.DATA, + RecordKind.DATA, + RecordKind.FINISH, + ] + assert records[1].supersession == Supersession("tokens", 1, 2) + assert [r.value for r in records if r.kind is RecordKind.DATA] == [ + "a", + "a", + "b", + "c", + ] + assert [(r.producer_id, r.attempt, r.sequence) for r in records[2:]] == [ + ("tokens", 2, 1), + ("tokens", 2, 2), + ("tokens", 2, 3), + ("tokens", 2, 4), + ] + + +@activity.defn +async def stream_without_provider() -> str: + try: + activity_handle() + except ValueError as error: + return str(error) + return "opened" + + +@workflow.defn +class RunsActivityWithoutProvider: + @workflow.run + async def run(self) -> str: + return await workflow.execute_activity( + stream_without_provider, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_an_activity_without_a_provider_is_told(client: Client): + async with new_worker( + client, RunsActivityWithoutProvider, activities=[stream_without_provider] + ) as worker: + said = await client.execute_workflow( + RunsActivityWithoutProvider.run, + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + assert "no stream provider is registered" in said + + +async def test_a_client_handle_follows_the_chain_or_pins_a_run(client: Client): + provider = MemoryStreams() + streams_client = client_with(client, provider) + follower = get_stream_handle(streams_client, "wf") + assert follower.ref == StreamRef.for_workflow("wf") + pinned = get_stream_handle(streams_client, "wf", run_id="run", topic=EVENTS) + assert pinned.ref == StreamRef.for_workflow("wf", run_id="run", topic=EVENTS) + assert get_stream_handle(streams_client, pinned.ref).ref == pinned.ref + with pytest.raises(ValueError, match="carries its own"): + get_stream_handle(streams_client, pinned.ref, run_id="other") + with pytest.raises(ValueError, match="no stream provider is registered"): + get_stream_handle(client_with(client, None), "wf") + + +async def test_a_client_producer_and_the_workflow_share_a_topic(client: Client): + provider = MemoryStreams() + streams_client = client_with(client, provider) + workflow_id = new_workflow_id() + stream = get_stream_handle(streams_client, workflow_id) + outside = stream.producer(topic=EVENTS, producer_id="backend", attempt=1) + await outside.append({"from": "backend"}) + async with new_worker(streams_client, Publisher) as worker: + await streams_client.execute_workflow( + Publisher.run, 0, id=workflow_id, task_queue=worker.task_queue + ) + records = await read_all(stream.read(topic=EVENTS)) + assert [(r.producer_id, r.value) for r in records] == [ + ("backend", {"from": "backend"}), + ("", {"step": "init"}), + ("", None), + ] + + +async def test_an_activity_with_no_workflow_has_no_workflow_stream(): + env = ActivityEnvironment() + env.info = dataclasses.replace( + ActivityEnvironment.default_info(), workflow_id=None, workflow_run_id=None + ) + + async def standalone() -> None: + activity_handle() + + with pytest.raises(ValueError, match="not scheduled by a Workflow"): + await env.run(standalone) diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py index 8a2ac76859..96401febc3 100644 --- a/tests/contrib/streams/test_records_wire.py +++ b/tests/contrib/streams/test_records_wire.py @@ -83,6 +83,47 @@ def test_supersession_is_synthesized_from_observations(): assert attempts.note("other", 0, topic="t", previous=Cursor("c4")) is None +def test_an_attempt_that_goes_backwards_is_reported(): + said: list[str] = [] + attempts = AttemptTracker(said.append) + assert attempts.note("model", 2, topic="t", previous=BEGINNING) is None + assert attempts.note("model", 1, topic="t", previous=Cursor("c3")) is None + assert len(said) == 1 + assert "attempt 1" in said[0] and "behind attempt 2" in said[0] + assert "'model'" in said[0] + + +def test_a_repeat_of_the_current_attempt_is_not_reported(): + said: list[str] = [] + attempts = AttemptTracker(said.append) + attempts.note("model", 1, topic="t", previous=BEGINNING) + assert attempts.note("model", 1, topic="t", previous=Cursor("c1")) is None + assert said == [] + + +def test_the_decoder_reports_a_backwards_attempt(): + said: list[str] = [] + converter = DataConverter.default.payload_converter + decoder = RecordDecoder(converter, int, after=BEGINNING, warn=said.append) + + def record(attempt: int) -> WireRecord: + return to_wire( + converter, + topic="t", + kind=RecordKind.DATA, + value=attempt, + producer_id="model", + attempt=attempt, + sequence=1, + ) + + decoder.decode(Cursor("c1"), record(2)) + out = decoder.decode(Cursor("c2"), record(1)) + # Still delivered, because dropping it would hide what the store holds. + assert [r.value for r in out] == [1] + assert len(said) == 1 and "behind attempt 2" in said[0] + + def test_the_decoder_skips_a_record_it_cannot_decode(): said: list[str] = [] converter = DataConverter.default.payload_converter From 155118f45866c663ef7f36cd1be762c925263909 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 17:26:05 -0700 Subject: [PATCH 2/7] Included the scheduling run id in an Activity's stream producer id. Activity ids repeat across the runs of a chain while the stream and its dedupe state span the chain, so run 2's first append was refused as stale or silently matched run 1's position. --- temporalio/contrib/streams/__init__.py | 4 +- temporalio/contrib/streams/_handles.py | 17 ++++-- tests/contrib/streams/test_outside_handles.py | 56 +++++++++++++++++-- 3 files changed, 64 insertions(+), 13 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 656e75974c..b31508a3de 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -36,8 +36,8 @@ - Workflow code publishes to its own stream with :func:`workflow_writer`. - An Activity reaches the stream of the Workflow that scheduled it with - :func:`activity_handle`, and writes as itself: its Activity id and its - Temporal attempt. + :func:`activity_handle`, and writes as itself: its Activity id with the + scheduling run id, and its Temporal attempt. - Any process holding a client reaches a Workflow's stream with :func:`get_stream_handle`, and writes with a producer id and an attempt of its own. diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 21d13b9ea6..a0ed54bb81 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -1,8 +1,10 @@ """Reaching a stream from an Activity or from client code. An Activity reaches the stream of the Workflow that scheduled it, and writes -as itself: its Activity id is the producer id and its Temporal attempt is -the producer attempt, so a retry is reported to readers as ``SUPERSEDED``. +as itself: its producer id is ``@`` and its +Temporal attempt is the producer attempt, so a retry is reported to readers +as ``SUPERSEDED``. The run is part of the id because a stream outlives a +run, and Activity ids repeat across the runs of a chain. Any process holding a client reaches a Workflow's stream by Workflow id and writes with a producer id and attempt of its own. """ @@ -98,9 +100,9 @@ def producer( ) -> StreamProducer[Any]: """A producer on ``topic`` that writes as this Activity attempt. - The producer id is the Activity id and the attempt is the Activity's - Temporal attempt, so the first record of a retry makes readers see - ``SUPERSEDED``. A retry starts a new sequence, and the store + The producer id is ``@`` and the + attempt is the Activity's Temporal attempt, so the first record of a + retry makes readers see ``SUPERSEDED``. A retry starts a new sequence, and the store deduplicates a repeated append within one attempt. """ inner: Any = self._inner @@ -131,7 +133,10 @@ def activity_handle() -> ActivityStreamHandle: provider = provider_for_activity() ref = StreamRef.for_workflow(info.workflow_id, run_id=info.workflow_run_id) inner = provider.get_stream_handle(temporalio.activity.client(), ref) - return ActivityStreamHandle(inner, info.activity_id, info.attempt) + # Activity ids are counters within one run, so two runs of a chain would + # otherwise share producer state in the stream they share. + producer_id = f"{info.activity_id}@{info.workflow_run_id}" + return ActivityStreamHandle(inner, producer_id, info.attempt) def get_stream_handle( diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index 2d2f0ee115..40ee21dae6 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -115,7 +115,8 @@ async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): RecordKind.DATA, RecordKind.FINISH, ] - assert records[1].supersession == Supersession("tokens", 1, 2) + producer_id = f"tokens@{(await streams_client.get_workflow_handle(workflow_id).describe()).run_id}" + assert records[1].supersession == Supersession(producer_id, 1, 2) assert [r.value for r in records if r.kind is RecordKind.DATA] == [ "a", "a", @@ -123,10 +124,10 @@ async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): "c", ] assert [(r.producer_id, r.attempt, r.sequence) for r in records[2:]] == [ - ("tokens", 2, 1), - ("tokens", 2, 2), - ("tokens", 2, 3), - ("tokens", 2, 4), + (producer_id, 2, 1), + (producer_id, 2, 2), + (producer_id, 2, 3), + (producer_id, 2, 4), ] @@ -204,3 +205,48 @@ async def standalone() -> None: with pytest.raises(ValueError, match="not scheduled by a Workflow"): await env.run(standalone) + + +@activity.defn +async def count_twice() -> str: + producer = activity_handle().producer(topic=TOKENS) + await producer.append("one") + await producer.append("two") + return producer.producer_id + + +@workflow.defn +class CountThenContinue: + @workflow.run + async def run(self, runs_left: int) -> list[str]: + # The same Activity id in every run, as a counter-based id would be. + producer_id = await workflow.execute_activity( + count_twice, + activity_id="count", + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy(maximum_attempts=1), + ) + if runs_left: + workflow.continue_as_new(runs_left - 1) + return [producer_id] + + +async def test_an_activity_producer_writes_in_every_run_of_a_chain(client: Client): + provider = MemoryStreams() + streams_client = client_with(client, provider) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, CountThenContinue, activities=[count_twice] + ) as worker: + handle = await streams_client.start_workflow( + CountThenContinue.run, 1, id=workflow_id, task_queue=worker.task_queue + ) + (last_producer,) = await handle.result() + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + records = await read_all(stream.read()) + # Both runs' records, under two producers: the run is part of the id. + assert [r.value for r in records] == ["one", "two", "one", "two"] + producers = [r.producer_id for r in records] + assert producers[0] == producers[1] != producers[2] == producers[3] + assert producers[3] == last_producer + assert last_producer.endswith("@" + (await handle.describe()).run_id) From 860890661f7bf8163f36472713c26311fba697f7 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 13:13:53 -0700 Subject: [PATCH 3/7] Marked records from an older attempt as stale. Such records reached the consumer as ordinary DATA and only the Worker log said otherwise. The package docs now also say an Activity's read ends with its scheduling run. --- temporalio/contrib/streams/__init__.py | 3 ++- temporalio/contrib/streams/_handles.py | 10 +++++++--- temporalio/contrib/streams/_policy.py | 9 +++++++-- temporalio/contrib/streams/_record.py | 6 ++++++ temporalio/contrib/streams/_wire.py | 3 +++ tests/contrib/streams/test_outside_handles.py | 3 ++- tests/contrib/streams/test_records_wire.py | 7 +++++-- 7 files changed, 32 insertions(+), 9 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index b2d98938d7..99ddd1dd94 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -37,7 +37,8 @@ - Workflow code publishes to its own stream with :func:`workflow_writer`. - An Activity reaches the stream of the Workflow that scheduled it with :func:`activity_handle`, and writes as itself: its Activity id with the - scheduling run id, and its Temporal attempt. + scheduling run id, and its Temporal attempt. The handle is pinned to that + run, so a read on it ends when the run closes, Continue-as-New included. - Any process holding a client reaches a Workflow's stream with :func:`get_stream_handle`, and writes with a producer id and an attempt of its own. diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index a0ed54bb81..99da0f61b4 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -12,7 +12,7 @@ from __future__ import annotations from collections.abc import AsyncGenerator -from typing import Any, TypeVar, overload +from typing import Any, TypeVar, cast, overload import temporalio.activity from temporalio.client import Client @@ -82,7 +82,9 @@ def read( result_type: type | None = None, ) -> AsyncGenerator[StreamRecord[Any], None]: """See :meth:`temporalio.contrib.streams.StreamHandle.read`.""" - inner: Any = self._inner + # One call forwards both overloads, which the checker can't match to + # the protocol's overloads separately. + inner = cast(Any, self._inner) return inner.read(topic=topic, after=after, result_type=result_type) async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: @@ -105,7 +107,9 @@ def producer( retry makes readers see ``SUPERSEDED``. A retry starts a new sequence, and the store deduplicates a repeated append within one attempt. """ - inner: Any = self._inner + # One call forwards both overloads, which the checker can't match to + # the protocol's overloads separately. + inner = cast(Any, self._inner) return inner.producer( topic=topic, producer_id=self._producer_id, attempt=self._attempt ) diff --git a/temporalio/contrib/streams/_policy.py b/temporalio/contrib/streams/_policy.py index 40899fed85..0e4b93dfe5 100644 --- a/temporalio/contrib/streams/_policy.py +++ b/temporalio/contrib/streams/_policy.py @@ -32,6 +32,10 @@ def __init__(self, warn: Callable[[str], None] | None = None) -> None: self._attempts: dict[str, int] = {} self._warn = warn + def behind(self, producer_id: str, attempt: int) -> bool: + """Whether this read already delivered a newer attempt of ``producer_id``.""" + return bool(producer_id) and 0 < attempt < self._attempts.get(producer_id, 0) + def note( self, producer_id: str, attempt: int, *, topic: str, previous: Cursor ) -> StreamRecord[Any] | None: @@ -46,8 +50,9 @@ def note( A producer that declares no attempt supersedes nothing, because there is no generation to compare. - An attempt that goes backwards supersedes nothing either, and is - reported through ``warn`` rather than passed off as ordinary data. + An attempt that goes backwards supersedes nothing either. It is + reported through ``warn``, and the reader marks its records + ``stale`` rather than passing them off as ordinary data. A lower attempt after a higher one means an older attempt was still writing after a newer one started, such as an Activity attempt that timed out but kept running. A consumer that reads it as the current diff --git a/temporalio/contrib/streams/_record.py b/temporalio/contrib/streams/_record.py index f8ba72ae48..af8f112d9c 100644 --- a/temporalio/contrib/streams/_record.py +++ b/temporalio/contrib/streams/_record.py @@ -110,3 +110,9 @@ class StreamRecord(Generic[T]): """The published value. Set on ``DATA`` only.""" supersession: Supersession | None = None """The attempt change being reported. Set on ``SUPERSEDED`` only.""" + stale: bool = False + """An older attempt wrote it after this read delivered a newer attempt. + + Such a record is not part of the producer's current answer, for example + the last writes of an Activity attempt that timed out but kept running. + """ diff --git a/temporalio/contrib/streams/_wire.py b/temporalio/contrib/streams/_wire.py index b4911e63cf..5811a6baa1 100644 --- a/temporalio/contrib/streams/_wire.py +++ b/temporalio/contrib/streams/_wire.py @@ -9,6 +9,7 @@ from __future__ import annotations +import dataclasses from collections.abc import Callable from typing import Any @@ -132,6 +133,8 @@ def decode(self, cursor: Cursor, wire: WireRecord) -> list[StreamRecord[Any]]: raise StreamRecordError( f"stream record at {cursor} could not be decoded: {error}", cursor ) from error + if self._attempts.behind(wire.producer_id, wire.attempt): + record = dataclasses.replace(record, stale=True) out: list[StreamRecord[Any]] = [] superseded = self._attempts.note( wire.producer_id, wire.attempt, topic=wire.topic, previous=self._previous diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index 40ee21dae6..61c3878335 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -115,7 +115,8 @@ async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): RecordKind.DATA, RecordKind.FINISH, ] - producer_id = f"tokens@{(await streams_client.get_workflow_handle(workflow_id).describe()).run_id}" + described = await streams_client.get_workflow_handle(workflow_id).describe() + producer_id = f"tokens@{described.run_id}" assert records[1].supersession == Supersession(producer_id, 1, 2) assert [r.value for r in records if r.kind is RecordKind.DATA] == [ "a", diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py index 82a02f3a88..f5c29cccc1 100644 --- a/tests/contrib/streams/test_records_wire.py +++ b/tests/contrib/streams/test_records_wire.py @@ -120,10 +120,13 @@ def record(attempt: int) -> WireRecord: sequence=1, ) - decoder.decode(Cursor("c1"), record(2)) + current = decoder.decode(Cursor("c1"), record(2)) out = decoder.decode(Cursor("c2"), record(1)) - # Still delivered, because dropping it would hide what the store holds. + # Still delivered, because dropping it would hide what the store holds, + # but marked, so the consumer can keep it out of the current answer. assert [r.value for r in out] == [1] + assert [r.stale for r in out] == [True] + assert [r.stale for r in current] == [False] assert len(said) == 1 and "behind attempt 2" in said[0] From 0aba92866c0e707f1e1f3ef579be21e3b86a9e3e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 22:09:00 -0700 Subject: [PATCH 4/7] Reached a Workflow's stream through Core's stream service. Handles and producers call Core's in-process service, which checks owners, mints cursors and reports supersession. Lang converts values, takes the digest and content hashes before the codec, and runs the codec both ways. An Activity sends its typed identity so Core derives its producer id. --- temporalio/contrib/streams/__init__.py | 9 + temporalio/contrib/streams/_handles.py | 391 ++++++++++++++++-- temporalio/contrib/streams/_plugin.py | 4 - tests/contrib/streams/test_outside_handles.py | 189 +++++++-- tests/contrib/streams/test_workflow_writer.py | 56 +-- 5 files changed, 520 insertions(+), 129 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index e9d11ca772..e7729deedc 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -63,6 +63,13 @@ StreamStorageError, StreamUnsupportedError, ) +from temporalio.contrib.streams._handles import ( + ActivityStreamHandle, + StreamHandle, + StreamProducer, + activity_handle, + get_stream_handle, +) from temporalio.contrib.streams._plugin import StreamStorePlugin from temporalio.contrib.streams._record import ( Cursor, @@ -85,9 +92,11 @@ "StreamCursorError", "StreamError", "StreamExpiredError", + "StreamHandle", "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamOwnerKind", + "StreamProducer", "StreamProducerError", "StreamRecord", "StreamRecordError", diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 99da0f61b4..46111feccd 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -1,35 +1,352 @@ """Reaching a stream from an Activity or from client code. +Every call goes to Core's stream service, which checks the owner, mints and +checks cursors, and synthesizes ``SUPERSEDED``. What stays here is what +needs the plaintext or the user's types: converting values, the two hashes +taken before the codec, and the codec itself. + An Activity reaches the stream of the Workflow that scheduled it, and writes -as itself: its producer id is ``@`` and its -Temporal attempt is the producer attempt, so a retry is reported to readers -as ``SUPERSEDED``. The run is part of the id because a stream outlives a -run, and Activity ids repeat across the runs of a chain. +as itself: Core derives its producer id as +``@`` and its Temporal attempt is the +producer attempt, so a retry is reported to readers as ``SUPERSEDED``. The +run is part of the id because a stream outlives a run, and Activity ids +repeat across the runs of a chain. + Any process holding a client reaches a Workflow's stream by Workflow id and writes with a producer id and attempt of its own. """ from __future__ import annotations +import asyncio from collections.abc import AsyncGenerator -from typing import Any, TypeVar, cast, overload +from datetime import timedelta +from typing import Any, Generic, TypeVar, cast, overload import temporalio.activity +from temporalio.bridge.proto.streams import ( + ActivityProducer, + AppendRecord, + AppendRequest, + LatestRequest, + NamedProducer, + ReadRequest, + StreamAddress, + StreamOwnerKind, +) from temporalio.client import Client +from temporalio.contrib.streams._body import batch_digest, content_hash, encode_bodies from temporalio.contrib.streams._cursor import BEGINNING from temporalio.contrib.streams._plugin import ( - provider_for_activity, - provider_for_client, + StreamStorePlugin, + call, + store_for_client, ) -from temporalio.contrib.streams._provider import StreamHandle, StreamProducer -from temporalio.contrib.streams._record import Cursor, StreamRecord +from temporalio.contrib.streams._record import Cursor, RecordKind, StreamRecord from temporalio.contrib.streams._ref import StreamRef -from temporalio.contrib.streams._topic import StreamTopic +from temporalio.contrib.streams._topic import StreamTopic, resolve_topic +from temporalio.contrib.streams._wire import WireRecord, from_read, to_wire -__all__ = ["ActivityStreamHandle", "activity_handle", "get_stream_handle"] +__all__ = [ + "ActivityStreamHandle", + "StreamHandle", + "StreamProducer", + "activity_handle", + "get_stream_handle", +] T = TypeVar("T") +_READ_WAIT = timedelta(seconds=30) + + +class StreamProducer(Generic[T]): + """Appends to one topic as one producer attempt. + + Every append is visible as soon as the store accepts it. Each record + carries the producer id, the attempt and a sequence that starts at one + for each attempt, which lets the store deduplicate a retry and lets a + reader tell a retry from a new attempt. + """ + + def __init__( + self, + handle: StreamHandle, + topic: str, + producer_id: str, + attempt: int, + activity: ActivityProducer | None = None, + ) -> None: + """Prefer :meth:`StreamHandle.producer`.""" + self._handle = handle + self._topic = topic + self._producer_id = producer_id + self._attempt = attempt + self._activity = activity + self._sequence = 1 + self._last = BEGINNING + # A batch reads the sequence, then awaits the codec and Core, then + # moves it on; calls one at a time keep each batch's sequences. + self._lock = asyncio.Lock() + + @property + def producer_id(self) -> str: + """Who this producer writes as.""" + return self._producer_id + + @property + def attempt(self) -> int: + """The attempt this producer writes.""" + return self._attempt + + async def append(self, *values: T) -> Cursor: + """Append ``values`` as one batch and return the cursor of its last record. + + The batch lands whole and in order. The producer moves its sequence + past the batch only when the append succeeds, so a call that raised + :class:`temporalio.contrib.streams.StreamOutcomeUnknownError` can be + repeated with the same values: the store answers a repeat of the + batch it holds last for this producer attempt with the original + position and writes nothing. + + An empty call writes nothing and returns the cursor of this + producer's last record, or ``BEGINNING`` when it has written none. + + Calls on one producer run one at a time, in the order they were + made, so concurrent calls take consecutive sequences. A call that was + cancelled may or may not have written its batch, and the producer + cannot tell: start a new attempt (a new producer with a higher + ``attempt``) rather than continuing this one. + + Raises: + StreamProducerError: The sequence was already used with different + content, or it is below the newest one the store holds for + this producer attempt. + StreamOutcomeUnknownError: The store may or may not have written + the batch. Retry the same values on this producer. + StreamRefusedError: The store refused the batch, for example + because it is out of memory. Nothing was written. + StreamClosedError: The stream refuses appends. + """ + async with self._lock: + if not values: + return self._last + return await self._write(RecordKind.DATA, values) + + async def finish(self) -> Cursor: + """Write ``FINISH`` for this producer on this topic and return its cursor. + + Says this producer has nothing more to send. It does not say the + work behind it succeeded, and it does not end anyone's read. + """ + async with self._lock: + return await self._write(RecordKind.FINISH, [None]) + + async def _write(self, kind: RecordKind, values: Any) -> Cursor: + converter = self._handle._client.data_converter + wires = [ + to_wire( + converter.payload_converter, + topic=self._topic, + kind=kind, + value=value, + producer_id=self._producer_id, + attempt=self._attempt, + sequence=self._sequence + index, + ) + for index, value in enumerate(values) + ] + request = AppendRequest( + stream=self._handle._address(self._topic), + sequence=self._sequence, + records=await _append_records(converter, wires), + digest=batch_digest(wires), + ) + if self._activity is not None: + request.activity.CopyFrom(self._activity) + else: + request.named.CopyFrom( + NamedProducer(producer_id=self._producer_id, attempt=self._attempt) + ) + service = await self._handle._plugin._service_for(self._handle._client) + response = await call(service.append(request)) + self._sequence += len(wires) + self._last = Cursor(response.last_cursor) + return self._last + + +async def _append_records( + converter: Any, wires: list[WireRecord] +) -> list[AppendRecord]: + bodies = iter( + await encode_bodies(converter, [w.body for w in wires if w.HasField("body")]) + ) + records = [] + for wire in wires: + record = AppendRecord(kind=wire.kind) + if wire.HasField("body"): + record.body.CopyFrom(next(bodies)) + record.content_hash = content_hash(wire.body) + records.append(record) + return records + + +class StreamHandle: + """One stream, addressed by topic, from outside Workflow code. + + A handle on a Workflow's stream follows the run chain unless its + :attr:`ref` is pinned to a run. A topic is a + :class:`temporalio.contrib.streams.StreamTopic`, which carries the record + type, or a plain string with ``result_type=`` for a name decided at + runtime. A call that names no topic addresses the topic of :attr:`ref`. + """ + + def __init__( + self, plugin: StreamStorePlugin, client: Client, ref: StreamRef + ) -> None: + """Prefer :func:`temporalio.contrib.streams.get_stream_handle`.""" + self._plugin = plugin + self._client = client + self._ref = ref + + @property + def ref(self) -> StreamRef: + """The stream this handle is on, as data another process can open.""" + return self._ref + + @overload + def read( + self, *, topic: StreamTopic[T], after: Cursor = ... + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: type[T], + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: ... + + def read( + self, + *, + topic: str | StreamTopic[Any] | None = None, + after: Cursor = BEGINNING, + result_type: type | None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: + """Yield the records on ``topic`` after ``after`` as they arrive. + + ``BEGINNING`` starts at the oldest record the topic retains, and + ``END`` at whatever is appended after the read starts. Any other + cursor came from a record of this stream, and the read resumes just + past it, so a reader that keeps the last cursor it handled and hands + it back sees every stored record once. A resumed read knows the + attempt of the record at its cursor, so a new attempt of that + producer still arrives as ``SUPERSEDED``. Earlier attempts of other + producers are not known to it. The read ends when the owner's run + chain is closed and every retained record has been delivered. The + result is a generator, so a caller that stops early calls + ``aclose()`` on it. + + Raises: + ValueError: ``result_type`` was passed with a topic definition, + or the topic is empty. Raised by this call. + StreamCursorError: ``after`` came from another store or another + stream. Raised by the first iteration. + StreamExpiredError: ``after`` names a record the store no longer + retains, or the read fell behind retention while it ran. + StreamRecordError: A record did not decode. Read on from its + cursor to skip it. + """ + name, result_type = self._resolve(topic, result_type) + return self._read(name, after, result_type) + + async def _read( + self, topic: str, after: Cursor, result_type: type | None + ) -> AsyncGenerator[StreamRecord[Any], None]: + service = await self._plugin._service_for(self._client) + request = ReadRequest(stream=self._address(topic), after=after.token) + request.wait.FromTimedelta(_READ_WAIT) + while True: + response = await call(service.read(request)) + for record in response.records: + yield await from_read(self._client.data_converter, record, result_type) + if response.done: + return + request.after = response.cursor + request.state = response.state + + async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: + """The cursor of the newest record on ``topic``, or ``BEGINNING`` when empty. + + ``read(after=latest())`` yields only what is appended after this call + returned, which is how a client positions itself before it sends + something the Workflow answers on the stream. + """ + name, _ = self._resolve(topic, None) + service = await self._plugin._service_for(self._client) + response = await call(service.latest(LatestRequest(stream=self._address(name)))) + return Cursor(response.cursor) + + @overload + def producer( + self, *, topic: StreamTopic[T], producer_id: str, attempt: int + ) -> StreamProducer[T]: ... + + @overload + def producer( + self, *, topic: str | None = None, producer_id: str, attempt: int + ) -> StreamProducer[Any]: ... + + def producer( + self, + *, + topic: str | StreamTopic[Any] | None = None, + producer_id: str, + attempt: int, + ) -> StreamProducer[Any]: + """A producer on ``topic`` that writes as ``producer_id`` in ``attempt``. + + A process that restarts its work raises ``attempt``, which starts a + new sequence and tells readers that what the earlier attempt wrote is + superseded. Keep one producer object per ``(topic, producer_id, + attempt)``. Each object numbers its own records from one, so a second + object for the same session is refused as stale. + + Raises: + ValueError: ``producer_id`` is empty or ``attempt`` is below one. + """ + if not producer_id: + raise ValueError("producer_id must not be empty") + if isinstance(attempt, bool) or not isinstance(attempt, int) or attempt < 1: + raise ValueError(f"attempt must be an int of at least 1, got {attempt!r}") + name, _ = self._resolve(topic, None) + return StreamProducer(self, name, producer_id, attempt) + + def _resolve( + self, topic: str | StreamTopic[Any] | None, result_type: type | None + ) -> tuple[str, type | None]: + return resolve_topic(self._ref.topic if topic is None else topic, result_type) + + def _address(self, topic: str) -> StreamAddress: + return StreamAddress( + namespace=self._client.namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=self._ref.workflow_id, + run_id=self._ref.run_id or "", + topic=topic, + ) + class ActivityStreamHandle: """The scheduling Workflow's stream, as the running Activity sees it. @@ -40,11 +357,10 @@ class ActivityStreamHandle: producer. """ - def __init__(self, inner: StreamHandle, producer_id: str, attempt: int) -> None: + def __init__(self, inner: StreamHandle, identity: ActivityProducer) -> None: """Prefer :func:`temporalio.contrib.streams.activity_handle`.""" self._inner = inner - self._producer_id = producer_id - self._attempt = attempt + self._identity = identity @property def ref(self) -> StreamRef: @@ -82,8 +398,7 @@ def read( result_type: type | None = None, ) -> AsyncGenerator[StreamRecord[Any], None]: """See :meth:`temporalio.contrib.streams.StreamHandle.read`.""" - # One call forwards both overloads, which the checker can't match to - # the protocol's overloads separately. + # One call forwards every overload, which the checker can't match one by one. inner = cast(Any, self._inner) return inner.read(topic=topic, after=after, result_type=result_type) @@ -104,14 +419,17 @@ def producer( The producer id is ``@`` and the attempt is the Activity's Temporal attempt, so the first record of a - retry makes readers see ``SUPERSEDED``. A retry starts a new sequence, and the store - deduplicates a repeated append within one attempt. + retry makes readers see ``SUPERSEDED``. A retry starts a new sequence, + and the store deduplicates a repeated append within one attempt. """ - # One call forwards both overloads, which the checker can't match to - # the protocol's overloads separately. - inner = cast(Any, self._inner) - return inner.producer( - topic=topic, producer_id=self._producer_id, attempt=self._attempt + name, _ = self._inner._resolve(topic, None) + identity = self._identity + return StreamProducer( + self._inner, + name, + f"{identity.activity_id}@{identity.run_id}", + identity.attempt, + activity=identity, ) @@ -125,22 +443,25 @@ def activity_handle() -> ActivityStreamHandle: Raises: ValueError: The Activity was not scheduled by a Workflow, or no stream - provider is registered on the Worker or its client. + store is registered on the Worker's client. RuntimeError: Not called from inside an Activity. """ info = temporalio.activity.info() - if info.workflow_id is None: + if info.workflow_id is None or info.workflow_run_id is None: raise ValueError( "this Activity was not scheduled by a Workflow, so it has no Workflow " "stream" ) - provider = provider_for_activity() + client = temporalio.activity.client() ref = StreamRef.for_workflow(info.workflow_id, run_id=info.workflow_run_id) - inner = provider.get_stream_handle(temporalio.activity.client(), ref) - # Activity ids are counters within one run, so two runs of a chain would - # otherwise share producer state in the stream they share. - producer_id = f"{info.activity_id}@{info.workflow_run_id}" - return ActivityStreamHandle(inner, producer_id, info.attempt) + identity = ActivityProducer( + workflow_id=info.workflow_id, + run_id=info.workflow_run_id, + activity_id=info.activity_id, + attempt=info.attempt, + ) + inner = StreamHandle(store_for_client(client), client, ref) + return ActivityStreamHandle(inner, identity) def get_stream_handle( @@ -150,7 +471,7 @@ def get_stream_handle( run_id: str | None = None, topic: str | StreamTopic[Any] | None = None, ) -> StreamHandle: - """A handle on a Workflow's stream, through the client's provider. + """A handle on a Workflow's stream, through the client's store. Without ``run_id`` the handle follows the Workflow's run chain, so a read continues across Continue-as-New and ends when the chain closes. @@ -158,7 +479,7 @@ def get_stream_handle( names none addresses. Args: - client: A client that carries a stream provider plugin. + client: A client that carries a stream store plugin. workflow_id: The Workflow that owns the stream, or a :class:`temporalio.contrib.streams.StreamRef` that names it. run_id: Pin the handle to one run. @@ -167,7 +488,7 @@ def get_stream_handle( Raises: ValueError: ``run_id`` or ``topic`` was given with a ref, or no - stream provider is registered on ``client``. + stream store is registered on ``client``. """ if isinstance(workflow_id, StreamRef): if run_id is not None or topic is not None: @@ -175,4 +496,4 @@ def get_stream_handle( ref = workflow_id else: ref = StreamRef.for_workflow(workflow_id, run_id=run_id, topic=topic) - return provider_for_client(client).get_stream_handle(client, ref) + return StreamHandle(store_for_client(client), client, ref) diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py index c2c08e8668..f8da83c641 100644 --- a/temporalio/contrib/streams/_plugin.py +++ b/temporalio/contrib/streams/_plugin.py @@ -43,10 +43,6 @@ T = TypeVar("T") -_activity_provider: contextvars.ContextVar[StreamProviderPlugin | None] = ( - contextvars.ContextVar("__temporal_contrib_streams_provider", default=None) -) - class StreamStorePlugin(SimplePlugin): """The base class of the store configuration classes. diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index 61c3878335..e4358f48bc 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -2,24 +2,30 @@ An Activity writes to the stream of the Workflow that scheduled it as itself, so its retry is reported to readers as ``SUPERSEDED``. A client -reaches a Workflow's stream by Workflow id through its provider. +reaches a Workflow's stream by Workflow id through its store. """ from __future__ import annotations -import asyncio import dataclasses import uuid +from collections.abc import Sequence from datetime import timedelta -from typing import Any import pytest from temporalio import activity, workflow +from temporalio.api.common.v1 import Payload from temporalio.client import Client from temporalio.common import RetryPolicy from temporalio.contrib.streams import ( + BEGINNING, + Cursor, RecordKind, + StreamClosedError, + StreamCursorError, + StreamNotFoundError, + StreamProducerError, StreamRef, Supersession, activity_handle, @@ -27,37 +33,35 @@ topic, ) from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.converter import DataConverter, PayloadCodec from temporalio.exceptions import ApplicationError from temporalio.testing import ActivityEnvironment +from tests.contrib.streams._support import connect_with, new_workflow_id, read_all from tests.contrib.streams.test_workflow_writer import EVENTS, Publisher from tests.helpers import new_worker TOKENS = topic("tokens", str) -def client_with(client: Client, provider: MemoryStreams | None) -> Client: - config = client.config() - config["plugins"] = [provider] if provider is not None else [] - return Client(**config) - - -def new_workflow_id() -> str: - return f"streams-publish-{uuid.uuid4().hex}" +@workflow.defn +class Waits: + def __init__(self) -> None: + self.done = False + @workflow.run + async def run(self) -> None: + await workflow.wait_condition(lambda: self.done) -async def read_all(records: Any, timeout: float = 10.0) -> list: - """Read until the stream ends, which is when its owner closes.""" - out = [] + @workflow.signal + def finish(self) -> None: + self.done = True - async def pull() -> None: - async for record in records: - out.append(record) - try: - await asyncio.wait_for(pull(), timeout) - finally: - await records.aclose() - return out +async def running_owner(client: Client, task_queue: str) -> str: + # Core asks the server about the owner, so the stream's Workflow must exist. + workflow_id = new_workflow_id() + await client.start_workflow(Waits.run, id=workflow_id, task_queue=task_queue) + return workflow_id @activity.defn @@ -91,8 +95,7 @@ async def run(self, tokens: list[str]) -> bool: async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): - provider = MemoryStreams() - streams_client = client_with(client, provider) + streams_client = await connect_with(client, MemoryStreams()) workflow_id = new_workflow_id() async with new_worker( streams_client, RunsStreamingActivity, activities=[stream_tokens] @@ -116,6 +119,7 @@ async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): RecordKind.FINISH, ] described = await streams_client.get_workflow_handle(workflow_id).describe() + # Core derives the same producer id the handle reports. producer_id = f"tokens@{described.run_id}" assert records[1].supersession == Supersession(producer_id, 1, 2) assert [r.value for r in records if r.kind is RecordKind.DATA] == [ @@ -133,7 +137,7 @@ async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): @activity.defn -async def stream_without_provider() -> str: +async def stream_without_store() -> str: try: activity_handle() except ValueError as error: @@ -142,29 +146,28 @@ async def stream_without_provider() -> str: @workflow.defn -class RunsActivityWithoutProvider: +class RunsActivityWithoutStore: @workflow.run async def run(self) -> str: return await workflow.execute_activity( - stream_without_provider, start_to_close_timeout=timedelta(seconds=10) + stream_without_store, start_to_close_timeout=timedelta(seconds=10) ) -async def test_an_activity_without_a_provider_is_told(client: Client): +async def test_an_activity_without_a_store_is_told(client: Client): async with new_worker( - client, RunsActivityWithoutProvider, activities=[stream_without_provider] + client, RunsActivityWithoutStore, activities=[stream_without_store] ) as worker: said = await client.execute_workflow( - RunsActivityWithoutProvider.run, + RunsActivityWithoutStore.run, id=new_workflow_id(), task_queue=worker.task_queue, ) - assert "no stream provider is registered" in said + assert "no stream store is registered" in said async def test_a_client_handle_follows_the_chain_or_pins_a_run(client: Client): - provider = MemoryStreams() - streams_client = client_with(client, provider) + streams_client = await connect_with(client, MemoryStreams()) follower = get_stream_handle(streams_client, "wf") assert follower.ref == StreamRef.for_workflow("wf") pinned = get_stream_handle(streams_client, "wf", run_id="run", topic=EVENTS) @@ -172,27 +175,126 @@ async def test_a_client_handle_follows_the_chain_or_pins_a_run(client: Client): assert get_stream_handle(streams_client, pinned.ref).ref == pinned.ref with pytest.raises(ValueError, match="carries its own"): get_stream_handle(streams_client, pinned.ref, run_id="other") - with pytest.raises(ValueError, match="no stream provider is registered"): - get_stream_handle(client_with(client, None), "wf") + with pytest.raises(ValueError, match="no stream store is registered"): + get_stream_handle(client, "wf") async def test_a_client_producer_and_the_workflow_share_a_topic(client: Client): - provider = MemoryStreams() - streams_client = client_with(client, provider) + streams_client = await connect_with(client, MemoryStreams()) workflow_id = new_workflow_id() - stream = get_stream_handle(streams_client, workflow_id) - outside = stream.producer(topic=EVENTS, producer_id="backend", attempt=1) - await outside.append({"from": "backend"}) async with new_worker(streams_client, Publisher) as worker: - await streams_client.execute_workflow( + handle = await streams_client.start_workflow( Publisher.run, 0, id=workflow_id, task_queue=worker.task_queue ) + await handle.result() + stream = get_stream_handle(streams_client, workflow_id) records = await read_all(stream.read(topic=EVENTS)) assert [(r.producer_id, r.value) for r in records] == [ - ("backend", {"from": "backend"}), ("", {"step": "init"}), ("", None), ] + # The chain closed, so an outside producer is refused there. + outside = stream.producer(topic=EVENTS, producer_id="backend", attempt=1) + with pytest.raises(StreamClosedError): + await outside.append({"from": "backend"}) + + +async def test_a_producer_retry_dedupes_and_a_divergent_one_is_refused( + client: Client, +): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, Waits) as worker: + workflow_id = await running_owner(streams_client, worker.task_queue) + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + assert await stream.latest() == BEGINNING + producer = stream.producer(producer_id="p", attempt=1) + assert await producer.append() == BEGINNING + first = await producer.append("a", "b") + assert await producer.append() == first + assert await stream.latest() == first + # A second object for the same session starts at one again: a repeat + # of the newest batch dedupes to the original position. + again = stream.producer(producer_id="p", attempt=1) + assert await again.append("a", "b") == first + divergent = stream.producer(producer_id="p", attempt=1) + with pytest.raises(StreamProducerError): + await divergent.append("x", "y") + owner = streams_client.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + records = await read_all(stream.read()) + assert [r.value for r in records] == ["a", "b"] + + +async def test_a_resumed_read_continues_after_its_cursor(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, Waits) as worker: + workflow_id = await running_owner(streams_client, worker.task_queue) + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + producer = stream.producer(producer_id="p", attempt=1) + middle = await producer.append("a", "b") + await producer.append("c") + # A read from END starts at its first iteration, so a reader that must + # not miss the next record positions itself with latest() first. + live = stream.read(after=await stream.latest()) + await producer.append("d") + assert (await anext(live)).value == "d" + await live.aclose() + with pytest.raises(StreamCursorError): + await anext(stream.read(after=Cursor("not a cursor"))) + owner = streams_client.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + assert [r.value for r in await read_all(stream.read(after=middle))] == ["c", "d"] + + +async def test_a_stream_without_an_owner_is_not_found(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + stream = get_stream_handle(streams_client, f"no-such-owner-{uuid.uuid4()}") + with pytest.raises(StreamNotFoundError): + await stream.producer(producer_id="p", attempt=1).append("a") + + +async def test_a_producer_refuses_a_bad_identity(client: Client): + stream = get_stream_handle(await connect_with(client, MemoryStreams()), "wf") + with pytest.raises(ValueError, match="producer_id"): + stream.producer(producer_id="", attempt=1) + for attempt in (0, True, 1.5): + with pytest.raises(ValueError, match="attempt"): + stream.producer(producer_id="p", attempt=attempt) # type: ignore[arg-type] + + +class NonceCodec(PayloadCodec): + """Encrypts with a fresh nonce per call, so two encodings never match.""" + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload( + metadata={"encoding": b"binary/nonce"}, + data=uuid.uuid4().bytes + p.SerializeToString(), + ) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [Payload.FromString(p.data[16:]) for p in payloads] + + +async def test_a_retry_through_a_nonce_codec_still_dedupes(client: Client): + coded = await connect_with( + client, MemoryStreams(), DataConverter(payload_codec=NonceCodec()) + ) + async with new_worker(coded, Waits) as worker: + workflow_id = await running_owner(coded, worker.task_queue) + stream = get_stream_handle(coded, workflow_id, topic=TOKENS) + first = await stream.producer(producer_id="p", attempt=1).append("a") + retry = stream.producer(producer_id="p", attempt=1) + # The ciphertext differs, and the digest lang took before the codec does not. + assert await retry.append("a") == first + owner = coded.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + assert [r.value for r in await read_all(stream.read())] == ["a"] async def test_an_activity_with_no_workflow_has_no_workflow_stream(): @@ -233,8 +335,7 @@ async def run(self, runs_left: int) -> list[str]: async def test_an_activity_producer_writes_in_every_run_of_a_chain(client: Client): - provider = MemoryStreams() - streams_client = client_with(client, provider) + streams_client = await connect_with(client, MemoryStreams()) workflow_id = new_workflow_id() async with new_worker( streams_client, CountThenContinue, activities=[count_twice] diff --git a/tests/contrib/streams/test_workflow_writer.py b/tests/contrib/streams/test_workflow_writer.py index 21ddcb09e6..f27ec1440e 100644 --- a/tests/contrib/streams/test_workflow_writer.py +++ b/tests/contrib/streams/test_workflow_writer.py @@ -12,66 +12,27 @@ import pytest from temporalio import workflow -from temporalio.bridge.proto.streams import ( - LatestRequest, - ReadRequest, - StreamAddress, - StreamOwnerKind, -) from temporalio.client import ( Client, WorkflowQueryFailedError, WorkflowUpdateFailedError, ) from temporalio.contrib.streams import ( + BEGINNING, DEFAULT_TOPIC, RecordKind, - StreamRecord, + get_stream_handle, topic, workflow_writer, ) -from temporalio.contrib.streams._plugin import call, store_for_client -from temporalio.contrib.streams._wire import from_read from temporalio.contrib.streams.memory import MemoryStreams from temporalio.exceptions import ApplicationError -from tests.contrib.streams._support import connect_with, new_workflow_id +from tests.contrib.streams._support import connect_with, new_workflow_id, read_all from tests.helpers import new_worker EVENTS = topic("events", dict) -def address(client: Client, workflow_id: str, topic: str) -> StreamAddress: - return StreamAddress( - namespace=client.namespace, - owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, - workflow_id=workflow_id, - topic=topic, - ) - - -async def read_records( - client: Client, workflow_id: str, topic: str, result_type: type | None = None -) -> list[StreamRecord]: - """Read through Core's stream service until the owner's chain has ended.""" - service = await store_for_client(client)._service_for(client) - request = ReadRequest(stream=address(client, workflow_id, topic)) - request.wait.FromSeconds(5) - records = [] - while True: - response = await call(service.read(request)) - for record in response.records: - records.append(await from_read(client.data_converter, record, result_type)) - if response.done: - return records - request.after, request.state = response.cursor, response.state - - -async def latest(client: Client, workflow_id: str, topic: str) -> str: - service = await store_for_client(client)._service_for(client) - request = LatestRequest(stream=address(client, workflow_id, topic)) - return (await call(service.latest(request))).cursor - - @workflow.defn class Publisher: def __init__(self) -> None: @@ -101,7 +62,8 @@ async def test_a_workflow_publishes_and_a_client_reads(client: Client): await streams_client.execute_workflow( Publisher.run, 3, id=workflow_id, task_queue=worker.task_queue ) - records = await read_records(streams_client, workflow_id, EVENTS.name, dict) + stream = get_stream_handle(streams_client, workflow_id) + records = await read_all(stream.read(topic=EVENTS)) assert [r.kind for r in records] == [RecordKind.DATA] * 4 + [RecordKind.FINISH] assert [r.value for r in records[:4]] == [ {"step": "init"}, @@ -112,7 +74,7 @@ async def test_a_workflow_publishes_and_a_client_reads(client: Client): # The owning Workflow writes with no producer identity: its task is the # boundary, not a producer attempt. assert {(r.producer_id, r.attempt, r.sequence) for r in records} == {("", 0, 0)} - (default,) = await read_records(streams_client, workflow_id, DEFAULT_TOPIC, str) + (default,) = await read_all(stream.read(result_type=str)) assert default.value == "on the default topic" assert default.topic == DEFAULT_TOPIC @@ -126,7 +88,8 @@ async def test_an_evicted_workflow_does_not_publish_twice(client: Client): await streams_client.execute_workflow( Publisher.run, 3, id=workflow_id, task_queue=worker.task_queue ) - assert len(await read_records(streams_client, workflow_id, EVENTS.name)) == 5 + stream = get_stream_handle(streams_client, workflow_id) + assert len(await read_all(stream.read(topic=EVENTS))) == 5 @workflow.defn @@ -176,7 +139,8 @@ async def test_a_query_or_validator_cannot_publish(client: Client): with pytest.raises(WorkflowUpdateFailedError) as failed: await handle.execute_update(ReadOnlyPublisher.poke) assert "publish to a stream" in str(failed.value.cause) - assert await latest(streams_client, workflow_id, EVENTS.name) == "" + stream = get_stream_handle(streams_client, workflow_id) + assert await stream.latest(topic=EVENTS) == BEGINNING await handle.terminate() From 342f31aefbb96f9a604dc348edaddc289b23fdb3 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 22:41:42 -0700 Subject: [PATCH 5/7] Said when a stream read starts and how not to miss the next record. Core checks the cursor and resolves END when the generator first runs, so a reader that must catch the next record positions itself with latest() first. --- temporalio/contrib/streams/_handles.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 46111feccd..58a769d1ce 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -258,6 +258,16 @@ def read( result is a generator, so a caller that stops early calls ``aclose()`` on it. + The read starts when the generator first runs, not when this call + returns, and ``END`` is resolved then. A reader that must not miss + the next record, such as one about to send something the Workflow + answers on the stream, takes ``latest()`` first and reads after it:: + + start = await handle.latest() + await client.execute_update(...) + async for record in handle.read(after=start): + ... + Raises: ValueError: ``result_type`` was passed with a topic definition, or the topic is empty. Raised by this call. From 9f22bcaec0830e69362996b549bfc32d46df269f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:39:40 -0700 Subject: [PATCH 6/7] Used the Workflow's codec context and one producer per topic. The Workflow encodes its records with its own serialization context, so readers and producers without it couldn't decode them with a context-aware codec. An Activity's second producer object could also dedupe a record away, and a Local Activity's attempt isn't durable, so it gets no producer. --- temporalio/contrib/streams/_handles.py | 88 +++++++++++-- tests/contrib/streams/test_outside_handles.py | 121 +++++++++++++++++- 2 files changed, 194 insertions(+), 15 deletions(-) diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 58a769d1ce..d3239ec236 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -19,6 +19,7 @@ from __future__ import annotations import asyncio +import weakref from collections.abc import AsyncGenerator from datetime import timedelta from typing import Any, Generic, TypeVar, cast, overload @@ -37,6 +38,7 @@ from temporalio.client import Client from temporalio.contrib.streams._body import batch_digest, content_hash, encode_bodies from temporalio.contrib.streams._cursor import BEGINNING +from temporalio.contrib.streams._errors import StreamUnsupportedError from temporalio.contrib.streams._plugin import ( StreamStorePlugin, call, @@ -46,6 +48,7 @@ from temporalio.contrib.streams._ref import StreamRef from temporalio.contrib.streams._topic import StreamTopic, resolve_topic from temporalio.contrib.streams._wire import WireRecord, from_read, to_wire +from temporalio.converter import DataConverter, WorkflowSerializationContext __all__ = [ "ActivityStreamHandle", @@ -143,7 +146,7 @@ async def finish(self) -> Cursor: return await self._write(RecordKind.FINISH, [None]) async def _write(self, kind: RecordKind, values: Any) -> Cursor: - converter = self._handle._client.data_converter + converter = self._handle._converter wires = [ to_wire( converter.payload_converter, @@ -208,6 +211,13 @@ def __init__( self._plugin = plugin self._client = client self._ref = ref + # The stream is the Workflow's, and the Workflow encodes its records with + # its own context, so every reader and producer on it uses that context. + self._converter: DataConverter = client.data_converter.with_context( + WorkflowSerializationContext( + namespace=client.namespace, workflow_id=ref.workflow_id + ) + ) @property def ref(self) -> StreamRef: @@ -277,6 +287,13 @@ def read( retains, or the read fell behind retention while it ran. StreamRecordError: A record did not decode. Read on from its cursor to skip it. + StreamNotFoundError: The owner doesn't exist. + StreamStorageError: The store failed or couldn't be reached, as + a Redis blip during a long read does. + + Any of these ends the generator. Start a new read after the last + cursor you handled. It reads on from there, though it no longer knows + which earlier attempts of other producers it delivered. """ name, result_type = self._resolve(topic, result_type) return self._read(name, after, result_type) @@ -290,7 +307,7 @@ async def _read( while True: response = await call(service.read(request)) for record in response.records: - yield await from_read(self._client.data_converter, record, result_type) + yield await from_read(self._converter, record, result_type) if response.done: return request.after = response.cursor @@ -330,8 +347,11 @@ def producer( A process that restarts its work raises ``attempt``, which starts a new sequence and tells readers that what the earlier attempt wrote is superseded. Keep one producer object per ``(topic, producer_id, - attempt)``. Each object numbers its own records from one, so a second - object for the same session is refused as stale. + attempt)``. Each object numbers its own records from one. A second + object for the same session is refused as stale once the first moved + past its first batch. Before that, a second object's first batch with + the same content as the first object's is taken as a retry, and the + store writes nothing, with no error. Raises: ValueError: ``producer_id`` is empty or ``attempt`` is below one. @@ -367,10 +387,16 @@ class ActivityStreamHandle: producer. """ - def __init__(self, inner: StreamHandle, identity: ActivityProducer) -> None: + def __init__( + self, + inner: StreamHandle, + identity: ActivityProducer, + producers: dict[str, StreamProducer[Any]], + ) -> None: """Prefer :func:`temporalio.contrib.streams.activity_handle`.""" self._inner = inner self._identity = identity + self._producers = producers @property def ref(self) -> StreamRef: @@ -431,16 +457,23 @@ def producer( attempt is the Activity's Temporal attempt, so the first record of a retry makes readers see ``SUPERSEDED``. A retry starts a new sequence, and the store deduplicates a repeated append within one attempt. + + Every call in one attempt returns the same producer for a topic, so + two parts of the Activity that each ask for it number their records + in one sequence. """ name, _ = self._inner._resolve(topic, None) - identity = self._identity - return StreamProducer( - self._inner, - name, - f"{identity.activity_id}@{identity.run_id}", - identity.attempt, - activity=identity, - ) + producer = self._producers.get(name) + if producer is None: + identity = self._identity + producer = self._producers[name] = StreamProducer( + self._inner, + name, + f"{identity.activity_id}@{identity.run_id}", + identity.attempt, + activity=identity, + ) + return producer def activity_handle() -> ActivityStreamHandle: @@ -451,12 +484,23 @@ def activity_handle() -> ActivityStreamHandle: :func:`get_stream_handle` with ``temporalio.activity.client()`` and a producer id of your own. + A Local Activity has no stream producer in this release. Its attempt is + not durable, so after a Worker crash it runs again as the same attempt, + and the store refuses its first append as stale. Publish from the + Workflow instead, which commits with the Workflow Task. + Raises: ValueError: The Activity was not scheduled by a Workflow, or no stream store is registered on the Worker's client. + StreamUnsupportedError: The Activity is a Local Activity. RuntimeError: Not called from inside an Activity. """ info = temporalio.activity.info() + if info.is_local: + raise StreamUnsupportedError( + "a Local Activity has no stream producer in this release, since its " + "attempt isn't durable; publish from the Workflow instead" + ) if info.workflow_id is None or info.workflow_run_id is None: raise ValueError( "this Activity was not scheduled by a Workflow, so it has no Workflow " @@ -471,7 +515,23 @@ def activity_handle() -> ActivityStreamHandle: attempt=info.attempt, ) inner = StreamHandle(store_for_client(client), client, ref) - return ActivityStreamHandle(inner, identity) + return ActivityStreamHandle(inner, identity, _attempt_producers()) + + +# The producers of each running Activity attempt, by topic, under the id of the +# attempt's context. The context isn't hashable, so a finalizer drops the entry +# with it. +_activity_producers: dict[int, dict[str, StreamProducer[Any]]] = {} + + +def _attempt_producers() -> dict[str, StreamProducer[Any]]: + context = temporalio.activity._Context.current() + key = id(context) + producers = _activity_producers.get(key) + if producers is None: + producers = _activity_producers[key] = {} + weakref.finalize(context, _activity_producers.pop, key, None) + return producers def get_stream_handle( diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index e4358f48bc..ec330a836a 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -27,13 +27,20 @@ StreamNotFoundError, StreamProducerError, StreamRef, + StreamUnsupportedError, Supersession, activity_handle, get_stream_handle, topic, + workflow_writer, ) from temporalio.contrib.streams.memory import MemoryStreams -from temporalio.converter import DataConverter, PayloadCodec +from temporalio.converter import ( + DataConverter, + PayloadCodec, + SerializationContext, + WithSerializationContext, +) from temporalio.exceptions import ApplicationError from temporalio.testing import ActivityEnvironment from tests.contrib.streams._support import connect_with, new_workflow_id, read_all @@ -352,3 +359,115 @@ async def test_an_activity_producer_writes_in_every_run_of_a_chain(client: Clien assert producers[0] == producers[1] != producers[2] == producers[3] assert producers[3] == last_producer assert last_producer.endswith("@" + (await handle.describe()).run_id) + + +class WorkflowKeyedCodec(PayloadCodec, WithSerializationContext): + """Encrypts each payload under the key of the Workflow its context names, as + a multi-tenant codec does. A payload decodes only under the same key.""" + + def __init__(self, key: str = "") -> None: + self.key = key + + def with_context(self, context: SerializationContext) -> WorkflowKeyedCodec: + workflow_id = getattr(context, "workflow_id", None) + return WorkflowKeyedCodec(workflow_id or "") + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload( + metadata={"encoding": b"binary/keyed", "key": self.key.encode()}, + data=p.SerializeToString(), + ) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + out = [] + for p in payloads: + if p.metadata.get("encoding") != b"binary/keyed": + out.append(p) + continue + if p.metadata["key"].decode() != self.key: + raise ValueError( + f"encrypted under {p.metadata['key']!r}, not {self.key!r}" + ) + out.append(Payload.FromString(p.data)) + return out + + +@activity.defn +async def append_one_token() -> None: + await activity_handle().producer(topic=TOKENS).append("from the activity") + + +@workflow.defn +class PublishesThenRunsAnActivity: + @workflow.run + async def run(self) -> None: + workflow_writer(TOKENS).publish("from the workflow") + await workflow.execute_activity( + append_one_token, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_both_sides_use_the_workflows_serialization_context(client: Client): + # The Workflow's records are encoded with its context, so readers and the + # other producers on its stream must use the same one. + keyed = await connect_with( + client, MemoryStreams(), DataConverter(payload_codec=WorkflowKeyedCodec()) + ) + workflow_id = new_workflow_id() + async with new_worker( + keyed, PublishesThenRunsAnActivity, activities=[append_one_token] + ) as worker: + await keyed.execute_workflow( + PublishesThenRunsAnActivity.run, + id=workflow_id, + task_queue=worker.task_queue, + ) + records = await read_all(get_stream_handle(keyed, workflow_id).read(topic=TOKENS)) + assert [r.value for r in records] == ["from the workflow", "from the activity"] + + +@activity.defn +async def logs_twice() -> None: + # A helper that asks for the producer on every call. + for line in ["started", "started"]: + await activity_handle().producer(topic=TOKENS).append(line) + + +@workflow.defn +class RunsLogsTwice: + @workflow.run + async def run(self) -> None: + await workflow.execute_activity( + logs_twice, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_an_activity_gets_one_producer_per_topic(client: Client): + # Two producer objects would both start at sequence 1, so the second equal + # record would dedupe into the first and vanish. + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, RunsLogsTwice, activities=[logs_twice] + ) as worker: + await streams_client.execute_workflow( + RunsLogsTwice.run, id=workflow_id, task_queue=worker.task_queue + ) + records = await read_all( + get_stream_handle(streams_client, workflow_id).read(topic=TOKENS) + ) + assert [r.value for r in records] == ["started", "started"] + + +async def test_a_local_activity_has_no_stream_producer(): + env = ActivityEnvironment() + env.info = dataclasses.replace(ActivityEnvironment.default_info(), is_local=True) + + async def local() -> None: + activity_handle() + + with pytest.raises(StreamUnsupportedError, match="Local Activity"): + await env.run(local) From 1557fcd598427fa86ab3c196c51f3ea35cc0189d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 10:25:51 -0700 Subject: [PATCH 7/7] Added a changelog entry for Workflow streams. This layer is where a Workflow publishes and outside code reads end to end, so the release note lands here. --- changelog/added/pirouetting-platypus.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/added/pirouetting-platypus.md diff --git a/changelog/added/pirouetting-platypus.md b/changelog/added/pirouetting-platypus.md new file mode 100644 index 0000000000..c8a7476d59 --- /dev/null +++ b/changelog/added/pirouetting-platypus.md @@ -0,0 +1 @@ +Add experimental `temporalio.contrib.streams` so a Workflow can publish ordered records to a Redis stream with its Workflow Task, and Activities and clients can read them.