diff --git a/temporalio/streams/providers/redis.py b/temporalio/streams/providers/redis.py index 46e60686e..335467179 100644 --- a/temporalio/streams/providers/redis.py +++ b/temporalio/streams/providers/redis.py @@ -54,19 +54,40 @@ - A workflow's streams are keyed by the chain's first run, so a handle follows continue-as-new by construction and ``run_id`` only decides whose close ends a read. +- An activity's own streams are one Redis stream per topic, keyed by the + namespace, the workflow id, empty for a standalone activity, the run the + activity execution belongs to, and the activity id. The run is the + workflow's for a workflow's activity and the activity's own for a + standalone one, so a retry writes to the same stream, which a reader sees + as ``SUPERSEDED``, and an id started again, in a new run, starts a new + one, as it does on the server-side provider. Inside the activity the run + is in its info; a handle opened outside without one describes the owner + and takes its current run. The owner is one key component joined with + ``/``, a character the chain keys percent-encode out of every id, so no + chain key and no key derived from one can name an activity's stream. + There is no input stream, because no workflow reads these, and no + staging, because an activity's append is visible as soon as the store + accepts it. A read ends when the owner is terminal and the retained tail + is delivered: a standalone activity is described directly, and a + workflow's activity through its workflow, whose close ends the read, as + does the activity leaving the pending set once its stream exists. An + activity that never wrote has no stream, so a read on it waits for the + workflow. Nothing gates an append after that, so a late attempt still + lands, and a read that has ended does not see it. - A task's publishes are staged as one batch. The transport's own per-task batch limits are lifted for this provider, because a synchronous publish cannot wait for the worker to stage a full batch; a batch it cannot stage fails the task. - Retention is trimming, with no consumer floor. By default a record older than :data:`DEFAULT_RETENTION`, seven days, is trimmed by the next append - the provider makes to its key, whatever any reader has reached; - ``retention=None`` turns the age trim off, and ``max_len`` adds a count cap - that is off by default. A replay that reaches a recorded range the trim - removed fails its Workflow Task, an outside cursor below the trim is - refused, and a fully trimmed topic reads as empty. A run that has to replay - cold after seven days of consuming fails, so a long-lived consumer - continues as new inside the window, or is configured with a longer one. + the provider makes to its key, an activity's stream included, whatever any + reader has reached; ``retention=None`` turns the age trim off, and + ``max_len`` adds a count cap that is off by default. A replay that reaches + a recorded range the trim removed fails its Workflow Task, an outside + cursor below the trim is refused, and a fully trimmed topic reads as + empty. A run that has to replay cold after seven days of consuming fails, + so a long-lived consumer continues as new inside the window, or is + configured with a longer one. - Every record carries the SHA-256 of its converted body under ``temporal.io/content-hash``, stamped before the payload codec runs, and the @@ -96,15 +117,16 @@ import re import time from collections.abc import AsyncGenerator, Awaitable, Callable, Coroutine, Sequence -from dataclasses import replace +from dataclasses import dataclass, replace from datetime import timedelta from typing import Any, Final, Generic, TypeVar, get_args +from urllib.parse import quote from google.protobuf.message import DecodeError from temporalio import workflow from temporalio.api.common.v1 import Payload -from temporalio.client import Client, WorkflowExecutionStatus +from temporalio.client import ActivityExecutionStatus, Client, WorkflowExecutionStatus from temporalio.contrib.external_workflow_streams import ( AFTER, AppendConflictError, @@ -161,6 +183,8 @@ ) from temporalio.contrib.external_workflow_streams._wake import WakeTransport from temporalio.converter import ( + ActivitySerializationContext, + SerializationContext, WorkflowSerializationContext, ) from temporalio.service import RPCError, RPCStatusCode @@ -412,7 +436,7 @@ def _stamp_hash(wire: WireRecord) -> None: class _LogAppend: - """One record on a topic's log, in one Redis call.""" + """One record on one log, a topic's or an activity's, in one Redis call.""" def __init__(self, backend: RedisStreamBackend) -> None: """Bind to ``backend``'s client.""" @@ -434,6 +458,97 @@ async def write(self, *, name: str, record: TransportRecord, digest: str) -> Off return Offset(_text(placed)) +def _prefix(backend: Any) -> str: + """The key prefix ``backend`` writes under, for keys the provider derives itself.""" + return backend._key_prefix + + +@dataclass(frozen=True) +class _ActivityOwner: + """The activity whose streams a handle addresses, and where Redis keeps them. + + ``workflow_id`` is ``None`` for a standalone activity. ``run_id`` is the + workflow's run for a workflow's activity and the activity's own run for a + standalone one. It is part of the key, so an activity execution's streams + are its own and an id started again in a new run starts new ones, and it + decides whose close ends a read. ``None`` means the run is not known yet: + a handle opened outside without one asks the server for the current run + before it touches a key. + """ + + namespace: str + workflow_id: str | None + activity_id: str + run_id: str | None + + def key(self, key_prefix: str, topic: str) -> str: + """The Redis stream holding ``topic`` of this activity's streams.""" + if self.run_id is None: + raise RuntimeError(f"the run of {self} was not resolved before its key") + # The chain keys percent-encode every id, so none of their components + # holds a "/", and a component built around one can never equal a + # chain key or a key derived from one. + namespace = quote(self.namespace, safe="") + owner = f"activity/{quote(self.workflow_id or '', safe='')}/" + owner += f"{quote(self.run_id, safe='')}/{quote(self.activity_id, safe='')}" + return f"{key_prefix}:{namespace}:{owner}:{quote(topic, safe='')}" + + def context(self) -> SerializationContext: + """What the owner's payloads are coded under. + + A workflow's activity writes the workflow's data, as it does on the + workflow's own topics; a standalone activity has no workflow, so its + own identity is the context. + """ + if self.workflow_id is not None: + return WorkflowSerializationContext( + namespace=self.namespace, workflow_id=self.workflow_id + ) + return ActivitySerializationContext( + namespace=self.namespace, + activity_id=self.activity_id, + activity_type=None, + activity_task_queue=None, + workflow_id=None, + workflow_type=None, + is_local=False, + ) + + def __str__(self) -> str: + """The owner, for messages.""" + if self.workflow_id is None: + return f"activity {self.activity_id!r}" + return f"activity {self.activity_id!r} of workflow {self.workflow_id!r}" + + +async def _resolve_owner(client: Client, owner: _ActivityOwner) -> _ActivityOwner: + """``owner`` with its run known, or ``StreamNotFoundError`` when the server does not know it. + + One describe: it says whether the owner exists and, for a handle opened + without a run, which run is current. A standalone activity is described + itself; a workflow's activity is described through its workflow, because + the server does not describe it on its own. + """ + try: + if owner.workflow_id is None: + described = await client.get_activity_handle( + owner.activity_id, run_id=owner.run_id + ).describe() + run_id = owner.run_id or described.activity_run_id + else: + description = await client.get_workflow_handle( + owner.workflow_id, run_id=owner.run_id + ).describe() + run_id = owner.run_id or description.run_id + except RPCError as error: + if error.status == RPCStatusCode.NOT_FOUND: + raise StreamNotFoundError(f"{owner} was not found") from error + raise + if not run_id: + raise StreamNotFoundError(f"the server reports no run for {owner}") + return replace(owner, run_id=run_id) + + async def _retained(client: Any, name: str, offset: Offset) -> bool: """Whether the record at ``offset`` on the stream ``name`` survived trimming. @@ -451,6 +566,14 @@ async def _retained(client: Any, name: str, offset: Offset) -> bool: return wanted > _entry_id(info["last-generated-id"]) +async def _newest(store: Any, name: str) -> Cursor: + """The cursor of the newest entry of the log ``name``, or ``BEGINNING``.""" + newest: Any = await store.xrevrange(name, "+", "-", count=1) + if not newest: + return BEGINNING + return mint_cursor(_PROVIDER, _text(newest[0][0])) + + def _is_staged(fields: Any) -> bool: """Whether a log entry is one the workflow staged, rather than a producer's.""" return _STAGE_FIELD in fields @@ -877,11 +1000,13 @@ def __init__( topic: str, producer_id: str, attempt: int, + owner: _ActivityOwner | None = None, ) -> None: - """Bind this producer to ``topic`` of the chain.""" + """Bind this producer to ``topic`` of the chain or of ``owner``.""" self._streams = streams self._client = client self._workflow_id = workflow_id + self._owner = owner self._topic = topic self._producer_id = producer_id self._attempt = attempt @@ -963,6 +1088,17 @@ async def _connect(self) -> None: return backend = self._streams._require_backend() self._append = _LogAppend(backend) + if self._owner is not None: + # No transport producer to bind and no one to wake: an activity's + # stream has no workflow reader and no chain to check the key against. + # Inside the activity the run is known and nothing is described. + if self._owner.run_id is None: + self._owner = await _resolve_owner(self._client, self._owner) + self._name = self._owner.key(_prefix(backend), self._topic) + self._codec = StreamPayloadCodec( + self._client.data_converter.with_context(self._owner.context()), bytes + ) + return assert self._workflow_id is not None chain = await _chain(self._client, self._workflow_id) try: @@ -1030,7 +1166,8 @@ async def _place(self, records: list[WireRecord]) -> Offset | None: last = await self._append.write( name=self._name, record=staged, digest=digest ) - await self._wake(last) + if self._owner is None: + await self._wake(last) return last async def _wake(self, position: Offset | None) -> None: @@ -1102,7 +1239,14 @@ async def _chain_is_terminal(self) -> bool: class RedisStreamHandle: - """One workflow's topics from outside, over the transport's output streams.""" + """One owner's topics from outside. + + The owner is ``workflow_id``'s workflow, whose topic logs are read through + the transport's output read so a staged batch is a barrier until its task + settles, or with ``activity_id`` an activity, read from the stream the + provider keeps for it: a standalone activity without ``workflow_id``, or + an activity that workflow scheduled. + """ def __init__( self, @@ -1110,19 +1254,30 @@ def __init__( client: Client, workflow_id: str | None, run_id: str | None, + activity_id: str | None = None, ) -> None: - """Address the workflow's topics; ``run_id`` decides whose close ends a read.""" + """Address the owner's topics; ``run_id`` decides whose close ends a read.""" self._streams = streams self._client = client self._workflow_id = workflow_id self._run_id = run_id - if workflow_id is None: - raise ValueError("a stream handle needs a workflow_id") - converter = client.data_converter.with_context( - WorkflowSerializationContext( - namespace=client.namespace, workflow_id=workflow_id + self._owner: _ActivityOwner | None = None + converter = client.data_converter + if activity_id is not None: + self._owner = _ActivityOwner( + client.namespace, workflow_id, activity_id, run_id + ) + converter = converter.with_context(self._owner.context()) + else: + if workflow_id is None: + raise ValueError( + "a stream handle needs a workflow_id or an activity_id" + ) + converter = converter.with_context( + WorkflowSerializationContext( + namespace=client.namespace, workflow_id=workflow_id + ) ) - ) self._converter = client.data_converter.payload_converter self._codec: StreamPayloadCodec[bytes] = StreamPayloadCodec(converter, bytes) @@ -1134,11 +1289,12 @@ def read( last: int | None = None, result_type: type | None = None, ) -> AsyncGenerator[StreamRecord[Any], None]: - """Yield records on ``topic`` from where the read starts until the workflow closes. + """Yield records on ``topic`` from where the read starts until the owner closes. - That is the chain, or the pinned run. ``END`` and ``last=`` are - positioned against the log on the first step of the generator, since - this call cannot reach the store. + For a workflow that is the chain, or the pinned run; for an activity it is + the activity reaching a terminal status, learned as the module docstring + says. ``END`` and ``last=`` are positioned against the log on the first + step of the generator, since this call cannot reach the store. Two refusals and they do not land together. A cursor another provider minted, or one that is not a Redis entry id, is refused by this call: reading the @@ -1158,8 +1314,116 @@ def read( # a foreign one fails this call, not the first iteration. tail = last if last is not None else (0 if after == END else None) position = None if tail is not None else _position(after) + if self._owner is not None: + return self._read_owned( + self._owner, topic, position, after, result_type, tail=tail + ) return self._read(topic, position, after, result_type, tail=tail) + async def _read_owned( + self, + owner: _ActivityOwner, + topic: str, + position: Offset | None, + after: Cursor, + result_type: type | None, + *, + tail: int | None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: + backend = self._streams._require_backend() + # Described on every read, so a handle without a run reads the run that + # is current when the read starts, and ends with it. + owner = await _resolve_owner(self._client, owner) + store = backend._client + name = owner.key(_prefix(backend), topic) + if tail is not None: + position = await _tail_after(store, name, tail) + after = ( + BEGINNING + if position is None + else mint_cursor(_PROVIDER, position.token) + ) + decoder = RecordDecoder( + self._converter, result_type, after=after, warn=logger.warning + ) + if ( + position is not None + and isinstance(backend, _TopicLogBackend) + and not await _retained(store, name, position) + ): + # Refused rather than resumed from the first retained record, + # which would skip whatever the trim took in between. + raise StreamCursorError( + f"cursor {after.token!r} names a record on {topic!r} that the " + f"provider's retention has trimmed ({backend.describe_window()})" + ) + start = _BEGINNING_SENTINEL if position is None else position.token + # XREAD BLOCK 0 waits forever, which is not what a zero poll asks for. + block = int(self._streams._poll.total_seconds() * 1000) or None + closed = False + while True: + found: Any = await store.xread( + {name: start}, count=_READ_BATCH, block=block + ) + entries = found[0][1] if found else [] + for entry_id, fields in entries: + placed = _to_record(entry_id, fields) + assert placed.offset is not None + start = placed.offset.token + if placed.kind is not TransportRecordKind.DATA: + continue + minted = mint_cursor(_PROVIDER, placed.offset.token) + wire = _parse( + minted, await self._codec.decode(placed.payload), logger.warning + ) + if wire is None: + continue + for record in decoder.decode(minted, wire): + yield record + if entries: + continue + if closed: + return + # One more pass after learning the owner is terminal, so a record + # that landed between the read and the describe is not lost. + closed = await self._owner_closed(owner, store, name) + + async def _owner_closed(self, owner: _ActivityOwner, store: Any, name: str) -> bool: + """Whether the owning activity is terminal, as far as this store can tell. + + A standalone activity says so itself. The server does not describe a + workflow's activity, so its workflow is asked: the workflow closing ends + the read, and so does the activity leaving the pending set once its + stream exists. Before the first write there is no stream, so an + activity that never wrote is read until its workflow closes. + """ + try: + if owner.workflow_id is None: + described = await self._client.get_activity_handle( + owner.activity_id, run_id=owner.run_id + ).describe() + return described.status != ActivityExecutionStatus.RUNNING + description = await self._client.get_workflow_handle( + owner.workflow_id, run_id=owner.run_id + ).describe() + except RPCError as error: + if error.status == RPCStatusCode.NOT_FOUND: + # The owner's History is gone, so there is nothing left to follow. + return True + raise + status = description.status + if status is not None and status != WorkflowExecutionStatus.RUNNING: + # An activity's streams are not the chain's: a run that continued + # as new took its activities with it. + return True + pending = any( + info.activity_id == owner.activity_id + for info in description.raw_description.pending_activities + ) + if pending: + return False + return bool(await store.exists(name)) + async def _read( self, topic: str, @@ -1292,6 +1556,9 @@ async def latest(self, *, topic: str | StreamTopic[Any]) -> Cursor: """ topic, _ = resolve_topic(topic) backend = self._streams._require_backend() + if self._owner is not None: + owner = await _resolve_owner(self._client, self._owner) + return await _newest(backend._client, owner.key(_prefix(backend), topic)) assert self._workflow_id is not None chain = await _chain(self._client, self._workflow_id) try: @@ -1307,6 +1574,13 @@ async def latest(self, *, topic: str | StreamTopic[Any]) -> Cursor: def ref(self, *, topic: str | StreamTopic[Any] | None = None) -> StreamRef: """A ref to ``topic`` of this owner's stream, pinned as this handle is.""" + if self._owner is not None: + return StreamRef.for_activity( + self._owner.activity_id, + workflow_id=self._owner.workflow_id, + run_id=self._owner.run_id, + topic=topic, + ) assert self._workflow_id is not None return StreamRef.for_workflow( self._workflow_id, run_id=self._run_id, topic=topic @@ -1336,6 +1610,7 @@ def producer( topic, producer_id, attempt, + owner=self._owner, ) @@ -1472,14 +1747,20 @@ def get_activity_stream_handle( workflow_id: str | None = None, run_id: str | None = None, ) -> RedisStreamHandle: - """Refuse: this provider keeps no stream an activity owns. - - Raises: - StreamUnsupportedError: Always. + """A handle on the topics ``activity_id`` owns, apart from any workflow's. + + Without ``workflow_id`` the activity is a standalone one and ``run_id`` + names its run; with one it is that workflow's activity and ``run_id`` + names the workflow's run. The streams are keyed by that run, so a + retry writes to the same ones and a reader sees the attempt change as + ``SUPERSEDED``, while an id started again in a new run starts new + ones. Without ``run_id`` each read, ``latest()`` and the first append + of a producer describe the owner and take the run current at that + moment. A read ends when the activity is terminal and the retained + tail is delivered; the store has no gate on an append after that, so + a late attempt still lands, and a read that has ended does not see it. """ - raise StreamUnsupportedError( - "the redis provider does not host activity streams" - ) + return RedisStreamHandle(self, client, workflow_id, run_id, activity_id) async def create_standalone_stream( self, diff --git a/tests/streams/test_activity_streams.py b/tests/streams/test_activity_streams.py index bd0a675ac..8d4f7c59c 100644 --- a/tests/streams/test_activity_streams.py +++ b/tests/streams/test_activity_streams.py @@ -16,6 +16,7 @@ from __future__ import annotations import asyncio +import os import uuid from collections.abc import AsyncIterator, Callable from dataclasses import dataclass @@ -34,6 +35,7 @@ topic, ) from temporalio.streams.providers.memory import MemoryStreams +from temporalio.streams.providers.redis import RedisStreams from temporalio.testing import WorkflowEnvironment from tests.helpers import new_worker @@ -57,9 +59,32 @@ async def _memory_setup(client: Client) -> AsyncIterator[ActivitySetup]: provider.reset() +async def _redis_setup(client: Client) -> AsyncIterator[ActivitySetup]: + # The store is a Redis the test environment does not start; the server + # is the environment's own unless TEMPORAL_ADDRESS names another. + address = os.environ.get("TEMPORAL_ADDRESS") + if address: + client = await Client.connect( + address, namespace=os.environ.get("TEMPORAL_NAMESPACE", "default") + ) + provider = RedisStreams( + url=os.environ.get("TEMPORAL_TEST_REDIS_URL") + or os.environ.get("AI198_REDIS_URL", "redis://127.0.0.1:6379"), + # A prefix per setup, because the store keeps what earlier runs wrote. + key_prefix=f"streams-activity-{uuid.uuid4().hex}", + poll_interval=timedelta(milliseconds=100), + ) + config = client.config() + config["plugins"] = [provider] + yield ActivitySetup("redis", provider, Client(**config)) + await provider.close() + + SETUPS: dict[str, Callable[[Client], AsyncIterator[ActivitySetup]]] = { "memory": _memory_setup } +if os.environ.get("STREAMS_LIVE") == "redis": + SETUPS["redis"] = _redis_setup @pytest.fixture(params=sorted(SETUPS)) diff --git a/tests/streams/test_redis_activity_streams.py b/tests/streams/test_redis_activity_streams.py new file mode 100644 index 000000000..9208fb54f --- /dev/null +++ b/tests/streams/test_redis_activity_streams.py @@ -0,0 +1,288 @@ +"""Live checks for what the Redis provider decides about an activity's streams. + +``test_activity_streams`` runs the shared cases on this provider behind +``STREAMS_LIVE=redis``. This module covers what only this store does: how a +read learns that a workflow's activity is terminal without the server saying +so, that an activity that never wrote is read until its workflow closes, +that retention trims an activity's stream like any other, that nothing +gates an append once the owner is terminal, and that an activity's streams +belong to the run its execution is in, so an id started again in a new run +starts new ones. All need a dev server (``TEMPORAL_ADDRESS`` or the test +environment's own) and a Redis (``TEMPORAL_TEST_REDIS_URL`` or +``AI198_REDIS_URL``). +""" + +from __future__ import annotations + +import asyncio +import os +import uuid +from collections.abc import AsyncIterator +from datetime import timedelta + +import pytest + +from temporalio import activity, workflow +from temporalio.client import Client +from temporalio.streams import BEGINNING, RecordKind, StreamCursorError +from temporalio.streams.providers.redis import RedisStreams +from temporalio.testing import WorkflowEnvironment +from tests.helpers import new_worker +from tests.streams.test_activity_streams import ( + TOKENS, + RunsOneActivity, + read_all, + summary, + write_by_default, + write_to_own_streams, +) +from tests.streams.test_streams_conformance import StreamHost, take + +pytestmark = pytest.mark.skipif( + os.environ.get("STREAMS_LIVE") != "redis", + reason="needs a live server and redis; run with STREAMS_LIVE=redis", +) + + +def redis_url() -> str: + return os.environ.get("TEMPORAL_TEST_REDIS_URL") or os.environ.get( + "AI198_REDIS_URL", "redis://127.0.0.1:6379" + ) + + +@pytest.fixture +async def live(client: Client, env: WorkflowEnvironment) -> AsyncIterator[Client]: + """A client with a fresh provider registered, on a server with standalone activities.""" + if env.supports_time_skipping: + pytest.skip("the time-skipping test server has no standalone activities") + address = os.environ.get("TEMPORAL_ADDRESS") + if address: + client = await Client.connect( + address, namespace=os.environ.get("TEMPORAL_NAMESPACE", "default") + ) + provider = RedisStreams( + url=redis_url(), + # A prefix per case, because the store keeps what earlier cases wrote. + key_prefix=f"streams-redis-activity-{uuid.uuid4().hex}", + poll_interval=timedelta(milliseconds=100), + ) + config = client.config() + config["plugins"] = [provider] + try: + yield Client(**config) + finally: + await provider.close() + + +@activity.defn +async def write_nothing(label: str) -> str: + return label + + +@workflow.defn +class RunsOneActivityThenWaits: + """Runs one activity by name, then stays open until released.""" + + def __init__(self) -> None: + self._released = False + + @workflow.signal + def release(self) -> None: + self._released = True + + @workflow.run + async def run(self, name: str, label: str) -> str: + result = await workflow.execute_activity( + name, + label, + activity_id="streamer", + start_to_close_timeout=timedelta(seconds=30), + ) + await workflow.wait_condition(lambda: self._released) + return result + + +async def test_a_read_opened_after_the_activity_finished_ends_by_itself( + live: Client, +): + # The server does not describe a workflow's activity, so the provider ends + # the read once the activity is no longer pending and its stream exists. + workflow_id = f"streams-redis-wfa-{uuid.uuid4().hex}" + async with new_worker( + live, RunsOneActivityThenWaits, activities=[write_to_own_streams] + ) as worker: + handle = await live.start_workflow( + RunsOneActivityThenWaits.run, + args=["write_to_own_streams", "to the activity"], + id=workflow_id, + task_queue=worker.task_queue, + ) + own = live.get_stream_handle(workflow_id, activity_id="streamer") + # The first read follows the activity to its FINISH while it runs. + await take(own.read(topic=TOKENS), 2, 30) + # A read opened afterwards ends on its own, with the workflow still open. + records = await read_all(own.read(topic=TOKENS), timeout=10) + assert summary(records) == [ + (RecordKind.DATA, 1, "to the activity"), + (RecordKind.FINISH, 1, None), + ] + assert (await handle.describe()).status is not None + await handle.signal(RunsOneActivityThenWaits.release) + await handle.result() + + +async def test_an_activity_that_never_wrote_is_read_until_its_workflow_closes( + live: Client, +): + workflow_id = f"streams-redis-wfa-silent-{uuid.uuid4().hex}" + async with new_worker( + live, RunsOneActivityThenWaits, activities=[write_nothing] + ) as worker: + handle = await live.start_workflow( + RunsOneActivityThenWaits.run, + args=["write_nothing", "quiet"], + id=workflow_id, + task_queue=worker.task_queue, + ) + assert await handle.query("__temporal_workflow_metadata") is not None + own = live.get_stream_handle(workflow_id, activity_id="streamer") + reading = asyncio.ensure_future(read_all(own.read(topic=TOKENS), timeout=30)) + # No stream exists for an activity that wrote nothing, so the read has + # nothing to say the activity is over and waits for the workflow. + await asyncio.sleep(1.0) + assert not reading.done() + await handle.signal(RunsOneActivityThenWaits.release) + await handle.result() + assert await reading == [] + + +async def test_retention_trims_an_activity_stream_too(client: Client): + address = os.environ.get("TEMPORAL_ADDRESS") + if address: + client = await Client.connect(address) + provider = RedisStreams( + url=redis_url(), + key_prefix=f"streams-redis-activity-{uuid.uuid4().hex}", + poll_interval=timedelta(milliseconds=100), + max_len=2, + ) + config = client.config() + config["plugins"] = [provider] + live = Client(**config) + workflow_id = f"streams-redis-wfa-trim-{uuid.uuid4().hex}" + try: + async with new_worker(live, StreamHost) as worker: + host = await live.start_workflow( + StreamHost.run, id=workflow_id, task_queue=worker.task_queue + ) + stream = live.get_stream_handle(workflow_id, activity_id="tool") + producer = stream.producer(topic=TOKENS, producer_id="tool", attempt=1) + first = await producer.append({"token": "one"}) + assert first is not None + await producer.append({"token": "two"}, {"token": "three"}) + # The window holds two entries, so the first record is gone. + records = await take(stream.read(topic=TOKENS), 2, 10) + assert [r.value["token"] for r in records] == ["two", "three"] + with pytest.raises(StreamCursorError, match="retention has trimmed"): + await take(stream.read(topic=TOKENS, after=first), 1, 10) + await host.terminate() + finally: + await provider.close() + + +async def test_an_append_after_the_owner_is_terminal_still_lands(live: Client): + # The store has no gate the server would have: a late attempt writes, and + # a reader that already ended is not told. The handle names no run, so it + # describes the finished activity and lands on that run's stream. + activity_id = f"streams-redis-saa-late-{uuid.uuid4().hex}" + async with new_worker(live, activities=[write_by_default]) as worker: + handle = await live.start_activity( + write_by_default, + "standalone", + id=activity_id, + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=30), + ) + assert await handle.result() == activity_id + stream = live.get_stream_handle(activity_id=activity_id) + before = await stream.latest(topic=TOKENS) + assert before != BEGINNING + late = stream.producer(topic=TOKENS, producer_id=activity_id, attempt=2) + after = await late.append({"token": "late"}) + assert after != before + records = await read_all(stream.read(topic=TOKENS), timeout=10) + assert summary(records) == [ + (RecordKind.DATA, 1, "standalone"), + (RecordKind.FINISH, 1, None), + (RecordKind.SUPERSEDED, 2, None), + (RecordKind.DATA, 2, "late"), + ] + # Pinned to the run, the same stream reads the same way. + pinned = live.get_stream_handle(activity_id=activity_id, run_id=handle.run_id) + assert summary(await read_all(pinned.read(topic=TOKENS), timeout=10)) == summary( + records + ) + + +async def test_a_standalone_activity_id_started_again_starts_a_new_stream( + live: Client, +): + # The stream belongs to the activity execution, which is its run, not to + # the id: the second execution under the same id writes to a new stream, + # a handle without a run reads the current one, and a run pins the other. + activity_id = f"streams-redis-saa-again-{uuid.uuid4().hex}" + async with new_worker(live, activities=[write_by_default]) as worker: + runs = [] + for label in ("first", "second"): + handle = await live.start_activity( + write_by_default, + label, + id=activity_id, + task_queue=worker.task_queue, + start_to_close_timeout=timedelta(seconds=30), + ) + assert await handle.result() == activity_id + runs.append(handle.run_id) + assert runs[0] != runs[1] + current = live.get_stream_handle(activity_id=activity_id) + assert summary(await read_all(current.read(topic=TOKENS), timeout=10)) == [ + (RecordKind.DATA, 1, "second"), + (RecordKind.FINISH, 1, None), + ] + earlier = live.get_stream_handle(activity_id=activity_id, run_id=runs[0]) + assert summary(await read_all(earlier.read(topic=TOKENS), timeout=10)) == [ + (RecordKind.DATA, 1, "first"), + (RecordKind.FINISH, 1, None), + ] + + +async def test_a_workflow_activity_in_a_new_run_starts_a_new_stream(live: Client): + # The same workflow id run again schedules the same activity id; keyed by + # the workflow's run, the two executions keep their streams apart. + workflow_id = f"streams-redis-wfa-again-{uuid.uuid4().hex}" + async with new_worker( + live, RunsOneActivity, activities=[write_to_own_streams] + ) as worker: + runs = [] + for label in ("first", "second"): + handle = await live.start_workflow( + RunsOneActivity.run, + args=["write_to_own_streams", label], + id=workflow_id, + task_queue=worker.task_queue, + ) + await handle.result() + runs.append(handle.result_run_id) + assert runs[0] != runs[1] + current = live.get_stream_handle(workflow_id, activity_id="streamer") + assert summary(await read_all(current.read(topic=TOKENS), timeout=10)) == [ + (RecordKind.DATA, 1, "second"), + (RecordKind.FINISH, 1, None), + ] + earlier = live.get_stream_handle( + workflow_id, activity_id="streamer", run_id=runs[0] + ) + assert summary(await read_all(earlier.read(topic=TOKENS), timeout=10)) == [ + (RecordKind.DATA, 1, "first"), + (RecordKind.FINISH, 1, None), + ] diff --git a/tests/streams/test_redis_provider.py b/tests/streams/test_redis_provider.py index 1786ee905..97dde8fa1 100644 --- a/tests/streams/test_redis_provider.py +++ b/tests/streams/test_redis_provider.py @@ -12,12 +12,15 @@ from temporalio.client import WorkflowExecutionStatus from temporalio.contrib.external_workflow_streams import ( - StreamError as TransportStreamError, + StreamDirection, + WakeNotAcknowledgedError, ) from temporalio.contrib.external_workflow_streams import ( - WakeNotAcknowledgedError, + StreamError as TransportStreamError, ) +from temporalio.contrib.external_workflow_streams._backend import StreamKey from temporalio.contrib.external_workflow_streams._record import Offset +from temporalio.contrib.external_workflow_streams._redis import RedisStreamBackend from temporalio.converter import DataConverter from temporalio.service import RPCError, RPCStatusCode from temporalio.streams import BEGINNING, Cursor, StreamCursorError, StreamError @@ -26,6 +29,7 @@ DEFAULT_RETENTION, RedisProducer, RedisStreams, + _ActivityOwner, _drive, _position, _wake_counter, @@ -239,6 +243,50 @@ def register_script(self, _script: str) -> None: return None +def test_activity_keys_encode_their_ids_and_never_meet_chain_keys(): + # The run is part of the key: a workflow's activity is keyed by the + # workflow's run and a standalone one by its own, so an id started again + # in a new run starts a new stream. + assert ( + _ActivityOwner("ns", "wf", "act", "run").key("p", "t") + == "p:ns:activity/wf/run/act:t" + ) + assert ( + _ActivityOwner("ns", None, "act", "run").key("p", "t") + == "p:ns:activity//run/act:t" + ) + # An id holding a separator is encoded, so it cannot move a boundary. + assert ( + _ActivityOwner("n:s", "w/f", "a:c", "r/1").key("p", "t/u") + == "p:n%3As:activity/w%2Ff/r%2F1/a%3Ac:t%2Fu" + ) + # A key needs the run; a handle opened without one resolves it first. + with pytest.raises(RuntimeError, match="not resolved"): + _ActivityOwner("ns", "wf", "act", None).key("p", "t") + # A chain key percent-encodes every id, so none of its components holds a + # "/" however the ids are chosen, and the owner component here always does. + backend = RedisStreamBackend(client=_NoRedis(), key_prefix="p") + forged = StreamKey( + namespace="ns", + workflow_id="activity/wf/act", + first_execution_run_id="t", + stream_name="t", + direction=StreamDirection.OUTPUT, + ) + assert "/" not in backend.stream_key(forged) + assert backend.stream_key(forged) != _ActivityOwner("ns", "wf", "act", "t").key( + "p", "t" + ) + + +def test_an_activity_owner_names_itself_for_messages(): + assert str(_ActivityOwner("ns", None, "act", None)) == "activity 'act'" + assert ( + str(_ActivityOwner("ns", "wf", "act", "run")) + == "activity 'act' of workflow 'wf'" + ) + + def test_cursors_name_entries_of_the_topics_one_log(): assert _position(BEGINNING) is None position = _position(Cursor("redis:1700000000000-3"))