diff --git a/crates/protos/build.rs b/crates/protos/build.rs index 55d4a24e39..fb94ca41d9 100644 --- a/crates/protos/build.rs +++ b/crates/protos/build.rs @@ -59,6 +59,7 @@ fn main() -> Result<(), Box> { let protos = &[ "./protos/local/temporal/sdk/core/core_interface.proto", "./protos/local/temporal/sdk/core/streams/streams.proto", + "./protos/local/temporal/sdk/core/streams/nexus.proto", "./protos/api_upstream/temporal/api/sdk/v1/workflow_metadata.proto", "./protos/api_upstream/temporal/api/workflowservice/v1/service.proto", "./protos/api_upstream/temporal/api/nexusservices/workerservice/v1/request_response.proto", diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto b/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto new file mode 100644 index 0000000000..ffdaa94bba --- /dev/null +++ b/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto @@ -0,0 +1,93 @@ +syntax = "proto3"; + +package coresdk.streams; +option ruby_package = "Temporalio::Internal::Bridge::Api::Streams"; + +import "google/protobuf/timestamp.proto"; +import "temporal/api/common/v1/message.proto"; +import "temporal/api/failure/v1/message.proto"; + +// The stream service's Nexus operations, served by Core. Lang forwards each call's input in the +// N0a contract's JSON transfer form, runs its payload codec when Core asks for it, and hands the +// failure Core answers with to its Nexus handler. The contract is +// `crates/streams/nexus/temporal_streams.nexusrpc.yaml`. + +// The calls Bridge adds beside `StreamService`, dispatched by name through the same entry. +service StreamNexusService { + // The stream service's Nexus `append` operation, over its JSON transfer form. + rpc NexusAppend(NexusCall) returns (NexusStep); + // The stream service's Nexus `read` operation, over its JSON transfer form. + rpc NexusRead(NexusCall) returns (NexusStep); + // Goes on with a call that asked for codec work. + rpc NexusResume(NexusResumeRequest) returns (NexusStep); +} + +// One Nexus operation call. +message NexusCall { + // The operation's input, the N0a model in its JSON transfer form. + bytes input = 1; + // When the Nexus request times out. Core answers a read in time for its answer to travel back. + google.protobuf.Timestamp deadline = 2; + NexusLimits limits = 3; + // The namespace of the lang Worker's client, which streams the call names live in. + string namespace = 4; +} + +message NexusLimits { + // The most bytes one read answer takes as JSON past its first record, with its tokens and state. + // Zero takes the default of 1 MiB, which keeps an answer below the server's blob size limit. + uint64 max_answer_bytes = 1; + // The most bytes one record may take in an answer. A record above it can never be read, so an + // append with one is refused. Zero takes the default of 1.5 MiB. + uint64 max_record_bytes = 2; +} + +// What Core answers a call or a resume with. A codec step comes before the output on a call whose +// bodies need lang's payload codec, and lang resumes it with the results. +message NexusStep { + oneof step { + // The operation's output, the N0a model in its JSON transfer form. + bytes output = 1; + CodecWork codec = 2; + NexusFailure failure = 3; + } +} + +// Bodies for lang's payload codec. Lang runs the namespace's codec over every payload, in order, +// and resumes with one result per payload. +message CodecWork { + // True encodes the bodies of an append, and false decodes the bodies of a read. + bool encode = 1; + // The stream's namespace and owner, for the codec's serialization context. + string namespace = 2; + string workflow_id = 3; + repeated temporal.api.common.v1.Payload payloads = 4; + // Hand back in `NexusResumeRequest`. Core forgets a continuation 60 s after it asked. + uint64 continuation = 5; +} + +message NexusResumeRequest { + uint64 continuation = 1; + // One per payload of the `CodecWork`, in order. + repeated CodecResult results = 2; +} + +message CodecResult { + oneof result { + temporal.api.common.v1.Payload payload = 1; + // Why the codec failed on this payload. A read hands the record over as an error at its + // cursor, and an append fails. + string error = 2; + } +} + +// How a Nexus operation failed, ready for lang's Nexus handler error. +message NexusFailure { + // The Nexus `HandlerErrorType` name, such as `BAD_REQUEST`. + string handler_error_type = 1; + bool retryable = 2; + string message = 3; + // The failure that crosses to the caller, with the stream condition's class as its cause's + // application error type. + temporal.api.failure.v1.Failure failure = 4; +} diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index 8b8f74ba24..c7a505c1bf 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -769,8 +769,8 @@ enum StreamFailureKind { // This release or this store does not offer the capability. STREAM_FAILURE_KIND_UNSUPPORTED = 11; // The call's input breaks the contract, such as a missing member or a value out of range, so - // the same call fails the same way. A lang built before this kind maps it, as any kind it - // doesn't know, to a storage failure. + // the same call fails the same way. The Nexus calls answer it, as BAD_REQUEST. A lang built + // before this kind maps it, as any kind it doesn't know, to a storage failure. STREAM_FAILURE_KIND_INVALID_ARGUMENT = 12; // A replay read found other records than the read it repeats: other content, another count // or another end. Lang reports it as nondeterminism. diff --git a/crates/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml new file mode 100644 index 0000000000..e4ed1e5811 --- /dev/null +++ b/crates/streams/nexus/temporal_streams.nexusrpc.yaml @@ -0,0 +1,309 @@ +# The stream service's wire contract. Core serves it, so this copy in Core is +# the canonical one. Every SDK generates its models, its forwarding handler and +# its HTTP caller from the copy in the Core it vendors. Core's serde types are +# checked against it by tests/nexus_contract.rs. +# +# Inputs reject unknown members so a typo fails at the handler. Outputs accept +# them so a handler can add a member before every caller is rebuilt. +nexusrpc: "1.0.0" + +services: + TemporalStreams: + fqn: temporal.sdk.streams.v1.TemporalStreams + description: >- + Appends to and reads from streams through one Nexus endpoint, so the + store behind it stays the endpoint's business and an operator can change + it without touching callers. An append carries one batch and a read + answers with a batch, because a Nexus operation per record costs too + much for token streams. A record crosses as the serialized + temporal.sdk.streams.v1.StreamRecord, the bytes every store keeps, so a + caller in any language decodes it with that proto alone. The generated + HTTP caller runs no payload codec, so it suits a namespace whose Workers + use none. A Workflow caller goes through its Worker, whose codec applies. + x-nexus-handle: + StreamHandle: [stream] + StreamProducer: [stream, producer_id, attempt] + operations: + append: + fqn: append + description: >- + Write one batch as one producer attempt, or finish the producer. The + store numbers the batch from sequence and deduplicates a retry: a + repeat of the producer attempt's newest batch with the same content + answers with where the original landed and writes nothing. A + sequence already used with other content, or below the newest one + the store holds for the attempt, is refused. Supersession records + are never written: a reader synthesizes them from the attempts it + observes. + input: { $ref: "#/$defs/AppendInput" } + output: { $ref: "#/$defs/AppendOutput" } + read: + fqn: read + description: >- + Answer with the records after the caller's cursor. The call waits + until it has max_records records or wait_ms passes, whichever comes + first, and answers with what it collected. It says when the stream + has ended for the reader, so the caller can stop. + x-nexus-long-poll: + wait-field: wait_ms + result-field: records + done-field: done + state-field: state + input: { $ref: "#/$defs/ReadInput" } + output: { $ref: "#/$defs/ReadOutput" } + +$defs: + StreamRef: + type: object + description: >- + One stream, named by its owner and a topic. It names no cursor and no + store, so the same reference is good behind any endpoint that serves the + owner. + x-nexus-stream-ref: { py: temporalio.contrib.streams.StreamRef } + properties: + kind: + type: string + description: >- + What owns the stream. Only "workflow" is opened in this release, and + a call with another kind is refused. + workflow_id: + type: string + description: "The owning Workflow." + chain_run_id: + oneOf: [{ type: string }, { type: "null" }] + description: >- + Names the run chain that holds this run. Absent or null names the + Workflow id's newest chain. It never stops a read at this run: a + read follows the chain across Continue-as-New and ends with the + chain. + topic: + type: string + description: "The topic on the owner's stream." + required: [kind, workflow_id, topic] + additionalProperties: false + + AppendInput: + type: object + description: "One append call: who writes, where, and what." + properties: + stream: { $ref: "#/$defs/StreamRef" } + producer_id: + type: string + description: >- + Names the writer across its retries, so the store can deduplicate a + retry and a reader can tell a retry from a new attempt. Never empty. + attempt: + type: integer + minimum: 1 + description: >- + The producer attempt this call writes. A higher attempt supersedes a + lower one, and its sequence starts again at one. + sequence: + type: integer + minimum: 1 + description: >- + The sequence of the first record this call writes. Records of an + attempt are numbered from one with no gaps, so it continues where + the attempt's previous call ended, or repeats that call's sequence + for a retry. + payloads: + type: array + items: + type: string + contentEncoding: base64 + description: >- + The record bodies, in order, each a serialized + temporal.api.common.v1.Payload as the caller's payload converter + left it, before any codec. The call crosses encoded as a whole by + the caller's codec, like any Nexus input, and the store encodes each + body with the serving Worker's codec, so a body is never plaintext + in transit or at rest. The store recognizes a retry by the bodies' + plaintext, so a codec that encodes with a fresh nonce does not make + a retry look like other content. A call with a record larger than + the handler's per-record limit (1.5 MiB by default, below what one + read result can carry) is refused, so every record can be read. + Empty on a call that finishes. + finish: + type: boolean + description: >- + Write FINISH for this producer at sequence: it writes nothing more on + the topic. A finishing call carries no payloads, so every call is one + store write and a retry of it is deduplicated whole. Says nothing + about the producer's outcome. + required: [stream, producer_id, attempt, sequence] + additionalProperties: false + + AppendOutput: + type: object + description: "Where the append landed." + properties: + cursor: + type: string + x-nexus-cursor: StreamCursor + description: >- + The cursor of the last record this call wrote, or of the original's + last record when the call was a deduplicated retry. + required: [cursor] + additionalProperties: true + + ReadInput: + type: object + description: "One read call: where to resume and how long to wait." + properties: + stream: { $ref: "#/$defs/StreamRef" } + after_token: + type: string + x-nexus-cursor: StreamCursor + description: >- + A cursor from an earlier record, read or append of this stream. The + read resumes strictly after the record it names. Absent or empty + starts at the oldest record the stream retains. A cursor from another + stream or another store is refused, and one whose record retention + dropped is refused as expired. + max_records: + type: integer + minimum: 1 + maximum: 1000 + description: >- + Answer with at most this many stored records. 100 when absent. A + supersession comes on top. + wait_ms: + type: integer + minimum: 0 + maximum: 60000 + description: >- + Wait at most this long for a record before answering. The handler + shortens it to fit the request deadline. A call that collects + nothing answers with no records and the caller's own cursor. + latest_only: + type: boolean + description: >- + Answer with the cursor of the newest record and no records, for a + reader that follows from now. Refused beside after_token. + state: + type: string + contentEncoding: base64 + description: >- + The state of this read's previous answer, opaque to the caller. + Absent on a read's first call. With it the endpoint knows which + producer attempts the reader already saw, so its superseded and + stale marks are exact across calls and across handler Workers. A + state that doesn't parse is refused as a cursor mistake. + required: [stream] + additionalProperties: false + + RecordWire: + type: object + description: "One record on the wire: its cursor and the record." + properties: + token: + type: string + x-nexus-cursor: StreamCursor + description: "The cursor of this record. Pass it as after_token to resume past it." + record: + type: string + contentEncoding: base64 + description: >- + The serialized temporal.sdk.streams.v1.StreamRecord: topic, kind, + producer, attempt, sequence and body. The body is the producer's + payload decoded by the serving Worker's data converter, payload + codec and external storage included. The answer as a whole is + encoded on its way back like any operation result, so the caller's + codec decodes it and the body crosses encoded once. Empty when + error or superseded is set. + error: + type: string + description: >- + Why the endpoint can't hand over the record at this cursor, such as + one written to the store directly that is larger than a read result + can carry. The record is left out, and a reader goes on past the + cursor. + stale: + type: boolean + description: >- + The record is from an attempt of its producer below the newest one + this read delivered: an older attempt wrote after a newer one + started. It is still delivered, so a reader can keep it out of the + current answer. + superseded: + $ref: "#/$defs/Supersession" + description: >- + Set instead of a record when a later attempt of a producer started + writing. The store's reader reports it, at the cursor before the + new attempt's first record, also on a read that resumes after a + cursor and saw no earlier attempt itself. + required: [token, record] + additionalProperties: true + + Supersession: + type: object + description: "A later attempt of a producer started writing." + properties: + producer_id: + type: string + description: "The producer whose attempt changed." + previous_attempt: + type: integer + description: "The attempt readers saw before." + attempt: + type: integer + description: "The attempt that started writing." + required: [producer_id, previous_attempt, attempt] + additionalProperties: true + + ReadOutput: + type: object + description: "What one read answered, and where to resume." + properties: + records: + type: array + items: { $ref: "#/$defs/RecordWire" } + description: "The records after the caller's cursor, in stream order." + next_token: + type: string + x-nexus-cursor: StreamCursor + description: >- + The cursor to pass as after_token next. The caller's own cursor when + the call collected nothing. + done: + type: boolean + description: >- + The stream ended for this reader: the stream is closed, or its + owner's run chain ended, and every retained record after the cursor + has been delivered. + Nothing more will arrive, so the caller stops. + state: + type: string + contentEncoding: base64 + description: >- + Pass as state on the next call of this read. It matches next_token, + also when the answer was cut short by its byte budget. It holds at + most 16 KiB, which is about 22 KiB as base64, so it can ride in a + Workflow's History. A longer state is refused. + required: [records, next_token, done] + additionalProperties: true + + StreamOperationToken: + type: object + description: >- + What a stream-returning operation's token carries. Core mints it when + the operation starts, and a caller reads the stream back from it. The + token is this model's transfer form as JSON with sorted keys and no + spaces, in base64url with no padding. + x-nexus-operation-token: { version-field: v } + properties: + v: + type: integer + const: 2 + description: "The token version. A token of another version is refused." + ref: { $ref: "#/$defs/StreamRef" } + attach: + type: string + description: "The request id of the start that attached the caller." + chain: + type: string + description: >- + The first run of the run chain the start attached to, so a cancel + detaches from it even after the Workflow id moved on to a new chain. + required: [v, ref, attach, chain] + additionalProperties: false