diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 8bb571572e..18f89fb525 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -15,12 +15,31 @@ 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`` as an ordinary payload. """ +from temporalio.contrib.streams._cursor import BEGINNING, END +from temporalio.contrib.streams._errors import ( + StreamClosedError, + StreamCursorError, + StreamError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamRecordError, + StreamRefusedError, + StreamStorageError, + StreamUnsupportedError, +) from temporalio.contrib.streams._record import ( Cursor, RecordKind, @@ -29,8 +48,21 @@ ) __all__ = [ + "BEGINNING", "Cursor", + "END", "RecordKind", + "StreamClosedError", + "StreamCursorError", + "StreamError", + "StreamExpiredError", + "StreamNotFoundError", + "StreamOutcomeUnknownError", + "StreamProducerError", "StreamRecord", + "StreamRecordError", + "StreamRefusedError", + "StreamStorageError", + "StreamUnsupportedError", "Supersession", ] diff --git a/temporalio/contrib/streams/_cursor.py b/temporalio/contrib/streams/_cursor.py new file mode 100644 index 0000000000..2e38359cab --- /dev/null +++ b/temporalio/contrib/streams/_cursor.py @@ -0,0 +1,20 @@ +"""The cursors a read starts from without having read anything. + +Core mints and checks every other cursor, so its token format is Core's own. +""" + +from __future__ import annotations + +from temporalio.contrib.streams._record import Cursor + +__all__ = ["BEGINNING", "END"] + +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. +""" diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py new file mode 100644 index 0000000000..5099b454b7 --- /dev/null +++ b/temporalio/contrib/streams/_errors.py @@ -0,0 +1,169 @@ +"""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:`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`). 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 + +import temporalio.exceptions +from temporalio.bridge.proto.streams import StreamFailure, StreamFailureKind +from temporalio.contrib.streams._record import Cursor + +__all__ = [ + "StreamClosedError", + "StreamCursorError", + "StreamError", + "StreamExpiredError", + "StreamNotFoundError", + "StreamOutcomeUnknownError", + "StreamProducerError", + "StreamRecordError", + "StreamRefusedError", + "StreamStorageError", + "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 store or for another stream, or it is not a + token the store 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 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 + retains stay readable. + """ + + +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 + 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. + + 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 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 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 | ValueError: + """The error a caller catches for Core's ``failure``. + + 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: + return StreamProducerError(failure.message, divergent=True) + 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 new file mode 100644 index 0000000000..2d75d38456 --- /dev/null +++ b/tests/contrib/streams/test_errors_cursors.py @@ -0,0 +1,117 @@ +"""The error family, Core's failure kinds, and the cursors a read starts from.""" + +from __future__ import annotations + +from temporalio.bridge.proto.streams import StreamFailure, StreamFailureKind +from temporalio.contrib.streams import ( + BEGINNING, + END, + Cursor, + StreamClosedError, + StreamCursorError, + StreamError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamRecordError, + StreamRefusedError, + StreamStorageError, + StreamUnsupportedError, +) +from temporalio.contrib.streams._errors import error_from_failure + + +def test_every_condition_is_a_stream_error(): + for error in ( + StreamClosedError, + StreamCursorError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamRecordError, + 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_refused_and_storage_failures_are_families_of_their_own(): + # 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) + + +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, + 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()) - { + 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" + + +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 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]