From a2acd1d44c5c9530ce26e45943fd8f32ef75254a Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Mon, 26 Oct 2026 22:23:13 -0700 Subject: [PATCH 1/9] Added the Stream service WIT. Each SDK copies it and generates its client with nexgen's client target. SDKs call the writer methods with `runtime.proto` messages, so there's no writer WIT. --- crates/streams/README.md | 3 + crates/streams/nexus/stream-service.wit | 396 ++++++++++++++++++++++++ crates/streams/tests/wit.rs | 14 + 3 files changed, 413 insertions(+) create mode 100644 crates/streams/nexus/stream-service.wit create mode 100644 crates/streams/tests/wit.rs diff --git a/crates/streams/README.md b/crates/streams/README.md index e325f177a..a784ef6ec 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. Each SDK, each Store and the diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit new file mode 100644 index 000000000..8538f15fc --- /dev/null +++ b/crates/streams/nexus/stream-service.wit @@ -0,0 +1,396 @@ +// Draft. The Stream service as a System Nexus service, for the Brook 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.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 after encoding." + 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 { + kind: option, + /// @nexus.doc "Stable across attempts. The proposal gives the form per kind." + id: string, + 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" + record append-record { + /// @nexus.doc "The value. It goes through the payload codec." + body: payload, + } + + /// @nexus.proto "temporal.api.enums.v1.StreamRecordKind" + enum stream-record-kind { + stream-record-kind-unspecified, + stream-record-kind-data, + stream-record-kind-superseded, + stream-record-kind-close, + } + + // 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 Record withdraws." + /// @nexus.proto "temporal.api.stream.v1.Supersession" + record supersession { + producer-kind: option, + producer-id: string, + epoch: option, + /// @nexus.doc "Workflow only. The previous completed Workflow Task of the run." + previous-completed-task: option, + /// @nexus.doc "Workflow only. Started event ids of failed or timed out Workflow Tasks." + failed-tasks: list, + /// @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, + } + + // An offset is the Redis entry id packed as ms << 20 | seq. It 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 Record that a read gives." + /// @nexus.proto "temporal.api.stream.v1.StreamRecord" + record stream-record { + kind: option, + /// @nexus.doc "The place of this Record in the Stream. Ordered and unique, with gaps." + offset: option, + producer: option, + /// @nexus.doc "Set on a data Record." + body: option, + /// @nexus.doc "Set on a SUPERSEDED Record, which the read builds from a flagged batch." + supersession: option, + /// @nexus.doc "Set on a close Record with an outcome." + close-result: option, + close-failure: option, + /// @nexus.doc "Set on a close Record, together with the result or the failure." + close-status: option, + } + + /// @nexus.doc "Create a Stream, or get the one that exists, which keeps its Retention." + /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamRequest" + record create-stream-request { + /// @nexus.doc "Required on Redis and memory. On native, empty asks the server for an id." + stream-id: string, + /// @nexus.doc "The age after which a Record is dropped. Empty is the Store default." + retention: option, + /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws." + withdraw-run: option, + /// @nexus.doc "The Workflow run that owns the Stream. The Workflow transport sets it." + owner: option, + // Native only. Brook's Stores don't read it. + /// @nexus.omit + close-with-run: placeholder, + /// @nexus.doc "The Stream's ref, with the owner fields." + %ref: option, + /// @nexus.omit + namespace: placeholder, + // Native only. Brook's 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: option, + /// @nexus.doc "Required outside a Workflow. In a Workflow the transport sets it." + producer: option, + /// @nexus.doc "The batch number in this epoch, 1 or more. It is part of the dedupe key." + sequence: option, + records: list, + /// @nexus.doc "Local Activity only. The run's last completed Workflow Task." + previous-completed-task: option, + /// @nexus.doc "Local Activity only. The run's failed or timed out Workflow Tasks." + failed-tasks: list, + /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws." + withdraw-run: option, + /// @nexus.omit + namespace: placeholder, + // Native only. Brook's 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: option, + producer: option, + %result: option, + failure: option, + /// @nexus.doc "Empty is completed, or failed with a failure. A caller may set canceled." + status: option, + /// @nexus.doc "Workflow only. The previous completed Workflow Task of the run." + previous-completed-task: option, + /// @nexus.doc "Workflow only. The run's failed or timed out Workflow Tasks." + failed-tasks: list, + /// @nexus.omit + namespace: placeholder, + // Native only. Brook's 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: option, + /// @nexus.doc "Start offset. Empty uses start. INVALID_ARGUMENT 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 "At most this many Records. 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 "How long to wait for an uncreated Stream. Empty fails at once. At most 60 s." + create-wait: option, + /// @nexus.doc "Read the newest this many Records, oldest first. No from-offset. No wait." + 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. Brook's 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. Brook's Stores never set it. + /// @nexus.omit + continue-as-new-suggested: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.DescribeStreamRequest" + record describe-stream-request { + %stream: option, + /// @nexus.omit + namespace: placeholder, + } + + /// @nexus.proto "temporal.api.streamservice.v1.DescribeStreamResponse" + record describe-stream-response { + status: option, + /// @nexus.doc "The offset of the newest Record. Empty when the Stream has no Record." + last-offset: option, + retention: option, + } + + // A handle never changes, since offsets are values that its methods return. Only a create's + // response sets created and status, so a handle built from a ref has neither. + /// @nexus.doc "A handle on one Stream. Build it from a stream-ref, or get it from a create." + resource %stream { + constructor(%stream: stream-ref, created: option, status: option); + + /// @nexus.doc "Append Records. The answer has the first and the last offset." + /// @nexus.operation "append-records" + append-records: func( + producer: option, + sequence: option, + records: list, + previous-completed-task: option, + failed-tasks: list, + withdraw-run: option, + ) -> append-records-response; + + /// @nexus.doc "Close the Stream. Readers drain what is left, then end." + /// @nexus.operation "close-stream" + close: func( + producer: option, + %result: option, + failure: option, + previous-completed-task: option, + failed-tasks: list, + ); + + /// @nexus.doc "Read one page of Records. The poll iterator reads them all." + /// @nexus.operation "read-records" + read-records: func( + from-offset: option, + start: option, + max-records: option, + wait: option, + create-wait: 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. status is closed when an + // earlier create and close used the id. + /// @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" + read-records: func(request: read-records-request) -> read-records-response; + + /// @nexus.operation name="DescribeStream" + describe-stream: func(request: describe-stream-request) -> describe-stream-response; +} 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:?}"); +} From c8d16eb27c389aad25e22361d2080c9d8b82e31e Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Tue, 27 Oct 2026 07:55:05 -0700 Subject: [PATCH 2/9] Added the close status to the stream's close. The close request has a status of its own, an enum apart from the stream's status. As a param, the caller sets it, and nothing fills it from the stream. --- crates/streams/nexus/stream-service.wit | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit index 8538f15fc..7abe8cd9b 100644 --- a/crates/streams/nexus/stream-service.wit +++ b/crates/streams/nexus/stream-service.wit @@ -349,6 +349,7 @@ interface stream-service { producer: option, %result: option, failure: option, + status: option, previous-completed-task: option, failed-tasks: list, ); From 4dbc7572e34e515f809e79da47d06a52a1613150 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 27 Oct 2026 09:49:47 -0700 Subject: [PATCH 3/9] Typed the read item and made each request's Stream required. The read item binds the API's `StreamRecord.item` oneof as a variant, and the error reason binds the new enum. The resource methods take the names users call, and each request lists its required fields first, so the generated models stay positional. --- crates/streams/nexus/stream-service.wit | 118 ++++++++++++++++-------- 1 file changed, 78 insertions(+), 40 deletions(-) diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit index 7abe8cd9b..480e8ddfe 100644 --- a/crates/streams/nexus/stream-service.wit +++ b/crates/streams/nexus/stream-service.wit @@ -89,9 +89,9 @@ interface stream-service { /// @nexus.doc "The Producer of a batch." /// @nexus.proto "temporal.api.stream.v1.Producer" record producer { - kind: option, /// @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, @@ -112,29 +112,21 @@ interface stream-service { body: payload, } - /// @nexus.proto "temporal.api.enums.v1.StreamRecordKind" - enum stream-record-kind { - stream-record-kind-unspecified, - stream-record-kind-data, - stream-record-kind-superseded, - stream-record-kind-close, - } - // 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 Record withdraws." + /// @nexus.doc "Which earlier Records a Superseded item withdraws." /// @nexus.proto "temporal.api.stream.v1.Supersession" record supersession { - producer-kind: option, 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 only. Started event ids of failed or timed out Workflow Tasks." - failed-tasks: list, /// @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." @@ -152,25 +144,43 @@ interface stream-service { 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. It goes through the payload codec." + body: payload, + 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 the Redis entry id packed as ms << 20 | seq. It 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 Record that a read gives." + /// @nexus.doc "One item that a read gives." /// @nexus.proto "temporal.api.stream.v1.StreamRecord" record stream-record { - kind: option, - /// @nexus.doc "The place of this Record in the Stream. Ordered and unique, with gaps." + item: stream-item, + /// @nexus.doc "The place of this item in the Stream. Ordered and unique, with gaps." offset: option, - producer: option, - /// @nexus.doc "Set on a data Record." - body: option, - /// @nexus.doc "Set on a SUPERSEDED Record, which the read builds from a flagged batch." - supersession: option, - /// @nexus.doc "Set on a close Record with an outcome." - close-result: option, - close-failure: option, - /// @nexus.doc "Set on a close Record, together with the result or the failure." - close-status: option, } /// @nexus.doc "Create a Stream, or get the one that exists, which keeps its Retention." @@ -184,11 +194,11 @@ interface stream-service { withdraw-run: option, /// @nexus.doc "The Workflow run that owns the Stream. The Workflow transport sets it." owner: option, + /// @nexus.doc "The Stream's ref, with the owner fields." + %ref: option, // Native only. Brook's Stores don't read it. /// @nexus.omit close-with-run: placeholder, - /// @nexus.doc "The Stream's ref, with the owner fields." - %ref: option, /// @nexus.omit namespace: placeholder, // Native only. Brook's Stores don't read it. @@ -210,16 +220,16 @@ interface stream-service { /// "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: option, + %stream: stream-ref, + records: list, + /// @nexus.doc "Local Activity only. The run's failed or timed out Workflow Tasks." + failed-tasks: list, /// @nexus.doc "Required outside a Workflow. In a Workflow the transport sets it." producer: option, /// @nexus.doc "The batch number in this epoch, 1 or more. It is part of the dedupe key." sequence: option, - records: list, /// @nexus.doc "Local Activity only. The run's last completed Workflow Task." previous-completed-task: option, - /// @nexus.doc "Local Activity only. The run's failed or timed out Workflow Tasks." - failed-tasks: list, /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws." withdraw-run: option, /// @nexus.omit @@ -240,7 +250,9 @@ interface stream-service { /// @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: option, + %stream: stream-ref, + /// @nexus.doc "Workflow only. The run's failed or timed out Workflow Tasks." + failed-tasks: list, producer: option, %result: option, failure: option, @@ -248,8 +260,6 @@ interface stream-service { status: option, /// @nexus.doc "Workflow only. The previous completed Workflow Task of the run." previous-completed-task: option, - /// @nexus.doc "Workflow only. The run's failed or timed out Workflow Tasks." - failed-tasks: list, /// @nexus.omit namespace: placeholder, // Native only. Brook's Stores don't read them. @@ -273,7 +283,7 @@ interface stream-service { /// @nexus.doc "Read Records from an offset." /// @nexus.proto "temporal.api.streamservice.v1.ReadRecordsRequest" record read-records-request { - %stream: option, + %stream: stream-ref, /// @nexus.doc "Start offset. Empty uses start. INVALID_ARGUMENT past the newest offset + 1." from-offset: option, /// @nexus.doc "Where to start when from-offset is empty. Unspecified is the beginning." @@ -313,7 +323,7 @@ interface stream-service { /// @nexus.proto "temporal.api.streamservice.v1.DescribeStreamRequest" record describe-stream-request { - %stream: option, + %stream: stream-ref, /// @nexus.omit namespace: placeholder, } @@ -332,9 +342,9 @@ interface stream-service { resource %stream { constructor(%stream: stream-ref, created: option, status: option); - /// @nexus.doc "Append Records. The answer has the first and the last offset." + /// @nexus.doc "Publish Records. The answer has the first and the last offset." /// @nexus.operation "append-records" - append-records: func( + publish: func( producer: option, sequence: option, records: list, @@ -356,7 +366,7 @@ interface stream-service { /// @nexus.doc "Read one page of Records. The poll iterator reads them all." /// @nexus.operation "read-records" - read-records: func( + read: func( from-offset: option, start: option, max-records: option, @@ -394,4 +404,32 @@ interface stream-service { /// @nexus.operation name="DescribeStream" describe-stream: func(request: describe-stream-request) -> describe-stream-response; + + // No operation returns these. A failed call carries them, so each SDK can raise one error type + // per reason. + /// @nexus.doc "Why a Stream call failed. Each reason has one gRPC status code." + /// @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, + stream-error-reason-seq-divergent, + stream-error-reason-not-owner, + stream-error-reason-format-too-new, + stream-error-reason-offset, + stream-error-reason-too-large, + stream-error-reason-invalid, + stream-error-reason-storage, + stream-error-reason-unknown-writer, + 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, + } } From 76f84a2a759c0f2e52415213bd04011a56a356ce Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 28 Oct 2026 00:30:22 -0700 Subject: [PATCH 4/9] Matched the Stream service WIT to the create and read fixes. A create of a closed, expired or other-owned Stream fails, so the handle no longer takes a create status. The read method takes `last` and `max-bytes`, which the request has. --- crates/streams/nexus/stream-service.wit | 29 +++++++++++++------------ 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit index 480e8ddfe..7d47c42e6 100644 --- a/crates/streams/nexus/stream-service.wit +++ b/crates/streams/nexus/stream-service.wit @@ -30,7 +30,7 @@ interface stream-service { /// @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 after encoding." + /// @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, @@ -172,9 +172,8 @@ interface stream-service { close(stream-close), } - // An offset is the Redis entry id packed as ms << 20 | seq. It 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. + // 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 { @@ -183,10 +182,11 @@ interface stream-service { offset: option, } - /// @nexus.doc "Create a Stream, or get the one that exists, which keeps its Retention." + /// @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 "Required on Redis and memory. On native, empty asks the server for an id." + /// @nexus.doc "The Stream id. Required, and never empty." stream-id: string, /// @nexus.doc "The age after which a Record is dropped. Empty is the Store default." retention: option, @@ -194,7 +194,7 @@ interface stream-service { withdraw-run: option, /// @nexus.doc "The Workflow run that owns the Stream. The Workflow transport sets it." owner: option, - /// @nexus.doc "The Stream's ref, with the owner fields." + /// @nexus.doc "The ref, whose id is stream-id. A create with an owner needs the ref's owner." %ref: option, // Native only. Brook's Stores don't read it. /// @nexus.omit @@ -284,11 +284,11 @@ interface stream-service { /// @nexus.proto "temporal.api.streamservice.v1.ReadRecordsRequest" record read-records-request { %stream: stream-ref, - /// @nexus.doc "Start offset. Empty uses start. INVALID_ARGUMENT past the newest offset + 1." + /// @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 "At most this many Records. 1 to 1000." + /// @nexus.doc "At most this many items, Superseded items included. 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, @@ -331,16 +331,16 @@ interface stream-service { /// @nexus.proto "temporal.api.streamservice.v1.DescribeStreamResponse" record describe-stream-response { status: option, - /// @nexus.doc "The offset of the newest Record. Empty when the Stream has no Record." + /// @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 and status, so a handle built from a ref has neither. + // 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." resource %stream { - constructor(%stream: stream-ref, created: option, status: option); + constructor(%stream: stream-ref, created: option); /// @nexus.doc "Publish Records. The answer has the first and the last offset." /// @nexus.operation "append-records" @@ -372,6 +372,8 @@ interface stream-service { max-records: option, wait: option, create-wait: option, + last: option, + max-bytes: option, ) -> read-records-response; /// @nexus.doc "Describe the Stream's status, newest offset and Retention." @@ -379,8 +381,7 @@ interface stream-service { describe: func() -> describe-stream-response; } - // created is false when the Stream existed, whatever its Retention. status is closed when an - // earlier create and close used the id. + // created is false when the Stream existed, whatever its Retention. /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamResponse" type create-stream-result = own<%stream>; From bb353352c2a82e4eb542f162cbce17faa68cd712 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 28 Oct 2026 15:57:39 -0700 Subject: [PATCH 5/9] Added the nexgen directives to the Stream service WIT. The generated Stream API is the user API. The directives hide the fields that the runtime sets, make the Stream generic in its value type, and shape publish, read and close. A read item is optional, so a reader can skip an item kind that it doesn't know. --- crates/streams/nexus/stream-service.wit | 95 ++++++++++++++++++------- 1 file changed, 71 insertions(+), 24 deletions(-) diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit index 7d47c42e6..e316fd665 100644 --- a/crates/streams/nexus/stream-service.wit +++ b/crates/streams/nexus/stream-service.wit @@ -19,6 +19,10 @@ interface stream-service { 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, @@ -107,9 +111,10 @@ interface stream-service { /// @nexus.doc "One Record to append." /// @nexus.proto "temporal.api.stream.v1.AppendRecord" + /// @nexus.unwrap record append-record { /// @nexus.doc "The value. It goes through the payload codec." - body: payload, + body: body-t, } // An Activity or Local Activity item withdraws the Records of its Producer with a lower epoch. A @@ -148,7 +153,7 @@ interface stream-service { /// @nexus.proto "temporal.api.stream.v1.StreamData" record stream-data { /// @nexus.doc "The value. It goes through the payload codec." - body: payload, + body: body-t, producer: option, } @@ -177,7 +182,8 @@ interface stream-service { /// @nexus.doc "One item that a read gives." /// @nexus.proto "temporal.api.stream.v1.StreamRecord" record stream-record { - item: stream-item, + // 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, } @@ -187,14 +193,23 @@ interface stream-service { /// @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, /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws." + /// @nexus.api-omit 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. Brook's Stores don't read it. /// @nexus.omit @@ -223,14 +238,28 @@ interface stream-service { %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, /// @nexus.doc "The batch number in this epoch, 1 or more. It is part of the dedupe key." + /// @nexus.api-omit 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, /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws." + /// @nexus.api-omit withdraw-run: option, /// @nexus.omit namespace: placeholder, @@ -252,13 +281,22 @@ interface stream-service { record close-stream-request { %stream: stream-ref, /// @nexus.doc "Workflow only. The run's failed or timed out Workflow Tasks." + /// @nexus.api-omit failed-tasks: list, + /// @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, failure: option, - /// @nexus.doc "Empty is completed, or failed with a failure. A caller may set canceled." + // The service sets canceled for a failure with canceled_failure_info, so the API omits status. + /// @nexus.doc + /// "Empty is completed, or failed with a failure. With a canceled failure, it is canceled." + /// @nexus.api-omit status: option, /// @nexus.doc "Workflow only. The previous completed Workflow Task of the run." + /// @nexus.api-omit previous-completed-task: option, /// @nexus.omit namespace: placeholder, @@ -339,30 +377,24 @@ interface stream-service { // 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 Records. The answer has the first and the last offset." + /// @nexus.doc "Publish values, and return the offset of the last one." /// @nexus.operation "append-records" - publish: func( - producer: option, - sequence: option, - records: list, - previous-completed-task: option, - failed-tasks: list, - withdraw-run: option, - ) -> append-records-response; + /// @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" - close: func( - producer: option, - %result: option, - failure: option, - status: option, - previous-completed-task: option, - failed-tasks: list, - ); + /// @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" @@ -400,31 +432,46 @@ interface stream-service { /// @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" + /// @nexus.poll python="next_read_request" 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 carries them, so each SDK can raise one error type - // per reason. + // 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" base="stream-error" + /// python-base="temporalio.exceptions.ApplicationError" + /// typescript-base="ApplicationFailure" typescript-base-import="@temporalio/common" + /// dotnet-base="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, } From 88adfdf216804417a2db6b06c60f305f0de27314 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 28 Oct 2026 21:52:59 -0700 Subject: [PATCH 6/9] Kept `last` out of Workflow reads and renamed the error base keys. Replay needs a fixed range, so the Workflow API omits `last`, and the resource `read` no longer takes it. nexgen renamed the error enum's `base` keys to `name` and `extends`. The `max-records` doc says that a Superseded item stays with its Record, as the proto does. --- crates/streams/nexus/stream-service.wit | 34 +++++++++++++------------ 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit index e316fd665..8aec7dfee 100644 --- a/crates/streams/nexus/stream-service.wit +++ b/crates/streams/nexus/stream-service.wit @@ -1,5 +1,5 @@ -// Draft. The Stream service as a System Nexus service, for the Brook API proposal. Each record maps -// to a message of the protos under ../protos/temporal/api (stream/v1, streamservice/v1 and the +// 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; @@ -113,7 +113,7 @@ interface stream-service { /// @nexus.proto "temporal.api.stream.v1.AppendRecord" /// @nexus.unwrap record append-record { - /// @nexus.doc "The value. It goes through the payload codec." + /// @nexus.doc "The value." body: body-t, } @@ -152,7 +152,7 @@ interface stream-service { /// @nexus.doc "One Record that a Producer appended." /// @nexus.proto "temporal.api.stream.v1.StreamData" record stream-data { - /// @nexus.doc "The value. It goes through the payload codec." + /// @nexus.doc "The value." body: body-t, producer: option, } @@ -211,12 +211,12 @@ interface stream-service { /// @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. Brook's Stores don't read it. + // Native only. The SDK Stores don't read it. /// @nexus.omit close-with-run: placeholder, /// @nexus.omit namespace: placeholder, - // Native only. Brook's Stores don't read it. + // Native only. The SDK Stores don't read it. /// @nexus.omit task-token: placeholder, } @@ -263,7 +263,7 @@ interface stream-service { withdraw-run: option, /// @nexus.omit namespace: placeholder, - // Native only. Brook's Stores don't read them. + // Native only. The SDK Stores don't read them. /// @nexus.omit defer-withdrawal: placeholder, /// @nexus.omit @@ -300,7 +300,7 @@ interface stream-service { previous-completed-task: option, /// @nexus.omit namespace: placeholder, - // Native only. Brook's Stores don't read them. + // Native only. The SDK Stores don't read them. /// @nexus.omit withdraw-run: placeholder, /// @nexus.omit @@ -326,19 +326,22 @@ interface stream-service { from-offset: option, /// @nexus.doc "Where to start when from-offset is empty. Unspecified is the beginning." start: option, - /// @nexus.doc "At most this many items, Superseded items included. 1 to 1000." + /// @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. Brook's Stores refuse to-offset and wake-only. + // Native only. The SDK Stores refuse to-offset and wake-only. /// @nexus.omit to-offset: placeholder, /// @nexus.omit @@ -354,7 +357,7 @@ interface stream-service { next-offset: option, /// @nexus.doc "True after the close Record. No later read gives a Record." done: option, - // Native only. Brook's Stores never set it. + // Native only. The SDK Stores never set it. /// @nexus.omit continue-as-new-suggested: placeholder, } @@ -404,7 +407,6 @@ interface stream-service { max-records: option, wait: option, create-wait: option, - last: option, max-bytes: option, ) -> read-records-response; @@ -443,10 +445,10 @@ interface stream-service { // 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" base="stream-error" - /// python-base="temporalio.exceptions.ApplicationError" - /// typescript-base="ApplicationFailure" typescript-base-import="@temporalio/common" - /// dotnet-base="Temporalio.Exceptions.ApplicationFailureException" + /// @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 From 3b8ac8beb38cafba3fa36d61b729e70172d269c0 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 30 Oct 2026 00:34:38 -0700 Subject: [PATCH 7/9] Omitted the close task fields from the WIT, as native only. The SDK Stores ignore failed-tasks and previous-completed-task on a close, as they do withdraw-run. --- crates/streams/nexus/stream-service.wit | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit index 8aec7dfee..ad7d3a693 100644 --- a/crates/streams/nexus/stream-service.wit +++ b/crates/streams/nexus/stream-service.wit @@ -280,9 +280,9 @@ interface stream-service { /// @nexus.proto "temporal.api.streamservice.v1.CloseStreamRequest" record close-stream-request { %stream: stream-ref, - /// @nexus.doc "Workflow only. The run's failed or timed out Workflow Tasks." - /// @nexus.api-omit - failed-tasks: list, + // 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()" @@ -295,9 +295,9 @@ interface stream-service { /// "Empty is completed, or failed with a failure. With a canceled failure, it is canceled." /// @nexus.api-omit status: option, - /// @nexus.doc "Workflow only. The previous completed Workflow Task of the run." - /// @nexus.api-omit - previous-completed-task: 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. From 9dc337d66b3c02f260779411f3faa166ce832647 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 29 Oct 2026 09:11:55 -0700 Subject: [PATCH 8/9] Added the TypeScript and .NET helpers to the read poll. nexgen's TypeScript and .NET client targets build the read iterator from these support helpers. --- crates/streams/nexus/stream-service.wit | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit index ad7d3a693..d894875ee 100644 --- a/crates/streams/nexus/stream-service.wit +++ b/crates/streams/nexus/stream-service.wit @@ -434,7 +434,8 @@ interface stream-service { /// @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" + /// @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; From e0d0155ce6fac544370d7dd3a9201f672a226264 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 30 Oct 2026 10:33:41 -0700 Subject: [PATCH 9/9] Made sequence a parameter and withdraw-run a source in the WIT. The close drops status, which the failure gives. An append with no sequence gets the next one from the Store, so publish works. --- crates/streams/nexus/stream-service.wit | 28 +++++++++++++++++-------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/crates/streams/nexus/stream-service.wit b/crates/streams/nexus/stream-service.wit index d894875ee..921994372 100644 --- a/crates/streams/nexus/stream-service.wit +++ b/crates/streams/nexus/stream-service.wit @@ -200,8 +200,13 @@ interface stream-service { 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.api-omit + /// @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()" @@ -249,8 +254,10 @@ interface stream-service { /// @nexus.source target="client" python="client_producer()" typescript="clientProducer()" /// dotnet="StreamActivityContext.Producer()" producer: option, - /// @nexus.doc "The batch number in this epoch, 1 or more. It is part of the dedupe key." - /// @nexus.api-omit + // 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" @@ -258,8 +265,13 @@ interface stream-service { /// 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.api-omit + /// @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, @@ -289,12 +301,10 @@ interface stream-service { /// 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, - // The service sets canceled for a failure with canceled_failure_info, so the API omits status. - /// @nexus.doc - /// "Empty is completed, or failed with a failure. With a canceled failure, it is canceled." - /// @nexus.api-omit - status: option, // Native only. The SDK Stores don't read it. /// @nexus.omit previous-completed-task: placeholder,