diff --git a/changelog/added/pirouetting-platypus.md b/changelog/added/pirouetting-platypus.md new file mode 100644 index 0000000000..7171967de3 --- /dev/null +++ b/changelog/added/pirouetting-platypus.md @@ -0,0 +1 @@ +Add experimental `temporalio.contrib.streams` so Activities and clients can append ordered records to a Workflow's stream in Redis, and outside readers can follow it from a cursor. diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 095d981512..7ab29b24ab 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -34,7 +34,22 @@ A store is configured as a plugin: ``Client.connect(..., plugins=[RedisStreams(...)])`` registers it on the -client and on every Worker built from that client. +client and on every Worker built from that client. Each +context then reaches a stream the same way: + +- An Activity reaches the stream of the Workflow that scheduled it with + :func:`activity_handle`, and writes as itself: its Activity id with the + scheduling run id, and its Temporal attempt. +- Any process holding a client reaches a Workflow's stream with + :func:`get_stream_handle`, and writes with a producer id and an attempt of + its own. + +Every read follows the Workflow's run chain across Continue-as-New and ends +when the chain closes. A reader can't pin one run. + +In this release only a Workflow owns a stream, and only Activities and +clients read one. Reading inside a Workflow (:func:`workflow_reader`) and +the other owner kinds raise :class:`StreamUnsupportedError`. The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` @@ -55,6 +70,13 @@ StreamStorageError, StreamUnsupportedError, ) +from temporalio.contrib.streams._handles import ( + ActivityStreamHandle, + StreamHandle, + StreamProducer, + activity_handle, + get_stream_handle, +) from temporalio.contrib.streams._plugin import StreamStorePlugin from temporalio.contrib.streams._record import ( Cursor, @@ -64,8 +86,10 @@ ) from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic +from temporalio.contrib.streams._workflow import workflow_reader __all__ = [ + "ActivityStreamHandle", "BEGINNING", "Cursor", "DEFAULT_TOPIC", @@ -75,9 +99,11 @@ "StreamCursorError", "StreamError", "StreamExpiredError", + "StreamHandle", "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamOwnerKind", + "StreamProducer", "StreamProducerError", "StreamRecord", "StreamRecordError", @@ -88,5 +114,8 @@ "StreamTopic", "StreamUnsupportedError", "Supersession", + "activity_handle", + "get_stream_handle", "topic", + "workflow_reader", ] diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py new file mode 100644 index 0000000000..b773d38576 --- /dev/null +++ b/temporalio/contrib/streams/_handles.py @@ -0,0 +1,626 @@ +"""Reaching a stream from an Activity or from client code. + +Every call goes to Core's stream service, which checks the owner, mints and +checks cursors, and synthesizes ``SUPERSEDED``. What stays here is what +needs the plaintext or the user's types: converting values, the two hashes +taken before the codec, and the codec itself. + +An Activity reaches the stream of the Workflow that scheduled it, and writes +as itself: Core derives its producer id as +``@`` and its Temporal attempt is the +producer attempt, so a retry is reported to readers as ``SUPERSEDED``. The +run is part of the id because a stream outlives a run, and Activity ids +repeat across the runs of a chain. + +Any process holding a client reaches a Workflow's stream by Workflow id and +writes with a producer id and attempt of its own. +""" + +from __future__ import annotations + +import asyncio +import weakref +from collections.abc import AsyncGenerator +from datetime import timedelta +from typing import Any, Generic, TypeVar, cast, overload + +import temporalio.activity +from temporalio.bridge.proto.streams import ( + ActivityProducer, + AppendRecord, + AppendRequest, + LatestRequest, + NamedProducer, + ReadRequest, + StreamAddress, + StreamOwnerKind, +) +from temporalio.client import Client +from temporalio.contrib.streams._body import batch_digest, content_hash, encode_bodies +from temporalio.contrib.streams._cursor import BEGINNING +from temporalio.contrib.streams._errors import ( + StreamError, + StreamNotFoundError, + StreamUnsupportedError, +) +from temporalio.contrib.streams._plugin import ( + StreamStorePlugin, + call, + store_for_client, +) +from temporalio.contrib.streams._record import Cursor, RecordKind, StreamRecord +from temporalio.contrib.streams._ref import StreamRef +from temporalio.contrib.streams._topic import StreamTopic, resolve_topic +from temporalio.contrib.streams._wire import WireRecord, from_read, to_wire +from temporalio.converter import DataConverter, WorkflowSerializationContext +from temporalio.service import RPCError, RPCStatusCode + +__all__ = [ + "ActivityStreamHandle", + "StreamHandle", + "StreamProducer", + "activity_handle", + "get_stream_handle", +] + +T = TypeVar("T") + +_READ_WAIT = timedelta(seconds=30) + + +class StreamProducer(Generic[T]): + """Appends to one topic as one producer attempt. + + Every append is visible as soon as the store accepts it. Each record + carries the producer id, the attempt and a sequence that starts at one + for each attempt, which lets the store deduplicate a retry and lets a + reader tell a retry from a new attempt. + """ + + def __init__( + self, + handle: StreamHandle, + topic: str, + producer_id: str, + attempt: int, + activity: ActivityProducer | None = None, + ) -> None: + """Prefer :meth:`StreamHandle.producer`.""" + self._handle = handle + self._topic = topic + self._producer_id = producer_id + self._attempt = attempt + self._activity = activity + self._sequence = 1 + self._last = BEGINNING + # A batch reads the sequence, then awaits the codec and Core, then + # moves it on; calls one at a time keep each batch's sequences. + self._lock = asyncio.Lock() + + @property + def producer_id(self) -> str: + """Who this producer writes as.""" + return self._producer_id + + @property + def attempt(self) -> int: + """The attempt this producer writes.""" + return self._attempt + + async def append(self, *values: T) -> Cursor: + """Append ``values`` as one batch and return the cursor of its last record. + + The batch lands whole and in order. The producer moves its sequence + past the batch only when the append succeeds, so a call that raised + :class:`temporalio.contrib.streams.StreamOutcomeUnknownError` can be + repeated with the same values: the store answers a repeat of the + batch it holds last for this producer attempt with the original + position and writes nothing. + + An empty call writes nothing and returns the cursor of this + producer's last record, or ``BEGINNING`` when it has written none. + + Calls on one producer run one at a time, in the order they were + made, so concurrent calls take consecutive sequences. A call that was + cancelled may or may not have written its batch, and the producer + cannot tell: start a new attempt (a new producer with a higher + ``attempt``) rather than continuing this one. + + Raises: + StreamProducerError: The sequence was already used with different + content, or it is below the newest one the store holds for + this producer attempt. + StreamOutcomeUnknownError: The store may or may not have written + the batch. Retry the same values on this producer. + StreamRefusedError: The store refused the batch, for example + because it is out of memory. Nothing was written. + StreamClosedError: The stream refuses appends. + """ + async with self._lock: + if not values: + return self._last + return await self._write(RecordKind.DATA, values) + + async def finish(self) -> Cursor: + """Write ``FINISH`` for this producer on this topic and return its cursor. + + Says this producer has nothing more to send. It does not say the + work behind it succeeded, and it does not end anyone's read. + """ + async with self._lock: + return await self._write(RecordKind.FINISH, [None]) + + async def _write(self, kind: RecordKind, values: Any) -> Cursor: + converter = self._handle._converter + wires = [ + to_wire( + converter.payload_converter, + topic=self._topic, + kind=kind, + value=value, + producer_id=self._producer_id, + attempt=self._attempt, + sequence=self._sequence + index, + ) + for index, value in enumerate(values) + ] + request = AppendRequest( + stream=self._handle._address(self._topic), + sequence=self._sequence, + records=await _append_records(converter, wires), + digest=batch_digest(wires), + ) + if self._activity is not None: + request.activity.CopyFrom(self._activity) + else: + request.named.CopyFrom( + NamedProducer(producer_id=self._producer_id, attempt=self._attempt) + ) + service = await self._handle._plugin._service_for(self._handle._client) + response = await call(service.append(request)) + self._sequence += len(wires) + self._last = Cursor(response.last_cursor) + return self._last + + +async def _append_records( + converter: Any, wires: list[WireRecord] +) -> list[AppendRecord]: + bodies = iter( + await encode_bodies(converter, [w.body for w in wires if w.HasField("body")]) + ) + records = [] + for wire in wires: + record = AppendRecord(kind=wire.kind) + if wire.HasField("body"): + record.body.CopyFrom(next(bodies)) + record.content_hash = content_hash(wire.body) + records.append(record) + return records + + +class StreamHandle: + """One stream, addressed by topic, from outside Workflow code. + + A handle on a Workflow's stream follows the run chain, so a read goes on + across Continue-as-New and ends when the chain closes. A topic is a + :class:`temporalio.contrib.streams.StreamTopic`, which carries the record + type, or a plain string with ``result_type=`` for a name decided at + runtime. A call that names no topic addresses the topic of :attr:`ref`. + """ + + def __init__( + self, + plugin: StreamStorePlugin, + client: Client, + ref: StreamRef, + ) -> None: + """Prefer :func:`temporalio.contrib.streams.get_stream_handle`.""" + self._plugin = plugin + self._client = client + self._ref = ref + # The stream is the Workflow's, and the Workflow encodes its records with + # its own context, so every reader and producer on it uses that context. + self._converter: DataConverter = client.data_converter.with_context( + WorkflowSerializationContext( + namespace=client.namespace, workflow_id=ref.workflow_id + ) + ) + + @property + def ref(self) -> StreamRef: + """The stream this handle is on, as data another process can open.""" + return self._ref + + @overload + def read( + self, *, topic: StreamTopic[T], after: Cursor = ... + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: type[T], + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: ... + + def read( + self, + *, + topic: str | StreamTopic[Any] | None = None, + after: Cursor = BEGINNING, + result_type: type | None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: + """Yield the records on ``topic`` after ``after`` as they arrive. + + ``BEGINNING`` starts at the oldest record the topic retains, and + ``END`` at whatever is appended after the read starts. Any other + cursor came from a record of this stream, and the read resumes just + past it, so a reader that keeps the last cursor it handled and hands + it back sees every stored record once. A resumed read knows the + attempt of the record at its cursor, so a new attempt of that + producer still arrives as ``SUPERSEDED``. Earlier attempts of other + producers are not known to it. A ``SUPERSEDED`` record withdraws only + what its producer wrote in earlier attempts. The other producers' + records on the topic stay valid. The read ends when the owner's run + chain is closed and every retained record has been delivered. The + result is a generator, so a caller that stops early calls + ``aclose()`` on it. + + The read starts when the generator first runs, not when this call + returns, and ``END`` is resolved then. A reader that must not miss + the next record, such as one about to send something the Workflow + answers on the stream, takes ``latest()`` first and reads after it:: + + start = await handle.latest() + await client.execute_update(...) + async for record in handle.read(after=start): + ... + + Raises: + ValueError: ``result_type`` was passed with a topic definition, + or the topic is empty. Raised by this call. + StreamCursorError: ``after`` came from another store or another + stream. Raised by the first iteration. + StreamExpiredError: ``after`` names a record the store no longer + retains, or the read fell behind retention while it ran. + StreamRecordError: A record did not decode. Read on from its + cursor to skip it. + StreamNotFoundError: The owner doesn't exist. + StreamStorageError: The store failed or couldn't be reached, as + a Redis blip during a long read does. + + Any of these ends the generator. Start a new read after the last + cursor you handled. It reads on from there, though it no longer knows + which earlier attempts of other producers it delivered. + """ + name, result_type = self._resolve(topic, result_type) + return self._read(name, after, result_type) + + async def _read( + self, topic: str, after: Cursor, result_type: type | None + ) -> AsyncGenerator[StreamRecord[Any], None]: + service = await self._plugin._service_for(self._client) + address = self._address(topic, run_id=await self._read_run_id()) + request = ReadRequest(stream=address, after=after.token) + request.wait.FromTimedelta(_READ_WAIT) + while True: + response = await call(service.read(request)) + for record in response.records: + yield await from_read(self._converter, record, result_type) + if response.done: + return + request.after = response.cursor + request.state = response.state + + async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: + """The cursor of the newest record on ``topic``, or ``BEGINNING`` when empty. + + ``read(after=latest())`` yields only what is appended after this call + returned, which is how a client positions itself before it sends + something the Workflow answers on the stream. + """ + name, _ = self._resolve(topic, None) + service = await self._plugin._service_for(self._client) + response = await call(service.latest(LatestRequest(stream=self._address(name)))) + return Cursor(response.cursor) + + @overload + def producer( + self, *, topic: StreamTopic[T], producer_id: str, attempt: int + ) -> StreamProducer[T]: ... + + @overload + def producer( + self, *, topic: str | None = None, producer_id: str, attempt: int + ) -> StreamProducer[Any]: ... + + def producer( + self, + *, + topic: str | StreamTopic[Any] | None = None, + producer_id: str, + attempt: int, + ) -> StreamProducer[Any]: + """A producer on ``topic`` that writes as ``producer_id`` in ``attempt``. + + A process that restarts its work raises ``attempt``, which starts a + new sequence and tells readers that what the earlier attempt of this + ``producer_id`` wrote is superseded. Records of other producers are + not. Keep one producer object per ``(topic, producer_id, + attempt)``. Each object numbers its own records from one. A second + object for the same session is refused as stale once the first moved + past its first batch. Before that, a second object's first batch with + the same content as the first object's is taken as a retry, and the + store writes nothing, with no error. + + Raises: + ValueError: ``producer_id`` is empty or ``attempt`` is below one. + """ + if not producer_id: + raise ValueError("producer_id must not be empty") + if isinstance(attempt, bool) or not isinstance(attempt, int) or attempt < 1: + raise ValueError(f"attempt must be an int of at least 1, got {attempt!r}") + name, _ = self._resolve(topic, None) + return StreamProducer(self, name, producer_id, attempt) + + def _resolve( + self, topic: str | StreamTopic[Any] | None, result_type: type | None + ) -> tuple[str, type | None]: + return resolve_topic(self._ref.topic if topic is None else topic, result_type) + + def _address(self, topic: str, *, run_id: str | None = None) -> StreamAddress: + return StreamAddress( + namespace=self._client.namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=self._ref.workflow_id, + run_id=(self._ref.chain_run_id or "") if run_id is None else run_id, + topic=topic, + ) + + async def _read_run_id(self) -> str: + """The run a read names to Core, which ends a read when that run closes. + + A read must end only with its chain. While the chain that holds + ``chain_run_id`` is the Workflow id's newest, the read names no run + and follows that chain to its end. Once a newer chain took the id, + the old chain has ended, and a read on any of its runs delivers what + is left and ends. + + Raises: + StreamNotFoundError: Temporal holds no run ``chain_run_id``. + StreamError: Describing the Workflow failed. + """ + run_id = self._ref.chain_run_id + if not run_id: + return "" + workflow_id = self._ref.workflow_id + try: + named = await self._client.get_workflow_handle( + workflow_id, run_id=run_id + ).describe() + newest = await self._client.get_workflow_handle(workflow_id).describe() + except RPCError as error: + if error.status == RPCStatusCode.NOT_FOUND: + raise StreamNotFoundError( + f"Workflow {workflow_id!r} has no run {run_id!r}" + ) from error + raise StreamError( + f"could not describe Workflow {workflow_id!r}: {error}" + ) from error + first = named.raw_description.workflow_execution_info.first_run_id + if first == newest.raw_description.workflow_execution_info.first_run_id: + return "" + return run_id + + +class ActivityStreamHandle: + """The scheduling Workflow's stream, as the running Activity sees it. + + Reads and positions as :class:`temporalio.contrib.streams.StreamHandle` + does. :meth:`producer` writes as this Activity, which is the only + identity that lets readers tell this Activity's retry from a new + producer. + """ + + def __init__( + self, + inner: StreamHandle, + identity: ActivityProducer, + producers: dict[str, StreamProducer[Any]], + ) -> None: + """Prefer :func:`temporalio.contrib.streams.activity_handle`.""" + self._inner = inner + self._identity = identity + self._producers = producers + + @property + def ref(self) -> StreamRef: + """The stream this handle is on, as data another process can open.""" + return self._inner.ref + + @overload + def read( + self, *, topic: StreamTopic[T], after: Cursor = ... + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: type[T], + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: ... + + def read( + self, + *, + topic: str | StreamTopic[Any] | None = None, + after: Cursor = BEGINNING, + result_type: type | None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: + """See :meth:`temporalio.contrib.streams.StreamHandle.read`.""" + # One call forwards every overload, which the checker can't match one by one. + inner = cast(Any, self._inner) + return inner.read(topic=topic, after=after, result_type=result_type) + + async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: + """See :meth:`temporalio.contrib.streams.StreamHandle.latest`.""" + return await self._inner.latest(topic=topic) + + @overload + def producer(self, *, topic: StreamTopic[T]) -> StreamProducer[T]: ... + + @overload + def producer(self, *, topic: str | None = None) -> StreamProducer[Any]: ... + + def producer( + self, *, topic: str | StreamTopic[Any] | None = None + ) -> StreamProducer[Any]: + """A producer on ``topic`` that writes as this Activity attempt. + + The producer id is ``@`` and the + attempt is the Activity's Temporal attempt, so the first record of a + retry makes readers see ``SUPERSEDED``, which withdraws what this + Activity's earlier attempts wrote and nothing else. A retry starts a + new sequence, and the store deduplicates a repeated append within one + attempt. + + Every call in one attempt returns the same producer for a topic, so + two parts of the Activity that each ask for it number their records + in one sequence. + """ + name, _ = self._inner._resolve(topic, None) + producer = self._producers.get(name) + if producer is None: + identity = self._identity + producer = self._producers[name] = StreamProducer( + self._inner, + name, + f"{identity.activity_id}@{identity.run_id}", + identity.attempt, + activity=identity, + ) + return producer + + +def activity_handle() -> ActivityStreamHandle: + """The stream of the Workflow that scheduled the running Activity. + + Its ref names the chain of the scheduling run with ``chain_run_id``, so + its producers write to that chain even once a new chain took the + Workflow id. A read on it follows that chain across Continue-as-New and + ends when the chain ends. To write to another + Workflow's stream, use + :func:`get_stream_handle` with ``temporalio.activity.client()`` and a + producer id of your own. + + A Local Activity has no stream producer in this release. Its attempt is + not durable, so after a Worker crash it runs again as the same attempt, + and the store refuses its first append as stale. Use a regular Activity + instead. + + Raises: + StreamUnsupportedError: The Activity is a Local Activity, or it was + not scheduled by a Workflow, since streams owned by an Activity + are not supported in this release. + ValueError: No stream store is registered on the Worker's client. + RuntimeError: Not called from inside an Activity. + """ + info = temporalio.activity.info() + if info.is_local: + raise StreamUnsupportedError( + "a Local Activity has no stream producer in this release, since its " + "attempt isn't durable; use a regular Activity instead" + ) + if info.workflow_id is None or info.workflow_run_id is None: + raise StreamUnsupportedError( + "this Activity was not scheduled by a Workflow, and streams owned by " + "an Activity are not supported in this release" + ) + client = temporalio.activity.client() + ref = StreamRef.for_workflow(info.workflow_id, chain_run_id=info.workflow_run_id) + identity = ActivityProducer( + workflow_id=info.workflow_id, + run_id=info.workflow_run_id, + activity_id=info.activity_id, + attempt=info.attempt, + ) + inner = StreamHandle(store_for_client(client), client, ref) + return ActivityStreamHandle(inner, identity, _attempt_producers()) + + +# The producers of each running Activity attempt, by topic, under the id of the +# attempt's context. The context isn't hashable, so a finalizer drops the entry +# with it. +_activity_producers: dict[int, dict[str, StreamProducer[Any]]] = {} + + +def _attempt_producers() -> dict[str, StreamProducer[Any]]: + context = temporalio.activity._Context.current() + key = id(context) + producers = _activity_producers.get(key) + if producers is None: + producers = _activity_producers[key] = {} + weakref.finalize(context, _activity_producers.pop, key, None) + return producers + + +def get_stream_handle( + client: Client, + workflow_id: str | StreamRef, + *, + topic: str | StreamTopic[Any] | None = None, + chain_run_id: str | None = None, +) -> StreamHandle: + """A handle on a Workflow's stream, through the client's store. + + The handle follows the Workflow's run chain, so a read continues across + Continue-as-New and ends when the chain closes. ``topic`` is the topic a + call that names none addresses. + + Args: + client: A client that carries a stream store plugin. + workflow_id: The Workflow that owns the stream, or a + :class:`temporalio.contrib.streams.StreamRef` that names it. + topic: The handle's default topic. Without one it is + :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + chain_run_id: Any run of the chain to open. Without one the handle + opens the Workflow id's newest chain. It does not end a read + when that run closes. + + Raises: + ValueError: ``topic`` or ``chain_run_id`` was given with a ref, or no + stream store is registered on ``client``. + """ + if isinstance(workflow_id, StreamRef): + if topic is not None or chain_run_id is not None: + raise ValueError("a StreamRef carries its own topic and chain_run_id") + ref = workflow_id + else: + ref = StreamRef.for_workflow( + workflow_id, topic=topic, chain_run_id=chain_run_id + ) + ref._require_supported() + return StreamHandle(store_for_client(client), client, ref) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index e41c7cb7d8..27bebd8af8 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -13,6 +13,7 @@ from dataclasses import dataclass from typing import Any, Literal +from temporalio.contrib.streams._errors import StreamUnsupportedError from temporalio.contrib.streams._topic import ( DEFAULT_TOPIC, StreamTopic, @@ -25,8 +26,7 @@ """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. +a later release adds still decodes, and is refused where a handle opens. """ @@ -55,19 +55,35 @@ class StreamRef: chain_run_id: str | None = None def __post_init__(self) -> None: - """Refuse a ref this release cannot open. + """Refuse a ref that names no stream. + + Any ``kind`` decodes, since a ref of a later release can arrive as + Workflow input, where raising would fail the Workflow Task on every + retry. :meth:`_require_supported` refuses it where a handle opens. Raises: - ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or - ``topic`` is empty. + ValueError: ``kind`` or ``topic`` is empty, or a Workflow ref has + no ``workflow_id``. """ - if self.kind != "workflow": - raise ValueError(f"unknown StreamRef kind {self.kind!r}") - if not self.workflow_id: + if not self.kind: + raise ValueError("a StreamRef needs a kind") + if self.kind == "workflow" and 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") + def _require_supported(self) -> None: + """Refuse to open a stream whose owner kind this release lacks. + + Raises: + StreamUnsupportedError: ``kind`` is not ``"workflow"``. + """ + if self.kind != "workflow": + raise StreamUnsupportedError( + f"streams owned by kind {self.kind!r} are not supported in this " + "release; only Workflow-owned streams are" + ) + @classmethod def for_workflow( cls, diff --git a/temporalio/contrib/streams/_workflow.py b/temporalio/contrib/streams/_workflow.py new file mode 100644 index 0000000000..8c371a4c54 --- /dev/null +++ b/temporalio/contrib/streams/_workflow.py @@ -0,0 +1,30 @@ +"""Streams inside Workflow code. + +Reading a stream inside a Workflow is not part of this release. +""" + +from __future__ import annotations + +from typing import Any + +from temporalio.contrib.streams._errors import StreamUnsupportedError + +__all__ = ["workflow_reader"] + + +def workflow_reader(*_args: Any, **_kwargs: Any) -> Any: + """Reading a stream inside a Workflow is not supported in this release. + + Read from an Activity or a client with + :func:`temporalio.contrib.streams.activity_handle` or + :func:`temporalio.contrib.streams.get_stream_handle`, and send the + Workflow what it needs, for example as a Signal. + + Raises: + StreamUnsupportedError: Always. + """ + raise StreamUnsupportedError( + "reading a stream inside a Workflow is not supported in this release; " + "read it from an Activity or a client and send the Workflow what it " + "needs, for example as a Signal" + ) diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py new file mode 100644 index 0000000000..3bb6a31ef7 --- /dev/null +++ b/tests/contrib/streams/test_outside_handles.py @@ -0,0 +1,551 @@ +"""Reaching a Workflow's stream from an Activity or a client. + +An Activity writes to the stream of the Workflow that scheduled it as +itself, so its retry is reported to readers as ``SUPERSEDED``. A client +reaches a Workflow's stream by Workflow id through its store. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +import uuid +from collections.abc import Sequence +from datetime import timedelta + +import pytest + +from temporalio import activity, workflow +from temporalio.api.common.v1 import Payload +from temporalio.client import Client +from temporalio.common import RetryPolicy +from temporalio.contrib.streams import ( + BEGINNING, + Cursor, + RecordKind, + StreamCursorError, + StreamNotFoundError, + StreamProducerError, + StreamRef, + StreamUnsupportedError, + Supersession, + activity_handle, + get_stream_handle, + topic, +) +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.converter import ( + DataConverter, + PayloadCodec, + SerializationContext, + WithSerializationContext, +) +from temporalio.exceptions import ApplicationError +from temporalio.testing import ActivityEnvironment +from tests.contrib.streams._support import connect_with, new_workflow_id, read_all +from tests.helpers import new_worker + +EVENTS = topic("events", dict) +TOKENS = topic("tokens", str) + + +@workflow.defn +class Waits: + def __init__(self) -> None: + self.done = False + + @workflow.run + async def run(self) -> None: + await workflow.wait_condition(lambda: self.done) + + @workflow.signal + def finish(self) -> None: + self.done = True + + +async def running_owner(client: Client, task_queue: str) -> str: + # Core asks the server about the owner, so the stream's Workflow must exist. + workflow_id = new_workflow_id() + await client.start_workflow(Waits.run, id=workflow_id, task_queue=task_queue) + return workflow_id + + +@activity.defn +async def stream_tokens(tokens: list[str]) -> bool: + stream = activity_handle() + producer = stream.producer(topic=TOKENS) + attempt = activity.info().attempt + # The first attempt writes part of its answer and fails; the retry + # writes the whole answer, which readers must see as superseding it. + await producer.append(*tokens[:attempt]) + if attempt == 1: + raise ApplicationError("retry me") + await producer.append(*tokens[attempt:]) + await producer.finish() + info = activity.info() + return stream.ref == StreamRef.for_workflow( + info.workflow_id or "", chain_run_id=info.workflow_run_id + ) + + +@workflow.defn +class RunsStreamingActivity: + @workflow.run + async def run(self, tokens: list[str]) -> bool: + return await workflow.execute_activity( + stream_tokens, + tokens, + activity_id="tokens", + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=50), maximum_attempts=2 + ), + ) + + +async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, RunsStreamingActivity, activities=[stream_tokens] + ) as worker: + names_its_chain = await streams_client.execute_workflow( + RunsStreamingActivity.run, + ["a", "b", "c"], + id=workflow_id, + task_queue=worker.task_queue, + ) + assert names_its_chain + records = await read_all( + get_stream_handle(streams_client, workflow_id).read(topic=TOKENS) + ) + assert [r.kind for r in records] == [ + RecordKind.DATA, + RecordKind.SUPERSEDED, + RecordKind.DATA, + RecordKind.DATA, + RecordKind.DATA, + RecordKind.FINISH, + ] + described = await streams_client.get_workflow_handle(workflow_id).describe() + # Core derives the same producer id the handle reports. + producer_id = f"tokens@{described.run_id}" + assert records[1].supersession == Supersession(producer_id, 1, 2) + assert [r.value for r in records if r.kind is RecordKind.DATA] == [ + "a", + "a", + "b", + "c", + ] + assert [(r.producer_id, r.attempt, r.sequence) for r in records[2:]] == [ + (producer_id, 2, 1), + (producer_id, 2, 2), + (producer_id, 2, 3), + (producer_id, 2, 4), + ] + + +@activity.defn +async def stream_without_store() -> str: + try: + activity_handle() + except ValueError as error: + return str(error) + return "opened" + + +@workflow.defn +class RunsActivityWithoutStore: + @workflow.run + async def run(self) -> str: + return await workflow.execute_activity( + stream_without_store, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_an_activity_without_a_store_is_told(client: Client): + async with new_worker( + client, RunsActivityWithoutStore, activities=[stream_without_store] + ) as worker: + said = await client.execute_workflow( + RunsActivityWithoutStore.run, + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + assert "no stream store is registered" in said + + +async def test_a_client_handle_follows_the_chain(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + follower = get_stream_handle(streams_client, "wf") + assert follower.ref == StreamRef.for_workflow("wf") + assert follower._address("output").run_id == "" + events = get_stream_handle(streams_client, "wf", topic=EVENTS) + assert events.ref == StreamRef.for_workflow("wf", topic=EVENTS) + assert get_stream_handle(streams_client, events.ref).ref == events.ref + with pytest.raises(ValueError, match="carries its own topic"): + get_stream_handle(streams_client, events.ref, topic=TOKENS) + named = get_stream_handle(streams_client, "wf", chain_run_id="run") + assert named.ref == StreamRef.for_workflow("wf", chain_run_id="run") + # Appends and positions name the chain's run. A read decides at its start. + assert named._address("output").run_id == "run" + with pytest.raises(ValueError, match="chain_run_id"): + get_stream_handle(streams_client, named.ref, chain_run_id="other") + with pytest.raises(ValueError, match="no stream store is registered"): + get_stream_handle(client, "wf") + + +async def test_a_producer_retry_dedupes_and_a_divergent_one_is_refused( + client: Client, +): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, Waits) as worker: + workflow_id = await running_owner(streams_client, worker.task_queue) + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + assert await stream.latest() == BEGINNING + producer = stream.producer(producer_id="p", attempt=1) + assert await producer.append() == BEGINNING + first = await producer.append("a", "b") + assert await producer.append() == first + assert await stream.latest() == first + # A second object for the same session starts at one again: a repeat + # of the newest batch dedupes to the original position. + again = stream.producer(producer_id="p", attempt=1) + assert await again.append("a", "b") == first + divergent = stream.producer(producer_id="p", attempt=1) + with pytest.raises(StreamProducerError): + await divergent.append("x", "y") + owner = streams_client.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + records = await read_all(stream.read()) + assert [r.value for r in records] == ["a", "b"] + + +async def test_a_resumed_read_continues_after_its_cursor(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, Waits) as worker: + workflow_id = await running_owner(streams_client, worker.task_queue) + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + producer = stream.producer(producer_id="p", attempt=1) + middle = await producer.append("a", "b") + await producer.append("c") + # A read from END starts at its first iteration, so a reader that must + # not miss the next record positions itself with latest() first. + live = stream.read(after=await stream.latest()) + await producer.append("d") + assert (await anext(live)).value == "d" + await live.aclose() + with pytest.raises(StreamCursorError): + await anext(stream.read(after=Cursor("not a cursor"))) + owner = streams_client.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + assert [r.value for r in await read_all(stream.read(after=middle))] == ["c", "d"] + + +async def test_a_stream_without_an_owner_is_not_found(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + stream = get_stream_handle(streams_client, f"no-such-owner-{uuid.uuid4()}") + with pytest.raises(StreamNotFoundError): + await stream.producer(producer_id="p", attempt=1).append("a") + + +async def test_a_producer_refuses_a_bad_identity(client: Client): + stream = get_stream_handle(await connect_with(client, MemoryStreams()), "wf") + with pytest.raises(ValueError, match="producer_id"): + stream.producer(producer_id="", attempt=1) + for attempt in (0, True, 1.5): + with pytest.raises(ValueError, match="attempt"): + stream.producer(producer_id="p", attempt=attempt) # type: ignore[arg-type] + + +class NonceCodec(PayloadCodec): + """Encrypts with a fresh nonce per call, so two encodings never match.""" + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload( + metadata={"encoding": b"binary/nonce"}, + data=uuid.uuid4().bytes + p.SerializeToString(), + ) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [Payload.FromString(p.data[16:]) for p in payloads] + + +async def test_a_retry_through_a_nonce_codec_still_dedupes(client: Client): + coded = await connect_with( + client, MemoryStreams(), DataConverter(payload_codec=NonceCodec()) + ) + async with new_worker(coded, Waits) as worker: + workflow_id = await running_owner(coded, worker.task_queue) + stream = get_stream_handle(coded, workflow_id, topic=TOKENS) + first = await stream.producer(producer_id="p", attempt=1).append("a") + retry = stream.producer(producer_id="p", attempt=1) + # The ciphertext differs, and the digest lang took before the codec does not. + assert await retry.append("a") == first + owner = coded.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + assert [r.value for r in await read_all(stream.read())] == ["a"] + + +async def test_an_activity_handle_names_the_chain_of_its_run(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + env = ActivityEnvironment(client=streams_client) + + async def chain_run_id() -> str | None: + return activity_handle().ref.chain_run_id + + assert await env.run(chain_run_id) == ( + ActivityEnvironment.default_info().workflow_run_id + ) + + +@workflow.defn +class ContinuesThenWaits: + def __init__(self) -> None: + self.done = False + + @workflow.run + async def run(self, runs_left: int) -> None: + if runs_left: + workflow.continue_as_new(runs_left - 1) + await workflow.wait_condition(lambda: self.done) + + @workflow.signal + def finish(self) -> None: + self.done = True + + +async def wait_for_run_after(client: Client, workflow_id: str, run_id: str) -> None: + async def continued() -> None: + while ( + await client.get_workflow_handle(workflow_id).describe() + ).run_id == run_id: + await asyncio.sleep(0.05) + + await asyncio.wait_for(continued(), 10) + + +async def test_a_read_on_a_named_chain_follows_continue_as_new(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker(streams_client, ContinuesThenWaits) as worker: + handle = await streams_client.start_workflow( + ContinuesThenWaits.run, 1, id=workflow_id, task_queue=worker.task_queue + ) + first_run = handle.first_execution_run_id or "" + await wait_for_run_after(streams_client, workflow_id, first_run) + stream = get_stream_handle( + streams_client, workflow_id, topic=TOKENS, chain_run_id=first_run + ) + producer = stream.producer(producer_id="p", attempt=1) + await producer.append("x") + reader = stream.read() + first = await asyncio.wait_for(anext(reader), 10) + waiting = asyncio.ensure_future(anext(reader)) + # Past Core's first owner check while the read waits. A read pinned to + # the named run would end there, since that run closed at + # Continue-as-New. + await asyncio.sleep(1.5) + await producer.append("y") + second = await asyncio.wait_for(waiting, 10) + await streams_client.get_workflow_handle(workflow_id).signal( + ContinuesThenWaits.finish + ) + rest = await read_all(reader) + assert [first.value, second.value] == ["x", "y"] + assert rest == [] + + +async def test_a_named_chain_stays_apart_from_a_newer_chain(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker(streams_client, ContinuesThenWaits) as worker: + old = await streams_client.start_workflow( + ContinuesThenWaits.run, 0, id=workflow_id, task_queue=worker.task_queue + ) + old_run = old.first_execution_run_id or "" + await ( + get_stream_handle(streams_client, workflow_id, chain_run_id=old_run) + .producer(producer_id="p", attempt=1) + .append("old") + ) + await old.signal(ContinuesThenWaits.finish) + await old.result() + new = await streams_client.start_workflow( + ContinuesThenWaits.run, 0, id=workflow_id, task_queue=worker.task_queue + ) + await ( + get_stream_handle(streams_client, workflow_id) + .producer(producer_id="p", attempt=1) + .append("new") + ) + # The old chain ended, so its read delivers what it holds and ends. + old_records = await read_all( + get_stream_handle(streams_client, workflow_id, chain_run_id=old_run).read() + ) + await new.signal(ContinuesThenWaits.finish) + await new.result() + new_records = await read_all(get_stream_handle(streams_client, workflow_id).read()) + assert [r.value for r in old_records] == ["old"] + assert [r.value for r in new_records] == ["new"] + + +@activity.defn +async def count_twice() -> str: + producer = activity_handle().producer(topic=TOKENS) + await producer.append("one") + await producer.append("two") + return producer.producer_id + + +@workflow.defn +class CountThenContinue: + @workflow.run + async def run(self, runs_left: int) -> list[str]: + # The same Activity id in every run, as a counter-based id would be. + producer_id = await workflow.execute_activity( + count_twice, + activity_id="count", + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy(maximum_attempts=1), + ) + if runs_left: + workflow.continue_as_new(runs_left - 1) + return [producer_id] + + +async def test_an_activity_producer_writes_in_every_run_of_a_chain(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, CountThenContinue, activities=[count_twice] + ) as worker: + handle = await streams_client.start_workflow( + CountThenContinue.run, 1, id=workflow_id, task_queue=worker.task_queue + ) + (last_producer,) = await handle.result() + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + records = await read_all(stream.read()) + # Both runs' records, under two producers: the run is part of the id. + assert [r.value for r in records] == ["one", "two", "one", "two"] + producers = [r.producer_id for r in records] + assert producers[0] == producers[1] != producers[2] == producers[3] + assert producers[3] == last_producer + assert last_producer.endswith("@" + (await handle.describe()).run_id) + + +class WorkflowKeyedCodec(PayloadCodec, WithSerializationContext): + """Encrypts each payload under the key of the Workflow its context names, as + a multi-tenant codec does. A payload decodes only under the same key.""" + + def __init__(self, key: str = "") -> None: + self.key = key + + def with_context(self, context: SerializationContext) -> WorkflowKeyedCodec: + workflow_id = getattr(context, "workflow_id", None) + return WorkflowKeyedCodec(workflow_id or "") + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload( + metadata={"encoding": b"binary/keyed", "key": self.key.encode()}, + data=p.SerializeToString(), + ) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + out = [] + for p in payloads: + if p.metadata.get("encoding") != b"binary/keyed": + out.append(p) + continue + if p.metadata["key"].decode() != self.key: + raise ValueError( + f"encrypted under {p.metadata['key']!r}, not {self.key!r}" + ) + out.append(Payload.FromString(p.data)) + return out + + +@activity.defn +async def append_one_token() -> None: + await activity_handle().producer(topic=TOKENS).append("from the activity") + + +@workflow.defn +class PublishesThenRunsAnActivity: + @workflow.run + async def run(self) -> None: + await workflow.execute_activity( + append_one_token, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_both_sides_use_the_workflows_serialization_context(client: Client): + # The stream is the Workflow's, so its producers and readers encode with + # the Workflow's context. + keyed = await connect_with( + client, MemoryStreams(), DataConverter(payload_codec=WorkflowKeyedCodec()) + ) + workflow_id = new_workflow_id() + async with new_worker( + keyed, PublishesThenRunsAnActivity, activities=[append_one_token] + ) as worker: + await keyed.execute_workflow( + PublishesThenRunsAnActivity.run, + id=workflow_id, + task_queue=worker.task_queue, + ) + records = await read_all(get_stream_handle(keyed, workflow_id).read(topic=TOKENS)) + assert [r.value for r in records] == ["from the activity"] + + +@activity.defn +async def logs_twice() -> None: + # A helper that asks for the producer on every call. + for line in ["started", "started"]: + await activity_handle().producer(topic=TOKENS).append(line) + + +@workflow.defn +class RunsLogsTwice: + @workflow.run + async def run(self) -> None: + await workflow.execute_activity( + logs_twice, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_an_activity_gets_one_producer_per_topic(client: Client): + # Two producer objects would both start at sequence 1, so the second equal + # record would dedupe into the first and vanish. + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, RunsLogsTwice, activities=[logs_twice] + ) as worker: + await streams_client.execute_workflow( + RunsLogsTwice.run, id=workflow_id, task_queue=worker.task_queue + ) + records = await read_all( + get_stream_handle(streams_client, workflow_id).read(topic=TOKENS) + ) + assert [r.value for r in records] == ["started", "started"] + + +async def test_a_local_activity_has_no_stream_producer(): + env = ActivityEnvironment() + env.info = dataclasses.replace(ActivityEnvironment.default_info(), is_local=True) + + async def local() -> None: + activity_handle() + + with pytest.raises(StreamUnsupportedError, match="Local Activity"): + await env.run(local) diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 2f51d457b5..d74d4e38b0 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -9,6 +9,7 @@ DEFAULT_TOPIC, StreamRef, StreamTopic, + StreamUnsupportedError, topic, ) from temporalio.contrib.streams._topic import ( @@ -63,24 +64,26 @@ def test_a_ref_with_the_old_run_id_decodes_as_the_newest_chain(): 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. +@pytest.mark.parametrize("kind", ["activity", "nexus"]) +def test_a_ref_from_a_later_release_decodes_and_fails_where_it_opens(kind: str): + # Decoding happens while a Workflow Task applies its input, where a raise + # fails the task on every retry. Opening the handle is where the caller + # can catch it. converter = DataConverter.default.payload_converter payload = Payload( metadata={"encoding": b"json/plain"}, - data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}', + data=f'{{"kind": "{kind}", "workflow_id": "wf", "topic": "out"}}'.encode(), ) - with pytest.raises(ValueError, match="'activity'"): - converter.from_payloads([payload], [StreamRef]) + ref = converter.from_payloads([payload], [StreamRef])[0] + assert ref.kind == kind + with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): + ref._require_supported() @pytest.mark.parametrize( diff --git a/tests/contrib/streams/test_unsupported_owners.py b/tests/contrib/streams/test_unsupported_owners.py new file mode 100644 index 0000000000..498957f430 --- /dev/null +++ b/tests/contrib/streams/test_unsupported_owners.py @@ -0,0 +1,73 @@ +"""Owners and readers this release does not support say so. + +Only a Workflow owns a stream in this release, and only Activities and +clients read one. Everything else raises ``StreamUnsupportedError``. +""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from temporalio import workflow +from temporalio.client import Client +from temporalio.contrib.streams import ( + StreamRef, + StreamUnsupportedError, + activity_handle, + get_stream_handle, + topic, + workflow_reader, +) +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.testing import ActivityEnvironment +from tests.contrib.streams._support import connect_with, new_workflow_id +from tests.helpers import new_worker + +EVENTS = topic("events", dict) + + +async def test_a_ref_refuses_owner_kinds_this_release_lacks_where_it_opens( + client: Client, +): + streams_client = await connect_with(client, MemoryStreams()) + for kind in ("activity", "standalone"): + ref = StreamRef(kind, "x") # type: ignore[arg-type] + with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): + get_stream_handle(streams_client, ref) + + +@workflow.defn +class TriesToRead: + @workflow.run + async def run(self) -> str: + try: + workflow_reader(EVENTS) + except StreamUnsupportedError as error: + return str(error) + return "read" + + +async def test_a_workflow_cannot_read_a_stream(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, TriesToRead) as worker: + said = await streams_client.execute_workflow( + TriesToRead.run, + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + assert "not supported in this release" in said + + +async def test_an_activity_with_no_workflow_has_no_stream(): + env = ActivityEnvironment() + env.info = dataclasses.replace( + ActivityEnvironment.default_info(), workflow_id=None, workflow_run_id=None + ) + + async def standalone() -> None: + activity_handle() + + with pytest.raises(StreamUnsupportedError, match="owned by an Activity"): + await env.run(standalone)