From 47d30c74728f3386a527c17c1389269ff21e5f89 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 14:56:17 -0700 Subject: [PATCH 1/6] Added the stream error family and stream-bound cursors. Errors tell a refused write from one with an unknown outcome, and expired cursors from invalid ones. A cursor names its stream, so a provider can refuse one from another stream. --- temporalio/contrib/streams/__init__.py | 26 ++++++ temporalio/contrib/streams/_cursor.py | 78 +++++++++++++++++ temporalio/contrib/streams/_errors.py | 88 ++++++++++++++++++++ tests/contrib/streams/test_errors_cursors.py | 73 ++++++++++++++++ tests/contrib/streams/test_records_wire.py | 3 +- 5 files changed, 266 insertions(+), 2 deletions(-) create mode 100644 temporalio/contrib/streams/_cursor.py create mode 100644 temporalio/contrib/streams/_errors.py create mode 100644 tests/contrib/streams/test_errors_cursors.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 3d98a6b03a..d71ef1947d 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -14,6 +14,11 @@ record of a producer's newer attempt, it yields a :attr:`RecordKind.SUPERSEDED` record first. No store holds that record, so every provider reports a retry the same way. +2. **A cursor belongs to one stream.** Hand it back to resume strictly after + the record it names. A provider refuses a cursor from another provider + or another stream with :class:`StreamCursorError`, and one whose record + retention dropped with :class:`StreamExpiredError`. A read with no cursor + starts at :data:`BEGINNING` or :data:`END`. The record on the wire is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.contrib.streams.proto.v1`, with the user's value in ``body`` @@ -27,6 +32,17 @@ decode_body, encode_body, ) +from temporalio.contrib.streams._cursor import BEGINNING, END +from temporalio.contrib.streams._errors import ( + StreamClosedError, + StreamCursorError, + StreamError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamUnsupportedError, +) from temporalio.contrib.streams._record import ( Cursor, RecordKind, @@ -35,10 +51,20 @@ ) __all__ = [ + "BEGINNING", "CONTENT_HASH_KEY", "Cursor", + "END", "RecordKind", + "StreamClosedError", + "StreamCursorError", + "StreamError", + "StreamExpiredError", + "StreamNotFoundError", + "StreamOutcomeUnknownError", + "StreamProducerError", "StreamRecord", + "StreamUnsupportedError", "Supersession", "content_fingerprint", "content_hash", diff --git a/temporalio/contrib/streams/_cursor.py b/temporalio/contrib/streams/_cursor.py new file mode 100644 index 0000000000..ba5bf11374 --- /dev/null +++ b/temporalio/contrib/streams/_cursor.py @@ -0,0 +1,78 @@ +"""Cursor tokens bound to the provider and the stream that minted them. + +A token reads ``::``. The stream hash is a +short digest of the stream's identity, so a provider refuses a cursor from +another stream at the call instead of resuming at an unrelated position +that happens to exist. The position is the provider's own and nothing else +reads it. +""" + +from __future__ import annotations + +import hashlib + +from temporalio.contrib.streams._errors import StreamCursorError +from temporalio.contrib.streams._record import Cursor + +__all__ = ["BEGINNING", "END", "cursor_position", "mint_cursor", "stream_hash"] + +BEGINNING = Cursor("") +"""Read from the oldest record the stream still retains.""" + +END = Cursor("$end") +"""Read only what is appended after the read starts. + +It is resolved when the read starts. To position a reader before the +reader's process writes something, use the handle's ``latest`` instead. +""" + +_HASH_LENGTH = 8 + + +def stream_hash(namespace: str, owner_kind: str, owner_id: str, topic: str) -> str: + """The short digest that binds a cursor to one stream. + + Taken over the namespace, the owner and the topic. A run id is never + part of it, because a stream follows its owner's run chain and a cursor + stays valid across Continue-as-New. + """ + digest = hashlib.sha256() + for part in (namespace, owner_kind, owner_id, topic): + encoded = part.encode() + digest.update(len(encoded).to_bytes(8, "big")) + digest.update(encoded) + return digest.hexdigest()[:_HASH_LENGTH] + + +def mint_cursor(provider: str, stream: str, position: str) -> Cursor: + """A cursor that names ``position`` on the stream whose hash is ``stream``.""" + return Cursor(f"{provider}:{stream}:{position}") + + +def cursor_position(cursor: Cursor, *, provider: str, stream: str) -> str | None: + """The position inside ``cursor``, or ``None`` for ``BEGINNING``. + + Raises: + StreamCursorError: The cursor is ``END``, which has no position, or it + was minted by another provider or for another stream, or it is + not a token at all. + """ + if cursor == BEGINNING: + return None + if cursor == END: + raise StreamCursorError("END has no position; resolve it at the read") + parts = cursor.token.split(":", 2) + if len(parts) != 3 or not parts[2]: + raise StreamCursorError(f"cursor {cursor.token!r} is not a stream cursor") + minted_by, minted_for, position = parts + if minted_by != provider: + raise StreamCursorError( + f"cursor {cursor.token!r} was minted by the {minted_by!r} provider, " + f"not the {provider!r} provider" + ) + if minted_for != stream: + raise StreamCursorError( + f"cursor {cursor.token!r} belongs to another stream; a cursor resumes " + "only the stream it was read from" + ) + return position diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py new file mode 100644 index 0000000000..3d7e59e0d4 --- /dev/null +++ b/temporalio/contrib/streams/_errors.py @@ -0,0 +1,88 @@ +"""The errors a stream call raises. + +Every stream condition is a :class:`StreamError`, so a caller can catch by +meaning the way it catches other :class:`temporalio.exceptions.TemporalError` +subclasses. Argument mistakes stay ``ValueError``. + +Two distinctions matter to a caller that retries. A write the store refused +(:class:`StreamProducerError`, :class:`StreamClosedError`) did not happen, +and repeating it gets the same answer. A write whose outcome is unknown +(:class:`StreamOutcomeUnknownError`) may have landed, and repeating it on +the same producer is safe because the store deduplicates the retry. A cursor +that names a record the store no longer keeps +(:class:`StreamExpiredError`) is told apart from one that is not valid for +the stream at all (:class:`StreamCursorError`). +""" + +from __future__ import annotations + +import temporalio.exceptions + +__all__ = [ + "StreamClosedError", + "StreamCursorError", + "StreamError", + "StreamExpiredError", + "StreamNotFoundError", + "StreamOutcomeUnknownError", + "StreamProducerError", + "StreamUnsupportedError", +] + + +class StreamError(temporalio.exceptions.TemporalError): + """Base for stream conditions.""" + + +class StreamNotFoundError(StreamError): + """The stream's owner does not exist, or nothing is known about the stream.""" + + +class StreamCursorError(StreamError): + """The cursor is not valid for this stream. + + It was minted by another provider or for another stream, or it is not a + token the provider can parse. + """ + + +class StreamExpiredError(StreamCursorError): + """The cursor names a record the store no longer retains. + + The cursor was valid once, but retention dropped the records after it, + so resuming from it would skip records silently. Start again from + :data:`temporalio.contrib.streams.BEGINNING` or + :data:`temporalio.contrib.streams.END` and decide what to do about the + gap. + """ + + +class StreamClosedError(StreamError): + """The stream is closed and refuses appends. + + The store refused the write, so nothing was written. Records the stream + retains stay readable. + """ + + +class StreamProducerError(StreamError): + """The store refused the append because of the producer's identity. + + The sequence was already used with different content, or it is below the + newest sequence the store holds for this producer and attempt. Nothing + was written, and repeating the call gets the same answer. + """ + + +class StreamOutcomeUnknownError(StreamError): + """The append may or may not have landed. + + The connection failed or timed out after the request left. Retrying the + same values on the same producer is safe: the producer keeps its + sequence until an append succeeds, and the store returns the original + position for a retry that repeats a batch it already holds. + """ + + +class StreamUnsupportedError(StreamError): + """The capability is not offered by this release or by this provider.""" diff --git a/tests/contrib/streams/test_errors_cursors.py b/tests/contrib/streams/test_errors_cursors.py new file mode 100644 index 0000000000..d2d4ff8296 --- /dev/null +++ b/tests/contrib/streams/test_errors_cursors.py @@ -0,0 +1,73 @@ +"""The error family and cursors bound to one provider and one stream.""" + +from __future__ import annotations + +import pytest + +from temporalio.contrib.streams import ( + BEGINNING, + END, + Cursor, + StreamClosedError, + StreamCursorError, + StreamError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamUnsupportedError, +) +from temporalio.contrib.streams._cursor import ( + cursor_position, + mint_cursor, + stream_hash, +) + + +def test_every_condition_is_a_stream_error(): + for error in ( + StreamClosedError, + StreamCursorError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamUnsupportedError, + ): + assert issubclass(error, StreamError) + # Expired is a kind of invalid cursor, so a caller that only resets on a + # bad cursor still catches it, and one that cares can tell them apart. + assert issubclass(StreamExpiredError, StreamCursorError) + # Refused and unknown are different answers to a retrying caller. + assert not issubclass(StreamOutcomeUnknownError, StreamProducerError) + assert not issubclass(StreamProducerError, StreamOutcomeUnknownError) + + +def test_a_cursor_is_bound_to_its_provider_and_stream(): + one = stream_hash("ns", "workflow", "wf", "a") + cursor = mint_cursor("memory", one, "7") + assert cursor_position(cursor, provider="memory", stream=one) == "7" + assert cursor_position(BEGINNING, provider="memory", stream=one) is None + + with pytest.raises(StreamCursorError, match="minted by the 'memory' provider"): + cursor_position(cursor, provider="redis", stream=one) + for other in ( + stream_hash("ns", "workflow", "wf", "b"), + stream_hash("ns", "workflow", "wf2", "a"), + stream_hash("ns2", "workflow", "wf", "a"), + stream_hash("ns", "activity", "wf", "a"), + ): + with pytest.raises(StreamCursorError, match="another stream"): + cursor_position(cursor, provider="memory", stream=other) + for token in ("garbage", "memory:", f"memory:{one}:"): + with pytest.raises(StreamCursorError): + cursor_position(Cursor(token), provider="memory", stream=one) + with pytest.raises(StreamCursorError, match="END"): + cursor_position(END, provider="memory", stream=one) + + +def test_the_stream_hash_is_short_and_length_delimited(): + assert len(stream_hash("ns", "workflow", "wf", "a")) == 8 + assert stream_hash("ns", "workflow", "a:b", "c") != stream_hash( + "ns", "workflow", "a", "b:c" + ) diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py index 6008fec8ba..8a2ac76859 100644 --- a/tests/contrib/streams/test_records_wire.py +++ b/tests/contrib/streams/test_records_wire.py @@ -10,6 +10,7 @@ from temporalio.api.common.v1 import Payload from temporalio.common import RawValue from temporalio.contrib.streams import ( + BEGINNING, CONTENT_HASH_KEY, Cursor, RecordKind, @@ -24,8 +25,6 @@ from temporalio.contrib.streams.proto.v1 import StreamRecord as WireRecord from temporalio.converter import DataConverter, PayloadCodec -BEGINNING = Cursor("") - def test_a_record_round_trips_through_the_wire(): converter = DataConverter.default.payload_converter From a876eee146c098c995e933194b77419296b37deb Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 17:19:47 -0700 Subject: [PATCH 2/6] Added refused and storage error families for streams. A caller must tell a write the store refused from a store that failed, and the store's own exceptions must not leak, so both get a StreamError family that the specific errors join. --- temporalio/contrib/streams/__init__.py | 4 +++ temporalio/contrib/streams/_errors.py | 35 ++++++++++++++++---- tests/contrib/streams/test_errors_cursors.py | 13 ++++++++ 3 files changed, 45 insertions(+), 7 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 7535d96bb8..ba6c5cd0c6 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -34,6 +34,8 @@ StreamNotFoundError, StreamOutcomeUnknownError, StreamProducerError, + StreamRefusedError, + StreamStorageError, StreamUnsupportedError, ) from temporalio.contrib.streams._record import ( @@ -56,6 +58,8 @@ "StreamOutcomeUnknownError", "StreamProducerError", "StreamRecord", + "StreamRefusedError", + "StreamStorageError", "StreamUnsupportedError", "Supersession", ] diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py index 3d7e59e0d4..6ea891289d 100644 --- a/temporalio/contrib/streams/_errors.py +++ b/temporalio/contrib/streams/_errors.py @@ -5,10 +5,13 @@ subclasses. Argument mistakes stay ``ValueError``. Two distinctions matter to a caller that retries. A write the store refused -(:class:`StreamProducerError`, :class:`StreamClosedError`) did not happen, -and repeating it gets the same answer. A write whose outcome is unknown -(:class:`StreamOutcomeUnknownError`) may have landed, and repeating it on -the same producer is safe because the store deduplicates the retry. A cursor +(:class:`StreamRefusedError`, which includes :class:`StreamProducerError` +and :class:`StreamClosedError`) did not happen, and repeating it gets the +same answer until something changes. A failure of the store or of the way +to it (:class:`StreamStorageError`) may hide a write that landed: for an +append that is :class:`StreamOutcomeUnknownError`, and repeating it on the +same producer is safe because the store deduplicates the retry. Errors from +the store's client library never escape: they arrive as one of these. A cursor that names a record the store no longer keeps (:class:`StreamExpiredError`) is told apart from one that is not valid for the stream at all (:class:`StreamCursorError`). @@ -26,6 +29,8 @@ "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamProducerError", + "StreamRefusedError", + "StreamStorageError", "StreamUnsupportedError", ] @@ -57,7 +62,23 @@ class StreamExpiredError(StreamCursorError): """ -class StreamClosedError(StreamError): +class StreamRefusedError(StreamError): + """The store refused the write, so nothing was written. + + For example the store is out of memory, or a key holds a value of the + wrong type. The subclasses name the refusals the contract defines. + """ + + +class StreamStorageError(StreamError): + """The store failed, or could not be reached. + + For a read, retry later. For an append, the outcome is unknown and + :class:`StreamOutcomeUnknownError` is raised. + """ + + +class StreamClosedError(StreamRefusedError): """The stream is closed and refuses appends. The store refused the write, so nothing was written. Records the stream @@ -65,7 +86,7 @@ class StreamClosedError(StreamError): """ -class StreamProducerError(StreamError): +class StreamProducerError(StreamRefusedError): """The store refused the append because of the producer's identity. The sequence was already used with different content, or it is below the @@ -74,7 +95,7 @@ class StreamProducerError(StreamError): """ -class StreamOutcomeUnknownError(StreamError): +class StreamOutcomeUnknownError(StreamStorageError): """The append may or may not have landed. The connection failed or timed out after the request left. Retrying the diff --git a/tests/contrib/streams/test_errors_cursors.py b/tests/contrib/streams/test_errors_cursors.py index d2d4ff8296..a2d17b1578 100644 --- a/tests/contrib/streams/test_errors_cursors.py +++ b/tests/contrib/streams/test_errors_cursors.py @@ -71,3 +71,16 @@ def test_the_stream_hash_is_short_and_length_delimited(): assert stream_hash("ns", "workflow", "a:b", "c") != stream_hash( "ns", "workflow", "a", "b:c" ) + + +def test_refused_and_storage_failures_are_families_of_their_own(): + from temporalio.contrib.streams import StreamRefusedError, StreamStorageError + + # A refused write did not happen; a storage failure may have. + assert issubclass(StreamProducerError, StreamRefusedError) + assert issubclass(StreamClosedError, StreamRefusedError) + assert issubclass(StreamOutcomeUnknownError, StreamStorageError) + assert not issubclass(StreamRefusedError, StreamStorageError) + assert not issubclass(StreamStorageError, StreamRefusedError) + assert issubclass(StreamRefusedError, StreamError) + assert issubclass(StreamStorageError, StreamError) From 8acda51ea357af5eba8d07c442395ba0b7775446 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 13:01:56 -0700 Subject: [PATCH 3/6] Raised StreamRecordError for a record a reader can't decode. Skipping it with a log warning lost data the reader never heard of. The error carries the cursor, so a caller can resume past it on purpose. --- temporalio/contrib/streams/__init__.py | 2 ++ temporalio/contrib/streams/_cursor.py | 5 +++- temporalio/contrib/streams/_errors.py | 24 +++++++++++++++++++- temporalio/contrib/streams/_wire.py | 17 +++++++------- tests/contrib/streams/test_errors_cursors.py | 2 ++ tests/contrib/streams/test_records_wire.py | 12 ++++++---- 6 files changed, 46 insertions(+), 16 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index ba6c5cd0c6..824a868d1b 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -34,6 +34,7 @@ StreamNotFoundError, StreamOutcomeUnknownError, StreamProducerError, + StreamRecordError, StreamRefusedError, StreamStorageError, StreamUnsupportedError, @@ -57,6 +58,7 @@ "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamProducerError", + "StreamRecordError", "StreamRecord", "StreamRefusedError", "StreamStorageError", diff --git a/temporalio/contrib/streams/_cursor.py b/temporalio/contrib/streams/_cursor.py index ba5bf11374..7ba6dca3f3 100644 --- a/temporalio/contrib/streams/_cursor.py +++ b/temporalio/contrib/streams/_cursor.py @@ -34,7 +34,10 @@ def stream_hash(namespace: str, owner_kind: str, owner_id: str, topic: str) -> s Taken over the namespace, the owner and the topic. A run id is never part of it, because a stream follows its owner's run chain and a cursor - stays valid across Continue-as-New. + stays valid across Continue-as-New. So a cursor does not tell two chains + of one Workflow id apart: one from an earlier chain passes the check on a + later chain. Eight hex characters catch a mistake, not a crafted token, + so the hash is no authorization check. """ digest = hashlib.sha256() for part in (namespace, owner_kind, owner_id, topic): diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py index 6ea891289d..455050edcb 100644 --- a/temporalio/contrib/streams/_errors.py +++ b/temporalio/contrib/streams/_errors.py @@ -14,13 +14,20 @@ the store's client library never escape: they arrive as one of these. A cursor that names a record the store no longer keeps (:class:`StreamExpiredError`) is told apart from one that is not valid for -the stream at all (:class:`StreamCursorError`). +the stream at all (:class:`StreamCursorError`). A stored record that the +reader can't decode (:class:`StreamRecordError`) carries its cursor, so the +caller can resume past it on purpose. """ from __future__ import annotations +from typing import TYPE_CHECKING + import temporalio.exceptions +if TYPE_CHECKING: + from temporalio.contrib.streams._record import Cursor + __all__ = [ "StreamClosedError", "StreamCursorError", @@ -29,6 +36,7 @@ "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamProducerError", + "StreamRecordError", "StreamRefusedError", "StreamStorageError", "StreamUnsupportedError", @@ -105,5 +113,19 @@ class StreamOutcomeUnknownError(StreamStorageError): """ +class StreamRecordError(StreamError): + """A stored record could not be turned into a record for this reader. + + Its body did not convert to the topic's type, or the codec or the store + could not decode it. Reading on from :attr:`cursor` skips it, which is a + choice the caller makes, since skipping loses the record. + """ + + def __init__(self, message: str, cursor: Cursor) -> None: + """Name the record that failed by its cursor.""" + super().__init__(message) + self.cursor = cursor + + class StreamUnsupportedError(StreamError): """The capability is not offered by this release or by this provider.""" diff --git a/temporalio/contrib/streams/_wire.py b/temporalio/contrib/streams/_wire.py index e52850cdbb..1f365382d1 100644 --- a/temporalio/contrib/streams/_wire.py +++ b/temporalio/contrib/streams/_wire.py @@ -13,6 +13,7 @@ from typing import Any import temporalio.converter +from temporalio.contrib.streams._errors import StreamRecordError from temporalio.contrib.streams._policy import AttemptTracker from temporalio.contrib.streams._record import Cursor, RecordKind, StreamRecord from temporalio.contrib.streams.proto.v1 import StreamRecord as WireRecord @@ -90,10 +91,10 @@ class RecordDecoder: """Turns stored records into the records a reader yields. One per read. It synthesizes ``SUPERSEDED`` from the attempts it - observes, positions each synthesized record at the cursor before the - record that triggered it, and skips a record it cannot decode with a - warning rather than raising, so one bad record cannot stop every reader - of the stream. + observes, and positions each synthesized record at the cursor before the + record that triggered it. A record it cannot decode raises + :class:`temporalio.contrib.streams.StreamRecordError` with that record's + cursor. Skipping it silently would lose data the reader never hears of. """ def __init__( @@ -128,11 +129,9 @@ def decode(self, cursor: Cursor, wire: WireRecord) -> list[StreamRecord[Any]]: try: record = from_wire(self._converter, cursor, wire, self._result_type) except Exception as error: - self._warn(f"skipping stream record at {cursor}: {error}") - # The skipped record keeps its position, so a resume after it - # moves on rather than tripping over it again. - self._previous = cursor - return [] + raise StreamRecordError( + f"stream record at {cursor} could not be decoded: {error}", cursor + ) from error out: list[StreamRecord[Any]] = [] superseded = self._attempts.note( wire.producer_id, wire.attempt, topic=wire.topic, previous=self._previous diff --git a/tests/contrib/streams/test_errors_cursors.py b/tests/contrib/streams/test_errors_cursors.py index a2d17b1578..ca898f0eae 100644 --- a/tests/contrib/streams/test_errors_cursors.py +++ b/tests/contrib/streams/test_errors_cursors.py @@ -15,6 +15,7 @@ StreamNotFoundError, StreamOutcomeUnknownError, StreamProducerError, + StreamRecordError, StreamUnsupportedError, ) from temporalio.contrib.streams._cursor import ( @@ -32,6 +33,7 @@ def test_every_condition_is_a_stream_error(): StreamNotFoundError, StreamOutcomeUnknownError, StreamProducerError, + StreamRecordError, StreamUnsupportedError, ): assert issubclass(error, StreamError) diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py index 13d0725e31..670dbe19d7 100644 --- a/tests/contrib/streams/test_records_wire.py +++ b/tests/contrib/streams/test_records_wire.py @@ -13,6 +13,7 @@ BEGINNING, Cursor, RecordKind, + StreamRecordError, Supersession, ) from temporalio.contrib.streams._body import ( @@ -85,17 +86,18 @@ def test_supersession_is_synthesized_from_observations(): assert attempts.note("other", 0, topic="t", previous=Cursor("c4")) is None -def test_the_decoder_skips_a_record_it_cannot_decode(): - said: list[str] = [] +def test_the_decoder_raises_with_the_cursor_of_a_record_it_cannot_decode(): converter = DataConverter.default.payload_converter - decoder = RecordDecoder(converter, int, after=BEGINNING, warn=said.append) + decoder = RecordDecoder(converter, int, after=BEGINNING, warn=lambda _: None) bad = WireRecord( topic="t", kind=RecordKind.DATA.value, # type: ignore[arg-type] body=Payload(metadata={"encoding": b"json/plain"}, data=b"{not json"), ) - assert decoder.decode(Cursor("c1"), bad) == [] - assert said and "c1" in said[0] + with pytest.raises(StreamRecordError) as raised: + decoder.decode(Cursor("c1"), bad) + # The caller can resume past it on purpose. + assert raised.value.cursor == Cursor("c1") good = to_wire(converter, topic="t", kind=RecordKind.DATA, value=3) assert [r.value for r in decoder.decode(Cursor("c2"), good)] == [3] From 28716218db4f48a1d387c11dafe719f138aae388 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 22:08:44 -0700 Subject: [PATCH 4/6] Mapped Core's stream failure kinds to the stream errors. Core reports every failed stream call as one failure kind, so each kind maps to the error a caller catches. Cursors are Core's tokens now, and lang keeps only the two a read starts from. --- temporalio/contrib/streams/__init__.py | 7 +- temporalio/contrib/streams/_cursor.py | 67 +------------ temporalio/contrib/streams/_errors.py | 38 ++++++-- temporalio/contrib/streams/_wire.py | 26 +++-- tests/contrib/streams/test_errors_cursors.py | 99 +++++++++++--------- tests/contrib/streams/test_public_surface.py | 1 + tests/contrib/streams/test_records_wire.py | 17 +++- 7 files changed, 129 insertions(+), 126 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 3154c1cb05..705f0b85ea 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -14,6 +14,11 @@ record of a producer's newer attempt, it yields a :attr:`RecordKind.SUPERSEDED` record first. No store holds that record, so every store reports a retry the same way. +2. **A cursor belongs to one stream.** Hand it back to resume strictly after + the record it names. A cursor from another store or another stream is + refused with :class:`StreamCursorError`, and one whose record + retention dropped with :class:`StreamExpiredError`. A read with no cursor + starts at :data:`BEGINNING` or :data:`END`. The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` @@ -53,8 +58,8 @@ "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamProducerError", - "StreamRecordError", "StreamRecord", + "StreamRecordError", "StreamRefusedError", "StreamStorageError", "StreamUnsupportedError", diff --git a/temporalio/contrib/streams/_cursor.py b/temporalio/contrib/streams/_cursor.py index 7ba6dca3f3..2e38359cab 100644 --- a/temporalio/contrib/streams/_cursor.py +++ b/temporalio/contrib/streams/_cursor.py @@ -1,20 +1,13 @@ -"""Cursor tokens bound to the provider and the stream that minted them. +"""The cursors a read starts from without having read anything. -A token reads ``::``. The stream hash is a -short digest of the stream's identity, so a provider refuses a cursor from -another stream at the call instead of resuming at an unrelated position -that happens to exist. The position is the provider's own and nothing else -reads it. +Core mints and checks every other cursor, so its token format is Core's own. """ from __future__ import annotations -import hashlib - -from temporalio.contrib.streams._errors import StreamCursorError from temporalio.contrib.streams._record import Cursor -__all__ = ["BEGINNING", "END", "cursor_position", "mint_cursor", "stream_hash"] +__all__ = ["BEGINNING", "END"] BEGINNING = Cursor("") """Read from the oldest record the stream still retains.""" @@ -25,57 +18,3 @@ It is resolved when the read starts. To position a reader before the reader's process writes something, use the handle's ``latest`` instead. """ - -_HASH_LENGTH = 8 - - -def stream_hash(namespace: str, owner_kind: str, owner_id: str, topic: str) -> str: - """The short digest that binds a cursor to one stream. - - Taken over the namespace, the owner and the topic. A run id is never - part of it, because a stream follows its owner's run chain and a cursor - stays valid across Continue-as-New. So a cursor does not tell two chains - of one Workflow id apart: one from an earlier chain passes the check on a - later chain. Eight hex characters catch a mistake, not a crafted token, - so the hash is no authorization check. - """ - digest = hashlib.sha256() - for part in (namespace, owner_kind, owner_id, topic): - encoded = part.encode() - digest.update(len(encoded).to_bytes(8, "big")) - digest.update(encoded) - return digest.hexdigest()[:_HASH_LENGTH] - - -def mint_cursor(provider: str, stream: str, position: str) -> Cursor: - """A cursor that names ``position`` on the stream whose hash is ``stream``.""" - return Cursor(f"{provider}:{stream}:{position}") - - -def cursor_position(cursor: Cursor, *, provider: str, stream: str) -> str | None: - """The position inside ``cursor``, or ``None`` for ``BEGINNING``. - - Raises: - StreamCursorError: The cursor is ``END``, which has no position, or it - was minted by another provider or for another stream, or it is - not a token at all. - """ - if cursor == BEGINNING: - return None - if cursor == END: - raise StreamCursorError("END has no position; resolve it at the read") - parts = cursor.token.split(":", 2) - if len(parts) != 3 or not parts[2]: - raise StreamCursorError(f"cursor {cursor.token!r} is not a stream cursor") - minted_by, minted_for, position = parts - if minted_by != provider: - raise StreamCursorError( - f"cursor {cursor.token!r} was minted by the {minted_by!r} provider, " - f"not the {provider!r} provider" - ) - if minted_for != stream: - raise StreamCursorError( - f"cursor {cursor.token!r} belongs to another stream; a cursor resumes " - "only the stream it was read from" - ) - return position diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py index 455050edcb..b7be17b860 100644 --- a/temporalio/contrib/streams/_errors.py +++ b/temporalio/contrib/streams/_errors.py @@ -21,12 +21,9 @@ from __future__ import annotations -from typing import TYPE_CHECKING - import temporalio.exceptions - -if TYPE_CHECKING: - from temporalio.contrib.streams._record import Cursor +from temporalio.bridge.proto.streams import StreamFailure, StreamFailureKind +from temporalio.contrib.streams._record import Cursor __all__ = [ "StreamClosedError", @@ -54,8 +51,8 @@ class StreamNotFoundError(StreamError): class StreamCursorError(StreamError): """The cursor is not valid for this stream. - It was minted by another provider or for another stream, or it is not a - token the provider can parse. + It was minted by another store or for another stream, or it is not a + token the store can parse. """ @@ -128,4 +125,29 @@ def __init__(self, message: str, cursor: Cursor) -> None: class StreamUnsupportedError(StreamError): - """The capability is not offered by this release or by this provider.""" + """The capability is not offered by this release or by this store.""" + + +_BY_KIND: dict[int, type[StreamError]] = { + StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: StreamProducerError, + StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_STALE: StreamProducerError, + StreamFailureKind.STREAM_FAILURE_KIND_OUTCOME_UNKNOWN: StreamOutcomeUnknownError, + StreamFailureKind.STREAM_FAILURE_KIND_REFUSED: StreamRefusedError, + StreamFailureKind.STREAM_FAILURE_KIND_CLOSED: StreamClosedError, + StreamFailureKind.STREAM_FAILURE_KIND_CURSOR: StreamCursorError, + StreamFailureKind.STREAM_FAILURE_KIND_EXPIRED: StreamExpiredError, + StreamFailureKind.STREAM_FAILURE_KIND_NOT_FOUND: StreamNotFoundError, + StreamFailureKind.STREAM_FAILURE_KIND_STORAGE: StreamStorageError, + StreamFailureKind.STREAM_FAILURE_KIND_UNSUPPORTED: StreamUnsupportedError, +} + + +def error_from_failure(failure: StreamFailure) -> StreamError: + """The error a caller catches for Core's ``failure``. + + A kind this release does not know is a storage error, since the call + failed for a reason the caller can't act on beyond retrying later. + """ + if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_RECORD: + return StreamRecordError(failure.message, Cursor(failure.cursor)) + return _BY_KIND.get(failure.kind, StreamStorageError)(failure.message) diff --git a/temporalio/contrib/streams/_wire.py b/temporalio/contrib/streams/_wire.py index f722752a25..9b3bb60e63 100644 --- a/temporalio/contrib/streams/_wire.py +++ b/temporalio/contrib/streams/_wire.py @@ -15,6 +15,7 @@ from temporalio.bridge.proto.streams.v1 import StreamRecord as WireRecord from temporalio.bridge.proto.streams.v1 import StreamRecordKind from temporalio.contrib.streams._body import decode_body +from temporalio.contrib.streams._errors import StreamRecordError from temporalio.contrib.streams._record import ( Cursor, RecordKind, @@ -75,8 +76,8 @@ async def from_read( """The record a reader yields for one record of Core's answer. Raises: - ValueError: The record holds a kind no store may hold, or one this - SDK does not know. + StreamRecordError: The body did not decode or convert, or the record + holds a kind this SDK does not know. """ cursor = Cursor(read.cursor) if read.HasField("superseded"): @@ -92,14 +93,19 @@ async def from_read( ), ) wire = read.stored - kind = RecordKind(wire.kind) - if kind is RecordKind.SUPERSEDED: - raise ValueError("a SUPERSEDED record is synthesized, never stored") - value: Any = None - if kind in (RecordKind.DATA, RecordKind.UNSPECIFIED) and wire.HasField("body"): - body = await decode_body(converter, wire.body) - hints = [result_type] if result_type is not None else None - value = converter.payload_converter.from_payloads([body], hints)[0] + try: + kind = RecordKind(wire.kind) + if kind is RecordKind.SUPERSEDED: + raise ValueError("a SUPERSEDED record is synthesized, never stored") + value: Any = None + if kind in (RecordKind.DATA, RecordKind.UNSPECIFIED) and wire.HasField("body"): + body = await decode_body(converter, wire.body) + hints = [result_type] if result_type is not None else None + value = converter.payload_converter.from_payloads([body], hints)[0] + except Exception as error: + raise StreamRecordError( + f"stream record at {cursor} could not be decoded: {error}", cursor + ) from error return StreamRecord( kind=RecordKind.DATA if kind is RecordKind.UNSPECIFIED else kind, cursor=cursor, diff --git a/tests/contrib/streams/test_errors_cursors.py b/tests/contrib/streams/test_errors_cursors.py index ca898f0eae..9381c7a9d2 100644 --- a/tests/contrib/streams/test_errors_cursors.py +++ b/tests/contrib/streams/test_errors_cursors.py @@ -1,9 +1,8 @@ -"""The error family and cursors bound to one provider and one stream.""" +"""The error family, Core's failure kinds, and the cursors a read starts from.""" from __future__ import annotations -import pytest - +from temporalio.bridge.proto.streams import StreamFailure, StreamFailureKind from temporalio.contrib.streams import ( BEGINNING, END, @@ -16,13 +15,11 @@ StreamOutcomeUnknownError, StreamProducerError, StreamRecordError, + StreamRefusedError, + StreamStorageError, StreamUnsupportedError, ) -from temporalio.contrib.streams._cursor import ( - cursor_position, - mint_cursor, - stream_hash, -) +from temporalio.contrib.streams._errors import error_from_failure def test_every_condition_is_a_stream_error(): @@ -45,44 +42,62 @@ def test_every_condition_is_a_stream_error(): assert not issubclass(StreamProducerError, StreamOutcomeUnknownError) -def test_a_cursor_is_bound_to_its_provider_and_stream(): - one = stream_hash("ns", "workflow", "wf", "a") - cursor = mint_cursor("memory", one, "7") - assert cursor_position(cursor, provider="memory", stream=one) == "7" - assert cursor_position(BEGINNING, provider="memory", stream=one) is None - - with pytest.raises(StreamCursorError, match="minted by the 'memory' provider"): - cursor_position(cursor, provider="redis", stream=one) - for other in ( - stream_hash("ns", "workflow", "wf", "b"), - stream_hash("ns", "workflow", "wf2", "a"), - stream_hash("ns2", "workflow", "wf", "a"), - stream_hash("ns", "activity", "wf", "a"), - ): - with pytest.raises(StreamCursorError, match="another stream"): - cursor_position(cursor, provider="memory", stream=other) - for token in ("garbage", "memory:", f"memory:{one}:"): - with pytest.raises(StreamCursorError): - cursor_position(Cursor(token), provider="memory", stream=one) - with pytest.raises(StreamCursorError, match="END"): - cursor_position(END, provider="memory", stream=one) - - -def test_the_stream_hash_is_short_and_length_delimited(): - assert len(stream_hash("ns", "workflow", "wf", "a")) == 8 - assert stream_hash("ns", "workflow", "a:b", "c") != stream_hash( - "ns", "workflow", "a", "b:c" - ) - - def test_refused_and_storage_failures_are_families_of_their_own(): - from temporalio.contrib.streams import StreamRefusedError, StreamStorageError - # A refused write did not happen; a storage failure may have. assert issubclass(StreamProducerError, StreamRefusedError) assert issubclass(StreamClosedError, StreamRefusedError) assert issubclass(StreamOutcomeUnknownError, StreamStorageError) assert not issubclass(StreamRefusedError, StreamStorageError) assert not issubclass(StreamStorageError, StreamRefusedError) - assert issubclass(StreamRefusedError, StreamError) - assert issubclass(StreamStorageError, StreamError) + + +def test_each_core_failure_kind_raises_its_error(): + expected = { + StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: StreamProducerError, + StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_STALE: StreamProducerError, + StreamFailureKind.STREAM_FAILURE_KIND_OUTCOME_UNKNOWN: StreamOutcomeUnknownError, + StreamFailureKind.STREAM_FAILURE_KIND_REFUSED: StreamRefusedError, + StreamFailureKind.STREAM_FAILURE_KIND_CLOSED: StreamClosedError, + StreamFailureKind.STREAM_FAILURE_KIND_CURSOR: StreamCursorError, + StreamFailureKind.STREAM_FAILURE_KIND_EXPIRED: StreamExpiredError, + StreamFailureKind.STREAM_FAILURE_KIND_NOT_FOUND: StreamNotFoundError, + StreamFailureKind.STREAM_FAILURE_KIND_RECORD: StreamRecordError, + StreamFailureKind.STREAM_FAILURE_KIND_STORAGE: StreamStorageError, + StreamFailureKind.STREAM_FAILURE_KIND_UNSUPPORTED: StreamUnsupportedError, + } + # Every kind Core defines has its error, so a new kind fails here first. + assert set(expected) == set(StreamFailureKind.values()) - { + StreamFailureKind.STREAM_FAILURE_KIND_UNSPECIFIED + } + for kind, error_type in expected.items(): + error = error_from_failure(StreamFailure(kind=kind, message="why")) + assert type(error) is error_type + assert str(error) == "why" + + +def test_a_record_failure_carries_the_cursor_to_resume_past(): + error = error_from_failure( + StreamFailure( + kind=StreamFailureKind.STREAM_FAILURE_KIND_RECORD, + message="bad record", + cursor="memory:abc:7", + ) + ) + assert isinstance(error, StreamRecordError) + assert error.cursor == Cursor("memory:abc:7") + + +def test_a_kind_this_release_does_not_know_is_a_storage_error(): + for kind in ( + StreamFailureKind.STREAM_FAILURE_KIND_UNSPECIFIED, + StreamFailureKind.ValueType(99), + ): + error = error_from_failure(StreamFailure(kind=kind, message="new")) + assert type(error) is StreamStorageError + + +def test_the_start_cursors_are_the_tokens_core_reads(): + # An empty `after` starts at the oldest record, and `$end` at what comes next. + assert BEGINNING.token == "" + assert END.token == "$end" + assert str(Cursor("memory:abc:7")) == "memory:abc:7" diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py index 0d23012a8d..a3647d4eba 100644 --- a/tests/contrib/streams/test_public_surface.py +++ b/tests/contrib/streams/test_public_surface.py @@ -11,6 +11,7 @@ "content_hash", "decode_body", "encode_bodies", + "error_from_failure", } diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py index 9a18b7d34c..bc5a3e300e 100644 --- a/tests/contrib/streams/test_records_wire.py +++ b/tests/contrib/streams/test_records_wire.py @@ -11,7 +11,7 @@ from temporalio.bridge.proto.streams import ReadRecord, Supersession from temporalio.bridge.proto.streams.v1 import StreamRecordKind from temporalio.common import RawValue -from temporalio.contrib.streams import Cursor, RecordKind +from temporalio.contrib.streams import Cursor, RecordKind, StreamRecordError from temporalio.contrib.streams import Supersession as ReportedSupersession from temporalio.contrib.streams._body import ( batch_digest, @@ -102,6 +102,21 @@ async def test_a_raw_value_passes_through_untouched(): assert record.value.payload == raw +async def test_a_record_that_does_not_decode_raises_with_its_cursor(): + bad = WireRecord( + topic="t", + kind=RecordKind.DATA.value, # type: ignore[arg-type] + body=Payload(metadata={"encoding": b"json/plain"}, data=b"{not json"), + ) + with pytest.raises(StreamRecordError) as raised: + await from_read(CONVERTER, stored(bad, "c1"), int) + # The caller can resume past it on purpose. + assert raised.value.cursor == Cursor("c1") + unknown = WireRecord(topic="t", kind=7) # type: ignore[arg-type] + with pytest.raises(StreamRecordError): + await from_read(CONVERTER, stored(unknown, "c2"), int) + + def test_the_content_hash_is_the_plaintext_payload_hash(): converter = CONVERTER.payload_converter one = converter.to_payloads([{"n": 1}])[0] From 44cfafbd6763e5737b520a35fe75ce2a69939d5c Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:28:57 -0700 Subject: [PATCH 5/6] Kept whether a producer refusal diverged or was stale. The two need different fixes, and parsing the message was the only way to tell them apart. --- temporalio/contrib/streams/_errors.py | 12 ++++++++++++ tests/contrib/streams/test_errors_cursors.py | 13 +++++++++++++ 2 files changed, 25 insertions(+) diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py index b7be17b860..da29e47dec 100644 --- a/temporalio/contrib/streams/_errors.py +++ b/temporalio/contrib/streams/_errors.py @@ -97,8 +97,18 @@ class StreamProducerError(StreamRefusedError): The sequence was already used with different content, or it is below the newest sequence the store holds for this producer and attempt. Nothing was written, and repeating the call gets the same answer. + + ``divergent`` is true when the sequence was already written with + different content, so the caller changed the batch on a retry. Otherwise + the sequence is stale, most often from a second producer object for the + same session. """ + def __init__(self, message: str, *, divergent: bool = False) -> None: + """Say whether the batch diverged or the sequence is stale.""" + super().__init__(message) + self.divergent = divergent + class StreamOutcomeUnknownError(StreamStorageError): """The append may or may not have landed. @@ -150,4 +160,6 @@ def error_from_failure(failure: StreamFailure) -> StreamError: """ if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_RECORD: return StreamRecordError(failure.message, Cursor(failure.cursor)) + if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: + return StreamProducerError(failure.message, divergent=True) return _BY_KIND.get(failure.kind, StreamStorageError)(failure.message) diff --git a/tests/contrib/streams/test_errors_cursors.py b/tests/contrib/streams/test_errors_cursors.py index 9381c7a9d2..272f1813f9 100644 --- a/tests/contrib/streams/test_errors_cursors.py +++ b/tests/contrib/streams/test_errors_cursors.py @@ -101,3 +101,16 @@ def test_the_start_cursors_are_the_tokens_core_reads(): assert BEGINNING.token == "" assert END.token == "$end" assert str(Cursor("memory:abc:7")) == "memory:abc:7" + + +def test_a_producer_refusal_says_whether_the_batch_diverged(): + # A divergent retry means the caller changed the batch, and a stale one + # usually means a second producer object for the same session. + divergent = error_from_failure( + StreamFailure(kind=StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT) + ) + stale = error_from_failure( + StreamFailure(kind=StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_STALE) + ) + assert isinstance(divergent, StreamProducerError) and divergent.divergent + assert isinstance(stale, StreamProducerError) and not stale.divergent From e8756324b57f7c8a322347e400e0c60393f49a5d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 13:25:40 -0700 Subject: [PATCH 6/6] Raised an invalid stream call argument as ValueError. Core now tells a call-input mistake apart from a store refusal. ValueError is the SDK's class for a bad argument, so a caller can catch the two apart. (cherry picked from commit d5acb4825651d25add41b0099aeb711f98d8ba74) --- temporalio/contrib/streams/_errors.py | 10 +++++++--- tests/contrib/streams/test_errors_cursors.py | 1 + 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py index da29e47dec..5099b454b7 100644 --- a/temporalio/contrib/streams/_errors.py +++ b/temporalio/contrib/streams/_errors.py @@ -152,12 +152,16 @@ class StreamUnsupportedError(StreamError): } -def error_from_failure(failure: StreamFailure) -> StreamError: +def error_from_failure(failure: StreamFailure) -> StreamError | ValueError: """The error a caller catches for Core's ``failure``. - A kind this release does not know is a storage error, since the call - failed for a reason the caller can't act on beyond retrying later. + An input that breaks the call's contract is a ``ValueError``, as in the + rest of the SDK, so a caller can tell it from a store refusal. A kind this + release does not know is a storage error, since the call failed for a + reason the caller can't act on beyond retrying later. """ + if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_INVALID_ARGUMENT: + return ValueError(failure.message) if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_RECORD: return StreamRecordError(failure.message, Cursor(failure.cursor)) if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: diff --git a/tests/contrib/streams/test_errors_cursors.py b/tests/contrib/streams/test_errors_cursors.py index 272f1813f9..2d75d38456 100644 --- a/tests/contrib/streams/test_errors_cursors.py +++ b/tests/contrib/streams/test_errors_cursors.py @@ -64,6 +64,7 @@ def test_each_core_failure_kind_raises_its_error(): StreamFailureKind.STREAM_FAILURE_KIND_RECORD: StreamRecordError, StreamFailureKind.STREAM_FAILURE_KIND_STORAGE: StreamStorageError, StreamFailureKind.STREAM_FAILURE_KIND_UNSUPPORTED: StreamUnsupportedError, + StreamFailureKind.STREAM_FAILURE_KIND_INVALID_ARGUMENT: ValueError, } # Every kind Core defines has its error, so a new kind fails here first. assert set(expected) == set(StreamFailureKind.values()) - {