diff --git a/crates/streams/README.md b/crates/streams/README.md index 2c4930d15..35000effa 100644 --- a/crates/streams/README.md +++ b/crates/streams/README.md @@ -15,6 +15,9 @@ Workflow makes each Stream call from a Local Activity. Core has no Stream code. needs no file outside itself. - `protos/temporal/sdk/streams/v1/runtime.proto` has the crate's own calls, such as the writer calls. +- `nexus/stream-service.wit` is the Stream service as a System Nexus service, with the messages of + the protos above. Each SDK copies it and generates its bindings with nexgen. A test checks that + every line is at most 100 columns. - `testdata/*.json` are the shared vectors. They cover Stream ids, Producer ids, Store keys and meta fields, offsets, epoch keys, the withdrawal rule and the batch digest of a dedupe entry. `testdata/gen_vectors.py` writes them from a second implementation, and the crate's tests read diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit new file mode 100644 index 000000000..921994372 --- /dev/null +++ b/crates/streams/nexus/stream-service.wit @@ -0,0 +1,496 @@ +// Draft. The Stream service as a System Nexus service, for the Streams API proposal. Each record +// maps to a message of the protos under ../protos/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, + failure, + payload, + placeholder, + }; + + /// @nexus.doc "The value type of a Stream's Records. A Stream definition sets it." + /// @nexus.type-parameter + type body-t = placeholder; + + /// @nexus.proto "google.protobuf.Timestamp" + record timestamp { + seconds: option, + nanos: option, + } + + // A Stream that a Workflow creates has the id ::. Each part is + // percent-encoded. So a caller that knows the first run id can build the id with no call. + /// @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 owning Workflow run chain or Standalone Activity. Empty on a client ref." + 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 { + 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.proto "temporal.api.enums.v1.StreamProducerKind" + enum stream-producer-kind { + stream-producer-kind-unspecified, + stream-producer-kind-workflow, + stream-producer-kind-local-activity, + stream-producer-kind-activity, + stream-producer-kind-client, + stream-producer-kind-standalone-activity, + } + + // Epochs compare field by field, in the order of the fields. A field that a kind does not use is + // empty and compares as zero. A timestamp compares seconds, then nanos. A Workflow epoch has + // task-started-event-id, task-attempt and task-started-time. A Local Activity epoch has the same + // three of the dispatching Workflow Task, then attempt. An Activity epoch has + // attempt-scheduled-time, then attempt. A client epoch has attempt. + /// @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 { + /// @nexus.doc "The started event id of the Workflow Task." + task-started-event-id: option, + /// @nexus.doc "The attempt of the Workflow Task." + task-attempt: option, + /// @nexus.doc "When the Workflow Task attempt started. It tells a reused started id apart." + task-started-time: option, + /// @nexus.doc "When the server scheduled this Activity attempt. A reset makes it later." + attempt-scheduled-time: option, + /// @nexus.doc "The Activity, Local Activity or client attempt." + attempt: option, + } + + /// @nexus.doc "The Producer of a batch." + /// @nexus.proto "temporal.api.stream.v1.Producer" + record producer { + /// @nexus.doc "Stable across attempts. The proposal gives the form per kind." + id: string, + kind: option, + epoch: option, + /// @nexus.doc "The run of the Producer, so nobody parses the id. Empty for a client." + 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, + started-time: 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, + } + + // An Activity or Local Activity item withdraws the Records of its Producer with a lower epoch. A + // Workflow item, with this task t in epoch and the previous completed task c, withdraws the + // Records of the run and of its Local Activities whose task-started-event-id 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. + /// @nexus.doc "Which earlier Records a Superseded item withdraws." + /// @nexus.proto "temporal.api.stream.v1.Supersession" + record supersession { + producer-id: string, + /// @nexus.doc "Workflow only. Started event ids of failed or timed out Workflow Tasks." + failed-tasks: list, + producer-kind: option, + epoch: option, + /// @nexus.doc "Workflow only. The previous completed Workflow Task of the run." + previous-completed-task: option, + /// @nexus.doc "Workflow retry only. The previous run, all of whose Records are withdrawn." + withdraw-run: option, + /// @nexus.doc "The withdrawing Producer's run, so no Reader parses ids. Empty for a client." + run-id: option, + } + + /// @nexus.doc "How a Stream closed. Only the owner's end on the server terminates or times out." + /// @nexus.proto "temporal.api.enums.v1.StreamCloseStatus" + enum stream-close-status { + stream-close-status-unspecified, + stream-close-status-completed, + stream-close-status-failed, + stream-close-status-canceled, + stream-close-status-terminated, + stream-close-status-timed-out, + } + + /// @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, + } + + /// @nexus.doc "The close of a Stream. A read gives it last." + /// @nexus.proto "temporal.api.stream.v1.StreamClose" + record stream-close { + /// @nexus.doc "The Producer that closed the Stream." + producer: option, + %result: option, + failure: option, + /// @nexus.doc "Set together with the result or the failure." + status: option, + } + + // The read builds a Superseded item from a flagged batch. + /// @nexus.doc "A read item: a Record, a Superseded item or the close." + /// @nexus.proto "temporal.api.stream.v1.StreamRecord.item" + variant stream-item { + data(stream-data), + superseded(supersession), + close(stream-close), + } + + // An offset is ordered and unique in one Stream, and sparse. Offsets have gaps, so they don't + // count Records. A read names 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, or get the open one. A closed, expired or another owner's Stream fails." + /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamRequest" + record create-stream-request { + /// @nexus.doc "The Stream id. Required, and never empty." + /// @nexus.name python="id" typescript="id" dotnet="Id" + /// @nexus.source target="workflow" python="workflow_stream_id(definition)" + /// typescript="workflowStreamId(definition)" + /// dotnet="StreamWorkflowContext.StreamId(definition)" + stream-id: string, + /// @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 target="workflow" python="withdraw_run()" typescript="withdrawRun()" + /// dotnet="StreamWorkflowContext.WithdrawRun()" + /// @nexus.source target="client" python="withdraw_run()" typescript="withdrawRun()" + /// dotnet="StreamWorkflowContext.WithdrawRun()" + withdraw-run: option, + /// @nexus.doc "The Workflow run that owns the Stream. The Workflow transport sets it." + /// @nexus.source target="workflow" python="workflow_owner()" typescript="workflowOwner()" + /// dotnet="StreamWorkflowContext.Owner()" + /// @nexus.api-omit target="client" + owner: option, + /// @nexus.doc "The ref, whose id is stream-id. A create with an owner needs the ref's owner." + /// @nexus.flatten-in-api fields="backend" fill="id=stream-id" + %ref: 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, + } + + /// @nexus.proto "temporal.api.enums.v1.StreamStatus" + enum stream-status { + stream-status-unspecified, + stream-status-open, + stream-status-closed, + stream-status-expired, + } + + // The meta key of each Stream has a format version v. Lua refuses a write to a Stream with a + // higher v with FAILED_PRECONDITION and the detail stream_format_too_new. + /// @nexus.doc + /// "Append. FAILED_PRECONDITION if closed, fenced or format too new. NOT_FOUND if absent." + /// @nexus.proto "temporal.api.streamservice.v1.AppendRecordsRequest" + record append-records-request { + %stream: stream-ref, + records: list, + /// @nexus.doc "Local Activity only. The run's failed or timed out Workflow Tasks." + /// @nexus.api-omit target="workflow" + /// @nexus.source target="client" python="local_activity_failed_tasks()" + /// typescript="localActivityFailedTasks()" + /// dotnet="StreamActivityContext.LocalActivityFailedTasks()" + failed-tasks: list, + /// @nexus.doc "Required outside a Workflow. In a Workflow the transport sets it." + /// @nexus.source target="workflow" python="workflow_producer()" + /// typescript="workflowProducer()" dotnet="StreamWorkflowContext.Producer()" + /// @nexus.source target="client" python="client_producer()" typescript="clientProducer()" + /// dotnet="StreamActivityContext.Producer()" + producer: 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.doc "Local Activity only. The run's last completed Workflow Task." + /// @nexus.api-omit target="workflow" + /// @nexus.source target="client" python="local_activity_previous_completed_task()" + /// typescript="localActivityPreviousCompletedTask()" + /// dotnet="StreamActivityContext.LocalActivityPreviousCompletedTask()" + previous-completed-task: 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 target="workflow" python="withdraw_run()" typescript="withdrawRun()" + /// dotnet="StreamWorkflowContext.WithdrawRun()" + /// @nexus.source target="client" python="withdraw_run()" typescript="withdrawRun()" + /// dotnet="StreamWorkflowContext.WithdrawRun()" + withdraw-run: 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, + } + + /// @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 { + %stream: stream-ref, + // Native only. The SDK Stores don't read it. + /// @nexus.omit + failed-tasks: placeholder, + /// @nexus.source target="workflow" python="workflow_producer()" + /// typescript="workflowProducer()" dotnet="StreamWorkflowContext.Producer()" + /// @nexus.source target="client" python="client_producer()" typescript="clientProducer()" + /// dotnet="StreamActivityContext.Producer()" + producer: option, + %result: option, + // Each SDK gives a canceled Activity, child Workflow or Nexus Operation as a failure of that + // call whose cause is the cancel. So such a failure also cancels. + /// @nexus.doc "Empty completes the Stream. A cancel failure cancels it. Any other fails it." + failure: option, + // Native only. The SDK Stores don't read it. + /// @nexus.omit + previous-completed-task: placeholder, + /// @nexus.omit + namespace: placeholder, + // Native only. The SDK Stores don't read them. + /// @nexus.omit + withdraw-run: placeholder, + /// @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 { + %stream: stream-ref, + /// @nexus.doc "Start offset. Empty uses start. OUT_OF_RANGE past the newest offset + 1." + 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 target="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 after the close Record. No later read gives a Record." + 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 { + %stream: 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, a Superseded item or the close. Empty for none." + last-offset: option, + retention: option, + } + + // A handle never changes, since offsets are values that its methods return. Only a create's + // response sets created, so a handle built from a ref has none. + /// @nexus.doc "A handle on one Stream. Build it from a stream-ref, or get it from a create." + /// @nexus.type-parameter "body-t" + /// @nexus.definition python="temporalio.common.StreamDefinition" + /// typescript="StreamDefinition" typescript-import="@temporalio/common" + /// dotnet="Temporalio.Common.StreamDefinition" + /// @nexus.flatten-in-api param="stream" fields="id backend" keep=true + resource %stream { + constructor(%stream: 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 failure="failure" + close: func(%result: option, failure: option); + + /// @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; + } + + // 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" + /// target="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. + /// @nexus.doc "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 { + 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, + /// @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, + /// @nexus.error + stream-error-reason-too-large, + stream-error-reason-invalid, + /// @nexus.error retryable=true + stream-error-reason-storage, + stream-error-reason-unknown-writer, + /// @nexus.error + stream-error-reason-unsupported, + } + + /// @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/crates/streams/tests/wit.rs b/crates/streams/tests/wit.rs new file mode 100644 index 000000000..a7cb3b4de --- /dev/null +++ b/crates/streams/tests/wit.rs @@ -0,0 +1,14 @@ +//! The Stream service WIT that each SDK copies and generates its bindings from. + +const WIT: &str = include_str!("../nexus/stream-service.wit"); + +#[test] +fn every_wit_line_fits_100_columns() { + let long: Vec<_> = WIT + .lines() + .enumerate() + .filter(|(_, line)| line.chars().count() > 100) + .map(|(n, _)| n + 1) + .collect(); + assert!(long.is_empty(), "lines over 100 columns: {long:?}"); +}