diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index f365744292..095d981512 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/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", ] diff --git a/temporalio/contrib/streams/_plugin.py b/temporalio/contrib/streams/_plugin.py new file mode 100644 index 0000000000..737f0de1c4 --- /dev/null +++ b/temporalio/contrib/streams/_plugin.py @@ -0,0 +1,201 @@ +"""The plugin a store configuration is, and how each context finds the store. + +A configuration class is registered once, on the client: +``Client.connect(plugins=[RedisStreams(...)])``. Connecting the client asks +Core for the store it names. Core serves that one store to the client's +handles and to the Activities of every Worker built from the client, so an +Activity's append and an outside read meet in the same place. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from typing import TYPE_CHECKING, Any, TypeVar, cast + +import temporalio.bridge.client +from temporalio.bridge.proto.streams import ( + DeleteOwnerRequest, + StreamOwnerKind, + StreamStoreConfig, +) +from temporalio.bridge.streams import StreamCallFailure, StreamStore +from temporalio.bridge.streams_generated import StreamService +from temporalio.contrib.streams._errors import error_from_failure +from temporalio.plugin import SimplePlugin + +if TYPE_CHECKING: + from temporalio.client import Client, ClientConfig + from temporalio.service import ConnectConfig, ServiceClient + from temporalio.worker import WorkerConfig + +__all__ = ["StreamStorePlugin"] + +T = TypeVar("T") + + +class StreamStorePlugin(SimplePlugin): + """The base class of the store configuration classes. + + Use :class:`temporalio.contrib.streams.redis.RedisStreams` or + :class:`temporalio.contrib.streams.memory.MemoryStreams`. One instance + is one store, on one client connection, since Core asks about each + stream's owner over the connection that connected the store. A second + connection, such as another namespace's endpoint, needs its own instance. + Registering a second, different store on the same client raises + ``ValueError``, because a handle could otherwise pick either. + """ + + def __init__(self, name: str, config: StreamStoreConfig) -> None: + """Name the plugin and the store Core connects to.""" + super().__init__(name) + self._config = config + self._service: StreamService | None = None + self._store: StreamStore | None = None + self._connecting: asyncio.Lock | None = None + self._service_client: ServiceClient | None = None + # Whether Client.connect connected the store, which a Worker needs. + self._connected_by_connect = False + + def configure_client(self, config: ClientConfig) -> ClientConfig: + """Register this store on the client. + + Raises: + ValueError: Another store is already registered on the client. + """ + for plugin in config.get("plugins", []): + if isinstance(plugin, StreamStorePlugin) and plugin is not self: + raise ValueError(_two_stores(plugin, self)) + return super().configure_client(config) + + async def connect_service_client( + self, + config: ConnectConfig, + next: Callable[[ConnectConfig], Awaitable[ServiceClient]], + ) -> ServiceClient: + """Connect the client, then the store, so Workers built from it have one. + + A lazy client connects here too, since Core needs its connection to + ask about a stream's owner. + + Raises: + temporalio.contrib.streams.StreamError: Core could not reach the + store. + """ + service_client = await next(config) + await self._connected(service_client) + self._connected_by_connect = True + return service_client + + def configure_worker(self, config: WorkerConfig) -> WorkerConfig: + """Check that the Worker's client carries this store. + + Raises: + ValueError: The plugin is not registered on the Worker's client, + or that client was not made with ``Client.connect``. A + client made otherwise is refused even after one of its handles + connected the store. + """ + config = super().configure_worker(config) + client = config["client"] # type:ignore[reportTypedDictNotRequiredAccess] + if ( + self not in client.config()["plugins"] + or not self._connected_by_connect + or client.service_client is not self._service_client + ): + raise ValueError( + f"stream store {self.name()!r} must be registered on the Worker's " + "client, as Client.connect(..., plugins=[store])" + ) + return config + + async def delete_workflow_streams(self, namespace: str, workflow_id: str) -> int: + """Delete every stream of ``workflow_id`` in ``namespace``. + + An admin helper for cleanup and compliance requests. Deleting a + Workflow does not delete its streams. Their retention bounds what is + left, and this removes it at once, across every run chain of the + Workflow id. It asks Temporal nothing, so it can run after the + Workflow is gone. On Redis it walks the keyspace, so run it rarely, + not on a hot path. + + Delete only after the chain closed and its producers stopped. A late + append writes a fresh stream with no dedupe state and no close flag. + + Returns: + How many store keys were deleted. + + Raises: + ValueError: No client connected this store yet. + temporalio.contrib.streams.StreamStorageError: The store failed; + some keys may be deleted. + """ + if self._service is None: + raise ValueError(f"no client has connected stream store {self.name()!r}") + request = DeleteOwnerRequest( + namespace=namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=workflow_id, + ) + return (await call(self._service.delete_owner(request))).deleted + + async def _service_for(self, client: Client) -> StreamService: + """The stream service, connecting the store if no client did yet. + + Raises: + ValueError: Another client connection already connected the store. + """ + await self._connected(client.service_client) + assert self._service is not None + return self._service + + async def _connected(self, service_client: ServiceClient) -> None: + if self._service_client is service_client: + return + if self._connecting is None: + self._connecting = asyncio.Lock() + async with self._connecting: + if self._service_client is service_client: + return + if self._service_client is not None: + raise ValueError( + f"stream store {self.name()!r} serves one client connection, " + "and another one already connected it; give each connection " + "a store instance of its own" + ) + bridge: temporalio.bridge.client.Client = await cast( + Any, service_client + )._connected_client() + self._store = await call(StreamStore.connect(bridge, self._config)) + self._service = StreamService(self._store) + self._service_client = service_client + + +async def call(pending: Awaitable[T]) -> T: + """Await a stream call, raising Core's failure as the error a caller catches.""" + try: + return await pending + except StreamCallFailure as failure: + raise error_from_failure(failure.failure) from None + + +def _two_stores(held: StreamStorePlugin, new: StreamStorePlugin) -> str: + return ( + f"stream store {held.name()!r} is already registered, so {new.name()!r} " + "cannot be; register one store per client" + ) + + +def store_for_client(client: Client) -> StreamStorePlugin: + """The store registered on ``client``. + + Raises: + ValueError: No store is registered on the client. + """ + for plugin in client.config()["plugins"]: + if isinstance(plugin, StreamStorePlugin): + return plugin + raise ValueError( + "no stream store is registered on this client; pass one as a plugin, for " + "example Client.connect(..., plugins=[RedisStreams(...)])" + ) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py new file mode 100644 index 0000000000..e41c7cb7d8 --- /dev/null +++ b/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) diff --git a/temporalio/contrib/streams/_topic.py b/temporalio/contrib/streams/_topic.py new file mode 100644 index 0000000000..0f2ec562a5 --- /dev/null +++ b/temporalio/contrib/streams/_topic.py @@ -0,0 +1,108 @@ +"""Typed topic definitions. + +A topic is defined once, at module level, with the type its records decode +to, and the Workflow, its Activities and the client code all refer to that +one definition, the way they share a Signal or an Activity. A plain string +names a topic too, for a name decided at runtime; then the decode hint +travels as ``result_type=`` on each read. + +On the wire a topic is a string, and +:attr:`temporalio.contrib.streams.StreamRecord.topic` is that string. +""" + +from __future__ import annotations + +import unicodedata +from dataclasses import dataclass +from typing import Any, Generic, TypeVar, overload + +__all__ = ["DEFAULT_TOPIC", "StreamTopic", "resolve_topic", "topic"] + +T = TypeVar("T") + +DEFAULT_TOPIC = "output" +"""The topic a call addresses when it names none. + +It is an ordinary name, so naming it explicitly addresses the same topic. +""" + + +@dataclass(frozen=True) +class StreamTopic(Generic[T]): + """A topic of a stream, with the type its records decode to. + + Made with :func:`topic`. Pass it wherever a call takes a topic, and the + record and value types follow from it. + """ + + name: str + result_type: type[T] | None = None + """The type records decode to, or ``None`` for the converter's default.""" + + +@overload +def topic(name: str, result_type: type[T]) -> StreamTopic[T]: ... + + +@overload +def topic(name: str, result_type: None = None) -> StreamTopic[Any]: ... + + +def topic(name: str, result_type: type | None = None) -> StreamTopic[Any]: + """Define a topic. + + Args: + name: The topic's name, as it appears on every record. + result_type: The type records on this topic decode to. Without one, + the payload converter's default applies. + + Raises: + ValueError: ``name`` is empty, longer than 256 UTF-8 bytes, or holds + a control character. + """ + _check_name(name) + return StreamTopic(name, result_type) + + +def resolve_topic( + topic: str | StreamTopic[Any] | None = None, result_type: type | None = None +) -> tuple[str, type | None]: + """The name and decode hint a call means, from either form of topic. + + ``None`` means :data:`DEFAULT_TOPIC`. A handle calls this once at the + top of each call that takes a topic, so a definition and a string reach + the store the same way. + + Raises: + ValueError: A definition was given together with ``result_type``, + which would name two types for one topic, or the name is empty, + longer than 256 UTF-8 bytes, or holds a control character. + """ + if isinstance(topic, StreamTopic): + if result_type is not None: + raise ValueError( + f"topic {topic.name!r} already carries its type; do not pass " + "result_type= with a definition" + ) + name, result_type = topic.name, topic.result_type + elif topic is None: + name = DEFAULT_TOPIC + else: + name = topic + _check_name(name) + return name, result_type + + +_MAX_NAME_BYTES = 256 + + +def _check_name(name: str) -> None: + # The same limit in every SDK: the name goes into store keys, cursor + # hashes and Core's manifest budget, and a store may join names with a + # control character. + if not name: + raise ValueError("topic name must not be empty") + if len(name.encode()) > _MAX_NAME_BYTES: + raise ValueError(f"topic name must be at most {_MAX_NAME_BYTES} UTF-8 bytes") + if any(unicodedata.category(c) == "Cc" for c in name): + raise ValueError(f"topic name {name!r} must not hold a control character") diff --git a/temporalio/contrib/streams/memory.py b/temporalio/contrib/streams/memory.py new file mode 100644 index 0000000000..198a518bfd --- /dev/null +++ b/temporalio/contrib/streams/memory.py @@ -0,0 +1,24 @@ +"""Streams in this process's memory, for tests and local development.""" + +from __future__ import annotations + +from temporalio.bridge.proto.streams import MemoryStoreConfig, StreamStoreConfig +from temporalio.contrib.streams._plugin import StreamStorePlugin + +__all__ = ["MemoryStreams"] + + +class MemoryStreams(StreamStorePlugin): + """A store in this process's memory. + + Core holds it, so the client's handles and every Worker built from the + client share it. Nothing survives the process, and another process sees + none of it. + """ + + def __init__(self) -> None: + """Create the store's configuration.""" + super().__init__( + "temporalio.contrib.streams.MemoryStreams", + StreamStoreConfig(memory=MemoryStoreConfig()), + ) diff --git a/temporalio/contrib/streams/redis.py b/temporalio/contrib/streams/redis.py new file mode 100644 index 0000000000..7d0be4409a --- /dev/null +++ b/temporalio/contrib/streams/redis.py @@ -0,0 +1,79 @@ +"""Streams in the application's Redis.""" + +from __future__ import annotations + +from collections.abc import Sequence +from datetime import timedelta + +from temporalio.bridge.proto.streams import RedisStoreConfig, StreamStoreConfig +from temporalio.contrib.streams._plugin import StreamStorePlugin + +__all__ = ["RedisStreams"] + + +class RedisStreams(StreamStorePlugin): + """A store in a Redis or a Redis Cluster that Core connects to. + + One instance per Redis deployment. Pass it to the client as a plugin, + and open handles from the client with + :func:`temporalio.contrib.streams.get_stream_handle`. Redis 7.0 or later + is required, with ``maxmemory-policy noeviction``. + """ + + def __init__( + self, + url: str | Sequence[str], + *, + cluster: bool = False, + key_prefix: str = "temporal-streams", + retention: timedelta = timedelta(days=7), + blocking_reads_per_node: int | None = None, + response_timeout: timedelta | None = None, + ) -> None: + """Create the store's configuration. + + Args: + url: A ``redis://`` or ``rediss://`` URL, or the seed nodes of a + cluster. Credentials and TLS settings come from the URL. In + this release TLS verifies the server against the system's + root certificates only, with no custom CA and no client + certificate. + cluster: Whether ``url`` names a Redis Cluster. Every key of one + Workflow's streams shares a hash tag, so a cluster serves + them from one slot. + key_prefix: Prepended to every key, so streams can share a Redis + with other data and an ACL can scope them. + retention: How long a stream keeps a record, and how long after + its last write the stream itself lives. + blocking_reads_per_node: How many reads may wait on one node at + once, each on a connection of its own. Core's default when + unset. + response_timeout: How long a short call waits for its reply. + Core's default when unset. + + Raises: + ValueError: No URL was given, ``key_prefix`` is empty, or + ``retention`` is shorter than a millisecond. + """ + urls = [url] if isinstance(url, str) else list(url) + if not urls: + raise ValueError("RedisStreams needs a URL") + if not key_prefix: + raise ValueError("key_prefix must not be empty") + if retention < timedelta(milliseconds=1): + raise ValueError( + f"retention must be at least a millisecond, got {retention}" + ) + config = RedisStoreConfig( + urls=urls, + cluster=cluster, + key_prefix=key_prefix, + blocking_reads_per_node=blocking_reads_per_node or 0, + ) + config.retention.FromTimedelta(retention) + if response_timeout is not None: + config.response_timeout.FromTimedelta(response_timeout) + super().__init__( + "temporalio.contrib.streams.RedisStreams", + StreamStoreConfig(redis=config), + ) diff --git a/tests/contrib/streams/_support.py b/tests/contrib/streams/_support.py new file mode 100644 index 0000000000..de4da231a5 --- /dev/null +++ b/tests/contrib/streams/_support.py @@ -0,0 +1,48 @@ +"""What the stream tests share.""" + +from __future__ import annotations + +import asyncio +import uuid +from typing import Any + +from temporalio.client import Client +from temporalio.contrib.streams import StreamStorePlugin +from temporalio.converter import DataConverter + + +async def connect_with( + client: Client, + store: StreamStorePlugin | None, + converter: DataConverter | None = None, +) -> Client: + """A client of the test server that carries ``store``. + + It connects, since connecting is what hands the store to Workers built + from it. + """ + return await Client.connect( + client.service_client.config.target_host, + namespace=client.namespace, + data_converter=converter or client.data_converter, + plugins=[store] if store is not None else [], + ) + + +def new_workflow_id() -> str: + return f"streams-{uuid.uuid4().hex}" + + +async def read_all(records: Any, timeout: float = 10.0) -> list: + """Read until the stream ends, which is when its owner closes.""" + out = [] + + async def pull() -> None: + async for record in records: + out.append(record) + + try: + await asyncio.wait_for(pull(), timeout) + finally: + await records.aclose() + return out diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py index a3647d4eba..1161287e40 100644 --- a/tests/contrib/streams/test_public_surface.py +++ b/tests/contrib/streams/test_public_surface.py @@ -12,6 +12,7 @@ "decode_body", "encode_bodies", "error_from_failure", + "resolve_topic", } diff --git a/tests/contrib/streams/test_store_plugin.py b/tests/contrib/streams/test_store_plugin.py new file mode 100644 index 0000000000..039527c1c3 --- /dev/null +++ b/tests/contrib/streams/test_store_plugin.py @@ -0,0 +1,198 @@ +"""A store configuration registers as a client plugin, and Core connects to it.""" + +from __future__ import annotations + +import hashlib +import uuid +from datetime import timedelta + +import pytest + +from temporalio import activity +from temporalio.api.common.v1 import Payload +from temporalio.bridge.proto.streams import ( + AppendRecord, + AppendRequest, + LatestRequest, + NamedProducer, + StreamAddress, + StreamOwnerKind, +) +from temporalio.bridge.proto.streams.v1 import StreamRecordKind +from temporalio.client import Client +from temporalio.contrib.streams import StreamError, StreamNotFoundError +from temporalio.contrib.streams._plugin import call +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.contrib.streams.redis import RedisStreams +from temporalio.worker import Worker +from tests.contrib.streams._support import connect_with + + +async def test_one_store_per_client(client: Client): + first, second = MemoryStreams(), MemoryStreams() + config = client.config() + config["plugins"] = [first, second] + with pytest.raises(ValueError, match="is already registered"): + Client(**config) + # The same instance twice is one store, not two. + config["plugins"] = [first, first] + Client(**config) + + +async def test_a_worker_takes_the_store_of_its_client(client: Client): + store = MemoryStreams() + streams_client = await connect_with(client, store) + Worker(streams_client, task_queue=str(uuid.uuid4()), activities=[_noop]) + # The store on the client and the Worker configures the Worker twice. + with pytest.warns(UserWarning, match="same plugin type"): + Worker( + streams_client, + task_queue=str(uuid.uuid4()), + activities=[_noop], + plugins=[store], + ) + + +async def test_a_store_on_the_worker_alone_is_refused(client: Client): + with pytest.raises(ValueError, match="registered on the Worker's client"): + Worker(client, task_queue="tq", activities=[_noop], plugins=[MemoryStreams()]) + # A client made without connecting has not connected its store either. + config = client.config() + config["plugins"] = [MemoryStreams()] + with pytest.raises(ValueError, match="registered on the Worker's client"): + Worker(Client(**config), task_queue="tq", activities=[_noop]) + + +async def test_a_client_made_without_connecting_connects_its_store_on_use( + client: Client, +): + store = MemoryStreams() + config = client.config() + config["plugins"] = [store] + service = await store._service_for(Client(**config)) + # An owner that does not exist is Core's answer, so the call reached the store. + stream = StreamAddress( + namespace=client.namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=f"no-such-owner-{uuid.uuid4()}", + topic="out", + ) + with pytest.raises(StreamNotFoundError): + await call(service.latest(LatestRequest(stream=stream))) + + +async def test_a_store_core_cannot_reach_fails_the_connect(client: Client): + with pytest.raises(StreamError): + await connect_with( + client, + RedisStreams("redis://127.0.0.1:1", response_timeout=timedelta(seconds=1)), + ) + + +def test_redis_settings_go_into_the_store_config(): + store = RedisStreams( + ["redis://a:6379", "redis://b:6379"], + cluster=True, + key_prefix="app", + retention=timedelta(hours=2), + blocking_reads_per_node=4, + response_timeout=timedelta(seconds=3), + ) + config = store._config.redis + assert list(config.urls) == ["redis://a:6379", "redis://b:6379"] + assert config.cluster + assert config.key_prefix == "app" + assert config.retention.ToTimedelta() == timedelta(hours=2) + assert config.blocking_reads_per_node == 4 + assert config.response_timeout.ToTimedelta() == timedelta(seconds=3) + # Unset settings leave Core's defaults. + defaults = RedisStreams("redis://a:6379")._config.redis + assert defaults.blocking_reads_per_node == 0 + assert not defaults.HasField("response_timeout") + + +def test_redis_setting_mistakes_are_value_errors(): + with pytest.raises(ValueError, match="needs a URL"): + RedisStreams([]) + with pytest.raises(ValueError, match="key_prefix"): + RedisStreams("redis://a", key_prefix="") + with pytest.raises(ValueError, match="retention"): + RedisStreams("redis://a", retention=timedelta(microseconds=10)) + + +async def test_deleting_removes_what_a_workflow_stream_holds(client: Client): + store = MemoryStreams() + with pytest.raises(ValueError, match="no client has connected"): + await store.delete_workflow_streams(client.namespace, "wf") + streams_client = await connect_with(client, store) + service = await store._service_for(streams_client) + workflow_id = f"stream-owner-{uuid.uuid4()}" + await streams_client.start_workflow( + "StreamOwner", id=workflow_id, task_queue=f"unpolled-{uuid.uuid4()}" + ) + stream = address(client, workflow_id) + await call( + service.append( + AppendRequest( + stream=stream, + named=NamedProducer(producer_id="p", attempt=1), + sequence=1, + records=[ + AppendRecord( + kind=StreamRecordKind.STREAM_RECORD_KIND_DATA, + body=Payload(data=b"a"), + content_hash=hashlib.sha256(b"a").digest(), + ) + ], + digest=hashlib.sha256(b"batch").digest(), + ) + ) + ) + assert (await call(service.latest(LatestRequest(stream=stream)))).cursor + + assert await store.delete_workflow_streams(client.namespace, workflow_id) > 0 + assert (await call(service.latest(LatestRequest(stream=stream)))).cursor == "" + + +async def test_one_store_serves_one_client_connection(client: Client): + store = MemoryStreams() + first = await connect_with(client, store) + # Core asks about owners over the connection that connected the store, so a + # second connection would get the first one's answers. + with pytest.raises(ValueError, match="one client connection"): + await connect_with(client, store) + config = client.config() + config["plugins"] = [store] + with pytest.raises(ValueError, match="one client connection"): + await store._service_for(Client(**config)) + # A client copied from the connected one shares its connection. + copied = first.config() + assert await store._service_for(Client(**copied)) + + +async def test_a_worker_needs_a_client_made_with_connect_after_a_lazy_connect( + client: Client, +): + store = MemoryStreams() + config = client.config() + config["plugins"] = [store] + lazy = Client(**config) + # A handle call connects the store, but the client still wasn't connected + # through the plugin, so Workers built from it are refused all the same. + await store._service_for(lazy) + with pytest.raises(ValueError, match="registered on the Worker's client"): + Worker(lazy, task_queue=str(uuid.uuid4()), activities=[_noop]) + + +@activity.defn +async def _noop() -> None: + pass + + +def address(client: Client, workflow_id: str) -> StreamAddress: + return StreamAddress( + namespace=client.namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=workflow_id, + topic="out", + ) diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py new file mode 100644 index 0000000000..2f51d457b5 --- /dev/null +++ b/tests/contrib/streams/test_topics_refs.py @@ -0,0 +1,100 @@ +"""Typed topics and the stream reference that crosses process boundaries.""" + +from __future__ import annotations + +import pytest + +from temporalio.api.common.v1 import Payload +from temporalio.contrib.streams import ( + DEFAULT_TOPIC, + StreamRef, + StreamTopic, + topic, +) +from temporalio.contrib.streams._topic import ( + resolve_topic, +) +from temporalio.converter import DataConverter + +OUT = topic("out", dict) + + +def test_a_definition_carries_its_name_and_type(): + assert OUT == StreamTopic("out", dict) + assert resolve_topic(OUT) == ("out", dict) + assert resolve_topic("free", int) == ("free", int) + assert resolve_topic(None) == (DEFAULT_TOPIC, None) + + +def test_topic_mistakes_are_value_errors(): + with pytest.raises(ValueError): + topic("") + with pytest.raises(ValueError): + resolve_topic("") + with pytest.raises(ValueError, match="already carries its type"): + resolve_topic(OUT, dict) + + +def test_a_ref_names_a_workflow_stream(): + ref = StreamRef.for_workflow("wf", topic=OUT) + assert ref == StreamRef("workflow", "wf", topic="out") + assert StreamRef.for_workflow("wf").topic == DEFAULT_TOPIC + assert ref.with_topic(None).topic == DEFAULT_TOPIC + assert ref.with_topic("other").workflow_id == "wf" + + +def test_a_ref_round_trips_through_the_default_converter(): + ref = StreamRef.for_workflow("wf", topic=OUT, chain_run_id="run") + converter = DataConverter.default.payload_converter + payload = converter.to_payloads([ref])[0] + assert converter.from_payloads([payload], [StreamRef])[0] == ref + + +def test_a_ref_with_the_old_run_id_decodes_as_the_newest_chain(): + # The old field also ended a read at that run, which no read does now, + # so it is dropped rather than read as chain_run_id. + converter = DataConverter.default.payload_converter + payload = Payload( + metadata={"encoding": b"json/plain"}, + data=b'{"kind": "workflow", "workflow_id": "wf", "run_id": "r", "topic": "out"}', + ) + ref = converter.from_payloads([payload], [StreamRef])[0] + assert ref == StreamRef.for_workflow("wf", topic=OUT) + + +def test_a_ref_refuses_what_it_cannot_name(): + with pytest.raises(ValueError, match="unknown"): + StreamRef("nexus", "x") + with pytest.raises(ValueError): + StreamRef("workflow", "") + with pytest.raises(ValueError): + StreamRef("workflow", "wf", topic="") + + +def test_a_ref_from_a_later_release_decodes_to_the_kind_check(): + # The kind is a plain string, so the converter builds the ref and the ref + # itself refuses the kind, rather than the converter failing a type check. + converter = DataConverter.default.payload_converter + payload = Payload( + metadata={"encoding": b"json/plain"}, + data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}', + ) + with pytest.raises(ValueError, match="'activity'"): + converter.from_payloads([payload], [StreamRef]) + + +@pytest.mark.parametrize( + "name", + ["x" * 257, "é" * 129, "a\x1fb", "line\nbreak", "\x00"], +) +def test_a_topic_name_that_is_too_long_or_has_a_control_character_is_refused(name: str): + # Every store puts the name in keys, cursors and Core's manifest budget. + with pytest.raises(ValueError): + topic(name) + with pytest.raises(ValueError): + resolve_topic(name) + + +def test_a_topic_name_of_256_bytes_is_accepted(): + assert topic("x" * 256).name == "x" * 256 + assert resolve_topic("é" * 128)[0] == "é" * 128