diff --git a/Makefile b/Makefile index a0b49fe91..091084483 100644 --- a/Makefile +++ b/Makefile @@ -35,7 +35,11 @@ PROTO_PATHS = paths=source_relative:$(PROTO_OUT) OAPI_OUT := openapi OAPI3_PATH := .components.schemas.Payload -NEX_GEN ?= nexgen +# The Stream WIT uses nexgen directives that no release has yet. Pin the fork commit that has them +# until a release does. +NEX_GEN_REV := f9578ddbd8dbc8a1a79320210d58d1b722cd542b +NEX_GEN_ROOT := $(CURDIR)/$(STAMPDIR)/nexgen-$(NEX_GEN_REV) +NEX_GEN ?= $(NEX_GEN_ROOT)/bin/nexgen $(PROTO_OUT): mkdir $(PROTO_OUT) @@ -142,6 +146,7 @@ nexus-rpc-yaml-install: @cd cmd/protoc-gen-nexus-rpc-yaml && go install . ##### Compile system Nexus WIT files ##### +# No Stream operation returns StreamErrorDetails, so add-message selects it by name. system-nexus-wit: system-nexus-wit-install nex-gen-install printf $(COLOR) "Generate system Nexus WIT..." tmp=$$(mktemp -d); trap 'rm -rf "$$tmp"' EXIT; \ @@ -152,21 +157,32 @@ system-nexus-wit: system-nexus-wit-install nex-gen-install --system-nexus-wit_opt=linked_input=nexus/deps \ --system-nexus-wit_out=. \ temporal/api/workflowservice/v1/service.proto \ - temporal/api/notificationservice/v1/request_response.proto && \ + temporal/api/notificationservice/v1/request_response.proto \ + temporal/api/streamservice/v1/request_response.proto && \ $(NEX_GEN) add-message \ --descriptors "$$tmp/system.bin" \ --file temporal/api/notificationservice/v1/request_response.proto \ --input nexus/notification-service.wit \ --input nexus/deps \ - --output nexus/notification-service.wit + --output nexus/notification-service.wit && \ + $(NEX_GEN) add-message \ + --descriptors "$$tmp/system.bin" \ + --file temporal/api/streamservice/v1/request_response.proto \ + --message temporal.api.stream.v1.StreamErrorDetails \ + --input nexus/stream-service.wit \ + --input nexus/deps \ + --output nexus/stream-service.wit system-nexus-wit-install: printf $(COLOR) "Build and install protoc-gen-system-nexus-wit..." @cd cmd/protoc-gen-system-nexus-wit && go install . +# The git CLI follows a local URL rewrite to SSH, which cargo's own git client can't authenticate. nex-gen-install: printf $(COLOR) "Install nexgen if missing..." - command -v $(NEX_GEN) >/dev/null || cargo install nexgen --version ^0.2 --features advanced + command -v $(NEX_GEN) >/dev/null || CARGO_NET_GIT_FETCH_WITH_CLI=true cargo install \ + --git https://github.com/moetemp/nexgen --rev $(NEX_GEN_REV) --locked --features advanced \ + --root $(NEX_GEN_ROOT) nexgen ##### Clean ##### clean: diff --git a/nexus/stream-service.wit b/nexus/stream-service.wit new file mode 100644 index 000000000..68bc13d1b --- /dev/null +++ b/nexus/stream-service.wit @@ -0,0 +1,467 @@ +// 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; + + /// @nexus.proto "google.protobuf.Timestamp" + record timestamp { + seconds: option, + nanos: option, + } + + /// @nexus.doc "Identifies one Stream. The id is opaque to callers." + /// @nexus.proto "temporal.api.stream.v1.StreamRef" + record stream-ref { + /// @nexus.doc "The Stream id, 1 to 4096 bytes." + id: string, + /// @nexus.doc "The run chain or Standalone Activity of a bound Stream. Empty if standalone." + owner: option, + /// @nexus.doc "Registered backend name, empty for the default. The Store never reads it." + backend: option, + } + + /// @nexus.proto "temporal.api.enums.v1.ExecutionType" + enum execution-type { + /// @nexus.fallback + execution-type-unspecified, + execution-type-workflow, + execution-type-activity, + execution-type-nexus-operation, + } + + /// @nexus.doc "A Workflow run chain or a Standalone Activity run." + /// @nexus.proto "temporal.api.common.v1.Execution" + record execution { + %type: option, + /// @nexus.doc "The Workflow id, or the Standalone Activity's id." + business-id: option, + /// @nexus.doc "The run chain's first run id, or the Standalone Activity's run id." + run-id: option, + } + + /// @nexus.doc "A Workflow Task attempt of the run." + /// @nexus.proto "temporal.api.stream.v1.WorkflowTaskEpoch" + record workflow-task-epoch { + started-event-id: option, + attempt: option, + /// @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: option, + /// @nexus.doc "The Local Activity's own attempt in that Workflow Task." + attempt: option, + } + + /// @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: option, + attempt: option, + } + + /// @nexus.doc "A client attempt, which the client picks." + /// @nexus.proto "temporal.api.stream.v1.ClientEpoch" + record client-epoch { + attempt: option, + } + + // The case is the kind of the Producer, so a kind sets only its own fields. + /// @nexus.doc "The epoch of each kind of Producer." + /// @nexus.proto "temporal.api.stream.v1.ProducerEpoch.kind" + variant producer-epoch-kind { + workflow(workflow-task-epoch), + local-activity(local-activity-epoch), + activity(activity-epoch), + standalone-activity(activity-epoch), + client(client-epoch), + } + + /// @nexus.doc "The ordered attempt of a Producer. A higher epoch supersedes a lower one." + /// @nexus.proto "temporal.api.stream.v1.ProducerEpoch" + record producer-epoch { + // A newer server can send a kind that this SDK doesn't know. Then the kind is empty. + kind: option, + } + + /// @nexus.doc "The Producer of a batch." + /// @nexus.proto "temporal.api.stream.v1.Producer" + record producer { + /// @nexus.doc "Stable across attempts, such as wf: for a Workflow run." + id: string, + /// @nexus.doc "Required. Its case gives the kind of the Producer." + epoch: option, + /// @nexus.doc "The run of the Producer, so nobody parses the id. Empty for a client." + run-id: option, + } + + /// @nexus.doc "One Record to append." + /// @nexus.proto "temporal.api.stream.v1.AppendRecord" + /// @nexus.unwrap + record append-record { + /// @nexus.doc "The value." + body: body-t, + } + + // A Local Activity, Activity or Standalone Activity write withdraws the Records of its Producer + // with a lower epoch. A Workflow write, with its task t and the previous completed task c, + // withdraws the Records of the run and of its Local Activities whose Workflow Task is in + // failed-tasks or strictly between c and t. It also withdraws those at the started event id of c + // or of t with a lower epoch than c or t. No task in that range can have completed. A client + // write withdraws nothing. + /// @nexus.doc "The Producer of a write, and which earlier Records the write withdraws." + /// @nexus.proto "temporal.api.stream.v1.StreamWriter" + record stream-writer { + /// @nexus.doc "Required on a write. Its epoch's case picks the withdrawal rule." + producer: option, + /// @nexus.doc "Workflow and Local Activity only. The run's last completed Workflow Task." + previous-completed-task: option, + /// @nexus.doc "Workflow and Local Activity only. The run's failed or timed out Workflow Tasks." + failed-tasks: list, + /// @nexus.doc "Retry run only. The previous run, all of whose Records are withdrawn." + withdraw-run: option, + } + + /// @nexus.doc "Which earlier Records a Superseded item withdraws." + /// @nexus.proto "temporal.api.stream.v1.StreamSuperseded" + record stream-superseded { + /// @nexus.doc "The writer that withdraws. Its epoch is empty when only withdraw-run applies." + writer: option, + } + + /// @nexus.doc "One Record that a Producer appended." + /// @nexus.proto "temporal.api.stream.v1.StreamData" + record stream-data { + /// @nexus.doc "The value." + body: body-t, + producer: option, + } + + // The read builds a Superseded item from a flagged batch. + /// @nexus.doc "A read item: a Record or a Superseded item. A close writes no item." + /// @nexus.proto "temporal.api.stream.v1.StreamRecord.item" + variant stream-item { + data(stream-data), + superseded(stream-superseded), + } + + // 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. + /// @nexus.doc "One item that a read gives." + /// @nexus.proto "temporal.api.stream.v1.StreamRecord" + record stream-record { + // A newer server can send an item kind that this SDK doesn't know. Then the item is empty. + item: option, + /// @nexus.doc "The place of this item in the Stream. Ordered and unique, with gaps." + offset: option, + } + + /// @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, which + // carries the owner. + /// @nexus.doc "The ref of the Stream to create. Its id is the Stream id." + /// @nexus.flatten-in-api fields="id backend" + stream-ref: stream-ref, + /// @nexus.doc "The age after which a Record is dropped. Empty is the Store default." + retention: option, + // A wrong run would silently withdraw real Records, so it is never a parameter. The helper + // gives the previous run in a retry run's Workflow, and nothing elsewhere. + /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws." + /// @nexus.source targets="workflow" python="withdraw_run()" typescript="withdrawRun()" + /// dotnet="StreamWorkflowContext.WithdrawRun()" + /// @nexus.source targets="client" python="withdraw_run()" typescript="withdrawRun()" + /// dotnet="StreamWorkflowContext.WithdrawRun()" + withdraw-run: option, + // A bound Stream's owner. An SDK creates only standalone Streams for now. + /// @nexus.omit + owner: placeholder, + /// @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 Stores don't read it. + /// @nexus.omit + close-with-run: placeholder, + /// @nexus.omit + namespace: placeholder, + // Native only. The SDK Stores don't read it. + /// @nexus.omit + task-token: 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 wrong run or task would silently withdraw real Records, so the writer is never a + // parameter. The helper of each context fills only the fields that apply where it runs. + /// @nexus.doc "Required outside a Workflow. In a Workflow the transport sets it." + /// @nexus.source targets="workflow" python="workflow_writer()" + /// typescript="workflowWriter()" dotnet="StreamWorkflowContext.Writer()" + /// @nexus.source targets="client" python="client_writer()" typescript="clientWriter()" + /// dotnet="StreamActivityContext.Writer()" + writer: option, + // A call with no writer state, such as publish, sends no sequence. 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, + /// @nexus.omit + namespace: placeholder, + // Native only. The SDK Stores don't read them. + /// @nexus.omit + defer-withdrawal: placeholder, + /// @nexus.omit + task-token: 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 publishes 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, + // The SDK Stores read only the Producer. The server also reads the withdrawal fields. + /// @nexus.source targets="workflow" python="workflow_writer()" + /// typescript="workflowWriter()" dotnet="StreamWorkflowContext.Writer()" + /// @nexus.source targets="client" python="client_writer()" typescript="clientWriter()" + /// dotnet="StreamActivityContext.Writer()" + writer: option, + /// @nexus.omit + namespace: placeholder, + // Native only. The SDK Stores don't read it. + /// @nexus.omit + task-token: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.CloseStreamResponse" + record close-stream-response { + } + + /// @nexus.proto "temporal.api.enums.v1.StreamReadStart" + enum stream-read-start { + stream-read-start-unspecified, + stream-read-start-beginning, + stream-read-start-end, + } + + /// @nexus.doc "Read Records from an offset." + /// @nexus.proto "temporal.api.streamservice.v1.ReadRecordsRequest" + record read-records-request { + ref: stream-ref, + /// @nexus.doc "Start offset. Empty uses start. An offset past the newest one + 1 fails." + from-offset: option, + /// @nexus.doc "Where to start when from-offset is empty. Unspecified is the beginning." + start: option, + /// @nexus.doc + /// "About this many items per page, 1 to 1000. A Superseded item stays with its Record." + 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 "How long to wait for an uncreated Stream. Empty fails at once. At most 60 s." + create-wait: option, + // Replay needs a fixed range, so the Workflow API omits `last`. + /// @nexus.doc "Read the newest this many Records, oldest first. No from-offset. No wait." + /// @nexus.api-omit targets="workflow" + last: option, + /// @nexus.doc "The maximum Record bytes in a page. A larger Record is still returned alone." + max-bytes: option, + /// @nexus.omit + namespace: placeholder, + // Native only. The SDK Stores refuse to-offset and wake-only. + /// @nexus.omit + to-offset: placeholder, + /// @nexus.omit + wake-only: placeholder, + /// @nexus.omit + floor-from: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.ReadRecordsResponse" + record read-records-response { + records: list, + /// @nexus.doc "The from-offset of the next read, which is the newest offset read + 1." + next-offset: option, + /// @nexus.doc "True when the Stream is closed and this read gave its last item." + done: option, + // Native only. The SDK Stores never set it. + /// @nexus.omit + continue-as-new-suggested: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.DescribeStreamRequest" + record describe-stream-request { + ref: stream-ref, + /// @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 or a Superseded item. Empty for none." + last-offset: option, + retention: 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." + /// @nexus.flatten-in-api param="ref" fields="id backend" keep=true + resource %stream { + constructor(ref: stream-ref, created: option); + + /// @nexus.doc "Publish values, and return the offset of the last one." + /// @nexus.operation "append-records" + /// @nexus.varargs name="values" + /// @nexus.result-field "last-offset" + publish: func(records: list) -> append-records-response; + + /// @nexus.doc "Close the Stream. Readers drain what is left, then end." + /// @nexus.operation "close-stream" + /// @nexus.close + close: func(); + + /// @nexus.doc "Read one page of Records. The poll iterator reads them all." + /// @nexus.operation "read-records" + read: func( + from-offset: option, + start: option, + max-records: option, + wait: option, + create-wait: option, + max-bytes: option, + ) -> read-records-response; + + /// @nexus.doc "Describe the Stream's status, newest offset and Retention." + /// @nexus.operation "describe-stream" + describe: func() -> describe-stream-response; + } + + // With use-existing, created is false when the Stream existed, whatever its Retention. + /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamResponse" + type create-stream-result = own<%stream>; + + /// @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" + append-records: func(request: append-records-request) -> append-records-response; + + /// @nexus.operation name="CloseStream" + 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" typescript="nextReadRequest" + /// dotnet="StreamPoll.NextReadRequest" items="records" python-resume="resume_read_request" + /// targets="client" + read-records: func(request: read-records-request) -> read-records-response; + + /// @nexus.operation name="DescribeStream" + describe-stream: func(request: describe-stream-request) -> describe-stream-response; + + // No operation returns these. A failed call has one in its details. Each SDK raises the class + // of its reason, or the base class for a reason with no class. + /// Why a Stream call failed. Each reason has one gRPC status code. + /// @nexus.proto "temporal.api.enums.v1.StreamErrorReason" + /// @nexus.error details="stream-error-details" field="reason" name="stream-error" + /// python-extends="temporalio.exceptions.ApplicationError" + /// typescript-extends="ApplicationFailure" typescript-extends-import="@temporalio/common" + /// dotnet-extends="Temporalio.Exceptions.ApplicationFailureException" + enum stream-error-reason { + /// @nexus.fallback + stream-error-reason-unspecified, + /// @nexus.error + stream-error-reason-not-found, + /// @nexus.error + stream-error-reason-expired, + /// @nexus.error + stream-error-reason-closed, + /// @nexus.error + stream-error-reason-fenced, + stream-error-reason-sequence-too-low, + /// @nexus.error + stream-error-reason-seq-divergent, + /// @nexus.error + stream-error-reason-not-owner, + stream-error-reason-format-too-new, + /// @nexus.error + stream-error-reason-offset-too-high, + /// @nexus.error + stream-error-reason-too-large, + stream-error-reason-invalid, + /// @nexus.error retryable=true + stream-error-reason-storage-unavailable, + stream-error-reason-unknown-writer, + /// @nexus.error + stream-error-reason-unsupported, + /// @nexus.error + stream-error-reason-already-exists, + stream-error-reason-storage-corrupt, + } + + /// @nexus.doc "The details of a failed Stream call." + /// @nexus.proto "temporal.api.stream.v1.StreamErrorDetails" + record stream-error-details { + reason: option, + } +} diff --git a/temporal/api/enums/v1/stream.proto b/temporal/api/enums/v1/stream.proto new file mode 100644 index 000000000..1a258f97d --- /dev/null +++ b/temporal/api/enums/v1/stream.proto @@ -0,0 +1,83 @@ +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. +// Where a read starts when it has no from_offset. +enum StreamReadStart { + STREAM_READ_START_UNSPECIFIED = 0; + STREAM_READ_START_BEGINNING = 1; + STREAM_READ_START_END = 2; +} + +// 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. An SDK raises its own error +// class for each reason that code can act on, and a base class for the others. +enum StreamErrorReason { + STREAM_ERROR_REASON_UNSPECIFIED = 0; + // Nobody created the Stream, even after `create_wait`. NOT_FOUND. + STREAM_ERROR_REASON_NOT_FOUND = 1; + // The Stream's Retention or safety 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, its Workflow Task was withdrawn, or a + // Workflow retry withdrew its run. 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; + // A create gives another owner than the Stream has, or the owner doesn't allow this close. + // 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 + 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_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; + // No writer has the id. The SDK's in-process writer only. NOT_FOUND. + STREAM_ERROR_REASON_UNKNOWN_WRITER = 13; + // The Store lacks a capability that the call needs. UNIMPLEMENTED. + STREAM_ERROR_REASON_UNSUPPORTED = 14; + // A create with the FAIL conflict policy is for a Stream that exists. ALREADY_EXISTS. + STREAM_ERROR_REASON_ALREADY_EXISTS = 15; + // 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 = 16; +} diff --git a/temporal/api/stream/v1/message.proto b/temporal/api/stream/v1/message.proto new file mode 100644 index 000000000..63bca5f6c --- /dev/null +++ b/temporal/api/stream/v1/message.proto @@ -0,0 +1,161 @@ +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. +message StreamRef { + string id = 1; + // 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. + 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. 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. +message ProducerEpoch { + oneof kind { + // 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. + WorkflowTaskEpoch workflow = 1; + // A Local Activity. A higher epoch withdraws the Producer's lower epochs. + LocalActivityEpoch local_activity = 2; + // An Activity of a Workflow. A higher epoch withdraws the Producer's lower epochs. + ActivityEpoch activity = 3; + // A Standalone Activity. A higher epoch withdraws the Producer's lower epochs. + ActivityEpoch standalone_activity = 4; + // 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. + ClientEpoch client = 5; + } +} + +// Experimental. Might change in the future. +// One Workflow Task attempt of a run. It orders by started_event_id, then attempt, then +// started_time. +message WorkflowTaskEpoch { + int64 started_event_id = 1; + int32 attempt = 2; + // Tells a reused started event id apart, for example after a dropped speculative task. + 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 a Local Activity of a +// later Workflow Task supersedes one of an earlier task. +message LocalActivityEpoch { + // The Workflow Task attempt that dispatched the Local Activity. + WorkflowTaskEpoch task = 1; + // The Local Activity's own attempt in that Workflow Task. + int32 attempt = 2; +} + +// Experimental. Might change in the future. +// The attempt of an Activity. It orders by attempt_scheduled_time, then attempt. +message ActivityEpoch { + // 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; + int32 attempt = 2; +} + +// Experimental. Might change in the future. +// The attempt of a client Producer, which the client picks. +message ClientEpoch { + int32 attempt = 1; +} + +// Experimental. Might change in the future. +// The code that wrote a batch. Its id stays the same across attempts, and its epoch orders them. +message Producer { + // Stable across attempts, such as `wf:` for the Workflow Producer of a run. + string id = 1; + // Required. Its case gives the kind of the Producer. + ProducerEpoch epoch = 2; + // The run of the Producer, so nobody parses the id. Empty for a client. + string run_id = 3; +} + +// 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. +message StreamWriter { + // Required on a write. Its epoch's case picks the withdrawal rule. + Producer producer = 1; + // 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. + WorkflowTaskEpoch previous_completed_task = 2; + // 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. + repeated int64 failed_tasks = 3; + // Retry run only. The previous run, on the retry run's first write. Every Record whose + // Producer carries it is withdrawn. + string withdraw_run = 4; +} + +// 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 { + temporal.api.common.v1.Payload body = 1; +} + +// 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. +message StreamSuperseded { + // The writer of the batch that withdraws. Its Producer's epoch is empty when only + // withdraw_run applies. + StreamWriter writer = 1; +} + +// Experimental. Might change in the future. +// One Record that a Producer appended. +message StreamData { + Producer producer = 1; + temporal.api.common.v1.Payload body = 2; +} + +// 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. +message StreamRecord { + int64 offset = 1; + // A read item is a Record or a Superseded item. A close writes no item. + oneof item { + StreamData data = 2; + // 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. + StreamSuperseded superseded = 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/temporal/api/streamservice/v1/request_response.proto b/temporal/api/streamservice/v1/request_response.proto new file mode 100644 index 000000000..cdd61456e --- /dev/null +++ b/temporal/api/streamservice/v1/request_response.proto @@ -0,0 +1,186 @@ +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 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. --) +message CreateStreamRequest { + string namespace = 1; + // 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. + temporal.api.stream.v1.StreamRef stream_ref = 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 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. + temporal.api.stream.v1.Producer owner = 5; + // 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. --) + bool close_with_run = 6; + // 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. + bytes task_token = 7; + // What the create does when the Stream exists. Unspecified is FAIL. + temporal.api.enums.v1.StreamIdConflictPolicy id_conflict_policy = 8; + // 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 = 9; +} + +// Experimental. Might change in the future. +// The Stream that a create made or got. +message CreateStreamResponse { + // The ref, with the owner of a bound Stream. + 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 { + string namespace = 1; + temporal.api.stream.v1.StreamRef ref = 2; + // Required outside a Workflow. In a Workflow the transport sets it. + temporal.api.stream.v1.StreamWriter writer = 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; + // 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. + bool defer_withdrawal = 6; + // Only read by the server. The server checks the Producer against the Activity's current + // attempt, so nobody else writes as that Activity. + bytes task_token = 7; +} + +// Experimental. Might change in the future. +// Where the batch's Records are in the Stream. +message AppendRecordsResponse { + int64 first_offset = 1; + int64 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. +// Only the owner's run chain closes a bound Stream. Only a Workflow or a client closes a +// standalone Stream. +message CloseStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef ref = 2; + // The writer that closes the Stream. An SDK Store reads only its Producer, and the server + // also reads its withdrawal fields. + temporal.api.stream.v1.StreamWriter writer = 3; + // Only read by the server. It checks the token as on the append. + bytes task_token = 4; +} + +// 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 items from an offset. At the end of the Stream, the read waits for a new +// Record, so readers long poll. +message ReadRecordsRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef ref = 2; + // 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. --) + int64 from_offset = 3; + // Where to start when from_offset is zero. Unspecified is the beginning. + temporal.api.enums.v1.StreamReadStart start = 4; + // 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. + int32 max_records = 5; + // 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 = 6; + // 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 = 7; + // 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. --) + int64 to_offset = 8; + // 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. + bool wake_only = 9; + // 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. + int32 last = 10; + // 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 = 11; + // 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. --) + int64 floor_from = 12; +} + +// Experimental. Might change in the future. +// One page of a read. +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; + // 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. + 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 set by the server. + bool continue_as_new_suggested = 4; +} + +// Experimental. Might change in the future. +// Gets the status, the newest offset and the Retention of a Stream. +message DescribeStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef ref = 2; +} + +// Experimental. Might change in the future. +// The state of a Stream. +message DescribeStreamResponse { + temporal.api.enums.v1.StreamStatus status = 1; + // The newest offset, of a Record or a Superseded item. Zero when the Stream has none. + int64 last_offset = 2; + google.protobuf.Duration retention = 3; +}