From 6e961de4560d8432955238f058d7d9e08ae9a703 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 14:57:48 -0700 Subject: [PATCH 1/5] Added the stream provider protocol and its plugin. A provider hands out handles and producers, and registers once on a client or Worker, so the accessors never pick between two stores silently. --- temporalio/contrib/streams/__init__.py | 26 ++- temporalio/contrib/streams/_plugin.py | 110 +++++++++ temporalio/contrib/streams/_provider.py | 218 ++++++++++++++++++ tests/contrib/streams/test_provider_plugin.py | 57 +++++ 4 files changed, 408 insertions(+), 3 deletions(-) create mode 100644 temporalio/contrib/streams/_plugin.py create mode 100644 temporalio/contrib/streams/_provider.py create mode 100644 tests/contrib/streams/test_provider_plugin.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 0221a926ca..9903d37ba9 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -10,19 +10,29 @@ 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 provider 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 provider refuses a cursor from another provider or another stream 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 provider is a plugin: ``Client.connect(..., plugins=[provider])`` +registers it on the client and on every Worker built from that client. + 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`` as an ordinary payload. @@ -46,6 +56,12 @@ StreamProducerError, StreamUnsupportedError, ) +from temporalio.contrib.streams._plugin import StreamProviderPlugin +from temporalio.contrib.streams._provider import ( + StreamHandle, + StreamProducer, + StreamProvider, +) from temporalio.contrib.streams._record import ( Cursor, RecordKind, @@ -71,10 +87,14 @@ "StreamCursorError", "StreamError", "StreamExpiredError", + "StreamHandle", "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamOwnerKind", + "StreamProducer", "StreamProducerError", + "StreamProvider", + "StreamProviderPlugin", "StreamRecord", "StreamRef", "StreamTopic", diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py new file mode 100644 index 0000000000..a88ff03b01 --- /dev/null +++ b/temporalio/contrib/streams/_plugin.py @@ -0,0 +1,110 @@ +"""The plugin a provider is. + +A provider is registered once, as a plugin: +``Client.connect(plugins=[provider])``. The client keeps it, and every Worker +built from that client inherits it. ``Worker(plugins=[provider])`` and +``Replayer(plugins=[provider])`` register it on a Worker alone. + +On a Worker the plugin adds one interceptor and nothing else, so the +Worker knows which provider it carries. No Worker internals change. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Any + +import temporalio.worker +from temporalio.contrib.streams._provider import StreamHandle +from temporalio.contrib.streams._ref import StreamRef +from temporalio.plugin import SimplePlugin + +if TYPE_CHECKING: + from temporalio.client import Client, ClientConfig + from temporalio.worker import ReplayerConfig, WorkerConfig + +__all__ = ["StreamProviderPlugin"] + + +class StreamProviderPlugin(SimplePlugin, ABC): + """The base class of every provider. + + A subclass implements :class:`temporalio.contrib.streams.StreamProvider`. + This class makes it a client and Worker plugin. One provider serves a + client and its Workers; registering a second, different provider on the + same client or Worker raises ``ValueError``, because the accessors would + otherwise pick one without saying so. A process that talks to two stores + opens the second store's handles from the provider object. + """ + + def __init__(self, name: str) -> None: + """Name the plugin; the name shows in the Worker's plugin list.""" + super().__init__(name) + self._interceptor = _StreamsInterceptor(self) + + @abstractmethod + def get_stream_handle(self, client: Client, ref: StreamRef) -> StreamHandle: + """See :meth:`temporalio.contrib.streams.StreamProvider.get_stream_handle`.""" + + @abstractmethod + async def close(self) -> None: + """See :meth:`temporalio.contrib.streams.StreamProvider.close`.""" + + def configure_client(self, config: ClientConfig) -> ClientConfig: + """Register this provider on the client. + + Raises: + ValueError: Another provider is already registered on the client. + """ + for plugin in config.get("plugins", []): + if isinstance(plugin, StreamProviderPlugin) and plugin is not self: + raise ValueError(_two_providers(plugin, self)) + return super().configure_client(config) + + def configure_worker(self, config: WorkerConfig) -> WorkerConfig: + """Register this provider on the Worker. + + Raises: + ValueError: Another provider is already registered on the Worker. + """ + config = super().configure_worker(config) + config["interceptors"] = self._with_interceptor(config.get("interceptors")) + return config + + def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig: + """Register this provider on the Replayer. + + Raises: + ValueError: Another provider is already registered on the Replayer. + """ + config = super().configure_replayer(config) + config["interceptors"] = self._with_interceptor(config.get("interceptors")) + return config + + def _with_interceptor( + self, interceptors: Any + ) -> list[temporalio.worker.Interceptor]: + held = list(interceptors or []) + for interceptor in held: + if isinstance(interceptor, _StreamsInterceptor): + if interceptor.provider is not self: + raise ValueError(_two_providers(interceptor.provider, self)) + # The same plugin on the client and the Worker configures the + # Worker twice; one interceptor is enough. + return held + return [*held, self._interceptor] + + +def _two_providers(held: StreamProviderPlugin, new: StreamProviderPlugin) -> str: + return ( + f"stream provider {held.name()!r} is already registered, so {new.name()!r} " + "cannot be; register one provider and open the other's handles from the " + "provider object" + ) + + +class _StreamsInterceptor(temporalio.worker.Interceptor): + """Marks the provider a Worker carries.""" + + def __init__(self, provider: StreamProviderPlugin) -> None: + self.provider = provider diff --git a/temporalio/contrib/streams/_provider.py b/temporalio/contrib/streams/_provider.py new file mode 100644 index 0000000000..2903e2565d --- /dev/null +++ b/temporalio/contrib/streams/_provider.py @@ -0,0 +1,218 @@ +"""What a provider implements. + +A provider is the store behind a stream. It hands out a +:class:`StreamHandle` per stream, and a handle reads, positions and makes +producers. A provider moves ``temporal.sdk.streams.v1.StreamRecord`` protos; +the shared helpers in this package convert values, synthesize +``SUPERSEDED``, mint stream-bound cursors and run bodies through the data +converter, so every provider behaves the same where the contract says so. + +What a provider owes a record's body: the handle converts a value into the +body with the payload converter, then takes the retry fingerprint +(:func:`temporalio.contrib.streams.content_fingerprint`) over the converted +records, then runs each body through +:func:`temporalio.contrib.streams.encode_body`, which stamps the plaintext +hash and applies the codec and external storage. On a read it runs +:func:`temporalio.contrib.streams.decode_body` before the record reaches the +reader. +""" + +from __future__ import annotations + +from collections.abc import AsyncGenerator +from typing import TYPE_CHECKING, Any, Protocol, TypeVar, overload + +from temporalio.contrib.streams._cursor import BEGINNING +from temporalio.contrib.streams._record import Cursor, StreamRecord +from temporalio.contrib.streams._ref import StreamRef +from temporalio.contrib.streams._topic import StreamTopic + +if TYPE_CHECKING: + from temporalio.client import Client + +__all__ = ["StreamHandle", "StreamProducer", "StreamProvider"] + +T = TypeVar("T") +T_contra = TypeVar("T_contra", contravariant=True) + + +class StreamProducer(Protocol[T_contra]): + """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. + """ + + @property + def producer_id(self) -> str: + """Who this producer writes as.""" + ... + + @property + def attempt(self) -> int: + """The attempt this producer writes.""" + ... + + async def append(self, *values: T_contra) -> 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. + + 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. + StreamClosedError: The stream refuses appends. + """ + ... + + 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. + """ + ... + + +class StreamHandle(Protocol): + """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`. + """ + + @property + def ref(self) -> StreamRef: + """The stream this handle is on, as data another process can open.""" + ... + + @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 record once. 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. + StreamCursorError: ``after`` came from another provider or + another stream. Raised by this call, not by the first + iteration. + StreamExpiredError: ``after`` names a record the store no longer + retains, or the read fell behind retention while it ran. + """ + ... + + 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. + """ + ... + + @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. + + Raises: + ValueError: ``producer_id`` is empty or ``attempt`` is below one. + """ + ... + + +class StreamProvider(Protocol): + """A store that holds streams. + + In this release only a Workflow owns a stream. The provider keeps the + stream for the owner's run chain, so a handle that is not pinned to a run + reads across Continue-as-New. + """ + + def get_stream_handle(self, client: Client, ref: StreamRef) -> StreamHandle: + """A handle on the stream ``ref`` names. + + ``client`` is how the handle learns that the owner closed and which + data converter to encode bodies with. + """ + ... + + async def close(self) -> None: + """Release what this provider holds for the process. + + The application calls this, not the Worker and not the client: one + provider serves every Worker built from a client and every handle + opened outside them, so none of those owns its lifetime. + """ + ... diff --git a/tests/contrib/streams/test_provider_plugin.py b/tests/contrib/streams/test_provider_plugin.py new file mode 100644 index 0000000000..8cf1bcf6da --- /dev/null +++ b/tests/contrib/streams/test_provider_plugin.py @@ -0,0 +1,57 @@ +"""A provider registers as a plugin, once per client or Worker.""" + +from __future__ import annotations + +import pytest + +from temporalio import activity +from temporalio.client import Client +from temporalio.contrib.streams import StreamHandle, StreamProviderPlugin, StreamRef +from temporalio.contrib.streams._plugin import _StreamsInterceptor +from temporalio.worker import Worker + + +class _StubProvider(StreamProviderPlugin): + def __init__(self, name: str) -> None: + super().__init__(name) + + def get_stream_handle(self, client: Client, ref: StreamRef) -> StreamHandle: + raise NotImplementedError + + async def close(self) -> None: + pass + + +async def test_one_provider_per_client(client: Client): + first, second = _StubProvider("first"), _StubProvider("second") + config = client.config() + config["plugins"] = [first, second] + with pytest.raises(ValueError, match="is already registered"): + Client(**config) + # The same instance twice is one provider, not two. + config["plugins"] = [first, first] + Client(**config) + + +class _OtherStubProvider(_StubProvider): + pass + + +async def test_one_provider_per_worker(client: Client): + first, second = _StubProvider("first"), _OtherStubProvider("second") + config = client.config() + config["plugins"] = [first] + with pytest.raises(ValueError, match="is already registered"): + Worker(Client(**config), task_queue="tq", activities=[_noop], plugins=[second]) + # The provider on the client and the Worker adds one interceptor. + with pytest.warns(UserWarning, match="same plugin type"): + worker = Worker( + Client(**config), task_queue="tq", activities=[_noop], plugins=[first] + ) + interceptors = worker.config(active_config=True).get("interceptors", []) + assert len([i for i in interceptors if isinstance(i, _StreamsInterceptor)]) == 1 + + +@activity.defn +async def _noop() -> None: + pass From 64f11b0c59d36754b7b40066fb89bce2c83eb6d8 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 17:21:31 -0700 Subject: [PATCH 2/5] Documented producer concurrency and recovery in the stream protocol. Providers must serialize calls on one producer, and a caller needs to know what to do after a cancelled or unknown append. --- temporalio/contrib/streams/_provider.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/temporalio/contrib/streams/_provider.py b/temporalio/contrib/streams/_provider.py index 2903e2565d..f024bc9dd3 100644 --- a/temporalio/contrib/streams/_provider.py +++ b/temporalio/contrib/streams/_provider.py @@ -9,11 +9,11 @@ What a provider owes a record's body: the handle converts a value into the body with the payload converter, then takes the retry fingerprint -(:func:`temporalio.contrib.streams.content_fingerprint`) over the converted +(:func:`temporalio.contrib.streams._body.content_fingerprint`) over the converted records, then runs each body through -:func:`temporalio.contrib.streams.encode_body`, which stamps the plaintext +:func:`temporalio.contrib.streams._body.encode_body`, which stamps the plaintext hash and applies the codec and external storage. On a read it runs -:func:`temporalio.contrib.streams.decode_body` before the record reaches the +:func:`temporalio.contrib.streams._body.decode_body` before the record reaches the reader. """ @@ -68,12 +68,20 @@ async def append(self, *values: T_contra) -> Cursor: 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. + 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. """ ... From 526e8dcac9dbbcaaa18681e3f857640e05b7cf94 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 13:05:06 -0700 Subject: [PATCH 3/5] Documented the resume limit, one producer per session, and the warn. The read contract promised every record once, which a resumed read keeps only for stored records. A second producer object for one session is refused, and registering the plugin twice warns. --- temporalio/contrib/streams/_plugin.py | 4 +++- temporalio/contrib/streams/_provider.py | 9 +++++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py index a88ff03b01..35beeece84 100644 --- a/temporalio/contrib/streams/_plugin.py +++ b/temporalio/contrib/streams/_plugin.py @@ -3,7 +3,9 @@ A provider is registered once, as a plugin: ``Client.connect(plugins=[provider])``. The client keeps it, and every Worker built from that client inherits it. ``Worker(plugins=[provider])`` and -``Replayer(plugins=[provider])`` register it on a Worker alone. +``Replayer(plugins=[provider])`` register it on a Worker alone. Registering it +on the client and again on a Worker built from it warns about a duplicate +plugin. On a Worker the plugin adds one interceptor and nothing else, so the Worker knows which provider it carries. No Worker internals change. diff --git a/temporalio/contrib/streams/_provider.py b/temporalio/contrib/streams/_provider.py index f024bc9dd3..e5a526ab6d 100644 --- a/temporalio/contrib/streams/_provider.py +++ b/temporalio/contrib/streams/_provider.py @@ -146,7 +146,10 @@ def read( ``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 record once. The read ends when the owner's run + 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. @@ -192,7 +195,9 @@ 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. + 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. From 414189a4594952d08373c74263006fdf4609235b Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 22:08:46 -0700 Subject: [PATCH 4/5] Made RedisStreams and MemoryStreams configure Core's stream store. Core runs the store, so the provider classes become client plugins that name one. Connecting the client connects the store, and every Worker built from that client hands it to Core. --- temporalio/contrib/streams/__init__.py | 16 +- temporalio/contrib/streams/_plugin.py | 239 ++++++++++++------ temporalio/contrib/streams/_provider.py | 231 ----------------- temporalio/contrib/streams/memory.py | 24 ++ temporalio/contrib/streams/redis.py | 76 ++++++ temporalio/worker/_worker.py | 10 + tests/contrib/streams/_support.py | 48 ++++ tests/contrib/streams/test_provider_plugin.py | 57 ----- tests/contrib/streams/test_store_plugin.py | 132 ++++++++++ 9 files changed, 465 insertions(+), 368 deletions(-) delete mode 100644 temporalio/contrib/streams/_provider.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 delete mode 100644 tests/contrib/streams/test_provider_plugin.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 25252961f1..12a738abbe 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -21,7 +21,7 @@ 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 @@ -30,6 +30,10 @@ 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. @@ -49,12 +53,7 @@ StreamStorageError, StreamUnsupportedError, ) -from temporalio.contrib.streams._plugin import StreamProviderPlugin -from temporalio.contrib.streams._provider import ( - StreamHandle, - StreamProducer, - StreamProvider, -) +from temporalio.contrib.streams._plugin import StreamStorePlugin from temporalio.contrib.streams._record import ( Cursor, RecordKind, @@ -74,17 +73,16 @@ "StreamCursorError", "StreamError", "StreamExpiredError", - "StreamHandle", "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamOwnerKind", - "StreamProducer", "StreamProducerError", "StreamRecord", "StreamRecordError", "StreamRef", "StreamRefusedError", "StreamStorageError", + "StreamStorePlugin", "StreamTopic", "StreamUnsupportedError", "Supersession", diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py index 35beeece84..f8da83c641 100644 --- a/temporalio/contrib/streams/_plugin.py +++ b/temporalio/contrib/streams/_plugin.py @@ -1,112 +1,209 @@ -"""The plugin a provider is. - -A provider is registered once, as a plugin: -``Client.connect(plugins=[provider])``. The client keeps it, and every Worker -built from that client inherits it. ``Worker(plugins=[provider])`` and -``Replayer(plugins=[provider])`` register it on a Worker alone. Registering it -on the client and again on a Worker built from it warns about a duplicate -plugin. - -On a Worker the plugin adds one interceptor and nothing else, so the -Worker knows which provider it carries. No Worker internals change. +"""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 every Worker built from the client, so a Workflow's publish +and an outside read meet in the same place. + +On a Worker the plugin hands Core the store, which commits a Workflow's +records with its Workflow Task, and adds one interceptor. The interceptor +tells Workflow code, through the Worker's extern functions, that the Worker +has a store, so a publish on a Worker without one fails at the call. """ from __future__ import annotations -from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, Any +import asyncio +from collections.abc import Awaitable, Callable +from typing import TYPE_CHECKING, Any, TypeVar, cast +import temporalio.bridge.client import temporalio.worker -from temporalio.contrib.streams._provider import StreamHandle -from temporalio.contrib.streams._ref import StreamRef +import temporalio.workflow +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.bridge.temporal_sdk_bridge import StreamStoreRef from temporalio.client import Client, ClientConfig - from temporalio.worker import ReplayerConfig, WorkerConfig + from temporalio.service import ConnectConfig, ServiceClient + from temporalio.worker import WorkerConfig -__all__ = ["StreamProviderPlugin"] +__all__ = ["StreamStorePlugin"] +_WORKFLOW_EXTERN = "__temporal_contrib_streams_store" -class StreamProviderPlugin(SimplePlugin, ABC): - """The base class of every provider. +T = TypeVar("T") - A subclass implements :class:`temporalio.contrib.streams.StreamProvider`. - This class makes it a client and Worker plugin. One provider serves a - client and its Workers; registering a second, different provider on the - same client or Worker raises ``ValueError``, because the accessors would - otherwise pick one without saying so. A process that talks to two stores - opens the second store's handles from the provider object. - """ - def __init__(self, name: str) -> None: - """Name the plugin; the name shows in the Worker's plugin list.""" - super().__init__(name) - self._interceptor = _StreamsInterceptor(self) +class StreamStorePlugin(SimplePlugin): + """The base class of the store configuration classes. - @abstractmethod - def get_stream_handle(self, client: Client, ref: StreamRef) -> StreamHandle: - """See :meth:`temporalio.contrib.streams.StreamProvider.get_stream_handle`.""" + Use :class:`temporalio.contrib.streams.redis.RedisStreams` or + :class:`temporalio.contrib.streams.memory.MemoryStreams`. One instance + is one store. Registering a second, different store on the same client + raises ``ValueError``, because a handle could otherwise pick either. + """ - @abstractmethod - async def close(self) -> None: - """See :meth:`temporalio.contrib.streams.StreamProvider.close`.""" + 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 def configure_client(self, config: ClientConfig) -> ClientConfig: - """Register this provider on the client. + """Register this store on the client. Raises: - ValueError: Another provider is already registered on the client. + ValueError: Another store is already registered on the client. """ for plugin in config.get("plugins", []): - if isinstance(plugin, StreamProviderPlugin) and plugin is not self: - raise ValueError(_two_providers(plugin, self)) + 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) + return service_client + def configure_worker(self, config: WorkerConfig) -> WorkerConfig: - """Register this provider on the Worker. + """Hand the client's store to the Worker. Raises: - ValueError: Another provider is already registered on the Worker. + ValueError: The plugin is not registered on the Worker's client, + or that client was not made with ``Client.connect``. """ config = super().configure_worker(config) - config["interceptors"] = self._with_interceptor(config.get("interceptors")) + client = config["client"] # type:ignore[reportTypedDictNotRequiredAccess] + if self not in client.config()["plugins"] or self._store is None: + raise ValueError( + f"stream store {self.name()!r} must be registered on the Worker's " + "client, as Client.connect(..., plugins=[store])" + ) + held = list(config.get("interceptors") or []) + # The same plugin on the client and the Worker configures the Worker twice. + if not any(isinstance(i, _StreamsInterceptor) for i in held): + held.append(_StreamsInterceptor()) + config["interceptors"] = held return config - def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig: - """Register this provider on the Replayer. + @property + def _temporal_stream_store(self) -> StreamStoreRef | None: + # The Worker looks for this name on its plugins and gives the store to Core. + return self._store.ref if self._store is not None else None + + 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: Another provider is already registered on the Replayer. + ValueError: No client connected this store yet. + temporalio.contrib.streams.StreamStorageError: The store failed; + some keys may be deleted. """ - config = super().configure_replayer(config) - config["interceptors"] = self._with_interceptor(config.get("interceptors")) - return config - - def _with_interceptor( - self, interceptors: Any - ) -> list[temporalio.worker.Interceptor]: - held = list(interceptors or []) - for interceptor in held: - if isinstance(interceptor, _StreamsInterceptor): - if interceptor.provider is not self: - raise ValueError(_two_providers(interceptor.provider, self)) - # The same plugin on the client and the Worker configures the - # Worker twice; one interceptor is enough. - return held - return [*held, self._interceptor] - - -def _two_providers(held: StreamProviderPlugin, new: StreamProviderPlugin) -> str: + 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.""" + if self._service is None: + 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._connecting is None: + self._connecting = asyncio.Lock() + async with self._connecting: + if self._store is not None: + return + 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) + + +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 provider {held.name()!r} is already registered, so {new.name()!r} " - "cannot be; register one provider and open the other's handles from the " - "provider object" + f"stream store {held.name()!r} is already registered, so {new.name()!r} " + "cannot be; register one store per client" ) class _StreamsInterceptor(temporalio.worker.Interceptor): - """Marks the provider a Worker carries.""" + def workflow_interceptor_class( + self, input: temporalio.worker.WorkflowInterceptorClassInput + ) -> type[temporalio.worker.WorkflowInboundInterceptor] | None: + input.unsafe_extern_functions[_WORKFLOW_EXTERN] = lambda: True + return None + + +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(...)])" + ) + - def __init__(self, provider: StreamProviderPlugin) -> None: - self.provider = provider +def worker_has_store() -> bool: + """Whether the Worker running the current Workflow has a stream store.""" + return _WORKFLOW_EXTERN in temporalio.workflow.extern_functions() diff --git a/temporalio/contrib/streams/_provider.py b/temporalio/contrib/streams/_provider.py deleted file mode 100644 index e5a526ab6d..0000000000 --- a/temporalio/contrib/streams/_provider.py +++ /dev/null @@ -1,231 +0,0 @@ -"""What a provider implements. - -A provider is the store behind a stream. It hands out a -:class:`StreamHandle` per stream, and a handle reads, positions and makes -producers. A provider moves ``temporal.sdk.streams.v1.StreamRecord`` protos; -the shared helpers in this package convert values, synthesize -``SUPERSEDED``, mint stream-bound cursors and run bodies through the data -converter, so every provider behaves the same where the contract says so. - -What a provider owes a record's body: the handle converts a value into the -body with the payload converter, then takes the retry fingerprint -(:func:`temporalio.contrib.streams._body.content_fingerprint`) over the converted -records, then runs each body through -:func:`temporalio.contrib.streams._body.encode_body`, which stamps the plaintext -hash and applies the codec and external storage. On a read it runs -:func:`temporalio.contrib.streams._body.decode_body` before the record reaches the -reader. -""" - -from __future__ import annotations - -from collections.abc import AsyncGenerator -from typing import TYPE_CHECKING, Any, Protocol, TypeVar, overload - -from temporalio.contrib.streams._cursor import BEGINNING -from temporalio.contrib.streams._record import Cursor, StreamRecord -from temporalio.contrib.streams._ref import StreamRef -from temporalio.contrib.streams._topic import StreamTopic - -if TYPE_CHECKING: - from temporalio.client import Client - -__all__ = ["StreamHandle", "StreamProducer", "StreamProvider"] - -T = TypeVar("T") -T_contra = TypeVar("T_contra", contravariant=True) - - -class StreamProducer(Protocol[T_contra]): - """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. - """ - - @property - def producer_id(self) -> str: - """Who this producer writes as.""" - ... - - @property - def attempt(self) -> int: - """The attempt this producer writes.""" - ... - - async def append(self, *values: T_contra) -> 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 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. - """ - ... - - -class StreamHandle(Protocol): - """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`. - """ - - @property - def ref(self) -> StreamRef: - """The stream this handle is on, as data another process can open.""" - ... - - @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. - StreamCursorError: ``after`` came from another provider or - another stream. Raised by this call, not by the first - iteration. - StreamExpiredError: ``after`` names a record the store no longer - retains, or the read fell behind retention while it ran. - """ - ... - - 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. - """ - ... - - @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. - """ - ... - - -class StreamProvider(Protocol): - """A store that holds streams. - - In this release only a Workflow owns a stream. The provider keeps the - stream for the owner's run chain, so a handle that is not pinned to a run - reads across Continue-as-New. - """ - - def get_stream_handle(self, client: Client, ref: StreamRef) -> StreamHandle: - """A handle on the stream ``ref`` names. - - ``client`` is how the handle learns that the owner closed and which - data converter to encode bodies with. - """ - ... - - async def close(self) -> None: - """Release what this provider holds for the process. - - The application calls this, not the Worker and not the client: one - provider serves every Worker built from a client and every handle - opened outside them, so none of those owns its lifetime. - """ - ... 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..bd699d9a11 --- /dev/null +++ b/temporalio/contrib/streams/redis.py @@ -0,0 +1,76 @@ +"""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. + 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/temporalio/worker/_worker.py b/temporalio/worker/_worker.py index 39faf615ae..22afcf8d3c 100644 --- a/temporalio/worker/_worker.py +++ b/temporalio/worker/_worker.py @@ -702,6 +702,16 @@ def check_activity(activity: str): ]._to_bridge(), # type: ignore[reportTypedDictNotRequiredAccess,reportOptionalMemberAccess] plugins=deduped_plugin_names, storage_drivers=deduped_storage_driver_types, + # A stream store plugin offers its store under this name, which keeps the bridge + # type off the public Worker options. + stream_store=next( + ( + store + for plugin in self._plugins + if (store := getattr(plugin, "_temporal_stream_store", None)) + ), + None, + ), disable_payload_error_limit=config.get( "disable_payload_error_limit", False ), 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_provider_plugin.py b/tests/contrib/streams/test_provider_plugin.py deleted file mode 100644 index 8cf1bcf6da..0000000000 --- a/tests/contrib/streams/test_provider_plugin.py +++ /dev/null @@ -1,57 +0,0 @@ -"""A provider registers as a plugin, once per client or Worker.""" - -from __future__ import annotations - -import pytest - -from temporalio import activity -from temporalio.client import Client -from temporalio.contrib.streams import StreamHandle, StreamProviderPlugin, StreamRef -from temporalio.contrib.streams._plugin import _StreamsInterceptor -from temporalio.worker import Worker - - -class _StubProvider(StreamProviderPlugin): - def __init__(self, name: str) -> None: - super().__init__(name) - - def get_stream_handle(self, client: Client, ref: StreamRef) -> StreamHandle: - raise NotImplementedError - - async def close(self) -> None: - pass - - -async def test_one_provider_per_client(client: Client): - first, second = _StubProvider("first"), _StubProvider("second") - config = client.config() - config["plugins"] = [first, second] - with pytest.raises(ValueError, match="is already registered"): - Client(**config) - # The same instance twice is one provider, not two. - config["plugins"] = [first, first] - Client(**config) - - -class _OtherStubProvider(_StubProvider): - pass - - -async def test_one_provider_per_worker(client: Client): - first, second = _StubProvider("first"), _OtherStubProvider("second") - config = client.config() - config["plugins"] = [first] - with pytest.raises(ValueError, match="is already registered"): - Worker(Client(**config), task_queue="tq", activities=[_noop], plugins=[second]) - # The provider on the client and the Worker adds one interceptor. - with pytest.warns(UserWarning, match="same plugin type"): - worker = Worker( - Client(**config), task_queue="tq", activities=[_noop], plugins=[first] - ) - interceptors = worker.config(active_config=True).get("interceptors", []) - assert len([i for i in interceptors if isinstance(i, _StreamsInterceptor)]) == 1 - - -@activity.defn -async def _noop() -> None: - pass diff --git a/tests/contrib/streams/test_store_plugin.py b/tests/contrib/streams/test_store_plugin.py new file mode 100644 index 0000000000..cd25125c71 --- /dev/null +++ b/tests/contrib/streams/test_store_plugin.py @@ -0,0 +1,132 @@ +"""A store configuration registers as a client plugin, and Core connects to it.""" + +from __future__ import annotations + +import uuid +from datetime import timedelta + +import pytest + +from temporalio import activity +from temporalio.bridge.proto.streams import ( + LatestRequest, + StreamAddress, + StreamOwnerKind, +) +from temporalio.client import Client +from temporalio.contrib.streams import StreamError, StreamNotFoundError +from temporalio.contrib.streams._plugin import _StreamsInterceptor, 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 = Worker(streams_client, task_queue=str(uuid.uuid4()), activities=[_noop]) + interceptors = worker.config(active_config=True).get("interceptors", []) + assert len([i for i in interceptors if isinstance(i, _StreamsInterceptor)]) == 1 + assert store._temporal_stream_store is not None + # The store on the client and the Worker still adds one interceptor. + with pytest.warns(UserWarning, match="same plugin type"): + worker = Worker( + streams_client, + task_queue=str(uuid.uuid4()), + activities=[_noop], + plugins=[store], + ) + interceptors = worker.config(active_config=True).get("interceptors", []) + assert len([i for i in interceptors if isinstance(i, _StreamsInterceptor)]) == 1 + + +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_reach_core(): + 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_needs_a_connected_store(client: Client): + store = MemoryStreams() + with pytest.raises(ValueError, match="no client has connected"): + await store.delete_workflow_streams(client.namespace, "wf") + await connect_with(client, store) + assert await store.delete_workflow_streams(client.namespace, "wf") == 0 + + +@activity.defn +async def _noop() -> None: + pass From f82049b953325317bdb8d669ce4958b7310df479 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:30:48 -0700 Subject: [PATCH 5/5] Bound a stream store to one client connection. Core asks about owners over the connection that connected the store, so a second connection got the first one's answers. A Worker now needs a client made with Client.connect whatever connected the store first, and the delete test checks a stream that held a record. --- temporalio/contrib/streams/_plugin.py | 41 ++++++++--- temporalio/contrib/streams/redis.py | 5 +- tests/contrib/streams/test_store_plugin.py | 79 ++++++++++++++++++++-- 3 files changed, 112 insertions(+), 13 deletions(-) diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py index f8da83c641..2966da98aa 100644 --- a/temporalio/contrib/streams/_plugin.py +++ b/temporalio/contrib/streams/_plugin.py @@ -49,8 +49,11 @@ class StreamStorePlugin(SimplePlugin): Use :class:`temporalio.contrib.streams.redis.RedisStreams` or :class:`temporalio.contrib.streams.memory.MemoryStreams`. One instance - is one store. Registering a second, different store on the same client - raises ``ValueError``, because a handle could otherwise pick either. + 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: @@ -60,6 +63,9 @@ def __init__(self, name: str, config: StreamStoreConfig) -> None: 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. @@ -88,6 +94,7 @@ async def connect_service_client( """ 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: @@ -95,11 +102,17 @@ def configure_worker(self, config: WorkerConfig) -> WorkerConfig: Raises: ValueError: The plugin is not registered on the Worker's client, - or that client was not made with ``Client.connect``. + 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 self._store is None: + 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])" @@ -147,23 +160,35 @@ async def delete_workflow_streams(self, namespace: str, workflow_id: str) -> int 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.""" - if self._service is None: - await self._connected(client.service_client) + """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._store is not None: + 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: diff --git a/temporalio/contrib/streams/redis.py b/temporalio/contrib/streams/redis.py index bd699d9a11..7d0be4409a 100644 --- a/temporalio/contrib/streams/redis.py +++ b/temporalio/contrib/streams/redis.py @@ -34,7 +34,10 @@ def __init__( Args: url: A ``redis://`` or ``rediss://`` URL, or the seed nodes of a - cluster. Credentials and TLS settings come from the URL. + 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. diff --git a/tests/contrib/streams/test_store_plugin.py b/tests/contrib/streams/test_store_plugin.py index cd25125c71..e4a236be27 100644 --- a/tests/contrib/streams/test_store_plugin.py +++ b/tests/contrib/streams/test_store_plugin.py @@ -2,17 +2,23 @@ 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 _StreamsInterceptor, call @@ -88,7 +94,7 @@ async def test_a_store_core_cannot_reach_fails_the_connect(client: Client): ) -def test_redis_settings_reach_core(): +def test_redis_settings_go_into_the_store_config(): store = RedisStreams( ["redis://a:6379", "redis://b:6379"], cluster=True, @@ -119,14 +125,79 @@ def test_redis_setting_mistakes_are_value_errors(): RedisStreams("redis://a", retention=timedelta(microseconds=10)) -async def test_deleting_needs_a_connected_store(client: Client): +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") - await connect_with(client, store) - assert await store.delete_workflow_streams(client.namespace, "wf") == 0 + 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", + )