[Brook] Added stream topics, references and the store plugins.

temporalio/contrib/streams/__init__.py

@@ -10,17 +10,31 @@
 
 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. It withdraws only what that
    producer wrote in its earlier attempts, not the whole stream and not the
    records of other producers. 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`.
+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``
@@ -41,16 +55,20 @@
     StreamStorageError,
     StreamUnsupportedError,
 )
+from temporalio.contrib.streams._plugin import StreamStorePlugin
 from temporalio.contrib.streams._record import (
     Cursor,
     RecordKind,
     StreamRecord,
     Supersession,
 )
+from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef
+from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic
 
 __all__ = [
     "BEGINNING",
     "Cursor",
+    "DEFAULT_TOPIC",
     "END",
     "RecordKind",
     "StreamClosedError",
@@ -59,11 +77,16 @@
     "StreamExpiredError",
     "StreamNotFoundError",
     "StreamOutcomeUnknownError",
+    "StreamOwnerKind",
     "StreamProducerError",
     "StreamRecord",
     "StreamRecordError",
+    "StreamRef",
     "StreamRefusedError",
     "StreamStorageError",
+    "StreamStorePlugin",
+    "StreamTopic",
     "StreamUnsupportedError",
     "Supersession",
+    "topic",
 ]

temporalio/contrib/streams/_plugin.py

@@ -0,0 +1,201 @@
+"""The plugin a store configuration is, and how each context finds the store.
+
+A configuration class is registered once, on the client:
+``Client.connect(plugins=[RedisStreams(...)])``. Connecting the client asks
+Core for the store it names. Core serves that one store to the client's
+handles and to the Activities of every Worker built from the client, so an
+Activity's append and an outside read meet in the same place.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import TYPE_CHECKING, Any, TypeVar, cast
+
+import temporalio.bridge.client
+from temporalio.bridge.proto.streams import (
+    DeleteOwnerRequest,
+    StreamOwnerKind,
+    StreamStoreConfig,
+)
+from temporalio.bridge.streams import StreamCallFailure, StreamStore
+from temporalio.bridge.streams_generated import StreamService
+from temporalio.contrib.streams._errors import error_from_failure
+from temporalio.plugin import SimplePlugin
+
+if TYPE_CHECKING:
+    from temporalio.client import Client, ClientConfig
+    from temporalio.service import ConnectConfig, ServiceClient
+    from temporalio.worker import WorkerConfig
+
+__all__ = ["StreamStorePlugin"]
+
+T = TypeVar("T")
+
+
+class StreamStorePlugin(SimplePlugin):
+    """The base class of the store configuration classes.
+
+    Use :class:`temporalio.contrib.streams.redis.RedisStreams` or
+    :class:`temporalio.contrib.streams.memory.MemoryStreams`. One instance
+    is one store, on one client connection, since Core asks about each
+    stream's owner over the connection that connected the store. A second
+    connection, such as another namespace's endpoint, needs its own instance.
+    Registering a second, different store on the same client raises
+    ``ValueError``, because a handle could otherwise pick either.
+    """
+
+    def __init__(self, name: str, config: StreamStoreConfig) -> None:
+        """Name the plugin and the store Core connects to."""
+        super().__init__(name)
+        self._config = config
+        self._service: StreamService | None = None
+        self._store: StreamStore | None = None
+        self._connecting: asyncio.Lock | None = None
+        self._service_client: ServiceClient | None = None
+        # Whether Client.connect connected the store, which a Worker needs.
+        self._connected_by_connect = False
+
+    def configure_client(self, config: ClientConfig) -> ClientConfig:
+        """Register this store on the client.
+
+        Raises:
+            ValueError: Another store is already registered on the client.
+        """
+        for plugin in config.get("plugins", []):
+            if isinstance(plugin, StreamStorePlugin) and plugin is not self:
+                raise ValueError(_two_stores(plugin, self))
+        return super().configure_client(config)
+
+    async def connect_service_client(
+        self,
+        config: ConnectConfig,
+        next: Callable[[ConnectConfig], Awaitable[ServiceClient]],
+    ) -> ServiceClient:
+        """Connect the client, then the store, so Workers built from it have one.
+
+        A lazy client connects here too, since Core needs its connection to
+        ask about a stream's owner.
+
+        Raises:
+            temporalio.contrib.streams.StreamError: Core could not reach the
+                store.
+        """
+        service_client = await next(config)
+        await self._connected(service_client)
+        self._connected_by_connect = True
+        return service_client
+
+    def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
+        """Check that the Worker's client carries this store.
+
+        Raises:
+            ValueError: The plugin is not registered on the Worker's client,
+                or that client was not made with ``Client.connect``. A
+                client made otherwise is refused even after one of its handles
+                connected the store.
+        """
+        config = super().configure_worker(config)
+        client = config["client"]  # type:ignore[reportTypedDictNotRequiredAccess]
+        if (
+            self not in client.config()["plugins"]
+            or not self._connected_by_connect
+            or client.service_client is not self._service_client
+        ):
+            raise ValueError(
+                f"stream store {self.name()!r} must be registered on the Worker's "
+                "client, as Client.connect(..., plugins=[store])"
+            )
+        return config
+
+    async def delete_workflow_streams(self, namespace: str, workflow_id: str) -> int:
+        """Delete every stream of ``workflow_id`` in ``namespace``.
+
+        An admin helper for cleanup and compliance requests. Deleting a
+        Workflow does not delete its streams. Their retention bounds what is
+        left, and this removes it at once, across every run chain of the
+        Workflow id. It asks Temporal nothing, so it can run after the
+        Workflow is gone. On Redis it walks the keyspace, so run it rarely,
+        not on a hot path.
+
+        Delete only after the chain closed and its producers stopped. A late
+        append writes a fresh stream with no dedupe state and no close flag.
+
+        Returns:
+            How many store keys were deleted.
+
+        Raises:
+            ValueError: No client connected this store yet.
+            temporalio.contrib.streams.StreamStorageError: The store failed;
+                some keys may be deleted.
+        """
+        if self._service is None:
+            raise ValueError(f"no client has connected stream store {self.name()!r}")
+        request = DeleteOwnerRequest(
+            namespace=namespace,
+            owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW,
+            workflow_id=workflow_id,
+        )
+        return (await call(self._service.delete_owner(request))).deleted
+
+    async def _service_for(self, client: Client) -> StreamService:
+        """The stream service, connecting the store if no client did yet.
+
+        Raises:
+            ValueError: Another client connection already connected the store.
+        """
+        await self._connected(client.service_client)
+        assert self._service is not None
+        return self._service
+
+    async def _connected(self, service_client: ServiceClient) -> None:
+        if self._service_client is service_client:
+            return
+        if self._connecting is None:
+            self._connecting = asyncio.Lock()
+        async with self._connecting:
+            if self._service_client is service_client:
+                return
+            if self._service_client is not None:
+                raise ValueError(
+                    f"stream store {self.name()!r} serves one client connection, "
+                    "and another one already connected it; give each connection "
+                    "a store instance of its own"
+                )
+            bridge: temporalio.bridge.client.Client = await cast(
+                Any, service_client
+            )._connected_client()
+            self._store = await call(StreamStore.connect(bridge, self._config))
+            self._service = StreamService(self._store)
+            self._service_client = service_client
+
+
+async def call(pending: Awaitable[T]) -> T:
+    """Await a stream call, raising Core's failure as the error a caller catches."""
+    try:
+        return await pending
+    except StreamCallFailure as failure:
+        raise error_from_failure(failure.failure) from None
+
+
+def _two_stores(held: StreamStorePlugin, new: StreamStorePlugin) -> str:
+    return (
+        f"stream store {held.name()!r} is already registered, so {new.name()!r} "
+        "cannot be; register one store per client"
+    )
+
+
+def store_for_client(client: Client) -> StreamStorePlugin:
+    """The store registered on ``client``.
+
+    Raises:
+        ValueError: No store is registered on the client.
+    """
+    for plugin in client.config()["plugins"]:
+        if isinstance(plugin, StreamStorePlugin):
+            return plugin
+    raise ValueError(
+        "no stream store is registered on this client; pass one as a plugin, for "
+        "example Client.connect(..., plugins=[RedisStreams(...)])"
+    )

temporalio/contrib/streams/_ref.py

@@ -0,0 +1,95 @@
+"""A reference to one stream that crosses a process boundary as data.
+
+A handle is bound to a client and its store, so it cannot be a Workflow
+argument or an Activity result. A :class:`StreamRef` can: it names the owner
+and the topic, nothing more, and whoever receives it opens the stream on the
+store its own client carries. It carries no cursor, because a position
+belongs to a reader, and no store name.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from dataclasses import dataclass
+from typing import Any, Literal
+
+from temporalio.contrib.streams._topic import (
+    DEFAULT_TOPIC,
+    StreamTopic,
+    resolve_topic,
+)
+
+__all__ = ["StreamOwnerKind", "StreamRef"]
+
+StreamOwnerKind = Literal["workflow"]
+"""The owner kinds this release opens. Only a Workflow owns a stream.
+
+:attr:`StreamRef.kind` is a plain string, so a ref that names an owner kind
+a later release adds still reaches the ref's own check when it is decoded,
+instead of failing the converter's type check.
+"""
+
+
+@dataclass(frozen=True)
+class StreamRef:
+    """One stream, named by its owner and its topic.
+
+    ``kind`` says what owns the stream; only ``"workflow"`` exists in this
+    release. A Workflow ref carries ``workflow_id``. The stream belongs to
+    the Workflow's run chain, and a read on it follows the chain across
+    Continue-as-New and ends only when the chain ends. ``topic`` is the topic
+    a handle opened from the ref addresses when a call names none.
+
+    ``chain_run_id`` picks which chain: the one that holds that run. A
+    Workflow id can start a new chain once the old one closed, and without
+    ``chain_run_id`` a ref names the newest chain. It never stops a read at
+    that run's end.
+
+    The default data converter carries it as JSON, so it can be a Workflow
+    argument or an Activity result.
+    """
+
+    kind: str
+    workflow_id: str
+    topic: str = DEFAULT_TOPIC
+    chain_run_id: str | None = None
+
+    def __post_init__(self) -> None:
+        """Refuse a ref this release cannot open.
+
+        Raises:
+            ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or
+                ``topic`` is empty.
+        """
+        if self.kind != "workflow":
+            raise ValueError(f"unknown StreamRef kind {self.kind!r}")
+        if not self.workflow_id:
+            raise ValueError("a Workflow StreamRef needs a workflow_id")
+        if not self.topic:
+            raise ValueError("a StreamRef needs a topic name")
+
+    @classmethod
+    def for_workflow(
+        cls,
+        workflow_id: str,
+        *,
+        topic: str | StreamTopic[Any] | None = None,
+        chain_run_id: str | None = None,
+    ) -> StreamRef:
+        """A ref to ``topic`` of ``workflow_id``'s stream.
+
+        Without ``topic`` it names
+        :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. Without
+        ``chain_run_id`` it names the Workflow id's newest chain, and with
+        one the chain that holds that run.
+        """
+        name, _ = resolve_topic(topic)
+        return cls("workflow", workflow_id, topic=name, chain_run_id=chain_run_id)
+
+    def with_topic(self, topic: str | StreamTopic[Any] | None) -> StreamRef:
+        """The same owner, naming ``topic`` instead.
+
+        ``None`` names :data:`temporalio.contrib.streams.DEFAULT_TOPIC`.
+        """
+        name, _ = resolve_topic(topic)
+        return dataclasses.replace(self, topic=name)