[Brook] Added stream records and the hashes lang owes Core.

temporalio/contrib/streams/__init__.py

@@ -0,0 +1,37 @@
+"""Streams: an ordered log a Workflow owns, written from many places.
+
+.. warning::
+    This package is experimental and may change in future versions.
+
+A Workflow owns a stream. Producers outside Workflow code, such as the
+Workflow's Activities and clients, append to it, and outside readers consume
+it from a cursor. Records live in a store the application runs, which
+Core reaches for every SDK. Record bodies never reach the server.
+
+The contract:
+
+1. **A new attempt supersedes the old one.** When a reader sees the first
+   record of a producer's newer attempt, it yields a
+   :attr:`RecordKind.SUPERSEDED` record first. It withdraws only what that
+   producer wrote in its earlier attempts, not the whole stream and not the
+   records of other producers. No store holds that record, so every store
+   reports a retry the same way.
+
+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._record import (
+    Cursor,
+    RecordKind,
+    StreamRecord,
+    Supersession,
+)
+
+__all__ = [
+    "Cursor",
+    "RecordKind",
+    "StreamRecord",
+    "Supersession",
+]

temporalio/contrib/streams/_body.py

@@ -0,0 +1,74 @@
+"""What lang owes a record's body between the converter and Core.
+
+The payload converter turns a value into the body and stops there. Every
+other payload the SDK sends then passes through the payload codec and
+external storage, and a stream body owes the same, or a codec-protected
+deployment would leak plaintext through its streams. Core never sees a
+plaintext body, so both hashes that need one are taken here, before the
+codec. A codec that encrypts with a fresh nonce makes every retry's bytes
+differ, so a store that compared encoded bytes would refuse a retry as a
+divergent write.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Sequence
+
+from temporalio.api.common.v1 import Payload
+from temporalio.bridge.proto.streams.v1 import StreamRecord as WireRecord
+from temporalio.converter import DataConverter
+
+__all__ = ["batch_digest", "content_hash", "decode_body", "encode_bodies"]
+
+
+def content_hash(payload: Payload) -> bytes:
+    """The SHA-256 of ``payload`` as the converter produced it.
+
+    Taken over the deterministic serialization of the whole payload, metadata
+    included, so two payloads that differ only in their encoding hash apart.
+    Core stores it on the record as hex under ``temporal.io/content-hash``.
+    """
+    return hashlib.sha256(payload.SerializeToString(deterministic=True)).digest()
+
+
+def batch_digest(records: Sequence[WireRecord]) -> bytes:
+    """The identity of one append, taken over its converted records.
+
+    Each record is its length in 8 bytes big-endian, then its deterministic
+    serialization, so a batch split differently can't collide with this one.
+    The store compares it when a producer repeats its newest batch. Only the
+    writer that made it compares it, since deterministic serialization is
+    stable within one protobuf implementation and not across languages, so a
+    retry must come from the same SDK as the batch it repeats.
+    """
+    digest = hashlib.sha256()
+    for record in records:
+        body = record.SerializeToString(deterministic=True)
+        digest.update(len(body).to_bytes(8, "big"))
+        digest.update(body)
+    return digest.digest()
+
+
+async def encode_bodies(
+    converter: DataConverter, payloads: Sequence[Payload]
+) -> list[Payload]:
+    """``payloads`` as the store keeps them.
+
+    Through the payload codec and then external storage, in the order
+    :meth:`temporalio.converter.DataConverter.encode` uses, so a body above
+    the external storage threshold is replaced by a claim.
+    """
+    encoded = await converter._encode_payload_sequence(payloads)
+    return await converter._external_store_payload_sequence(encoded)
+
+
+async def decode_body(converter: DataConverter, payload: Payload) -> Payload:
+    """Undo :func:`encode_bodies` on a body read back from the store.
+
+    Raises:
+        RuntimeError: The body is a claim and ``converter`` has no external
+            storage to redeem it with.
+    """
+    retrieved = await converter._external_retrieve_payload_sequence([payload])
+    return (await converter._decode_payload_sequence(retrieved))[0]

temporalio/contrib/streams/_record.py

