[Brook] Added the Redis and memory stream store plugins.

temporalio/contrib/streams/__init__.py

@@ -11,19 +11,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.
@@ -43,6 +54,7 @@
     StreamStorageError,
     StreamUnsupportedError,
 )
+from temporalio.contrib.streams._plugin import StreamStorePlugin
 from temporalio.contrib.streams._record import (
     Cursor,
     RecordKind,
@@ -71,6 +83,7 @@
     "StreamRef",
     "StreamRefusedError",
     "StreamStorageError",
+    "StreamStorePlugin",
     "StreamTopic",
     "StreamUnsupportedError",
     "Supersession",

temporalio/contrib/streams/_plugin.py

@@ -0,0 +1,234 @@
+"""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
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import TYPE_CHECKING, Any, TypeVar, cast
+
+import temporalio.bridge.client
+import temporalio.worker
+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.service import ConnectConfig, ServiceClient
+    from temporalio.worker import WorkerConfig
+
+__all__ = ["StreamStorePlugin"]
+
+_WORKFLOW_EXTERN = "__temporal_contrib_streams_store"
+
+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:
+        """Hand the client's store to the Worker.
+
+        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])"
+            )
+        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
+
+    @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: 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"
+    )
+
+
+class _StreamsInterceptor(temporalio.worker.Interceptor):
+    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 worker_has_store() -> bool:
+    """Whether the Worker running the current Workflow has a stream store."""
+    return _WORKFLOW_EXTERN in temporalio.workflow.extern_functions()

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()),
+        )

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),
+        )