diff --git a/.gitmodules b/.gitmodules index a4c9116312..914ed43b1f 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,3 +1,3 @@ [submodule "sdk-core"] path = temporalio/bridge/sdk-core - url = https://github.com/temporalio/sdk-rust.git + url = https://github.com/moetemp/sdk-rust.git diff --git a/pyproject.toml b/pyproject.toml index f803404db6..7eb3c5df3e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,8 +97,8 @@ dev = [ ] [tool.poe.tasks] -build-develop = "uv run maturin develop --uv" -build-develop-with-release = { cmd = "uv run maturin develop --release --uv" } +build-develop = "uv run maturin develop --uv --features streams" +build-develop-with-release = { cmd = "uv run maturin develop --release --uv --features streams" } # FIPS builds: swap the rustls stack onto aws-lc-rs FIPS mode (aws-lc-fips-sys), eliminating `ring`. build-develop-fips = { cmd = "uv run maturin develop --uv --no-default-features --features fips", env = { TEMPORALIO_FIPS = "1" } } build-wheel-fips = { cmd = "uv run maturin build --release --no-default-features --features fips", env = { TEMPORALIO_FIPS = "1" } } @@ -163,18 +163,20 @@ filterwarnings = [ before-all = "pip install protoc-wheel-0" build = "cp310-win_amd64 cp310-manylinux_x86_64 cp310-manylinux_aarch64 cp310-musllinux_x86_64 cp310-musllinux_aarch64 cp310-macosx_x86_64 cp310-macosx_arm64" build-verbosity = 1 +# The official wheels link the Stream runtime. +environment = { MATURIN_PEP517_ARGS = "--features streams" } [tool.cibuildwheel.macos] -environment = { MACOSX_DEPLOYMENT_TARGET = "10.12" } +environment = { MACOSX_DEPLOYMENT_TARGET = "10.12", MATURIN_PEP517_ARGS = "--features streams" } [tool.cibuildwheel.linux] before-all = "sh scripts/cibuildwheel_before_all_linux.sh" before-build = "sh scripts/cibuildwheel_before_build_linux.sh" -environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true" } +environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true", MATURIN_PEP517_ARGS = "--features streams" } [[tool.cibuildwheel.overrides]] select = "*musllinux*" -environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true", RUSTFLAGS = "-C target-feature=-crt-static" } +environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true", RUSTFLAGS = "-C target-feature=-crt-static", MATURIN_PEP517_ARGS = "--features streams" } [tool.mypy] ignore_missing_imports = true @@ -183,6 +185,8 @@ exclude = [ 'temporalio/api', 'temporalio/bridge/proto', 'temporalio/nexus/system/workflow_service', + # The Core submodule, which has a script of the crate's test vectors + 'temporalio/bridge/sdk-core', ] [[tool.mypy.overrides]] @@ -192,7 +196,7 @@ ignore_errors = true [tool.pydocstyle] convention = "google" # https://github.com/PyCQA/pydocstyle/issues/363#issuecomment-625563088 -match_dir = "^(?!(docs|scripts|tests|api|proto|system|_support|notifications|\\.)).*" +match_dir = "^(?!(docs|scripts|tests|api|proto|system|_support|notifications|sdk-core|\\.)).*" add_ignore = [ # We like to wrap at a certain number of chars, even long summary sentences. # https://github.com/PyCQA/pydocstyle/issues/184 @@ -259,10 +263,14 @@ exclude = [ "temporalio/bridge/proto", "temporalio/bridge/_visitor.py", "tests/worker/workflow_sandbox/testmodules/proto", + # The Core submodule, which has a script of the crate's test vectors + "temporalio/bridge/sdk-core", ] [tool.ruff] target-version = "py310" +# The Core submodule, which has a script of the crate's test vectors +extend-exclude = ["temporalio/bridge/sdk-core"] [build-system] requires = ["maturin>=1.0,<2.0"] diff --git a/temporalio/api/enums/v1/__init__.py b/temporalio/api/enums/v1/__init__.py index 518a2f32bc..db645f19b3 100644 --- a/temporalio/api/enums/v1/__init__.py +++ b/temporalio/api/enums/v1/__init__.py @@ -44,6 +44,12 @@ from .query_pb2 import QueryRejectCondition, QueryResultType from .reset_pb2 import ResetReapplyExcludeType, ResetReapplyType, ResetType from .schedule_pb2 import ScheduleOverlapPolicy +from .stream_pb2 import ( + StreamErrorReason, + StreamIdConflictPolicy, + StreamReadStart, + StreamStatus, +) from .task_queue_pb2 import ( BuildIdTaskReachability, DescribeTaskQueueMode, @@ -120,6 +126,10 @@ "Severity", "SignalExternalWorkflowExecutionFailedCause", "StartChildWorkflowExecutionFailedCause", + "StreamErrorReason", + "StreamIdConflictPolicy", + "StreamReadStart", + "StreamStatus", "SuggestContinueAsNewReason", "TaskQueueKind", "TaskQueueType", diff --git a/temporalio/api/enums/v1/stream_pb2.py b/temporalio/api/enums/v1/stream_pb2.py new file mode 100644 index 0000000000..b587fa6e9a --- /dev/null +++ b/temporalio/api/enums/v1/stream_pb2.py @@ -0,0 +1,70 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/api/enums/v1/stream.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() + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n"temporal/api/enums/v1/stream.proto\x12\x15temporal.api.enums.v1*z\n\x0cStreamStatus\x12\x1d\n\x19STREAM_STATUS_UNSPECIFIED\x10\x00\x12\x16\n\x12STREAM_STATUS_OPEN\x10\x01\x12\x18\n\x14STREAM_STATUS_CLOSED\x10\x02\x12\x19\n\x15STREAM_STATUS_EXPIRED\x10\x03*p\n\x0fStreamReadStart\x12!\n\x1dSTREAM_READ_START_UNSPECIFIED\x10\x00\x12\x1f\n\x1bSTREAM_READ_START_BEGINNING\x10\x01\x12\x19\n\x15STREAM_READ_START_END\x10\x02*\x93\x01\n\x16StreamIdConflictPolicy\x12)\n%STREAM_ID_CONFLICT_POLICY_UNSPECIFIED\x10\x00\x12"\n\x1eSTREAM_ID_CONFLICT_POLICY_FAIL\x10\x01\x12*\n&STREAM_ID_CONFLICT_POLICY_USE_EXISTING\x10\x02*\x90\x05\n\x11StreamErrorReason\x12#\n\x1fSTREAM_ERROR_REASON_UNSPECIFIED\x10\x00\x12!\n\x1dSTREAM_ERROR_REASON_NOT_FOUND\x10\x01\x12\x1f\n\x1bSTREAM_ERROR_REASON_EXPIRED\x10\x02\x12\x1e\n\x1aSTREAM_ERROR_REASON_CLOSED\x10\x03\x12\x1e\n\x1aSTREAM_ERROR_REASON_FENCED\x10\x04\x12(\n$STREAM_ERROR_REASON_SEQUENCE_TOO_LOW\x10\x05\x12%\n!STREAM_ERROR_REASON_SEQ_DIVERGENT\x10\x06\x12!\n\x1dSTREAM_ERROR_REASON_NOT_OWNER\x10\x07\x12&\n"STREAM_ERROR_REASON_FORMAT_TOO_NEW\x10\x08\x12\'\n#STREAM_ERROR_REASON_OFFSET_TOO_HIGH\x10\t\x12!\n\x1dSTREAM_ERROR_REASON_TOO_LARGE\x10\n\x12\x1f\n\x1bSTREAM_ERROR_REASON_INVALID\x10\x0b\x12+\n\'STREAM_ERROR_REASON_STORAGE_UNAVAILABLE\x10\x0c\x12&\n"STREAM_ERROR_REASON_UNKNOWN_WRITER\x10\r\x12#\n\x1fSTREAM_ERROR_REASON_UNSUPPORTED\x10\x0e\x12&\n"STREAM_ERROR_REASON_ALREADY_EXISTS\x10\x0f\x12\'\n#STREAM_ERROR_REASON_STORAGE_CORRUPT\x10\x10\x42\x83\x01\n\x18io.temporal.api.enums.v1B\x0bStreamProtoP\x01Z!go.temporal.io/api/enums/v1;enums\xaa\x02\x17Temporalio.Api.Enums.V1\xea\x02\x1aTemporalio::Api::Enums::V1b\x06proto3' +) + +_STREAMSTATUS = DESCRIPTOR.enum_types_by_name["StreamStatus"] +StreamStatus = enum_type_wrapper.EnumTypeWrapper(_STREAMSTATUS) +_STREAMREADSTART = DESCRIPTOR.enum_types_by_name["StreamReadStart"] +StreamReadStart = enum_type_wrapper.EnumTypeWrapper(_STREAMREADSTART) +_STREAMIDCONFLICTPOLICY = DESCRIPTOR.enum_types_by_name["StreamIdConflictPolicy"] +StreamIdConflictPolicy = enum_type_wrapper.EnumTypeWrapper(_STREAMIDCONFLICTPOLICY) +_STREAMERRORREASON = DESCRIPTOR.enum_types_by_name["StreamErrorReason"] +StreamErrorReason = enum_type_wrapper.EnumTypeWrapper(_STREAMERRORREASON) +STREAM_STATUS_UNSPECIFIED = 0 +STREAM_STATUS_OPEN = 1 +STREAM_STATUS_CLOSED = 2 +STREAM_STATUS_EXPIRED = 3 +STREAM_READ_START_UNSPECIFIED = 0 +STREAM_READ_START_BEGINNING = 1 +STREAM_READ_START_END = 2 +STREAM_ID_CONFLICT_POLICY_UNSPECIFIED = 0 +STREAM_ID_CONFLICT_POLICY_FAIL = 1 +STREAM_ID_CONFLICT_POLICY_USE_EXISTING = 2 +STREAM_ERROR_REASON_UNSPECIFIED = 0 +STREAM_ERROR_REASON_NOT_FOUND = 1 +STREAM_ERROR_REASON_EXPIRED = 2 +STREAM_ERROR_REASON_CLOSED = 3 +STREAM_ERROR_REASON_FENCED = 4 +STREAM_ERROR_REASON_SEQUENCE_TOO_LOW = 5 +STREAM_ERROR_REASON_SEQ_DIVERGENT = 6 +STREAM_ERROR_REASON_NOT_OWNER = 7 +STREAM_ERROR_REASON_FORMAT_TOO_NEW = 8 +STREAM_ERROR_REASON_OFFSET_TOO_HIGH = 9 +STREAM_ERROR_REASON_TOO_LARGE = 10 +STREAM_ERROR_REASON_INVALID = 11 +STREAM_ERROR_REASON_STORAGE_UNAVAILABLE = 12 +STREAM_ERROR_REASON_UNKNOWN_WRITER = 13 +STREAM_ERROR_REASON_UNSUPPORTED = 14 +STREAM_ERROR_REASON_ALREADY_EXISTS = 15 +STREAM_ERROR_REASON_STORAGE_CORRUPT = 16 + + +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = b"\n\030io.temporal.api.enums.v1B\013StreamProtoP\001Z!go.temporal.io/api/enums/v1;enums\252\002\027Temporalio.Api.Enums.V1\352\002\032Temporalio::Api::Enums::V1" + _STREAMSTATUS._serialized_start = 61 + _STREAMSTATUS._serialized_end = 183 + _STREAMREADSTART._serialized_start = 185 + _STREAMREADSTART._serialized_end = 297 + _STREAMIDCONFLICTPOLICY._serialized_start = 300 + _STREAMIDCONFLICTPOLICY._serialized_end = 447 + _STREAMERRORREASON._serialized_start = 450 + _STREAMERRORREASON._serialized_end = 1106 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/api/enums/v1/stream_pb2.pyi b/temporalio/api/enums/v1/stream_pb2.pyi new file mode 100644 index 0000000000..e314317c76 --- /dev/null +++ b/temporalio/api/enums/v1/stream_pb2.pyi @@ -0,0 +1,211 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import sys +import typing + +import google.protobuf.descriptor +import google.protobuf.internal.enum_type_wrapper + +if sys.version_info >= (3, 10): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class _StreamStatus: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamStatusEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamStatus.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_STATUS_UNSPECIFIED: _StreamStatus.ValueType # 0 + STREAM_STATUS_OPEN: _StreamStatus.ValueType # 1 + STREAM_STATUS_CLOSED: _StreamStatus.ValueType # 2 + STREAM_STATUS_EXPIRED: _StreamStatus.ValueType # 3 + +class StreamStatus(_StreamStatus, metaclass=_StreamStatusEnumTypeWrapper): + """Experimental. Might change in the future. + The state of a Stream, as a describe gives it. + (-- api-linter: core::0216::synonyms=disabled + aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) + """ + +STREAM_STATUS_UNSPECIFIED: StreamStatus.ValueType # 0 +STREAM_STATUS_OPEN: StreamStatus.ValueType # 1 +STREAM_STATUS_CLOSED: StreamStatus.ValueType # 2 +STREAM_STATUS_EXPIRED: StreamStatus.ValueType # 3 +global___StreamStatus = StreamStatus + +class _StreamReadStart: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamReadStartEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamReadStart.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_READ_START_UNSPECIFIED: _StreamReadStart.ValueType # 0 + STREAM_READ_START_BEGINNING: _StreamReadStart.ValueType # 1 + STREAM_READ_START_END: _StreamReadStart.ValueType # 2 + +class StreamReadStart(_StreamReadStart, metaclass=_StreamReadStartEnumTypeWrapper): + """Experimental. Might change in the future. + Where a read starts when it has no from_offset. + """ + +STREAM_READ_START_UNSPECIFIED: StreamReadStart.ValueType # 0 +STREAM_READ_START_BEGINNING: StreamReadStart.ValueType # 1 +STREAM_READ_START_END: StreamReadStart.ValueType # 2 +global___StreamReadStart = StreamReadStart + +class _StreamIdConflictPolicy: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamIdConflictPolicyEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamIdConflictPolicy.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_ID_CONFLICT_POLICY_UNSPECIFIED: _StreamIdConflictPolicy.ValueType # 0 + STREAM_ID_CONFLICT_POLICY_FAIL: _StreamIdConflictPolicy.ValueType # 1 + """The create fails with ALREADY_EXISTS.""" + STREAM_ID_CONFLICT_POLICY_USE_EXISTING: _StreamIdConflictPolicy.ValueType # 2 + """The create gets the open Stream, which keeps its own Retention.""" + +class StreamIdConflictPolicy( + _StreamIdConflictPolicy, metaclass=_StreamIdConflictPolicyEnumTypeWrapper +): + """Experimental. Might change in the future. + What a create does when its Stream exists. Unspecified is FAIL, as for a Workflow start. + """ + +STREAM_ID_CONFLICT_POLICY_UNSPECIFIED: StreamIdConflictPolicy.ValueType # 0 +STREAM_ID_CONFLICT_POLICY_FAIL: StreamIdConflictPolicy.ValueType # 1 +"""The create fails with ALREADY_EXISTS.""" +STREAM_ID_CONFLICT_POLICY_USE_EXISTING: StreamIdConflictPolicy.ValueType # 2 +"""The create gets the open Stream, which keeps its own Retention.""" +global___StreamIdConflictPolicy = StreamIdConflictPolicy + +class _StreamErrorReason: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamErrorReasonEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamErrorReason.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_ERROR_REASON_UNSPECIFIED: _StreamErrorReason.ValueType # 0 + STREAM_ERROR_REASON_NOT_FOUND: _StreamErrorReason.ValueType # 1 + """Nobody created the Stream, even after `create_wait`. NOT_FOUND.""" + STREAM_ERROR_REASON_EXPIRED: _StreamErrorReason.ValueType # 2 + """The Stream's Retention or safety TTL passed. NOT_FOUND.""" + STREAM_ERROR_REASON_CLOSED: _StreamErrorReason.ValueType # 3 + """The Stream is closed. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_FENCED: _StreamErrorReason.ValueType # 4 + """The Producer's epoch is lower than the newest one, its Workflow Task was withdrawn, or a + Workflow retry withdrew its run. FAILED_PRECONDITION. + """ + STREAM_ERROR_REASON_SEQUENCE_TOO_LOW: _StreamErrorReason.ValueType # 5 + """The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_SEQ_DIVERGENT: _StreamErrorReason.ValueType # 6 + """The batch has the `sequence` of the last one of its epoch, and other content. + FAILED_PRECONDITION. + """ + STREAM_ERROR_REASON_NOT_OWNER: _StreamErrorReason.ValueType # 7 + """A create gives another owner than the Stream has, or the owner doesn't allow this close. + FAILED_PRECONDITION. + """ + STREAM_ERROR_REASON_FORMAT_TOO_NEW: _StreamErrorReason.ValueType # 8 + """A newer format version wrote the Stream. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_OFFSET_TOO_HIGH: _StreamErrorReason.ValueType # 9 + """The offset is past the newest offset + 1, for example after a failover lost the newest + Records. A read from it works again once the Stream has Records up to it. OUT_OF_RANGE. + """ + STREAM_ERROR_REASON_TOO_LARGE: _StreamErrorReason.ValueType # 10 + """A Record or a batch is over the Store's limit. INVALID_ARGUMENT.""" + STREAM_ERROR_REASON_INVALID: _StreamErrorReason.ValueType # 11 + """The request is malformed. INVALID_ARGUMENT.""" + STREAM_ERROR_REASON_STORAGE_UNAVAILABLE: _StreamErrorReason.ValueType # 12 + """The Store failed or did not answer. A retry can succeed. UNAVAILABLE.""" + STREAM_ERROR_REASON_UNKNOWN_WRITER: _StreamErrorReason.ValueType # 13 + """No writer has the id. The SDK's in-process writer only. NOT_FOUND.""" + STREAM_ERROR_REASON_UNSUPPORTED: _StreamErrorReason.ValueType # 14 + """The Store lacks a capability that the call needs. UNIMPLEMENTED.""" + STREAM_ERROR_REASON_ALREADY_EXISTS: _StreamErrorReason.ValueType # 15 + """A create with the FAIL conflict policy is for a Stream that exists. ALREADY_EXISTS.""" + STREAM_ERROR_REASON_STORAGE_CORRUPT: _StreamErrorReason.ValueType # 16 + """The Store gave back data that isn't valid, such as a stored value with the wrong length. A + retry reads the same data. DATA_LOSS. + """ + +class StreamErrorReason( + _StreamErrorReason, metaclass=_StreamErrorReasonEnumTypeWrapper +): + """Experimental. Might change in the future. + Why a Stream call failed. Each reason has one gRPC status code. An SDK raises its own error + class for each reason that code can act on, and a base class for the others. + """ + +STREAM_ERROR_REASON_UNSPECIFIED: StreamErrorReason.ValueType # 0 +STREAM_ERROR_REASON_NOT_FOUND: StreamErrorReason.ValueType # 1 +"""Nobody created the Stream, even after `create_wait`. NOT_FOUND.""" +STREAM_ERROR_REASON_EXPIRED: StreamErrorReason.ValueType # 2 +"""The Stream's Retention or safety TTL passed. NOT_FOUND.""" +STREAM_ERROR_REASON_CLOSED: StreamErrorReason.ValueType # 3 +"""The Stream is closed. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_FENCED: StreamErrorReason.ValueType # 4 +"""The Producer's epoch is lower than the newest one, its Workflow Task was withdrawn, or a +Workflow retry withdrew its run. FAILED_PRECONDITION. +""" +STREAM_ERROR_REASON_SEQUENCE_TOO_LOW: StreamErrorReason.ValueType # 5 +"""The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_SEQ_DIVERGENT: StreamErrorReason.ValueType # 6 +"""The batch has the `sequence` of the last one of its epoch, and other content. +FAILED_PRECONDITION. +""" +STREAM_ERROR_REASON_NOT_OWNER: StreamErrorReason.ValueType # 7 +"""A create gives another owner than the Stream has, or the owner doesn't allow this close. +FAILED_PRECONDITION. +""" +STREAM_ERROR_REASON_FORMAT_TOO_NEW: StreamErrorReason.ValueType # 8 +"""A newer format version wrote the Stream. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_OFFSET_TOO_HIGH: StreamErrorReason.ValueType # 9 +"""The offset is past the newest offset + 1, for example after a failover lost the newest +Records. A read from it works again once the Stream has Records up to it. OUT_OF_RANGE. +""" +STREAM_ERROR_REASON_TOO_LARGE: StreamErrorReason.ValueType # 10 +"""A Record or a batch is over the Store's limit. INVALID_ARGUMENT.""" +STREAM_ERROR_REASON_INVALID: StreamErrorReason.ValueType # 11 +"""The request is malformed. INVALID_ARGUMENT.""" +STREAM_ERROR_REASON_STORAGE_UNAVAILABLE: StreamErrorReason.ValueType # 12 +"""The Store failed or did not answer. A retry can succeed. UNAVAILABLE.""" +STREAM_ERROR_REASON_UNKNOWN_WRITER: StreamErrorReason.ValueType # 13 +"""No writer has the id. The SDK's in-process writer only. NOT_FOUND.""" +STREAM_ERROR_REASON_UNSUPPORTED: StreamErrorReason.ValueType # 14 +"""The Store lacks a capability that the call needs. UNIMPLEMENTED.""" +STREAM_ERROR_REASON_ALREADY_EXISTS: StreamErrorReason.ValueType # 15 +"""A create with the FAIL conflict policy is for a Stream that exists. ALREADY_EXISTS.""" +STREAM_ERROR_REASON_STORAGE_CORRUPT: StreamErrorReason.ValueType # 16 +"""The Store gave back data that isn't valid, such as a stored value with the wrong length. A +retry reads the same data. DATA_LOSS. +""" +global___StreamErrorReason = StreamErrorReason diff --git a/temporalio/api/stream/__init__.py b/temporalio/api/stream/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/temporalio/api/stream/v1/__init__.py b/temporalio/api/stream/v1/__init__.py new file mode 100644 index 0000000000..6232b08cc0 --- /dev/null +++ b/temporalio/api/stream/v1/__init__.py @@ -0,0 +1,31 @@ +from .message_pb2 import ( + ActivityEpoch, + AppendRecord, + ClientEpoch, + LocalActivityEpoch, + Producer, + ProducerEpoch, + StreamData, + StreamErrorDetails, + StreamRecord, + StreamRef, + StreamSuperseded, + StreamWriter, + WorkflowTaskEpoch, +) + +__all__ = [ + "ActivityEpoch", + "AppendRecord", + "ClientEpoch", + "LocalActivityEpoch", + "Producer", + "ProducerEpoch", + "StreamData", + "StreamErrorDetails", + "StreamRecord", + "StreamRef", + "StreamSuperseded", + "StreamWriter", + "WorkflowTaskEpoch", +] diff --git a/temporalio/api/stream/v1/message_pb2.py b/temporalio/api/stream/v1/message_pb2.py new file mode 100644 index 0000000000..c3f561c549 --- /dev/null +++ b/temporalio/api/stream/v1/message_pb2.py @@ -0,0 +1,216 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/api/stream/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 + +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2 + +from temporalio.api.common.v1 import ( + message_pb2 as temporal_dot_api_dot_common_dot_v1_dot_message__pb2, +) +from temporalio.api.enums.v1 import ( + stream_pb2 as temporal_dot_api_dot_enums_dot_v1_dot_stream__pb2, +) + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n$temporal/api/stream/v1/message.proto\x12\x16temporal.api.stream.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a"temporal/api/enums/v1/stream.proto"Z\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\x30\n\x05owner\x18\x02 \x01(\x0b\x32!.temporal.api.common.v1.Execution\x12\x0f\n\x07\x62\x61\x63kend\x18\x03 \x01(\t"\xd4\x02\n\rProducerEpoch\x12=\n\x08workflow\x18\x01 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpochH\x00\x12\x44\n\x0elocal_activity\x18\x02 \x01(\x0b\x32*.temporal.api.stream.v1.LocalActivityEpochH\x00\x12\x39\n\x08\x61\x63tivity\x18\x03 \x01(\x0b\x32%.temporal.api.stream.v1.ActivityEpochH\x00\x12\x44\n\x13standalone_activity\x18\x04 \x01(\x0b\x32%.temporal.api.stream.v1.ActivityEpochH\x00\x12\x35\n\x06\x63lient\x18\x05 \x01(\x0b\x32#.temporal.api.stream.v1.ClientEpochH\x00\x42\x06\n\x04kind"p\n\x11WorkflowTaskEpoch\x12\x18\n\x10started_event_id\x18\x01 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05\x12\x30\n\x0cstarted_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp"^\n\x12LocalActivityEpoch\x12\x37\n\x04task\x18\x01 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpoch\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05"\\\n\rActivityEpoch\x12:\n\x16\x61ttempt_scheduled_time\x18\x01 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05"\x1e\n\x0b\x43lientEpoch\x12\x0f\n\x07\x61ttempt\x18\x01 \x01(\x05"\\\n\x08Producer\x12\n\n\x02id\x18\x01 \x01(\t\x12\x34\n\x05\x65poch\x18\x02 \x01(\x0b\x32%.temporal.api.stream.v1.ProducerEpoch\x12\x0e\n\x06run_id\x18\x03 \x01(\t"\xba\x01\n\x0cStreamWriter\x12\x32\n\x08producer\x18\x01 \x01(\x0b\x32 .temporal.api.stream.v1.Producer\x12J\n\x17previous_completed_task\x18\x02 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x03 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x04 \x01(\t"=\n\x0c\x41ppendRecord\x12-\n\x04\x62ody\x18\x01 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload"H\n\x10StreamSuperseded\x12\x34\n\x06writer\x18\x01 \x01(\x0b\x32$.temporal.api.stream.v1.StreamWriter"o\n\nStreamData\x12\x32\n\x08producer\x18\x01 \x01(\x0b\x32 .temporal.api.stream.v1.Producer\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload"\x9a\x01\n\x0cStreamRecord\x12\x0e\n\x06offset\x18\x01 \x01(\x03\x12\x32\n\x04\x64\x61ta\x18\x02 \x01(\x0b\x32".temporal.api.stream.v1.StreamDataH\x00\x12>\n\nsuperseded\x18\x03 \x01(\x0b\x32(.temporal.api.stream.v1.StreamSupersededH\x00\x42\x06\n\x04item"N\n\x12StreamErrorDetails\x12\x38\n\x06reason\x18\x01 \x01(\x0e\x32(.temporal.api.enums.v1.StreamErrorReasonB\x89\x01\n\x19io.temporal.api.stream.v1B\x0cMessageProtoP\x01Z#go.temporal.io/api/stream/v1;stream\xaa\x02\x18Temporalio.Api.Stream.V1\xea\x02\x1bTemporalio::Api::Stream::V1b\x06proto3' +) + + +_STREAMREF = DESCRIPTOR.message_types_by_name["StreamRef"] +_PRODUCEREPOCH = DESCRIPTOR.message_types_by_name["ProducerEpoch"] +_WORKFLOWTASKEPOCH = DESCRIPTOR.message_types_by_name["WorkflowTaskEpoch"] +_LOCALACTIVITYEPOCH = DESCRIPTOR.message_types_by_name["LocalActivityEpoch"] +_ACTIVITYEPOCH = DESCRIPTOR.message_types_by_name["ActivityEpoch"] +_CLIENTEPOCH = DESCRIPTOR.message_types_by_name["ClientEpoch"] +_PRODUCER = DESCRIPTOR.message_types_by_name["Producer"] +_STREAMWRITER = DESCRIPTOR.message_types_by_name["StreamWriter"] +_APPENDRECORD = DESCRIPTOR.message_types_by_name["AppendRecord"] +_STREAMSUPERSEDED = DESCRIPTOR.message_types_by_name["StreamSuperseded"] +_STREAMDATA = DESCRIPTOR.message_types_by_name["StreamData"] +_STREAMRECORD = DESCRIPTOR.message_types_by_name["StreamRecord"] +_STREAMERRORDETAILS = DESCRIPTOR.message_types_by_name["StreamErrorDetails"] +StreamRef = _reflection.GeneratedProtocolMessageType( + "StreamRef", + (_message.Message,), + { + "DESCRIPTOR": _STREAMREF, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamRef) + }, +) +_sym_db.RegisterMessage(StreamRef) + +ProducerEpoch = _reflection.GeneratedProtocolMessageType( + "ProducerEpoch", + (_message.Message,), + { + "DESCRIPTOR": _PRODUCEREPOCH, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.ProducerEpoch) + }, +) +_sym_db.RegisterMessage(ProducerEpoch) + +WorkflowTaskEpoch = _reflection.GeneratedProtocolMessageType( + "WorkflowTaskEpoch", + (_message.Message,), + { + "DESCRIPTOR": _WORKFLOWTASKEPOCH, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.WorkflowTaskEpoch) + }, +) +_sym_db.RegisterMessage(WorkflowTaskEpoch) + +LocalActivityEpoch = _reflection.GeneratedProtocolMessageType( + "LocalActivityEpoch", + (_message.Message,), + { + "DESCRIPTOR": _LOCALACTIVITYEPOCH, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.LocalActivityEpoch) + }, +) +_sym_db.RegisterMessage(LocalActivityEpoch) + +ActivityEpoch = _reflection.GeneratedProtocolMessageType( + "ActivityEpoch", + (_message.Message,), + { + "DESCRIPTOR": _ACTIVITYEPOCH, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.ActivityEpoch) + }, +) +_sym_db.RegisterMessage(ActivityEpoch) + +ClientEpoch = _reflection.GeneratedProtocolMessageType( + "ClientEpoch", + (_message.Message,), + { + "DESCRIPTOR": _CLIENTEPOCH, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.ClientEpoch) + }, +) +_sym_db.RegisterMessage(ClientEpoch) + +Producer = _reflection.GeneratedProtocolMessageType( + "Producer", + (_message.Message,), + { + "DESCRIPTOR": _PRODUCER, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.Producer) + }, +) +_sym_db.RegisterMessage(Producer) + +StreamWriter = _reflection.GeneratedProtocolMessageType( + "StreamWriter", + (_message.Message,), + { + "DESCRIPTOR": _STREAMWRITER, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamWriter) + }, +) +_sym_db.RegisterMessage(StreamWriter) + +AppendRecord = _reflection.GeneratedProtocolMessageType( + "AppendRecord", + (_message.Message,), + { + "DESCRIPTOR": _APPENDRECORD, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.AppendRecord) + }, +) +_sym_db.RegisterMessage(AppendRecord) + +StreamSuperseded = _reflection.GeneratedProtocolMessageType( + "StreamSuperseded", + (_message.Message,), + { + "DESCRIPTOR": _STREAMSUPERSEDED, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamSuperseded) + }, +) +_sym_db.RegisterMessage(StreamSuperseded) + +StreamData = _reflection.GeneratedProtocolMessageType( + "StreamData", + (_message.Message,), + { + "DESCRIPTOR": _STREAMDATA, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamData) + }, +) +_sym_db.RegisterMessage(StreamData) + +StreamRecord = _reflection.GeneratedProtocolMessageType( + "StreamRecord", + (_message.Message,), + { + "DESCRIPTOR": _STREAMRECORD, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamRecord) + }, +) +_sym_db.RegisterMessage(StreamRecord) + +StreamErrorDetails = _reflection.GeneratedProtocolMessageType( + "StreamErrorDetails", + (_message.Message,), + { + "DESCRIPTOR": _STREAMERRORDETAILS, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamErrorDetails) + }, +) +_sym_db.RegisterMessage(StreamErrorDetails) + +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = b"\n\031io.temporal.api.stream.v1B\014MessageProtoP\001Z#go.temporal.io/api/stream/v1;stream\252\002\030Temporalio.Api.Stream.V1\352\002\033Temporalio::Api::Stream::V1" + _STREAMREF._serialized_start = 171 + _STREAMREF._serialized_end = 261 + _PRODUCEREPOCH._serialized_start = 264 + _PRODUCEREPOCH._serialized_end = 604 + _WORKFLOWTASKEPOCH._serialized_start = 606 + _WORKFLOWTASKEPOCH._serialized_end = 718 + _LOCALACTIVITYEPOCH._serialized_start = 720 + _LOCALACTIVITYEPOCH._serialized_end = 814 + _ACTIVITYEPOCH._serialized_start = 816 + _ACTIVITYEPOCH._serialized_end = 908 + _CLIENTEPOCH._serialized_start = 910 + _CLIENTEPOCH._serialized_end = 940 + _PRODUCER._serialized_start = 942 + _PRODUCER._serialized_end = 1034 + _STREAMWRITER._serialized_start = 1037 + _STREAMWRITER._serialized_end = 1223 + _APPENDRECORD._serialized_start = 1225 + _APPENDRECORD._serialized_end = 1286 + _STREAMSUPERSEDED._serialized_start = 1288 + _STREAMSUPERSEDED._serialized_end = 1360 + _STREAMDATA._serialized_start = 1362 + _STREAMDATA._serialized_end = 1473 + _STREAMRECORD._serialized_start = 1476 + _STREAMRECORD._serialized_end = 1630 + _STREAMERRORDETAILS._serialized_start = 1632 + _STREAMERRORDETAILS._serialized_end = 1710 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/api/stream/v1/message_pb2.pyi b/temporalio/api/stream/v1/message_pb2.pyi new file mode 100644 index 0000000000..a8dc53d3ec --- /dev/null +++ b/temporalio/api/stream/v1/message_pb2.pyi @@ -0,0 +1,537 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import collections.abc +import sys + +import google.protobuf.descriptor +import google.protobuf.internal.containers +import google.protobuf.message +import google.protobuf.timestamp_pb2 + +import temporalio.api.common.v1.message_pb2 +import temporalio.api.enums.v1.stream_pb2 + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class StreamRef(google.protobuf.message.Message): + """Experimental. Might change in the future. + Identifies one Stream. The id is opaque to callers. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + ID_FIELD_NUMBER: builtins.int + OWNER_FIELD_NUMBER: builtins.int + BACKEND_FIELD_NUMBER: builtins.int + id: builtins.str + @property + def owner(self) -> temporalio.api.common.v1.message_pb2.Execution: + """Unset for a standalone Stream. For a Stream bound to a Workflow, business_id is the Workflow + ID and run_id is the first execution run id of its run chain. For a Standalone Activity, + business_id is the activity ID and run_id is its run id. A create and a close of a + Workflow-owned Stream must carry its owner. + """ + backend: builtins.str + """The name of the backend that holds the Stream, as the Worker and client config register + it, so every process routes a ref it receives. Empty is the default backend. The Store + never reads it. + """ + def __init__( + self, + *, + id: builtins.str = ..., + owner: temporalio.api.common.v1.message_pb2.Execution | None = ..., + backend: builtins.str = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["owner", b"owner"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "backend", b"backend", "id", b"id", "owner", b"owner" + ], + ) -> None: ... + +global___StreamRef = StreamRef + +class ProducerEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + The ordered attempt of a Producer. A higher epoch supersedes the Records of a lower epoch of the + same Producer. The case is the kind of the Producer. It decides how the Store orders the epochs, + and which earlier Records a higher epoch withdraws. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WORKFLOW_FIELD_NUMBER: builtins.int + LOCAL_ACTIVITY_FIELD_NUMBER: builtins.int + ACTIVITY_FIELD_NUMBER: builtins.int + STANDALONE_ACTIVITY_FIELD_NUMBER: builtins.int + CLIENT_FIELD_NUMBER: builtins.int + @property + def workflow(self) -> global___WorkflowTaskEpoch: + """A Workflow run, which writes in its Workflow Task. A higher epoch withdraws the Records + of the run's Workflow Tasks that never completed, also those of their Local Activities. + """ + @property + def local_activity(self) -> global___LocalActivityEpoch: + """A Local Activity. A higher epoch withdraws the Producer's lower epochs.""" + @property + def activity(self) -> global___ActivityEpoch: + """An Activity of a Workflow. A higher epoch withdraws the Producer's lower epochs.""" + @property + def standalone_activity(self) -> global___ActivityEpoch: + """A Standalone Activity. A higher epoch withdraws the Producer's lower epochs.""" + @property + def client(self) -> global___ClientEpoch: + """A client. A higher epoch only fences the lower ones, since a restarted client's earlier + Records are real messages, not the Records of a failed attempt. + """ + def __init__( + self, + *, + workflow: global___WorkflowTaskEpoch | None = ..., + local_activity: global___LocalActivityEpoch | None = ..., + activity: global___ActivityEpoch | None = ..., + standalone_activity: global___ActivityEpoch | None = ..., + client: global___ClientEpoch | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "activity", + b"activity", + "client", + b"client", + "kind", + b"kind", + "local_activity", + b"local_activity", + "standalone_activity", + b"standalone_activity", + "workflow", + b"workflow", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "activity", + b"activity", + "client", + b"client", + "kind", + b"kind", + "local_activity", + b"local_activity", + "standalone_activity", + b"standalone_activity", + "workflow", + b"workflow", + ], + ) -> None: ... + def WhichOneof( + self, oneof_group: typing_extensions.Literal["kind", b"kind"] + ) -> ( + typing_extensions.Literal[ + "workflow", "local_activity", "activity", "standalone_activity", "client" + ] + | None + ): ... + +global___ProducerEpoch = ProducerEpoch + +class WorkflowTaskEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + One Workflow Task attempt of a run. It orders by started_event_id, then attempt, then + started_time. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STARTED_EVENT_ID_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + STARTED_TIME_FIELD_NUMBER: builtins.int + started_event_id: builtins.int + attempt: builtins.int + @property + def started_time(self) -> google.protobuf.timestamp_pb2.Timestamp: + """Tells a reused started event id apart, for example after a dropped speculative task.""" + def __init__( + self, + *, + started_event_id: builtins.int = ..., + attempt: builtins.int = ..., + started_time: google.protobuf.timestamp_pb2.Timestamp | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["started_time", b"started_time"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "attempt", + b"attempt", + "started_event_id", + b"started_event_id", + "started_time", + b"started_time", + ], + ) -> None: ... + +global___WorkflowTaskEpoch = WorkflowTaskEpoch + +class LocalActivityEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + The attempt of a Local Activity. It orders by task, then attempt, so a Local Activity of a + later Workflow Task supersedes one of an earlier task. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + TASK_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + @property + def task(self) -> global___WorkflowTaskEpoch: + """The Workflow Task attempt that dispatched the Local Activity.""" + attempt: builtins.int + """The Local Activity's own attempt in that Workflow Task.""" + def __init__( + self, + *, + task: global___WorkflowTaskEpoch | None = ..., + attempt: builtins.int = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["task", b"task"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal["attempt", b"attempt", "task", b"task"], + ) -> None: ... + +global___LocalActivityEpoch = LocalActivityEpoch + +class ActivityEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + The attempt of an Activity. It orders by attempt_scheduled_time, then attempt. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + ATTEMPT_SCHEDULED_TIME_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + @property + def attempt_scheduled_time(self) -> google.protobuf.timestamp_pb2.Timestamp: + """When the server scheduled this attempt. An Activity reset starts the attempt count again, + so only this time orders the attempts across a reset. + """ + attempt: builtins.int + def __init__( + self, + *, + attempt_scheduled_time: google.protobuf.timestamp_pb2.Timestamp | None = ..., + attempt: builtins.int = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "attempt_scheduled_time", b"attempt_scheduled_time" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "attempt", b"attempt", "attempt_scheduled_time", b"attempt_scheduled_time" + ], + ) -> None: ... + +global___ActivityEpoch = ActivityEpoch + +class ClientEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + The attempt of a client Producer, which the client picks. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + ATTEMPT_FIELD_NUMBER: builtins.int + attempt: builtins.int + def __init__( + self, + *, + attempt: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["attempt", b"attempt"] + ) -> None: ... + +global___ClientEpoch = ClientEpoch + +class Producer(google.protobuf.message.Message): + """Experimental. Might change in the future. + The code that wrote a batch. Its id stays the same across attempts, and its epoch orders them. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + ID_FIELD_NUMBER: builtins.int + EPOCH_FIELD_NUMBER: builtins.int + RUN_ID_FIELD_NUMBER: builtins.int + id: builtins.str + """Stable across attempts, such as `wf:` for the Workflow Producer of a run.""" + @property + def epoch(self) -> global___ProducerEpoch: + """Required. Its case gives the kind of the Producer.""" + run_id: builtins.str + """The run of the Producer, so nobody parses the id. Empty for a client.""" + def __init__( + self, + *, + id: builtins.str = ..., + epoch: global___ProducerEpoch | None = ..., + run_id: builtins.str = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["epoch", b"epoch"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "epoch", b"epoch", "id", b"id", "run_id", b"run_id" + ], + ) -> None: ... + +global___Producer = Producer + +class StreamWriter(google.protobuf.message.Message): + """Experimental. Might change in the future. + The Producer of a write, and which earlier Records the write withdraws. An append, a close and + a writer carry the same fields, so the Store orders and withdraws each write the same way. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + PRODUCER_FIELD_NUMBER: builtins.int + PREVIOUS_COMPLETED_TASK_FIELD_NUMBER: builtins.int + FAILED_TASKS_FIELD_NUMBER: builtins.int + WITHDRAW_RUN_FIELD_NUMBER: builtins.int + @property + def producer(self) -> global___Producer: + """Required on a write. Its epoch's case picks the withdrawal rule.""" + @property + def previous_completed_task(self) -> global___WorkflowTaskEpoch: + """Workflow and Local Activity only. The run's last completed Workflow Task, from a Local + Activity's start. Its started_event_id is 0 before the first. The write withdraws the run's + Local Activity Records whose Workflow Task never completed. + """ + @property + def failed_tasks( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.int]: + """Workflow and Local Activity only. The started event ids of the run's failed or timed out + Workflow Tasks. The write withdraws the run's Local Activity Records of these tasks. + """ + withdraw_run: builtins.str + """Retry run only. The previous run, on the retry run's first write. Every Record whose + Producer carries it is withdrawn. + """ + def __init__( + self, + *, + producer: global___Producer | None = ..., + previous_completed_task: global___WorkflowTaskEpoch | None = ..., + failed_tasks: collections.abc.Iterable[builtins.int] | None = ..., + withdraw_run: builtins.str = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "previous_completed_task", + b"previous_completed_task", + "producer", + b"producer", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "failed_tasks", + b"failed_tasks", + "previous_completed_task", + b"previous_completed_task", + "producer", + b"producer", + "withdraw_run", + b"withdraw_run", + ], + ) -> None: ... + +global___StreamWriter = StreamWriter + +class AppendRecord(google.protobuf.message.Message): + """Experimental. Might change in the future. + One Record of a batch to append. The Store adds the batch's Producer and an offset. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + BODY_FIELD_NUMBER: builtins.int + @property + def body(self) -> temporalio.api.common.v1.message_pb2.Payload: ... + def __init__( + self, + *, + body: temporalio.api.common.v1.message_pb2.Payload | None = ..., + ) -> 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"] + ) -> None: ... + +global___AppendRecord = AppendRecord + +class StreamSuperseded(google.protobuf.message.Message): + """Experimental. Might change in the future. + Which earlier Records a retried Producer withdraws. The Store keeps these facts for the first + batch of a higher epoch, or for a retry run's first write, and a read builds a Superseded item + from them. A reader drops the earlier Records that the facts match. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_FIELD_NUMBER: builtins.int + @property + def writer(self) -> global___StreamWriter: + """The writer of the batch that withdraws. Its Producer's epoch is empty when only + withdraw_run applies. + """ + def __init__( + self, + *, + writer: global___StreamWriter | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["writer", b"writer"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing_extensions.Literal["writer", b"writer"] + ) -> None: ... + +global___StreamSuperseded = StreamSuperseded + +class StreamData(google.protobuf.message.Message): + """Experimental. Might change in the future. + One Record that a Producer appended. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + PRODUCER_FIELD_NUMBER: builtins.int + BODY_FIELD_NUMBER: builtins.int + @property + def producer(self) -> global___Producer: ... + @property + def body(self) -> temporalio.api.common.v1.message_pb2.Payload: ... + def __init__( + self, + *, + producer: global___Producer | None = ..., + body: temporalio.api.common.v1.message_pb2.Payload | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal["body", b"body", "producer", b"producer"], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal["body", b"body", "producer", b"producer"], + ) -> None: ... + +global___StreamData = StreamData + +class StreamRecord(google.protobuf.message.Message): + """Experimental. Might change in the future. + An offset is ordered and unique in one Stream, and it is sparse. Offsets have gaps, so they + don't count Records. Zero is never an offset. A read gives its Stream, so the read checks the + offset against that Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + OFFSET_FIELD_NUMBER: builtins.int + DATA_FIELD_NUMBER: builtins.int + SUPERSEDED_FIELD_NUMBER: builtins.int + offset: builtins.int + @property + def data(self) -> global___StreamData: ... + @property + def superseded(self) -> global___StreamSuperseded: + """The Store keeps only the withdrawal facts of the first batch of a higher epoch. A read + builds this item from them. It withdraws the earlier Records that the facts match, and a + reader drops them. + """ + def __init__( + self, + *, + offset: builtins.int = ..., + data: global___StreamData | None = ..., + superseded: global___StreamSuperseded | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "data", b"data", "item", b"item", "superseded", b"superseded" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "data", + b"data", + "item", + b"item", + "offset", + b"offset", + "superseded", + b"superseded", + ], + ) -> None: ... + def WhichOneof( + self, oneof_group: typing_extensions.Literal["item", b"item"] + ) -> typing_extensions.Literal["data", "superseded"] | None: ... + +global___StreamRecord = StreamRecord + +class StreamErrorDetails(google.protobuf.message.Message): + """Experimental. Might change in the future. + The details of a failed Stream call, in the gRPC status details and in the details of the + application failure of a Nexus operation. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + REASON_FIELD_NUMBER: builtins.int + reason: temporalio.api.enums.v1.stream_pb2.StreamErrorReason.ValueType + def __init__( + self, + *, + reason: temporalio.api.enums.v1.stream_pb2.StreamErrorReason.ValueType = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["reason", b"reason"] + ) -> None: ... + +global___StreamErrorDetails = StreamErrorDetails diff --git a/temporalio/api/streamservice/__init__.py b/temporalio/api/streamservice/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/temporalio/api/streamservice/v1/__init__.py b/temporalio/api/streamservice/v1/__init__.py new file mode 100644 index 0000000000..40801cf9a5 --- /dev/null +++ b/temporalio/api/streamservice/v1/__init__.py @@ -0,0 +1,25 @@ +from .request_response_pb2 import ( + AppendRecordsRequest, + AppendRecordsResponse, + CloseStreamRequest, + CloseStreamResponse, + CreateStreamRequest, + CreateStreamResponse, + DescribeStreamRequest, + DescribeStreamResponse, + ReadRecordsRequest, + ReadRecordsResponse, +) + +__all__ = [ + "AppendRecordsRequest", + "AppendRecordsResponse", + "CloseStreamRequest", + "CloseStreamResponse", + "CreateStreamRequest", + "CreateStreamResponse", + "DescribeStreamRequest", + "DescribeStreamResponse", + "ReadRecordsRequest", + "ReadRecordsResponse", +] diff --git a/temporalio/api/streamservice/v1/request_response_pb2.py b/temporalio/api/streamservice/v1/request_response_pb2.py new file mode 100644 index 0000000000..177dfb34e3 --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2.py @@ -0,0 +1,174 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/api/streamservice/v1/request_response.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 + +# @@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.enums.v1 import ( + stream_pb2 as temporal_dot_api_dot_enums_dot_v1_dot_stream__pb2, +) +from temporalio.api.stream.v1 import ( + message_pb2 as temporal_dot_api_dot_stream_dot_v1_dot_message__pb2, +) + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n4temporal/api/streamservice/v1/request_response.proto\x12\x1dtemporal.api.streamservice.v1\x1a\x1egoogle/protobuf/duration.proto\x1a"temporal/api/enums/v1/stream.proto\x1a$temporal/api/stream/v1/message.proto"\xdf\x02\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x35\n\nstream_ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x14\n\x0cwithdraw_run\x18\x04 \x01(\t\x12/\n\x05owner\x18\x05 \x01(\x0b\x32 .temporal.api.stream.v1.Producer\x12\x16\n\x0e\x63lose_with_run\x18\x06 \x01(\x08\x12\x12\n\ntask_token\x18\x07 \x01(\x0c\x12I\n\x12id_conflict_policy\x18\x08 \x01(\x0e\x32-.temporal.api.enums.v1.StreamIdConflictPolicy\x12\x12\n\nrequest_id\x18\t \x01(\t"W\n\x14\x43reateStreamResponse\x12.\n\x03ref\x18\x01 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08"\x86\x02\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x34\n\x06writer\x18\x03 \x01(\x0b\x32$.temporal.api.stream.v1.StreamWriter\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12\x35\n\x07records\x18\x05 \x03(\x0b\x32$.temporal.api.stream.v1.AppendRecord\x12\x18\n\x10\x64\x65\x66\x65r_withdrawal\x18\x06 \x01(\x08\x12\x12\n\ntask_token\x18\x07 \x01(\x0c"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\xa1\x01\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x34\n\x06writer\x18\x03 \x01(\x0b\x32$.temporal.api.stream.v1.StreamWriter\x12\x12\n\ntask_token\x18\x04 \x01(\x0c"\x15\n\x13\x43loseStreamResponse"\xec\x02\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x03\x12\x35\n\x05start\x18\x04 \x01(\x0e\x32&.temporal.api.enums.v1.StreamReadStart\x12\x13\n\x0bmax_records\x18\x05 \x01(\x05\x12\'\n\x04wait\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration\x12.\n\x0b\x63reate_wait\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x11\n\tto_offset\x18\x08 \x01(\x03\x12\x11\n\twake_only\x18\t \x01(\x08\x12\x0c\n\x04last\x18\n \x01(\x05\x12\x11\n\tmax_bytes\x18\x0b \x01(\x03\x12\x12\n\nfloor_from\x18\x0c \x01(\x03"\x92\x01\n\x13ReadRecordsResponse\x12\x35\n\x07records\x18\x01 \x03(\x0b\x32$.temporal.api.stream.v1.StreamRecord\x12\x13\n\x0bnext_offset\x18\x02 \x01(\x03\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08\x12!\n\x19\x63ontinue_as_new_suggested\x18\x04 \x01(\x08"Z\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef"\x90\x01\n\x16\x44\x65scribeStreamResponse\x12\x33\n\x06status\x18\x01 \x01(\x0e\x32#.temporal.api.enums.v1.StreamStatus\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.DurationB\xb4\x01\n io.temporal.api.streamservice.v1B\x14RequestResponseProtoP\x01Z1go.temporal.io/api/streamservice/v1;streamservice\xaa\x02\x1fTemporalio.Api.StreamService.V1\xea\x02"Temporalio::Api::StreamService::V1b\x06proto3' +) + + +_CREATESTREAMREQUEST = DESCRIPTOR.message_types_by_name["CreateStreamRequest"] +_CREATESTREAMRESPONSE = DESCRIPTOR.message_types_by_name["CreateStreamResponse"] +_APPENDRECORDSREQUEST = DESCRIPTOR.message_types_by_name["AppendRecordsRequest"] +_APPENDRECORDSRESPONSE = DESCRIPTOR.message_types_by_name["AppendRecordsResponse"] +_CLOSESTREAMREQUEST = DESCRIPTOR.message_types_by_name["CloseStreamRequest"] +_CLOSESTREAMRESPONSE = DESCRIPTOR.message_types_by_name["CloseStreamResponse"] +_READRECORDSREQUEST = DESCRIPTOR.message_types_by_name["ReadRecordsRequest"] +_READRECORDSRESPONSE = DESCRIPTOR.message_types_by_name["ReadRecordsResponse"] +_DESCRIBESTREAMREQUEST = DESCRIPTOR.message_types_by_name["DescribeStreamRequest"] +_DESCRIBESTREAMRESPONSE = DESCRIPTOR.message_types_by_name["DescribeStreamResponse"] +CreateStreamRequest = _reflection.GeneratedProtocolMessageType( + "CreateStreamRequest", + (_message.Message,), + { + "DESCRIPTOR": _CREATESTREAMREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.CreateStreamRequest) + }, +) +_sym_db.RegisterMessage(CreateStreamRequest) + +CreateStreamResponse = _reflection.GeneratedProtocolMessageType( + "CreateStreamResponse", + (_message.Message,), + { + "DESCRIPTOR": _CREATESTREAMRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.CreateStreamResponse) + }, +) +_sym_db.RegisterMessage(CreateStreamResponse) + +AppendRecordsRequest = _reflection.GeneratedProtocolMessageType( + "AppendRecordsRequest", + (_message.Message,), + { + "DESCRIPTOR": _APPENDRECORDSREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.AppendRecordsRequest) + }, +) +_sym_db.RegisterMessage(AppendRecordsRequest) + +AppendRecordsResponse = _reflection.GeneratedProtocolMessageType( + "AppendRecordsResponse", + (_message.Message,), + { + "DESCRIPTOR": _APPENDRECORDSRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.AppendRecordsResponse) + }, +) +_sym_db.RegisterMessage(AppendRecordsResponse) + +CloseStreamRequest = _reflection.GeneratedProtocolMessageType( + "CloseStreamRequest", + (_message.Message,), + { + "DESCRIPTOR": _CLOSESTREAMREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.CloseStreamRequest) + }, +) +_sym_db.RegisterMessage(CloseStreamRequest) + +CloseStreamResponse = _reflection.GeneratedProtocolMessageType( + "CloseStreamResponse", + (_message.Message,), + { + "DESCRIPTOR": _CLOSESTREAMRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.CloseStreamResponse) + }, +) +_sym_db.RegisterMessage(CloseStreamResponse) + +ReadRecordsRequest = _reflection.GeneratedProtocolMessageType( + "ReadRecordsRequest", + (_message.Message,), + { + "DESCRIPTOR": _READRECORDSREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.ReadRecordsRequest) + }, +) +_sym_db.RegisterMessage(ReadRecordsRequest) + +ReadRecordsResponse = _reflection.GeneratedProtocolMessageType( + "ReadRecordsResponse", + (_message.Message,), + { + "DESCRIPTOR": _READRECORDSRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.ReadRecordsResponse) + }, +) +_sym_db.RegisterMessage(ReadRecordsResponse) + +DescribeStreamRequest = _reflection.GeneratedProtocolMessageType( + "DescribeStreamRequest", + (_message.Message,), + { + "DESCRIPTOR": _DESCRIBESTREAMREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.DescribeStreamRequest) + }, +) +_sym_db.RegisterMessage(DescribeStreamRequest) + +DescribeStreamResponse = _reflection.GeneratedProtocolMessageType( + "DescribeStreamResponse", + (_message.Message,), + { + "DESCRIPTOR": _DESCRIBESTREAMRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.DescribeStreamResponse) + }, +) +_sym_db.RegisterMessage(DescribeStreamResponse) + +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = b'\n io.temporal.api.streamservice.v1B\024RequestResponseProtoP\001Z1go.temporal.io/api/streamservice/v1;streamservice\252\002\037Temporalio.Api.StreamService.V1\352\002"Temporalio::Api::StreamService::V1' + _CREATESTREAMREQUEST._serialized_start = 194 + _CREATESTREAMREQUEST._serialized_end = 545 + _CREATESTREAMRESPONSE._serialized_start = 547 + _CREATESTREAMRESPONSE._serialized_end = 634 + _APPENDRECORDSREQUEST._serialized_start = 637 + _APPENDRECORDSREQUEST._serialized_end = 899 + _APPENDRECORDSRESPONSE._serialized_start = 901 + _APPENDRECORDSRESPONSE._serialized_end = 967 + _CLOSESTREAMREQUEST._serialized_start = 970 + _CLOSESTREAMREQUEST._serialized_end = 1131 + _CLOSESTREAMRESPONSE._serialized_start = 1133 + _CLOSESTREAMRESPONSE._serialized_end = 1154 + _READRECORDSREQUEST._serialized_start = 1157 + _READRECORDSREQUEST._serialized_end = 1521 + _READRECORDSRESPONSE._serialized_start = 1524 + _READRECORDSRESPONSE._serialized_end = 1670 + _DESCRIBESTREAMREQUEST._serialized_start = 1672 + _DESCRIBESTREAMREQUEST._serialized_end = 1762 + _DESCRIBESTREAMRESPONSE._serialized_start = 1765 + _DESCRIBESTREAMRESPONSE._serialized_end = 1909 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/api/streamservice/v1/request_response_pb2.pyi b/temporalio/api/streamservice/v1/request_response_pb2.pyi new file mode 100644 index 0000000000..f5becf5a53 --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2.pyi @@ -0,0 +1,570 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import collections.abc +import sys + +import google.protobuf.descriptor +import google.protobuf.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.message + +import temporalio.api.enums.v1.stream_pb2 +import temporalio.api.stream.v1.message_pb2 + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class CreateStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Creates the Stream. When the Stream exists, FAIL fails with ALREADY_EXISTS, and USE_EXISTING + gets it with created false and its own Retention. A retry of the create that made the Stream, + with the same request_id, gets created true. Each policy fails with CLOSED on a closed Stream, + EXPIRED on an expired one, and NOT_OWNER on one with another owner. + (-- api-linter: core::0133=disabled + aip.dev/not-precedent: The request identifies a Stream by namespace and ref. --) + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + STREAM_REF_FIELD_NUMBER: builtins.int + RETENTION_FIELD_NUMBER: builtins.int + WITHDRAW_RUN_FIELD_NUMBER: builtins.int + OWNER_FIELD_NUMBER: builtins.int + CLOSE_WITH_RUN_FIELD_NUMBER: builtins.int + TASK_TOKEN_FIELD_NUMBER: builtins.int + ID_CONFLICT_POLICY_FIELD_NUMBER: builtins.int + REQUEST_ID_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def stream_ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """The ref of the Stream to create. Its id is the Stream id, and is never empty. A bound + Stream's ref has the owner's run chain as its owner. + """ + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: + """The age after which a Record is dropped. Empty is the Store default.""" + withdraw_run: builtins.str + """Workflow only. The previous run on a retry run's first create, to withdraw its Records.""" + @property + def owner(self) -> temporalio.api.stream.v1.message_pb2.Producer: + """The Workflow run that owns a bound Stream, with its run_id and a Workflow epoch. It needs a + ref whose owner is the run's chain. Unset for a standalone Stream. + """ + close_with_run: builtins.bool + """Only read by the server. When set, the server closes the Stream when the owner's run chain + ends, also on a terminate or a timeout. It needs owner, and a ref with a Workflow owner. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: "with" ties the close to the owner's run chain. --) + """ + task_token: builtins.bytes + """Only read by the server, for a Stream bound to a Standalone Activity. The server checks that + the token is the Activity's current attempt, so a caller can't bind a Stream to a Standalone + Activity that it isn't. The server knows the run of a Workflow's call, and a standalone + Stream has no owner, so neither needs a token. + """ + id_conflict_policy: ( + temporalio.api.enums.v1.stream_pb2.StreamIdConflictPolicy.ValueType + ) + """What the create does when the Stream exists. Unspecified is FAIL.""" + request_id: builtins.str + """Unique for each create call, and the same on its retries. The Store keeps the request_id of + the create that made the Stream, so a retry after a lost answer doesn't get ALREADY_EXISTS. + """ + def __init__( + self, + *, + namespace: builtins.str = ..., + stream_ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + retention: google.protobuf.duration_pb2.Duration | None = ..., + withdraw_run: builtins.str = ..., + owner: temporalio.api.stream.v1.message_pb2.Producer | None = ..., + close_with_run: builtins.bool = ..., + task_token: builtins.bytes = ..., + id_conflict_policy: temporalio.api.enums.v1.stream_pb2.StreamIdConflictPolicy.ValueType = ..., + request_id: builtins.str = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "owner", b"owner", "retention", b"retention", "stream_ref", b"stream_ref" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "close_with_run", + b"close_with_run", + "id_conflict_policy", + b"id_conflict_policy", + "namespace", + b"namespace", + "owner", + b"owner", + "request_id", + b"request_id", + "retention", + b"retention", + "stream_ref", + b"stream_ref", + "task_token", + b"task_token", + "withdraw_run", + b"withdraw_run", + ], + ) -> None: ... + +global___CreateStreamRequest = CreateStreamRequest + +class CreateStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + The Stream that a create made or got. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + REF_FIELD_NUMBER: builtins.int + CREATED_FIELD_NUMBER: builtins.int + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """The ref, with the owner of a bound Stream.""" + created: builtins.bool + """False when USE_EXISTING got a Stream that existed.""" + def __init__( + self, + *, + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + created: builtins.bool = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["ref", b"ref"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal["created", b"created", "ref", b"ref"], + ) -> None: ... + +global___CreateStreamResponse = CreateStreamResponse + +class AppendRecordsRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Appends one batch of Records as one Producer. The Store writes a resent batch, with the same + Producer, epoch and sequence, only once. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + WRITER_FIELD_NUMBER: builtins.int + SEQUENCE_FIELD_NUMBER: builtins.int + RECORDS_FIELD_NUMBER: builtins.int + DEFER_WITHDRAWAL_FIELD_NUMBER: builtins.int + TASK_TOKEN_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + @property + def writer(self) -> temporalio.api.stream.v1.message_pb2.StreamWriter: + """Required outside a Workflow. In a Workflow the transport sets it.""" + sequence: builtins.int + """The batch number in this epoch. With the Producer and epoch it is the dedupe key. Zero takes + the next number of the epoch. Such a batch has no dedupe, so a retry writes it again. + """ + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ]: ... + defer_withdrawal: builtins.bool + """Only read by the server, on a Workflow Producer's batch. The batch withdraws no Workflow + Task and keeps no task facts, but the writer's withdraw_run still applies. + """ + task_token: builtins.bytes + """Only read by the server. The server checks the Producer against the Activity's current + attempt, so nobody else writes as that Activity. + """ + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + writer: temporalio.api.stream.v1.message_pb2.StreamWriter | None = ..., + sequence: builtins.int = ..., + records: collections.abc.Iterable[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ] + | None = ..., + defer_withdrawal: builtins.bool = ..., + task_token: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["ref", b"ref", "writer", b"writer"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "defer_withdrawal", + b"defer_withdrawal", + "namespace", + b"namespace", + "records", + b"records", + "ref", + b"ref", + "sequence", + b"sequence", + "task_token", + b"task_token", + "writer", + b"writer", + ], + ) -> None: ... + +global___AppendRecordsRequest = AppendRecordsRequest + +class AppendRecordsResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + Where the batch's Records are in the Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FIRST_OFFSET_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + first_offset: builtins.int + last_offset: builtins.int + def __init__( + self, + *, + first_offset: builtins.int = ..., + last_offset: builtins.int = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "first_offset", b"first_offset", "last_offset", b"last_offset" + ], + ) -> None: ... + +global___AppendRecordsResponse = AppendRecordsResponse + +class CloseStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Closes a Stream. Readers drain what is left, then end, and later appends fail with CLOSED. A + close writes no Record and carries no value, so code that has a last value publishes it first. + Only the owner's run chain closes a bound Stream. Only a Workflow or a client closes a + standalone Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + WRITER_FIELD_NUMBER: builtins.int + TASK_TOKEN_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + @property + def writer(self) -> temporalio.api.stream.v1.message_pb2.StreamWriter: + """The writer that closes the Stream. An SDK Store reads only its Producer, and the server + also reads its withdrawal fields. + """ + task_token: builtins.bytes + """Only read by the server. It checks the token as on the append.""" + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + writer: temporalio.api.stream.v1.message_pb2.StreamWriter | None = ..., + task_token: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["ref", b"ref", "writer", b"writer"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "namespace", + b"namespace", + "ref", + b"ref", + "task_token", + b"task_token", + "writer", + b"writer", + ], + ) -> None: ... + +global___CloseStreamRequest = CloseStreamRequest + +class CloseStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + A close answers when the Store has the close. It gives nothing back. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___CloseStreamResponse = CloseStreamResponse + +class ReadRecordsRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Reads one page of items from an offset. At the end of the Stream, the read waits for a new + Record, so readers long poll. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + FROM_OFFSET_FIELD_NUMBER: builtins.int + START_FIELD_NUMBER: builtins.int + MAX_RECORDS_FIELD_NUMBER: builtins.int + WAIT_FIELD_NUMBER: builtins.int + CREATE_WAIT_FIELD_NUMBER: builtins.int + TO_OFFSET_FIELD_NUMBER: builtins.int + WAKE_ONLY_FIELD_NUMBER: builtins.int + LAST_FIELD_NUMBER: builtins.int + MAX_BYTES_FIELD_NUMBER: builtins.int + FLOOR_FROM_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + from_offset: builtins.int + """Start at this offset. Zero uses start. It fails with OFFSET_TOO_HIGH if it is past the newest + offset + 1. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --) + """ + start: temporalio.api.enums.v1.stream_pb2.StreamReadStart.ValueType + """Where to start when from_offset is zero. Unspecified is the beginning.""" + max_records: builtins.int + """The most items in one page, Superseded items included. Zero is 1000. A Superseded item and + the Record after it stay in one page, so a page of one Record can hold two items. + """ + @property + def wait(self) -> google.protobuf.duration_pb2.Duration: + """How long the read waits for a new Record before it answers with an empty page. Empty answers + at once, and a wait longer than 60 s waits 60 s. + """ + @property + def create_wait(self) -> google.protobuf.duration_pb2.Duration: + """How long to wait for a Stream that nobody created yet. Empty fails at once with NOT_FOUND. + At most 60 s. + """ + to_offset: builtins.int + """The exclusive end of the read. The read scans no entry at or after it, so a replay reads the + same range. Zero means no end. Only the server serves it, and a Store refuses it. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --) + """ + wake_only: builtins.bool + """Park as a normal read, then answer with no Records and next_offset = from_offset, when a + Record at or after from_offset exists, when the Stream closes (done), or when wait ends. Its + wait may be days, since a parked read adds nothing to History while it waits. Only the + server serves it, and a Store refuses it. + """ + last: builtins.int + """Read the newest `last` Records, oldest first. It needs no from_offset and no start. Not with + wake_only, and not in Workflow code, where replay needs a fixed range. + """ + max_bytes: builtins.int + """The most Record bytes in one page, so a page fits the caller's limit, such as a History + marker. A larger Record still comes alone, so the read always moves on. Zero is no cap. + """ + floor_from: builtins.int + """The lowest offset this run read on replay. A reset run copies the earlier run's reads into + its History, so its floor must cover them too, not only its first live page. Only read by + the server. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: The floor is the lowest from_offset of the run's reads. --) + """ + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + from_offset: builtins.int = ..., + start: temporalio.api.enums.v1.stream_pb2.StreamReadStart.ValueType = ..., + max_records: builtins.int = ..., + wait: google.protobuf.duration_pb2.Duration | None = ..., + create_wait: google.protobuf.duration_pb2.Duration | None = ..., + to_offset: builtins.int = ..., + wake_only: builtins.bool = ..., + last: builtins.int = ..., + max_bytes: builtins.int = ..., + floor_from: builtins.int = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "create_wait", b"create_wait", "ref", b"ref", "wait", b"wait" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "create_wait", + b"create_wait", + "floor_from", + b"floor_from", + "from_offset", + b"from_offset", + "last", + b"last", + "max_bytes", + b"max_bytes", + "max_records", + b"max_records", + "namespace", + b"namespace", + "ref", + b"ref", + "start", + b"start", + "to_offset", + b"to_offset", + "wait", + b"wait", + "wake_only", + b"wake_only", + ], + ) -> None: ... + +global___ReadRecordsRequest = ReadRecordsRequest + +class ReadRecordsResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + One page of a read. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RECORDS_FIELD_NUMBER: builtins.int + NEXT_OFFSET_FIELD_NUMBER: builtins.int + DONE_FIELD_NUMBER: builtins.int + CONTINUE_AS_NEW_SUGGESTED_FIELD_NUMBER: builtins.int + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.stream.v1.message_pb2.StreamRecord + ]: ... + next_offset: builtins.int + """The from_offset of the next read, which is the newest offset read + 1.""" + done: builtins.bool + """True when the Stream is closed and this read gave its last item. A read of a closed Stream + with no item left answers at once. + """ + continue_as_new_suggested: builtins.bool + """This reader's floor keeps more than half of the Stream's byte cap. The writer fails when the + cap fills, and only the reader's continue-as-new frees it. Only set by the server. + """ + def __init__( + self, + *, + records: collections.abc.Iterable[ + temporalio.api.stream.v1.message_pb2.StreamRecord + ] + | None = ..., + next_offset: builtins.int = ..., + done: builtins.bool = ..., + continue_as_new_suggested: builtins.bool = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "continue_as_new_suggested", + b"continue_as_new_suggested", + "done", + b"done", + "next_offset", + b"next_offset", + "records", + b"records", + ], + ) -> None: ... + +global___ReadRecordsResponse = ReadRecordsResponse + +class DescribeStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Gets the status, the newest offset and the Retention of a Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["ref", b"ref"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal["namespace", b"namespace", "ref", b"ref"], + ) -> None: ... + +global___DescribeStreamRequest = DescribeStreamRequest + +class DescribeStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + The state of a Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STATUS_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + RETENTION_FIELD_NUMBER: builtins.int + status: temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType + last_offset: builtins.int + """The newest offset, of a Record or a Superseded item. Zero when the Stream has none.""" + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: ... + def __init__( + self, + *, + status: temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType = ..., + last_offset: builtins.int = ..., + retention: google.protobuf.duration_pb2.Duration | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["retention", b"retention"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "last_offset", + b"last_offset", + "retention", + b"retention", + "status", + b"status", + ], + ) -> None: ... + +global___DescribeStreamResponse = DescribeStreamResponse diff --git a/temporalio/api/streamservice/v1/request_response_pb2_grpc.py b/temporalio/api/streamservice/v1/request_response_pb2_grpc.py new file mode 100644 index 0000000000..bf947056a2 --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2_grpc.py @@ -0,0 +1,4 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" + +import grpc diff --git a/temporalio/api/streamservice/v1/request_response_pb2_grpc.pyi b/temporalio/api/streamservice/v1/request_response_pb2_grpc.pyi new file mode 100644 index 0000000000..f3a5a087ef --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2_grpc.pyi @@ -0,0 +1,4 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index 48e4a9416b..464c895cb2 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.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "380962b3d93b99642e50b9febf72c071e97e8b32d86eae9cf4ef582bb960ae6e" +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 = "7.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" @@ -1493,6 +1597,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" @@ -2033,6 +2143,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" @@ -2382,6 +2525,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" @@ -2591,6 +2751,8 @@ dependencies = [ "temporalio-client", "temporalio-common", "temporalio-sdk-core", + "temporalio-streams", + "temporalio-streams-redis", "tokio", "tokio-rustls", "tokio-stream", @@ -2774,6 +2936,41 @@ dependencies = [ "zip", ] +[[package]] +name = "temporalio-streams" +version = "0.1.0" +dependencies = [ + "async-trait", + "parking_lot", + "prost", + "prost-wkt-types", + "sha2", + "temporalio-common", + "thiserror", + "tokio", + "tonic", +] + +[[package]] +name = "temporalio-streams-redis" +version = "0.1.0" +source = "git+https://github.com/temporal-community/sdk-streams?rev=dc93cba6bd49c89bb255a3aae9f4d970f4ffc997#dc93cba6bd49c89bb255a3aae9f4d970f4ffc997" +dependencies = [ + "async-trait", + "aws-lc-rs", + "crc16", + "parking_lot", + "prost", + "prost-wkt-types", + "redis", + "temporalio-common", + "temporalio-streams", + "thiserror", + "tokio", + "tonic-prost-build", + "tracing", +] + [[package]] name = "termtree" version = "0.5.1" @@ -3138,6 +3335,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" @@ -3620,6 +3823,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..a2aa91c3f7 100644 --- a/temporalio/bridge/Cargo.toml +++ b/temporalio/bridge/Cargo.toml @@ -29,14 +29,20 @@ default = ["tls-ring"] tls-ring = [ "temporalio-sdk-core/tls-ring", "temporalio-client/tls-ring", + "temporalio-streams-redis?/tls-ring", "tokio-rustls/ring", ] tls-aws-lc = [ "temporalio-sdk-core/tls-aws-lc", "temporalio-client/tls-aws-lc", + "temporalio-streams-redis?/tls-aws-lc", "tokio-rustls/aws_lc_rs", ] -fips = ["tls-aws-lc", "dep:aws-lc-rs"] +fips = ["tls-aws-lc", "dep:aws-lc-rs", "streams", "temporalio-streams-redis?/fips"] +# Off by default, so a source build doesn't compile the Stream runtime and its Redis client. +# The official builds turn it on. Cargo fetches the git dependency below for every build, with +# this feature off too. +streams = ["dep:temporalio-streams", "dep:temporalio-streams-redis"] [dependencies] anyhow = "1.0" @@ -84,6 +90,25 @@ tonic = "0.14" tracing = "0.1" url = "2.2" +# The Stream runtime and its memory Store, from the submodule. It runs every Store. +[dependencies.temporalio-streams] +path = "./sdk-core/crates/streams" +optional = true + +# The Redis Store, which lives in its own repo. +[dependencies.temporalio-streams-redis] +git = "https://github.com/temporal-community/sdk-streams" +rev = "dc93cba6bd49c89bb255a3aae9f4d970f4ffc997" +default-features = false +optional = true + +# The Redis crate takes temporalio-common and temporalio-streams from crates.io. This makes it use +# the submodule's crates, so the bridge sees one copy of each type. The versions must be semver +# compatible. +[patch.crates-io] +temporalio-common = { path = "./sdk-core/crates/common" } +temporalio-streams = { path = "./sdk-core/crates/streams" } + [profile.release] opt-level = 3 debug = false diff --git a/temporalio/bridge/proto/streams/__init__.py b/temporalio/bridge/proto/streams/__init__.py new file mode 100644 index 0000000000..57506046a0 --- /dev/null +++ b/temporalio/bridge/proto/streams/__init__.py @@ -0,0 +1,27 @@ +from .streams_pb2 import ( + CloseWriterRequest, + CloseWriterResponse, + CloseWritersRequest, + CloseWritersResponse, + FreeWriterRequest, + FreeWriterResponse, + MemoryStreamConfig, + OpenWriterRequest, + OpenWriterResponse, + WriteRecordsRequest, + WriteRecordsResponse, +) + +__all__ = [ + "CloseWriterRequest", + "CloseWriterResponse", + "CloseWritersRequest", + "CloseWritersResponse", + "FreeWriterRequest", + "FreeWriterResponse", + "MemoryStreamConfig", + "OpenWriterRequest", + "OpenWriterResponse", + "WriteRecordsRequest", + "WriteRecordsResponse", +] diff --git a/temporalio/bridge/proto/streams/streams_pb2.py b/temporalio/bridge/proto/streams/streams_pb2.py new file mode 100644 index 0000000000..78dfba638b --- /dev/null +++ b/temporalio/bridge/proto/streams/streams_pb2.py @@ -0,0 +1,187 @@ +# -*- 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 + +# @@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.stream.v1 import ( + message_pb2 as temporal_dot_api_dot_stream_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/stream/v1/message.proto"\x82\x01\n\x12MemoryStreamConfig\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nsafety_ttl\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x0f\n\x07\x62\x61\x63kend\x18\n \x01(\t"\xdb\x01\n\x11OpenWriterRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x34\n\x06writer\x18\x03 \x01(\x0b\x32$.temporal.api.stream.v1.StreamWriter\x12\x16\n\x0emax_batch_size\x18\x04 \x01(\x05\x12\x35\n\x12max_retry_duration\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration"\'\n\x12OpenWriterResponse\x12\x11\n\twriter_id\x18\x01 \x01(\x04"_\n\x13WriteRecordsRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04\x12\x35\n\x07records\x18\x02 \x03(\x0b\x32$.temporal.api.stream.v1.AppendRecord"A\n\x14WriteRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\'\n\x12\x43loseWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x43loseWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"&\n\x11\x46reeWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"\x14\n\x12\x46reeWriterResponse"\x15\n\x13\x43loseWritersRequest"&\n\x14\x43loseWritersResponse\x12\x0e\n\x06\x66\x61iled\x18\x01 \x01(\x05\x42-\xea\x02*Temporalio::Internal::Bridge::Api::Streamsb\x06proto3' +) + + +_MEMORYSTREAMCONFIG = DESCRIPTOR.message_types_by_name["MemoryStreamConfig"] +_OPENWRITERREQUEST = DESCRIPTOR.message_types_by_name["OpenWriterRequest"] +_OPENWRITERRESPONSE = DESCRIPTOR.message_types_by_name["OpenWriterResponse"] +_WRITERECORDSREQUEST = DESCRIPTOR.message_types_by_name["WriteRecordsRequest"] +_WRITERECORDSRESPONSE = DESCRIPTOR.message_types_by_name["WriteRecordsResponse"] +_CLOSEWRITERREQUEST = DESCRIPTOR.message_types_by_name["CloseWriterRequest"] +_CLOSEWRITERRESPONSE = DESCRIPTOR.message_types_by_name["CloseWriterResponse"] +_FREEWRITERREQUEST = DESCRIPTOR.message_types_by_name["FreeWriterRequest"] +_FREEWRITERRESPONSE = DESCRIPTOR.message_types_by_name["FreeWriterResponse"] +_CLOSEWRITERSREQUEST = DESCRIPTOR.message_types_by_name["CloseWritersRequest"] +_CLOSEWRITERSRESPONSE = DESCRIPTOR.message_types_by_name["CloseWritersResponse"] +MemoryStreamConfig = _reflection.GeneratedProtocolMessageType( + "MemoryStreamConfig", + (_message.Message,), + { + "DESCRIPTOR": _MEMORYSTREAMCONFIG, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.MemoryStreamConfig) + }, +) +_sym_db.RegisterMessage(MemoryStreamConfig) + +OpenWriterRequest = _reflection.GeneratedProtocolMessageType( + "OpenWriterRequest", + (_message.Message,), + { + "DESCRIPTOR": _OPENWRITERREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.OpenWriterRequest) + }, +) +_sym_db.RegisterMessage(OpenWriterRequest) + +OpenWriterResponse = _reflection.GeneratedProtocolMessageType( + "OpenWriterResponse", + (_message.Message,), + { + "DESCRIPTOR": _OPENWRITERRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.OpenWriterResponse) + }, +) +_sym_db.RegisterMessage(OpenWriterResponse) + +WriteRecordsRequest = _reflection.GeneratedProtocolMessageType( + "WriteRecordsRequest", + (_message.Message,), + { + "DESCRIPTOR": _WRITERECORDSREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.WriteRecordsRequest) + }, +) +_sym_db.RegisterMessage(WriteRecordsRequest) + +WriteRecordsResponse = _reflection.GeneratedProtocolMessageType( + "WriteRecordsResponse", + (_message.Message,), + { + "DESCRIPTOR": _WRITERECORDSRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.WriteRecordsResponse) + }, +) +_sym_db.RegisterMessage(WriteRecordsResponse) + +CloseWriterRequest = _reflection.GeneratedProtocolMessageType( + "CloseWriterRequest", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEWRITERREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.CloseWriterRequest) + }, +) +_sym_db.RegisterMessage(CloseWriterRequest) + +CloseWriterResponse = _reflection.GeneratedProtocolMessageType( + "CloseWriterResponse", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEWRITERRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.CloseWriterResponse) + }, +) +_sym_db.RegisterMessage(CloseWriterResponse) + +FreeWriterRequest = _reflection.GeneratedProtocolMessageType( + "FreeWriterRequest", + (_message.Message,), + { + "DESCRIPTOR": _FREEWRITERREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.FreeWriterRequest) + }, +) +_sym_db.RegisterMessage(FreeWriterRequest) + +FreeWriterResponse = _reflection.GeneratedProtocolMessageType( + "FreeWriterResponse", + (_message.Message,), + { + "DESCRIPTOR": _FREEWRITERRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.FreeWriterResponse) + }, +) +_sym_db.RegisterMessage(FreeWriterResponse) + +CloseWritersRequest = _reflection.GeneratedProtocolMessageType( + "CloseWritersRequest", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEWRITERSREQUEST, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.CloseWritersRequest) + }, +) +_sym_db.RegisterMessage(CloseWritersRequest) + +CloseWritersResponse = _reflection.GeneratedProtocolMessageType( + "CloseWritersResponse", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEWRITERSRESPONSE, + "__module__": "temporal.sdk.core.streams.streams_pb2", + # @@protoc_insertion_point(class_scope:coresdk.streams.CloseWritersResponse) + }, +) +_sym_db.RegisterMessage(CloseWritersResponse) + +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = ( + b"\352\002*Temporalio::Internal::Bridge::Api::Streams" + ) + _MEMORYSTREAMCONFIG._serialized_start = 131 + _MEMORYSTREAMCONFIG._serialized_end = 261 + _OPENWRITERREQUEST._serialized_start = 264 + _OPENWRITERREQUEST._serialized_end = 483 + _OPENWRITERRESPONSE._serialized_start = 485 + _OPENWRITERRESPONSE._serialized_end = 524 + _WRITERECORDSREQUEST._serialized_start = 526 + _WRITERECORDSREQUEST._serialized_end = 621 + _WRITERECORDSRESPONSE._serialized_start = 623 + _WRITERECORDSRESPONSE._serialized_end = 688 + _CLOSEWRITERREQUEST._serialized_start = 690 + _CLOSEWRITERREQUEST._serialized_end = 729 + _CLOSEWRITERRESPONSE._serialized_start = 731 + _CLOSEWRITERRESPONSE._serialized_end = 773 + _FREEWRITERREQUEST._serialized_start = 775 + _FREEWRITERREQUEST._serialized_end = 813 + _FREEWRITERRESPONSE._serialized_start = 815 + _FREEWRITERRESPONSE._serialized_end = 835 + _CLOSEWRITERSREQUEST._serialized_start = 837 + _CLOSEWRITERSREQUEST._serialized_end = 858 + _CLOSEWRITERSRESPONSE._serialized_start = 860 + _CLOSEWRITERSRESPONSE._serialized_end = 898 +# @@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..b9b9655523 --- /dev/null +++ b/temporalio/bridge/proto/streams/streams_pb2.pyi @@ -0,0 +1,319 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import collections.abc +import sys + +import google.protobuf.descriptor +import google.protobuf.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.message + +import temporalio.api.stream.v1.message_pb2 + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class MemoryStreamConfig(google.protobuf.message.Message): + """The messages of the calls that `temporalio-streams` adds to the five Stream service operations. + Lang sends them through the bridge's call on the runtime object, by the method name in each + comment. One generic call keeps Stream logic out of every bridge. + + The settings of the memory Stream runtime, for tests. Each field has the number of the same + field in the config of the other Stores, such as Redis, so a backend can send that config here + too. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RETENTION_FIELD_NUMBER: builtins.int + SAFETY_TTL_FIELD_NUMBER: builtins.int + BACKEND_FIELD_NUMBER: builtins.int + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: + """The default Retention of a new Stream. Empty is 7 days.""" + @property + def safety_ttl(self) -> google.protobuf.duration_pb2.Duration: + """How long an open Workflow-owned Stream stays after its last write. Empty is 90 days.""" + backend: builtins.str + """The backend name that lang registered this runtime under. The runtime sets it on each + StreamRef that it builds, such as the one in a create response. + """ + def __init__( + self, + *, + retention: google.protobuf.duration_pb2.Duration | None = ..., + safety_ttl: google.protobuf.duration_pb2.Duration | None = ..., + backend: builtins.str = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "retention", b"retention", "safety_ttl", b"safety_ttl" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "backend", + b"backend", + "retention", + b"retention", + "safety_ttl", + b"safety_ttl", + ], + ) -> None: ... + +global___MemoryStreamConfig = MemoryStreamConfig + +class OpenWriterRequest(google.protobuf.message.Message): + """Method `OpenWriter`, asynchronous. A writer for an Activity, a Local Activity or a client + Producer, which keeps its epoch and `sequence` order. A writer of a Local Activity, or of an + Activity attempt above 1, first sends an empty batch, so that the new epoch withdraws the earlier + one's Records. The opens of one Producer epoch on one Stream share one writer. Each open adds a + reference, and each `CloseWriter` or `FreeWriter` releases one. The batches of every writer in a + process take their `sequence` from one counter, so a writer that opens again after its close + continues above its earlier batches. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + WRITER_FIELD_NUMBER: builtins.int + MAX_BATCH_SIZE_FIELD_NUMBER: builtins.int + MAX_RETRY_DURATION_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + @property + def writer(self) -> temporalio.api.stream.v1.message_pb2.StreamWriter: + """Each batch sends it as AppendRecords does. For a Local Activity it has the run's task facts, + so the writes withdraw the run's orphaned Local Activity Records. Its withdraw_run must be + empty, since only a retry run's first append withdraws a run, and a writer sends many. + """ + max_batch_size: builtins.int + """A write sends at most this many Records in one batch. Zero, or a value above 1000, is 1000.""" + @property + def max_retry_duration(self) -> google.protobuf.duration_pb2.Duration: + """How long a write retries a Store failure. Empty is 10 s.""" + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + writer: temporalio.api.stream.v1.message_pb2.StreamWriter | None = ..., + max_batch_size: builtins.int = ..., + max_retry_duration: google.protobuf.duration_pb2.Duration | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "max_retry_duration", + b"max_retry_duration", + "ref", + b"ref", + "writer", + b"writer", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "max_batch_size", + b"max_batch_size", + "max_retry_duration", + b"max_retry_duration", + "namespace", + b"namespace", + "ref", + b"ref", + "writer", + b"writer", + ], + ) -> None: ... + +global___OpenWriterRequest = OpenWriterRequest + +class OpenWriterResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + writer_id: builtins.int + """The writer that the other writer calls name.""" + def __init__( + self, + *, + writer_id: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["writer_id", b"writer_id"] + ) -> None: ... + +global___OpenWriterResponse = OpenWriterResponse + +class WriteRecordsRequest(google.protobuf.message.Message): + """Method `WriteRecords`, asynchronous. Appends Records at once for an open writer and responds when + the Store has them. It uses the writer's Producer epoch and `sequence` order. After a Store + failure it resends the same batch, so that the Store writes it once. Records past the Store's + batch limit go in more batches, in order. An awaited publish of an Activity or a client uses this + call. A Store failure after the retries fails only that write, and the next write takes a higher + `sequence`. Any other failure fails the writer. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + RECORDS_FIELD_NUMBER: builtins.int + writer_id: builtins.int + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ]: ... + def __init__( + self, + *, + writer_id: builtins.int = ..., + records: collections.abc.Iterable[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ] + | None = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "records", b"records", "writer_id", b"writer_id" + ], + ) -> None: ... + +global___WriteRecordsRequest = WriteRecordsRequest + +class WriteRecordsResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FIRST_OFFSET_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + first_offset: builtins.int + """The offsets of the first and the last Record. Zero when the request has no Record.""" + last_offset: builtins.int + def __init__( + self, + *, + first_offset: builtins.int = ..., + last_offset: builtins.int = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "first_offset", b"first_offset", "last_offset", b"last_offset" + ], + ) -> None: ... + +global___WriteRecordsResponse = WriteRecordsResponse + +class CloseWriterRequest(google.protobuf.message.Message): + """Method `CloseWriter`, asynchronous. Releases one reference of the writer. The last reference + waits for a write in flight, then drops the writer. It doesn't close the Stream. It fails with + the writer's error if the writer failed. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + writer_id: builtins.int + def __init__( + self, + *, + writer_id: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["writer_id", b"writer_id"] + ) -> None: ... + +global___CloseWriterRequest = CloseWriterRequest + +class CloseWriterResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + LAST_OFFSET_FIELD_NUMBER: builtins.int + last_offset: builtins.int + """The offset of the newest Record the writer wrote. Zero if it wrote none.""" + def __init__( + self, + *, + last_offset: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["last_offset", b"last_offset"] + ) -> None: ... + +global___CloseWriterResponse = CloseWriterResponse + +class FreeWriterRequest(google.protobuf.message.Message): + """Method `FreeWriter`. Releases one reference of the writer with no wait, for a reference that + lang drops, such as from a finalizer. The bridge may also call it with no async runtime. A write + in flight still completes. The writer's error, if any, is lost. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + writer_id: builtins.int + def __init__( + self, + *, + writer_id: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["writer_id", b"writer_id"] + ) -> None: ... + +global___FreeWriterRequest = FreeWriterRequest + +class FreeWriterResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___FreeWriterResponse = FreeWriterResponse + +class CloseWritersRequest(google.protobuf.message.Message): + """Method `CloseWriters`, asynchronous. Drops every open writer, for example when the backend's + client or Worker shuts down. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___CloseWritersRequest = CloseWritersRequest + +class CloseWritersResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FAILED_FIELD_NUMBER: builtins.int + failed: builtins.int + """The writers that had failed.""" + def __init__( + self, + *, + failed: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["failed", b"failed"] + ) -> None: ... + +global___CloseWritersResponse = CloseWritersResponse diff --git a/temporalio/bridge/sdk-core b/temporalio/bridge/sdk-core index 4b3af8dba2..872c90fcc4 160000 --- a/temporalio/bridge/sdk-core +++ b/temporalio/bridge/sdk-core @@ -1 +1 @@ -Subproject commit 4b3af8dba206bdc0be1ec357084a766ac16ad165 +Subproject commit 872c90fcc4782380c661de2e763484f9f205fc8c diff --git a/temporalio/bridge/src/lib.rs b/temporalio/bridge/src/lib.rs index a4fda26db6..f69a261bb5 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; @@ -69,6 +70,12 @@ fn temporal_sdk_bridge(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_function(wrap_pyfunction!(new_worker, m)?)?; m.add_function(wrap_pyfunction!(new_replay_worker, m)?)?; + m.add("StreamCallError", py.get_type::())?; + #[cfg(feature = "streams")] + { + m.add_class::()?; + m.add_function(wrap_pyfunction!(new_stream_runtime, m)?)?; + } // envconfig let envconfig_module = PyModule::new(py, "envconfig")?; @@ -131,6 +138,16 @@ fn new_worker( worker::new_worker(runtime_ref, client, config) } +#[cfg(feature = "streams")] +#[pyfunction] +fn new_stream_runtime( + runtime_ref: &runtime::RuntimeRef, + memory: bool, + config: Vec, +) -> PyResult { + streams::new_stream_runtime(runtime_ref, memory, config) +} + #[pyfunction] fn new_replay_worker<'a>( py: Python<'a>, diff --git a/temporalio/bridge/src/streams.rs b/temporalio/bridge/src/streams.rs new file mode 100644 index 0000000000..8e37f71b5e --- /dev/null +++ b/temporalio/bridge/src/streams.rs @@ -0,0 +1,80 @@ +//! Bindings to the Stream runtime of the temporalio-streams crate, on the Redis Store of +//! temporalio-streams-redis or on the memory Store. A Stream backend keeps one runtime for each +//! Core runtime, and the clients of that Core runtime share it and its connections. Python calls +//! it by method name. + +use pyo3::exceptions::PyException; +#[cfg(feature = "streams")] +use pyo3::prelude::*; + +#[cfg(feature = "streams")] +use crate::runtime; + +// The exception args are the gRPC code, the message and the encoded details, as in `RPCError`. +pyo3::create_exception!(temporal_sdk_bridge, StreamCallError, PyException); + +/// The crate's runtime object. Python keeps one for each Core runtime, which its clients share. +#[cfg(feature = "streams")] +#[pyclass(frozen)] +pub struct StreamRuntimeRef { + streams: std::sync::Arc, + runtime: runtime::Runtime, +} + +#[cfg(feature = "streams")] +#[pymethods] +impl StreamRuntimeRef { + #[getter] + fn name(&self) -> String { + self.streams.name().to_string() + } + + fn call<'p>( + &self, + py: Python<'p>, + method: String, + input: Vec, + ) -> PyResult> { + self.runtime + .assert_same_process("call the Stream runtime")?; + let streams = self.streams.clone(); + self.runtime.future_into_py(py, async move { + streams.call(&method, &input).await.map_err(call_error) + }) + } + + /// Releases one handle of a writer without waiting. It needs no Tokio runtime, so a Python + /// finalizer can call it. + fn free_writer(&self, writer_id: u64) -> PyResult<()> { + self.streams.free_writer(writer_id).map_err(call_error) + } +} + +#[cfg(feature = "streams")] +fn call_error(err: temporalio_streams::StreamError) -> PyErr { + StreamCallError::new_err((err.code() as u32, err.message.clone(), err.details())) +} + +/// Builds the runtime on Redis, or on the memory Store with `memory`, from its encoded config. +#[cfg(feature = "streams")] +pub fn new_stream_runtime( + runtime_ref: &runtime::RuntimeRef, + memory: bool, + config: Vec, +) -> PyResult { + use pyo3::exceptions::PyValueError; + + let runtime = runtime_ref.runtime.clone(); + runtime.assert_same_process("create a Stream runtime")?; + // Building the Stream runtime can spawn Tokio tasks, which need an entered Tokio runtime. + let _guard = runtime.core.tokio_handle().enter(); + let meter = runtime.core.telemetry().get_temporal_metric_meter(); + let streams = if memory { + temporalio_streams::MemoryStreamRuntime::from_config_bytes(&config, meter) + .map_err(|err| PyValueError::new_err(err.to_string()))? + } else { + temporalio_streams_redis::RedisStreamRuntime::from_config_bytes(&config, meter) + .map_err(|err| PyValueError::new_err(err.to_string()))? + }; + Ok(StreamRuntimeRef { streams, runtime }) +} diff --git a/temporalio/bridge/streams.py b/temporalio/bridge/streams.py new file mode 100644 index 0000000000..4b4a7f59dd --- /dev/null +++ b/temporalio/bridge/streams.py @@ -0,0 +1,47 @@ +"""The Stream runtime that this bridge build links. (unstable) + +Nothing in this module should be considered stable. The API may change. +""" + +from __future__ import annotations + +from typing import Any + +import temporalio.bridge.runtime +import temporalio.bridge.temporal_sdk_bridge + +StreamCallError: type[Exception] = temporalio.bridge.temporal_sdk_bridge.StreamCallError + + +class StreamRuntime: + """The temporalio-streams runtime, on Redis or on the memory Store.""" + + def __init__( + self, runtime: temporalio.bridge.runtime.Runtime, memory: bool, config: bytes + ) -> None: + """Build the runtime on Redis, or on memory with ``memory``, from its encoded config. + + Raises: + ValueError: The bridge build doesn't link the crate, or the config is not valid. + """ + new = getattr(temporalio.bridge.temporal_sdk_bridge, "new_stream_runtime", None) + if new is None: + raise ValueError("This bridge build has no Stream runtime") + self._ref: Any = new(runtime._ref, memory, config) + + @property + def name(self) -> str: + """Name that the runtime reports.""" + return self._ref.name + + async def call(self, method: str, input: bytes) -> bytes: + """Call ``method`` on the runtime and await its output. + + Raises: + StreamCallError: The call failed, with the gRPC code, message and details as args. + """ + return await self._ref.call(method, input) + + def free_writer(self, writer_id: int) -> None: + """Release one handle of a writer without waiting, such as from a finalizer.""" + self._ref.free_writer(writer_id) diff --git a/tests/bridge/test_streams.py b/tests/bridge/test_streams.py new file mode 100644 index 0000000000..22fc8ab8be --- /dev/null +++ b/tests/bridge/test_streams.py @@ -0,0 +1,29 @@ +import pytest + +import temporalio.bridge.streams +from temporalio.runtime import Runtime +from temporalio.service import RPCStatusCode + + +def _stream_runtime( + memory: bool, config: bytes = b"" +) -> temporalio.bridge.streams.StreamRuntime: + return temporalio.bridge.streams.StreamRuntime( + Runtime.default()._core_runtime, memory, config + ) + + +async def test_memory_runtime_takes_calls_by_method_name(): + runtime = _stream_runtime(True) + # The args are the gRPC code, the message and the encoded details, as in RPCError. + with pytest.raises(temporalio.bridge.streams.StreamCallError) as err: + await runtime.call("NoSuchMethod", b"") + code, _, details = err.value.args + assert code == RPCStatusCode.UNIMPLEMENTED + # StreamErrorDetails with the reason STREAM_ERROR_REASON_UNSUPPORTED (14). + assert details == b"\x08\x0e" + + +def test_runtime_refuses_a_config_that_does_not_decode(): + with pytest.raises(ValueError, match="config"): + _stream_runtime(False, b"\xff\xff")