From 6527082bb2fa1ad3695bc6694cdae2a6f790d5bb Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:21:30 -0700 Subject: [PATCH] Refused stream owners this release does not support. --- temporalio/contrib/streams/__init__.py | 6 ++ temporalio/contrib/streams/_handles.py | 14 ++-- temporalio/contrib/streams/_ref.py | 32 ++++++-- temporalio/contrib/streams/_workflow.py | 30 ++++++++ tests/contrib/streams/test_outside_handles.py | 13 ---- tests/contrib/streams/test_topics_refs.py | 19 +++-- .../streams/test_unsupported_owners.py | 73 +++++++++++++++++++ 7 files changed, 152 insertions(+), 35 deletions(-) create mode 100644 temporalio/contrib/streams/_workflow.py create mode 100644 tests/contrib/streams/test_unsupported_owners.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index c6976630bb..70d9e4d175 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -43,6 +43,10 @@ :func:`get_stream_handle`, and writes with a producer id and an attempt of its own. +In this release only a Workflow owns a stream, and only Activities and +clients read one. Reading inside a Workflow (:func:`workflow_reader`) and +the other owner kinds raise :class:`StreamUnsupportedError`. + The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` as an ordinary payload. @@ -78,6 +82,7 @@ ) from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic +from temporalio.contrib.streams._workflow import workflow_reader __all__ = [ "ActivityStreamHandle", @@ -108,4 +113,5 @@ "activity_handle", "get_stream_handle", "topic", + "workflow_reader", ] diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index e922484e30..4c43eaa69d 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -490,9 +490,10 @@ def activity_handle() -> ActivityStreamHandle: instead. 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. + StreamUnsupportedError: The Activity is a Local Activity, or it was + not scheduled by a Workflow, since streams owned by an Activity + are not supported in this release. + ValueError: No stream store is registered on the Worker's client. RuntimeError: Not called from inside an Activity. """ info = temporalio.activity.info() @@ -502,9 +503,9 @@ def activity_handle() -> ActivityStreamHandle: "attempt isn't durable; use a regular Activity 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 " - "stream" + raise StreamUnsupportedError( + "this Activity was not scheduled by a Workflow, and streams owned by " + "an Activity are not supported in this release" ) client = temporalio.activity.client() ref = StreamRef.for_workflow(info.workflow_id, run_id=info.workflow_run_id) @@ -566,4 +567,5 @@ def get_stream_handle( ref = workflow_id else: ref = StreamRef.for_workflow(workflow_id, run_id=run_id, topic=topic) + ref._require_supported() return StreamHandle(store_for_client(client), client, ref) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index bc8308398a..91f78a8c4e 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -13,6 +13,7 @@ from dataclasses import dataclass from typing import Any, Literal +from temporalio.contrib.streams._errors import StreamUnsupportedError from temporalio.contrib.streams._topic import ( DEFAULT_TOPIC, StreamTopic, @@ -25,8 +26,7 @@ """The owner kinds this release opens. Only a Workflow owns a stream. :attr:`StreamRef.kind` is a plain string, so a ref that names an owner kind -a later release adds still reaches the ref's own check when it is decoded, -instead of failing the converter's type check. +a later release adds still decodes, and is refused where a handle opens. """ @@ -53,19 +53,35 @@ class StreamRef: topic: str = DEFAULT_TOPIC def __post_init__(self) -> None: - """Refuse a ref this release cannot open. + """Refuse a ref that names no stream. + + Any ``kind`` decodes, since a ref of a later release can arrive as + Workflow input, where raising would fail the Workflow Task on every + retry. :meth:`_require_supported` refuses it where a handle opens. Raises: - ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or - ``topic`` is empty. + ValueError: ``kind`` or ``topic`` is empty, or a Workflow ref has + no ``workflow_id``. """ - if self.kind != "workflow": - raise ValueError(f"unknown StreamRef kind {self.kind!r}") - if not self.workflow_id: + if not self.kind: + raise ValueError("a StreamRef needs a kind") + if self.kind == "workflow" and not self.workflow_id: raise ValueError("a Workflow StreamRef needs a workflow_id") if not self.topic: raise ValueError("a StreamRef needs a topic name") + def _require_supported(self) -> None: + """Refuse to open a stream whose owner kind this release lacks. + + Raises: + StreamUnsupportedError: ``kind`` is not ``"workflow"``. + """ + if self.kind != "workflow": + raise StreamUnsupportedError( + f"streams owned by kind {self.kind!r} are not supported in this " + "release; only Workflow-owned streams are" + ) + @classmethod def for_workflow( cls, diff --git a/temporalio/contrib/streams/_workflow.py b/temporalio/contrib/streams/_workflow.py new file mode 100644 index 0000000000..8c371a4c54 --- /dev/null +++ b/temporalio/contrib/streams/_workflow.py @@ -0,0 +1,30 @@ +"""Streams inside Workflow code. + +Reading a stream inside a Workflow is not part of this release. +""" + +from __future__ import annotations + +from typing import Any + +from temporalio.contrib.streams._errors import StreamUnsupportedError + +__all__ = ["workflow_reader"] + + +def workflow_reader(*_args: Any, **_kwargs: Any) -> Any: + """Reading a stream inside a Workflow is not supported in this release. + + Read from an Activity or a client with + :func:`temporalio.contrib.streams.activity_handle` or + :func:`temporalio.contrib.streams.get_stream_handle`, and send the + Workflow what it needs, for example as a Signal. + + Raises: + StreamUnsupportedError: Always. + """ + raise StreamUnsupportedError( + "reading a stream inside a Workflow is not supported in this release; " + "read it from an Activity or a client and send the Workflow what it " + "needs, for example as a Signal" + ) diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index cad6b4a6e0..b466aa8ca6 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -282,19 +282,6 @@ async def test_a_retry_through_a_nonce_codec_still_dedupes(client: Client): assert [r.value for r in await read_all(stream.read())] == ["a"] -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) - - @activity.defn async def count_twice() -> str: producer = activity_handle().producer(topic=TOKENS) diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 3cbbe1e6b0..509d8cbbf1 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -9,6 +9,7 @@ DEFAULT_TOPIC, StreamRef, StreamTopic, + StreamUnsupportedError, topic, ) from temporalio.contrib.streams._topic import ( @@ -51,24 +52,26 @@ def test_a_ref_round_trips_through_the_default_converter(): def test_a_ref_refuses_what_it_cannot_name(): - with pytest.raises(ValueError, match="unknown"): - StreamRef("nexus", "x") with pytest.raises(ValueError): StreamRef("workflow", "") with pytest.raises(ValueError): StreamRef("workflow", "wf", topic="") -def test_a_ref_from_a_later_release_decodes_to_the_kind_check(): - # The kind is a plain string, so the converter builds the ref and the ref - # itself refuses the kind, rather than the converter failing a type check. +@pytest.mark.parametrize("kind", ["activity", "nexus"]) +def test_a_ref_from_a_later_release_decodes_and_fails_where_it_opens(kind: str): + # Decoding happens while a Workflow Task applies its input, where a raise + # fails the task on every retry. Opening the handle is where the caller + # can catch it. converter = DataConverter.default.payload_converter payload = Payload( metadata={"encoding": b"json/plain"}, - data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}', + data=f'{{"kind": "{kind}", "workflow_id": "wf", "topic": "out"}}'.encode(), ) - with pytest.raises(ValueError, match="'activity'"): - converter.from_payloads([payload], [StreamRef]) + ref = converter.from_payloads([payload], [StreamRef])[0] + assert ref.kind == kind + with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): + ref._require_supported() @pytest.mark.parametrize( diff --git a/tests/contrib/streams/test_unsupported_owners.py b/tests/contrib/streams/test_unsupported_owners.py new file mode 100644 index 0000000000..498957f430 --- /dev/null +++ b/tests/contrib/streams/test_unsupported_owners.py @@ -0,0 +1,73 @@ +"""Owners and readers this release does not support say so. + +Only a Workflow owns a stream in this release, and only Activities and +clients read one. Everything else raises ``StreamUnsupportedError``. +""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from temporalio import workflow +from temporalio.client import Client +from temporalio.contrib.streams import ( + StreamRef, + StreamUnsupportedError, + activity_handle, + get_stream_handle, + topic, + workflow_reader, +) +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.testing import ActivityEnvironment +from tests.contrib.streams._support import connect_with, new_workflow_id +from tests.helpers import new_worker + +EVENTS = topic("events", dict) + + +async def test_a_ref_refuses_owner_kinds_this_release_lacks_where_it_opens( + client: Client, +): + streams_client = await connect_with(client, MemoryStreams()) + for kind in ("activity", "standalone"): + ref = StreamRef(kind, "x") # type: ignore[arg-type] + with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): + get_stream_handle(streams_client, ref) + + +@workflow.defn +class TriesToRead: + @workflow.run + async def run(self) -> str: + try: + workflow_reader(EVENTS) + except StreamUnsupportedError as error: + return str(error) + return "read" + + +async def test_a_workflow_cannot_read_a_stream(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, TriesToRead) as worker: + said = await streams_client.execute_workflow( + TriesToRead.run, + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + assert "not supported in this release" in said + + +async def test_an_activity_with_no_workflow_has_no_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(StreamUnsupportedError, match="owned by an Activity"): + await env.run(standalone)