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/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