@@ -0,0 +1,135 @@
+"""The value types the stream contract is expressed in.
+
+Nothing here touches Temporal or a store, so every store shares it
+unchanged.
+"""
+
+from __future__ import annotations
+
+import enum
+from dataclasses import dataclass
+from typing import Generic, TypeVar
+
+__all__ = [
+    "Cursor",
+    "RecordKind",
+    "StreamRecord",
+    "Supersession",
+]
+
+T = TypeVar("T")
+
+
+@enum.unique
+class RecordKind(enum.IntEnum):
+    """What a record is.
+
+    The stored kinds match ``temporal.sdk.streams.v1.StreamRecordKind``
+    value for value, so a kind crosses the wire as the integer the proto
+    holds. :attr:`SUPERSEDED` is never stored.
+    """
+
+    UNSPECIFIED = 0
+    """The proto's zero value.
+
+    A stored record whose writer set no kind is read as :attr:`DATA`, so a
+    reader never sees this kind on a record.
+    """
+
+    DATA = 1
+    """Carries a value published by a Workflow or a producer."""
+
+    FINISH = 2
+    """The producer in ``producer_id`` will write nothing more on this topic.
+
+    An empty ``producer_id`` names the owning Workflow. A Workflow's
+    ``FINISH`` covers one run: after Continue-as-New the next run can publish
+    on the topic again, under the same empty ``producer_id``. It does not end
+    a read and says nothing about the producer's outcome: an Activity can
+    still fail after it wrote ``FINISH``.
+    """
+
+    SUPERSEDED = 3
+    """A later attempt of the producer in ``producer_id`` started writing.
+
+    It withdraws only what that producer wrote on this topic in its earlier
+    attempts. Records of other producers stay valid, and so does the rest of
+    the stream. Match the earlier records on ``producer_id`` and an
+    ``attempt`` below :attr:`Supersession.attempt`.
+
+    Core's reader synthesizes it from the records it observed. No store
+    holds it, so every store reports a retry the same way. Its cursor is the
+    position before the new attempt's first record. A reader that resumes
+    there gets this record again, then the new attempt, since dropping the
+    earlier attempt's records twice is harmless and missing it is not.
+    """
+
+
+@dataclass(frozen=True)
+class Cursor:
+    """A position in one stream.
+
+    Opaque on purpose. The token names the store that minted it and the
+    stream it belongs to, and a token from another store or another stream
+    is refused with
+    :class:`temporalio.contrib.streams.StreamCursorError`. Do not compare two
+    cursors or do arithmetic on one: hand a cursor back to resume strictly
+    after the record it names.
+    """
+
+    token: str
+
+    def __str__(self) -> str:
+        """The store's position token."""
+        return self.token
+
+
+@dataclass(frozen=True)
+class Supersession:
+    """What a :attr:`RecordKind.SUPERSEDED` record reports.
+
+    ``producer_id`` started ``attempt``, so the records it wrote in
+    ``previous_attempt`` and before are withdrawn. No other producer's
+    records are.
+    """
+
+    producer_id: str
+    previous_attempt: int
+    attempt: int
+
+
+@dataclass(frozen=True)
+class StreamRecord(Generic[T]):
+    """One record as a reader sees it.
+
+    ``value`` is set on a :attr:`RecordKind.DATA` record and
+    ``supersession`` on a :attr:`RecordKind.SUPERSEDED` one. Other kinds
+    carry neither, so a consumer narrows on ``kind`` first.
+    """
+
+    kind: RecordKind
+    cursor: Cursor
+    topic: str
+    producer_id: str = ""
+    """Who wrote it, or empty when the owning Workflow wrote it."""
+    attempt: int = 0
+    """The producer's attempt, or 0 when the producer declared none."""
+    sequence: int = 0
+    """The producer's position within its attempt, or 0 when it does not number."""
+    run_id: str = ""
+    """The run that published it, when the owning Workflow did; empty otherwise.
+
+    A stream follows its Workflow's run chain, so this tells a successor run
+    or a reset branch apart. An Activity's record leaves it empty, and the
+    scheduling run is in its ``producer_id``, as ``<Activity id>@<run id>``.
+    """
+    value: T | None = None
+    """The published value. Set on ``DATA`` only."""
+    supersession: Supersession | None = None
+    """The attempt change being reported. Set on ``SUPERSEDED`` only."""
+    stale: bool = False
+    """An older attempt wrote it after this read delivered a newer attempt.
+
+    Such a record is not part of the producer's current answer, for example
+    the last writes of an Activity attempt that timed out but kept running.
+    """

temporalio/contrib/streams/_wire.py

@@ -0,0 +1,115 @@
+"""How a record crosses to Core and back: the proto is the record.
+
+``temporal.sdk.streams.v1.StreamRecord`` is the stored format on every
+store, and Core builds and parses it. ``body`` is the user's payload,
+produced and consumed through the payload converter, so a pre-encoded
+:class:`temporalio.common.RawValue` passes through untouched.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import temporalio.converter
+from temporalio.bridge.proto.streams import ReadRecord
+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._record import (
+    Cursor,
+    RecordKind,
+    StreamRecord,
+    Supersession,
+)
+
+__all__ = ["RUN_ID_KEY", "WireRecord", "from_read", "to_wire"]
+
+RUN_ID_KEY = "temporal.io/run-id"
+"""The record metadata key the producing run id is stored under.
+
+Core stamps it on every record the owning Workflow publishes, because a
+stream follows the Workflow's run chain and a reader needs the run to tell a
+reset branch or a successor run apart. Its value is a payload with
+``encoding`` ``binary/plain`` whose data is the run id.
+"""
+
+
+def to_wire(
+    converter: temporalio.converter.PayloadConverter,
+    *,
+    topic: str,
+    kind: RecordKind,
+    value: Any = None,
+    producer_id: str = "",
+    attempt: int = 0,
+    sequence: int = 0,
+) -> WireRecord:
+    """The record as the converter produced it, before the codec.
+
+    Only a ``DATA`` record carries a body; the converter encodes ``value``
+    into it.
+
+    Raises:
+        ValueError: ``kind`` is ``SUPERSEDED`` or ``UNSPECIFIED``, which no
+            writer stores.
+    """
+    if kind not in (RecordKind.DATA, RecordKind.FINISH):
+        raise ValueError(f"a writer stores DATA or FINISH records, not {kind.name}")
+    record = WireRecord(
+        topic=topic,
+        kind=StreamRecordKind.ValueType(int(kind)),
+        producer_id=producer_id,
+        attempt=attempt,
+        sequence=sequence,
+    )
+    if kind is RecordKind.DATA:
+        record.body.CopyFrom(converter.to_payloads([value])[0])
+    return record
+
+
+async def from_read(
+    converter: temporalio.converter.DataConverter,
+    read: ReadRecord,
+    result_type: type | None,
+) -> StreamRecord[Any]:
+    """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.
+    """
+    cursor = Cursor(read.cursor)
+    if read.HasField("superseded"):
+        superseded = read.superseded
+        return StreamRecord(
+            kind=RecordKind.SUPERSEDED,
+            cursor=cursor,
+            topic=superseded.topic,
+            producer_id=superseded.producer_id,
+            attempt=superseded.attempt,
+            supersession=Supersession(
+                superseded.producer_id, superseded.previous_attempt, superseded.attempt
+            ),
+        )
+    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]
+    return StreamRecord(
+        kind=RecordKind.DATA if kind is RecordKind.UNSPECIFIED else kind,
+        cursor=cursor,
+        topic=wire.topic,
+        producer_id=wire.producer_id,
+        attempt=wire.attempt,
+        sequence=wire.sequence,
+        run_id=wire.metadata[RUN_ID_KEY].data.decode()
+        if RUN_ID_KEY in wire.metadata
+        else "",
+        value=value,
+        stale=read.stale,
+    )

tests/contrib/streams/test_public_surface.py

@@ -0,0 +1,20 @@
+"""The package exports only what applications use."""
+
+from __future__ import annotations
+
+import temporalio.contrib.streams as streams
+
+# The helpers behind the handles stay importable from their private modules,
+# but they are not part of the public surface.
+_PRIVATE = {
+    "batch_digest",
+    "content_hash",
+    "decode_body",
+    "encode_bodies",
+}
+
+
+def test_internal_helpers_are_not_public():
+    assert _PRIVATE.isdisjoint(streams.__all__)
+    for name in _PRIVATE:
+        assert not hasattr(streams, name), name