diff --git a/.github/actions/install-redis/action.yml b/.github/actions/install-redis/action.yml new file mode 100644 index 0000000000..0c96b66ec1 --- /dev/null +++ b/.github/actions/install-redis/action.yml @@ -0,0 +1,17 @@ +name: Install Redis +description: Install redis-server for the Stream tests, whose Redis lane then fails 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" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f498f08f09..7e179d471c 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/temporalio/client/_client.py b/temporalio/client/_client.py index f9e9a4e025..158fa016fe 100644 --- a/temporalio/client/_client.py +++ b/temporalio/client/_client.py @@ -232,8 +232,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.streams.StreamBackend`. .. warning:: diff --git a/temporalio/contrib/stream/__init__.py b/temporalio/contrib/stream/__init__.py new file mode 100644 index 0000000000..f61b3f5e80 --- /dev/null +++ b/temporalio/contrib/stream/__init__.py @@ -0,0 +1,91 @@ +"""The Redis Stream backend. + +Register it on the client. A client creates a Stream with +:py:attr:`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.streams._extension import ExtensionStreamBackend +from temporalio.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 its ``name``, ``redis`` by default, and + each Stream ref that it creates keeps 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 Stream bound to a Workflow is kept after its last + write. Defaults to 90 days. A standalone Stream, the only kind that this SDK + creates, expires one Retention after its last write instead. + response_timeout: The longest time that one Redis call can take. Defaults to 5 s. + name: The backend name that Stream refs store, so every process can route a + ref 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"] diff --git a/temporalio/contrib/stream/testing.py b/temporalio/contrib/stream/testing.py new file mode 100644 index 0000000000..cc9edb9e44 --- /dev/null +++ b/temporalio/contrib/stream/testing.py @@ -0,0 +1,57 @@ +"""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.streams._extension import ExtensionStreamBackend +from temporalio.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 its ``name``, ``memory`` by default. 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 Stream bound to a Workflow is kept after its last + write. Defaults to 90 days. A standalone Stream, the only kind that this SDK + creates, expires one Retention after its last write instead. + runtime: The runtime of the backend's memory Store. Defaults to the default + runtime. + name: The backend name that Stream refs 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 + ) diff --git a/temporalio/streams/_extension.py b/temporalio/streams/_extension.py new file mode 100644 index 0000000000..a0986b3dee --- /dev/null +++ b/temporalio/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.streams._backend import BackendCalls, _ExtensionCalls +from temporalio.streams._errors import StreamConfigError +from temporalio.streams._plugin import StreamBackend +from temporalio.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 refs 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 diff --git a/temporalio/worker/_replayer.py b/temporalio/worker/_replayer.py index 7f103fef9e..2d0902e9e8 100644 --- a/temporalio/worker/_replayer.py +++ b/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 diff --git a/tests/contrib/streams/conftest.py b/tests/contrib/streams/conftest.py new file mode 100644 index 0000000000..fd7d97bce5 --- /dev/null +++ b/tests/contrib/streams/conftest.py @@ -0,0 +1,73 @@ +import os +import shutil +import socket +import subprocess +import time +import uuid +from collections.abc import AsyncIterator, Iterator + +import pytest + +from temporalio.client import Client +from temporalio.contrib.stream import RedisStreamBackend +from temporalio.contrib.stream.testing import MemoryStreamBackend +from temporalio.streams import StreamBackend + + +def _free_port() -> int: + # The OS gives a port that is free now, so the scratch server takes no port of another + # service. + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +# One scratch server serves every test module, and each test uses its own key prefix. To use a +# known port, set TEMPORAL_STREAMS_REDIS_URL to your own server. +@pytest.fixture(scope="session") +def redis_url() -> Iterator[str]: + given = os.environ.get("TEMPORAL_STREAMS_REDIS_URL") + if given: + yield given + return + server = shutil.which("redis-server") + if server is None: + # CI sets this where it installs Redis, so the Redis lane can't skip there unseen. + if os.environ.get("TEMPORAL_STREAMS_REQUIRE_REDIS"): + pytest.fail("redis-server is not installed") + pytest.skip("redis-server is not installed") + port = _free_port() + process = subprocess.Popen( + [server, "--port", str(port), "--save", "", "--appendonly", "no"], + stdout=subprocess.DEVNULL, + ) + try: + for _ in range(100): + try: + socket.create_connection(("127.0.0.1", port), timeout=0.1).close() + break + except OSError: + time.sleep(0.05) + yield f"redis://127.0.0.1:{port}" + finally: + process.terminate() + process.wait() + + +@pytest.fixture(params=["memory", "redis"]) +async def stream_client( + client: Client, request: pytest.FixtureRequest +) -> AsyncIterator[Client]: + # Every live test runs on the memory Store and on Redis, since both must have the same + # rules. The backend is the default, so a create with no backend name uses it. + backend: StreamBackend + if request.param == "memory": + backend = MemoryStreamBackend(default=True) + else: + redis_url = request.getfixturevalue("redis_url") + backend = RedisStreamBackend( + redis_url, key_prefix=f"t-{uuid.uuid4()}", default=True + ) + config = client.config() + config["stream_backends"] = [backend] + yield Client(**config) diff --git a/tests/contrib/streams/test_stream_client.py b/tests/contrib/streams/test_stream_client.py index f731c6c9c2..dfda0625ec 100644 --- a/tests/contrib/streams/test_stream_client.py +++ b/tests/contrib/streams/test_stream_client.py @@ -1,13 +1,18 @@ import asyncio +import dataclasses import uuid from dataclasses import dataclass +from datetime import timedelta from typing import Any import pytest from temporalio import streams, workflow -from temporalio.client import Client +from temporalio.api.enums.v1 import StreamErrorReason +from temporalio.client import Client, Stream +from temporalio.contrib.stream.testing import MemoryStreamBackend from temporalio.converter import DataConverter +from temporalio.streams.system import models from tests.contrib.streams.fake_extension import fake_client from tests.helpers import new_worker @@ -153,3 +158,270 @@ async def test_query_and_workflow_results_give_refs(client: Client): "from the query result", "from the Workflow result", ] + + +async def test_async_with_closes_a_client_stream(stream_client: Client): + async with await stream_client.create_stream[str](f"block-{uuid.uuid4()}") as done: + await done.publish("a") + failed = await stream_client.create_stream[str](f"block-{uuid.uuid4()}") + with pytest.raises(ValueError): + async with failed: + raise ValueError("turn failed") + early = await stream_client.create_stream[str](f"block-{uuid.uuid4()}") + async with early: + await early.close() + cancelled = await stream_client.create_stream[str](f"block-{uuid.uuid4()}") + with pytest.raises(asyncio.CancelledError): + async with cancelled: + raise asyncio.CancelledError() + for stream in (done, failed, early, cancelled): + # A read of a closed Stream ends once it gave the last item, with no wait. + items = await asyncio.wait_for( + _records(stream_client.get_stream[str](stream.ref.id)), 5 + ) + assert items == (["a"] if stream is done else []) + with pytest.raises(streams.StreamClosedError): + await stream_client.get_stream[str](stream.ref.id).publish("late") + + +async def _records(stream: Stream[Any]) -> list[Any]: + return [r.data async for r in stream.subscribe() if isinstance(r, streams.Record)] + + +async def test_create_fails_on_a_stream_that_exists(stream_client: Client): + stream_id = f"again-{uuid.uuid4()}" + first = await stream_client.create_stream[str]( + stream_id, retention=timedelta(days=1) + ) + await first.publish("a") + with pytest.raises(streams.StreamAlreadyExistsError) as exists: + await stream_client.create_stream[str](stream_id) + assert exists.value.reason == StreamErrorReason.STREAM_ERROR_REASON_ALREADY_EXISTS + # The create-or-get ignores the other Retention and returns the Stream. + again = await stream_client.create_or_get_stream[str]( + stream_id, retention=timedelta(days=2) + ) + assert again.ref == first.ref + assert await again.get_offset() == await first.get_offset() > 0 + await first.close() + # A closed Stream fails both creates with CLOSED, which comes before ALREADY_EXISTS. + for create in (stream_client.create_stream, stream_client.create_or_get_stream): + with pytest.raises(streams.StreamClosedError) as closed: + await create[str](stream_id) + assert closed.value.reason == StreamErrorReason.STREAM_ERROR_REASON_CLOSED + + +async def test_a_create_retry_with_the_same_request_id_succeeds(stream_client: Client): + # A retry after a lost answer sends the same request id, so it isn't a conflict. + [backend] = stream_client.config(active_config=True)["stream_backends"] + service = backend._backend_for(stream_client).service(stream_client.namespace) + request = models.CreateStreamRequest( + stream_ref=models.StreamRef(id=f"retry-{uuid.uuid4()}"), request_id="create-1" + ) + await service._stream_service_create_stream(request, object) + await service._stream_service_create_stream(request, object) + with pytest.raises(streams.StreamAlreadyExistsError): + await service._stream_service_create_stream( + dataclasses.replace(request, request_id="create-2"), object + ) + + +async def test_publish_to_a_stream_that_nobody_created_raises(stream_client: Client): + stream = stream_client.get_stream[str](f"missing-{uuid.uuid4()}") + with pytest.raises(streams.StreamNotFoundError) as missing: + await stream.publish("lost") + assert missing.value.reason == StreamErrorReason.STREAM_ERROR_REASON_NOT_FOUND + + +async def test_next_offset_resumes_after_the_last_item(stream_client: Client): + stream = await stream_client.create_stream[str](f"resume-{uuid.uuid4()}") + last = await stream.publish("a", "b") + tokens = stream.subscribe(wait_new_records=False) + assert tokens.next_offset == 0 + assert [item.data async for item in tokens if isinstance(item, streams.Record)] == [ + "a", + "b", + ] + assert tokens.next_offset == last + 1 + new = stream.subscribe(offset=streams.END, wait_new_records=False) + assert new.next_offset is None + assert [item async for item in new] == [] + assert new.next_offset == last + 1 + await stream.publish("c") + resumed = stream.subscribe(offset=tokens.next_offset, wait_new_records=False) + assert [ + item.data async for item in resumed if isinstance(item, streams.Record) + ] == ["c"] + + +async def test_a_waiting_read_ends_when_the_stream_closes(stream_client: Client): + stream = await stream_client.create_stream[str](f"wakes-{uuid.uuid4()}") + await stream.publish("a") + reader = asyncio.create_task(_records(stream_client.get_stream[str](stream.ref.id))) + await asyncio.sleep(0.2) + assert not reader.done() + await stream.close() + assert await asyncio.wait_for(reader, 10) == ["a"] + # A read from the end of a closed Stream has no item left, so it answers at once. + tail = stream_client.get_stream[str](stream.ref.id).subscribe(offset=streams.END) + assert await asyncio.wait_for(_items(tail), 5) == [] + + +async def _items(read: streams.Subscription[Any]) -> list[Any]: + return [item async for item in read] + + +async def test_last_reads_the_newest_records_then_follows(stream_client: Client): + stream = await stream_client.create_stream[str](f"last-{uuid.uuid4()}") + await stream.publish("a", "b", "c") + # A chat view shows the last few Records first, then waits for new ones. + tail = stream.subscribe(last=2) + assert tail.next_offset is None + shown = [await anext(tail), await anext(tail)] + await stream.publish("d") + shown.append(await anext(tail)) + await tail.aclose() + assert [item.data for item in shown if isinstance(item, streams.Record)] == [ + "b", + "c", + "d", + ] + newest = stream.subscribe(last=1, wait_new_records=False) + assert [item.data async for item in newest if isinstance(item, streams.Record)] == [ + "d" + ] + with pytest.raises(ValueError, match="no offset"): + stream.subscribe(last=2, offset=0) + + +async def test_payload_type_of_the_call_decodes_each_value(stream_client: Client): + stream = await stream_client.create_stream[Turn](f"typed-{uuid.uuid4()}") + await stream.publish(Turn("model")) + turns = stream_client.get_stream[Turn](stream.ref.id) + dicts = stream_client.get_stream[dict[str, Any]](stream.ref.id) + for reader, expected in ( + (turns, Turn("model")), + (dicts, {"kind": "model", "text": ""}), + ): + read = reader.subscribe(wait_new_records=False) + assert [r.data async for r in read if isinstance(r, streams.Record)] == [ + expected + ] + + +async def test_record_that_fails_to_decode_raises_and_resumes_past_it( + stream_client: Client, +): + stream_id = f"poison-{uuid.uuid4()}" + writer = await stream_client.create_stream[str](stream_id) + bad = await writer.publish("not a turn") + reader = stream_client.get_stream[Turn](stream_id) + await reader.publish(Turn("after")) + read = reader.subscribe(wait_new_records=False) + with pytest.raises(streams.StreamError, match=f"offset {bad}"): + await anext(read) + assert read.next_offset == bad + 1 + resumed = reader.subscribe(offset=read.next_offset, wait_new_records=False) + assert [ + item.data async for item in resumed if isinstance(item, streams.Record) + ] == [Turn("after")] + + +async def test_record_over_the_store_limit_raises_too_large(stream_client: Client): + # The client's writer splits a batch at the Store's limit, so only one Record can be over it. + stream = await stream_client.create_stream[str](f"large-{uuid.uuid4()}") + with pytest.raises(streams.StreamTooLargeError) as refused: + await stream.publish("x" * 2_000_000) + assert refused.value.reason == StreamErrorReason.STREAM_ERROR_REASON_TOO_LARGE + assert refused.value.non_retryable + assert await stream.get_offset() == 0 + + +async def test_read_keeps_no_offsets_of_client_records(stream_client: Client): + # Nothing withdraws a client's Records, so a long UI read of them keeps no state. + stream = await stream_client.create_stream[str](f"chat-{uuid.uuid4()}") + await stream.publish("a", "b", "c") + read = stream.subscribe(wait_new_records=False) + assert [item.data async for item in read if isinstance(item, streams.Record)] == [ + "a", + "b", + "c", + ] + assert read._shown == {} + + +async def test_create_of_an_expired_stream_raises(stream_client: Client): + # An open standalone Stream expires one Retention after its last write. + stream_id = f"expired-{uuid.uuid4()}" + retention = timedelta(milliseconds=200) + stream = await stream_client.create_stream[str](stream_id, retention=retention) + await stream.publish("old") + await asyncio.sleep(0.5) + with pytest.raises(streams.StreamExpiredError): + await stream_client.create_or_get_stream[str](stream_id, retention=retention) + + +async def test_named_producer_fences_its_earlier_epoch(stream_client: Client): + stream = await stream_client.create_stream[str](f"named-{uuid.uuid4()}") + # A server that restarts writes as the same name with a higher epoch, so the zombie that it + # replaces can't write after it. + before = stream.producer("ingest", epoch=1) + await before.publish("before restart") + after = stream.producer("ingest", epoch=2) + await after.publish("after restart") + with pytest.raises(streams.StreamFencedError): + await before.publish("zombie") + # The earlier epoch's Records are real messages, so nothing withdraws them. + items = [item async for item in stream.subscribe(wait_new_records=False)] + assert [item.kind for item in items] == [streams.ItemKind.RECORD] * 2 + with pytest.raises(ValueError): + stream.producer("ingest", epoch=0) + + +async def test_clients_of_one_runtime_share_the_backend_calls(stream_client: Client): + # A service that makes a client for each tenant keeps one set of Redis connections. + other = Client(**stream_client.config()) + [backend] = stream_client.config(active_config=True)["stream_backends"] + assert backend._backend_for(other) is backend._backend_for(stream_client) + stream = await other.create_stream[str](f"shared-{uuid.uuid4()}") + await stream.publish("from the other client") + again = stream_client.get_stream[str](stream.ref.id) + items = [item async for item in again.subscribe(wait_new_records=False)] + assert [item.data for item in items if isinstance(item, streams.Record)] == [ + "from the other client" + ] + + +async def test_default_is_native_with_no_default_backend(client: Client): + config = client.config() + config["stream_backends"] = [MemoryStreamBackend()] + one = Client(**config) + with pytest.raises(streams.StreamConfigError, match="default is native"): + one.get_stream[str]("d-2") + with pytest.raises(streams.StreamConfigError, match="default is native"): + await one.create_stream[str]("d-2") + with pytest.raises(streams.StreamConfigError, match="default is native"): + one.to_stream(streams.StreamRef(id="d-2")) + on_memory = one.get_stream[str]("d-2", backend="memory") + assert on_memory.ref.backend == "memory" + created = await one.create_stream[str]("d-2", backend="memory") + assert created.ref.backend == "memory" + + +async def test_stream_takes_the_named_backend_or_the_default(client: Client): + config = client.config() + config["stream_backends"] = [ + MemoryStreamBackend(name="a", default=True), + MemoryStreamBackend(name="b"), + ] + two = Client(**config) + created = await two.create_stream[str]("moving", backend="b") + await created.publish("on b") + on_b = two.get_stream[str]("moving", backend="b") + assert on_b.ref == created.ref and await on_b.get_offset() > 0 + assert two.get_stream[str]("moving").ref.backend == "a" + # A ref with no backend name uses the default backend, which doesn't have the Stream. + with pytest.raises(streams.StreamNotFoundError): + await two.to_stream(streams.StreamRef(id="moving")).get_offset() + with pytest.raises(ValueError): + two.get_stream[str]("")