From 088e22129367d230f36b93ba9da3cab3f8aa6729 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Mon, 26 Oct 2026 22:10:42 -0700 Subject: [PATCH 1/3] Added the temporalio-streams crate with the Stream protos. The crate is the Stream runtime that the SDKs link. It is experimental. The Stream protos are a copy of the API proposal, and they import this repo's API protos. --- Cargo.toml | 1 + crates/sdk-core/CHANGELOG.md | 6 + crates/streams/Cargo.toml | 36 ++++ crates/streams/README.md | 17 ++ crates/streams/build.rs | 50 +++++ .../protos/temporal/api/enums/v1/stream.proto | 90 +++++++++ .../temporal/api/stream/v1/message.proto | 127 ++++++++++++ .../streamservice/v1/request_response.proto | 187 ++++++++++++++++++ .../temporal/sdk/streams/v1/runtime.proto | 83 ++++++++ crates/streams/src/error.rs | 162 +++++++++++++++ crates/streams/src/lib.rs | 20 ++ crates/streams/src/protos.rs | 35 ++++ 12 files changed, 814 insertions(+) create mode 100644 crates/streams/Cargo.toml create mode 100644 crates/streams/README.md create mode 100644 crates/streams/build.rs create mode 100644 crates/streams/protos/temporal/api/enums/v1/stream.proto create mode 100644 crates/streams/protos/temporal/api/stream/v1/message.proto create mode 100644 crates/streams/protos/temporal/api/streamservice/v1/request_response.proto create mode 100644 crates/streams/protos/temporal/sdk/streams/v1/runtime.proto create mode 100644 crates/streams/src/error.rs create mode 100644 crates/streams/src/lib.rs create mode 100644 crates/streams/src/protos.rs diff --git a/Cargo.toml b/Cargo.toml index 76bdd46f64..e44afa24cd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "crates/macros", "crates/sdk", "crates/workflow", + "crates/streams", "crates/sdk-core-c-bridge", "crates/changelog-release-notes", ] diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 75470adb68..cc8f65cd77 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -40,6 +40,12 @@ relevant information. name the run's last completed workflow task, and `failed_tasks`, the started event ids of the run's last 100 failed or timed out workflow tasks. Retries that Core runs locally keep the value. Core leaves it unset for other activities. + also has `previous_started_event_id`, the started event id of the run's last completed workflow + task, and `failed_tasks`, the started event ids of the run's last 100 failed or timed out + workflow tasks. Retries that Core runs locally keep the value. Core leaves it unset for other + activities. +* A new experimental crate, `temporalio-streams`, has the Stream runtime that the SDKs link + into their bridges. Core itself has no Stream code. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml new file mode 100644 index 0000000000..3153fccf2b --- /dev/null +++ b/crates/streams/Cargo.toml @@ -0,0 +1,36 @@ +[package] +name = "temporalio-streams" +version = "0.1.0" +edition = "2024" +rust-version = "1.88.0" +authors = ["Temporal Technologies Inc. "] +license-file = { workspace = true } +description = "The experimental Stream runtime that Temporal SDK bridges link" +homepage = "https://temporal.io/" +repository = "https://github.com/temporalio/sdk-rust" +keywords = ["temporal", "workflow", "streams"] +categories = ["development-tools"] + +[dependencies] +prost = { workspace = true } +prost-types = { workspace = true } +thiserror = { workspace = true } +tonic = { workspace = true } + +[dependencies.temporalio-common] +path = "../common" +version = "~1.1.0" +default-features = false + +# Only for its descriptor set, which the build script reads for the API protos that the Stream +# protos import. +[dependencies.temporalio-protos] +path = "../protos" +version = "~0.10.0" + +[build-dependencies] +protox = "0.9.1" +tonic-prost-build = { workspace = true } + +[lints] +workspace = true diff --git a/crates/streams/README.md b/crates/streams/README.md new file mode 100644 index 0000000000..d1102dd094 --- /dev/null +++ b/crates/streams/README.md @@ -0,0 +1,17 @@ +# temporalio-streams + +Experimental. The Stream runtime that the Temporal SDKs link into their bridges. A bridge builds +the runtime per client, as a plain object, and calls it with a method name and proto bytes. A +Workflow makes each Stream call from a Local Activity. Core has no Stream code. + +## Protos + +- `protos/temporal/api/stream/v1/message.proto`, + `protos/temporal/api/streamservice/v1/request_response.proto` and + `protos/temporal/api/enums/v1/stream.proto` are the Stream protos of the API proposal, byte for + byte. The API protos of this repo don't have them yet, so the crate generates them. Lang + generates the same files. They import API protos that the build reads from the descriptor set + of `temporalio-protos`, and that map to the types of `temporalio-common`. So the packaged crate + needs no file outside itself. +- `protos/temporal/sdk/streams/v1/runtime.proto` has the crate's own calls, such as the writer + calls. diff --git a/crates/streams/build.rs b/crates/streams/build.rs new file mode 100644 index 0000000000..289758e829 --- /dev/null +++ b/crates/streams/build.rs @@ -0,0 +1,50 @@ +use protox::{ + Compiler, + file::{ChainFileResolver, DescriptorSetFileResolver, GoogleFileResolver, IncludeFileResolver}, +}; + +fn main() -> Result<(), Box> { + println!("cargo:rerun-if-changed=protos"); + let protos = [ + "protos/temporal/api/enums/v1/stream.proto", + "protos/temporal/api/stream/v1/message.proto", + "protos/temporal/api/streamservice/v1/request_response.proto", + "protos/temporal/sdk/streams/v1/runtime.proto", + ]; + // The build reads the API protos that these import from the descriptor set of + // `temporalio-protos`, so the packaged crate needs no file outside itself. + let api = std::env::var("DEP_TEMPORALIO_PROTOS_DESCRIPTOR_PATH")?; + println!("cargo:rerun-if-changed={api}"); + let mut resolver = ChainFileResolver::new(); + resolver.add(IncludeFileResolver::new("protos".into())); + resolver.add(DescriptorSetFileResolver::decode( + std::fs::read(&api)?.as_slice(), + )?); + resolver.add(GoogleFileResolver::new()); + let mut compiler = Compiler::with_file_resolver(resolver); + compiler + .include_source_info(true) + .include_imports(true) + .open_files(protos)?; + // protox compiles the protos in Rust, so the crate builds with no protoc, as Core's + // `vendored-protox` build does. + let descriptor_file = std::path::PathBuf::from(std::env::var("OUT_DIR")?).join("streams.bin"); + std::fs::write(&descriptor_file, compiler.encode_file_descriptor_set())?; + tonic_prost_build::configure() + .build_server(false) + .build_client(false) + .skip_protoc_run() + .file_descriptor_set_path(&descriptor_file) + // Use the Core types for the Temporal API messages, so that values need no conversion + // between the crates. Not `.temporal.api.enums`, since Core lacks the Stream enums. + .extern_path( + ".temporal.api.common", + "::temporalio_common::protos::temporal::api::common", + ) + .extern_path( + ".temporal.api.failure", + "::temporalio_common::protos::temporal::api::failure", + ) + .compile_protos(&protos, &["protos"])?; + Ok(()) +} diff --git a/crates/streams/protos/temporal/api/enums/v1/stream.proto b/crates/streams/protos/temporal/api/enums/v1/stream.proto new file mode 100644 index 0000000000..b64f7d5d43 --- /dev/null +++ b/crates/streams/protos/temporal/api/enums/v1/stream.proto @@ -0,0 +1,90 @@ +syntax = "proto3"; + +package temporal.api.enums.v1; + +option go_package = "go.temporal.io/api/enums/v1;enums"; +option java_package = "io.temporal.api.enums.v1"; +option java_multiple_files = true; +option java_outer_classname = "StreamProto"; +option ruby_package = "Temporalio::Api::Enums::V1"; +option csharp_namespace = "Temporalio.Api.Enums.V1"; + +// Experimental. Might change in the future. +enum StreamProducerKind { + STREAM_PRODUCER_KIND_UNSPECIFIED = 0; + STREAM_PRODUCER_KIND_WORKFLOW = 1; + STREAM_PRODUCER_KIND_LOCAL_ACTIVITY = 2; + STREAM_PRODUCER_KIND_ACTIVITY = 3; + STREAM_PRODUCER_KIND_CLIENT = 4; + STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY = 5; +} + +// Experimental. Might change in the future. +// How a Stream closed, so a reader tells a terminated or timed out owner from a failed one +// without parsing the failure. +// (-- api-linter: core::0216::synonyms=disabled +// aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) +enum StreamCloseStatus { + STREAM_CLOSE_STATUS_UNSPECIFIED = 0; + STREAM_CLOSE_STATUS_COMPLETED = 1; + STREAM_CLOSE_STATUS_FAILED = 2; + STREAM_CLOSE_STATUS_CANCELED = 3; + // TERMINATED and TIMED_OUT come only from the owner's end, never from a caller. + STREAM_CLOSE_STATUS_TERMINATED = 4; + STREAM_CLOSE_STATUS_TIMED_OUT = 5; +} + +// Experimental. Might change in the future. +// (-- api-linter: core::0216::synonyms=disabled +// aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) +enum StreamStatus { + STREAM_STATUS_UNSPECIFIED = 0; + STREAM_STATUS_OPEN = 1; + STREAM_STATUS_CLOSED = 2; + STREAM_STATUS_EXPIRED = 3; +} + +// Experimental. Might change in the future. +enum StreamReadStart { + STREAM_READ_START_UNSPECIFIED = 0; + STREAM_READ_START_BEGINNING = 1; + STREAM_READ_START_END = 2; +} + +// Experimental. Might change in the future. +// Why a Stream call failed. Each reason has one gRPC status code. +enum StreamErrorReason { + STREAM_ERROR_REASON_UNSPECIFIED = 0; + // Nobody created the Stream, even after `create_wait`. NOT_FOUND. + STREAM_ERROR_REASON_NOT_FOUND = 1; + // The Stream's Retention or safety TTL passed. NOT_FOUND. + STREAM_ERROR_REASON_EXPIRED = 2; + // The Stream is closed. FAILED_PRECONDITION. + STREAM_ERROR_REASON_CLOSED = 3; + // The Producer's epoch is lower than the newest one, its Workflow Task was withdrawn, or a + // Workflow retry withdrew its run. FAILED_PRECONDITION. + STREAM_ERROR_REASON_FENCED = 4; + // The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION. + STREAM_ERROR_REASON_SEQUENCE = 5; + // The batch has the `sequence` of the last one of its epoch, and other content. + // FAILED_PRECONDITION. + STREAM_ERROR_REASON_SEQ_DIVERGENT = 6; + // A create names another owner than the Stream has, or the owner doesn't allow this close. + // FAILED_PRECONDITION. + STREAM_ERROR_REASON_NOT_OWNER = 7; + // A newer format version wrote the Stream. FAILED_PRECONDITION. + STREAM_ERROR_REASON_FORMAT_TOO_NEW = 8; + // The offset is past the newest offset + 1, for example after a failover lost the newest + // Records. A read from it works again once the Stream has Records up to it. OUT_OF_RANGE. + STREAM_ERROR_REASON_OFFSET = 9; + // A Record or a batch is over the Store's limit. INVALID_ARGUMENT. + STREAM_ERROR_REASON_TOO_LARGE = 10; + // The request is malformed. INVALID_ARGUMENT. + STREAM_ERROR_REASON_INVALID = 11; + // The Store failed or did not answer. UNAVAILABLE. + STREAM_ERROR_REASON_STORAGE = 12; + // No writer has the id. The SDK's in-process writer only. NOT_FOUND. + STREAM_ERROR_REASON_UNKNOWN_WRITER = 13; + // The Store lacks a capability that the call needs. UNIMPLEMENTED. + STREAM_ERROR_REASON_UNSUPPORTED = 14; +} diff --git a/crates/streams/protos/temporal/api/stream/v1/message.proto b/crates/streams/protos/temporal/api/stream/v1/message.proto new file mode 100644 index 0000000000..e4984081ed --- /dev/null +++ b/crates/streams/protos/temporal/api/stream/v1/message.proto @@ -0,0 +1,127 @@ +syntax = "proto3"; + +package temporal.api.stream.v1; + +option go_package = "go.temporal.io/api/stream/v1;stream"; +option java_package = "io.temporal.api.stream.v1"; +option java_multiple_files = true; +option java_outer_classname = "MessageProto"; +option ruby_package = "Temporalio::Api::Stream::V1"; +option csharp_namespace = "Temporalio.Api.Stream.V1"; + +import "google/protobuf/timestamp.proto"; + +import "temporal/api/common/v1/message.proto"; +import "temporal/api/enums/v1/stream.proto"; +import "temporal/api/failure/v1/message.proto"; + +// Experimental. Might change in the future. +// Identifies one Stream. The id is opaque to callers. +message StreamRef { + string id = 1; + // Unset for a client's Stream. For a Workflow, business_id is the Workflow ID and run_id is + // the first execution run id of its run chain. For a Standalone Activity, business_id is the + // activity ID and run_id is its run id. A create and a close of a Workflow-owned Stream must + // name its owner. + temporal.api.common.v1.Execution owner = 2; + // The name of the backend that holds the Stream, as the Worker and client config register + // it, so every process routes a ref it receives. Empty is the default backend. The Store + // never reads it. + string backend = 3; +} + +// Experimental. Might change in the future. +// The ordered attempt of a Producer. Compared in this order: task_started_event_id, +// task_attempt, task_started_time, attempt_scheduled_time, attempt. A field that a kind does +// not use is empty and compares as zero. A timestamp compares seconds, then nanos. +message ProducerEpoch { + // The Workflow Task of a Workflow, or the one that dispatched a Local Activity. + int64 task_started_event_id = 1; + int32 task_attempt = 2; + // Tells a reused started event id apart, for example after a dropped speculative task. + google.protobuf.Timestamp task_started_time = 5; + // When the server scheduled an Activity's attempt. A reset makes it later. + google.protobuf.Timestamp attempt_scheduled_time = 3; + // Activity, Local Activity and client attempt. + int32 attempt = 4; +} + +// Experimental. Might change in the future. +message Producer { + temporal.api.enums.v1.StreamProducerKind kind = 1; + string id = 2; + ProducerEpoch epoch = 3; + // The run of the Producer, so nobody parses the id. Empty for a client. + string run_id = 4; +} + +// Experimental. Might change in the future. +message WorkflowTaskEpoch { + int64 started_event_id = 1; + int32 attempt = 2; + google.protobuf.Timestamp started_time = 3; +} + +// Experimental. Might change in the future. +message AppendRecord { + reserved 1; + reserved "topic"; + temporal.api.common.v1.Payload body = 2; +} + +// Experimental. Might change in the future. +message Supersession { + temporal.api.enums.v1.StreamProducerKind producer_kind = 1; + string producer_id = 2; + ProducerEpoch epoch = 3; + // Workflow only. + WorkflowTaskEpoch previous_completed_task = 4; + repeated int64 failed_tasks = 5; + // Workflow retry only. The previous run. Every Record whose Producer carries it is withdrawn. + string withdraw_run = 6; + // The run of the Producer, so nobody parses the id. Empty for a client. + string run_id = 7; +} + +// Experimental. Might change in the future. +// One Record that a Producer appended. +message StreamData { + reserved 1; + reserved "topic"; + Producer producer = 2; + temporal.api.common.v1.Payload body = 3; +} + +// Experimental. Might change in the future. +// The close of a Stream. A Stream has at most one, and a read returns it last. +message StreamClose { + // The Producer that closed the Stream. + Producer producer = 1; + temporal.api.common.v1.Payload result = 2; + temporal.api.failure.v1.Failure failure = 3; + // The result or the failure stays next to it. + temporal.api.enums.v1.StreamCloseStatus status = 4; +} + +// Experimental. Might change in the future. +// An offset is ordered and unique in one Stream, and it is sparse. Offsets have gaps, so they +// don't count Records. Zero is never an offset. A read names its Stream, so the read checks the +// offset against that Stream. +message StreamRecord { + int64 offset = 1; + // A read item is a Record, a Superseded item or the close. + oneof item { + StreamData data = 2; + // The first batch of a higher epoch carries it. It withdraws the earlier Records that its + // facts name, and a reader drops them. + Supersession superseded = 3; + StreamClose close = 4; + } +} + +// Experimental. Might change in the future. +// The details of a failed Stream call, in the gRPC status details and in the details of the +// application failure of a Nexus operation. +message StreamErrorDetails { + temporal.api.enums.v1.StreamErrorReason reason = 1; +} diff --git a/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto b/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto new file mode 100644 index 0000000000..f48dfcf658 --- /dev/null +++ b/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto @@ -0,0 +1,187 @@ +syntax = "proto3"; + +package temporal.api.streamservice.v1; + +option go_package = "go.temporal.io/api/streamservice/v1;streamservice"; +option java_package = "io.temporal.api.streamservice.v1"; +option java_multiple_files = true; +option java_outer_classname = "RequestResponseProto"; +option ruby_package = "Temporalio::Api::StreamService::V1"; +option csharp_namespace = "Temporalio.Api.StreamService.V1"; + +// The messages of the Stream service. An SDK Store (Redis or memory) and the server answer the +// same five operations with them. A field that only the server reads says so. + +import "google/protobuf/duration.proto"; + +import "temporal/api/common/v1/message.proto"; +import "temporal/api/enums/v1/stream.proto"; +import "temporal/api/failure/v1/message.proto"; +import "temporal/api/stream/v1/message.proto"; + +// Experimental. Might change in the future. +// Creates the Stream, or gets it when it exists. Then created is false, and the Stream keeps its +// own Retention. It fails with CLOSED on a closed Stream, EXPIRED on an expired one, and NOT_OWNER +// on one with another owner. +// (-- api-linter: core::0133=disabled +// aip.dev/not-precedent: The request names a Stream by namespace and stream_id. --) +message CreateStreamRequest { + reserved 5; + reserved "auto_close"; + string namespace = 1; + // Required, and never empty. + string stream_id = 2; + // The age after which a Record is dropped. Empty is the Store default. + google.protobuf.Duration retention = 3; + // Workflow only. The previous run on a retry run's first create, to withdraw its Records. + string withdraw_run = 4; + // The Workflow run that owns the Stream. Workflow kind, with its run_id and epoch. It needs a + // ref whose owner names the run's chain. + temporal.api.stream.v1.Producer owner = 6; + // Only the server reads it. The server closes the Stream with the outcome of the owner's run + // chain, terminate and timeout included. It needs owner, and a ref with a Workflow owner. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "with" ties the close to the owner's run chain. --) + bool close_with_run = 7; + // The ref of the Stream, with stream_id as its id. A Workflow-owned Stream sets the ref's + // owner, which names the owner's run chain. + temporal.api.stream.v1.StreamRef ref = 8; + // Only the server reads it. The server checks that the token is the Activity's current + // attempt, so a caller can't put a Stream into a Standalone Activity it isn't. + bytes task_token = 9; +} + +// Experimental. Might change in the future. +message CreateStreamResponse { + reserved 3; + reserved "status"; + // The ref, with the owner of a Workflow-owned Stream. + temporal.api.stream.v1.StreamRef stream = 1; + bool created = 2; +} + +// Experimental. Might change in the future. +// FAILED_PRECONDITION if closed, fenced, or if the Stream's meta has a higher format version v +// (detail stream_format_too_new). NOT_FOUND if absent. +message AppendRecordsRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + // Required outside a Workflow. In a Workflow the transport sets it. + temporal.api.stream.v1.Producer producer = 3; + // The batch number in this epoch. With the Producer and epoch it is the dedupe key. + int64 sequence = 4; + repeated temporal.api.stream.v1.AppendRecord records = 5; + // The run's last completed Workflow Task (started_event_id 0 before the first) and its failed + // tasks, from a Local Activity's start. The batch withdraws the run's Local Activity Records + // whose task never completed. + temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6; + repeated int64 failed_tasks = 7; + // Local Activity only. The previous run on a retry run's first batch, to withdraw its Records. + string withdraw_run = 8; + // Only the server reads it, on a Workflow Producer's batch. The batch withdraws no Workflow + // Task and keeps no task facts, but withdraw_run still applies. + bool defer_withdrawal = 9; + // Only the server reads it. The server checks the Producer against the Activity's current + // attempt, so nobody else writes as that Activity. + bytes task_token = 10; +} + +// Experimental. Might change in the future. +message AppendRecordsResponse { + int64 first_offset = 1; + int64 last_offset = 2; +} + +// Experimental. Might change in the future. +message CloseStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + temporal.api.stream.v1.Producer producer = 3; + temporal.api.common.v1.Payload result = 4; + temporal.api.failure.v1.Failure failure = 5; + // Only the server reads these. They are the run's task facts, as on the append. + temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6; + repeated int64 failed_tasks = 7; + // Only the server reads it. The previous run on a retry run's close, to withdraw its Records. + string withdraw_run = 8; + // Empty is COMPLETED with no failure, CANCELED with a failure that has canceled_failure_info, + // and FAILED with any other failure. + temporal.api.enums.v1.StreamCloseStatus status = 9; + // Only the server reads it. It checks the token as on the append. + bytes task_token = 10; +} + +// Experimental. Might change in the future. +message CloseStreamResponse { +} + +// Experimental. Might change in the future. +message ReadRecordsRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + // Start at this offset. Zero uses start. OUT_OF_RANGE if past the newest offset + 1. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --) + int64 from_offset = 3; + temporal.api.enums.v1.StreamReadStart start = 4; + reserved 5, 10; + reserved "topics", "topic_prefixes"; + // The most items in one page, Superseded items included. Zero is 1000. A Superseded item and + // the Record after it stay in one page, so a page of one Record can hold two items. + int32 max_records = 6; + google.protobuf.Duration wait = 7; + // How long to wait for a Stream that nobody created yet. Empty fails at once with + // NOT_FOUND. At most 60 s. + google.protobuf.Duration create_wait = 8; + // The exclusive end of the read. The read scans no entry at or after it, the close Record + // included, so a replay reads the same range. Zero means no end. Only the server serves it, + // and a Store refuses it. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --) + int64 to_offset = 11; + // Park as a normal read, then answer with no Records and next_offset = from_offset, when a + // Record at or after from_offset exists, when the Stream closes (done), or when wait ends. Its + // wait may be days, since a parked read adds nothing to History while it waits. Only the + // server serves it, and a Store refuses it. + bool wake_only = 12; + // Read the newest `last` Records, oldest first. It needs no from_offset and no start. Not with + // wake_only, and not in Workflow code, where replay needs a fixed range. + int32 last = 13; + // The most Record bytes in one page, so a page fits the caller's limit, such as a History + // marker. A larger Record still comes alone, so the read always moves on. Zero is no cap. + int64 max_bytes = 14; + // The lowest offset this run read on replay. A reset run copies the earlier run's reads into + // its History, so its floor must cover them too, not only its first live page. Only the + // server reads it. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: The floor is the lowest from_offset of the run's reads. --) + int64 floor_from = 15; +} + +// Experimental. Might change in the future. +message ReadRecordsResponse { + repeated temporal.api.stream.v1.StreamRecord records = 1; + // The from_offset of the next read, which is the newest offset read + 1. + int64 next_offset = 2; + bool done = 3; + // This reader's floor keeps more than half of the Stream's byte cap. The writer fails when the + // cap fills, and only the reader's continue-as-new frees it. Only the server sets it. + bool continue_as_new_suggested = 4; +} + +// Experimental. Might change in the future. +message DescribeStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; +} + +// Experimental. Might change in the future. +message DescribeStreamResponse { + reserved 4; + reserved "auto_close"; + temporal.api.enums.v1.StreamStatus status = 1; + // The newest offset, of a Record, a Superseded item or the close. Zero when the Stream has + // none. + int64 last_offset = 2; + google.protobuf.Duration retention = 3; +} diff --git a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto new file mode 100644 index 0000000000..cc0de59d7a --- /dev/null +++ b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto @@ -0,0 +1,83 @@ +syntax = "proto3"; + +// The messages of the crate's own calls, in addition to the five Stream service operations. Lang +// sends them through the bridge's call on the runtime object, by the method name in each comment. +// Field numbers are part of the contract with every SDK. +package temporal.sdk.streams.v1; + +import "google/protobuf/duration.proto"; +import "temporal/api/stream/v1/message.proto"; + +// Method `OpenWriter`, asynchronous. A writer for an Activity, a Local Activity or a client +// Producer, which keeps its epoch and `sequence` order. A writer of a Local Activity, or of an +// Activity attempt above 1, first sends an empty batch, so that the new epoch withdraws the earlier +// one's Records. The handles of one Producer epoch on one Stream share one writer. Each open adds a +// handle, and each `CloseWriter` or `FreeWriter` releases one. The batches of every writer in a +// process take their `sequence` from one counter, so a writer that opens again after its close +// continues above its earlier batches. +message OpenWriterRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + temporal.api.stream.v1.Producer producer = 3; + // A write sends at most this many Records in one batch. Zero is 1000. + int32 max_batch_size = 5; + // How long a write retries a Store failure. Empty is 10 s. + google.protobuf.Duration max_retry_duration = 6; + // Local Activity only. The run's task facts, which each batch sends as AppendRecords does, so + // the writes withdraw the run's orphaned Local Activity Records. + temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 8; + repeated int64 failed_tasks = 9; +} + +message OpenWriterResponse { + uint64 writer_id = 1; +} + +// Method `WriteRecords`, asynchronous. Appends Records at once for an open writer and responds when +// the Store has them. It uses the writer's Producer epoch and `sequence` order. After a Store +// failure it resends the same batch, so that the Store writes it once. Records past the Store's +// batch limit go in more batches, in order. An awaited publish of an Activity or a client uses this +// call. A Store failure after the retries fails only that write, and the next write takes a higher +// `sequence`. Any other failure fails the writer. +message WriteRecordsRequest { + uint64 writer_id = 1; + repeated temporal.api.stream.v1.AppendRecord records = 2; +} + +message WriteRecordsResponse { + // The offsets of the first and the last Record. Zero when the request has no Record. + int64 first_offset = 1; + int64 last_offset = 2; +} + +// Method `CloseWriter`, asynchronous. Releases one handle of the writer. The last handle waits for +// a write in flight, then drops the writer. It doesn't close the Stream. It fails with the +// writer's error if the writer failed. +message CloseWriterRequest { + uint64 writer_id = 1; +} + +message CloseWriterResponse { + // The offset of the newest Record the writer wrote. Zero if it wrote none. + int64 last_offset = 1; +} + +// Method `FreeWriter`. Releases one handle of the writer with no wait, for a handle that lang +// drops, such as from a finalizer. The bridge may also call it with no async runtime. A write in +// flight still completes. The writer's error, if any, is lost. +message FreeWriterRequest { + uint64 writer_id = 1; +} + +message FreeWriterResponse { +} + +// Method `CloseWriters`, asynchronous. Drops every open writer, for example when the backend's +// client or Worker shuts down. +message CloseWritersRequest { +} + +message CloseWritersResponse { + // The writers that had failed. + int32 failed = 1; +} diff --git a/crates/streams/src/error.rs b/crates/streams/src/error.rs new file mode 100644 index 0000000000..a149d9127e --- /dev/null +++ b/crates/streams/src/error.rs @@ -0,0 +1,162 @@ +//! The error of every failed call, and its form for lang. + +use crate::proto::{StreamErrorDetails, StreamErrorReason}; +use prost::Message; +use tonic::Code; + +/// The result of every fallible call in the crate. +pub type StreamResult = Result; + +/// Why a call failed. Each reason has one status code and one case of `StreamErrorReason`, and lang +/// maps the reason to one error class. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum Reason { + /// Nobody created the Stream, even after `create_wait`. + NotFound, + /// The Stream's Retention or safety TTL passed. + Expired, + /// The Stream is closed. + Closed, + /// The Producer's epoch is lower than the newest one, its Workflow Task was withdrawn, or a + /// Workflow retry withdrew its run. + Fenced, + /// The batch has a lower `sequence` than the last one of its epoch. + Sequence, + /// The batch has the `sequence` of the last one of its epoch, and other content. + SeqDivergent, + /// The Stream's owner doesn't allow this Producer to close it. + NotOwner, + /// A newer format version wrote the Stream. + FormatTooNew, + /// The offset is past the newest offset + 1, for example after a failover lost the newest + /// Records. + Offset, + /// A Record or a batch is over the Store's limit. + TooLarge, + /// The request is malformed. + Invalid, + /// The Store failed or did not answer. + Storage, + /// No writer has the id. + UnknownWriter, + /// The Store lacks a capability that the call needs. + Unsupported, +} + +impl Reason { + /// The token lang reads from the error details. + pub fn token(self) -> &'static str { + match self { + Reason::NotFound => "stream_not_found", + Reason::Expired => "stream_expired", + Reason::Closed => "stream_closed", + Reason::Fenced => "stream_fenced", + Reason::Sequence => "stream_sequence", + Reason::SeqDivergent => "seq_divergent", + Reason::NotOwner => "stream_not_owner", + Reason::FormatTooNew => "stream_format_too_new", + Reason::Offset => "stream_offset", + Reason::TooLarge => "stream_too_large", + Reason::Invalid => "stream_invalid", + Reason::Storage => "stream_storage", + Reason::UnknownWriter => "stream_unknown_writer", + Reason::Unsupported => "stream_unsupported", + } + } + + /// The reason in the error details. + pub fn proto(self) -> StreamErrorReason { + match self { + Reason::NotFound => StreamErrorReason::NotFound, + Reason::Expired => StreamErrorReason::Expired, + Reason::Closed => StreamErrorReason::Closed, + Reason::Fenced => StreamErrorReason::Fenced, + Reason::Sequence => StreamErrorReason::Sequence, + Reason::SeqDivergent => StreamErrorReason::SeqDivergent, + Reason::NotOwner => StreamErrorReason::NotOwner, + Reason::FormatTooNew => StreamErrorReason::FormatTooNew, + Reason::Offset => StreamErrorReason::Offset, + Reason::TooLarge => StreamErrorReason::TooLarge, + Reason::Invalid => StreamErrorReason::Invalid, + Reason::Storage => StreamErrorReason::Storage, + Reason::UnknownWriter => StreamErrorReason::UnknownWriter, + Reason::Unsupported => StreamErrorReason::Unsupported, + } + } + + /// The status code of the reason, as in the proposal's Appendix A. + pub fn code(self) -> Code { + match self { + Reason::NotFound | Reason::Expired | Reason::UnknownWriter => Code::NotFound, + Reason::Closed + | Reason::Fenced + | Reason::Sequence + | Reason::SeqDivergent + | Reason::NotOwner + | Reason::FormatTooNew => Code::FailedPrecondition, + Reason::TooLarge | Reason::Invalid => Code::InvalidArgument, + // A read from the offset works again once the Stream has Records up to it. + Reason::Offset => Code::OutOfRange, + Reason::Storage => Code::Unavailable, + Reason::Unsupported => Code::Unimplemented, + } + } +} + +/// A failed Stream call. +#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] +#[error("{}: {message}", reason.token())] +pub struct StreamError { + /// Why the call failed. + pub reason: Reason, + /// What happened, for a log or the facade error. + pub message: String, +} + +impl StreamError { + /// Creates an error. + pub fn new(reason: Reason, message: impl Into) -> Self { + Self { + reason, + message: message.into(), + } + } + + /// A malformed request. + pub fn invalid(message: impl Into) -> Self { + Self::new(Reason::Invalid, message) + } + + /// A Store failure. + pub fn storage(message: impl Into) -> Self { + Self::new(Reason::Storage, message) + } + + /// The status code that lang raises the error with. + pub fn code(&self) -> Code { + self.reason.code() + } + + /// The encoded [StreamErrorDetails], which lang reads the reason from, as from the details of + /// a gRPC status. + pub fn details(&self) -> Vec { + StreamErrorDetails { + reason: self.reason.proto() as i32, + } + .encode_to_vec() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_error_carries_its_code_and_reason_to_lang() { + let error = StreamError::new(Reason::FormatTooNew, "format 2 is newer than 1"); + assert_eq!(error.code(), Code::FailedPrecondition); + let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap(); + assert_eq!(details.reason(), StreamErrorReason::FormatTooNew); + } +} diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs new file mode 100644 index 0000000000..95e6f23b41 --- /dev/null +++ b/crates/streams/src/lib.rs @@ -0,0 +1,20 @@ +#![warn(missing_docs)] + +//! The Stream runtime of Temporal SDKs. It is experimental. +//! +//! The crate serves the Stream service on a Stream Store. Lang's bridge builds the runtime per +//! client and calls it with an operation name and proto bytes. Core has no hook for it. A Workflow +//! makes each call from a Local Activity. + +mod error; +pub mod protos; + +pub use error::{Reason, StreamError, StreamResult}; + +/// The protos that lang and the crate exchange. +pub mod proto { + pub use crate::protos::temporal::{ + api::{enums::v1::*, stream::v1::*, streamservice::v1::*}, + sdk::streams::v1::*, + }; +} diff --git a/crates/streams/src/protos.rs b/crates/streams/src/protos.rs new file mode 100644 index 0000000000..7b939ed842 --- /dev/null +++ b/crates/streams/src/protos.rs @@ -0,0 +1,35 @@ +//! The generated protos of the Stream service of the API proposal and of the crate's own calls. + +#![allow(missing_docs, unreachable_pub, clippy::all)] + +/// The Temporal API packages that the crate generates. Others map to `temporalio-common`, and +/// `enums` has only the Stream enums. +pub mod temporal { + pub mod api { + pub mod enums { + pub mod v1 { + tonic::include_proto!("temporal.api.enums.v1"); + } + } + + pub mod stream { + pub mod v1 { + tonic::include_proto!("temporal.api.stream.v1"); + } + } + + pub mod streamservice { + pub mod v1 { + tonic::include_proto!("temporal.api.streamservice.v1"); + } + } + } + + pub mod sdk { + pub mod streams { + pub mod v1 { + tonic::include_proto!("temporal.sdk.streams.v1"); + } + } + } +} From 15225360e216c1e22d62060fc3734a469ef046e3 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 30 Oct 2026 00:16:58 -0700 Subject: [PATCH 2/3] Fixed the K1 CHANGELOG, the token doc and the writer proto. The CHANGELOG repeated part of c1's entry. Lang reads the reason from the error details, not from the token. OpenWriterRequest reserves the numbers of its removed fields. --- crates/sdk-core/CHANGELOG.md | 4 ---- crates/streams/protos/temporal/sdk/streams/v1/runtime.proto | 1 + crates/streams/src/error.rs | 3 ++- 3 files changed, 3 insertions(+), 5 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index bbd0f766bf..c4887f9651 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -40,10 +40,6 @@ relevant information. name the run's last completed workflow task, and `failed_tasks`, the started event ids of the run's last 100 failed or timed out workflow tasks. Retries that Core runs locally keep the value. Core leaves it unset for other activities. - also has `previous_started_event_id`, the started event id of the run's last completed workflow - task, and `failed_tasks`, the started event ids of the run's last 100 failed or timed out - workflow tasks. Retries that Core runs locally keep the value. Core leaves it unset for other - activities. * A new experimental crate, `temporalio-streams`, has the Stream runtime that the SDKs link into their bridges. Core itself has no Stream code. diff --git a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto index cc0de59d7a..4ca123e949 100644 --- a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto +++ b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto @@ -16,6 +16,7 @@ import "temporal/api/stream/v1/message.proto"; // process take their `sequence` from one counter, so a writer that opens again after its close // continues above its earlier batches. message OpenWriterRequest { + reserved 4, 7; string namespace = 1; temporal.api.stream.v1.StreamRef stream = 2; temporal.api.stream.v1.Producer producer = 3; diff --git a/crates/streams/src/error.rs b/crates/streams/src/error.rs index a149d9127e..0c0aa4a529 100644 --- a/crates/streams/src/error.rs +++ b/crates/streams/src/error.rs @@ -45,7 +45,8 @@ pub enum Reason { } impl Reason { - /// The token lang reads from the error details. + /// The token that starts the error's text. Lang reads the reason from + /// `StreamErrorDetails.reason`, not from this text. pub fn token(self) -> &'static str { match self { Reason::NotFound => "stream_not_found", From a0413041960471ba00e79a82b6e01a09f6c8f2ca Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 30 Oct 2026 10:29:21 -0700 Subject: [PATCH 3/3] Copied A1's request protos, which drop the close status. The failure gives the close status. A zero sequence now takes the next batch number of the epoch. --- .../api/streamservice/v1/request_response.proto | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto b/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto index f48dfcf658..a2bfeba350 100644 --- a/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto +++ b/crates/streams/protos/temporal/api/streamservice/v1/request_response.proto @@ -68,7 +68,8 @@ message AppendRecordsRequest { temporal.api.stream.v1.StreamRef stream = 2; // Required outside a Workflow. In a Workflow the transport sets it. temporal.api.stream.v1.Producer producer = 3; - // The batch number in this epoch. With the Producer and epoch it is the dedupe key. + // The batch number in this epoch. With the Producer and epoch it is the dedupe key. Zero takes + // the next number of the epoch. Such a batch has no dedupe, so a retry writes it again. int64 sequence = 4; repeated temporal.api.stream.v1.AppendRecord records = 5; // The run's last completed Workflow Task (started_event_id 0 before the first) and its failed @@ -98,15 +99,17 @@ message CloseStreamRequest { temporal.api.stream.v1.StreamRef stream = 2; temporal.api.stream.v1.Producer producer = 3; temporal.api.common.v1.Payload result = 4; + // The close status comes from it. No failure is COMPLETED. A failure is CANCELED when it has + // canceled_failure_info, or when it is an Activity, child Workflow or Nexus Operation failure + // whose cause has canceled_failure_info. Any other failure is FAILED. temporal.api.failure.v1.Failure failure = 5; // Only the server reads these. They are the run's task facts, as on the append. temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6; repeated int64 failed_tasks = 7; // Only the server reads it. The previous run on a retry run's close, to withdraw its Records. string withdraw_run = 8; - // Empty is COMPLETED with no failure, CANCELED with a failure that has canceled_failure_info, - // and FAILED with any other failure. - temporal.api.enums.v1.StreamCloseStatus status = 9; + reserved 9; + reserved "status"; // Only the server reads it. It checks the token as on the append. bytes task_token = 10; }