From 3fa37e2f2f111939537fb452399ae688141e4643 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 25 Oct 2026 14:25:33 -0700 Subject: [PATCH] Added the Stream service messages as experimental. The crate, sdk-python and later the server read one copy of these messages. The gRPC service and its Nexus annotations come with the server. --- temporal/api/enums/v1/stream.proto | 61 ++++++ temporal/api/stream/v1/message.proto | 111 +++++++++++ .../streamservice/v1/request_response.proto | 177 ++++++++++++++++++ 3 files changed, 349 insertions(+) create mode 100644 temporal/api/enums/v1/stream.proto create mode 100644 temporal/api/stream/v1/message.proto create mode 100644 temporal/api/streamservice/v1/request_response.proto diff --git a/temporal/api/enums/v1/stream.proto b/temporal/api/enums/v1/stream.proto new file mode 100644 index 000000000..acefa18e7 --- /dev/null +++ b/temporal/api/enums/v1/stream.proto @@ -0,0 +1,61 @@ +syntax = "proto3"; + +package temporal.api.enums.v1; + +option go_package = "go.temporal.io/api/enums/v1;enums"; +option java_package = "io.temporal.api.enums.v1"; +option java_multiple_files = true; +option java_outer_classname = "StreamProto"; +option ruby_package = "Temporalio::Api::Enums::V1"; +option csharp_namespace = "Temporalio.Api.Enums.V1"; + +// Experimental. Might change in the future. +enum StreamProducerKind { + STREAM_PRODUCER_KIND_UNSPECIFIED = 0; + STREAM_PRODUCER_KIND_WORKFLOW = 1; + STREAM_PRODUCER_KIND_LOCAL_ACTIVITY = 2; + STREAM_PRODUCER_KIND_ACTIVITY = 3; + STREAM_PRODUCER_KIND_CLIENT = 4; + STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY = 5; +} + +// Experimental. Might change in the future. +enum StreamRecordKind { + STREAM_RECORD_KIND_UNSPECIFIED = 0; + STREAM_RECORD_KIND_DATA = 1; + // Lua flags the first batch of a higher epoch. The read makes this Record from the flag. + STREAM_RECORD_KIND_SUPERSEDED = 2; + STREAM_RECORD_KIND_CLOSE = 3; +} + +// Experimental. Might change in the future. +// How a Stream closed, so a reader tells a terminated or timed out owner from a failed one +// without parsing the failure. +// (-- api-linter: core::0216::synonyms=disabled +// aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) +enum StreamCloseStatus { + STREAM_CLOSE_STATUS_UNSPECIFIED = 0; + STREAM_CLOSE_STATUS_COMPLETED = 1; + STREAM_CLOSE_STATUS_FAILED = 2; + STREAM_CLOSE_STATUS_CANCELED = 3; + // TERMINATED and TIMED_OUT come only from the owner's end, never from a caller. + STREAM_CLOSE_STATUS_TERMINATED = 4; + STREAM_CLOSE_STATUS_TIMED_OUT = 5; +} + +// Experimental. Might change in the future. +// (-- api-linter: core::0216::synonyms=disabled +// aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) +enum StreamStatus { + STREAM_STATUS_UNSPECIFIED = 0; + STREAM_STATUS_OPEN = 1; + STREAM_STATUS_CLOSED = 2; + STREAM_STATUS_EXPIRED = 3; +} + +// Experimental. Might change in the future. +enum StreamReadStart { + STREAM_READ_START_UNSPECIFIED = 0; + STREAM_READ_START_BEGINNING = 1; + STREAM_READ_START_END = 2; +} diff --git a/temporal/api/stream/v1/message.proto b/temporal/api/stream/v1/message.proto new file mode 100644 index 000000000..b285ad8b9 --- /dev/null +++ b/temporal/api/stream/v1/message.proto @@ -0,0 +1,111 @@ +syntax = "proto3"; + +package temporal.api.stream.v1; + +option go_package = "go.temporal.io/api/stream/v1;stream"; +option java_package = "io.temporal.api.stream.v1"; +option java_multiple_files = true; +option java_outer_classname = "MessageProto"; +option ruby_package = "Temporalio::Api::Stream::V1"; +option csharp_namespace = "Temporalio.Api.Stream.V1"; + +import "google/protobuf/timestamp.proto"; + +import "temporal/api/common/v1/message.proto"; +import "temporal/api/enums/v1/stream.proto"; +import "temporal/api/failure/v1/message.proto"; + +// Experimental. Might change in the future. +// Identifies one Stream. The id is opaque to callers. +message StreamRef { + string id = 1; + // Unset for a client's Stream. For 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 Store keys on id, and reads only a Workflow owner's + // business_id, to check that a close's ref names the owner's run chain. + temporal.api.common.v1.Execution owner = 2; + // 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. + string backend = 3; +} + +// Experimental. Might change in the future. +// The ordered attempt of a Producer. Compared in this order: task_started_event_id, +// task_attempt, task_started_time, attempt_scheduled_time, attempt. A field that a kind does +// not use is empty and compares as zero. A timestamp compares seconds, then nanos. +message ProducerEpoch { + // The Workflow Task of a Workflow, or the one that dispatched a Local Activity. + int64 task_started_event_id = 1; + int32 task_attempt = 2; + // Tells a reused started event id apart, for example after a dropped speculative task. + google.protobuf.Timestamp task_started_time = 5; + // When the server scheduled an Activity's attempt. A reset makes it later. + google.protobuf.Timestamp attempt_scheduled_time = 3; + // Activity, Local Activity and client attempt. + int32 attempt = 4; +} + +// Experimental. Might change in the future. +message Producer { + temporal.api.enums.v1.StreamProducerKind kind = 1; + string id = 2; + ProducerEpoch epoch = 3; + // The run of the Producer, so nobody parses the id. Empty for a client. + string run_id = 4; +} + +// Experimental. Might change in the future. +message WorkflowTaskEpoch { + int64 started_event_id = 1; + int32 attempt = 2; + google.protobuf.Timestamp started_time = 3; +} + +// Experimental. Might change in the future. +message AppendRecord { + // Empty is the default Topic. + string topic = 1; + temporal.api.common.v1.Payload body = 2; +} + +// Experimental. Might change in the future. +message Supersession { + temporal.api.enums.v1.StreamProducerKind producer_kind = 1; + string producer_id = 2; + ProducerEpoch epoch = 3; + // Workflow only. + WorkflowTaskEpoch previous_completed_task = 4; + repeated int64 failed_tasks = 5; + // Workflow retry only. The previous run. Every Record whose Producer carries it is withdrawn. + string withdraw_run = 6; + // The run of the Producer, so nobody parses the id. Empty for a client. + string run_id = 7; +} + +// Experimental. Might change in the future. +// An offset is the Redis entry id packed as ms << 20 | seq. It 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 names its Stream, so the read checks the offset against that Stream. +message StreamRecord { + temporal.api.enums.v1.StreamRecordKind kind = 1; + int64 offset = 2; + string topic = 3; + Producer producer = 4; + temporal.api.common.v1.Payload body = 5; + Supersession supersession = 6; + temporal.api.common.v1.Payload close_result = 7; + temporal.api.failure.v1.Failure close_failure = 8; + // CLOSE only. The result or the failure stays next to it. + temporal.api.enums.v1.StreamCloseStatus close_status = 9; +} + +// 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. +message StreamErrorDetails { + // A token such as `stream_closed`, `stream_fenced` or `stream_format_too_new`. + // `seq_divergent` is a resend with the sequence of the last batch and other content. + // `stream_not_owner` is a close or a Producer that the Stream's owner doesn't allow. + string reason = 1; +} diff --git a/temporal/api/streamservice/v1/request_response.proto b/temporal/api/streamservice/v1/request_response.proto new file mode 100644 index 000000000..3957b91bb --- /dev/null +++ b/temporal/api/streamservice/v1/request_response.proto @@ -0,0 +1,177 @@ +syntax = "proto3"; + +package temporal.api.streamservice.v1; + +option go_package = "go.temporal.io/api/streamservice/v1;streamservice"; +option java_package = "io.temporal.api.streamservice.v1"; +option java_multiple_files = true; +option java_outer_classname = "RequestResponseProto"; +option ruby_package = "Temporalio::Api::StreamService::V1"; +option csharp_namespace = "Temporalio.Api.StreamService.V1"; + +// The messages of the Stream service. An SDK Store (Redis or memory) and the server answer the +// same five operations with them. A field that only the server reads says so. + +import "google/protobuf/duration.proto"; + +import "temporal/api/common/v1/message.proto"; +import "temporal/api/enums/v1/stream.proto"; +import "temporal/api/failure/v1/message.proto"; +import "temporal/api/stream/v1/message.proto"; + +// Experimental. Might change in the future. +// Idempotent for one id and one Retention. ALREADY_EXISTS for another Retention. +// (-- api-linter: core::0133=disabled +// aip.dev/not-precedent: The request names a Stream by namespace and stream_id. --) +message CreateStreamRequest { + string namespace = 1; + // A Store requires it. Outside Workflow code, the server makes an id when it is empty. + string stream_id = 2; + // The age after which a Record is dropped. Empty is the Store default. + google.protobuf.Duration retention = 3; + // Workflow only. The previous run on a retry run's first create, to withdraw its Records. + string withdraw_run = 4; + // The Workflow run that owns the Stream. Workflow kind, with its run_id and epoch. + temporal.api.stream.v1.Producer owner = 6; + // Only the server reads it. The server closes the Stream with the outcome of the owner's run + // chain, terminate and timeout included. 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. --) + bool close_with_run = 7; + // The ref of the Stream. A Workflow-owned Stream sets the ref's owner, which names the + // owner's run chain. + temporal.api.stream.v1.StreamRef ref = 8; + // Only the server reads it. The server checks that the token is the Activity's current + // attempt, so a caller can't put a Stream into a Standalone Activity it isn't. + bytes task_token = 9; +} + +// Experimental. Might change in the future. +message CreateStreamResponse { + temporal.api.stream.v1.StreamRef stream = 1; + bool created = 2; + temporal.api.enums.v1.StreamStatus status = 3; +} + +// Experimental. Might change in the future. +// FAILED_PRECONDITION if closed, fenced, or if the Stream's meta has a higher format version v +// (detail stream_format_too_new). NOT_FOUND if absent. +message AppendRecordsRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + // Required outside a Workflow. In a Workflow the transport sets it. + temporal.api.stream.v1.Producer producer = 3; + // The batch number in this epoch. With the Producer and epoch it is the dedupe key. + int64 sequence = 4; + repeated temporal.api.stream.v1.AppendRecord records = 5; + // The run's last completed Workflow Task (started_event_id 0 before the first) and its failed + // tasks, from a Local Activity's start. The batch withdraws the run's Local Activity Records + // whose task never completed. + temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6; + repeated int64 failed_tasks = 7; + // Local Activity only. The previous run on a retry run's first batch, to withdraw its Records. + string withdraw_run = 8; + // Only the server reads it, on a Workflow Producer's batch. The batch withdraws no Workflow + // Task and keeps no task facts, but withdraw_run still applies. + bool defer_withdrawal = 9; + // Only the server reads it. The server checks the Producer against the Activity's current + // attempt, so nobody else writes as that Activity. + bytes task_token = 10; +} + +// Experimental. Might change in the future. +message AppendRecordsResponse { + int64 first_offset = 1; + int64 last_offset = 2; +} + +// Experimental. Might change in the future. +message CloseStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + temporal.api.stream.v1.Producer producer = 3; + temporal.api.common.v1.Payload result = 4; + temporal.api.failure.v1.Failure failure = 5; + // Only the server reads these. They are the run's task facts, as on the append. + temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6; + repeated int64 failed_tasks = 7; + // Only the server reads it. The previous run on a retry run's close, to withdraw its Records. + string withdraw_run = 8; + // Empty is COMPLETED with no failure and FAILED with one. A caller sets CANCELED itself. + temporal.api.enums.v1.StreamCloseStatus status = 9; + // Only the server reads it. It checks the token as on the append. + bytes task_token = 10; +} + +// Experimental. Might change in the future. +message CloseStreamResponse { +} + +// Experimental. Might change in the future. +message ReadRecordsRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + // Start at this offset. Zero uses start. INVALID_ARGUMENT if 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. --) + int64 from_offset = 3; + temporal.api.enums.v1.StreamReadStart start = 4; + // Empty reads every Topic in one order. + repeated string topics = 5; + int32 max_records = 6; + google.protobuf.Duration wait = 7; + // How long to wait for a Stream that nobody created yet. Empty fails at once with + // NOT_FOUND. At most 60 s. + google.protobuf.Duration create_wait = 8; + // Also read the Topics that start with one of these. A Record passes when its Topic is in + // topics or starts with a prefix. Both empty reads every Topic. + repeated string topic_prefixes = 10; + // The exclusive end of the read. The read scans no entry at or after it, the close Record + // included, 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. --) + int64 to_offset = 11; + // Park as a normal read, then answer with no Records and next_offset = from_offset, when a + // Record that passes the filter 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. + bool wake_only = 12; + // Read the newest `last` Records that pass the filter, 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. + int32 last = 13; + // 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. + int64 max_bytes = 14; + // 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 the + // server reads it. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: The floor is the lowest from_offset of the run's reads. --) + int64 floor_from = 15; +} + +// Experimental. Might change in the future. +message ReadRecordsResponse { + repeated temporal.api.stream.v1.StreamRecord records = 1; + // The from_offset of the next read, which is the newest offset read + 1. + int64 next_offset = 2; + bool done = 3; + // 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 the server sets it. + bool continue_as_new_suggested = 4; +} + +// Experimental. Might change in the future. +message DescribeStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; +} + +// Experimental. Might change in the future. +message DescribeStreamResponse { + temporal.api.enums.v1.StreamStatus status = 1; + // Zero when the Stream has no Record. + int64 last_offset = 2; + google.protobuf.Duration retention = 3; +}