From 7a4fe16a3c70fa62beaff68afd3e86c195748a83 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:20:02 -0700 Subject: [PATCH 1/4] Added stream topics and references. --- temporalio/contrib/streams/__init__.py | 10 ++ temporalio/contrib/streams/_ref.py | 91 ++++++++++++++++ temporalio/contrib/streams/_topic.py | 108 +++++++++++++++++++ tests/contrib/streams/test_public_surface.py | 1 + tests/contrib/streams/test_topics_refs.py | 88 +++++++++++++++ 5 files changed, 298 insertions(+) create mode 100644 temporalio/contrib/streams/_ref.py create mode 100644 temporalio/contrib/streams/_topic.py create mode 100644 tests/contrib/streams/test_topics_refs.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index d61e0ea763..4363d17636 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -19,6 +19,9 @@ refused with :class:`StreamCursorError`, and one whose record retention dropped with :class:`StreamExpiredError`. A read with no cursor starts at :data:`BEGINNING` or :data:`END`. +3. **Topics are defined once.** :func:`topic` defines a topic with the type + its records decode to, and every party shares that definition. A call + that names no topic addresses :data:`DEFAULT_TOPIC`. The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` @@ -45,10 +48,13 @@ StreamRecord, Supersession, ) +from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef +from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic __all__ = [ "BEGINNING", "Cursor", + "DEFAULT_TOPIC", "END", "RecordKind", "StreamClosedError", @@ -57,11 +63,15 @@ "StreamExpiredError", "StreamNotFoundError", "StreamOutcomeUnknownError", + "StreamOwnerKind", "StreamProducerError", "StreamRecord", "StreamRecordError", + "StreamRef", "StreamRefusedError", "StreamStorageError", + "StreamTopic", "StreamUnsupportedError", "Supersession", + "topic", ] diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py new file mode 100644 index 0000000000..bc8308398a --- /dev/null +++ b/temporalio/contrib/streams/_ref.py @@ -0,0 +1,91 @@ +"""A reference to one stream that crosses a process boundary as data. + +A handle is bound to a client and its store, so it cannot be a Workflow +argument or an Activity result. A :class:`StreamRef` can: it names the owner +and the topic, nothing more, and whoever receives it opens the stream on the +store its own client carries. It carries no cursor, because a position +belongs to a reader, and no store name. +""" + +from __future__ import annotations + +import dataclasses +from dataclasses import dataclass +from typing import Any, Literal + +from temporalio.contrib.streams._topic import ( + DEFAULT_TOPIC, + StreamTopic, + resolve_topic, +) + +__all__ = ["StreamOwnerKind", "StreamRef"] + +StreamOwnerKind = Literal["workflow"] +"""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. +""" + + +@dataclass(frozen=True) +class StreamRef: + """One stream, named by its owner and its topic. + + ``kind`` says what owns the stream; only ``"workflow"`` exists in this + release. A Workflow ref carries ``workflow_id`` and, when pinned to one + run, ``run_id``. The stream belongs to the run chain either way, and a + read returns the chain's records. ``run_id`` changes two things: which + chain the handle looks up, and that a read ends when that run closes, + Continue-as-New included. Without it a reader keeps reading across + Continue-as-New. ``topic`` is the topic a handle opened from the ref + addresses when a call names none. + + The default data converter carries it as JSON, so it can be a Workflow + argument or an Activity result. + """ + + kind: str + workflow_id: str + run_id: str | None = None + topic: str = DEFAULT_TOPIC + + def __post_init__(self) -> None: + """Refuse a ref this release cannot open. + + Raises: + ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or + ``topic`` is empty. + """ + if self.kind != "workflow": + raise ValueError(f"unknown StreamRef kind {self.kind!r}") + if 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") + + @classmethod + def for_workflow( + cls, + workflow_id: str, + *, + run_id: str | None = None, + topic: str | StreamTopic[Any] | None = None, + ) -> StreamRef: + """A ref to ``topic`` of ``workflow_id``'s stream. + + Without ``topic`` it names + :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + """ + name, _ = resolve_topic(topic) + return cls("workflow", workflow_id, run_id=run_id, topic=name) + + def with_topic(self, topic: str | StreamTopic[Any] | None) -> StreamRef: + """The same owner, naming ``topic`` instead. + + ``None`` names :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + """ + name, _ = resolve_topic(topic) + return dataclasses.replace(self, topic=name) diff --git a/temporalio/contrib/streams/_topic.py b/temporalio/contrib/streams/_topic.py new file mode 100644 index 0000000000..0f2ec562a5 --- /dev/null +++ b/temporalio/contrib/streams/_topic.py @@ -0,0 +1,108 @@ +"""Typed topic definitions. + +A topic is defined once, at module level, with the type its records decode +to, and the Workflow, its Activities and the client code all refer to that +one definition, the way they share a Signal or an Activity. A plain string +names a topic too, for a name decided at runtime; then the decode hint +travels as ``result_type=`` on each read. + +On the wire a topic is a string, and +:attr:`temporalio.contrib.streams.StreamRecord.topic` is that string. +""" + +from __future__ import annotations + +import unicodedata +from dataclasses import dataclass +from typing import Any, Generic, TypeVar, overload + +__all__ = ["DEFAULT_TOPIC", "StreamTopic", "resolve_topic", "topic"] + +T = TypeVar("T") + +DEFAULT_TOPIC = "output" +"""The topic a call addresses when it names none. + +It is an ordinary name, so naming it explicitly addresses the same topic. +""" + + +@dataclass(frozen=True) +class StreamTopic(Generic[T]): + """A topic of a stream, with the type its records decode to. + + Made with :func:`topic`. Pass it wherever a call takes a topic, and the + record and value types follow from it. + """ + + name: str + result_type: type[T] | None = None + """The type records decode to, or ``None`` for the converter's default.""" + + +@overload +def topic(name: str, result_type: type[T]) -> StreamTopic[T]: ... + + +@overload +def topic(name: str, result_type: None = None) -> StreamTopic[Any]: ... + + +def topic(name: str, result_type: type | None = None) -> StreamTopic[Any]: + """Define a topic. + + Args: + name: The topic's name, as it appears on every record. + result_type: The type records on this topic decode to. Without one, + the payload converter's default applies. + + Raises: + ValueError: ``name`` is empty, longer than 256 UTF-8 bytes, or holds + a control character. + """ + _check_name(name) + return StreamTopic(name, result_type) + + +def resolve_topic( + topic: str | StreamTopic[Any] | None = None, result_type: type | None = None +) -> tuple[str, type | None]: + """The name and decode hint a call means, from either form of topic. + + ``None`` means :data:`DEFAULT_TOPIC`. A handle calls this once at the + top of each call that takes a topic, so a definition and a string reach + the store the same way. + + Raises: + ValueError: A definition was given together with ``result_type``, + which would name two types for one topic, or the name is empty, + longer than 256 UTF-8 bytes, or holds a control character. + """ + if isinstance(topic, StreamTopic): + if result_type is not None: + raise ValueError( + f"topic {topic.name!r} already carries its type; do not pass " + "result_type= with a definition" + ) + name, result_type = topic.name, topic.result_type + elif topic is None: + name = DEFAULT_TOPIC + else: + name = topic + _check_name(name) + return name, result_type + + +_MAX_NAME_BYTES = 256 + + +def _check_name(name: str) -> None: + # The same limit in every SDK: the name goes into store keys, cursor + # hashes and Core's manifest budget, and a store may join names with a + # control character. + if not name: + raise ValueError("topic name must not be empty") + if len(name.encode()) > _MAX_NAME_BYTES: + raise ValueError(f"topic name must be at most {_MAX_NAME_BYTES} UTF-8 bytes") + if any(unicodedata.category(c) == "Cc" for c in name): + raise ValueError(f"topic name {name!r} must not hold a control character") diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py index a3647d4eba..1161287e40 100644 --- a/tests/contrib/streams/test_public_surface.py +++ b/tests/contrib/streams/test_public_surface.py @@ -12,6 +12,7 @@ "decode_body", "encode_bodies", "error_from_failure", + "resolve_topic", } diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py new file mode 100644 index 0000000000..3cbbe1e6b0 --- /dev/null +++ b/tests/contrib/streams/test_topics_refs.py @@ -0,0 +1,88 @@ +"""Typed topics and the stream reference that crosses process boundaries.""" + +from __future__ import annotations + +import pytest + +from temporalio.api.common.v1 import Payload +from temporalio.contrib.streams import ( + DEFAULT_TOPIC, + StreamRef, + StreamTopic, + topic, +) +from temporalio.contrib.streams._topic import ( + resolve_topic, +) +from temporalio.converter import DataConverter + +OUT = topic("out", dict) + + +def test_a_definition_carries_its_name_and_type(): + assert OUT == StreamTopic("out", dict) + assert resolve_topic(OUT) == ("out", dict) + assert resolve_topic("free", int) == ("free", int) + assert resolve_topic(None) == (DEFAULT_TOPIC, None) + + +def test_topic_mistakes_are_value_errors(): + with pytest.raises(ValueError): + topic("") + with pytest.raises(ValueError): + resolve_topic("") + with pytest.raises(ValueError, match="already carries its type"): + resolve_topic(OUT, dict) + + +def test_a_ref_names_a_workflow_stream(): + ref = StreamRef.for_workflow("wf", topic=OUT) + assert ref == StreamRef("workflow", "wf", topic="out") + assert StreamRef.for_workflow("wf").topic == DEFAULT_TOPIC + assert ref.with_topic(None).topic == DEFAULT_TOPIC + assert ref.with_topic("other").workflow_id == "wf" + + +def test_a_ref_round_trips_through_the_default_converter(): + ref = StreamRef.for_workflow("wf", run_id="run", topic=OUT) + converter = DataConverter.default.payload_converter + payload = converter.to_payloads([ref])[0] + assert converter.from_payloads([payload], [StreamRef])[0] == ref + + +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. + converter = DataConverter.default.payload_converter + payload = Payload( + metadata={"encoding": b"json/plain"}, + data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}', + ) + with pytest.raises(ValueError, match="'activity'"): + converter.from_payloads([payload], [StreamRef]) + + +@pytest.mark.parametrize( + "name", + ["x" * 257, "é" * 129, "a\x1fb", "line\nbreak", "\x00"], +) +def test_a_topic_name_that_is_too_long_or_has_a_control_character_is_refused(name: str): + # Every store puts the name in keys, cursors and Core's manifest budget. + with pytest.raises(ValueError): + topic(name) + with pytest.raises(ValueError): + resolve_topic(name) + + +def test_a_topic_name_of_256_bytes_is_accepted(): + assert topic("x" * 256).name == "x" * 256 + assert resolve_topic("é" * 128)[0] == "é" * 128 From 868348d18e0797e7b88c8bb361d7ee0de4716aae Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:20:13 -0700 Subject: [PATCH 2/4] Added the Redis and memory stream store plugins. --- temporalio/contrib/streams/__init__.py | 19 +- temporalio/contrib/streams/_plugin.py | 201 +++++++++++++++++++++ temporalio/contrib/streams/memory.py | 24 +++ temporalio/contrib/streams/redis.py | 79 ++++++++ tests/contrib/streams/_support.py | 48 +++++ tests/contrib/streams/test_store_plugin.py | 198 ++++++++++++++++++++ 6 files changed, 566 insertions(+), 3 deletions(-) create mode 100644 temporalio/contrib/streams/_plugin.py create mode 100644 temporalio/contrib/streams/memory.py create mode 100644 temporalio/contrib/streams/redis.py create mode 100644 tests/contrib/streams/_support.py create mode 100644 tests/contrib/streams/test_store_plugin.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 4363d17636..d672e8a52e 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -10,19 +10,30 @@ The contract: -1. **A new attempt supersedes the old one.** When a reader sees the first +1. **Producers write with an identity.** A producer has an id, an attempt + and a sequence that starts at one for each attempt. A retry of the + newest batch with the same content is written once and returns the + original position. A retry with different content, or with a lower + sequence, is refused with :class:`StreamProducerError`. Content is + compared over the payloads the converter produced, before the codec, so + a codec with a fresh nonce per call still deduplicates a retry. +2. **A new attempt supersedes the old one.** When a reader sees the first record of a producer's newer attempt, it yields a :attr:`RecordKind.SUPERSEDED` record first. No store holds that record, so every store reports a retry the same way. -2. **A cursor belongs to one stream.** Hand it back to resume strictly after +3. **A cursor belongs to one stream.** Hand it back to resume strictly after the record it names. A cursor from another store or another stream is refused with :class:`StreamCursorError`, and one whose record retention dropped with :class:`StreamExpiredError`. A read with no cursor starts at :data:`BEGINNING` or :data:`END`. -3. **Topics are defined once.** :func:`topic` defines a topic with the type +4. **Topics are defined once.** :func:`topic` defines a topic with the type its records decode to, and every party shares that definition. A call that names no topic addresses :data:`DEFAULT_TOPIC`. +A store is configured as a plugin: +``Client.connect(..., plugins=[RedisStreams(...)])`` registers it on the +client and on every Worker built from that client. + 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. @@ -42,6 +53,7 @@ StreamStorageError, StreamUnsupportedError, ) +from temporalio.contrib.streams._plugin import StreamStorePlugin from temporalio.contrib.streams._record import ( Cursor, RecordKind, @@ -70,6 +82,7 @@ "StreamRef", "StreamRefusedError", "StreamStorageError", + "StreamStorePlugin", "StreamTopic", "StreamUnsupportedError", "Supersession", diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py new file mode 100644 index 0000000000..737f0de1c4 --- /dev/null +++ b/temporalio/contrib/streams/_plugin.py @@ -0,0 +1,201 @@ +"""The plugin a store configuration is, and how each context finds the store. + +A configuration class is registered once, on the client: +``Client.connect(plugins=[RedisStreams(...)])``. Connecting the client asks +Core for the store it names. Core serves that one store to the client's +handles and to the Activities of every Worker built from the client, so an +Activity's append and an outside read meet in the same place. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from typing import TYPE_CHECKING, Any, TypeVar, cast + +import temporalio.bridge.client +from temporalio.bridge.proto.streams import ( + DeleteOwnerRequest, + StreamOwnerKind, + StreamStoreConfig, +) +from temporalio.bridge.streams import StreamCallFailure, StreamStore +from temporalio.bridge.streams_generated import StreamService +from temporalio.contrib.streams._errors import error_from_failure +from temporalio.plugin import SimplePlugin + +if TYPE_CHECKING: + from temporalio.client import Client, ClientConfig + from temporalio.service import ConnectConfig, ServiceClient + from temporalio.worker import WorkerConfig + +__all__ = ["StreamStorePlugin"] + +T = TypeVar("T") + + +class StreamStorePlugin(SimplePlugin): + """The base class of the store configuration classes. + + Use :class:`temporalio.contrib.streams.redis.RedisStreams` or + :class:`temporalio.contrib.streams.memory.MemoryStreams`. One instance + is one store, on one client connection, since Core asks about each + stream's owner over the connection that connected the store. A second + connection, such as another namespace's endpoint, needs its own instance. + Registering a second, different store on the same client raises + ``ValueError``, because a handle could otherwise pick either. + """ + + def __init__(self, name: str, config: StreamStoreConfig) -> None: + """Name the plugin and the store Core connects to.""" + super().__init__(name) + self._config = config + self._service: StreamService | None = None + self._store: StreamStore | None = None + self._connecting: asyncio.Lock | None = None + self._service_client: ServiceClient | None = None + # Whether Client.connect connected the store, which a Worker needs. + self._connected_by_connect = False + + def configure_client(self, config: ClientConfig) -> ClientConfig: + """Register this store on the client. + + Raises: + ValueError: Another store is already registered on the client. + """ + for plugin in config.get("plugins", []): + if isinstance(plugin, StreamStorePlugin) and plugin is not self: + raise ValueError(_two_stores(plugin, self)) + return super().configure_client(config) + + async def connect_service_client( + self, + config: ConnectConfig, + next: Callable[[ConnectConfig], Awaitable[ServiceClient]], + ) -> ServiceClient: + """Connect the client, then the store, so Workers built from it have one. + + A lazy client connects here too, since Core needs its connection to + ask about a stream's owner. + + Raises: + temporalio.contrib.streams.StreamError: Core could not reach the + store. + """ + service_client = await next(config) + await self._connected(service_client) + self._connected_by_connect = True + return service_client + + def configure_worker(self, config: WorkerConfig) -> WorkerConfig: + """Check that the Worker's client carries this store. + + Raises: + ValueError: The plugin is not registered on the Worker's client, + or that client was not made with ``Client.connect``. A + client made otherwise is refused even after one of its handles + connected the store. + """ + config = super().configure_worker(config) + client = config["client"] # type:ignore[reportTypedDictNotRequiredAccess] + if ( + self not in client.config()["plugins"] + or not self._connected_by_connect + or client.service_client is not self._service_client + ): + raise ValueError( + f"stream store {self.name()!r} must be registered on the Worker's " + "client, as Client.connect(..., plugins=[store])" + ) + return config + + async def delete_workflow_streams(self, namespace: str, workflow_id: str) -> int: + """Delete every stream of ``workflow_id`` in ``namespace``. + + An admin helper for cleanup and compliance requests. Deleting a + Workflow does not delete its streams. Their retention bounds what is + left, and this removes it at once, across every run chain of the + Workflow id. It asks Temporal nothing, so it can run after the + Workflow is gone. On Redis it walks the keyspace, so run it rarely, + not on a hot path. + + Delete only after the chain closed and its producers stopped. A late + append writes a fresh stream with no dedupe state and no close flag. + + Returns: + How many store keys were deleted. + + Raises: + ValueError: No client connected this store yet. + temporalio.contrib.streams.StreamStorageError: The store failed; + some keys may be deleted. + """ + if self._service is None: + raise ValueError(f"no client has connected stream store {self.name()!r}") + request = DeleteOwnerRequest( + namespace=namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=workflow_id, + ) + return (await call(self._service.delete_owner(request))).deleted + + async def _service_for(self, client: Client) -> StreamService: + """The stream service, connecting the store if no client did yet. + + Raises: + ValueError: Another client connection already connected the store. + """ + await self._connected(client.service_client) + assert self._service is not None + return self._service + + async def _connected(self, service_client: ServiceClient) -> None: + if self._service_client is service_client: + return + if self._connecting is None: + self._connecting = asyncio.Lock() + async with self._connecting: + if self._service_client is service_client: + return + if self._service_client is not None: + raise ValueError( + f"stream store {self.name()!r} serves one client connection, " + "and another one already connected it; give each connection " + "a store instance of its own" + ) + bridge: temporalio.bridge.client.Client = await cast( + Any, service_client + )._connected_client() + self._store = await call(StreamStore.connect(bridge, self._config)) + self._service = StreamService(self._store) + self._service_client = service_client + + +async def call(pending: Awaitable[T]) -> T: + """Await a stream call, raising Core's failure as the error a caller catches.""" + try: + return await pending + except StreamCallFailure as failure: + raise error_from_failure(failure.failure) from None + + +def _two_stores(held: StreamStorePlugin, new: StreamStorePlugin) -> str: + return ( + f"stream store {held.name()!r} is already registered, so {new.name()!r} " + "cannot be; register one store per client" + ) + + +def store_for_client(client: Client) -> StreamStorePlugin: + """The store registered on ``client``. + + Raises: + ValueError: No store is registered on the client. + """ + for plugin in client.config()["plugins"]: + if isinstance(plugin, StreamStorePlugin): + return plugin + raise ValueError( + "no stream store is registered on this client; pass one as a plugin, for " + "example Client.connect(..., plugins=[RedisStreams(...)])" + ) diff --git a/temporalio/contrib/streams/memory.py b/temporalio/contrib/streams/memory.py new file mode 100644 index 0000000000..198a518bfd --- /dev/null +++ b/temporalio/contrib/streams/memory.py @@ -0,0 +1,24 @@ +"""Streams in this process's memory, for tests and local development.""" + +from __future__ import annotations + +from temporalio.bridge.proto.streams import MemoryStoreConfig, StreamStoreConfig +from temporalio.contrib.streams._plugin import StreamStorePlugin + +__all__ = ["MemoryStreams"] + + +class MemoryStreams(StreamStorePlugin): + """A store in this process's memory. + + Core holds it, so the client's handles and every Worker built from the + client share it. Nothing survives the process, and another process sees + none of it. + """ + + def __init__(self) -> None: + """Create the store's configuration.""" + super().__init__( + "temporalio.contrib.streams.MemoryStreams", + StreamStoreConfig(memory=MemoryStoreConfig()), + ) diff --git a/temporalio/contrib/streams/redis.py b/temporalio/contrib/streams/redis.py new file mode 100644 index 0000000000..7d0be4409a --- /dev/null +++ b/temporalio/contrib/streams/redis.py @@ -0,0 +1,79 @@ +"""Streams in the application's Redis.""" + +from __future__ import annotations + +from collections.abc import Sequence +from datetime import timedelta + +from temporalio.bridge.proto.streams import RedisStoreConfig, StreamStoreConfig +from temporalio.contrib.streams._plugin import StreamStorePlugin + +__all__ = ["RedisStreams"] + + +class RedisStreams(StreamStorePlugin): + """A store in a Redis or a Redis Cluster that Core connects to. + + One instance per Redis deployment. Pass it to the client as a plugin, + and open handles from the client with + :func:`temporalio.contrib.streams.get_stream_handle`. Redis 7.0 or later + is required, with ``maxmemory-policy noeviction``. + """ + + def __init__( + self, + url: str | Sequence[str], + *, + cluster: bool = False, + key_prefix: str = "temporal-streams", + retention: timedelta = timedelta(days=7), + blocking_reads_per_node: int | None = None, + response_timeout: timedelta | None = None, + ) -> None: + """Create the store's configuration. + + Args: + url: A ``redis://`` or ``rediss://`` URL, or the seed nodes of a + cluster. Credentials and TLS settings come from the URL. In + this release TLS verifies the server against the system's + root certificates only, with no custom CA and no client + certificate. + cluster: Whether ``url`` names a Redis Cluster. Every key of one + Workflow's streams shares a hash tag, so a cluster serves + them from one slot. + key_prefix: Prepended to every key, so streams can share a Redis + with other data and an ACL can scope them. + retention: How long a stream keeps a record, and how long after + its last write the stream itself lives. + blocking_reads_per_node: How many reads may wait on one node at + once, each on a connection of its own. Core's default when + unset. + response_timeout: How long a short call waits for its reply. + Core's default when unset. + + Raises: + ValueError: No URL was given, ``key_prefix`` is empty, or + ``retention`` is shorter than a millisecond. + """ + urls = [url] if isinstance(url, str) else list(url) + if not urls: + raise ValueError("RedisStreams needs a URL") + if not key_prefix: + raise ValueError("key_prefix must not be empty") + if retention < timedelta(milliseconds=1): + raise ValueError( + f"retention must be at least a millisecond, got {retention}" + ) + config = RedisStoreConfig( + urls=urls, + cluster=cluster, + key_prefix=key_prefix, + blocking_reads_per_node=blocking_reads_per_node or 0, + ) + config.retention.FromTimedelta(retention) + if response_timeout is not None: + config.response_timeout.FromTimedelta(response_timeout) + super().__init__( + "temporalio.contrib.streams.RedisStreams", + StreamStoreConfig(redis=config), + ) diff --git a/tests/contrib/streams/_support.py b/tests/contrib/streams/_support.py new file mode 100644 index 0000000000..de4da231a5 --- /dev/null +++ b/tests/contrib/streams/_support.py @@ -0,0 +1,48 @@ +"""What the stream tests share.""" + +from __future__ import annotations + +import asyncio +import uuid +from typing import Any + +from temporalio.client import Client +from temporalio.contrib.streams import StreamStorePlugin +from temporalio.converter import DataConverter + + +async def connect_with( + client: Client, + store: StreamStorePlugin | None, + converter: DataConverter | None = None, +) -> Client: + """A client of the test server that carries ``store``. + + It connects, since connecting is what hands the store to Workers built + from it. + """ + return await Client.connect( + client.service_client.config.target_host, + namespace=client.namespace, + data_converter=converter or client.data_converter, + plugins=[store] if store is not None else [], + ) + + +def new_workflow_id() -> str: + return f"streams-{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 diff --git a/tests/contrib/streams/test_store_plugin.py b/tests/contrib/streams/test_store_plugin.py new file mode 100644 index 0000000000..039527c1c3 --- /dev/null +++ b/tests/contrib/streams/test_store_plugin.py @@ -0,0 +1,198 @@ +"""A store configuration registers as a client plugin, and Core connects to it.""" + +from __future__ import annotations + +import hashlib +import uuid +from datetime import timedelta + +import pytest + +from temporalio import activity +from temporalio.api.common.v1 import Payload +from temporalio.bridge.proto.streams import ( + AppendRecord, + AppendRequest, + LatestRequest, + NamedProducer, + StreamAddress, + StreamOwnerKind, +) +from temporalio.bridge.proto.streams.v1 import StreamRecordKind +from temporalio.client import Client +from temporalio.contrib.streams import StreamError, StreamNotFoundError +from temporalio.contrib.streams._plugin import call +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.contrib.streams.redis import RedisStreams +from temporalio.worker import Worker +from tests.contrib.streams._support import connect_with + + +async def test_one_store_per_client(client: Client): + first, second = MemoryStreams(), MemoryStreams() + config = client.config() + config["plugins"] = [first, second] + with pytest.raises(ValueError, match="is already registered"): + Client(**config) + # The same instance twice is one store, not two. + config["plugins"] = [first, first] + Client(**config) + + +async def test_a_worker_takes_the_store_of_its_client(client: Client): + store = MemoryStreams() + streams_client = await connect_with(client, store) + Worker(streams_client, task_queue=str(uuid.uuid4()), activities=[_noop]) + # The store on the client and the Worker configures the Worker twice. + with pytest.warns(UserWarning, match="same plugin type"): + Worker( + streams_client, + task_queue=str(uuid.uuid4()), + activities=[_noop], + plugins=[store], + ) + + +async def test_a_store_on_the_worker_alone_is_refused(client: Client): + with pytest.raises(ValueError, match="registered on the Worker's client"): + Worker(client, task_queue="tq", activities=[_noop], plugins=[MemoryStreams()]) + # A client made without connecting has not connected its store either. + config = client.config() + config["plugins"] = [MemoryStreams()] + with pytest.raises(ValueError, match="registered on the Worker's client"): + Worker(Client(**config), task_queue="tq", activities=[_noop]) + + +async def test_a_client_made_without_connecting_connects_its_store_on_use( + client: Client, +): + store = MemoryStreams() + config = client.config() + config["plugins"] = [store] + service = await store._service_for(Client(**config)) + # An owner that does not exist is Core's answer, so the call reached the store. + stream = StreamAddress( + namespace=client.namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=f"no-such-owner-{uuid.uuid4()}", + topic="out", + ) + with pytest.raises(StreamNotFoundError): + await call(service.latest(LatestRequest(stream=stream))) + + +async def test_a_store_core_cannot_reach_fails_the_connect(client: Client): + with pytest.raises(StreamError): + await connect_with( + client, + RedisStreams("redis://127.0.0.1:1", response_timeout=timedelta(seconds=1)), + ) + + +def test_redis_settings_go_into_the_store_config(): + store = RedisStreams( + ["redis://a:6379", "redis://b:6379"], + cluster=True, + key_prefix="app", + retention=timedelta(hours=2), + blocking_reads_per_node=4, + response_timeout=timedelta(seconds=3), + ) + config = store._config.redis + assert list(config.urls) == ["redis://a:6379", "redis://b:6379"] + assert config.cluster + assert config.key_prefix == "app" + assert config.retention.ToTimedelta() == timedelta(hours=2) + assert config.blocking_reads_per_node == 4 + assert config.response_timeout.ToTimedelta() == timedelta(seconds=3) + # Unset settings leave Core's defaults. + defaults = RedisStreams("redis://a:6379")._config.redis + assert defaults.blocking_reads_per_node == 0 + assert not defaults.HasField("response_timeout") + + +def test_redis_setting_mistakes_are_value_errors(): + with pytest.raises(ValueError, match="needs a URL"): + RedisStreams([]) + with pytest.raises(ValueError, match="key_prefix"): + RedisStreams("redis://a", key_prefix="") + with pytest.raises(ValueError, match="retention"): + RedisStreams("redis://a", retention=timedelta(microseconds=10)) + + +async def test_deleting_removes_what_a_workflow_stream_holds(client: Client): + store = MemoryStreams() + with pytest.raises(ValueError, match="no client has connected"): + await store.delete_workflow_streams(client.namespace, "wf") + streams_client = await connect_with(client, store) + service = await store._service_for(streams_client) + workflow_id = f"stream-owner-{uuid.uuid4()}" + await streams_client.start_workflow( + "StreamOwner", id=workflow_id, task_queue=f"unpolled-{uuid.uuid4()}" + ) + stream = address(client, workflow_id) + await call( + service.append( + AppendRequest( + stream=stream, + named=NamedProducer(producer_id="p", attempt=1), + sequence=1, + records=[ + AppendRecord( + kind=StreamRecordKind.STREAM_RECORD_KIND_DATA, + body=Payload(data=b"a"), + content_hash=hashlib.sha256(b"a").digest(), + ) + ], + digest=hashlib.sha256(b"batch").digest(), + ) + ) + ) + assert (await call(service.latest(LatestRequest(stream=stream)))).cursor + + assert await store.delete_workflow_streams(client.namespace, workflow_id) > 0 + assert (await call(service.latest(LatestRequest(stream=stream)))).cursor == "" + + +async def test_one_store_serves_one_client_connection(client: Client): + store = MemoryStreams() + first = await connect_with(client, store) + # Core asks about owners over the connection that connected the store, so a + # second connection would get the first one's answers. + with pytest.raises(ValueError, match="one client connection"): + await connect_with(client, store) + config = client.config() + config["plugins"] = [store] + with pytest.raises(ValueError, match="one client connection"): + await store._service_for(Client(**config)) + # A client copied from the connected one shares its connection. + copied = first.config() + assert await store._service_for(Client(**copied)) + + +async def test_a_worker_needs_a_client_made_with_connect_after_a_lazy_connect( + client: Client, +): + store = MemoryStreams() + config = client.config() + config["plugins"] = [store] + lazy = Client(**config) + # A handle call connects the store, but the client still wasn't connected + # through the plugin, so Workers built from it are refused all the same. + await store._service_for(lazy) + with pytest.raises(ValueError, match="registered on the Worker's client"): + Worker(lazy, task_queue=str(uuid.uuid4()), activities=[_noop]) + + +@activity.defn +async def _noop() -> None: + pass + + +def address(client: Client, workflow_id: str) -> StreamAddress: + return StreamAddress( + namespace=client.namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=workflow_id, + topic="out", + ) From 90a2facca07ebaf590ef5e8df8cf5d0082fa1d1a Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 17:51:37 -0700 Subject: [PATCH 3/4] Removed the run id from StreamRef, so a reader follows the chain. --- temporalio/contrib/streams/_ref.py | 15 +++++---------- tests/contrib/streams/test_topics_refs.py | 14 +++++++++++++- 2 files changed, 18 insertions(+), 11 deletions(-) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index bc8308398a..b63ae73f15 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -35,13 +35,10 @@ class StreamRef: """One stream, named by its owner and its topic. ``kind`` says what owns the stream; only ``"workflow"`` exists in this - release. A Workflow ref carries ``workflow_id`` and, when pinned to one - run, ``run_id``. The stream belongs to the run chain either way, and a - read returns the chain's records. ``run_id`` changes two things: which - chain the handle looks up, and that a read ends when that run closes, - Continue-as-New included. Without it a reader keeps reading across - Continue-as-New. ``topic`` is the topic a handle opened from the ref - addresses when a call names none. + release. A Workflow ref carries ``workflow_id`` and names no run: the + stream belongs to the Workflow's run chain, and a read on it follows the + chain across Continue-as-New. ``topic`` is the topic a handle opened from + the ref addresses when a call names none. The default data converter carries it as JSON, so it can be a Workflow argument or an Activity result. @@ -49,7 +46,6 @@ class StreamRef: kind: str workflow_id: str - run_id: str | None = None topic: str = DEFAULT_TOPIC def __post_init__(self) -> None: @@ -71,7 +67,6 @@ def for_workflow( cls, workflow_id: str, *, - run_id: str | None = None, topic: str | StreamTopic[Any] | None = None, ) -> StreamRef: """A ref to ``topic`` of ``workflow_id``'s stream. @@ -80,7 +75,7 @@ def for_workflow( :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. """ name, _ = resolve_topic(topic) - return cls("workflow", workflow_id, run_id=run_id, topic=name) + return cls("workflow", workflow_id, topic=name) def with_topic(self, topic: str | StreamTopic[Any] | None) -> StreamRef: """The same owner, naming ``topic`` instead. diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 3cbbe1e6b0..deb3053f85 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -44,12 +44,24 @@ def test_a_ref_names_a_workflow_stream(): def test_a_ref_round_trips_through_the_default_converter(): - ref = StreamRef.for_workflow("wf", run_id="run", topic=OUT) + ref = StreamRef.for_workflow("wf", topic=OUT) converter = DataConverter.default.payload_converter payload = converter.to_payloads([ref])[0] assert converter.from_payloads([payload], [StreamRef])[0] == ref +def test_a_ref_that_names_a_run_decodes_as_the_chain(): + # A reader follows the run chain, so a run id from another process is + # dropped rather than refused. + converter = DataConverter.default.payload_converter + payload = Payload( + metadata={"encoding": b"json/plain"}, + data=b'{"kind": "workflow", "workflow_id": "wf", "run_id": "r", "topic": "out"}', + ) + ref = converter.from_payloads([payload], [StreamRef])[0] + assert ref == StreamRef.for_workflow("wf", topic=OUT) + + def test_a_ref_refuses_what_it_cannot_name(): with pytest.raises(ValueError, match="unknown"): StreamRef("nexus", "x") From 17f61e1525c6346d69ac44c32a4ba6fae198a1ab Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 18:16:53 -0700 Subject: [PATCH 4/4] Let a StreamRef name its chain by a run, without ending reads there. --- temporalio/contrib/streams/_ref.py | 21 +++++++++++++++------ tests/contrib/streams/test_topics_refs.py | 8 ++++---- 2 files changed, 19 insertions(+), 10 deletions(-) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index b63ae73f15..e41c7cb7d8 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -35,10 +35,15 @@ class StreamRef: """One stream, named by its owner and its topic. ``kind`` says what owns the stream; only ``"workflow"`` exists in this - release. A Workflow ref carries ``workflow_id`` and names no run: the - stream belongs to the Workflow's run chain, and a read on it follows the - chain across Continue-as-New. ``topic`` is the topic a handle opened from - the ref addresses when a call names none. + release. A Workflow ref carries ``workflow_id``. The stream belongs to + the Workflow's run chain, and a read on it follows the chain across + Continue-as-New and ends only when the chain ends. ``topic`` is the topic + a handle opened from the ref addresses when a call names none. + + ``chain_run_id`` picks which chain: the one that holds that run. A + Workflow id can start a new chain once the old one closed, and without + ``chain_run_id`` a ref names the newest chain. It never stops a read at + that run's end. The default data converter carries it as JSON, so it can be a Workflow argument or an Activity result. @@ -47,6 +52,7 @@ class StreamRef: kind: str workflow_id: str topic: str = DEFAULT_TOPIC + chain_run_id: str | None = None def __post_init__(self) -> None: """Refuse a ref this release cannot open. @@ -68,14 +74,17 @@ def for_workflow( workflow_id: str, *, topic: str | StreamTopic[Any] | None = None, + chain_run_id: str | None = None, ) -> StreamRef: """A ref to ``topic`` of ``workflow_id``'s stream. Without ``topic`` it names - :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. Without + ``chain_run_id`` it names the Workflow id's newest chain, and with + one the chain that holds that run. """ name, _ = resolve_topic(topic) - return cls("workflow", workflow_id, topic=name) + return cls("workflow", workflow_id, topic=name, chain_run_id=chain_run_id) def with_topic(self, topic: str | StreamTopic[Any] | None) -> StreamRef: """The same owner, naming ``topic`` instead. diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index deb3053f85..2f51d457b5 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -44,15 +44,15 @@ def test_a_ref_names_a_workflow_stream(): def test_a_ref_round_trips_through_the_default_converter(): - ref = StreamRef.for_workflow("wf", topic=OUT) + ref = StreamRef.for_workflow("wf", topic=OUT, chain_run_id="run") converter = DataConverter.default.payload_converter payload = converter.to_payloads([ref])[0] assert converter.from_payloads([payload], [StreamRef])[0] == ref -def test_a_ref_that_names_a_run_decodes_as_the_chain(): - # A reader follows the run chain, so a run id from another process is - # dropped rather than refused. +def test_a_ref_with_the_old_run_id_decodes_as_the_newest_chain(): + # The old field also ended a read at that run, which no read does now, + # so it is dropped rather than read as chain_run_id. converter = DataConverter.default.payload_converter payload = Payload( metadata={"encoding": b"json/plain"},