diff --git a/.gitattributes b/.gitattributes index a19ade077d..49c3ecfce9 100644 --- a/.gitattributes +++ b/.gitattributes @@ -1 +1,11 @@ CHANGELOG.md merge=union +# GitHub folds generated files in diffs. +temporalio/api/** linguist-generated=true +temporalio/bridge/proto/** linguist-generated=true +*_pb2*.py linguist-generated=true +*_pb2*.pyi linguist-generated=true +*_generated.py linguist-generated=true +*_generated.rs linguist-generated=true +**/_generated/** linguist-generated=true +**/testdata/** linguist-generated=true +*_vectors.json linguist-generated=true diff --git a/changelog/added/cornfed-tanesha.md b/changelog/added/cornfed-tanesha.md new file mode 100644 index 0000000000..426c69a5c0 --- /dev/null +++ b/changelog/added/cornfed-tanesha.md @@ -0,0 +1 @@ +Core: Experimental: the stream service's `RetireProducer` call drops one producer attempt's dedupe state on a stream, which the Redis store keeps as one field of the topic's meta hash until retention. A producer that writes once per short-lived attempt calls it after the attempt's last append returned, so the meta hash tracks the attempts in flight. Calling it earlier lets a retried batch land twice. diff --git a/changelog/added/hypermagical-celena.md b/changelog/added/hypermagical-celena.md new file mode 100644 index 0000000000..9f2af80511 --- /dev/null +++ b/changelog/added/hypermagical-celena.md @@ -0,0 +1,2 @@ +Core: Experimental: behind the new `streams` feature, `streams::connect_stream_service` builds one stream store from a `StreamStoreConfig` (Redis or memory) and serves lang's stream calls in process: append, read, latest, close and owner delete, as serialized `coresdk.streams` requests. +Core: Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the stream service, built with the Redis store. diff --git a/changelog/breaking-changes/hypermagical-celena.md b/changelog/breaking-changes/hypermagical-celena.md new file mode 100644 index 0000000000..c13dc8f714 --- /dev/null +++ b/changelog/breaking-changes/hypermagical-celena.md @@ -0,0 +1 @@ +Core: Replay detects changes to the service or operation of scheduled Nexus operations, reporting nondeterminism while retaining compatibility with histories recorded before these checks. diff --git a/changelog/fixed/hypermagical-celena.md b/changelog/fixed/hypermagical-celena.md new file mode 100644 index 0000000000..73e2df8151 --- /dev/null +++ b/changelog/fixed/hypermagical-celena.md @@ -0,0 +1,2 @@ +Core: Local activities still running when their workflow is evicted or the worker shuts down now always receive a cancellation. Local activities no longer retry ordinary failures or start-to-close timeouts after cancellation has been requested, matching non-local activities. Workflow code no longer observes a spurious local activity cancellation when its run is evicted or completes while the local activity is backing off between retries. +Core: Workers no longer panic when evicting a workflow with buffered local activity results. Completed activations' output is recorded before eviction, without recording eviction-induced cancellations. diff --git a/pyproject.toml b/pyproject.toml index 444e4d0c5b..87004a15fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -183,6 +183,8 @@ exclude = [ 'temporalio/api', 'temporalio/bridge/proto', 'temporalio/nexus/system/workflow_service', + # Core's own sources, which hold a Python script for Core's cross-SDK tests + 'temporalio/bridge/sdk-core', ] [[tool.mypy.overrides]] @@ -258,11 +260,15 @@ exclude = [ "temporalio/api", "temporalio/bridge/proto", "temporalio/bridge/_visitor.py", + # Core's own sources, which hold a Python script for Core's cross-SDK tests + "temporalio/bridge/sdk-core", "tests/worker/workflow_sandbox/testmodules/proto", ] [tool.ruff] target-version = "py310" +# Core's own sources, which hold a Python script for Core's cross-SDK tests +extend-exclude = ["temporalio/bridge/sdk-core"] [build-system] requires = ["maturin>=1.0,<2.0"] diff --git a/scripts/gen_bridge_client.py b/scripts/gen_bridge_client.py index f06dd29e65..ffd3ac7a5f 100644 --- a/scripts/gen_bridge_client.py +++ b/scripts/gen_bridge_client.py @@ -13,6 +13,7 @@ import temporalio.api.testservice.v1.service_pb2 as test_service import temporalio.api.workflowservice.v1.service_pb2 as workflow_service import temporalio.bridge.proto.health.v1.health_pb2 as health_service +import temporalio.bridge.proto.streams.streams_pb2 as stream_service def generate_python_services( @@ -231,6 +232,85 @@ def generate_rust_match_arm( ) +def python_module_for(file_name: str) -> str: + """The Python package that re-exports the messages of a proto file.""" + directory = file_name.rsplit("/", 1)[0] + for proto_prefix, python_prefix in ( + ("temporal/sdk/core/", "temporalio/bridge/proto/"), + ("temporal/api/", "temporalio/api/"), + ): + if directory.startswith(proto_prefix): + return (python_prefix + directory[len(proto_prefix) :]).replace("/", ".") + raise ValueError(f"no Python package for {file_name}") + + +def generate_python_stream_service( + service_descriptor: ServiceDescriptor, + output_file: str = "temporalio/bridge/streams_generated.py", +): + print("generating python stream service") + + def type_name(message) -> str: + return f"{python_module_for(message.file.name)}.{message.name}" + + methods = sorted( + ( + method + for method in service_descriptor.methods + if not method.client_streaming and not method.server_streaming + ), + key=lambda m: m.name, + ) + modules = sorted( + {python_module_for(service_descriptor.file.name)} + | {python_module_for(m.input_type.file.name) for m in methods} + | {python_module_for(m.output_type.file.name) for m in methods} + ) + method_template = Template(''' + async def $method_name(self, req: $request_type) -> $response_type: + """Invokes the $service_name.$method_name call.""" + return await self._store.call("$rpc_name", req, $response_type) +''') + method_calls = [ + method_template.substitute( + service_name=service_descriptor.name, + method_name=pascal_to_snake(method.name), + rpc_name=method.name, + request_type=type_name(method.input_type), + response_type=type_name(method.output_type), + ) + for method in methods + ] + file_template = Template('''# Generated file. DO NOT EDIT +"""Generated calls for Core's in-process $service_name.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +$imports + +if TYPE_CHECKING: + from temporalio.bridge.streams import StreamStore + + +class $service_name: + """Calls for the $service_name, which Core dispatches in process.""" + + def __init__(self, store: StreamStore) -> None: + """Initialize service with the provided store.""" + self._store = store +$method_calls''') + with open(output_file, "w") as f: + f.write( + file_template.substitute( + service_name=service_descriptor.name, + imports="\n".join(f"import {module}" for module in modules), + method_calls="".join(method_calls), + ) + ) + + def pascal_to_snake(input: str) -> str: return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", input).lower() @@ -273,3 +353,7 @@ def sanitize_proto_name(input: str) -> str: health_service.DESCRIPTOR, ] ) + + generate_python_stream_service( + stream_service.DESCRIPTOR.services_by_name["StreamService"] + ) diff --git a/scripts/gen_protos.py b/scripts/gen_protos.py index 4b2ea04569..b75e7cdb5b 100644 --- a/scripts/gen_protos.py +++ b/scripts/gen_protos.py @@ -50,6 +50,10 @@ partial( re.compile(r"from temporal\.sdk\.core\.").sub, r"from temporalio.bridge.proto." ), + partial( + re.compile(r"from temporal\.sdk\.streams\.").sub, + r"from temporalio.bridge.proto.streams.", + ), partial( re.compile(r"'__module__' : 'temporal\.api\.").sub, r"'__module__' : 'temporalio.api.", @@ -71,6 +75,9 @@ r"temporalio.api.dependencies.nexusannotations.", ), partial(re.compile(r"temporal\.sdk\.core\.").sub, r"temporalio.bridge.proto."), + partial( + re.compile(r"temporal\.sdk\.streams\.").sub, r"temporalio.bridge.proto.streams." + ), ] find_class_re = re.compile(r"\nclass ([^_\(\:]+)") @@ -218,6 +225,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) + # The stream record envelope sits beside Core's own stream messages, which the loop above + # has just moved in. + (output_dir / "temporal" / "sdk" / "streams" / "v1").replace( + sdk_out_dir / "streams" / "v1" + ) shutil.rmtree(sdk_out_dir / "health", ignore_errors=True) (output_dir / "health").replace(sdk_out_dir / "health") # Move test protos diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index e61e990850..ce0d48ee3e 100644 --- a/temporalio/bridge/Cargo.lock +++ b/temporalio/bridge/Cargo.lock @@ -29,6 +29,32 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "arc-swap" +version = "1.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b" +dependencies = [ + "rustversion", +] + +[[package]] +name = "arcstr" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d" + +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + [[package]] name = "async-trait" version = "0.1.92" @@ -182,6 +208,15 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bon" version = "3.10.2" @@ -266,7 +301,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.3.1", "rand_core 0.10.1", ] @@ -307,7 +342,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e" dependencies = [ "bytes", + "futures-core", "memchr", + "pin-project-lite", + "tokio", + "tokio-util", ] [[package]] @@ -335,6 +374,15 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + [[package]] name = "cpufeatures" version = "0.3.1" @@ -344,6 +392,12 @@ dependencies = [ "libc", ] +[[package]] +name = "crc16" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "338089f42c427b86394a5ee60ff321da23a5c89c9d89514c829687b26359fcff" + [[package]] name = "crc32fast" version = "1.5.2" @@ -368,6 +422,16 @@ version = "0.8.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "darling" version = "0.24.1" @@ -425,6 +489,16 @@ dependencies = [ "unicode-xid", ] +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "dirs" version = "6.0.0" @@ -540,6 +614,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", +] + [[package]] name = "fastrand" version = "2.5.0" @@ -720,6 +814,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "gethostname" version = "1.1.0" @@ -1481,6 +1585,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -2021,6 +2131,39 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "redis" +version = "1.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5b98ae99461e02895c3a96a25dbe4d852a3d5e5b815d31d5d9e4b5344d99e9" +dependencies = [ + "arc-swap", + "arcstr", + "async-lock", + "backon", + "bytes", + "cfg-if", + "combine", + "crc16", + "futures-channel", + "futures-util", + "itoa", + "log", + "percent-encoding", + "pin-project-lite", + "rand 0.10.3", + "rustls", + "rustls-native-certs", + "ryu", + "sha1_smol", + "socket2", + "tokio", + "tokio-rustls", + "tokio-util", + "url", + "xxhash-rust", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -2370,6 +2513,23 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1_smol" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" + +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -2579,6 +2739,7 @@ dependencies = [ "temporalio-client", "temporalio-common", "temporalio-sdk-core", + "temporalio-streams", "tokio", "tokio-rustls", "tokio-stream", @@ -2751,6 +2912,7 @@ dependencies = [ "temporalio-client", "temporalio-common", "temporalio-macros", + "temporalio-streams", "thiserror", "tokio", "tokio-stream", @@ -2762,6 +2924,21 @@ dependencies = [ "zip", ] +[[package]] +name = "temporalio-streams" +version = "0.1.0" +dependencies = [ + "async-trait", + "crc16", + "prost", + "redis", + "sha2", + "temporalio-common", + "thiserror", + "tokio", + "tracing", +] + [[package]] name = "termtree" version = "0.5.1" @@ -3126,6 +3303,12 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c" +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "typetag" version = "0.2.23" @@ -3608,6 +3791,12 @@ dependencies = [ "rustix", ] +[[package]] +name = "xxhash-rust" +version = "0.8.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "550a2b930b62486a393c52d5c3b84bff264b28aa437ed64694d31e93b1757af7" + [[package]] name = "yoke" version = "0.8.3" diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml index 4e45b00ba3..f00c2e54c7 100644 --- a/temporalio/bridge/Cargo.toml +++ b/temporalio/bridge/Cargo.toml @@ -71,7 +71,11 @@ temporalio-sdk-core = { version = "0.10.0", path = "./sdk-core/crates/sdk-core", "ephemeral-server", "envconfig", "prometheus", + "streams", ] } +# The stream stores Core serves to Python. Redis takes the rustls provider the +# process installs, so it follows the tls-ring and tls-aws-lc features above. +temporalio-streams = { version = "0.1.0", path = "./sdk-core/crates/streams", features = ["redis"] } tokio = "1.26" # Matches the sdk-core client crate's rustls stack; used to build the custom # server certificate verifier for ClientTlsConfig.verification_server_name. diff --git a/temporalio/bridge/proto/streams/__init__.py b/temporalio/bridge/proto/streams/__init__.py new file mode 100644 index 0000000000..0873b3cad3 --- /dev/null +++ b/temporalio/bridge/proto/streams/__init__.py @@ -0,0 +1,69 @@ +from .streams_pb2 import ( + ActivityProducer, + AppendRecord, + AppendRequest, + AppendResponse, + ChainId, + CloseRequest, + CloseResponse, + DeleteOwnerRequest, + DeleteOwnerResponse, + LatestRequest, + LatestResponse, + MemoryStoreConfig, + NamedProducer, + ReadRecord, + ReadRequest, + ReadResponse, + RedisStoreConfig, + RetireProducerRequest, + RetireProducerResponse, + StoreAppendRequest, + StoreAppendResponse, + StoredRecord, + StoreLatestRequest, + StoreLatestResponse, + StoreReadRequest, + StoreReadResponse, + StreamAddress, + StreamFailure, + StreamFailureKind, + StreamOwnerKind, + StreamStoreConfig, + Supersession, +) + +__all__ = [ + "ActivityProducer", + "AppendRecord", + "AppendRequest", + "AppendResponse", + "ChainId", + "CloseRequest", + "CloseResponse", + "DeleteOwnerRequest", + "DeleteOwnerResponse", + "LatestRequest", + "LatestResponse", + "MemoryStoreConfig", + "NamedProducer", + "ReadRecord", + "ReadRequest", + "ReadResponse", + "RedisStoreConfig", + "RetireProducerRequest", + "RetireProducerResponse", + "StoreAppendRequest", + "StoreAppendResponse", + "StoreLatestRequest", + "StoreLatestResponse", + "StoreReadRequest", + "StoreReadResponse", + "StoredRecord", + "StreamAddress", + "StreamFailure", + "StreamFailureKind", + "StreamOwnerKind", + "StreamStoreConfig", + "Supersession", +] diff --git a/temporalio/bridge/proto/streams/streams_pb2.py b/temporalio/bridge/proto/streams/streams_pb2.py new file mode 100644 index 0000000000..5c9a74aa8e --- /dev/null +++ b/temporalio/bridge/proto/streams/streams_pb2.py @@ -0,0 +1,484 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/sdk/core/streams/streams.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 google.protobuf import duration_pb2 as google_dot_protobuf_dot_duration__pb2 + +from temporalio.api.common.v1 import ( + message_pb2 as temporal_dot_api_dot_common_dot_v1_dot_message__pb2, +) +from temporalio.bridge.proto.streams.v1 import ( + message_pb2 as temporal_dot_sdk_dot_streams_dot_v1_dot_message__pb2, +) + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n\'temporal/sdk/core/streams/streams.proto\x12\x0f\x63oresdk.streams\x1a\x1egoogle/protobuf/duration.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/sdk/streams/v1/message.proto"\x86\x01\n\x11StreamStoreConfig\x12\x32\n\x05redis\x18\x01 \x01(\x0b\x32!.coresdk.streams.RedisStoreConfigH\x00\x12\x34\n\x06memory\x18\x02 \x01(\x0b\x32".coresdk.streams.MemoryStoreConfigH\x00\x42\x07\n\x05store"\xc9\x01\n\x10RedisStoreConfig\x12\x0c\n\x04urls\x18\x01 \x03(\t\x12\x0f\n\x07\x63luster\x18\x02 \x01(\x08\x12\x12\n\nkey_prefix\x18\x03 \x01(\t\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1f\n\x17\x62locking_reads_per_node\x18\x05 \x01(\r\x12\x33\n\x10response_timeout\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration"\x13\n\x11MemoryStoreConfig"\x8c\x01\n\rStreamAddress\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x34\n\nowner_kind\x18\x02 \x01(\x0e\x32 .coresdk.streams.StreamOwnerKind\x12\x13\n\x0bworkflow_id\x18\x03 \x01(\t\x12\x0e\n\x06run_id\x18\x04 \x01(\t\x12\r\n\x05topic\x18\x05 \x01(\t"G\n\x07\x43hainId\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\x12\x14\n\x0c\x66irst_run_id\x18\x03 \x01(\t"5\n\rNamedProducer\x12\x13\n\x0bproducer_id\x18\x01 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x03"]\n\x10\x41\x63tivityProducer\x12\x13\n\x0bworkflow_id\x18\x01 \x01(\t\x12\x0e\n\x06run_id\x18\x02 \x01(\t\x12\x13\n\x0b\x61\x63tivity_id\x18\x03 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x05"\x8c\x01\n\x0c\x41ppendRecord\x12\x37\n\x04kind\x18\x01 \x01(\x0e\x32).temporal.sdk.streams.v1.StreamRecordKind\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x14\n\x0c\x63ontent_hash\x18\x03 \x01(\x0c"\x85\x02\n\rAppendRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12/\n\x05named\x18\x02 \x01(\x0b\x32\x1e.coresdk.streams.NamedProducerH\x00\x12\x35\n\x08\x61\x63tivity\x18\x03 \x01(\x0b\x32!.coresdk.streams.ActivityProducerH\x00\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12.\n\x07records\x18\x05 \x03(\x0b\x32\x1d.coresdk.streams.AppendRecord\x12\x0e\n\x06\x64igest\x18\x06 \x01(\x0c\x42\n\n\x08producer";\n\x0e\x41ppendResponse\x12\x14\n\x0c\x66irst_cursor\x18\x01 \x01(\t\x12\x13\n\x0blast_cursor\x18\x02 \x01(\t"\x99\x01\n\x0bReadRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12\r\n\x05\x61\x66ter\x18\x02 \x01(\t\x12\'\n\x04wait\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x13\n\x0bmax_records\x18\x04 \x01(\r\x12\r\n\x05state\x18\x05 \x01(\x0c"]\n\x0cSupersession\x12\r\n\x05topic\x18\x01 \x01(\t\x12\x13\n\x0bproducer_id\x18\x02 \x01(\t\x12\x18\n\x10previous_attempt\x18\x03 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x03"\xa3\x01\n\nReadRecord\x12\x0e\n\x06\x63ursor\x18\x01 \x01(\t\x12\x37\n\x06stored\x18\x02 \x01(\x0b\x32%.temporal.sdk.streams.v1.StreamRecordH\x00\x12\x33\n\nsuperseded\x18\x03 \x01(\x0b\x32\x1d.coresdk.streams.SupersessionH\x00\x12\r\n\x05stale\x18\x04 \x01(\x08\x42\x08\n\x06record"i\n\x0cReadResponse\x12,\n\x07records\x18\x01 \x03(\x0b\x32\x1b.coresdk.streams.ReadRecord\x12\x0e\n\x06\x63ursor\x18\x02 \x01(\t\x12\r\n\x05state\x18\x03 \x01(\x0c\x12\x0c\n\x04\x64one\x18\x04 \x01(\x08"?\n\rLatestRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress" \n\x0eLatestResponse\x12\x0e\n\x06\x63ursor\x18\x01 \x01(\t">\n\x0c\x43loseRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress"\x0f\n\rCloseResponse"r\n\x12\x44\x65leteOwnerRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x34\n\nowner_kind\x18\x02 \x01(\x0e\x32 .coresdk.streams.StreamOwnerKind\x12\x13\n\x0bworkflow_id\x18\x03 \x01(\t"&\n\x13\x44\x65leteOwnerResponse\x12\x0f\n\x07\x64\x65leted\x18\x01 \x01(\x04"\xbb\x01\n\x15RetireProducerRequest\x12.\n\x06stream\x18\x01 \x01(\x0b\x32\x1e.coresdk.streams.StreamAddress\x12/\n\x05named\x18\x02 \x01(\x0b\x32\x1e.coresdk.streams.NamedProducerH\x00\x12\x35\n\x08\x61\x63tivity\x18\x03 \x01(\x0b\x32!.coresdk.streams.ActivityProducerH\x00\x42\n\n\x08producer"\x18\n\x16RetireProducerResponse"b\n\rStreamFailure\x12\x30\n\x04kind\x18\x01 \x01(\x0e\x32".coresdk.streams.StreamFailureKind\x12\x0f\n\x07message\x18\x02 \x01(\t\x12\x0e\n\x06\x63ursor\x18\x03 \x01(\t"\xa5\x01\n\x12StoreAppendRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x13\n\x0bproducer_id\x18\x03 \x01(\t\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x03\x12\x10\n\x08sequence\x18\x05 \x01(\x03\x12\x0e\n\x06\x64igest\x18\x06 \x01(\x0c\x12\x0f\n\x07records\x18\x07 \x03(\x0c"D\n\x13StoreAppendResponse\x12\x16\n\x0e\x66irst_position\x18\x01 \x01(\t\x12\x15\n\rlast_position\x18\x02 \x01(\t"\xa0\x01\n\x10StoreReadRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t\x12\x16\n\x0e\x61\x66ter_position\x18\x03 \x01(\t\x12\'\n\x04wait\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x13\n\x0bmax_records\x18\x05 \x01(\r"0\n\x0cStoredRecord\x12\x10\n\x08position\x18\x01 \x01(\t\x12\x0e\n\x06record\x18\x02 \x01(\x0c"S\n\x11StoreReadResponse\x12.\n\x07records\x18\x01 \x03(\x0b\x32\x1d.coresdk.streams.StoredRecord\x12\x0e\n\x06\x63losed\x18\x02 \x01(\x08"L\n\x12StoreLatestRequest\x12\'\n\x05\x63hain\x18\x01 \x01(\x0b\x32\x18.coresdk.streams.ChainId\x12\r\n\x05topic\x18\x02 \x01(\t"\'\n\x13StoreLatestResponse\x12\x10\n\x08position\x18\x01 \x01(\t*T\n\x0fStreamOwnerKind\x12!\n\x1dSTREAM_OWNER_KIND_UNSPECIFIED\x10\x00\x12\x1e\n\x1aSTREAM_OWNER_KIND_WORKFLOW\x10\x01*\xea\x03\n\x11StreamFailureKind\x12#\n\x1fSTREAM_FAILURE_KIND_UNSPECIFIED\x10\x00\x12*\n&STREAM_FAILURE_KIND_PRODUCER_DIVERGENT\x10\x01\x12&\n"STREAM_FAILURE_KIND_PRODUCER_STALE\x10\x02\x12\'\n#STREAM_FAILURE_KIND_OUTCOME_UNKNOWN\x10\x03\x12\x1f\n\x1bSTREAM_FAILURE_KIND_REFUSED\x10\x04\x12\x1e\n\x1aSTREAM_FAILURE_KIND_CLOSED\x10\x05\x12\x1e\n\x1aSTREAM_FAILURE_KIND_CURSOR\x10\x06\x12\x1f\n\x1bSTREAM_FAILURE_KIND_EXPIRED\x10\x07\x12!\n\x1dSTREAM_FAILURE_KIND_NOT_FOUND\x10\x08\x12\x1e\n\x1aSTREAM_FAILURE_KIND_RECORD\x10\t\x12\x1f\n\x1bSTREAM_FAILURE_KIND_STORAGE\x10\n\x12#\n\x1fSTREAM_FAILURE_KIND_UNSUPPORTED\x10\x0b\x12(\n$STREAM_FAILURE_KIND_INVALID_ARGUMENT\x10\x0c\x32\xef\x03\n\rStreamService\x12I\n\x06\x41ppend\x12\x1e.coresdk.streams.AppendRequest\x1a\x1f.coresdk.streams.AppendResponse\x12\x43\n\x04Read\x12\x1c.coresdk.streams.ReadRequest\x1a\x1d.coresdk.streams.ReadResponse\x12I\n\x06Latest\x12\x1e.coresdk.streams.LatestRequest\x1a\x1f.coresdk.streams.LatestResponse\x12\x46\n\x05\x43lose\x12\x1d.coresdk.streams.CloseRequest\x1a\x1e.coresdk.streams.CloseResponse\x12X\n\x0b\x44\x65leteOwner\x12#.coresdk.streams.DeleteOwnerRequest\x1a$.coresdk.streams.DeleteOwnerResponse\x12\x61\n\x0eRetireProducer\x12&.coresdk.streams.RetireProducerRequest\x1a\'.coresdk.streams.RetireProducerResponseB-\xea\x02*Temporalio::Internal::Bridge::Api::Streamsb\x06proto3' +) + +_STREAMOWNERKIND = DESCRIPTOR.enum_types_by_name["StreamOwnerKind"] +StreamOwnerKind = enum_type_wrapper.EnumTypeWrapper(_STREAMOWNERKIND) +_STREAMFAILUREKIND = DESCRIPTOR.enum_types_by_name["StreamFailureKind"] +StreamFailureKind = enum_type_wrapper.EnumTypeWrapper(_STREAMFAILUREKIND) +STREAM_OWNER_KIND_UNSPECIFIED = 0 +STREAM_OWNER_KIND_WORKFLOW = 1 +STREAM_FAILURE_KIND_UNSPECIFIED = 0 +STREAM_FAILURE_KIND_PRODUCER_DIVERGENT = 1 +STREAM_FAILURE_KIND_PRODUCER_STALE = 2 +STREAM_FAILURE_KIND_OUTCOME_UNKNOWN = 3 +STREAM_FAILURE_KIND_REFUSED = 4 +STREAM_FAILURE_KIND_CLOSED = 5 +STREAM_FAILURE_KIND_CURSOR = 6 +STREAM_FAILURE_KIND_EXPIRED = 7 +STREAM_FAILURE_KIND_NOT_FOUND = 8 +STREAM_FAILURE_KIND_RECORD = 9 +STREAM_FAILURE_KIND_STORAGE = 10 +STREAM_FAILURE_KIND_UNSUPPORTED = 11 +STREAM_FAILURE_KIND_INVALID_ARGUMENT = 12 + + +_STREAMSTORECONFIG = DESCRIPTOR.message_types_by_name["StreamStoreConfig"] +_REDISSTORECONFIG = DESCRIPTOR.message_types_by_name["RedisStoreConfig"] +_MEMORYSTORECONFIG = DESCRIPTOR.message_types_by_name["MemoryStoreConfig"] +_STREAMADDRESS = DESCRIPTOR.message_types_by_name["StreamAddress"] +_CHAINID = DESCRIPTOR.message_types_by_name["ChainId"] +_NAMEDPRODUCER = DESCRIPTOR.message_types_by_name["NamedProducer"] +_ACTIVITYPRODUCER = DESCRIPTOR.message_types_by_name["ActivityProducer"] +_APPENDRECORD = DESCRIPTOR.message_types_by_name["AppendRecord"] +_APPENDREQUEST = DESCRIPTOR.message_types_by_name["AppendRequest"] +_APPENDRESPONSE = DESCRIPTOR.message_types_by_name["AppendResponse"] +_READREQUEST = DESCRIPTOR.message_types_by_name["ReadRequest"] +_SUPERSESSION = DESCRIPTOR.message_types_by_name["Supersession"] +_READRECORD = DESCRIPTOR.message_types_by_name["ReadRecord"] +_READRESPONSE = DESCRIPTOR.message_types_by_name["ReadResponse"] +_LATESTREQUEST = DESCRIPTOR.message_types_by_name["LatestRequest"] +_LATESTRESPONSE = DESCRIPTOR.message_types_by_name["LatestResponse"] +_CLOSEREQUEST = DESCRIPTOR.message_types_by_name["CloseRequest"] +_CLOSERESPONSE = DESCRIPTOR.message_types_by_name["CloseResponse"] +_DELETEOWNERREQUEST = DESCRIPTOR.message_types_by_name["DeleteOwnerRequest"] +_DELETEOWNERRESPONSE = DESCRIPTOR.message_types_by_name["DeleteOwnerResponse"] +_RETIREPRODUCERREQUEST = DESCRIPTOR.message_types_by_name["RetireProducerRequest"] +_RETIREPRODUCERRESPONSE = DESCRIPTOR.message_types_by_name["RetireProducerResponse"] +_STREAMFAILURE = DESCRIPTOR.message_types_by_name["StreamFailure"] +_STOREAPPENDREQUEST = DESCRIPTOR.message_types_by_name["StoreAppendRequest"] +_STOREAPPENDRESPONSE = DESCRIPTOR.message_types_by_name["StoreAppendResponse"] +_STOREREADREQUEST = DESCRIPTOR.message_types_by_name["StoreReadRequest"] +_STOREDRECORD = DESCRIPTOR.message_types_by_name["StoredRecord"] +_STOREREADRESPONSE = DESCRIPTOR.message_types_by_name["StoreReadResponse"] +_STORELATESTREQUEST = DESCRIPTOR.message_types_by_name["StoreLatestRequest"] +_STORELATESTRESPONSE = DESCRIPTOR.message_types_by_name["StoreLatestResponse"] +StreamStoreConfig = _reflection.GeneratedProtocolMessageType( + "StreamStoreConfig", + (_message.Message,), + { + "DESCRIPTOR": _STREAMSTORECONFIG, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StreamStoreConfig) + }, +) +_sym_db.RegisterMessage(StreamStoreConfig) + +RedisStoreConfig = _reflection.GeneratedProtocolMessageType( + "RedisStoreConfig", + (_message.Message,), + { + "DESCRIPTOR": _REDISSTORECONFIG, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.RedisStoreConfig) + }, +) +_sym_db.RegisterMessage(RedisStoreConfig) + +MemoryStoreConfig = _reflection.GeneratedProtocolMessageType( + "MemoryStoreConfig", + (_message.Message,), + { + "DESCRIPTOR": _MEMORYSTORECONFIG, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.MemoryStoreConfig) + }, +) +_sym_db.RegisterMessage(MemoryStoreConfig) + +StreamAddress = _reflection.GeneratedProtocolMessageType( + "StreamAddress", + (_message.Message,), + { + "DESCRIPTOR": _STREAMADDRESS, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StreamAddress) + }, +) +_sym_db.RegisterMessage(StreamAddress) + +ChainId = _reflection.GeneratedProtocolMessageType( + "ChainId", + (_message.Message,), + { + "DESCRIPTOR": _CHAINID, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.ChainId) + }, +) +_sym_db.RegisterMessage(ChainId) + +NamedProducer = _reflection.GeneratedProtocolMessageType( + "NamedProducer", + (_message.Message,), + { + "DESCRIPTOR": _NAMEDPRODUCER, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.NamedProducer) + }, +) +_sym_db.RegisterMessage(NamedProducer) + +ActivityProducer = _reflection.GeneratedProtocolMessageType( + "ActivityProducer", + (_message.Message,), + { + "DESCRIPTOR": _ACTIVITYPRODUCER, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.ActivityProducer) + }, +) +_sym_db.RegisterMessage(ActivityProducer) + +AppendRecord = _reflection.GeneratedProtocolMessageType( + "AppendRecord", + (_message.Message,), + { + "DESCRIPTOR": _APPENDRECORD, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.AppendRecord) + }, +) +_sym_db.RegisterMessage(AppendRecord) + +AppendRequest = _reflection.GeneratedProtocolMessageType( + "AppendRequest", + (_message.Message,), + { + "DESCRIPTOR": _APPENDREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.AppendRequest) + }, +) +_sym_db.RegisterMessage(AppendRequest) + +AppendResponse = _reflection.GeneratedProtocolMessageType( + "AppendResponse", + (_message.Message,), + { + "DESCRIPTOR": _APPENDRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.AppendResponse) + }, +) +_sym_db.RegisterMessage(AppendResponse) + +ReadRequest = _reflection.GeneratedProtocolMessageType( + "ReadRequest", + (_message.Message,), + { + "DESCRIPTOR": _READREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.ReadRequest) + }, +) +_sym_db.RegisterMessage(ReadRequest) + +Supersession = _reflection.GeneratedProtocolMessageType( + "Supersession", + (_message.Message,), + { + "DESCRIPTOR": _SUPERSESSION, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.Supersession) + }, +) +_sym_db.RegisterMessage(Supersession) + +ReadRecord = _reflection.GeneratedProtocolMessageType( + "ReadRecord", + (_message.Message,), + { + "DESCRIPTOR": _READRECORD, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.ReadRecord) + }, +) +_sym_db.RegisterMessage(ReadRecord) + +ReadResponse = _reflection.GeneratedProtocolMessageType( + "ReadResponse", + (_message.Message,), + { + "DESCRIPTOR": _READRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.ReadResponse) + }, +) +_sym_db.RegisterMessage(ReadResponse) + +LatestRequest = _reflection.GeneratedProtocolMessageType( + "LatestRequest", + (_message.Message,), + { + "DESCRIPTOR": _LATESTREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.LatestRequest) + }, +) +_sym_db.RegisterMessage(LatestRequest) + +LatestResponse = _reflection.GeneratedProtocolMessageType( + "LatestResponse", + (_message.Message,), + { + "DESCRIPTOR": _LATESTRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.LatestResponse) + }, +) +_sym_db.RegisterMessage(LatestResponse) + +CloseRequest = _reflection.GeneratedProtocolMessageType( + "CloseRequest", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.CloseRequest) + }, +) +_sym_db.RegisterMessage(CloseRequest) + +CloseResponse = _reflection.GeneratedProtocolMessageType( + "CloseResponse", + (_message.Message,), + { + "DESCRIPTOR": _CLOSERESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.CloseResponse) + }, +) +_sym_db.RegisterMessage(CloseResponse) + +DeleteOwnerRequest = _reflection.GeneratedProtocolMessageType( + "DeleteOwnerRequest", + (_message.Message,), + { + "DESCRIPTOR": _DELETEOWNERREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.DeleteOwnerRequest) + }, +) +_sym_db.RegisterMessage(DeleteOwnerRequest) + +DeleteOwnerResponse = _reflection.GeneratedProtocolMessageType( + "DeleteOwnerResponse", + (_message.Message,), + { + "DESCRIPTOR": _DELETEOWNERRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.DeleteOwnerResponse) + }, +) +_sym_db.RegisterMessage(DeleteOwnerResponse) + +RetireProducerRequest = _reflection.GeneratedProtocolMessageType( + "RetireProducerRequest", + (_message.Message,), + { + "DESCRIPTOR": _RETIREPRODUCERREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.RetireProducerRequest) + }, +) +_sym_db.RegisterMessage(RetireProducerRequest) + +RetireProducerResponse = _reflection.GeneratedProtocolMessageType( + "RetireProducerResponse", + (_message.Message,), + { + "DESCRIPTOR": _RETIREPRODUCERRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.RetireProducerResponse) + }, +) +_sym_db.RegisterMessage(RetireProducerResponse) + +StreamFailure = _reflection.GeneratedProtocolMessageType( + "StreamFailure", + (_message.Message,), + { + "DESCRIPTOR": _STREAMFAILURE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StreamFailure) + }, +) +_sym_db.RegisterMessage(StreamFailure) + +StoreAppendRequest = _reflection.GeneratedProtocolMessageType( + "StoreAppendRequest", + (_message.Message,), + { + "DESCRIPTOR": _STOREAPPENDREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StoreAppendRequest) + }, +) +_sym_db.RegisterMessage(StoreAppendRequest) + +StoreAppendResponse = _reflection.GeneratedProtocolMessageType( + "StoreAppendResponse", + (_message.Message,), + { + "DESCRIPTOR": _STOREAPPENDRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StoreAppendResponse) + }, +) +_sym_db.RegisterMessage(StoreAppendResponse) + +StoreReadRequest = _reflection.GeneratedProtocolMessageType( + "StoreReadRequest", + (_message.Message,), + { + "DESCRIPTOR": _STOREREADREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StoreReadRequest) + }, +) +_sym_db.RegisterMessage(StoreReadRequest) + +StoredRecord = _reflection.GeneratedProtocolMessageType( + "StoredRecord", + (_message.Message,), + { + "DESCRIPTOR": _STOREDRECORD, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StoredRecord) + }, +) +_sym_db.RegisterMessage(StoredRecord) + +StoreReadResponse = _reflection.GeneratedProtocolMessageType( + "StoreReadResponse", + (_message.Message,), + { + "DESCRIPTOR": _STOREREADRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StoreReadResponse) + }, +) +_sym_db.RegisterMessage(StoreReadResponse) + +StoreLatestRequest = _reflection.GeneratedProtocolMessageType( + "StoreLatestRequest", + (_message.Message,), + { + "DESCRIPTOR": _STORELATESTREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StoreLatestRequest) + }, +) +_sym_db.RegisterMessage(StoreLatestRequest) + +StoreLatestResponse = _reflection.GeneratedProtocolMessageType( + "StoreLatestResponse", + (_message.Message,), + { + "DESCRIPTOR": _STORELATESTRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.StoreLatestResponse) + }, +) +_sym_db.RegisterMessage(StoreLatestResponse) + +_STREAMSERVICE = DESCRIPTOR.services_by_name["StreamService"] +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = ( + b"\352\002*Temporalio::Internal::Bridge::Api::Streams" + ) + _STREAMOWNERKIND._serialized_start = 3196 + _STREAMOWNERKIND._serialized_end = 3280 + _STREAMFAILUREKIND._serialized_start = 3283 + _STREAMFAILUREKIND._serialized_end = 3773 + _STREAMSTORECONFIG._serialized_start = 170 + _STREAMSTORECONFIG._serialized_end = 304 + _REDISSTORECONFIG._serialized_start = 307 + _REDISSTORECONFIG._serialized_end = 508 + _MEMORYSTORECONFIG._serialized_start = 510 + _MEMORYSTORECONFIG._serialized_end = 529 + _STREAMADDRESS._serialized_start = 532 + _STREAMADDRESS._serialized_end = 672 + _CHAINID._serialized_start = 674 + _CHAINID._serialized_end = 745 + _NAMEDPRODUCER._serialized_start = 747 + _NAMEDPRODUCER._serialized_end = 800 + _ACTIVITYPRODUCER._serialized_start = 802 + _ACTIVITYPRODUCER._serialized_end = 895 + _APPENDRECORD._serialized_start = 898 + _APPENDRECORD._serialized_end = 1038 + _APPENDREQUEST._serialized_start = 1041 + _APPENDREQUEST._serialized_end = 1302 + _APPENDRESPONSE._serialized_start = 1304 + _APPENDRESPONSE._serialized_end = 1363 + _READREQUEST._serialized_start = 1366 + _READREQUEST._serialized_end = 1519 + _SUPERSESSION._serialized_start = 1521 + _SUPERSESSION._serialized_end = 1614 + _READRECORD._serialized_start = 1617 + _READRECORD._serialized_end = 1780 + _READRESPONSE._serialized_start = 1782 + _READRESPONSE._serialized_end = 1887 + _LATESTREQUEST._serialized_start = 1889 + _LATESTREQUEST._serialized_end = 1952 + _LATESTRESPONSE._serialized_start = 1954 + _LATESTRESPONSE._serialized_end = 1986 + _CLOSEREQUEST._serialized_start = 1988 + _CLOSEREQUEST._serialized_end = 2050 + _CLOSERESPONSE._serialized_start = 2052 + _CLOSERESPONSE._serialized_end = 2067 + _DELETEOWNERREQUEST._serialized_start = 2069 + _DELETEOWNERREQUEST._serialized_end = 2183 + _DELETEOWNERRESPONSE._serialized_start = 2185 + _DELETEOWNERRESPONSE._serialized_end = 2223 + _RETIREPRODUCERREQUEST._serialized_start = 2226 + _RETIREPRODUCERREQUEST._serialized_end = 2413 + _RETIREPRODUCERRESPONSE._serialized_start = 2415 + _RETIREPRODUCERRESPONSE._serialized_end = 2439 + _STREAMFAILURE._serialized_start = 2441 + _STREAMFAILURE._serialized_end = 2539 + _STOREAPPENDREQUEST._serialized_start = 2542 + _STOREAPPENDREQUEST._serialized_end = 2707 + _STOREAPPENDRESPONSE._serialized_start = 2709 + _STOREAPPENDRESPONSE._serialized_end = 2777 + _STOREREADREQUEST._serialized_start = 2780 + _STOREREADREQUEST._serialized_end = 2940 + _STOREDRECORD._serialized_start = 2942 + _STOREDRECORD._serialized_end = 2990 + _STOREREADRESPONSE._serialized_start = 2992 + _STOREREADRESPONSE._serialized_end = 3075 + _STORELATESTREQUEST._serialized_start = 3077 + _STORELATESTREQUEST._serialized_end = 3153 + _STORELATESTRESPONSE._serialized_start = 3155 + _STORELATESTRESPONSE._serialized_end = 3194 + _STREAMSERVICE._serialized_start = 3776 + _STREAMSERVICE._serialized_end = 4271 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/bridge/proto/streams/streams_pb2.pyi b/temporalio/bridge/proto/streams/streams_pb2.pyi new file mode 100644 index 0000000000..0ae73e6f34 --- /dev/null +++ b/temporalio/bridge/proto/streams/streams_pb2.pyi @@ -0,0 +1,1150 @@ +""" +@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.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.internal.enum_type_wrapper +import google.protobuf.message + +import temporalio.api.common.v1.message_pb2 +import temporalio.bridge.proto.streams.v1.message_pb2 + +if sys.version_info >= (3, 10): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class _StreamOwnerKind: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamOwnerKindEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamOwnerKind.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_OWNER_KIND_UNSPECIFIED: _StreamOwnerKind.ValueType # 0 + STREAM_OWNER_KIND_WORKFLOW: _StreamOwnerKind.ValueType # 1 + +class StreamOwnerKind(_StreamOwnerKind, metaclass=_StreamOwnerKindEnumTypeWrapper): + """What owns a stream. The owner decides when the stream closes.""" + +STREAM_OWNER_KIND_UNSPECIFIED: StreamOwnerKind.ValueType # 0 +STREAM_OWNER_KIND_WORKFLOW: StreamOwnerKind.ValueType # 1 +global___StreamOwnerKind = StreamOwnerKind + +class _StreamFailureKind: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamFailureKindEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamFailureKind.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_FAILURE_KIND_UNSPECIFIED: _StreamFailureKind.ValueType # 0 + STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: _StreamFailureKind.ValueType # 1 + """The sequence was already written with different content.""" + STREAM_FAILURE_KIND_PRODUCER_STALE: _StreamFailureKind.ValueType # 2 + """The sequence is below the next one the store expects for this producer attempt.""" + STREAM_FAILURE_KIND_OUTCOME_UNKNOWN: _StreamFailureKind.ValueType # 3 + """The append may or may not have landed. Retrying the same batch is safe.""" + STREAM_FAILURE_KIND_REFUSED: _StreamFailureKind.ValueType # 4 + """The store refused the write, so nothing was written.""" + STREAM_FAILURE_KIND_CLOSED: _StreamFailureKind.ValueType # 5 + """The stream is closed and refuses appends.""" + STREAM_FAILURE_KIND_CURSOR: _StreamFailureKind.ValueType # 6 + """The cursor is not valid for this stream.""" + STREAM_FAILURE_KIND_EXPIRED: _StreamFailureKind.ValueType # 7 + """The cursor names a record the store no longer retains.""" + STREAM_FAILURE_KIND_NOT_FOUND: _StreamFailureKind.ValueType # 8 + """The owner does not exist, or nothing is known about the stream.""" + STREAM_FAILURE_KIND_RECORD: _StreamFailureKind.ValueType # 9 + """A stored record could not be read.""" + STREAM_FAILURE_KIND_STORAGE: _StreamFailureKind.ValueType # 10 + """The store failed or could not be reached.""" + STREAM_FAILURE_KIND_UNSUPPORTED: _StreamFailureKind.ValueType # 11 + """This release or this store does not offer the capability.""" + STREAM_FAILURE_KIND_INVALID_ARGUMENT: _StreamFailureKind.ValueType # 12 + """The call's input breaks the contract, such as a missing member or a value out of range, so + the same call fails the same way. A lang built before this kind maps it, as any kind it + doesn't know, to a storage failure. + """ + +class StreamFailureKind( + _StreamFailureKind, metaclass=_StreamFailureKindEnumTypeWrapper +): + """One class per way a stream call can fail, so every lang maps the same failure to the same + error. + """ + +STREAM_FAILURE_KIND_UNSPECIFIED: StreamFailureKind.ValueType # 0 +STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: StreamFailureKind.ValueType # 1 +"""The sequence was already written with different content.""" +STREAM_FAILURE_KIND_PRODUCER_STALE: StreamFailureKind.ValueType # 2 +"""The sequence is below the next one the store expects for this producer attempt.""" +STREAM_FAILURE_KIND_OUTCOME_UNKNOWN: StreamFailureKind.ValueType # 3 +"""The append may or may not have landed. Retrying the same batch is safe.""" +STREAM_FAILURE_KIND_REFUSED: StreamFailureKind.ValueType # 4 +"""The store refused the write, so nothing was written.""" +STREAM_FAILURE_KIND_CLOSED: StreamFailureKind.ValueType # 5 +"""The stream is closed and refuses appends.""" +STREAM_FAILURE_KIND_CURSOR: StreamFailureKind.ValueType # 6 +"""The cursor is not valid for this stream.""" +STREAM_FAILURE_KIND_EXPIRED: StreamFailureKind.ValueType # 7 +"""The cursor names a record the store no longer retains.""" +STREAM_FAILURE_KIND_NOT_FOUND: StreamFailureKind.ValueType # 8 +"""The owner does not exist, or nothing is known about the stream.""" +STREAM_FAILURE_KIND_RECORD: StreamFailureKind.ValueType # 9 +"""A stored record could not be read.""" +STREAM_FAILURE_KIND_STORAGE: StreamFailureKind.ValueType # 10 +"""The store failed or could not be reached.""" +STREAM_FAILURE_KIND_UNSUPPORTED: StreamFailureKind.ValueType # 11 +"""This release or this store does not offer the capability.""" +STREAM_FAILURE_KIND_INVALID_ARGUMENT: StreamFailureKind.ValueType # 12 +"""The call's input breaks the contract, such as a missing member or a value out of range, so +the same call fails the same way. A lang built before this kind maps it, as any kind it +doesn't know, to a storage failure. +""" +global___StreamFailureKind = StreamFailureKind + +class StreamStoreConfig(google.protobuf.message.Message): + """Which store holds a process's streams, and how to reach it. Lang builds one store from it and + sends its stream calls there. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + REDIS_FIELD_NUMBER: builtins.int + MEMORY_FIELD_NUMBER: builtins.int + @property + def redis(self) -> global___RedisStoreConfig: ... + @property + def memory(self) -> global___MemoryStoreConfig: ... + def __init__( + self, + *, + redis: global___RedisStoreConfig | None = ..., + memory: global___MemoryStoreConfig | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "memory", b"memory", "redis", b"redis", "store", b"store" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "memory", b"memory", "redis", b"redis", "store", b"store" + ], + ) -> None: ... + def WhichOneof( + self, oneof_group: typing_extensions.Literal["store", b"store"] + ) -> typing_extensions.Literal["redis", "memory"] | None: ... + +global___StreamStoreConfig = StreamStoreConfig + +class RedisStoreConfig(google.protobuf.message.Message): + """Streams in a Redis or Redis Cluster that Core connects to itself.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + URLS_FIELD_NUMBER: builtins.int + CLUSTER_FIELD_NUMBER: builtins.int + KEY_PREFIX_FIELD_NUMBER: builtins.int + RETENTION_FIELD_NUMBER: builtins.int + BLOCKING_READS_PER_NODE_FIELD_NUMBER: builtins.int + RESPONSE_TIMEOUT_FIELD_NUMBER: builtins.int + @property + def urls( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: + """One `redis://` or `rediss://` URL, or the seed nodes of a cluster. Credentials and TLS come + from the URL. + """ + cluster: builtins.bool + key_prefix: builtins.str + """Prepended to every key, so streams can share a Redis with other data and an ACL can scope + them. Empty uses `temporal-streams`. + """ + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: + """How long a stream keeps a record. Unset keeps seven days.""" + blocking_reads_per_node: builtins.int + """How many reads may wait on one node at once, each on a connection of its own. Zero uses the + default. + """ + @property + def response_timeout(self) -> google.protobuf.duration_pb2.Duration: + """How long a short call waits for its reply. Unset uses the default.""" + def __init__( + self, + *, + urls: collections.abc.Iterable[builtins.str] | None = ..., + cluster: builtins.bool = ..., + key_prefix: builtins.str = ..., + retention: google.protobuf.duration_pb2.Duration | None = ..., + blocking_reads_per_node: builtins.int = ..., + response_timeout: google.protobuf.duration_pb2.Duration | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "response_timeout", b"response_timeout", "retention", b"retention" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "blocking_reads_per_node", + b"blocking_reads_per_node", + "cluster", + b"cluster", + "key_prefix", + b"key_prefix", + "response_timeout", + b"response_timeout", + "retention", + b"retention", + "urls", + b"urls", + ], + ) -> None: ... + +global___RedisStoreConfig = RedisStoreConfig + +class MemoryStoreConfig(google.protobuf.message.Message): + """Streams in this process's memory, for tests and local development. Nothing survives the + process, and each connect makes a store of its own that shares no streams with another. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___MemoryStoreConfig = MemoryStoreConfig + +class StreamAddress(google.protobuf.message.Message): + """One stream, a topic of an owner in a namespace.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + OWNER_KIND_FIELD_NUMBER: builtins.int + WORKFLOW_ID_FIELD_NUMBER: builtins.int + RUN_ID_FIELD_NUMBER: builtins.int + TOPIC_FIELD_NUMBER: builtins.int + namespace: builtins.str + owner_kind: global___StreamOwnerKind.ValueType + workflow_id: builtins.str + run_id: builtins.str + """Pins the stream to the run chain that holds this run. Empty follows the chain that holds the + Workflow id when the call starts. + """ + topic: builtins.str + def __init__( + self, + *, + namespace: builtins.str = ..., + owner_kind: global___StreamOwnerKind.ValueType = ..., + workflow_id: builtins.str = ..., + run_id: builtins.str = ..., + topic: builtins.str = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "namespace", + b"namespace", + "owner_kind", + b"owner_kind", + "run_id", + b"run_id", + "topic", + b"topic", + "workflow_id", + b"workflow_id", + ], + ) -> None: ... + +global___StreamAddress = StreamAddress + +class ChainId(google.protobuf.message.Message): + """A Workflow's run chain. Its streams are keyed by its first run, so they follow + Continue-as-New, and a new chain on the same Workflow id gets streams of its own. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + WORKFLOW_ID_FIELD_NUMBER: builtins.int + FIRST_RUN_ID_FIELD_NUMBER: builtins.int + namespace: builtins.str + workflow_id: builtins.str + first_run_id: builtins.str + def __init__( + self, + *, + namespace: builtins.str = ..., + workflow_id: builtins.str = ..., + first_run_id: builtins.str = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "first_run_id", + b"first_run_id", + "namespace", + b"namespace", + "workflow_id", + b"workflow_id", + ], + ) -> None: ... + +global___ChainId = ChainId + +class NamedProducer(google.protobuf.message.Message): + """A producer that names itself.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + PRODUCER_ID_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + producer_id: builtins.str + attempt: builtins.int + """A process that restarts its work raises the attempt, which starts a new sequence and tells + readers that the earlier attempt's records are superseded. From one. + """ + def __init__( + self, + *, + producer_id: builtins.str = ..., + attempt: builtins.int = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "attempt", b"attempt", "producer_id", b"producer_id" + ], + ) -> None: ... + +global___NamedProducer = NamedProducer + +class ActivityProducer(google.protobuf.message.Message): + """An Activity attempt that writes. Core derives the producer id from it, so every SDK writes the + same one, and the attempt is the Activity's. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WORKFLOW_ID_FIELD_NUMBER: builtins.int + RUN_ID_FIELD_NUMBER: builtins.int + ACTIVITY_ID_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + workflow_id: builtins.str + run_id: builtins.str + activity_id: builtins.str + attempt: builtins.int + def __init__( + self, + *, + workflow_id: builtins.str = ..., + run_id: builtins.str = ..., + activity_id: builtins.str = ..., + attempt: builtins.int = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "activity_id", + b"activity_id", + "attempt", + b"attempt", + "run_id", + b"run_id", + "workflow_id", + b"workflow_id", + ], + ) -> None: ... + +global___ActivityProducer = ActivityProducer + +class AppendRecord(google.protobuf.message.Message): + """One record lang appends.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + KIND_FIELD_NUMBER: builtins.int + BODY_FIELD_NUMBER: builtins.int + CONTENT_HASH_FIELD_NUMBER: builtins.int + kind: temporalio.bridge.proto.streams.v1.message_pb2.StreamRecordKind.ValueType + @property + def body(self) -> temporalio.api.common.v1.message_pb2.Payload: + """Set on DATA only, after lang's payload codec and external storage.""" + content_hash: builtins.bytes + """The SHA-256 of the body as the payload converter made it, before the codec. A codec that + encrypts with a fresh nonce makes a retry's bytes differ, so a retry is recognized by this. + Empty when there is no body. + """ + def __init__( + self, + *, + kind: temporalio.bridge.proto.streams.v1.message_pb2.StreamRecordKind.ValueType = ..., + body: temporalio.api.common.v1.message_pb2.Payload | None = ..., + content_hash: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["body", b"body"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "body", b"body", "content_hash", b"content_hash", "kind", b"kind" + ], + ) -> None: ... + +global___AppendRecord = AppendRecord + +class AppendRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STREAM_FIELD_NUMBER: builtins.int + NAMED_FIELD_NUMBER: builtins.int + ACTIVITY_FIELD_NUMBER: builtins.int + SEQUENCE_FIELD_NUMBER: builtins.int + RECORDS_FIELD_NUMBER: builtins.int + DIGEST_FIELD_NUMBER: builtins.int + @property + def stream(self) -> global___StreamAddress: ... + @property + def named(self) -> global___NamedProducer: ... + @property + def activity(self) -> global___ActivityProducer: ... + sequence: builtins.int + """The first record's position within the producer attempt, from one.""" + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + global___AppendRecord + ]: ... + digest: builtins.bytes + """The batch's identity, which the store compares when the producer repeats its newest batch. + Lang takes it before the codec, as the SHA-256 over each converted record as its length in 8 + bytes big-endian, then its deterministic serialization. Deterministic serialization is stable + within one protobuf library only, so only the SDK that made a digest can repeat the batch and + dedupe. Another SDK's retry of the same batch fails as PRODUCER_DIVERGENT. Required. + """ + def __init__( + self, + *, + stream: global___StreamAddress | None = ..., + named: global___NamedProducer | None = ..., + activity: global___ActivityProducer | None = ..., + sequence: builtins.int = ..., + records: collections.abc.Iterable[global___AppendRecord] | None = ..., + digest: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "activity", + b"activity", + "named", + b"named", + "producer", + b"producer", + "stream", + b"stream", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "activity", + b"activity", + "digest", + b"digest", + "named", + b"named", + "producer", + b"producer", + "records", + b"records", + "sequence", + b"sequence", + "stream", + b"stream", + ], + ) -> None: ... + def WhichOneof( + self, oneof_group: typing_extensions.Literal["producer", b"producer"] + ) -> typing_extensions.Literal["named", "activity"] | None: ... + +global___AppendRequest = AppendRequest + +class AppendResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FIRST_CURSOR_FIELD_NUMBER: builtins.int + LAST_CURSOR_FIELD_NUMBER: builtins.int + first_cursor: builtins.str + """The cursors of the batch's first and last records. A retry of the attempt's newest batch + answers with the cursors the batch got the first time. + """ + last_cursor: builtins.str + def __init__( + self, + *, + first_cursor: builtins.str = ..., + last_cursor: builtins.str = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "first_cursor", b"first_cursor", "last_cursor", b"last_cursor" + ], + ) -> None: ... + +global___AppendResponse = AppendResponse + +class ReadRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STREAM_FIELD_NUMBER: builtins.int + AFTER_FIELD_NUMBER: builtins.int + WAIT_FIELD_NUMBER: builtins.int + MAX_RECORDS_FIELD_NUMBER: builtins.int + STATE_FIELD_NUMBER: builtins.int + @property + def stream(self) -> global___StreamAddress: ... + after: builtins.str + """Read strictly after this cursor. Empty starts at the oldest record the stream retains, and + `$end` at whatever is appended after the read starts. + """ + @property + def wait(self) -> google.protobuf.duration_pb2.Duration: + """How long to wait for a record when none is there. Zero answers at once. Core waits at most + 60 seconds, the stream service contract's long-poll maximum. + """ + max_records: builtins.int + """How many stored records one answer holds at most. A synthesized supersession comes on top. + Zero lets Core choose. Core answers with at most 1,000 records. + """ + state: builtins.bytes + """The `state` of the previous answer of the same read. Empty on its first call.""" + def __init__( + self, + *, + stream: global___StreamAddress | None = ..., + after: builtins.str = ..., + wait: google.protobuf.duration_pb2.Duration | None = ..., + max_records: builtins.int = ..., + state: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal["stream", b"stream", "wait", b"wait"], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "after", + b"after", + "max_records", + b"max_records", + "state", + b"state", + "stream", + b"stream", + "wait", + b"wait", + ], + ) -> None: ... + +global___ReadRequest = ReadRequest + +class Supersession(google.protobuf.message.Message): + """A later attempt of a producer started writing. Readers synthesize it, and no store holds it.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + TOPIC_FIELD_NUMBER: builtins.int + PRODUCER_ID_FIELD_NUMBER: builtins.int + PREVIOUS_ATTEMPT_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + topic: builtins.str + producer_id: builtins.str + previous_attempt: builtins.int + attempt: builtins.int + def __init__( + self, + *, + topic: builtins.str = ..., + producer_id: builtins.str = ..., + previous_attempt: builtins.int = ..., + attempt: builtins.int = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "attempt", + b"attempt", + "previous_attempt", + b"previous_attempt", + "producer_id", + b"producer_id", + "topic", + b"topic", + ], + ) -> None: ... + +global___Supersession = Supersession + +class ReadRecord(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + CURSOR_FIELD_NUMBER: builtins.int + STORED_FIELD_NUMBER: builtins.int + SUPERSEDED_FIELD_NUMBER: builtins.int + STALE_FIELD_NUMBER: builtins.int + cursor: builtins.str + """Hand this back as `after` to resume strictly past this record. A supersession carries the + cursor before the record that started the new attempt. + """ + @property + def stored(self) -> temporalio.bridge.proto.streams.v1.message_pb2.StreamRecord: + """As stored. Its body is still encoded.""" + @property + def superseded(self) -> global___Supersession: ... + stale: builtins.bool + """An older attempt wrote it after this read delivered a newer attempt of the same producer.""" + def __init__( + self, + *, + cursor: builtins.str = ..., + stored: temporalio.bridge.proto.streams.v1.message_pb2.StreamRecord + | None = ..., + superseded: global___Supersession | None = ..., + stale: builtins.bool = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "record", b"record", "stored", b"stored", "superseded", b"superseded" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "cursor", + b"cursor", + "record", + b"record", + "stale", + b"stale", + "stored", + b"stored", + "superseded", + b"superseded", + ], + ) -> None: ... + def WhichOneof( + self, oneof_group: typing_extensions.Literal["record", b"record"] + ) -> typing_extensions.Literal["stored", "superseded"] | None: ... + +global___ReadRecord = ReadRecord + +class ReadResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RECORDS_FIELD_NUMBER: builtins.int + CURSOR_FIELD_NUMBER: builtins.int + STATE_FIELD_NUMBER: builtins.int + DONE_FIELD_NUMBER: builtins.int + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + global___ReadRecord + ]: ... + cursor: builtins.str + """Where the next call resumes. Past the last record of this answer, or the request's cursor + when the answer is empty, with `$end` resolved. + """ + state: builtins.bytes + """Opaque to lang. Pass it to the next call of the same read, so attempts it has seen, the trim + watermark and the owner check carry over. + """ + done: builtins.bool + """The owner's chain ended and the read delivered every record the stream retains. No later + call returns a record. + """ + def __init__( + self, + *, + records: collections.abc.Iterable[global___ReadRecord] | None = ..., + cursor: builtins.str = ..., + state: builtins.bytes = ..., + done: builtins.bool = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "cursor", + b"cursor", + "done", + b"done", + "records", + b"records", + "state", + b"state", + ], + ) -> None: ... + +global___ReadResponse = ReadResponse + +class LatestRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STREAM_FIELD_NUMBER: builtins.int + @property + def stream(self) -> global___StreamAddress: ... + def __init__( + self, + *, + stream: global___StreamAddress | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["stream", b"stream"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing_extensions.Literal["stream", b"stream"] + ) -> None: ... + +global___LatestRequest = LatestRequest + +class LatestResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + CURSOR_FIELD_NUMBER: builtins.int + cursor: builtins.str + """Empty when the stream holds no record.""" + def __init__( + self, + *, + cursor: builtins.str = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["cursor", b"cursor"] + ) -> None: ... + +global___LatestResponse = LatestResponse + +class CloseRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STREAM_FIELD_NUMBER: builtins.int + @property + def stream(self) -> global___StreamAddress: ... + def __init__( + self, + *, + stream: global___StreamAddress | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["stream", b"stream"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing_extensions.Literal["stream", b"stream"] + ) -> None: ... + +global___CloseRequest = CloseRequest + +class CloseResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___CloseResponse = CloseResponse + +class DeleteOwnerRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + OWNER_KIND_FIELD_NUMBER: builtins.int + WORKFLOW_ID_FIELD_NUMBER: builtins.int + namespace: builtins.str + owner_kind: global___StreamOwnerKind.ValueType + workflow_id: builtins.str + def __init__( + self, + *, + namespace: builtins.str = ..., + owner_kind: global___StreamOwnerKind.ValueType = ..., + workflow_id: builtins.str = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "namespace", + b"namespace", + "owner_kind", + b"owner_kind", + "workflow_id", + b"workflow_id", + ], + ) -> None: ... + +global___DeleteOwnerRequest = DeleteOwnerRequest + +class DeleteOwnerResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + DELETED_FIELD_NUMBER: builtins.int + deleted: builtins.int + """How much the delete removed, counted the store's own way, such as Redis keys. Only zero + means the same on every store: nothing was left to delete. + """ + def __init__( + self, + *, + deleted: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["deleted", b"deleted"] + ) -> None: ... + +global___DeleteOwnerResponse = DeleteOwnerResponse + +class RetireProducerRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STREAM_FIELD_NUMBER: builtins.int + NAMED_FIELD_NUMBER: builtins.int + ACTIVITY_FIELD_NUMBER: builtins.int + @property + def stream(self) -> global___StreamAddress: ... + @property + def named(self) -> global___NamedProducer: ... + @property + def activity(self) -> global___ActivityProducer: ... + def __init__( + self, + *, + stream: global___StreamAddress | None = ..., + named: global___NamedProducer | None = ..., + activity: global___ActivityProducer | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "activity", + b"activity", + "named", + b"named", + "producer", + b"producer", + "stream", + b"stream", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "activity", + b"activity", + "named", + b"named", + "producer", + b"producer", + "stream", + b"stream", + ], + ) -> None: ... + def WhichOneof( + self, oneof_group: typing_extensions.Literal["producer", b"producer"] + ) -> typing_extensions.Literal["named", "activity"] | None: ... + +global___RetireProducerRequest = RetireProducerRequest + +class RetireProducerResponse(google.protobuf.message.Message): + """A producer attempt with no dedupe state left, or a stream that holds nothing, answers the same.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___RetireProducerResponse = RetireProducerResponse + +class StreamFailure(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + KIND_FIELD_NUMBER: builtins.int + MESSAGE_FIELD_NUMBER: builtins.int + CURSOR_FIELD_NUMBER: builtins.int + kind: global___StreamFailureKind.ValueType + message: builtins.str + cursor: builtins.str + """The cursor of the record that failed, for RECORD.""" + def __init__( + self, + *, + kind: global___StreamFailureKind.ValueType = ..., + message: builtins.str = ..., + cursor: builtins.str = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "cursor", b"cursor", "kind", b"kind", "message", b"message" + ], + ) -> None: ... + +global___StreamFailure = StreamFailure + +class StoreAppendRequest(google.protobuf.message.Message): + """The calls below are what Core asks of a store, below owner checks, cursors and supersession. + They name chains and store positions, not addresses and cursors. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + CHAIN_FIELD_NUMBER: builtins.int + TOPIC_FIELD_NUMBER: builtins.int + PRODUCER_ID_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + SEQUENCE_FIELD_NUMBER: builtins.int + DIGEST_FIELD_NUMBER: builtins.int + RECORDS_FIELD_NUMBER: builtins.int + @property + def chain(self) -> global___ChainId: ... + topic: builtins.str + producer_id: builtins.str + attempt: builtins.int + sequence: builtins.int + digest: builtins.bytes + """The batch's fingerprint, compared on a retry of the attempt's newest batch.""" + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[ + builtins.bytes + ]: + """Serialized `temporalio.bridge.proto.streams.v1.StreamRecord`s, ready to store.""" + def __init__( + self, + *, + chain: global___ChainId | None = ..., + topic: builtins.str = ..., + producer_id: builtins.str = ..., + attempt: builtins.int = ..., + sequence: builtins.int = ..., + digest: builtins.bytes = ..., + records: collections.abc.Iterable[builtins.bytes] | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["chain", b"chain"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "attempt", + b"attempt", + "chain", + b"chain", + "digest", + b"digest", + "producer_id", + b"producer_id", + "records", + b"records", + "sequence", + b"sequence", + "topic", + b"topic", + ], + ) -> None: ... + +global___StoreAppendRequest = StoreAppendRequest + +class StoreAppendResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FIRST_POSITION_FIELD_NUMBER: builtins.int + LAST_POSITION_FIELD_NUMBER: builtins.int + first_position: builtins.str + last_position: builtins.str + def __init__( + self, + *, + first_position: builtins.str = ..., + last_position: builtins.str = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "first_position", b"first_position", "last_position", b"last_position" + ], + ) -> None: ... + +global___StoreAppendResponse = StoreAppendResponse + +class StoreReadRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + CHAIN_FIELD_NUMBER: builtins.int + TOPIC_FIELD_NUMBER: builtins.int + AFTER_POSITION_FIELD_NUMBER: builtins.int + WAIT_FIELD_NUMBER: builtins.int + MAX_RECORDS_FIELD_NUMBER: builtins.int + @property + def chain(self) -> global___ChainId: ... + topic: builtins.str + after_position: builtins.str + """Read strictly after this position. Empty starts at the oldest retained record.""" + @property + def wait(self) -> google.protobuf.duration_pb2.Duration: ... + max_records: builtins.int + def __init__( + self, + *, + chain: global___ChainId | None = ..., + topic: builtins.str = ..., + after_position: builtins.str = ..., + wait: google.protobuf.duration_pb2.Duration | None = ..., + max_records: builtins.int = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["chain", b"chain", "wait", b"wait"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "after_position", + b"after_position", + "chain", + b"chain", + "max_records", + b"max_records", + "topic", + b"topic", + "wait", + b"wait", + ], + ) -> None: ... + +global___StoreReadRequest = StoreReadRequest + +class StoredRecord(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + POSITION_FIELD_NUMBER: builtins.int + RECORD_FIELD_NUMBER: builtins.int + position: builtins.str + record: builtins.bytes + """A serialized `temporalio.bridge.proto.streams.v1.StreamRecord`.""" + def __init__( + self, + *, + position: builtins.str = ..., + record: builtins.bytes = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "position", b"position", "record", b"record" + ], + ) -> None: ... + +global___StoredRecord = StoredRecord + +class StoreReadResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RECORDS_FIELD_NUMBER: builtins.int + CLOSED_FIELD_NUMBER: builtins.int + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + global___StoredRecord + ]: ... + closed: builtins.bool + """The chain or this topic is marked closed.""" + def __init__( + self, + *, + records: collections.abc.Iterable[global___StoredRecord] | None = ..., + closed: builtins.bool = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "closed", b"closed", "records", b"records" + ], + ) -> None: ... + +global___StoreReadResponse = StoreReadResponse + +class StoreLatestRequest(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + CHAIN_FIELD_NUMBER: builtins.int + TOPIC_FIELD_NUMBER: builtins.int + @property + def chain(self) -> global___ChainId: ... + topic: builtins.str + def __init__( + self, + *, + chain: global___ChainId | None = ..., + topic: builtins.str = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["chain", b"chain"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal["chain", b"chain", "topic", b"topic"], + ) -> None: ... + +global___StoreLatestRequest = StoreLatestRequest + +class StoreLatestResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + POSITION_FIELD_NUMBER: builtins.int + position: builtins.str + """Empty when the topic holds no record.""" + def __init__( + self, + *, + position: builtins.str = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["position", b"position"] + ) -> None: ... + +global___StoreLatestResponse = StoreLatestResponse diff --git a/temporalio/bridge/proto/streams/v1/__init__.py b/temporalio/bridge/proto/streams/v1/__init__.py new file mode 100644 index 0000000000..fc6311caae --- /dev/null +++ b/temporalio/bridge/proto/streams/v1/__init__.py @@ -0,0 +1,6 @@ +from .message_pb2 import StreamRecord, StreamRecordKind + +__all__ = [ + "StreamRecord", + "StreamRecordKind", +] diff --git a/temporalio/bridge/proto/streams/v1/message_pb2.py b/temporalio/bridge/proto/streams/v1/message_pb2.py new file mode 100644 index 0000000000..a9a9fc3285 --- /dev/null +++ b/temporalio/bridge/proto/streams/v1/message_pb2.py @@ -0,0 +1,69 @@ +# -*- 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\x42\x32\xea\x02/Temporalio::Internal::Bridge::Api::StreamRecordb\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 + DESCRIPTOR._serialized_options = ( + b"\352\002/Temporalio::Internal::Bridge::Api::StreamRecord" + ) + _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/bridge/proto/streams/v1/message_pb2.pyi b/temporalio/bridge/proto/streams/v1/message_pb2.pyi new file mode 100644 index 0000000000..8fac1a5972 --- /dev/null +++ b/temporalio/bridge/proto/streams/v1/message_pb2.pyi @@ -0,0 +1,162 @@ +""" +@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. Lang runs it through the payload + codec and external storage before it reaches any store, so a store and + Core see only the encoded payload. + """ + @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/temporalio/bridge/sdk-core b/temporalio/bridge/sdk-core index 472f94eef7..70f5d3e3e1 160000 --- a/temporalio/bridge/sdk-core +++ b/temporalio/bridge/sdk-core @@ -1 +1 @@ -Subproject commit 472f94eef798a79cb3424f02f85ae03160791d34 +Subproject commit 70f5d3e3e1da0ddf9b531055d25fdf6f6a711cb0 diff --git a/temporalio/bridge/src/lib.rs b/temporalio/bridge/src/lib.rs index a4fda26db6..2202a75e5f 100644 --- a/temporalio/bridge/src/lib.rs +++ b/temporalio/bridge/src/lib.rs @@ -6,6 +6,7 @@ mod client_rpc_generated; mod envconfig; mod metric; mod runtime; +mod streams; mod testing; mod worker; @@ -58,6 +59,14 @@ fn temporal_sdk_bridge(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { "PollShutdownError", py.get_type::(), )?; + // Streams + m.add( + "StreamFailureError", + py.get_type::(), + )?; + m.add_class::()?; + m.add_function(wrap_pyfunction!(streams::connect_stream_store, m)?)?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/temporalio/bridge/src/streams.rs b/temporalio/bridge/src/streams.rs new file mode 100644 index 0000000000..d9b342ff2f --- /dev/null +++ b/temporalio/bridge/src/streams.rs @@ -0,0 +1,66 @@ +//! The stream service Core serves in process, reached from Python. + +use std::sync::Arc; + +use prost::Message; +use pyo3::{create_exception, exceptions::PyException, exceptions::PyValueError, prelude::*}; +use temporalio_sdk_core::streams::{connect_stream_service, proto, StreamService}; + +use crate::{client::ClientRef, runtime}; + +// Carries a serialized `coresdk.streams.StreamFailure`, so Python maps every failure the same way +// whatever store raised it. +create_exception!(temporal_sdk_bridge, StreamFailureError, PyException); + +/// The stream service one connect call built over its store, so two calls get two stores. +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct StreamStoreRef { + pub(crate) service: Arc, + runtime: runtime::Runtime, +} + +/// Connects to the store a serialized `StreamStoreConfig` names, asking `client`'s server about +/// streams' owners. +#[pyfunction] +pub fn connect_stream_store<'p>( + py: Python<'p>, + client: &ClientRef, + config: Vec, +) -> PyResult> { + client + .runtime + .assert_same_process("connect a stream store")?; + let config = proto::StreamStoreConfig::decode(&*config) + .map_err(|err| PyValueError::new_err(format!("Invalid stream store config: {err}")))?; + let connection = client.connection.clone(); + let runtime = client.runtime.clone(); + client.runtime.future_into_py(py, async move { + let service = connect_stream_service(config, connection) + .await + .map_err(|err| failure_error(proto::StreamFailure::from(err)))?; + Ok(StreamStoreRef { service, runtime }) + }) +} + +#[pymethods] +impl StreamStoreRef { + /// Makes one stream service call with a serialized request, answering with the serialized + /// response. + fn call<'p>( + &self, + py: Python<'p>, + rpc: String, + request: Vec, + ) -> PyResult> { + self.runtime.assert_same_process("call a stream store")?; + let service = self.service.clone(); + self.runtime.future_into_py(py, async move { + service.call(&rpc, &request).await.map_err(failure_error) + }) + } +} + +fn failure_error(failure: proto::StreamFailure) -> PyErr { + StreamFailureError::new_err(failure.encode_to_vec()) +} diff --git a/temporalio/bridge/streams.py b/temporalio/bridge/streams.py new file mode 100644 index 0000000000..014b1daf68 --- /dev/null +++ b/temporalio/bridge/streams.py @@ -0,0 +1,85 @@ +"""Core's stream store, reached in process. + +Core serves the stream service against one store per process. Python sends each call as a +serialized request and gets the serialized response back, so the typed calls in +:mod:`temporalio.bridge.streams_generated` are generated from Core's protos. +""" + +from __future__ import annotations + +from typing import TypeVar + +import google.protobuf.message + +import temporalio.bridge.client +import temporalio.bridge.proto.streams +import temporalio.bridge.temporal_sdk_bridge + +ProtoMessage = TypeVar("ProtoMessage", bound=google.protobuf.message.Message) + + +class StreamCallFailure(Exception): + """A stream call that failed, with Core's account of why.""" + + def __init__(self, failure: temporalio.bridge.proto.streams.StreamFailure) -> None: + """Wrap Core's failure.""" + super().__init__(failure.message) + self.failure = failure + + +class StreamStore: + """The stream store Core serves to this process.""" + + @staticmethod + async def connect( + client: temporalio.bridge.client.Client, + config: temporalio.bridge.proto.streams.StreamStoreConfig, + ) -> StreamStore: + """Connect to the store ``config`` names. + + Core asks ``client``'s server about streams' owners. + + Raises: + StreamCallFailure: Core could not reach the store. + """ + try: + ref = await temporalio.bridge.temporal_sdk_bridge.connect_stream_store( + client._ref, config.SerializeToString() + ) + except temporalio.bridge.temporal_sdk_bridge.StreamFailureError as error: + raise _call_failure(error) from None + return StreamStore(ref) + + def __init__( + self, ref: temporalio.bridge.temporal_sdk_bridge.StreamStoreRef + ) -> None: + """Wrap the bridge's store reference.""" + self.ref = ref + + async def call( + self, + rpc: str, + req: google.protobuf.message.Message, + resp_type: type[ProtoMessage], + ) -> ProtoMessage: + """Make one stream service call and decode its answer. + + Raises: + StreamCallFailure: The call failed. + """ + try: + data = await self.ref.call(rpc, req.SerializeToString()) + except temporalio.bridge.temporal_sdk_bridge.StreamFailureError as error: + raise _call_failure(error) from None + resp = resp_type() + resp.ParseFromString(data) + return resp + + +def _call_failure( + error: temporalio.bridge.temporal_sdk_bridge.StreamFailureError, +) -> StreamCallFailure: + # The bridge raises with the serialized failure as its only argument. + failure = temporalio.bridge.proto.streams.StreamFailure() + failure.ParseFromString(error.args[0]) + return StreamCallFailure(failure) diff --git a/temporalio/bridge/streams_generated.py b/temporalio/bridge/streams_generated.py new file mode 100644 index 0000000000..9c6b6971ce --- /dev/null +++ b/temporalio/bridge/streams_generated.py @@ -0,0 +1,69 @@ +# Generated file. DO NOT EDIT +"""Generated calls for Core's in-process StreamService.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import temporalio.bridge.proto.streams + +if TYPE_CHECKING: + from temporalio.bridge.streams import StreamStore + + +class StreamService: + """Calls for the StreamService, which Core dispatches in process.""" + + def __init__(self, store: StreamStore) -> None: + """Initialize service with the provided store.""" + self._store = store + + async def append( + self, req: temporalio.bridge.proto.streams.AppendRequest + ) -> temporalio.bridge.proto.streams.AppendResponse: + """Invokes the StreamService.append call.""" + return await self._store.call( + "Append", req, temporalio.bridge.proto.streams.AppendResponse + ) + + async def close( + self, req: temporalio.bridge.proto.streams.CloseRequest + ) -> temporalio.bridge.proto.streams.CloseResponse: + """Invokes the StreamService.close call.""" + return await self._store.call( + "Close", req, temporalio.bridge.proto.streams.CloseResponse + ) + + async def delete_owner( + self, req: temporalio.bridge.proto.streams.DeleteOwnerRequest + ) -> temporalio.bridge.proto.streams.DeleteOwnerResponse: + """Invokes the StreamService.delete_owner call.""" + return await self._store.call( + "DeleteOwner", req, temporalio.bridge.proto.streams.DeleteOwnerResponse + ) + + async def latest( + self, req: temporalio.bridge.proto.streams.LatestRequest + ) -> temporalio.bridge.proto.streams.LatestResponse: + """Invokes the StreamService.latest call.""" + return await self._store.call( + "Latest", req, temporalio.bridge.proto.streams.LatestResponse + ) + + async def read( + self, req: temporalio.bridge.proto.streams.ReadRequest + ) -> temporalio.bridge.proto.streams.ReadResponse: + """Invokes the StreamService.read call.""" + return await self._store.call( + "Read", req, temporalio.bridge.proto.streams.ReadResponse + ) + + async def retire_producer( + self, req: temporalio.bridge.proto.streams.RetireProducerRequest + ) -> temporalio.bridge.proto.streams.RetireProducerResponse: + """Invokes the StreamService.retire_producer call.""" + return await self._store.call( + "RetireProducer", + req, + temporalio.bridge.proto.streams.RetireProducerResponse, + ) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py new file mode 100644 index 0000000000..f365744292 --- /dev/null +++ b/temporalio/contrib/streams/__init__.py @@ -0,0 +1,69 @@ +"""Streams: an ordered log a Workflow owns, written from many places. + +.. warning:: + This package is experimental and may change in future versions. + +A Workflow owns a stream. Producers outside Workflow code, such as the +Workflow's Activities and clients, append to it, and outside readers consume +it from a cursor. Records live in a store the application runs, which +Core reaches for every SDK. Record bodies never reach the server. + +The contract: + +1. **A new attempt supersedes the old one.** When a reader sees the first + record of a producer's newer attempt, it yields a + :attr:`RecordKind.SUPERSEDED` record first. It withdraws only what that + producer wrote in its earlier attempts, not the whole stream and not the + records of other producers. No store holds that record, so every store + reports a retry the same way. +2. **A cursor belongs to one stream.** Hand it back to resume strictly after + the record it names. A cursor from another store or another stream is + refused with :class:`StreamCursorError`, and one whose record + retention dropped with :class:`StreamExpiredError`. A read with no cursor + starts at :data:`BEGINNING` or :data:`END`. + +The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in +:mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` +as an ordinary payload. +""" + +from temporalio.contrib.streams._cursor import BEGINNING, END +from temporalio.contrib.streams._errors import ( + StreamClosedError, + StreamCursorError, + StreamError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamRecordError, + StreamRefusedError, + StreamStorageError, + StreamUnsupportedError, +) +from temporalio.contrib.streams._record import ( + Cursor, + RecordKind, + StreamRecord, + Supersession, +) + +__all__ = [ + "BEGINNING", + "Cursor", + "END", + "RecordKind", + "StreamClosedError", + "StreamCursorError", + "StreamError", + "StreamExpiredError", + "StreamNotFoundError", + "StreamOutcomeUnknownError", + "StreamProducerError", + "StreamRecord", + "StreamRecordError", + "StreamRefusedError", + "StreamStorageError", + "StreamUnsupportedError", + "Supersession", +] diff --git a/temporalio/contrib/streams/_body.py b/temporalio/contrib/streams/_body.py new file mode 100644 index 0000000000..7ecbfe361b --- /dev/null +++ b/temporalio/contrib/streams/_body.py @@ -0,0 +1,74 @@ +"""What lang owes a record's body between the converter and Core. + +The payload converter turns a value into the body and stops there. Every +other payload the SDK sends then passes through the payload codec and +external storage, and a stream body owes the same, or a codec-protected +deployment would leak plaintext through its streams. Core never sees a +plaintext body, so both hashes that need one are taken here, before the +codec. A codec that encrypts with a fresh nonce makes every retry's bytes +differ, so a store that compared encoded bytes would refuse a retry as a +divergent write. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Sequence + +from temporalio.api.common.v1 import Payload +from temporalio.bridge.proto.streams.v1 import StreamRecord as WireRecord +from temporalio.converter import DataConverter + +__all__ = ["batch_digest", "content_hash", "decode_body", "encode_bodies"] + + +def content_hash(payload: Payload) -> bytes: + """The SHA-256 of ``payload`` as the converter produced it. + + Taken over the deterministic serialization of the whole payload, metadata + included, so two payloads that differ only in their encoding hash apart. + Core stores it on the record as hex under ``temporal.io/content-hash``. + """ + return hashlib.sha256(payload.SerializeToString(deterministic=True)).digest() + + +def batch_digest(records: Sequence[WireRecord]) -> bytes: + """The identity of one append, taken over its converted records. + + Each record is its length in 8 bytes big-endian, then its deterministic + serialization, so a batch split differently can't collide with this one. + The store compares it when a producer repeats its newest batch. Only the + writer that made it compares it, since deterministic serialization is + stable within one protobuf implementation and not across languages, so a + retry must come from the same SDK as the batch it repeats. + """ + digest = hashlib.sha256() + for record in records: + body = record.SerializeToString(deterministic=True) + digest.update(len(body).to_bytes(8, "big")) + digest.update(body) + return digest.digest() + + +async def encode_bodies( + converter: DataConverter, payloads: Sequence[Payload] +) -> list[Payload]: + """``payloads`` as the store keeps them. + + Through the payload codec and then external storage, in the order + :meth:`temporalio.converter.DataConverter.encode` uses, so a body above + the external storage threshold is replaced by a claim. + """ + encoded = await converter._encode_payload_sequence(payloads) + return await converter._external_store_payload_sequence(encoded) + + +async def decode_body(converter: DataConverter, payload: Payload) -> Payload: + """Undo :func:`encode_bodies` on a body read back from the store. + + Raises: + RuntimeError: The body is a claim and ``converter`` has no external + storage to redeem it with. + """ + retrieved = await converter._external_retrieve_payload_sequence([payload]) + return (await converter._decode_payload_sequence(retrieved))[0] diff --git a/temporalio/contrib/streams/_cursor.py b/temporalio/contrib/streams/_cursor.py new file mode 100644 index 0000000000..2e38359cab --- /dev/null +++ b/temporalio/contrib/streams/_cursor.py @@ -0,0 +1,20 @@ +"""The cursors a read starts from without having read anything. + +Core mints and checks every other cursor, so its token format is Core's own. +""" + +from __future__ import annotations + +from temporalio.contrib.streams._record import Cursor + +__all__ = ["BEGINNING", "END"] + +BEGINNING = Cursor("") +"""Read from the oldest record the stream still retains.""" + +END = Cursor("$end") +"""Read only what is appended after the read starts. + +It is resolved when the read starts. To position a reader before the +reader's process writes something, use the handle's ``latest`` instead. +""" diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py new file mode 100644 index 0000000000..5099b454b7 --- /dev/null +++ b/temporalio/contrib/streams/_errors.py @@ -0,0 +1,169 @@ +"""The errors a stream call raises. + +Every stream condition is a :class:`StreamError`, so a caller can catch by +meaning the way it catches other :class:`temporalio.exceptions.TemporalError` +subclasses. Argument mistakes stay ``ValueError``. + +Two distinctions matter to a caller that retries. A write the store refused +(:class:`StreamRefusedError`, which includes :class:`StreamProducerError` +and :class:`StreamClosedError`) did not happen, and repeating it gets the +same answer until something changes. A failure of the store or of the way +to it (:class:`StreamStorageError`) may hide a write that landed: for an +append that is :class:`StreamOutcomeUnknownError`, and repeating it on the +same producer is safe because the store deduplicates the retry. Errors from +the store's client library never escape: they arrive as one of these. A cursor +that names a record the store no longer keeps +(:class:`StreamExpiredError`) is told apart from one that is not valid for +the stream at all (:class:`StreamCursorError`). A stored record that the +reader can't decode (:class:`StreamRecordError`) carries its cursor, so the +caller can resume past it on purpose. +""" + +from __future__ import annotations + +import temporalio.exceptions +from temporalio.bridge.proto.streams import StreamFailure, StreamFailureKind +from temporalio.contrib.streams._record import Cursor + +__all__ = [ + "StreamClosedError", + "StreamCursorError", + "StreamError", + "StreamExpiredError", + "StreamNotFoundError", + "StreamOutcomeUnknownError", + "StreamProducerError", + "StreamRecordError", + "StreamRefusedError", + "StreamStorageError", + "StreamUnsupportedError", +] + + +class StreamError(temporalio.exceptions.TemporalError): + """Base for stream conditions.""" + + +class StreamNotFoundError(StreamError): + """The stream's owner does not exist, or nothing is known about the stream.""" + + +class StreamCursorError(StreamError): + """The cursor is not valid for this stream. + + It was minted by another store or for another stream, or it is not a + token the store can parse. + """ + + +class StreamExpiredError(StreamCursorError): + """The cursor names a record the store no longer retains. + + The cursor was valid once, but retention dropped the records after it, + so resuming from it would skip records silently. Start again from + :data:`temporalio.contrib.streams.BEGINNING` or + :data:`temporalio.contrib.streams.END` and decide what to do about the + gap. + """ + + +class StreamRefusedError(StreamError): + """The store refused the write, so nothing was written. + + For example the store is out of memory, or a key holds a value of the + wrong type. The subclasses name the refusals the contract defines. + """ + + +class StreamStorageError(StreamError): + """The store failed, or could not be reached. + + For a read, retry later. For an append, the outcome is unknown and + :class:`StreamOutcomeUnknownError` is raised. + """ + + +class StreamClosedError(StreamRefusedError): + """The stream is closed and refuses appends. + + The store refused the write, so nothing was written. Records the stream + retains stay readable. + """ + + +class StreamProducerError(StreamRefusedError): + """The store refused the append because of the producer's identity. + + The sequence was already used with different content, or it is below the + newest sequence the store holds for this producer and attempt. Nothing + was written, and repeating the call gets the same answer. + + ``divergent`` is true when the sequence was already written with + different content, so the caller changed the batch on a retry. Otherwise + the sequence is stale, most often from a second producer object for the + same session. + """ + + def __init__(self, message: str, *, divergent: bool = False) -> None: + """Say whether the batch diverged or the sequence is stale.""" + super().__init__(message) + self.divergent = divergent + + +class StreamOutcomeUnknownError(StreamStorageError): + """The append may or may not have landed. + + The connection failed or timed out after the request left. Retrying the + same values on the same producer is safe: the producer keeps its + sequence until an append succeeds, and the store returns the original + position for a retry that repeats a batch it already holds. + """ + + +class StreamRecordError(StreamError): + """A stored record could not be turned into a record for this reader. + + Its body did not convert to the topic's type, or the codec or the store + could not decode it. Reading on from :attr:`cursor` skips it, which is a + choice the caller makes, since skipping loses the record. + """ + + def __init__(self, message: str, cursor: Cursor) -> None: + """Name the record that failed by its cursor.""" + super().__init__(message) + self.cursor = cursor + + +class StreamUnsupportedError(StreamError): + """The capability is not offered by this release or by this store.""" + + +_BY_KIND: dict[int, type[StreamError]] = { + StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: StreamProducerError, + StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_STALE: StreamProducerError, + StreamFailureKind.STREAM_FAILURE_KIND_OUTCOME_UNKNOWN: StreamOutcomeUnknownError, + StreamFailureKind.STREAM_FAILURE_KIND_REFUSED: StreamRefusedError, + StreamFailureKind.STREAM_FAILURE_KIND_CLOSED: StreamClosedError, + StreamFailureKind.STREAM_FAILURE_KIND_CURSOR: StreamCursorError, + StreamFailureKind.STREAM_FAILURE_KIND_EXPIRED: StreamExpiredError, + StreamFailureKind.STREAM_FAILURE_KIND_NOT_FOUND: StreamNotFoundError, + StreamFailureKind.STREAM_FAILURE_KIND_STORAGE: StreamStorageError, + StreamFailureKind.STREAM_FAILURE_KIND_UNSUPPORTED: StreamUnsupportedError, +} + + +def error_from_failure(failure: StreamFailure) -> StreamError | ValueError: + """The error a caller catches for Core's ``failure``. + + An input that breaks the call's contract is a ``ValueError``, as in the + rest of the SDK, so a caller can tell it from a store refusal. A kind this + release does not know is a storage error, since the call failed for a + reason the caller can't act on beyond retrying later. + """ + if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_INVALID_ARGUMENT: + return ValueError(failure.message) + if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_RECORD: + return StreamRecordError(failure.message, Cursor(failure.cursor)) + if failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: + return StreamProducerError(failure.message, divergent=True) + return _BY_KIND.get(failure.kind, StreamStorageError)(failure.message) diff --git a/temporalio/contrib/streams/_record.py b/temporalio/contrib/streams/_record.py new file mode 100644 index 0000000000..b1624b63bd --- /dev/null +++ b/temporalio/contrib/streams/_record.py @@ -0,0 +1,135 @@ +"""The value types the stream contract is expressed in. + +Nothing here touches Temporal or a store, so every store shares it +unchanged. +""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass +from typing import Generic, TypeVar + +__all__ = [ + "Cursor", + "RecordKind", + "StreamRecord", + "Supersession", +] + +T = TypeVar("T") + + +@enum.unique +class RecordKind(enum.IntEnum): + """What a record is. + + The stored kinds match ``temporal.sdk.streams.v1.StreamRecordKind`` + value for value, so a kind crosses the wire as the integer the proto + holds. :attr:`SUPERSEDED` is never stored. + """ + + UNSPECIFIED = 0 + """The proto's zero value. + + A stored record whose writer set no kind is read as :attr:`DATA`, so a + reader never sees this kind on a record. + """ + + DATA = 1 + """Carries a value published by a Workflow or a producer.""" + + FINISH = 2 + """The producer in ``producer_id`` will write nothing more on this topic. + + An empty ``producer_id`` names the owning Workflow. A Workflow's + ``FINISH`` covers one run: after Continue-as-New the next run can publish + on the topic again, under the same empty ``producer_id``. It does not end + a read and says nothing about the producer's outcome: an Activity can + still fail after it wrote ``FINISH``. + """ + + SUPERSEDED = 3 + """A later attempt of the producer in ``producer_id`` started writing. + + It withdraws only what that producer wrote on this topic in its earlier + attempts. Records of other producers stay valid, and so does the rest of + the stream. Match the earlier records on ``producer_id`` and an + ``attempt`` below :attr:`Supersession.attempt`. + + Core's reader synthesizes it from the records it observed. No store + holds it, so every store reports a retry the same way. Its cursor is the + position before the new attempt's first record. A reader that resumes + there gets this record again, then the new attempt, since dropping the + earlier attempt's records twice is harmless and missing it is not. + """ + + +@dataclass(frozen=True) +class Cursor: + """A position in one stream. + + Opaque on purpose. The token names the store that minted it and the + stream it belongs to, and a token from another store or another stream + is refused with + :class:`temporalio.contrib.streams.StreamCursorError`. Do not compare two + cursors or do arithmetic on one: hand a cursor back to resume strictly + after the record it names. + """ + + token: str + + def __str__(self) -> str: + """The store's position token.""" + return self.token + + +@dataclass(frozen=True) +class Supersession: + """What a :attr:`RecordKind.SUPERSEDED` record reports. + + ``producer_id`` started ``attempt``, so the records it wrote in + ``previous_attempt`` and before are withdrawn. No other producer's + records are. + """ + + producer_id: str + previous_attempt: int + attempt: int + + +@dataclass(frozen=True) +class StreamRecord(Generic[T]): + """One record as a reader sees it. + + ``value`` is set on a :attr:`RecordKind.DATA` record and + ``supersession`` on a :attr:`RecordKind.SUPERSEDED` one. Other kinds + carry neither, so a consumer narrows on ``kind`` first. + """ + + kind: RecordKind + cursor: Cursor + topic: str + producer_id: str = "" + """Who wrote it, or empty when the owning Workflow wrote it.""" + attempt: int = 0 + """The producer's attempt, or 0 when the producer declared none.""" + sequence: int = 0 + """The producer's position within its attempt, or 0 when it does not number.""" + run_id: str = "" + """The run that published it, when the owning Workflow did; empty otherwise. + + A stream follows its Workflow's run chain, so this tells a successor run + or a reset branch apart. An Activity's record leaves it empty, and the + scheduling run is in its ``producer_id``, as ``@``. + """ + value: T | None = None + """The published value. Set on ``DATA`` only.""" + supersession: Supersession | None = None + """The attempt change being reported. Set on ``SUPERSEDED`` only.""" + stale: bool = False + """An older attempt wrote it after this read delivered a newer attempt. + + Such a record is not part of the producer's current answer, for example + the last writes of an Activity attempt that timed out but kept running. + """ diff --git a/temporalio/contrib/streams/_wire.py b/temporalio/contrib/streams/_wire.py new file mode 100644 index 0000000000..9b3bb60e63 --- /dev/null +++ b/temporalio/contrib/streams/_wire.py @@ -0,0 +1,121 @@ +"""How a record crosses to Core and back: the proto is the record. + +``temporal.sdk.streams.v1.StreamRecord`` is the stored format on every +store, and Core builds and parses it. ``body`` is the user's payload, +produced and consumed through the payload converter, so a pre-encoded +:class:`temporalio.common.RawValue` passes through untouched. +""" + +from __future__ import annotations + +from typing import Any + +import temporalio.converter +from temporalio.bridge.proto.streams import ReadRecord +from temporalio.bridge.proto.streams.v1 import StreamRecord as WireRecord +from temporalio.bridge.proto.streams.v1 import StreamRecordKind +from temporalio.contrib.streams._body import decode_body +from temporalio.contrib.streams._errors import StreamRecordError +from temporalio.contrib.streams._record import ( + Cursor, + RecordKind, + StreamRecord, + Supersession, +) + +__all__ = ["RUN_ID_KEY", "WireRecord", "from_read", "to_wire"] + +RUN_ID_KEY = "temporal.io/run-id" +"""The record metadata key the producing run id is stored under. + +Core stamps it on every record the owning Workflow publishes, because a +stream follows the Workflow's run chain and a reader needs the run to tell a +reset branch or a successor run apart. Its value is a payload with +``encoding`` ``binary/plain`` whose data is the run id. +""" + + +def to_wire( + converter: temporalio.converter.PayloadConverter, + *, + topic: str, + kind: RecordKind, + value: Any = None, + producer_id: str = "", + attempt: int = 0, + sequence: int = 0, +) -> WireRecord: + """The record as the converter produced it, before the codec. + + Only a ``DATA`` record carries a body; the converter encodes ``value`` + into it. + + Raises: + ValueError: ``kind`` is ``SUPERSEDED`` or ``UNSPECIFIED``, which no + writer stores. + """ + if kind not in (RecordKind.DATA, RecordKind.FINISH): + raise ValueError(f"a writer stores DATA or FINISH records, not {kind.name}") + record = WireRecord( + topic=topic, + kind=StreamRecordKind.ValueType(int(kind)), + producer_id=producer_id, + attempt=attempt, + sequence=sequence, + ) + if kind is RecordKind.DATA: + record.body.CopyFrom(converter.to_payloads([value])[0]) + return record + + +async def from_read( + converter: temporalio.converter.DataConverter, + read: ReadRecord, + result_type: type | None, +) -> StreamRecord[Any]: + """The record a reader yields for one record of Core's answer. + + Raises: + StreamRecordError: The body did not decode or convert, or the record + holds a kind this SDK does not know. + """ + cursor = Cursor(read.cursor) + if read.HasField("superseded"): + 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 + try: + kind = RecordKind(wire.kind) + if kind is RecordKind.SUPERSEDED: + raise ValueError("a SUPERSEDED record is synthesized, never stored") + value: Any = None + if kind in (RecordKind.DATA, RecordKind.UNSPECIFIED) and wire.HasField("body"): + body = await decode_body(converter, wire.body) + hints = [result_type] if result_type is not None else None + value = converter.payload_converter.from_payloads([body], hints)[0] + except Exception as error: + raise StreamRecordError( + f"stream record at {cursor} could not be decoded: {error}", cursor + ) from error + return StreamRecord( + kind=RecordKind.DATA if kind is RecordKind.UNSPECIFIED else kind, + cursor=cursor, + topic=wire.topic, + producer_id=wire.producer_id, + attempt=wire.attempt, + sequence=wire.sequence, + run_id=wire.metadata[RUN_ID_KEY].data.decode() + if RUN_ID_KEY in wire.metadata + else "", + value=value, + stale=read.stale, + ) diff --git a/tests/bridge/test_stream_store.py b/tests/bridge/test_stream_store.py new file mode 100644 index 0000000000..0323ac16ca --- /dev/null +++ b/tests/bridge/test_stream_store.py @@ -0,0 +1,116 @@ +"""Core's in-process stream service, reached through the bridge.""" + +from __future__ import annotations + +import hashlib +import uuid +from typing import Any, cast + +import pytest + +import temporalio.bridge.client +from temporalio.api.common.v1 import Payload +from temporalio.bridge.proto.streams import ( + AppendRecord, + AppendRequest, + LatestRequest, + MemoryStoreConfig, + NamedProducer, + ReadRequest, + StreamAddress, + StreamFailureKind, + StreamOwnerKind, + StreamStoreConfig, +) +from temporalio.bridge.proto.streams.v1 import StreamRecordKind +from temporalio.bridge.streams import StreamCallFailure, StreamStore +from temporalio.bridge.streams_generated import StreamService +from temporalio.client import Client + + +async def memory_service(client: Client) -> StreamService: + bridge = cast(Any, client.service_client) + connected: temporalio.bridge.client.Client = await bridge._connected_client() + store = await StreamStore.connect( + connected, StreamStoreConfig(memory=MemoryStoreConfig()) + ) + return StreamService(store) + + +async def running_owner(client: Client) -> str: + # The service asks the server about the owner, so it has to exist. No Worker runs it. + workflow_id = f"stream-owner-{uuid.uuid4()}" + await client.start_workflow( + "StreamOwner", id=workflow_id, task_queue=f"unpolled-{uuid.uuid4()}" + ) + return workflow_id + + +def address(client: Client, workflow_id: str) -> StreamAddress: + return StreamAddress( + namespace=client.namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=workflow_id, + topic="out", + ) + + +def append_request( + stream: StreamAddress, *bodies: bytes, sequence: int = 1 +) -> AppendRequest: + return AppendRequest( + stream=stream, + named=NamedProducer(producer_id="p", attempt=1), + sequence=sequence, + records=[ + AppendRecord( + kind=StreamRecordKind.STREAM_RECORD_KIND_DATA, + body=Payload(data=body), + content_hash=hashlib.sha256(body).digest(), + ) + for body in bodies + ], + # Stands in for lang's batch digest, which the store compares on a retry. + digest=hashlib.sha256(b"".join(bodies)).digest(), + ) + + +async def test_an_append_reads_back_through_the_generated_stub(client: Client): + service = await memory_service(client) + stream = address(client, await running_owner(client)) + + appended = await service.append(append_request(stream, b"a", b"b")) + read = await service.read(ReadRequest(stream=stream)) + latest = await service.latest(LatestRequest(stream=stream)) + + assert [record.stored.body.data for record in read.records] == [b"a", b"b"] + assert read.records[0].cursor == appended.first_cursor + assert read.records[-1].cursor == appended.last_cursor + assert read.cursor == appended.last_cursor + assert latest.cursor == appended.last_cursor + + +async def test_a_refused_call_crosses_as_core_s_failure(client: Client): + service = await memory_service(client) + stream = address(client, await running_owner(client)) + await service.append(append_request(stream, b"a")) + + # The same sequence with other content is a divergent retry. + with pytest.raises(StreamCallFailure) as refused: + await service.append(append_request(stream, b"other")) + + assert ( + refused.value.failure.kind + == StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT + ) + assert refused.value.failure.message + + +async def test_a_stream_without_an_owner_is_not_found(client: Client): + service = await memory_service(client) + stream = address(client, f"no-such-owner-{uuid.uuid4()}") + + with pytest.raises(StreamCallFailure) as refused: + await service.append(append_request(stream, b"a")) + + assert refused.value.failure.kind == StreamFailureKind.STREAM_FAILURE_KIND_NOT_FOUND 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_errors_cursors.py b/tests/contrib/streams/test_errors_cursors.py new file mode 100644 index 0000000000..2d75d38456 --- /dev/null +++ b/tests/contrib/streams/test_errors_cursors.py @@ -0,0 +1,117 @@ +"""The error family, Core's failure kinds, and the cursors a read starts from.""" + +from __future__ import annotations + +from temporalio.bridge.proto.streams import StreamFailure, StreamFailureKind +from temporalio.contrib.streams import ( + BEGINNING, + END, + Cursor, + StreamClosedError, + StreamCursorError, + StreamError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamRecordError, + StreamRefusedError, + StreamStorageError, + StreamUnsupportedError, +) +from temporalio.contrib.streams._errors import error_from_failure + + +def test_every_condition_is_a_stream_error(): + for error in ( + StreamClosedError, + StreamCursorError, + StreamExpiredError, + StreamNotFoundError, + StreamOutcomeUnknownError, + StreamProducerError, + StreamRecordError, + StreamUnsupportedError, + ): + assert issubclass(error, StreamError) + # Expired is a kind of invalid cursor, so a caller that only resets on a + # bad cursor still catches it, and one that cares can tell them apart. + assert issubclass(StreamExpiredError, StreamCursorError) + # Refused and unknown are different answers to a retrying caller. + assert not issubclass(StreamOutcomeUnknownError, StreamProducerError) + assert not issubclass(StreamProducerError, StreamOutcomeUnknownError) + + +def test_refused_and_storage_failures_are_families_of_their_own(): + # A refused write did not happen; a storage failure may have. + assert issubclass(StreamProducerError, StreamRefusedError) + assert issubclass(StreamClosedError, StreamRefusedError) + assert issubclass(StreamOutcomeUnknownError, StreamStorageError) + assert not issubclass(StreamRefusedError, StreamStorageError) + assert not issubclass(StreamStorageError, StreamRefusedError) + + +def test_each_core_failure_kind_raises_its_error(): + expected = { + StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT: StreamProducerError, + StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_STALE: StreamProducerError, + StreamFailureKind.STREAM_FAILURE_KIND_OUTCOME_UNKNOWN: StreamOutcomeUnknownError, + StreamFailureKind.STREAM_FAILURE_KIND_REFUSED: StreamRefusedError, + StreamFailureKind.STREAM_FAILURE_KIND_CLOSED: StreamClosedError, + StreamFailureKind.STREAM_FAILURE_KIND_CURSOR: StreamCursorError, + StreamFailureKind.STREAM_FAILURE_KIND_EXPIRED: StreamExpiredError, + StreamFailureKind.STREAM_FAILURE_KIND_NOT_FOUND: StreamNotFoundError, + StreamFailureKind.STREAM_FAILURE_KIND_RECORD: StreamRecordError, + StreamFailureKind.STREAM_FAILURE_KIND_STORAGE: StreamStorageError, + StreamFailureKind.STREAM_FAILURE_KIND_UNSUPPORTED: StreamUnsupportedError, + StreamFailureKind.STREAM_FAILURE_KIND_INVALID_ARGUMENT: ValueError, + } + # Every kind Core defines has its error, so a new kind fails here first. + assert set(expected) == set(StreamFailureKind.values()) - { + StreamFailureKind.STREAM_FAILURE_KIND_UNSPECIFIED + } + for kind, error_type in expected.items(): + error = error_from_failure(StreamFailure(kind=kind, message="why")) + assert type(error) is error_type + assert str(error) == "why" + + +def test_a_record_failure_carries_the_cursor_to_resume_past(): + error = error_from_failure( + StreamFailure( + kind=StreamFailureKind.STREAM_FAILURE_KIND_RECORD, + message="bad record", + cursor="memory:abc:7", + ) + ) + assert isinstance(error, StreamRecordError) + assert error.cursor == Cursor("memory:abc:7") + + +def test_a_kind_this_release_does_not_know_is_a_storage_error(): + for kind in ( + StreamFailureKind.STREAM_FAILURE_KIND_UNSPECIFIED, + StreamFailureKind.ValueType(99), + ): + error = error_from_failure(StreamFailure(kind=kind, message="new")) + assert type(error) is StreamStorageError + + +def test_the_start_cursors_are_the_tokens_core_reads(): + # An empty `after` starts at the oldest record, and `$end` at what comes next. + assert BEGINNING.token == "" + assert END.token == "$end" + assert str(Cursor("memory:abc:7")) == "memory:abc:7" + + +def test_a_producer_refusal_says_whether_the_batch_diverged(): + # A divergent retry means the caller changed the batch, and a stale one + # usually means a second producer object for the same session. + divergent = error_from_failure( + StreamFailure(kind=StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_DIVERGENT) + ) + stale = error_from_failure( + StreamFailure(kind=StreamFailureKind.STREAM_FAILURE_KIND_PRODUCER_STALE) + ) + assert isinstance(divergent, StreamProducerError) and divergent.divergent + assert isinstance(stale, StreamProducerError) and not stale.divergent diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py new file mode 100644 index 0000000000..a3647d4eba --- /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 + +# The helpers behind the handles stay importable from their private modules, +# but they are not part of the public surface. +_PRIVATE = { + "batch_digest", + "content_hash", + "decode_body", + "encode_bodies", + "error_from_failure", +} + + +def test_internal_helpers_are_not_public(): + assert _PRIVATE.isdisjoint(streams.__all__) + for name in _PRIVATE: + assert not hasattr(streams, name), name diff --git a/tests/contrib/streams/test_records_wire.py b/tests/contrib/streams/test_records_wire.py new file mode 100644 index 0000000000..bc5a3e300e --- /dev/null +++ b/tests/contrib/streams/test_records_wire.py @@ -0,0 +1,195 @@ +"""Records, their stored form, and what lang owes a record's body.""" + +from __future__ import annotations + +import uuid +from collections.abc import Sequence + +import pytest + +from temporalio.api.common.v1 import Payload +from temporalio.bridge.proto.streams import ReadRecord, Supersession +from temporalio.bridge.proto.streams.v1 import StreamRecordKind +from temporalio.common import RawValue +from temporalio.contrib.streams import Cursor, RecordKind, StreamRecordError +from temporalio.contrib.streams import Supersession as ReportedSupersession +from temporalio.contrib.streams._body import ( + batch_digest, + content_hash, + decode_body, + encode_bodies, +) +from temporalio.contrib.streams._wire import RUN_ID_KEY, WireRecord, from_read, to_wire +from temporalio.converter import DataConverter, PayloadCodec + +CONVERTER = DataConverter.default + + +def stored(wire: WireRecord, cursor: str = "c", *, stale: bool = False) -> ReadRecord: + return ReadRecord(cursor=cursor, stored=wire, stale=stale) + + +async def test_a_record_reads_back_as_it_was_written(): + wire = to_wire( + CONVERTER.payload_converter, + topic="out", + kind=RecordKind.DATA, + value={"n": 1}, + producer_id="p", + attempt=2, + sequence=3, + ) + record = await from_read(CONVERTER, stored(wire, "c1", stale=True), dict) + assert record.kind is RecordKind.DATA + assert record.value == {"n": 1} + assert record.cursor == Cursor("c1") + assert (record.producer_id, record.attempt, record.sequence) == ("p", 2, 3) + assert record.stale + + finish = to_wire(CONVERTER.payload_converter, topic="out", kind=RecordKind.FINISH) + assert not finish.HasField("body") + assert (await from_read(CONVERTER, stored(finish), None)).kind is RecordKind.FINISH + + unset = WireRecord( + topic="out", body=CONVERTER.payload_converter.to_payloads([1])[0] + ) + assert (await from_read(CONVERTER, stored(unset), int)).kind is RecordKind.DATA + + +async def test_the_run_id_core_stamps_reaches_the_reader(): + wire = to_wire( + CONVERTER.payload_converter, topic="out", kind=RecordKind.DATA, value=1 + ) + assert (await from_read(CONVERTER, stored(wire), int)).run_id == "" + wire.metadata[RUN_ID_KEY].CopyFrom( + Payload(metadata={"encoding": b"binary/plain"}, data=b"run-1") + ) + assert (await from_read(CONVERTER, stored(wire), int)).run_id == "run-1" + + +async def test_a_supersession_reads_as_the_record_core_synthesized(): + read = ReadRecord( + cursor="c2", + superseded=Supersession( + topic="t", producer_id="model", previous_attempt=1, attempt=2 + ), + ) + record = await from_read(CONVERTER, read, int) + assert record.kind is RecordKind.SUPERSEDED + assert record.cursor == Cursor("c2") + assert (record.topic, record.producer_id, record.attempt) == ("t", "model", 2) + assert record.supersession == ReportedSupersession("model", 1, 2) + assert record.value is None + + +def test_no_writer_stores_a_synthesized_kind(): + for kind in (RecordKind.SUPERSEDED, RecordKind.UNSPECIFIED): + with pytest.raises(ValueError): + to_wire(CONVERTER.payload_converter, topic="out", kind=kind) + + +async def test_a_raw_value_passes_through_untouched(): + raw = Payload(metadata={"encoding": b"binary/custom"}, data=b"\x00\x01") + wire = to_wire( + CONVERTER.payload_converter, + topic="out", + kind=RecordKind.DATA, + value=RawValue(raw), + ) + assert wire.body == raw + record = await from_read(CONVERTER, stored(wire), RawValue) + assert isinstance(record.value, RawValue) + assert record.value.payload == raw + + +async def test_a_record_that_does_not_decode_raises_with_its_cursor(): + bad = WireRecord( + topic="t", + kind=RecordKind.DATA.value, # type: ignore[arg-type] + body=Payload(metadata={"encoding": b"json/plain"}, data=b"{not json"), + ) + with pytest.raises(StreamRecordError) as raised: + await from_read(CONVERTER, stored(bad, "c1"), int) + # The caller can resume past it on purpose. + assert raised.value.cursor == Cursor("c1") + unknown = WireRecord(topic="t", kind=7) # type: ignore[arg-type] + with pytest.raises(StreamRecordError): + await from_read(CONVERTER, stored(unknown, "c2"), int) + + +def test_the_content_hash_is_the_plaintext_payload_hash(): + converter = CONVERTER.payload_converter + one = converter.to_payloads([{"n": 1}])[0] + assert content_hash(one) == content_hash(converter.to_payloads([{"n": 1}])[0]) + assert content_hash(one) != content_hash(converter.to_payloads([{"n": 2}])[0]) + # Core refuses a hash that isn't a SHA-256. + assert len(content_hash(one)) == 32 + + +def records(*values: str) -> list[WireRecord]: + return [ + to_wire(CONVERTER.payload_converter, topic="t", kind=RecordKind.DATA, value=v) + for v in values + ] + + +def test_the_digest_is_length_delimited_over_the_batch(): + assert batch_digest(records("a", "b")) == batch_digest(records("a", "b")) + assert batch_digest(records("a", "b")) != batch_digest(records("b", "a")) + assert batch_digest(records("ab")) != batch_digest(records("a", "b")) + + +def test_the_digest_matches_what_every_sdk_takes(): + # The vector Core's tests hold, so a retry through another SDK still dedupes. + batch = [ + to_wire( + CONVERTER.payload_converter, + topic="events", + kind=RecordKind.DATA, + value={"n": n}, + producer_id="p", + attempt=1, + sequence=n, + ) + for n in (1, 2) + ] + assert batch_digest(batch).hex() == ( + "0499768af1856bc6fd19cf6cf90e28e971099f1680e8347d70bdb4b65273f30f" + ) + + +class NonceCodec(PayloadCodec): + """Encrypts with a fresh nonce per call, so two encodings never match.""" + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload( + metadata={"encoding": b"binary/nonce"}, + data=uuid.uuid4().bytes + p.SerializeToString(), + ) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [Payload.FromString(p.data[16:]) for p in payloads] + + +async def test_the_codec_runs_on_the_body_and_never_on_the_hashes(): + converter = DataConverter(payload_codec=NonceCodec()) + plain = records("a")[0].body + first, second = await encode_bodies(converter, [plain, plain]) + # The codec made the stored bytes differ, which the plaintext hash can't see. + assert first != second + assert first.metadata["encoding"] == b"binary/nonce" + assert await decode_body(converter, first) == plain + + record = await from_read( + converter, + stored( + WireRecord( + topic="t", kind=StreamRecordKind.STREAM_RECORD_KIND_DATA, body=first + ) + ), + str, + ) + assert record.value == "a"