From 6171613bd06069b9c756dde88d53461c9d9f924c Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:20:02 -0700 Subject: [PATCH 1/2] Added stream records and the hashes lang owes Core. --- temporalio/contrib/streams/__init__.py | 35 ++++ temporalio/contrib/streams/_body.py | 74 ++++++++ temporalio/contrib/streams/_record.py | 125 +++++++++++++ temporalio/contrib/streams/_wire.py | 115 ++++++++++++ tests/contrib/streams/__init__.py | 0 tests/contrib/streams/test_public_surface.py | 20 +++ tests/contrib/streams/test_records_wire.py | 180 +++++++++++++++++++ 7 files changed, 549 insertions(+) create mode 100644 temporalio/contrib/streams/__init__.py create mode 100644 temporalio/contrib/streams/_body.py create mode 100644 temporalio/contrib/streams/_record.py create mode 100644 temporalio/contrib/streams/_wire.py create mode 100644 tests/contrib/streams/__init__.py create mode 100644 tests/contrib/streams/test_public_surface.py create mode 100644 tests/contrib/streams/test_records_wire.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py new file mode 100644 index 0000000000..7ad19da515 --- /dev/null +++ b/temporalio/contrib/streams/__init__.py @@ -0,0 +1,35 @@ +"""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. 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", +] diff --git a/temporalio/contrib/streams/_body.py b/temporalio/contrib/streams/_body.py new file mode 100644 index 0000000000..7ecbfe361b --- /dev/null +++ b/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] diff --git a/temporalio/contrib/streams/_record.py b/temporalio/contrib/streams/_record.py new file mode 100644 index 0000000000..8a9c6f8010 --- /dev/null +++ b/temporalio/contrib/streams/_record.py @@ -0,0 +1,125 @@ +"""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 same producer started writing. + + 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: 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 ``@``. + """ + 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. + """ diff --git a/temporalio/contrib/streams/_wire.py b/temporalio/contrib/streams/_wire.py new file mode 100644 index 0000000000..f722752a25 --- /dev/null +++ b/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, + ) diff --git a/tests/contrib/streams/__init__.py b/tests/contrib/streams/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py new file mode 100644 index 0000000000..0d23012a8d --- /dev/null +++ b/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 diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py new file mode 100644 index 0000000000..9a18b7d34c --- /dev/null +++ b/tests/contrib/streams/test_records_wire.py @@ -0,0 +1,180 @@ +"""Records, their stored form, and what lang owes a record's body.""" + +from __future__ import annotations + +import uuid +from collections.abc import Sequence + +import pytest + +from temporalio.api.common.v1 import Payload +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 Supersession as ReportedSupersession +from temporalio.contrib.streams._body import ( + batch_digest, + content_hash, + decode_body, + encode_bodies, +) +from temporalio.contrib.streams._wire import RUN_ID_KEY, WireRecord, from_read, to_wire +from temporalio.converter import DataConverter, PayloadCodec + +CONVERTER = DataConverter.default + + +def stored(wire: WireRecord, cursor: str = "c", *, stale: bool = False) -> ReadRecord: + return ReadRecord(cursor=cursor, stored=wire, stale=stale) + + +async def test_a_record_reads_back_as_it_was_written(): + wire = to_wire( + CONVERTER.payload_converter, + topic="out", + kind=RecordKind.DATA, + value={"n": 1}, + producer_id="p", + attempt=2, + sequence=3, + ) + record = await from_read(CONVERTER, stored(wire, "c1", stale=True), dict) + assert record.kind is RecordKind.DATA + assert record.value == {"n": 1} + assert record.cursor == Cursor("c1") + assert (record.producer_id, record.attempt, record.sequence) == ("p", 2, 3) + assert record.stale + + finish = to_wire(CONVERTER.payload_converter, topic="out", kind=RecordKind.FINISH) + assert not finish.HasField("body") + assert (await from_read(CONVERTER, stored(finish), None)).kind is RecordKind.FINISH + + unset = WireRecord( + topic="out", body=CONVERTER.payload_converter.to_payloads([1])[0] + ) + assert (await from_read(CONVERTER, stored(unset), int)).kind is RecordKind.DATA + + +async def test_the_run_id_core_stamps_reaches_the_reader(): + wire = to_wire( + CONVERTER.payload_converter, topic="out", kind=RecordKind.DATA, value=1 + ) + assert (await from_read(CONVERTER, stored(wire), int)).run_id == "" + wire.metadata[RUN_ID_KEY].CopyFrom( + Payload(metadata={"encoding": b"binary/plain"}, data=b"run-1") + ) + assert (await from_read(CONVERTER, stored(wire), int)).run_id == "run-1" + + +async def test_a_supersession_reads_as_the_record_core_synthesized(): + read = ReadRecord( + cursor="c2", + superseded=Supersession( + topic="t", producer_id="model", previous_attempt=1, attempt=2 + ), + ) + record = await from_read(CONVERTER, read, int) + assert record.kind is RecordKind.SUPERSEDED + assert record.cursor == Cursor("c2") + assert (record.topic, record.producer_id, record.attempt) == ("t", "model", 2) + assert record.supersession == ReportedSupersession("model", 1, 2) + assert record.value is None + + +def test_no_writer_stores_a_synthesized_kind(): + for kind in (RecordKind.SUPERSEDED, RecordKind.UNSPECIFIED): + with pytest.raises(ValueError): + to_wire(CONVERTER.payload_converter, topic="out", kind=kind) + + +async def test_a_raw_value_passes_through_untouched(): + raw = Payload(metadata={"encoding": b"binary/custom"}, data=b"\x00\x01") + wire = to_wire( + CONVERTER.payload_converter, + topic="out", + kind=RecordKind.DATA, + value=RawValue(raw), + ) + assert wire.body == raw + record = await from_read(CONVERTER, stored(wire), RawValue) + assert isinstance(record.value, RawValue) + assert record.value.payload == raw + + +def test_the_content_hash_is_the_plaintext_payload_hash(): + converter = CONVERTER.payload_converter + one = converter.to_payloads([{"n": 1}])[0] + assert content_hash(one) == content_hash(converter.to_payloads([{"n": 1}])[0]) + assert content_hash(one) != content_hash(converter.to_payloads([{"n": 2}])[0]) + # Core refuses a hash that isn't a SHA-256. + assert len(content_hash(one)) == 32 + + +def records(*values: str) -> list[WireRecord]: + return [ + to_wire(CONVERTER.payload_converter, topic="t", kind=RecordKind.DATA, value=v) + for v in values + ] + + +def test_the_digest_is_length_delimited_over_the_batch(): + assert batch_digest(records("a", "b")) == batch_digest(records("a", "b")) + assert batch_digest(records("a", "b")) != batch_digest(records("b", "a")) + assert batch_digest(records("ab")) != batch_digest(records("a", "b")) + + +def test_the_digest_matches_what_every_sdk_takes(): + # The vector Core's tests hold, so a retry through another SDK still dedupes. + batch = [ + to_wire( + CONVERTER.payload_converter, + topic="events", + kind=RecordKind.DATA, + value={"n": n}, + producer_id="p", + attempt=1, + sequence=n, + ) + for n in (1, 2) + ] + assert batch_digest(batch).hex() == ( + "0499768af1856bc6fd19cf6cf90e28e971099f1680e8347d70bdb4b65273f30f" + ) + + +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_the_codec_runs_on_the_body_and_never_on_the_hashes(): + converter = DataConverter(payload_codec=NonceCodec()) + plain = records("a")[0].body + first, second = await encode_bodies(converter, [plain, plain]) + # The codec made the stored bytes differ, which the plaintext hash can't see. + assert first != second + assert first.metadata["encoding"] == b"binary/nonce" + assert await decode_body(converter, first) == plain + + record = await from_read( + converter, + stored( + WireRecord( + topic="t", kind=StreamRecordKind.STREAM_RECORD_KIND_DATA, body=first + ) + ), + str, + ) + assert record.value == "a" From a1d2a7ba610f63416f95d5903a65e43bbb3b14f1 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 17:50:01 -0700 Subject: [PATCH 2/2] Said that SUPERSEDED withdraws one producer's earlier attempts. --- temporalio/contrib/streams/__init__.py | 6 ++++-- temporalio/contrib/streams/_record.py | 14 ++++++++++++-- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 7ad19da515..c4dac3a7db 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -12,8 +12,10 @@ 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. No store holds that record, - so every store reports a retry the same way. + :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`` diff --git a/temporalio/contrib/streams/_record.py b/temporalio/contrib/streams/_record.py index 8a9c6f8010..b1624b63bd 100644 --- a/temporalio/contrib/streams/_record.py +++ b/temporalio/contrib/streams/_record.py @@ -50,7 +50,12 @@ class RecordKind(enum.IntEnum): """ SUPERSEDED = 3 - """A later attempt of the same producer started writing. + """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 @@ -81,7 +86,12 @@ def __str__(self) -> str: @dataclass(frozen=True) class Supersession: - """What a :attr:`RecordKind.SUPERSEDED` record reports.""" + """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