From eb190ba9656f66ecbbcbc3467fa0a46e1e3536c8 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 14:54:20 -0700 Subject: [PATCH 1/8] Added the SDK-local stream record envelope proto. Streams store records as temporal.sdk.streams.v1.StreamRecord. Only the Python SDK reads it, so it is generated here rather than in the API repo. --- pyproject.toml | 2 + scripts/_proto/Dockerfile | 4 +- .../temporal/sdk/streams/v1/message.proto | 42 +++++ scripts/gen_protos.py | 6 + scripts/gen_protos_docker.py | 3 + temporalio/contrib/streams/__init__.py | 14 ++ temporalio/contrib/streams/proto/__init__.py | 1 + .../contrib/streams/proto/v1/__init__.py | 6 + .../contrib/streams/proto/v1/message_pb2.py | 66 +++++++ .../contrib/streams/proto/v1/message_pb2.pyi | 161 ++++++++++++++++++ tests/contrib/streams/__init__.py | 0 tests/contrib/streams/test_envelope_proto.py | 42 +++++ 12 files changed, 345 insertions(+), 2 deletions(-) create mode 100644 scripts/_proto/temporal/sdk/streams/v1/message.proto create mode 100644 temporalio/contrib/streams/__init__.py create mode 100644 temporalio/contrib/streams/proto/__init__.py create mode 100644 temporalio/contrib/streams/proto/v1/__init__.py create mode 100644 temporalio/contrib/streams/proto/v1/message_pb2.py create mode 100644 temporalio/contrib/streams/proto/v1/message_pb2.pyi create mode 100644 tests/contrib/streams/__init__.py create mode 100644 tests/contrib/streams/test_envelope_proto.py diff --git a/pyproject.toml b/pyproject.toml index 444e4d0c5b..a09b044a1a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -182,6 +182,7 @@ exclude = [ # Ignore generated code 'temporalio/api', 'temporalio/bridge/proto', + 'temporalio/contrib/streams/proto', 'temporalio/nexus/system/workflow_service', ] @@ -258,6 +259,7 @@ exclude = [ "temporalio/api", "temporalio/bridge/proto", "temporalio/bridge/_visitor.py", + "temporalio/contrib/streams/proto", "tests/worker/workflow_sandbox/testmodules/proto", ] diff --git a/scripts/_proto/Dockerfile b/scripts/_proto/Dockerfile index 0f0fb74b06..16e90bb400 100644 --- a/scripts/_proto/Dockerfile +++ b/scripts/_proto/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.10 RUN curl -LsSf https://astral.sh/uv/install.sh | sh ENV PATH="$PATH:/root/.local/bin" RUN uv tool install poethepoet -VOLUME ["/api_new", "/bridge_new"] +VOLUME ["/api_new", "/bridge_new", "/streams_new"] COPY ./ ./ @@ -13,4 +13,4 @@ RUN uv add "protobuf<4" RUN uv sync --all-extras RUN uv run scripts/gen_protos.py -CMD ["sh", "-c", "cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new"] +CMD ["sh", "-c", "cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new && cp -r ./temporalio/contrib/streams/proto/* /streams_new"] diff --git a/scripts/_proto/temporal/sdk/streams/v1/message.proto b/scripts/_proto/temporal/sdk/streams/v1/message.proto new file mode 100644 index 0000000000..514a724343 --- /dev/null +++ b/scripts/_proto/temporal/sdk/streams/v1/message.proto @@ -0,0 +1,42 @@ +syntax = "proto3"; + +package temporal.sdk.streams.v1; + +import "temporal/api/common/v1/message.proto"; + +// One entry in a stream. The record is the wire format: stores keep it +// serialized as is and readers in every language decode the same bytes. +// +// Field numbers and enum values are part of the stored format. The package +// name is not, so this message can move to another package without a data +// change. +message StreamRecord { + // The value the producer published. A provider runs it through the + // payload codec and external storage before it stores the record. + temporal.api.common.v1.Payload body = 1; + // Provenance stamped by the SDK, such as the plaintext content hash. + map metadata = 2; + // The topic the record was published on. + string topic = 3; + // How to read this record. Unspecified is read as DATA. + StreamRecordKind kind = 4; + // Who wrote the record. Empty when the owning Workflow did. + string producer_id = 5; + // The producer's attempt. Readers treat a later attempt by the same + // producer as superseding what the earlier one wrote. + int64 attempt = 6; + // The producer's position within its attempt, starting at one, or zero + // when it does not number its records. + int64 sequence = 7; +} + +// What a record means to a reader. +enum StreamRecordKind { + // Read as DATA. + STREAM_RECORD_KIND_UNSPECIFIED = 0; + // A value the producer published; `body` carries it. + STREAM_RECORD_KIND_DATA = 1; + // The producer named by `producer_id` writes nothing more on `topic`. + // Says nothing about that producer's outcome and does not end the stream. + STREAM_RECORD_KIND_FINISH = 2; +} diff --git a/scripts/gen_protos.py b/scripts/gen_protos.py index 4b2ea04569..c7cca33fd1 100644 --- a/scripts/gen_protos.py +++ b/scripts/gen_protos.py @@ -33,6 +33,7 @@ api_out_dir = base_dir / "temporalio" / "api" sdk_out_dir = base_dir / "temporalio" / "bridge" / "proto" +streams_out_dir = base_dir / "temporalio" / "contrib" / "streams" / "proto" py_fixes = [ partial(re.compile(r"from temporal\.api\.").sub, r"from temporalio.api."), @@ -218,6 +219,11 @@ def generate_protos(output_dir: Path): for p in (output_dir / "temporal" / "sdk" / "core").iterdir(): shutil.rmtree(sdk_out_dir / p.name, ignore_errors=True) p.replace(sdk_out_dir / p.name) + streams_out_dir.mkdir(exist_ok=True) + for p in (output_dir / "temporal" / "sdk" / "streams").iterdir(): + if p.is_dir(): + shutil.rmtree(streams_out_dir / p.name, ignore_errors=True) + p.replace(streams_out_dir / p.name) shutil.rmtree(sdk_out_dir / "health", ignore_errors=True) (output_dir / "health").replace(sdk_out_dir / "health") # Move test protos diff --git a/scripts/gen_protos_docker.py b/scripts/gen_protos_docker.py index 6735bdf90d..d4df953c01 100644 --- a/scripts/gen_protos_docker.py +++ b/scripts/gen_protos_docker.py @@ -34,6 +34,9 @@ os.path.join(os.getcwd(), "temporalio", "api") + ":/api_new", "-v", os.path.join(os.getcwd(), "temporalio", "bridge", "proto") + ":/bridge_new", + "-v", + os.path.join(os.getcwd(), "temporalio", "contrib", "streams", "proto") + + ":/streams_new", image_id, ] ) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py new file mode 100644 index 0000000000..116a87583f --- /dev/null +++ b/temporalio/contrib/streams/__init__.py @@ -0,0 +1,14 @@ +"""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, reached +through a provider, and never pass through Temporal or its History. + +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`` +as an ordinary payload. +""" diff --git a/temporalio/contrib/streams/proto/__init__.py b/temporalio/contrib/streams/proto/__init__.py new file mode 100644 index 0000000000..3100855386 --- /dev/null +++ b/temporalio/contrib/streams/proto/__init__.py @@ -0,0 +1 @@ +"""Generated protobuf models for the stream record envelope.""" diff --git a/temporalio/contrib/streams/proto/v1/__init__.py b/temporalio/contrib/streams/proto/v1/__init__.py new file mode 100644 index 0000000000..fc6311caae --- /dev/null +++ b/temporalio/contrib/streams/proto/v1/__init__.py @@ -0,0 +1,6 @@ +from .message_pb2 import StreamRecord, StreamRecordKind + +__all__ = [ + "StreamRecord", + "StreamRecordKind", +] diff --git a/temporalio/contrib/streams/proto/v1/message_pb2.py b/temporalio/contrib/streams/proto/v1/message_pb2.py new file mode 100644 index 0000000000..79699cb045 --- /dev/null +++ b/temporalio/contrib/streams/proto/v1/message_pb2.py @@ -0,0 +1,66 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/sdk/streams/v1/message.proto +"""Generated protocol buffer code.""" + +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import enum_type_wrapper + +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from temporalio.api.common.v1 import ( + message_pb2 as temporal_dot_api_dot_common_dot_v1_dot_message__pb2, +) + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n%temporal/sdk/streams/v1/message.proto\x12\x17temporal.sdk.streams.v1\x1a$temporal/api/common/v1/message.proto"\xd6\x02\n\x0cStreamRecord\x12-\n\x04\x62ody\x18\x01 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x45\n\x08metadata\x18\x02 \x03(\x0b\x32\x33.temporal.sdk.streams.v1.StreamRecord.MetadataEntry\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x37\n\x04kind\x18\x04 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12\x13\n\x0bproducer_id\x18\x05 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x06 \x01(\x03\x12\x10\n\x08sequence\x18\x07 \x01(\x03\x1aP\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12.\n\x05value\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload:\x02\x38\x01*r\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12\x1d\n\x19STREAM_RECORD_KIND_FINISH\x10\x02\x62\x06proto3' +) + +_STREAMRECORDKIND = DESCRIPTOR.enum_types_by_name["StreamRecordKind"] +StreamRecordKind = enum_type_wrapper.EnumTypeWrapper(_STREAMRECORDKIND) +STREAM_RECORD_KIND_UNSPECIFIED = 0 +STREAM_RECORD_KIND_DATA = 1 +STREAM_RECORD_KIND_FINISH = 2 + + +_STREAMRECORD = DESCRIPTOR.message_types_by_name["StreamRecord"] +_STREAMRECORD_METADATAENTRY = _STREAMRECORD.nested_types_by_name["MetadataEntry"] +StreamRecord = _reflection.GeneratedProtocolMessageType( + "StreamRecord", + (_message.Message,), + { + "MetadataEntry": _reflection.GeneratedProtocolMessageType( + "MetadataEntry", + (_message.Message,), + { + "DESCRIPTOR": _STREAMRECORD_METADATAENTRY, + "__module__": "temporal.sdk.streams.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.StreamRecord.MetadataEntry) + }, + ), + "DESCRIPTOR": _STREAMRECORD, + "__module__": "temporal.sdk.streams.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.StreamRecord) + }, +) +_sym_db.RegisterMessage(StreamRecord) +_sym_db.RegisterMessage(StreamRecord.MetadataEntry) + +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + _STREAMRECORD_METADATAENTRY._options = None + _STREAMRECORD_METADATAENTRY._serialized_options = b"8\001" + _STREAMRECORDKIND._serialized_start = 449 + _STREAMRECORDKIND._serialized_end = 563 + _STREAMRECORD._serialized_start = 105 + _STREAMRECORD._serialized_end = 447 + _STREAMRECORD_METADATAENTRY._serialized_start = 367 + _STREAMRECORD_METADATAENTRY._serialized_end = 447 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/contrib/streams/proto/v1/message_pb2.pyi b/temporalio/contrib/streams/proto/v1/message_pb2.pyi new file mode 100644 index 0000000000..d716502b86 --- /dev/null +++ b/temporalio/contrib/streams/proto/v1/message_pb2.pyi @@ -0,0 +1,161 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import collections.abc +import sys +import typing + +import google.protobuf.descriptor +import google.protobuf.internal.containers +import google.protobuf.internal.enum_type_wrapper +import google.protobuf.message + +import temporalio.api.common.v1.message_pb2 + +if sys.version_info >= (3, 10): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class _StreamRecordKind: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamRecordKindEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamRecordKind.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_RECORD_KIND_UNSPECIFIED: _StreamRecordKind.ValueType # 0 + """Read as DATA.""" + STREAM_RECORD_KIND_DATA: _StreamRecordKind.ValueType # 1 + """A value the producer published; `body` carries it.""" + STREAM_RECORD_KIND_FINISH: _StreamRecordKind.ValueType # 2 + """The producer named by `producer_id` writes nothing more on `topic`. + Says nothing about that producer's outcome and does not end the stream. + """ + +class StreamRecordKind(_StreamRecordKind, metaclass=_StreamRecordKindEnumTypeWrapper): + """What a record means to a reader.""" + +STREAM_RECORD_KIND_UNSPECIFIED: StreamRecordKind.ValueType # 0 +"""Read as DATA.""" +STREAM_RECORD_KIND_DATA: StreamRecordKind.ValueType # 1 +"""A value the producer published; `body` carries it.""" +STREAM_RECORD_KIND_FINISH: StreamRecordKind.ValueType # 2 +"""The producer named by `producer_id` writes nothing more on `topic`. +Says nothing about that producer's outcome and does not end the stream. +""" +global___StreamRecordKind = StreamRecordKind + +class StreamRecord(google.protobuf.message.Message): + """One entry in a stream. The record is the wire format: stores keep it + serialized as is and readers in every language decode the same bytes. + + Field numbers and enum values are part of the stored format. The package + name is not, so this message can move to another package without a data + change. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + class MetadataEntry(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + KEY_FIELD_NUMBER: builtins.int + VALUE_FIELD_NUMBER: builtins.int + key: builtins.str + @property + def value(self) -> temporalio.api.common.v1.message_pb2.Payload: ... + def __init__( + self, + *, + key: builtins.str = ..., + value: temporalio.api.common.v1.message_pb2.Payload | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["value", b"value"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal["key", b"key", "value", b"value"], + ) -> None: ... + + BODY_FIELD_NUMBER: builtins.int + METADATA_FIELD_NUMBER: builtins.int + TOPIC_FIELD_NUMBER: builtins.int + KIND_FIELD_NUMBER: builtins.int + PRODUCER_ID_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + SEQUENCE_FIELD_NUMBER: builtins.int + @property + def body(self) -> temporalio.api.common.v1.message_pb2.Payload: + """The value the producer published. A provider runs it through the + payload codec and external storage before it stores the record. + """ + @property + def metadata( + self, + ) -> google.protobuf.internal.containers.MessageMap[ + builtins.str, temporalio.api.common.v1.message_pb2.Payload + ]: + """Provenance stamped by the SDK, such as the plaintext content hash.""" + topic: builtins.str + """The topic the record was published on.""" + kind: global___StreamRecordKind.ValueType + """How to read this record. Unspecified is read as DATA.""" + producer_id: builtins.str + """Who wrote the record. Empty when the owning Workflow did.""" + attempt: builtins.int + """The producer's attempt. Readers treat a later attempt by the same + producer as superseding what the earlier one wrote. + """ + sequence: builtins.int + """The producer's position within its attempt, starting at one, or zero + when it does not number its records. + """ + def __init__( + self, + *, + body: temporalio.api.common.v1.message_pb2.Payload | None = ..., + metadata: collections.abc.Mapping[ + builtins.str, temporalio.api.common.v1.message_pb2.Payload + ] + | None = ..., + topic: builtins.str = ..., + kind: global___StreamRecordKind.ValueType = ..., + producer_id: builtins.str = ..., + attempt: builtins.int = ..., + sequence: builtins.int = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["body", b"body"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "attempt", + b"attempt", + "body", + b"body", + "kind", + b"kind", + "metadata", + b"metadata", + "producer_id", + b"producer_id", + "sequence", + b"sequence", + "topic", + b"topic", + ], + ) -> None: ... + +global___StreamRecord = StreamRecord 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_envelope_proto.py b/tests/contrib/streams/test_envelope_proto.py new file mode 100644 index 0000000000..cbb06920f8 --- /dev/null +++ b/tests/contrib/streams/test_envelope_proto.py @@ -0,0 +1,42 @@ +"""The stream record envelope keeps its stored format.""" + +from __future__ import annotations + +from temporalio.api.common.v1 import Payload +from temporalio.contrib.streams.proto.v1 import StreamRecord, StreamRecordKind + + +def test_the_envelope_keeps_its_field_numbers(): + # The bytes are the stored format; a renumbered field would make every + # retained record unreadable. + fields = {f.name: f.number for f in StreamRecord.DESCRIPTOR.fields} + assert fields == { + "body": 1, + "metadata": 2, + "topic": 3, + "kind": 4, + "producer_id": 5, + "attempt": 6, + "sequence": 7, + } + assert StreamRecord.DESCRIPTOR.full_name == "temporal.sdk.streams.v1.StreamRecord" + kinds = StreamRecord.DESCRIPTOR.fields_by_name["kind"].enum_type + assert kinds is not None + assert {v.name: v.number for v in kinds.values} == { + "STREAM_RECORD_KIND_UNSPECIFIED": 0, + "STREAM_RECORD_KIND_DATA": 1, + "STREAM_RECORD_KIND_FINISH": 2, + } + + +def test_the_envelope_round_trips_through_bytes(): + record = StreamRecord( + body=Payload(metadata={"encoding": b"json/plain"}, data=b"1"), + metadata={"k": Payload(data=b"v")}, + topic="out", + kind=StreamRecordKind.STREAM_RECORD_KIND_FINISH, + producer_id="p", + attempt=2, + sequence=3, + ) + assert StreamRecord.FromString(record.SerializeToString()) == record From 5fe3357e22ca4545ed8f6d8219c900839b5d3368 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 14:55:30 -0700 Subject: [PATCH 2/8] Added stream records and their wire mapping. Records map to the envelope proto in one place, and the body helpers hash the converted payload before the codec so a retry deduplicates. --- temporalio/contrib/streams/__init__.py | 33 ++++ temporalio/contrib/streams/_body.py | 124 +++++++++++++++ temporalio/contrib/streams/_policy.py | 63 ++++++++ temporalio/contrib/streams/_record.py | 109 +++++++++++++ temporalio/contrib/streams/_wire.py | 132 ++++++++++++++++ tests/contrib/streams/test_records_wire.py | 170 +++++++++++++++++++++ 6 files changed, 631 insertions(+) create mode 100644 temporalio/contrib/streams/_body.py create mode 100644 temporalio/contrib/streams/_policy.py create mode 100644 temporalio/contrib/streams/_record.py create mode 100644 temporalio/contrib/streams/_wire.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 index 116a87583f..3d98a6b03a 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -8,7 +8,40 @@ it from a cursor. Records live in a store the application runs, reached through a provider, and never pass through Temporal or its History. +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 provider reports a retry the same way. + 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`` as an ordinary payload. """ + +from temporalio.contrib.streams._body import ( + CONTENT_HASH_KEY, + content_fingerprint, + content_hash, + decode_body, + encode_body, +) +from temporalio.contrib.streams._record import ( + Cursor, + RecordKind, + StreamRecord, + Supersession, +) + +__all__ = [ + "CONTENT_HASH_KEY", + "Cursor", + "RecordKind", + "StreamRecord", + "Supersession", + "content_fingerprint", + "content_hash", + "decode_body", + "encode_body", +] diff --git a/temporalio/contrib/streams/_body.py b/temporalio/contrib/streams/_body.py new file mode 100644 index 0000000000..56b51d2638 --- /dev/null +++ b/temporalio/contrib/streams/_body.py @@ -0,0 +1,124 @@ +"""What a provider owes a record's body between the converter and its store. + +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. A provider runs the +body through :func:`encode_body` before it stores a record and through +:func:`decode_body` after it reads one back. + +The order inside :func:`encode_body` is the point. The plaintext hash is +taken first and stamped on the record, and :func:`content_fingerprint` is +taken over the converted records too, before the codec runs. 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.contrib.streams.proto.v1 import StreamRecord as WireRecord +from temporalio.converter import DataConverter + +__all__ = [ + "CONTENT_HASH_KEY", + "content_fingerprint", + "content_hash", + "decode_body", + "encode_body", + "stamp_content_hash", +] + +CONTENT_HASH_KEY = "temporal.io/content-hash" +"""The record metadata key the plaintext hash of the body is stored under. + +Its value is a payload with ``encoding`` ``binary/plain`` whose data is the +hex digest :func:`content_hash` returns. A record with no body carries no +hash. +""" + +_HASH_ENCODING = b"binary/plain" + + +def content_hash(payload: Payload) -> str: + """The hex 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. + """ + return hashlib.sha256(payload.SerializeToString(deterministic=True)).hexdigest() + + +def content_fingerprint(records: Sequence[WireRecord]) -> bytes: + """The identity of one append, taken over its converted records. + + Length-delimited, so a batch split differently cannot collide with this + one. Take it before :func:`encode_body`, while the bodies are still what + the converter produced, so a retry through a codec with a fresh nonce + matches its original. + """ + 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() + + +def stamp_content_hash(record: WireRecord) -> WireRecord: + """Stamp the plaintext hash of ``record``'s body under :data:`CONTENT_HASH_KEY`. + + In place, and returned for convenience. Synchronous, so it can run where + the codec cannot, such as on the Workflow thread. A record without a body + is returned untouched. + """ + if record.HasField("body"): + record.metadata[CONTENT_HASH_KEY].CopyFrom( + Payload( + metadata={"encoding": _HASH_ENCODING}, + data=content_hash(record.body).encode(), + ) + ) + return record + + +async def encode_body(converter: DataConverter, record: WireRecord) -> WireRecord: + """Stamp the plaintext hash on ``record`` and encode its body for the store. + + In place, and returned for convenience. The hash goes under + :data:`CONTENT_HASH_KEY` first. Then the body passes through + ``converter``'s payload codec and external storage, in the order + :meth:`temporalio.converter.DataConverter.encode` uses, so a body above + the external storage threshold is replaced by a claim and the store holds + the claim. A record without a body is returned untouched. + """ + if not record.HasField("body"): + return record + stamp_content_hash(record) + encoded = await converter._encode_payload_sequence([record.body]) + stored = await converter._external_store_payload_sequence(encoded) + record.body.CopyFrom(stored[0]) + return record + + +async def decode_body(converter: DataConverter, record: WireRecord) -> WireRecord: + """Undo :func:`encode_body` on a record read back from the store. + + In place, and returned for convenience. The body is retrieved from + external storage when it is a claim and then run through the payload + codec, in the order :meth:`temporalio.converter.DataConverter.decode` + uses. The hash stays on the record. + + Raises: + RuntimeError: The body is a claim and ``converter`` has no external + storage to redeem it with. + """ + if not record.HasField("body"): + return record + retrieved = await converter._external_retrieve_payload_sequence([record.body]) + decoded = await converter._decode_payload_sequence(retrieved) + record.body.CopyFrom(decoded[0]) + return record diff --git a/temporalio/contrib/streams/_policy.py b/temporalio/contrib/streams/_policy.py new file mode 100644 index 0000000000..0fcf4db5da --- /dev/null +++ b/temporalio/contrib/streams/_policy.py @@ -0,0 +1,63 @@ +"""Turning a producer's newer attempt into something a reader can act on. + +An Activity that streams half an answer and then fails leaves those records +in the stream. Its retry writes different ones. No provider can undo the +first half, so the reader is told that a new attempt began and the +application decides what to do. + +This runs in the reader over records it already observed, so it costs no +round trip and every provider reports a retry the same way. +""" + +from __future__ import annotations + +from typing import Any + +from temporalio.contrib.streams._record import ( + Cursor, + RecordKind, + StreamRecord, + Supersession, +) + +__all__ = ["AttemptTracker"] + + +class AttemptTracker: + """Watches producer attempts on one read.""" + + def __init__(self) -> None: + """Start with no producer seen.""" + self._attempts: dict[str, int] = {} + + def note( + self, producer_id: str, attempt: int, *, topic: str, previous: Cursor + ) -> StreamRecord[Any] | None: + """A ``SUPERSEDED`` record when this record starts a newer attempt. + + ``previous`` is the cursor of the last record delivered before the + one being noted, or the cursor the read started from. The synthesized + record carries it, so a consumer that checkpoints the supersession and + resumes after it gets the new attempt's first record next. + + A producer that declares no attempt supersedes nothing, because there + is no generation to compare. + """ + if not producer_id or attempt <= 0: + return None + seen = self._attempts.get(producer_id, 0) + if attempt <= seen: + return None + self._attempts[producer_id] = attempt + if seen == 0: + return None + return StreamRecord( + kind=RecordKind.SUPERSEDED, + cursor=previous, + topic=topic, + producer_id=producer_id, + attempt=attempt, + supersession=Supersession( + producer_id=producer_id, previous_attempt=seen, attempt=attempt + ), + ) diff --git a/temporalio/contrib/streams/_record.py b/temporalio/contrib/streams/_record.py new file mode 100644 index 0000000000..268a1517e2 --- /dev/null +++ b/temporalio/contrib/streams/_record.py @@ -0,0 +1,109 @@ +"""The value types the stream contract is expressed in. + +Nothing here touches Temporal or a provider, so every provider 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. 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. + + The reader synthesizes it from the records it observed. No store holds + it, so every provider reports a retry the same way. Its cursor is the + position before the new attempt's first record, so a reader that resumes + after it gets that record next. + """ + + +@dataclass(frozen=True) +class Cursor: + """A position in one stream. + + Opaque on purpose. The token names the provider that minted it and the + stream it belongs to, and that provider refuses a token from another + provider or another stream 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 provider'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.""" + value: T | None = None + """The published value. Set on ``DATA`` only.""" + supersession: Supersession | None = None + """The attempt change being reported. Set on ``SUPERSEDED`` only.""" diff --git a/temporalio/contrib/streams/_wire.py b/temporalio/contrib/streams/_wire.py new file mode 100644 index 0000000000..6741e4943d --- /dev/null +++ b/temporalio/contrib/streams/_wire.py @@ -0,0 +1,132 @@ +"""How a record crosses a provider: the proto is the record. + +``temporal.sdk.streams.v1.StreamRecord`` is the wire format on every +provider. A store that keeps bytes keeps ``SerializeToString()`` of it, and a +reader in any language parses the same bytes. ``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 collections.abc import Callable +from typing import Any + +import temporalio.converter +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 +from temporalio.contrib.streams.proto.v1 import StreamRecordKind + +__all__ = ["RecordDecoder", "WireRecord", "from_wire", "to_wire"] + + +def to_wire( + converter: temporalio.converter.PayloadConverter, + *, + topic: str, + kind: RecordKind, + value: Any = None, + producer_id: str = "", + attempt: int = 0, + sequence: int = 0, +) -> WireRecord: + """Build the record a provider stores. + + 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 + + +def from_wire( + converter: temporalio.converter.PayloadConverter, + cursor: Cursor, + wire: WireRecord, + result_type: type | None, +) -> StreamRecord[Any]: + """Turn a stored record into the record a reader yields. + + Raises: + ValueError: The kind is one no store may hold, such as a synthesized + ``SUPERSEDED`` or a value this SDK does not know. + """ + kind = RecordKind(wire.kind) + if kind is RecordKind.SUPERSEDED: + raise ValueError("a SUPERSEDED record is synthesized by readers, never stored") + if kind is RecordKind.UNSPECIFIED: + kind = RecordKind.DATA + value: Any = None + if kind is RecordKind.DATA and wire.HasField("body"): + hints = [result_type] if result_type is not None else None + value = converter.from_payloads([wire.body], hints)[0] + return StreamRecord( + kind=kind, + cursor=cursor, + topic=wire.topic, + producer_id=wire.producer_id, + attempt=wire.attempt, + sequence=wire.sequence, + value=value, + ) + + +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. + """ + + def __init__( + self, + converter: temporalio.converter.PayloadConverter, + result_type: type | None, + *, + after: Cursor, + warn: Callable[[str], None], + ) -> None: + """Decode with ``converter`` into ``result_type``, resuming after ``after``.""" + self._converter = converter + self._result_type = result_type + self._previous = after + self._warn = warn + self._attempts = AttemptTracker() + + def decode(self, cursor: Cursor, wire: WireRecord) -> list[StreamRecord[Any]]: + """The records to yield for one stored record, in order.""" + 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 [] + out: list[StreamRecord[Any]] = [] + superseded = self._attempts.note( + wire.producer_id, wire.attempt, topic=wire.topic, previous=self._previous + ) + if superseded is not None: + out.append(superseded) + out.append(record) + self._previous = cursor + return out diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py new file mode 100644 index 0000000000..6008fec8ba --- /dev/null +++ b/tests/contrib/streams/test_records_wire.py @@ -0,0 +1,170 @@ +"""Records, their wire form, and what a provider 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.common import RawValue +from temporalio.contrib.streams import ( + CONTENT_HASH_KEY, + Cursor, + RecordKind, + Supersession, + content_fingerprint, + content_hash, + decode_body, + encode_body, +) +from temporalio.contrib.streams._policy import AttemptTracker +from temporalio.contrib.streams._wire import RecordDecoder, from_wire, to_wire +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 + wire = to_wire( + converter, + topic="out", + kind=RecordKind.DATA, + value={"n": 1}, + producer_id="p", + attempt=2, + sequence=3, + ) + parsed = WireRecord.FromString(wire.SerializeToString()) + record = from_wire(converter, Cursor("c"), parsed, dict) + assert record.kind is RecordKind.DATA + assert record.value == {"n": 1} + assert (record.producer_id, record.attempt, record.sequence) == ("p", 2, 3) + + finish = to_wire(converter, topic="out", kind=RecordKind.FINISH) + assert not finish.HasField("body") + assert from_wire(converter, Cursor("c"), finish, None).kind is RecordKind.FINISH + + unset = WireRecord(topic="out", body=converter.to_payloads([1])[0]) + assert from_wire(converter, Cursor("c"), unset, int).kind is RecordKind.DATA + + +def test_no_writer_stores_a_synthesized_kind(): + converter = DataConverter.default.payload_converter + for kind in (RecordKind.SUPERSEDED, RecordKind.UNSPECIFIED): + with pytest.raises(ValueError): + to_wire(converter, topic="out", kind=kind) + + +def test_a_raw_value_passes_through_untouched(): + converter = DataConverter.default.payload_converter + raw = Payload(metadata={"encoding": b"binary/custom"}, data=b"\x00\x01") + wire = to_wire(converter, topic="out", kind=RecordKind.DATA, value=RawValue(raw)) + assert wire.body == raw + record = from_wire(converter, Cursor("c"), wire, RawValue) + assert isinstance(record.value, RawValue) + assert record.value.payload == raw + + +def test_supersession_is_synthesized_from_observations(): + attempts = AttemptTracker() + assert attempts.note("model", 1, topic="t", previous=BEGINNING) is None + assert attempts.note("model", 1, topic="t", previous=Cursor("c1")) is None + superseded = attempts.note("model", 2, topic="t", previous=Cursor("c2")) + assert superseded is not None + assert superseded.kind is RecordKind.SUPERSEDED + assert superseded.cursor == Cursor("c2") + assert superseded.supersession == Supersession("model", 1, 2) + assert attempts.note("model", 2, topic="t", previous=Cursor("c3")) is None + # Without an id or an attempt there is no generation to compare. + assert attempts.note("", 5, topic="t", previous=Cursor("c4")) is None + 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] = [] + converter = DataConverter.default.payload_converter + decoder = RecordDecoder(converter, int, after=BEGINNING, warn=said.append) + 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] + good = to_wire(converter, topic="t", kind=RecordKind.DATA, value=3) + assert [r.value for r in decoder.decode(Cursor("c2"), good)] == [3] + + +def test_the_content_hash_is_the_plaintext_payload_hash(): + converter = DataConverter.default.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]) + assert len(content_hash(one)) == 64 + + +def test_the_fingerprint_is_length_delimited_over_the_batch(): + converter = DataConverter.default.payload_converter + + def records(*values: str) -> list[WireRecord]: + return [ + to_wire(converter, topic="t", kind=RecordKind.DATA, value=v) for v in values + ] + + assert content_fingerprint(records("a", "b")) == content_fingerprint( + records("a", "b") + ) + assert content_fingerprint(records("a", "b")) != content_fingerprint( + records("b", "a") + ) + assert content_fingerprint(records("ab")) != content_fingerprint(records("a", "b")) + + +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_body_is_hashed_before_the_codec_and_decoded_after(): + converter = DataConverter(payload_codec=NonceCodec()) + plain = to_wire( + converter.payload_converter, topic="t", kind=RecordKind.DATA, value={"n": 1} + ) + expected = content_hash(plain.body) + + first = WireRecord() + first.CopyFrom(plain) + second = WireRecord() + second.CopyFrom(plain) + await encode_body(converter, first) + await encode_body(converter, second) + # The codec made the stored bytes differ, and the hash did not move. + assert first.body != second.body + assert first.body.metadata["encoding"] == b"binary/nonce" + for stamped in (first, second): + assert stamped.metadata[CONTENT_HASH_KEY].data.decode() == expected + assert stamped.metadata[CONTENT_HASH_KEY].metadata["encoding"] == ( + b"binary/plain" + ) + + await decode_body(converter, first) + assert first.body == plain.body + + finish = to_wire(converter.payload_converter, topic="t", kind=RecordKind.FINISH) + await encode_body(converter, finish) + assert CONTENT_HASH_KEY not in finish.metadata From 9225fcde9f47ab7d2f3edf356ca5abe8c2c8b362 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 17:17:07 -0700 Subject: [PATCH 3/8] Marked the generated stream envelope package experimental. The generated v1 modules are rewritten on every proto run, so the warning goes on the hand-written package above them. --- temporalio/contrib/streams/proto/__init__.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/temporalio/contrib/streams/proto/__init__.py b/temporalio/contrib/streams/proto/__init__.py index 3100855386..1ede5611cf 100644 --- a/temporalio/contrib/streams/proto/__init__.py +++ b/temporalio/contrib/streams/proto/__init__.py @@ -1 +1,8 @@ -"""Generated protobuf models for the stream record envelope.""" +"""Generated protobuf models for the stream record envelope. + +.. warning:: + This package is experimental and may change in future versions. + +The modules under ``v1`` are generated by ``scripts/gen_protos.py`` and +rewritten on every run, so this note lives here rather than in them. +""" From 0e4425e6355ace2f030f32af7c7f4ccccc89befc Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 17:18:07 -0700 Subject: [PATCH 4/8] Kept the stream body helpers out of the public surface. Only provider code uses them, so exporting them would commit the package to an interface no application needs. --- temporalio/contrib/streams/__init__.py | 12 ----------- tests/contrib/streams/test_public_surface.py | 21 ++++++++++++++++++++ tests/contrib/streams/test_records_wire.py | 4 +++- 3 files changed, 24 insertions(+), 13 deletions(-) create mode 100644 tests/contrib/streams/test_public_surface.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 3d98a6b03a..f8b9f74294 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -20,13 +20,6 @@ as an ordinary payload. """ -from temporalio.contrib.streams._body import ( - CONTENT_HASH_KEY, - content_fingerprint, - content_hash, - decode_body, - encode_body, -) from temporalio.contrib.streams._record import ( Cursor, RecordKind, @@ -35,13 +28,8 @@ ) __all__ = [ - "CONTENT_HASH_KEY", "Cursor", "RecordKind", "StreamRecord", "Supersession", - "content_fingerprint", - "content_hash", - "decode_body", - "encode_body", ] diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py new file mode 100644 index 0000000000..76a1e09099 --- /dev/null +++ b/tests/contrib/streams/test_public_surface.py @@ -0,0 +1,21 @@ +"""The package exports only what applications use.""" + +from __future__ import annotations + +import temporalio.contrib.streams as streams + +# Provider-author helpers stay importable from their private modules, but +# they are not part of the public surface. +_PRIVATE = { + "CONTENT_HASH_KEY", + "content_fingerprint", + "content_hash", + "decode_body", + "encode_body", +} + + +def test_provider_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 index 6008fec8ba..0fed5fd2b9 100644 --- a/tests/contrib/streams/test_records_wire.py +++ b/tests/contrib/streams/test_records_wire.py @@ -10,10 +10,12 @@ from temporalio.api.common.v1 import Payload from temporalio.common import RawValue from temporalio.contrib.streams import ( - CONTENT_HASH_KEY, Cursor, RecordKind, Supersession, +) +from temporalio.contrib.streams._body import ( + CONTENT_HASH_KEY, content_fingerprint, content_hash, decode_body, From 80b50701d487303687281f0769f88ddced0e2765 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 12:56:43 -0700 Subject: [PATCH 5/8] Reserved the record kind readers synthesize for SUPERSEDED. A stored kind at 3 would decode as SUPERSEDED, which readers refuse. The format test now pins field types and labels too, since a retyped field breaks stored records. --- scripts/_proto/temporal/sdk/streams/v1/message.proto | 3 +++ temporalio/contrib/streams/proto/v1/message_pb2.py | 4 ++-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/scripts/_proto/temporal/sdk/streams/v1/message.proto b/scripts/_proto/temporal/sdk/streams/v1/message.proto index 514a724343..23b19017d1 100644 --- a/scripts/_proto/temporal/sdk/streams/v1/message.proto +++ b/scripts/_proto/temporal/sdk/streams/v1/message.proto @@ -39,4 +39,7 @@ enum StreamRecordKind { // The producer named by `producer_id` writes nothing more on `topic`. // Says nothing about that producer's outcome and does not end the stream. STREAM_RECORD_KIND_FINISH = 2; + // A reader synthesizes SUPERSEDED with this value and never stores it, so + // no stored kind may take it. + reserved 3; } diff --git a/temporalio/contrib/streams/proto/v1/message_pb2.py b/temporalio/contrib/streams/proto/v1/message_pb2.py index 79699cb045..2df6179aa0 100644 --- a/temporalio/contrib/streams/proto/v1/message_pb2.py +++ b/temporalio/contrib/streams/proto/v1/message_pb2.py @@ -20,7 +20,7 @@ ) DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( - b'\n%temporal/sdk/streams/v1/message.proto\x12\x17temporal.sdk.streams.v1\x1a$temporal/api/common/v1/message.proto"\xd6\x02\n\x0cStreamRecord\x12-\n\x04\x62ody\x18\x01 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x45\n\x08metadata\x18\x02 \x03(\x0b\x32\x33.temporal.sdk.streams.v1.StreamRecord.MetadataEntry\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x37\n\x04kind\x18\x04 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12\x13\n\x0bproducer_id\x18\x05 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x06 \x01(\x03\x12\x10\n\x08sequence\x18\x07 \x01(\x03\x1aP\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12.\n\x05value\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload:\x02\x38\x01*r\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12\x1d\n\x19STREAM_RECORD_KIND_FINISH\x10\x02\x62\x06proto3' + b'\n%temporal/sdk/streams/v1/message.proto\x12\x17temporal.sdk.streams.v1\x1a$temporal/api/common/v1/message.proto"\xd6\x02\n\x0cStreamRecord\x12-\n\x04\x62ody\x18\x01 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x45\n\x08metadata\x18\x02 \x03(\x0b\x32\x33.temporal.sdk.streams.v1.StreamRecord.MetadataEntry\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x37\n\x04kind\x18\x04 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12\x13\n\x0bproducer_id\x18\x05 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x06 \x01(\x03\x12\x10\n\x08sequence\x18\x07 \x01(\x03\x1aP\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12.\n\x05value\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload:\x02\x38\x01*x\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12\x1d\n\x19STREAM_RECORD_KIND_FINISH\x10\x02"\x04\x08\x03\x10\x03\x62\x06proto3' ) _STREAMRECORDKIND = DESCRIPTOR.enum_types_by_name["StreamRecordKind"] @@ -58,7 +58,7 @@ _STREAMRECORD_METADATAENTRY._options = None _STREAMRECORD_METADATAENTRY._serialized_options = b"8\001" _STREAMRECORDKIND._serialized_start = 449 - _STREAMRECORDKIND._serialized_end = 563 + _STREAMRECORDKIND._serialized_end = 569 _STREAMRECORD._serialized_start = 105 _STREAMRECORD._serialized_end = 447 _STREAMRECORD_METADATAENTRY._serialized_start = 367 From 9376eae6887b8ce0e81d1048585ca4515209472b Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 12:59:34 -0700 Subject: [PATCH 6/8] Let a reader prime its attempt tracker at the resume cursor. A resumed read started with no attempts seen, so a retry right after the cursor reached the reader as plain DATA with no SUPERSEDED. --- temporalio/contrib/streams/_wire.py | 12 ++++++++++++ tests/contrib/streams/test_records_wire.py | 16 ++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/temporalio/contrib/streams/_wire.py b/temporalio/contrib/streams/_wire.py index 6741e4943d..e52850cdbb 100644 --- a/temporalio/contrib/streams/_wire.py +++ b/temporalio/contrib/streams/_wire.py @@ -111,6 +111,18 @@ def __init__( self._warn = warn self._attempts = AttemptTracker() + def prime(self, wire: WireRecord) -> None: + """Note the attempt of the record at the resume cursor, which was already delivered. + + A read that resumes would otherwise take the next attempt of that + producer as its first, and report no ``SUPERSEDED``. Only the cursor + record's producer is primed. Another producer's earlier attempts are + not known to a resumed read. + """ + self._attempts.note( + wire.producer_id, wire.attempt, topic=wire.topic, previous=self._previous + ) + def decode(self, cursor: Cursor, wire: WireRecord) -> list[StreamRecord[Any]]: """The records to yield for one stored record, in order.""" try: diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py index 0fed5fd2b9..a379ade57e 100644 --- a/tests/contrib/streams/test_records_wire.py +++ b/tests/contrib/streams/test_records_wire.py @@ -101,6 +101,22 @@ def test_the_decoder_skips_a_record_it_cannot_decode(): assert [r.value for r in decoder.decode(Cursor("c2"), good)] == [3] +def test_a_decoder_primed_at_its_resume_cursor_reports_a_new_attempt(): + converter = DataConverter.default.payload_converter + decoder = RecordDecoder(converter, int, after=Cursor("c2"), warn=lambda _: None) + + def written(attempt: int, value: int) -> WireRecord: + wire = to_wire(converter, topic="t", kind=RecordKind.DATA, value=value) + wire.producer_id, wire.attempt = "model", attempt + return wire + + # The record at the cursor was already delivered, before the reader stopped. + decoder.prime(written(1, 2)) + out = decoder.decode(Cursor("c3"), written(2, 3)) + assert [r.kind for r in out] == [RecordKind.SUPERSEDED, RecordKind.DATA] + assert out[0].cursor == Cursor("c2") + + def test_the_content_hash_is_the_plaintext_payload_hash(): converter = DataConverter.default.payload_converter one = converter.to_payloads([{"n": 1}])[0] From 165bdf9b3673f63bebd069e7582c934fb82a9b1d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 22:08:35 -0700 Subject: [PATCH 7/8] Rebuilt stream records and their hashes on Core's envelope. Core now owns the envelope proto and builds every stored record, so lang keeps conversion and the two hashes that need the plaintext. The content hash is raw bytes for Core, and the batch digest feeds the store's dedupe. --- pyproject.toml | 2 - scripts/_proto/Dockerfile | 4 +- .../temporal/sdk/streams/v1/message.proto | 45 ---- scripts/gen_protos.py | 1 - scripts/gen_protos_docker.py | 3 - temporalio/contrib/streams/__init__.py | 10 +- temporalio/contrib/streams/_body.py | 110 +++------ temporalio/contrib/streams/_policy.py | 63 ----- temporalio/contrib/streams/_record.py | 38 ++- temporalio/contrib/streams/_wire.py | 137 +++++------ temporalio/contrib/streams/proto/__init__.py | 8 - .../contrib/streams/proto/v1/__init__.py | 6 - .../contrib/streams/proto/v1/message_pb2.py | 66 ----- .../contrib/streams/proto/v1/message_pb2.pyi | 161 ------------- tests/contrib/streams/test_envelope_proto.py | 42 ---- tests/contrib/streams/test_public_surface.py | 11 +- tests/contrib/streams/test_records_wire.py | 228 +++++++++--------- 17 files changed, 232 insertions(+), 703 deletions(-) delete mode 100644 scripts/_proto/temporal/sdk/streams/v1/message.proto delete mode 100644 temporalio/contrib/streams/_policy.py delete mode 100644 temporalio/contrib/streams/proto/__init__.py delete mode 100644 temporalio/contrib/streams/proto/v1/__init__.py delete mode 100644 temporalio/contrib/streams/proto/v1/message_pb2.py delete mode 100644 temporalio/contrib/streams/proto/v1/message_pb2.pyi delete mode 100644 tests/contrib/streams/test_envelope_proto.py diff --git a/pyproject.toml b/pyproject.toml index a09b044a1a..444e4d0c5b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -182,7 +182,6 @@ exclude = [ # Ignore generated code 'temporalio/api', 'temporalio/bridge/proto', - 'temporalio/contrib/streams/proto', 'temporalio/nexus/system/workflow_service', ] @@ -259,7 +258,6 @@ exclude = [ "temporalio/api", "temporalio/bridge/proto", "temporalio/bridge/_visitor.py", - "temporalio/contrib/streams/proto", "tests/worker/workflow_sandbox/testmodules/proto", ] diff --git a/scripts/_proto/Dockerfile b/scripts/_proto/Dockerfile index 16e90bb400..0f0fb74b06 100644 --- a/scripts/_proto/Dockerfile +++ b/scripts/_proto/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.10 RUN curl -LsSf https://astral.sh/uv/install.sh | sh ENV PATH="$PATH:/root/.local/bin" RUN uv tool install poethepoet -VOLUME ["/api_new", "/bridge_new", "/streams_new"] +VOLUME ["/api_new", "/bridge_new"] COPY ./ ./ @@ -13,4 +13,4 @@ RUN uv add "protobuf<4" RUN uv sync --all-extras RUN uv run scripts/gen_protos.py -CMD ["sh", "-c", "cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new && cp -r ./temporalio/contrib/streams/proto/* /streams_new"] +CMD ["sh", "-c", "cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new"] diff --git a/scripts/_proto/temporal/sdk/streams/v1/message.proto b/scripts/_proto/temporal/sdk/streams/v1/message.proto deleted file mode 100644 index 23b19017d1..0000000000 --- a/scripts/_proto/temporal/sdk/streams/v1/message.proto +++ /dev/null @@ -1,45 +0,0 @@ -syntax = "proto3"; - -package temporal.sdk.streams.v1; - -import "temporal/api/common/v1/message.proto"; - -// One entry in a stream. The record is the wire format: stores keep it -// serialized as is and readers in every language decode the same bytes. -// -// Field numbers and enum values are part of the stored format. The package -// name is not, so this message can move to another package without a data -// change. -message StreamRecord { - // The value the producer published. A provider runs it through the - // payload codec and external storage before it stores the record. - temporal.api.common.v1.Payload body = 1; - // Provenance stamped by the SDK, such as the plaintext content hash. - map metadata = 2; - // The topic the record was published on. - string topic = 3; - // How to read this record. Unspecified is read as DATA. - StreamRecordKind kind = 4; - // Who wrote the record. Empty when the owning Workflow did. - string producer_id = 5; - // The producer's attempt. Readers treat a later attempt by the same - // producer as superseding what the earlier one wrote. - int64 attempt = 6; - // The producer's position within its attempt, starting at one, or zero - // when it does not number its records. - int64 sequence = 7; -} - -// What a record means to a reader. -enum StreamRecordKind { - // Read as DATA. - STREAM_RECORD_KIND_UNSPECIFIED = 0; - // A value the producer published; `body` carries it. - STREAM_RECORD_KIND_DATA = 1; - // The producer named by `producer_id` writes nothing more on `topic`. - // Says nothing about that producer's outcome and does not end the stream. - STREAM_RECORD_KIND_FINISH = 2; - // A reader synthesizes SUPERSEDED with this value and never stores it, so - // no stored kind may take it. - reserved 3; -} diff --git a/scripts/gen_protos.py b/scripts/gen_protos.py index 9dab37ea5e..b75e7cdb5b 100644 --- a/scripts/gen_protos.py +++ b/scripts/gen_protos.py @@ -33,7 +33,6 @@ api_out_dir = base_dir / "temporalio" / "api" sdk_out_dir = base_dir / "temporalio" / "bridge" / "proto" -streams_out_dir = base_dir / "temporalio" / "contrib" / "streams" / "proto" py_fixes = [ partial(re.compile(r"from temporal\.api\.").sub, r"from temporalio.api."), diff --git a/scripts/gen_protos_docker.py b/scripts/gen_protos_docker.py index d4df953c01..6735bdf90d 100644 --- a/scripts/gen_protos_docker.py +++ b/scripts/gen_protos_docker.py @@ -34,9 +34,6 @@ os.path.join(os.getcwd(), "temporalio", "api") + ":/api_new", "-v", os.path.join(os.getcwd(), "temporalio", "bridge", "proto") + ":/bridge_new", - "-v", - os.path.join(os.getcwd(), "temporalio", "contrib", "streams", "proto") - + ":/streams_new", image_id, ] ) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index f8b9f74294..ccf428fc74 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -5,18 +5,18 @@ 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, reached -through a provider, and never pass through Temporal or its History. +it from a cursor. Records live in a store the application runs, which +Core reaches for every SDK, and never pass through Temporal or its History. 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 provider reports a retry the same way. + so every store reports a retry the same way. -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`` +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. """ diff --git a/temporalio/contrib/streams/_body.py b/temporalio/contrib/streams/_body.py index 56b51d2638..4ed14ee6f1 100644 --- a/temporalio/contrib/streams/_body.py +++ b/temporalio/contrib/streams/_body.py @@ -1,17 +1,13 @@ -"""What a provider owes a record's body between the converter and its store. +"""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. A provider runs the -body through :func:`encode_body` before it stores a record and through -:func:`decode_body` after it reads one back. - -The order inside :func:`encode_body` is the point. The plaintext hash is -taken first and stamped on the record, and :func:`content_fingerprint` is -taken over the converted records too, before the codec runs. 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. +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 @@ -20,45 +16,29 @@ from collections.abc import Sequence from temporalio.api.common.v1 import Payload -from temporalio.contrib.streams.proto.v1 import StreamRecord as WireRecord +from temporalio.bridge.proto.streams.v1 import StreamRecord as WireRecord from temporalio.converter import DataConverter -__all__ = [ - "CONTENT_HASH_KEY", - "content_fingerprint", - "content_hash", - "decode_body", - "encode_body", - "stamp_content_hash", -] - -CONTENT_HASH_KEY = "temporal.io/content-hash" -"""The record metadata key the plaintext hash of the body is stored under. - -Its value is a payload with ``encoding`` ``binary/plain`` whose data is the -hex digest :func:`content_hash` returns. A record with no body carries no -hash. -""" - -_HASH_ENCODING = b"binary/plain" +__all__ = ["batch_digest", "content_hash", "decode_body", "encode_bodies"] -def content_hash(payload: Payload) -> str: - """The hex SHA-256 of ``payload`` as the converter produced it. +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)).hexdigest() + return hashlib.sha256(payload.SerializeToString(deterministic=True)).digest() -def content_fingerprint(records: Sequence[WireRecord]) -> bytes: +def batch_digest(records: Sequence[WireRecord]) -> bytes: """The identity of one append, taken over its converted records. - Length-delimited, so a batch split differently cannot collide with this - one. Take it before :func:`encode_body`, while the bodies are still what - the converter produced, so a retry through a codec with a fresh nonce - matches its original. + 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. + Every SDK takes it the same way, and the store compares it when a + producer repeats its newest batch. """ digest = hashlib.sha256() for record in records: @@ -68,57 +48,25 @@ def content_fingerprint(records: Sequence[WireRecord]) -> bytes: return digest.digest() -def stamp_content_hash(record: WireRecord) -> WireRecord: - """Stamp the plaintext hash of ``record``'s body under :data:`CONTENT_HASH_KEY`. +async def encode_bodies( + converter: DataConverter, payloads: Sequence[Payload] +) -> list[Payload]: + """``payloads`` as the store keeps them. - In place, and returned for convenience. Synchronous, so it can run where - the codec cannot, such as on the Workflow thread. A record without a body - is returned untouched. - """ - if record.HasField("body"): - record.metadata[CONTENT_HASH_KEY].CopyFrom( - Payload( - metadata={"encoding": _HASH_ENCODING}, - data=content_hash(record.body).encode(), - ) - ) - return record - - -async def encode_body(converter: DataConverter, record: WireRecord) -> WireRecord: - """Stamp the plaintext hash on ``record`` and encode its body for the store. - - In place, and returned for convenience. The hash goes under - :data:`CONTENT_HASH_KEY` first. Then the body passes through - ``converter``'s payload codec and external storage, in the order + 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 and the store holds - the claim. A record without a body is returned untouched. + the external storage threshold is replaced by a claim. """ - if not record.HasField("body"): - return record - stamp_content_hash(record) - encoded = await converter._encode_payload_sequence([record.body]) - stored = await converter._external_store_payload_sequence(encoded) - record.body.CopyFrom(stored[0]) - return record - + encoded = await converter._encode_payload_sequence(payloads) + return await converter._external_store_payload_sequence(encoded) -async def decode_body(converter: DataConverter, record: WireRecord) -> WireRecord: - """Undo :func:`encode_body` on a record read back from the store. - In place, and returned for convenience. The body is retrieved from - external storage when it is a claim and then run through the payload - codec, in the order :meth:`temporalio.converter.DataConverter.decode` - uses. The hash stays on the record. +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. """ - if not record.HasField("body"): - return record - retrieved = await converter._external_retrieve_payload_sequence([record.body]) - decoded = await converter._decode_payload_sequence(retrieved) - record.body.CopyFrom(decoded[0]) - return record + retrieved = await converter._external_retrieve_payload_sequence([payload]) + return (await converter._decode_payload_sequence(retrieved))[0] diff --git a/temporalio/contrib/streams/_policy.py b/temporalio/contrib/streams/_policy.py deleted file mode 100644 index 0fcf4db5da..0000000000 --- a/temporalio/contrib/streams/_policy.py +++ /dev/null @@ -1,63 +0,0 @@ -"""Turning a producer's newer attempt into something a reader can act on. - -An Activity that streams half an answer and then fails leaves those records -in the stream. Its retry writes different ones. No provider can undo the -first half, so the reader is told that a new attempt began and the -application decides what to do. - -This runs in the reader over records it already observed, so it costs no -round trip and every provider reports a retry the same way. -""" - -from __future__ import annotations - -from typing import Any - -from temporalio.contrib.streams._record import ( - Cursor, - RecordKind, - StreamRecord, - Supersession, -) - -__all__ = ["AttemptTracker"] - - -class AttemptTracker: - """Watches producer attempts on one read.""" - - def __init__(self) -> None: - """Start with no producer seen.""" - self._attempts: dict[str, int] = {} - - def note( - self, producer_id: str, attempt: int, *, topic: str, previous: Cursor - ) -> StreamRecord[Any] | None: - """A ``SUPERSEDED`` record when this record starts a newer attempt. - - ``previous`` is the cursor of the last record delivered before the - one being noted, or the cursor the read started from. The synthesized - record carries it, so a consumer that checkpoints the supersession and - resumes after it gets the new attempt's first record next. - - A producer that declares no attempt supersedes nothing, because there - is no generation to compare. - """ - if not producer_id or attempt <= 0: - return None - seen = self._attempts.get(producer_id, 0) - if attempt <= seen: - return None - self._attempts[producer_id] = attempt - if seen == 0: - return None - return StreamRecord( - kind=RecordKind.SUPERSEDED, - cursor=previous, - topic=topic, - producer_id=producer_id, - attempt=attempt, - supersession=Supersession( - producer_id=producer_id, previous_attempt=seen, attempt=attempt - ), - ) diff --git a/temporalio/contrib/streams/_record.py b/temporalio/contrib/streams/_record.py index 268a1517e2..8a9c6f8010 100644 --- a/temporalio/contrib/streams/_record.py +++ b/temporalio/contrib/streams/_record.py @@ -1,6 +1,6 @@ """The value types the stream contract is expressed in. -Nothing here touches Temporal or a provider, so every provider shares it +Nothing here touches Temporal or a store, so every store shares it unchanged. """ @@ -42,18 +42,21 @@ class RecordKind(enum.IntEnum): FINISH = 2 """The producer in ``producer_id`` will write nothing more on this topic. - An empty ``producer_id`` names the owning Workflow. It does not end a - read and says nothing about the producer's outcome: an Activity can + 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. - The reader synthesizes it from the records it observed. No store holds - it, so every provider reports a retry the same way. Its cursor is the - position before the new attempt's first record, so a reader that resumes - after it gets that record next. + 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. """ @@ -61,9 +64,9 @@ class RecordKind(enum.IntEnum): class Cursor: """A position in one stream. - Opaque on purpose. The token names the provider that minted it and the - stream it belongs to, and that provider refuses a token from another - provider or another stream with + 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. @@ -72,7 +75,7 @@ class Cursor: token: str def __str__(self) -> str: - """The provider's position token.""" + """The store's position token.""" return self.token @@ -103,7 +106,20 @@ class StreamRecord(Generic[T]): """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 index e52850cdbb..f722752a25 100644 --- a/temporalio/contrib/streams/_wire.py +++ b/temporalio/contrib/streams/_wire.py @@ -1,24 +1,37 @@ -"""How a record crosses a provider: the proto is the record. +"""How a record crosses to Core and back: the proto is the record. -``temporal.sdk.streams.v1.StreamRecord`` is the wire format on every -provider. A store that keeps bytes keeps ``SerializeToString()`` of it, and a -reader in any language parses the same bytes. ``body`` is the user's payload, +``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 collections.abc import Callable from typing import Any import temporalio.converter -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 -from temporalio.contrib.streams.proto.v1 import StreamRecordKind - -__all__ = ["RecordDecoder", "WireRecord", "from_wire", "to_wire"] +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( @@ -31,7 +44,7 @@ def to_wire( attempt: int = 0, sequence: int = 0, ) -> WireRecord: - """Build the record a provider stores. + """The record as the converter produced it, before the codec. Only a ``DATA`` record carries a body; the converter encodes ``value`` into it. @@ -54,91 +67,49 @@ def to_wire( return record -def from_wire( - converter: temporalio.converter.PayloadConverter, - cursor: Cursor, - wire: WireRecord, +async def from_read( + converter: temporalio.converter.DataConverter, + read: ReadRecord, result_type: type | None, ) -> StreamRecord[Any]: - """Turn a stored record into the record a reader yields. + """The record a reader yields for one record of Core's answer. Raises: - ValueError: The kind is one no store may hold, such as a synthesized - ``SUPERSEDED`` or a value this SDK does not know. + 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 by readers, never stored") - if kind is RecordKind.UNSPECIFIED: - kind = RecordKind.DATA + raise ValueError("a SUPERSEDED record is synthesized, never stored") value: Any = None - if kind is RecordKind.DATA and wire.HasField("body"): + 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.from_payloads([wire.body], hints)[0] + value = converter.payload_converter.from_payloads([body], hints)[0] return StreamRecord( - kind=kind, + 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, ) - - -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. - """ - - def __init__( - self, - converter: temporalio.converter.PayloadConverter, - result_type: type | None, - *, - after: Cursor, - warn: Callable[[str], None], - ) -> None: - """Decode with ``converter`` into ``result_type``, resuming after ``after``.""" - self._converter = converter - self._result_type = result_type - self._previous = after - self._warn = warn - self._attempts = AttemptTracker() - - def prime(self, wire: WireRecord) -> None: - """Note the attempt of the record at the resume cursor, which was already delivered. - - A read that resumes would otherwise take the next attempt of that - producer as its first, and report no ``SUPERSEDED``. Only the cursor - record's producer is primed. Another producer's earlier attempts are - not known to a resumed read. - """ - self._attempts.note( - wire.producer_id, wire.attempt, topic=wire.topic, previous=self._previous - ) - - def decode(self, cursor: Cursor, wire: WireRecord) -> list[StreamRecord[Any]]: - """The records to yield for one stored record, in order.""" - 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 [] - out: list[StreamRecord[Any]] = [] - superseded = self._attempts.note( - wire.producer_id, wire.attempt, topic=wire.topic, previous=self._previous - ) - if superseded is not None: - out.append(superseded) - out.append(record) - self._previous = cursor - return out diff --git a/temporalio/contrib/streams/proto/__init__.py b/temporalio/contrib/streams/proto/__init__.py deleted file mode 100644 index 1ede5611cf..0000000000 --- a/temporalio/contrib/streams/proto/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -"""Generated protobuf models for the stream record envelope. - -.. warning:: - This package is experimental and may change in future versions. - -The modules under ``v1`` are generated by ``scripts/gen_protos.py`` and -rewritten on every run, so this note lives here rather than in them. -""" diff --git a/temporalio/contrib/streams/proto/v1/__init__.py b/temporalio/contrib/streams/proto/v1/__init__.py deleted file mode 100644 index fc6311caae..0000000000 --- a/temporalio/contrib/streams/proto/v1/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -from .message_pb2 import StreamRecord, StreamRecordKind - -__all__ = [ - "StreamRecord", - "StreamRecordKind", -] diff --git a/temporalio/contrib/streams/proto/v1/message_pb2.py b/temporalio/contrib/streams/proto/v1/message_pb2.py deleted file mode 100644 index 2df6179aa0..0000000000 --- a/temporalio/contrib/streams/proto/v1/message_pb2.py +++ /dev/null @@ -1,66 +0,0 @@ -# -*- coding: utf-8 -*- -# Generated by the protocol buffer compiler. DO NOT EDIT! -# source: temporal/sdk/streams/v1/message.proto -"""Generated protocol buffer code.""" - -from google.protobuf import descriptor as _descriptor -from google.protobuf import descriptor_pool as _descriptor_pool -from google.protobuf import message as _message -from google.protobuf import reflection as _reflection -from google.protobuf import symbol_database as _symbol_database -from google.protobuf.internal import enum_type_wrapper - -# @@protoc_insertion_point(imports) - -_sym_db = _symbol_database.Default() - - -from temporalio.api.common.v1 import ( - message_pb2 as temporal_dot_api_dot_common_dot_v1_dot_message__pb2, -) - -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( - b'\n%temporal/sdk/streams/v1/message.proto\x12\x17temporal.sdk.streams.v1\x1a$temporal/api/common/v1/message.proto"\xd6\x02\n\x0cStreamRecord\x12-\n\x04\x62ody\x18\x01 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x45\n\x08metadata\x18\x02 \x03(\x0b\x32\x33.temporal.sdk.streams.v1.StreamRecord.MetadataEntry\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x37\n\x04kind\x18\x04 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12\x13\n\x0bproducer_id\x18\x05 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x06 \x01(\x03\x12\x10\n\x08sequence\x18\x07 \x01(\x03\x1aP\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12.\n\x05value\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload:\x02\x38\x01*x\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12\x1d\n\x19STREAM_RECORD_KIND_FINISH\x10\x02"\x04\x08\x03\x10\x03\x62\x06proto3' -) - -_STREAMRECORDKIND = DESCRIPTOR.enum_types_by_name["StreamRecordKind"] -StreamRecordKind = enum_type_wrapper.EnumTypeWrapper(_STREAMRECORDKIND) -STREAM_RECORD_KIND_UNSPECIFIED = 0 -STREAM_RECORD_KIND_DATA = 1 -STREAM_RECORD_KIND_FINISH = 2 - - -_STREAMRECORD = DESCRIPTOR.message_types_by_name["StreamRecord"] -_STREAMRECORD_METADATAENTRY = _STREAMRECORD.nested_types_by_name["MetadataEntry"] -StreamRecord = _reflection.GeneratedProtocolMessageType( - "StreamRecord", - (_message.Message,), - { - "MetadataEntry": _reflection.GeneratedProtocolMessageType( - "MetadataEntry", - (_message.Message,), - { - "DESCRIPTOR": _STREAMRECORD_METADATAENTRY, - "__module__": "temporal.sdk.streams.v1.message_pb2", - # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.StreamRecord.MetadataEntry) - }, - ), - "DESCRIPTOR": _STREAMRECORD, - "__module__": "temporal.sdk.streams.v1.message_pb2", - # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.StreamRecord) - }, -) -_sym_db.RegisterMessage(StreamRecord) -_sym_db.RegisterMessage(StreamRecord.MetadataEntry) - -if _descriptor._USE_C_DESCRIPTORS == False: - DESCRIPTOR._options = None - _STREAMRECORD_METADATAENTRY._options = None - _STREAMRECORD_METADATAENTRY._serialized_options = b"8\001" - _STREAMRECORDKIND._serialized_start = 449 - _STREAMRECORDKIND._serialized_end = 569 - _STREAMRECORD._serialized_start = 105 - _STREAMRECORD._serialized_end = 447 - _STREAMRECORD_METADATAENTRY._serialized_start = 367 - _STREAMRECORD_METADATAENTRY._serialized_end = 447 -# @@protoc_insertion_point(module_scope) diff --git a/temporalio/contrib/streams/proto/v1/message_pb2.pyi b/temporalio/contrib/streams/proto/v1/message_pb2.pyi deleted file mode 100644 index d716502b86..0000000000 --- a/temporalio/contrib/streams/proto/v1/message_pb2.pyi +++ /dev/null @@ -1,161 +0,0 @@ -""" -@generated by mypy-protobuf. Do not edit manually! -isort:skip_file -""" - -import builtins -import collections.abc -import sys -import typing - -import google.protobuf.descriptor -import google.protobuf.internal.containers -import google.protobuf.internal.enum_type_wrapper -import google.protobuf.message - -import temporalio.api.common.v1.message_pb2 - -if sys.version_info >= (3, 10): - import typing as typing_extensions -else: - import typing_extensions - -DESCRIPTOR: google.protobuf.descriptor.FileDescriptor - -class _StreamRecordKind: - ValueType = typing.NewType("ValueType", builtins.int) - V: typing_extensions.TypeAlias = ValueType - -class _StreamRecordKindEnumTypeWrapper( - google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ - _StreamRecordKind.ValueType - ], - builtins.type, -): # noqa: F821 - DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor - STREAM_RECORD_KIND_UNSPECIFIED: _StreamRecordKind.ValueType # 0 - """Read as DATA.""" - STREAM_RECORD_KIND_DATA: _StreamRecordKind.ValueType # 1 - """A value the producer published; `body` carries it.""" - STREAM_RECORD_KIND_FINISH: _StreamRecordKind.ValueType # 2 - """The producer named by `producer_id` writes nothing more on `topic`. - Says nothing about that producer's outcome and does not end the stream. - """ - -class StreamRecordKind(_StreamRecordKind, metaclass=_StreamRecordKindEnumTypeWrapper): - """What a record means to a reader.""" - -STREAM_RECORD_KIND_UNSPECIFIED: StreamRecordKind.ValueType # 0 -"""Read as DATA.""" -STREAM_RECORD_KIND_DATA: StreamRecordKind.ValueType # 1 -"""A value the producer published; `body` carries it.""" -STREAM_RECORD_KIND_FINISH: StreamRecordKind.ValueType # 2 -"""The producer named by `producer_id` writes nothing more on `topic`. -Says nothing about that producer's outcome and does not end the stream. -""" -global___StreamRecordKind = StreamRecordKind - -class StreamRecord(google.protobuf.message.Message): - """One entry in a stream. The record is the wire format: stores keep it - serialized as is and readers in every language decode the same bytes. - - Field numbers and enum values are part of the stored format. The package - name is not, so this message can move to another package without a data - change. - """ - - DESCRIPTOR: google.protobuf.descriptor.Descriptor - - class MetadataEntry(google.protobuf.message.Message): - DESCRIPTOR: google.protobuf.descriptor.Descriptor - - KEY_FIELD_NUMBER: builtins.int - VALUE_FIELD_NUMBER: builtins.int - key: builtins.str - @property - def value(self) -> temporalio.api.common.v1.message_pb2.Payload: ... - def __init__( - self, - *, - key: builtins.str = ..., - value: temporalio.api.common.v1.message_pb2.Payload | None = ..., - ) -> None: ... - def HasField( - self, field_name: typing_extensions.Literal["value", b"value"] - ) -> builtins.bool: ... - def ClearField( - self, - field_name: typing_extensions.Literal["key", b"key", "value", b"value"], - ) -> None: ... - - BODY_FIELD_NUMBER: builtins.int - METADATA_FIELD_NUMBER: builtins.int - TOPIC_FIELD_NUMBER: builtins.int - KIND_FIELD_NUMBER: builtins.int - PRODUCER_ID_FIELD_NUMBER: builtins.int - ATTEMPT_FIELD_NUMBER: builtins.int - SEQUENCE_FIELD_NUMBER: builtins.int - @property - def body(self) -> temporalio.api.common.v1.message_pb2.Payload: - """The value the producer published. A provider runs it through the - payload codec and external storage before it stores the record. - """ - @property - def metadata( - self, - ) -> google.protobuf.internal.containers.MessageMap[ - builtins.str, temporalio.api.common.v1.message_pb2.Payload - ]: - """Provenance stamped by the SDK, such as the plaintext content hash.""" - topic: builtins.str - """The topic the record was published on.""" - kind: global___StreamRecordKind.ValueType - """How to read this record. Unspecified is read as DATA.""" - producer_id: builtins.str - """Who wrote the record. Empty when the owning Workflow did.""" - attempt: builtins.int - """The producer's attempt. Readers treat a later attempt by the same - producer as superseding what the earlier one wrote. - """ - sequence: builtins.int - """The producer's position within its attempt, starting at one, or zero - when it does not number its records. - """ - def __init__( - self, - *, - body: temporalio.api.common.v1.message_pb2.Payload | None = ..., - metadata: collections.abc.Mapping[ - builtins.str, temporalio.api.common.v1.message_pb2.Payload - ] - | None = ..., - topic: builtins.str = ..., - kind: global___StreamRecordKind.ValueType = ..., - producer_id: builtins.str = ..., - attempt: builtins.int = ..., - sequence: builtins.int = ..., - ) -> None: ... - def HasField( - self, field_name: typing_extensions.Literal["body", b"body"] - ) -> builtins.bool: ... - def ClearField( - self, - field_name: typing_extensions.Literal[ - "attempt", - b"attempt", - "body", - b"body", - "kind", - b"kind", - "metadata", - b"metadata", - "producer_id", - b"producer_id", - "sequence", - b"sequence", - "topic", - b"topic", - ], - ) -> None: ... - -global___StreamRecord = StreamRecord diff --git a/tests/contrib/streams/test_envelope_proto.py b/tests/contrib/streams/test_envelope_proto.py deleted file mode 100644 index cbb06920f8..0000000000 --- a/tests/contrib/streams/test_envelope_proto.py +++ /dev/null @@ -1,42 +0,0 @@ -"""The stream record envelope keeps its stored format.""" - -from __future__ import annotations - -from temporalio.api.common.v1 import Payload -from temporalio.contrib.streams.proto.v1 import StreamRecord, StreamRecordKind - - -def test_the_envelope_keeps_its_field_numbers(): - # The bytes are the stored format; a renumbered field would make every - # retained record unreadable. - fields = {f.name: f.number for f in StreamRecord.DESCRIPTOR.fields} - assert fields == { - "body": 1, - "metadata": 2, - "topic": 3, - "kind": 4, - "producer_id": 5, - "attempt": 6, - "sequence": 7, - } - assert StreamRecord.DESCRIPTOR.full_name == "temporal.sdk.streams.v1.StreamRecord" - kinds = StreamRecord.DESCRIPTOR.fields_by_name["kind"].enum_type - assert kinds is not None - assert {v.name: v.number for v in kinds.values} == { - "STREAM_RECORD_KIND_UNSPECIFIED": 0, - "STREAM_RECORD_KIND_DATA": 1, - "STREAM_RECORD_KIND_FINISH": 2, - } - - -def test_the_envelope_round_trips_through_bytes(): - record = StreamRecord( - body=Payload(metadata={"encoding": b"json/plain"}, data=b"1"), - metadata={"k": Payload(data=b"v")}, - topic="out", - kind=StreamRecordKind.STREAM_RECORD_KIND_FINISH, - producer_id="p", - attempt=2, - sequence=3, - ) - assert StreamRecord.FromString(record.SerializeToString()) == record diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py index 76a1e09099..0d23012a8d 100644 --- a/tests/contrib/streams/test_public_surface.py +++ b/tests/contrib/streams/test_public_surface.py @@ -4,18 +4,17 @@ import temporalio.contrib.streams as streams -# Provider-author helpers stay importable from their private modules, but -# they are not part of the public surface. +# The helpers behind the handles stay importable from their private modules, +# but they are not part of the public surface. _PRIVATE = { - "CONTENT_HASH_KEY", - "content_fingerprint", + "batch_digest", "content_hash", "decode_body", - "encode_body", + "encode_bodies", } -def test_provider_helpers_are_not_public(): +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 index a379ade57e..9a18b7d34c 100644 --- a/tests/contrib/streams/test_records_wire.py +++ b/tests/contrib/streams/test_records_wire.py @@ -1,4 +1,4 @@ -"""Records, their wire form, and what a provider owes a record's body.""" +"""Records, their stored form, and what lang owes a record's body.""" from __future__ import annotations @@ -8,31 +8,30 @@ 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, - Supersession, -) +from temporalio.contrib.streams import Cursor, RecordKind +from temporalio.contrib.streams import Supersession as ReportedSupersession from temporalio.contrib.streams._body import ( - CONTENT_HASH_KEY, - content_fingerprint, + batch_digest, content_hash, decode_body, - encode_body, + encode_bodies, ) -from temporalio.contrib.streams._policy import AttemptTracker -from temporalio.contrib.streams._wire import RecordDecoder, from_wire, to_wire -from temporalio.contrib.streams.proto.v1 import StreamRecord as WireRecord +from temporalio.contrib.streams._wire import RUN_ID_KEY, WireRecord, from_read, to_wire from temporalio.converter import DataConverter, PayloadCodec -BEGINNING = Cursor("") +CONVERTER = DataConverter.default + + +def stored(wire: WireRecord, cursor: str = "c", *, stale: bool = False) -> ReadRecord: + return ReadRecord(cursor=cursor, stored=wire, stale=stale) -def test_a_record_round_trips_through_the_wire(): - converter = DataConverter.default.payload_converter +async def test_a_record_reads_back_as_it_was_written(): wire = to_wire( - converter, + CONVERTER.payload_converter, topic="out", kind=RecordKind.DATA, value={"n": 1}, @@ -40,106 +39,108 @@ def test_a_record_round_trips_through_the_wire(): attempt=2, sequence=3, ) - parsed = WireRecord.FromString(wire.SerializeToString()) - record = from_wire(converter, Cursor("c"), parsed, dict) + 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, topic="out", kind=RecordKind.FINISH) + finish = to_wire(CONVERTER.payload_converter, topic="out", kind=RecordKind.FINISH) assert not finish.HasField("body") - assert from_wire(converter, Cursor("c"), finish, None).kind is RecordKind.FINISH + 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 + - unset = WireRecord(topic="out", body=converter.to_payloads([1])[0]) - assert from_wire(converter, Cursor("c"), 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(): - converter = DataConverter.default.payload_converter for kind in (RecordKind.SUPERSEDED, RecordKind.UNSPECIFIED): with pytest.raises(ValueError): - to_wire(converter, topic="out", kind=kind) + to_wire(CONVERTER.payload_converter, topic="out", kind=kind) -def test_a_raw_value_passes_through_untouched(): - converter = DataConverter.default.payload_converter +async def test_a_raw_value_passes_through_untouched(): raw = Payload(metadata={"encoding": b"binary/custom"}, data=b"\x00\x01") - wire = to_wire(converter, topic="out", kind=RecordKind.DATA, value=RawValue(raw)) + wire = to_wire( + CONVERTER.payload_converter, + topic="out", + kind=RecordKind.DATA, + value=RawValue(raw), + ) assert wire.body == raw - record = from_wire(converter, Cursor("c"), wire, RawValue) + record = await from_read(CONVERTER, stored(wire), RawValue) assert isinstance(record.value, RawValue) assert record.value.payload == raw -def test_supersession_is_synthesized_from_observations(): - attempts = AttemptTracker() - assert attempts.note("model", 1, topic="t", previous=BEGINNING) is None - assert attempts.note("model", 1, topic="t", previous=Cursor("c1")) is None - superseded = attempts.note("model", 2, topic="t", previous=Cursor("c2")) - assert superseded is not None - assert superseded.kind is RecordKind.SUPERSEDED - assert superseded.cursor == Cursor("c2") - assert superseded.supersession == Supersession("model", 1, 2) - assert attempts.note("model", 2, topic="t", previous=Cursor("c3")) is None - # Without an id or an attempt there is no generation to compare. - assert attempts.note("", 5, topic="t", previous=Cursor("c4")) is None - 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] = [] - converter = DataConverter.default.payload_converter - decoder = RecordDecoder(converter, int, after=BEGINNING, warn=said.append) - 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] - good = to_wire(converter, topic="t", kind=RecordKind.DATA, value=3) - assert [r.value for r in decoder.decode(Cursor("c2"), good)] == [3] - - -def test_a_decoder_primed_at_its_resume_cursor_reports_a_new_attempt(): - converter = DataConverter.default.payload_converter - decoder = RecordDecoder(converter, int, after=Cursor("c2"), warn=lambda _: None) - - def written(attempt: int, value: int) -> WireRecord: - wire = to_wire(converter, topic="t", kind=RecordKind.DATA, value=value) - wire.producer_id, wire.attempt = "model", attempt - return wire - - # The record at the cursor was already delivered, before the reader stopped. - decoder.prime(written(1, 2)) - out = decoder.decode(Cursor("c3"), written(2, 3)) - assert [r.kind for r in out] == [RecordKind.SUPERSEDED, RecordKind.DATA] - assert out[0].cursor == Cursor("c2") - - def test_the_content_hash_is_the_plaintext_payload_hash(): - converter = DataConverter.default.payload_converter + 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]) - assert len(content_hash(one)) == 64 - - -def test_the_fingerprint_is_length_delimited_over_the_batch(): - converter = DataConverter.default.payload_converter - - def records(*values: str) -> list[WireRecord]: - return [ - to_wire(converter, topic="t", kind=RecordKind.DATA, value=v) for v in values - ] - - assert content_fingerprint(records("a", "b")) == content_fingerprint( - records("a", "b") - ) - assert content_fingerprint(records("a", "b")) != content_fingerprint( - records("b", "a") + # 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" ) - assert content_fingerprint(records("ab")) != content_fingerprint(records("a", "b")) class NonceCodec(PayloadCodec): @@ -158,31 +159,22 @@ async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: return [Payload.FromString(p.data[16:]) for p in payloads] -async def test_the_body_is_hashed_before_the_codec_and_decoded_after(): +async def test_the_codec_runs_on_the_body_and_never_on_the_hashes(): converter = DataConverter(payload_codec=NonceCodec()) - plain = to_wire( - converter.payload_converter, topic="t", kind=RecordKind.DATA, value={"n": 1} + 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, ) - expected = content_hash(plain.body) - - first = WireRecord() - first.CopyFrom(plain) - second = WireRecord() - second.CopyFrom(plain) - await encode_body(converter, first) - await encode_body(converter, second) - # The codec made the stored bytes differ, and the hash did not move. - assert first.body != second.body - assert first.body.metadata["encoding"] == b"binary/nonce" - for stamped in (first, second): - assert stamped.metadata[CONTENT_HASH_KEY].data.decode() == expected - assert stamped.metadata[CONTENT_HASH_KEY].metadata["encoding"] == ( - b"binary/plain" - ) - - await decode_body(converter, first) - assert first.body == plain.body - - finish = to_wire(converter.payload_converter, topic="t", kind=RecordKind.FINISH) - await encode_body(converter, finish) - assert CONTENT_HASH_KEY not in finish.metadata + assert record.value == "a" From 9b11bf1a79dd6b3923e493add30308a37aeb9ade Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:28:27 -0700 Subject: [PATCH 8/8] Said only the writer compares its digest, and what reaches History. Deterministic serialization differs across protobuf implementations, so a retry through another SDK wouldn't dedupe. A publishing task does add one marker event, so the package shouldn't promise none. --- temporalio/contrib/streams/__init__.py | 3 ++- temporalio/contrib/streams/_body.py | 6 ++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index ccf428fc74..8bb571572e 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -6,7 +6,8 @@ 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, and never pass through Temporal or its History. +Core reaches for every SDK. Record bodies never reach the server. A Workflow +Task that publishes adds one marker event with a small manifest. The contract: diff --git a/temporalio/contrib/streams/_body.py b/temporalio/contrib/streams/_body.py index 4ed14ee6f1..7ecbfe361b 100644 --- a/temporalio/contrib/streams/_body.py +++ b/temporalio/contrib/streams/_body.py @@ -37,8 +37,10 @@ def batch_digest(records: Sequence[WireRecord]) -> bytes: 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. - Every SDK takes it the same way, and the store compares it when a - producer repeats its newest batch. + 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: