[Brook S6] Added the Redis and in-memory Stream backends.

.github/actions/install-redis/action.yml

@@ -0,0 +1,17 @@
+name: Install Redis
+description: Install redis-server for the Stream tests on Linux and macOS, and make their Redis lane fail without it.
+runs:
+  using: composite
+  steps:
+    - if: runner.os == 'Linux'
+      shell: bash
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y --no-install-recommends redis-server
+    - if: runner.os == 'macOS'
+      shell: bash
+      run: brew install redis
+    # Redis has no official Windows build, so the Redis lane skips there.
+    - if: runner.os != 'Windows'
+      shell: bash
+      run: echo "TEMPORAL_STREAMS_REQUIRE_REDIS=1" >> "$GITHUB_ENV"

.github/workflows/ci.yml

@@ -60,6 +60,7 @@ jobs:
         if: ${{ matrix.clippyLinter }}
       - run: poe build-develop
       - run: poe lint
+      - uses: ./.github/actions/install-redis
       - run: mkdir junit-xml
       - run: poe test ${{matrix.pytestExtraArgs}} -s --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}.xml
         timeout-minutes: 15
@@ -158,6 +159,7 @@ jobs:
       - run: poe build-develop-fips
       - name: Confirm the FIPS build is loaded
         run: uv run python -c "from temporalio.bridge import temporal_sdk_bridge; assert temporal_sdk_bridge.FIPS, 'not a FIPS build'"
+      - uses: ./.github/actions/install-redis
       - run: mkdir junit-xml
       - name: Run tests against the FIPS build
         run: |
@@ -237,6 +239,7 @@ jobs:
       - name: Check generation unchanged
         run: |
           [[ -z $(git status --porcelain temporalio tests) ]] || (git diff temporalio tests; echo "Protos changed"; exit 1)
+      - uses: ./.github/actions/install-redis
       - name: Test with protobuf 3.x
         run: poe test -s
         env:
@@ -270,6 +273,7 @@ jobs:
       - run: uv sync --all-extras
       - run: poe build-develop
       - run: poe lint
+      - uses: ./.github/actions/install-redis
       - run: mkdir junit-xml
       - run: poe test -s --junit-xml=junit-xml/latest-deps.xml
         timeout-minutes: 15
@@ -331,6 +335,7 @@ jobs:
         env:
           TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }}
           TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1
+      - uses: ./.github/actions/install-redis
       - run: mkdir junit-xml
       - run: poe test -s --workflow-environment envconfig --junit-xml=junit-xml/cloud.xml
         timeout-minutes: 15

temporalio/client/_client.py

