[Brook] Added stream errors and Core's failure kinds.

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",
 ]

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.
+"""

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)

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,