diff --git a/crates/protos/build.rs b/crates/protos/build.rs index ddcdebf389..2e645bf3be 100644 --- a/crates/protos/build.rs +++ b/crates/protos/build.rs @@ -39,6 +39,8 @@ const SERDE_DERIVE_PREFIXES: &[&str] = &[ ".temporal.api.replication", ".temporal.api.rules", ".temporal.api.schedule", + ".temporal.api.stream", + ".temporal.api.streamservice", ".temporal.api.taskqueue", ".temporal.api.testservice", ".temporal.api.update", @@ -60,6 +62,7 @@ fn main() -> Result<(), Box> { "./protos/api_upstream/temporal/api/workflowservice/v1/service.proto", "./protos/api_upstream/temporal/api/nexusservices/workerservice/v1/request_response.proto", "./protos/api_upstream/temporal/api/operatorservice/v1/service.proto", + "./protos/api_upstream/temporal/api/streamservice/v1/request_response.proto", "./protos/api_upstream/temporal/api/errordetails/v1/message.proto", "./protos/api_cloud_upstream/temporal/api/cloud/cloudservice/v1/service.proto", "./protos/testsrv_upstream/temporal/api/testservice/v1/service.proto", diff --git a/crates/protos/protos/api_upstream/nexus/stream-service.wit b/crates/protos/protos/api_upstream/nexus/stream-service.wit new file mode 100644 index 0000000000..def79a37de --- /dev/null +++ b/crates/protos/protos/api_upstream/nexus/stream-service.wit @@ -0,0 +1,459 @@ +// Draft. The Stream service as a System Nexus service, for the Streams API proposal. Each record +// maps to a message of the protos under ../temporal/api (stream/v1, streamservice/v1 and the +// Stream enums in enums/v1). This file is the source. Each SDK copies it and generates its Workflow +// API and its client with nexgen (`--targets workflow,client --system-nexus`). +package temporal:nexus@1.0.0; + +world system { + export stream-service; +} + +/// @nexus.endpoint "__temporal_system" +/// @nexus.service-name "temporal.api.streamservice.v1.StreamService" +/// @nexus.experimental +interface stream-service { + use nexus:temporal-types/model@1.0.0.{ + duration, + placeholder, + }; + + /// The value type of a Stream's Records. A call's type argument sets it. + /// @nexus.type-parameter + type body-t = placeholder; + + /// The value that `reduce` folds the Records into. A call's initial value sets it. + /// @nexus.type-parameter + type value-t = placeholder; + + // The WIT has no named offset type yet, so this comment tells what each `list` offset holds. + // An offset is the opaque position of a Record in one Stream, and it is proto `bytes`. A caller + // never parses it. It only gives it back to a read, or compares two offsets of one Stream. They + // compare bytewise in the order of their Records, so a caller can tell which one is newer + // without a call to the Store. + // TODO: use `type stream-offset = list` once nexgen reads a named `list` as `bytes`. + + /// @nexus.proto "google.protobuf.Timestamp" + record timestamp { + seconds: s64, + nanos: s32, + } + + /// @nexus.doc "Identifies one Stream. The id is opaque to callers." + /// @nexus.proto "temporal.api.stream.v1.StreamRef" + /// @nexus.generic "body-t" + record stream-ref { + /// @nexus.doc "The Stream id, 1 to 4096 bytes." + id: string, + /// @nexus.doc "Registered Store name. With none, the SDK picks one. Only an SDK reads it." + store: option, + } + + /// @nexus.doc "A Workflow Task attempt of the run." + /// @nexus.proto "temporal.api.stream.v1.WorkflowTaskEpoch" + record workflow-task-epoch { + started-event-id: s64, + /// @nexus.doc "1 or more." + attempt: s32, + /// @nexus.doc "When the Workflow Task attempt started. It tells a reused started id apart." + started-time: option, + } + + /// @nexus.doc "A Local Activity attempt. Its Workflow Task orders it first." + /// @nexus.proto "temporal.api.stream.v1.LocalActivityEpoch" + record local-activity-epoch { + /// @nexus.doc "The Workflow Task attempt that dispatched the Local Activity." + task: workflow-task-epoch, + /// @nexus.doc "The Local Activity's own attempt in that Workflow Task, 1 or more." + attempt: s32, + } + + /// @nexus.doc "An Activity or Standalone Activity attempt." + /// @nexus.proto "temporal.api.stream.v1.ActivityEpoch" + record activity-epoch { + /// @nexus.doc "When the server scheduled this attempt. A reset makes it later." + attempt-scheduled-time: timestamp, + /// @nexus.doc "1 or more." + attempt: s32, + } + + /// @nexus.doc "A Workflow run Producer." + /// @nexus.proto "temporal.api.stream.v1.WorkflowProducer" + record workflow-producer { + /// @nexus.doc "The run id of the Workflow run, not of its chain." + run-id: string, + // The Workflow Task of the write. + epoch: option, + } + + /// @nexus.doc "A Local Activity Producer." + /// @nexus.proto "temporal.api.stream.v1.LocalActivityProducer" + record local-activity-producer { + run-id: string, + epoch: local-activity-epoch, + } + + /// @nexus.doc "An Activity Producer." + /// @nexus.proto "temporal.api.stream.v1.ActivityProducer" + record activity-producer { + /// @nexus.doc "The run of the Workflow that scheduled the Activity. Empty when standalone." + run-id: option, + epoch: activity-epoch, + } + + /// @nexus.doc "A client Producer. It has no run and no epoch, so nothing fences it." + /// @nexus.proto "temporal.api.stream.v1.ClientProducer" + record client-producer { + } + + // The case is the kind of the Producer, so a kind sets only its own fields. + /// @nexus.doc "The kind of a Producer, with its run and epoch." + /// @nexus.proto "temporal.api.stream.v1.Producer.kind" + variant producer-kind { + workflow(workflow-producer), + local-activity(local-activity-producer), + workflow-activity(activity-producer), + standalone-activity(activity-producer), + client(client-producer), + } + + /// @nexus.doc "The code that wrote a batch. A higher epoch fences and supersedes a lower one." + /// @nexus.proto "temporal.api.stream.v1.Producer" + record producer { + /// @nexus.doc "Stable across attempts, such as wf: for a Workflow run." + id: string, + // A newer server can send a kind that this SDK doesn't know. Then the kind is empty. + /// @nexus.doc "Required. The case is the kind of the Producer." + kind: option, + } + + /// @nexus.doc "One Record to append." + /// @nexus.proto "temporal.api.stream.v1.AppendRecord" + /// @nexus.flatten-in-api + record append-record { + /// @nexus.doc "The value." + body: body-t, + } + + // An offset is ordered and unique in one Stream, and sparse. Offsets have gaps, so they don't + // count Records. A read gives its Stream, so the read checks the offset against that Stream. + // A read gives only Records. A close stores no Record, so the read response tells of it with + // done. + /// @nexus.doc "One Record that a Producer appended, as a read gives it." + /// @nexus.proto "temporal.api.stream.v1.StreamRecord" + record stream-record { + /// @nexus.doc "The value." + body: body-t, + producer: option, + /// @nexus.doc "The place of this Record in the Stream. Ordered and unique, with gaps." + offset: list, + } + + /// Why a read of many pages ended. + enum stream-read-stop { + match, + done, + max-records, + max-bytes, + idle, + timeout, + } + + // The SDK's support package builds this from many pages of one call. It has no proto message, + // since it never goes on the wire. + /// @nexus.doc "What one read of many pages gave." + record stream-read { + records: list, + /// @nexus.doc "Where the scan stopped. Give it as the from-offset of the next call." + last-offset: option>, + stop: stream-read-stop, + done: bool, + } + + // A function of the caller stays in the process, so this signature is no operation. + record-predicate: func(%record: stream-record) -> bool; + + /// @nexus.function signature="record-predicate" + type read-predicate = placeholder; + + /// @nexus.doc "What one fold of many pages gave." + record stream-reduce { + value: value-t, + count: s32, + last-offset: option>, + stop: stream-read-stop, + done: bool, + } + + fold-function: func(value: value-t, %record: stream-record) -> value-t; + + fold-predicate: func(value: value-t, %record: stream-record) -> bool; + + /// @nexus.function signature="fold-function" + type fold = placeholder; + + /// @nexus.function signature="fold-predicate" + type fold-until = placeholder; + + /// @nexus.doc + /// "Create a Stream. If it exists, fail, or with use-existing get the open one." + /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamRequest" + record create-stream-request { + // Its field name isn't ref, so the created Stream takes its ref from the response. + /// @nexus.doc "The ref of the Stream to create. Its id is the Stream id." + /// @nexus.flatten-in-api + stream-ref: stream-ref, + /// @nexus.doc "What the create does when the Stream exists. Unspecified is fail." + id-conflict-policy: option, + // A retry of the create that made the Stream sends the same id, so it doesn't fail as a + // conflict. In a Workflow the helper gives a deterministic id. + /// @nexus.doc "Unique for each create call, and the same on its retries." + /// @nexus.source python="request_id()" typescript="requestId()" + /// dotnet="StreamWorkflowContext.RequestId()" + request-id: option, + // Native only. The SDK sets it, and the SDK Stores don't read it. + /// @nexus.omit + namespace: placeholder, + } + + /// What a create does when its Stream exists. Unspecified is fail. + /// @nexus.proto "temporal.api.enums.v1.StreamIdConflictPolicy" + enum stream-id-conflict-policy { + stream-id-conflict-policy-unspecified, + stream-id-conflict-policy-fail, + stream-id-conflict-policy-use-existing, + } + + /// @nexus.proto "temporal.api.enums.v1.StreamStatus" + enum stream-status { + /// @nexus.fallback + stream-status-unspecified, + stream-status-open, + stream-status-closed, + stream-status-expired, + } + + /// @nexus.doc + /// "Append one batch. The Store keeps a resent batch once, by Producer, epoch and sequence." + /// @nexus.proto "temporal.api.streamservice.v1.AppendRecordsRequest" + record append-records-request { + ref: stream-ref, + records: list, + // A call can pass its own Producer. A call that passes none gets the default, which the + // helper of each context picks for where it runs. + /// @nexus.doc "The Producer of the batch. Defaults to the Producer of the calling context." + /// @nexus.default targets="workflow" python="_support.workflow_producer()" + /// typescript="support.workflowProducer()" dotnet="StreamWorkflowContext.Producer()" + /// @nexus.default targets="client" python="_support.default_producer()" + /// typescript="support.defaultProducer()" dotnet="StreamActivityContext.Producer()" + producer: producer, + // A call that passes no sequence sends none. The Store then numbers the batch, so nothing + // dedupes it. + /// @nexus.doc + /// "The batch number in this epoch. Empty takes the next one, and a retry writes it again." + sequence: option, + // Native only. The SDK sets it, and the SDK Stores don't read it. + /// @nexus.omit + namespace: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.AppendRecordsResponse" + record append-records-response { + first-offset: option>, + last-offset: option>, + } + + // A close carries no value, so code that has a last value appends it first. + /// @nexus.doc "Close a Stream. Readers drain what is left, then end. Later appends are refused." + /// @nexus.proto "temporal.api.streamservice.v1.CloseStreamRequest" + record close-stream-request { + ref: stream-ref, + /// @nexus.doc + /// "The Producer that closes the Stream. Defaults to the Producer of the calling context." + /// @nexus.default targets="workflow" python="_support.workflow_producer()" + /// typescript="support.workflowProducer()" dotnet="StreamWorkflowContext.Producer()" + /// @nexus.default targets="client" python="_support.default_producer()" + /// typescript="support.defaultProducer()" dotnet="StreamActivityContext.Producer()" + producer: producer, + // Native only. The SDK sets it, and the SDK Stores don't read it. + /// @nexus.omit + namespace: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.CloseStreamResponse" + record close-stream-response { + } + + /// @nexus.doc "Read Records after an offset." + /// @nexus.proto "temporal.api.streamservice.v1.ReadRecordsRequest" + record read-records-request { + ref: stream-ref, + // To read only new Records, a caller gets last-offset from a describe and reads after it. + /// @nexus.doc "Read after this offset. Empty is the beginning. A too high offset fails." + from-offset: option>, + /// @nexus.doc "About this many Records per page, 1 to 1000." + max-records: option, + /// @nexus.doc "How long to wait for a new Record before an empty answer. At most 60 s." + wait: option, + /// @nexus.doc "The maximum Record bytes in a page. A larger Record is still returned alone." + max-bytes: option, + // Native only. The SDK sets it, and the SDK Stores don't read it. + /// @nexus.omit + namespace: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.ReadRecordsResponse" + record read-records-response { + records: list, + /// @nexus.doc "The last offset that this read covered. The next read starts after it." + last-offset: option>, + /// @nexus.doc "True when the Stream is closed and this read gave its last Record." + done: bool, + } + + /// @nexus.proto "temporal.api.streamservice.v1.DescribeStreamRequest" + record describe-stream-request { + ref: stream-ref, + // Native only. The SDK sets it, and the SDK Stores don't read it. + /// @nexus.omit + namespace: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.DescribeStreamResponse" + record describe-stream-response { + status: option, + /// @nexus.doc "The newest offset, of a Record. Empty for none." + last-offset: option>, + } + + // A stream never changes, since offsets are values that its methods return. Only a create's + // response sets created, so a stream built from a ref has none. + /// @nexus.doc "One Stream. Build it from a stream-ref, or get it from a create." + // TODO: implement @nexus.flatten-in-api param="ref". + resource %stream { + constructor(ref: stream-ref, created: option); + + /// @nexus.doc "Append values, and return the offsets of the batch." + /// @nexus.operation "append-records" + /// @nexus.varargs "records" + append: func( + records: list, + producer: option, + sequence: option, + ) -> append-records-response; + + /// @nexus.doc "Close the Stream. Readers drain what is left, then end." + /// @nexus.operation "close-stream" + /// @nexus.close + close: func(producer: option); + + // Workflow code reads with `take`, `filter` and `reduce`, not with a page read. So the + // Workflow target omits `read`, and the client target keeps it. + /// @nexus.doc "Read one page of Records. The poll iterator reads them all." + /// @nexus.operation "read-records" + /// @nexus.api-omit targets="workflow" + read: func( + from-offset: option>, + max-records: option, + wait: option, + max-bytes: option, + ) -> read-records-response; + + /// @nexus.doc "Describe the Stream's status and newest offset." + /// @nexus.operation "describe-stream" + describe: func() -> describe-stream-response; + + // The SDK's support package reads many pages in one call, so no operation binds these + // methods. It finds out where it runs, and Workflow code reads in one Local Activity. + /// @nexus.doc "Read every Record up to a stop." + /// @nexus.helper python="take_stream" + take: func( + timeout: duration, + until: option, + from-offset: option>, + idle-timeout: option, + max-records: option, + max-bytes: option, + ) -> stream-read; + + /// @nexus.doc "Read the Records that the predicate keeps, up to a stop." + /// @nexus.helper python="filter_stream" + filter: func( + predicate: read-predicate, + timeout: duration, + until: option, + from-offset: option>, + idle-timeout: option, + max-records: option, + max-bytes: option, + ) -> stream-read; + + /// @nexus.doc "Fold the Records into one value, up to a stop." + /// @nexus.helper python="reduce_stream" + reduce: func( + function: fold, + initial: value-t, + timeout: duration, + until: option, + from-offset: option>, + idle-timeout: option, + max-records: option, + max-bytes: option, + ) -> stream-reduce; + } + + // With use-existing, created is false when the Stream existed. + /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamResponse" + type create-stream-result = own<%stream>; + + // The Workflow API has no other way to create a Stream, so this function stays public. + /// @nexus.operation name="CreateStream" + create-stream: func(request: create-stream-request) -> create-stream-result; + + /// @nexus.doc "Append one batch. Workflow code makes the call from a Local Activity." + /// @nexus.operation name="AppendRecords" + /// @nexus.api-omit + append-records: func(request: append-records-request) -> append-records-response; + + /// @nexus.operation name="CloseStream" + /// @nexus.api-omit + close-stream: func(request: close-stream-request) -> close-stream-response; + + // A WIT stream can't cross a Nexus call, so a read is a request and a response. The SDK's support + // helper builds the next request from a page for the poll iterator. + /// @nexus.doc + /// "A long poll. NOT_FOUND for an uncreated Stream. Workflow code reads from a Local Activity." + /// @nexus.operation name="ReadRecords" + /// @nexus.poll python="next_read_request" items="records" python-resume="resume_read_request" + /// targets="client" + /// @nexus.api-omit + // TODO: implement polling for typescript="nextReadRequest" dotnet="StreamPoll.NextReadRequest". + read-records: func(request: read-records-request) -> read-records-response; + + /// @nexus.operation name="DescribeStream" + /// @nexus.api-omit + describe-stream: func(request: describe-stream-request) -> describe-stream-response; + /// @nexus.proto "temporal.api.enums.v1.StreamErrorReason" + enum stream-error-reason { + stream-error-reason-unspecified, + stream-error-reason-not-found, + stream-error-reason-expired, + stream-error-reason-closed, + stream-error-reason-fenced, + stream-error-reason-sequence-too-low, + stream-error-reason-seq-divergent, + stream-error-reason-not-owner, + stream-error-reason-format-too-new, + stream-error-reason-offset-too-high, + stream-error-reason-too-large, + stream-error-reason-invalid, + stream-error-reason-storage-unavailable, + stream-error-reason-unsupported, + stream-error-reason-already-exists, + stream-error-reason-storage-corrupt, + } + + /// @nexus.proto "temporal.api.stream.v1.StreamErrorDetails" + record stream-error-details { + reason: option, + } +} diff --git a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto new file mode 100644 index 0000000000..20d87ba48f --- /dev/null +++ b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto @@ -0,0 +1,72 @@ +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. +// The state of a Stream, as a describe gives it. +// (-- 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. +// What a create does when its Stream exists. Unspecified is FAIL, as for a Workflow start. +enum StreamIdConflictPolicy { + STREAM_ID_CONFLICT_POLICY_UNSPECIFIED = 0; + // The create fails with ALREADY_EXISTS. + STREAM_ID_CONFLICT_POLICY_FAIL = 1; + // The create gets the open Stream, which keeps its own Retention. + STREAM_ID_CONFLICT_POLICY_USE_EXISTING = 2; +} + +// Experimental. Might change in the future. +// Why a Stream call failed. Each reason has one gRPC status code, and several reasons share a +// code. So code matches on the reason, and not on the status code. An SDK raises one error type +// that carries the reason. +enum StreamErrorReason { + STREAM_ERROR_REASON_UNSPECIFIED = 0; + // Nobody created the Stream. NOT_FOUND. + STREAM_ERROR_REASON_NOT_FOUND = 1; + // The Stream's Retention or its last-write TTL passed. NOT_FOUND. + STREAM_ERROR_REASON_EXPIRED = 2; + // The Stream is closed. FAILED_PRECONDITION. + STREAM_ERROR_REASON_CLOSED = 3; + // The Producer's epoch is lower than the newest one. FAILED_PRECONDITION. + STREAM_ERROR_REASON_FENCED = 4; + // The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION. + STREAM_ERROR_REASON_SEQUENCE_TOO_LOW = 5; + // The batch has the `sequence` of the last one of its epoch, and other content. + // FAILED_PRECONDITION. + STREAM_ERROR_REASON_SEQ_DIVERGENT = 6; + // The caller may not create or close this Stream. FAILED_PRECONDITION. + STREAM_ERROR_REASON_NOT_OWNER = 7; + // A newer format version wrote the Stream. FAILED_PRECONDITION. + STREAM_ERROR_REASON_FORMAT_TOO_NEW = 8; + // The offset is past the newest offset, 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_OFFSET_TOO_HIGH = 9; + // A Record or a batch is over the Store's limit. INVALID_ARGUMENT. + STREAM_ERROR_REASON_TOO_LARGE = 10; + // The request is malformed. INVALID_ARGUMENT. + STREAM_ERROR_REASON_INVALID = 11; + // The Store failed or did not answer. A retry can succeed. UNAVAILABLE. + STREAM_ERROR_REASON_STORAGE_UNAVAILABLE = 12; + // The Store lacks a capability that the call needs. UNIMPLEMENTED. + STREAM_ERROR_REASON_UNSUPPORTED = 13; + // A create with the FAIL conflict policy is for a Stream that exists. ALREADY_EXISTS. + STREAM_ERROR_REASON_ALREADY_EXISTS = 14; + // 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. + STREAM_ERROR_REASON_STORAGE_CORRUPT = 15; +} diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto new file mode 100644 index 0000000000..376f883d69 --- /dev/null +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -0,0 +1,153 @@ +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"; + +// Experimental. Might change in the future. +// Identifies one Stream. The id is opaque to callers. A Stream has no incarnations. A create of a +// closed or expired Stream fails, and it doesn't make a second Stream with the same id. +message StreamRef { + string id = 1; + // The name of the Store that holds the Stream, as the Worker and client config register it, + // so every process routes a ref it receives. With no name, the SDK picks the Store from its + // config. Only an SDK sets and reads it, since only the SDK knows the name that its config + // gives a Store. A request carries the whole ref, and the server and a Store ignore this + // field. + string store = 3; +} + +// Experimental. Might change in the future. +// One Workflow Task attempt of a run. It orders by started_event_id, then attempt, then +// started_time. An epoch orders the attempts of one Producer only, and a Producer belongs to one +// run. So a reset run and the next run of a chain have other Producers, and their epochs start +// again. +message WorkflowTaskEpoch { + // Required. + int64 started_event_id = 1; + // Required. 1 or more. + int32 attempt = 2; + // Tells a reused started event id apart, for example after a dropped speculative task. Core + // leaves it empty when the server gave the task no started time. + google.protobuf.Timestamp started_time = 3; +} + +// Experimental. Might change in the future. +// The attempt of a Local Activity. It orders by task, then attempt, so the same call in a later +// Workflow Task is a higher epoch. The Producer carries the run. Records of two runs have no +// order by epoch, only their order in the Stream. +message LocalActivityEpoch { + // Required. The Workflow Task attempt that dispatched the Local Activity. + WorkflowTaskEpoch task = 1; + // Required. The Local Activity's own attempt in that Workflow Task, 1 or more. + int32 attempt = 2; +} + +// Experimental. Might change in the future. +// The attempt of an Activity. It orders by attempt_scheduled_time, then attempt. +message ActivityEpoch { + // Required. When the server scheduled this attempt. An Activity reset starts the attempt count + // again, so only this time orders the attempts across a reset. + google.protobuf.Timestamp attempt_scheduled_time = 1; + // Required. 1 or more. + int32 attempt = 2; +} + +// Experimental. Might change in the future. +// A Workflow run Producer. On the server's Streams, a Workflow writes as this Producer in its +// Workflow Task. On a Store, Workflow code writes through a Local Activity, so its writes carry a +// Local Activity Producer. +message WorkflowProducer { + // Required. The run id of the Workflow run, not of its chain. Each run is its own Producer. + string run_id = 1; + // The Workflow Task of the write. + WorkflowTaskEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// A Local Activity Producer, one for each call of a run. +message LocalActivityProducer { + // Required. The run id of the Workflow run. + string run_id = 1; + // Required. + LocalActivityEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// An Activity Producer, for an Activity of a Workflow or a Standalone Activity. +message ActivityProducer { + // The run id of the Workflow run that scheduled the Activity. Empty for a Standalone + // Activity, whose own run id is in the Producer id. + string run_id = 1; + // Required. + ActivityEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// A client Producer. A client has no run and no epoch. So nothing fences a client's writes, and a +// reader never hides them. +message ClientProducer { +} + +// Experimental. Might change in the future. +// The code that wrote a batch, which is one unit of work. Its id stays the same across attempts, +// and an epoch is one attempt. The kind's case orders its epochs. After a write of a higher epoch, +// a write of a lower epoch of the same Producer fails with FENCED, and a reader may hide the +// Records of the lower epoch. The rule is the same for every kind that has an epoch. A client has +// none. The epochs of two Producers never compare. A new run has new Producers, so no write +// fences or supersedes the Records of an earlier run, such as the base run of a reset. +message Producer { + // Required. Stable across attempts, such as `wf:` for the Workflow Producer of a run. + // It is unique in one Stream. The id of a run-based kind has its run id, so it stays unique + // when Producers of more than one namespace write to one Stream. + string id = 1; + // Required. The case is the kind of the Producer. + oneof kind { + WorkflowProducer workflow = 2; + LocalActivityProducer local_activity = 3; + // An Activity that a Workflow scheduled. + ActivityProducer workflow_activity = 4; + // An Activity with no Workflow. Both Activity cases order their epochs the same way. The + // case tells which identity the Producer has, since only the first has a Workflow run. + ActivityProducer standalone_activity = 5; + ClientProducer client = 6; + } +} + +// Experimental. Might change in the future. +// One Record of a batch to append. The Store adds the batch's Producer and an offset. +message AppendRecord { + // Required. + temporal.api.common.v1.Payload body = 1; +} + +// Experimental. Might change in the future. +// One Record that a Producer appended, as a read gives it. A read gives only Records. A close +// stores no Record, so a read tells of it with the done flag of its response. +message StreamRecord { + Producer producer = 1; + temporal.api.common.v1.Payload body = 2; + // Required. The Record's offset. An offset is ordered and unique in one Stream, and it is + // sparse. Offsets have gaps, so they don't count Records. Two offsets of one Stream compare + // bytewise, as unsigned bytes, in the order of their Records. So a reader, a Store and the + // server order them without decoding them. An offset is never empty. A read gives its + // Stream, so the read checks the offset against that Stream. + bytes offset = 3; +} + +// 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 { + temporal.api.enums.v1.StreamErrorReason reason = 1; +} diff --git a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto new file mode 100644 index 0000000000..4d23665dab --- /dev/null +++ b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto @@ -0,0 +1,152 @@ +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 is only read by the server says so. + +import "google/protobuf/duration.proto"; + +import "temporal/api/enums/v1/stream.proto"; +import "temporal/api/stream/v1/message.proto"; + +// 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. 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 and EXPIRED on +// an expired one. A create takes no Retention. The Store config holds it, as a namespace holds +// the Retention of its Workflows. +// (-- api-linter: core::0133=disabled +// aip.dev/not-precedent: The request identifies a Stream by namespace and ref. --) +message CreateStreamRequest { + // The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + // read it. + string namespace = 1; + // Required. The ref of the Stream to create. Its id is the Stream id, and is never empty. + temporal.api.stream.v1.StreamRef stream_ref = 2; + // What the create does when the Stream exists. Unspecified is FAIL. + temporal.api.enums.v1.StreamIdConflictPolicy id_conflict_policy = 3; + // 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. + string request_id = 4; +} + +// Experimental. Might change in the future. +// The Stream that a create made or got. +message CreateStreamResponse { + // The ref of the Stream, which the caller passes on as data. The SDK sets the name of the + // Store that it picked, also when the request left the name empty. The server gives the ref + // back as the request had it, since it doesn't know that name. + temporal.api.stream.v1.StreamRef ref = 1; + // False when USE_EXISTING got a Stream that existed. + bool created = 2; +} + +// 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. +message AppendRecordsRequest { + // The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + // read it. + string namespace = 1; + // Required. + temporal.api.stream.v1.StreamRef ref = 2; + // Required. The Producer of the batch. An append and a close of one Producer carry the same + // Producer, so the Store orders each write the same way. + temporal.api.stream.v1.Producer producer = 3; + // 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. + int64 sequence = 4; + repeated temporal.api.stream.v1.AppendRecord records = 5; +} + +// Experimental. Might change in the future. +// Where the batch's Records are in the Stream. +message AppendRecordsResponse { + bytes first_offset = 1; + bytes last_offset = 2; +} + +// 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. +message CloseStreamRequest { + // The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + // read it. + string namespace = 1; + // Required. + temporal.api.stream.v1.StreamRef ref = 2; + // Required. The Producer that closes the Stream. + temporal.api.stream.v1.Producer producer = 3; +} + +// Experimental. Might change in the future. +// A close answers when the Store has the close. It gives nothing back. +message CloseStreamResponse { +} + +// Experimental. Might change in the future. +// Reads one page of Records after an offset. At the end of the Stream, the read waits for a new +// Record, so readers long poll. +message ReadRecordsRequest { + // The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + // read it. + string namespace = 1; + // Required. + temporal.api.stream.v1.StreamRef ref = 2; + // Read after this offset. It fails with OFFSET_TOO_HIGH if it is past the newest offset. + // Leave empty to read from the beginning. To read only new Records, a caller gets last_offset + // from a describe and reads after it. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "from" marks the start of a range of offsets. --) + bytes from_offset = 3; + // The most Records in one page. Zero is 1000. + int32 max_records = 4; + // 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. + google.protobuf.Duration wait = 5; + // 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 = 6; +} + +// Experimental. Might change in the future. +// One page of a read. +message ReadRecordsResponse { + repeated temporal.api.stream.v1.StreamRecord records = 1; + // The last offset that this read covered, which is the offset of the newest Record of the + // page. The next read gives it as from_offset and starts after it. An empty page gives the + // request's from_offset, and that is empty when the request had none. So it is always + // known, and an empty value means that the next read starts as this one did. + bytes last_offset = 2; + // True when the Stream is closed and this read gave its last Record. It is the only sign of a + // close in a read, since a close stores no Record. A read of a closed Stream with no Record + // left answers at once with done. + bool done = 3; +} + +// Experimental. Might change in the future. +// Gets the status and the newest offset of a Stream. +message DescribeStreamRequest { + // The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + // read it. + string namespace = 1; + // Required. + temporal.api.stream.v1.StreamRef ref = 2; +} + +// Experimental. Might change in the future. +// The state of a Stream. It has no Retention, since the Retention is a setting of the Store and +// not of one Stream. +message DescribeStreamResponse { + temporal.api.enums.v1.StreamStatus status = 1; + // The newest offset, of a Record. Empty when the Stream has none. + bytes last_offset = 2; +} diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index a5d5a4a827..1bd681159e 100644 --- a/crates/protos/src/protos/mod.rs +++ b/crates/protos/src/protos/mod.rs @@ -2619,6 +2619,16 @@ pub mod temporal { include_proto_with_serde!("temporal.api.sdk.v1"); } } + pub mod stream { + pub mod v1 { + tonic::include_proto!("temporal.api.stream.v1"); + } + } + pub mod streamservice { + pub mod v1 { + tonic::include_proto!("temporal.api.streamservice.v1"); + } + } pub mod taskqueue { pub mod v1 { tonic::include_proto!("temporal.api.taskqueue.v1");