@@ -238,8 +238,9 @@ async def connect(
             payload_limits: Warning thresholds for outbound payload/memo sizes. Over-threshold
                 fields are logged but still sent. Set a threshold to 0 to disable it.
             header_codec_behavior: Encoding behavior for headers sent by the client.
-            stream_backends: The Stream backends of this client and its Workers. Names must be
-                unique. With more than one backend, exactly one must have ``default=True``. See
+            stream_backends: The Stream backends of this client and its Workers, such as
+                :py:class:`temporalio.contrib.stream.RedisStreamBackend`. Names must be unique.
+                With more than one backend, exactly one must have ``default=True``. See
                 :py:class:`temporalio.common.StreamBackend`.
 
                 .. warning::

temporalio/common/_streams/_extension.py

@@ -0,0 +1,145 @@
+"""The Stream backends that call the temporalio-streams runtime in the bridge."""
+
+from __future__ import annotations
+
+import logging
+import weakref
+from collections.abc import Sequence
+from typing import Any
+
+import temporalio.bridge.streams
+import temporalio.client
+import temporalio.converter
+import temporalio.runtime
+import temporalio.service
+import temporalio.worker
+from temporalio.common._streams._backend import BackendCalls, _ExtensionCalls
+from temporalio.common._streams._errors import StreamConfigError
+from temporalio.common._streams._plugin import StreamBackend
+from temporalio.common._streams.proto import v1 as ext_pb
+
+logger = logging.getLogger(__name__)
+
+
+class ExtensionStreamBackend(StreamBackend):
+    """Base class of the Redis and memory backends, which call the temporalio-streams runtime.
+
+    .. warning::
+        This API is experimental.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        memory: bool,
+        config: ext_pb.RedisStreamConfig | ext_pb.MemoryStreamConfig,
+        runtime: temporalio.runtime.Runtime | None,
+        backend_name: str,
+        default: bool,
+    ) -> None:
+        """Create the backend over the bridge's Stream runtime, on Redis or in memory.
+
+        ``backend_name`` is the name that Stream handles store for this backend. ``default``
+        makes it the backend of a create with no backend name.
+        """
+        config.backend = backend_name
+        self._config = config.SerializeToString()
+        # On Redis, the clients of one Core runtime share one runtime object and its connections.
+        # The memory backend has one runtime object, so all its clients share its Streams.
+        self._backends: weakref.WeakKeyDictionary[
+            temporalio.runtime.Runtime, _ExtensionCalls
+        ] = weakref.WeakKeyDictionary()
+        self._memory: _ExtensionCalls | None = None
+        if memory:
+            self._memory = self._new_backend(
+                runtime or temporalio.runtime.Runtime.default(), True
+            )
+        super().__init__(
+            name,
+            backend_name=backend_name,
+            default=default,
+        )
+
+    def _new_backend(
+        self, runtime: temporalio.runtime.Runtime, memory: bool = False
+    ) -> _ExtensionCalls:
+        if self._memory is not None and not memory:
+            return self._memory
+        backend = self._backends.get(runtime)
+        if backend is None:
+            backend = _ExtensionCalls(
+                temporalio.bridge.streams.StreamRuntime(
+                    runtime._core_runtime, memory, self._config
+                )
+            )
+            self._backends[runtime] = backend
+        return backend
+
+    def _configure_client(
+        self, config: temporalio.client.ClientConfig
+    ) -> temporalio.client.ClientConfig:
+        config = super()._configure_client(config)
+        backend = self._new_backend(_runtime_of(config["service_client"]))
+        config["interceptors"] = [
+            *config.get("interceptors", []),
+            _StreamInterceptor(self, backend, config["data_converter"]),
+        ]
+        return config
+
+    def _backend_for(self, client: temporalio.client.Client) -> BackendCalls:
+        # The Stream runtime that this backend gave the client.
+        interceptor = self._interceptor_of(
+            client.config(active_config=True).get("interceptors", [])
+        )
+        if interceptor is None or interceptor.backend is None:
+            raise StreamConfigError(
+                f"The client has no {type(self).__name__}. Register it with "
+                "stream_backends=."
+            )
+        return interceptor.backend
+
+    def _interceptor_of(self, interceptors: Sequence[Any]) -> _StreamInterceptor | None:
+        for interceptor in interceptors:
+            if (
+                isinstance(interceptor, _StreamInterceptor)
+                and interceptor.plugin is self
+            ):
+                return interceptor
+        return None
+
+    async def shutdown(self) -> None:
+        """Close every open writer, including the client's.
+
+        Call it when the process stops writing to Streams.
+        """
+        failed = 0
+        for backend in list(self._backends.values()):
+            failed += await backend.close_writers()
+        if failed:
+            logger.warning("Stream shutdown left %d failed writers", failed)
+
+
+def _runtime_of(
+    service: temporalio.service.ServiceClient,
+) -> temporalio.runtime.Runtime:
+    # Find the runtime of a client's connection the same way the Worker does.
+    service = getattr(service, "worker_service_client", service)
+    runtime = getattr(service, "_runtime", None) or getattr(
+        getattr(service, "config", None), "runtime", None
+    )
+    return runtime or temporalio.runtime.Runtime.default()
+
+
+class _StreamInterceptor(temporalio.client.Interceptor, temporalio.worker.Interceptor):
+    # There is one for each client of the backend, with that client's Stream runtime. The
+    # client's Workers get it from the client.
+    def __init__(
+        self,
+        plugin: ExtensionStreamBackend,
+        backend: _ExtensionCalls | None,
+        converter: temporalio.converter.DataConverter | None,
+    ) -> None:
+        super().__init__()
+        self.plugin = plugin
+        self.backend = backend
+        self.converter = converter or temporalio.converter.DataConverter.default

temporalio/contrib/stream/__init__.py

@@ -0,0 +1,90 @@
+"""The Redis Stream backend.
+
+Register it on the client. A client creates a Stream with
+:py:meth:`temporalio.client.Client.create_stream`. ``backend=`` picks a backend that isn't the
+default::
+
+    from temporalio.contrib.stream import RedisStreamBackend
+
+    client = await Client.connect(
+        "localhost:7233",
+        stream_backends=[RedisStreamBackend("redis://localhost", default=True)],
+    )
+
+.. warning::
+    This package is experimental.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import timedelta
+
+from temporalio.common._streams._extension import ExtensionStreamBackend
+from temporalio.common._streams.proto import v1 as ext_pb
+
+
+class RedisStreamBackend(ExtensionStreamBackend):
+    """Sends Stream calls to Redis, through the temporalio-streams runtime in the bridge.
+
+    Register it on the client with ``stream_backends=``. A Worker uses its client's backends,
+    so one backend serves Workflow code, Activities and the client. The clients of the backend
+    that share a :py:class:`temporalio.runtime.Runtime` share one runtime object and its
+    connections, and so do their Workers. It registers the backend name ``redis``, and each
+    Stream handle that it creates stores that name. A Worker serves only one backend. A client
+    can have more than one, and ``backend=`` picks one.
+
+    .. warning::
+        This API is experimental.
+    """
+
+    def __init__(
+        self,
+        url: str | Sequence[str],
+        *,
+        cluster: bool = False,
+        key_prefix: str | None = None,
+        retention: timedelta | None = None,
+        safety_ttl: timedelta | None = None,
+        response_timeout: timedelta | None = None,
+        name: str = "redis",
+        default: bool = False,
+    ) -> None:
+        """Create the backend.
+
+        The backend connects to Redis at the first Stream call on a runtime, and the clients of
+        that runtime share the connection.
+
+        Args:
+            url: A Redis URL, or the seed URLs of a cluster. Credentials and TLS come from the
+                URL.
+            cluster: Whether the URLs are of a Redis cluster.
+            key_prefix: The prefix of every key. Defaults to ``temporal-streams``.
+            retention: The default Retention of a new Stream. Defaults to 7 days.
+            safety_ttl: How long an open Workflow-owned Stream is kept after its last write.
+                Defaults to 90 days.
+            response_timeout: The longest time that one Redis call can take. Defaults to 5 s.
+            name: The backend name that Stream handles store, so every process can route a
+                handle that it receives.
+            default: Make this the backend of a create with no backend name. With more than
+                one registered backend, exactly one must have it, or the client raises
+                ``StreamConfigError``.
+        """
+        config = ext_pb.RedisStreamConfig(
+            urls=[url] if isinstance(url, str) else list(url),
+            cluster=cluster,
+            key_prefix=key_prefix or "",
+        )
+        for field, value in (
+            ("retention", retention),
+            ("safety_ttl", safety_ttl),
+            ("response_timeout", response_timeout),
+        ):
+            if value is not None:
+                getattr(config, field).FromTimedelta(value)
+        super().__init__(
+            "temporalio.RedisStreamBackend", False, config, None, name, default
+        )
+
+
+__all__ = ["RedisStreamBackend"]

temporalio/contrib/stream/testing.py

@@ -0,0 +1,56 @@
+"""A Stream backend for tests, with no Redis.
+
+.. warning::
+    This module is experimental.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+import temporalio.runtime
+from temporalio.common._streams._extension import ExtensionStreamBackend
+from temporalio.common._streams.proto import v1 as ext_pb
+
+
+class MemoryStreamBackend(ExtensionStreamBackend):
+    """Keeps Streams in this process, with the same rules as Redis, for tests.
+
+    Register it on the client, as with ``RedisStreamBackend``. Every Worker and client that
+    shares one backend shares its Streams. The Streams are lost when the process ends. It
+    registers the backend name ``memory``. It has Redis's rules, so tests can swap the Store
+    without changing Workflow code.
+
+    .. warning::
+        This API is experimental.
+    """
+
+    def __init__(
+        self,
+        *,
+        retention: timedelta | None = None,
+        safety_ttl: timedelta | None = None,
+        runtime: temporalio.runtime.Runtime | None = None,
+        name: str = "memory",
+        default: bool = False,
+    ) -> None:
+        """Create the backend.
+
+        Args:
+            retention: The default Retention of a new Stream. Defaults to 7 days.
+            safety_ttl: How long an open Workflow-owned Stream is kept after its last write.
+                Defaults to 90 days.
+            runtime: The runtime of the backend's memory Store. Defaults to the default
+                runtime.
+            name: The backend name that Stream handles store.
+            default: Make this the backend of a create with no backend name. With more than
+                one registered backend, exactly one must have it.
+        """
+        config = ext_pb.MemoryStreamConfig()
+        if retention is not None:
+            config.retention.FromTimedelta(retention)
+        if safety_ttl is not None:
+            config.safety_ttl.FromTimedelta(safety_ttl)
+        super().__init__(
+            "temporalio.MemoryStreamBackend", True, config, runtime, name, default
+        )

temporalio/worker/_replayer.py

@@ -68,8 +68,8 @@ def __init__(
         most of the arguments. Most of the same arguments need to be passed to
         the replayer that were passed to the worker when the workflow originally
         ran. Pass ``stream_backends`` with the same backends as the Worker's
-        client. A replay makes no Stream call, so the Store doesn't need to be
-        reachable.
+        client, such as :py:class:`temporalio.contrib.stream.RedisStreamBackend`.
+        A replay makes no Stream call, so the Store doesn't need to be reachable.
 
         Note, unlike the worker, for the replayer the workflow_task_executor
         will default to a new thread pool executor with no max_workers set that