[Brook K1] Added the temporalio-streams crate with the Stream protos.

Cargo.toml

@@ -8,6 +8,7 @@ members = [
     "crates/macros",
     "crates/sdk",
     "crates/workflow",
+    "crates/streams",
     "crates/sdk-core-c-bridge",
     "crates/changelog-release-notes",
 ]

crates/sdk-core/CHANGELOG.md

@@ -40,6 +40,8 @@ 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.
+* 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

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. <sdk@temporal.io>"]
+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

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.

crates/streams/build.rs

@@ -0,0 +1,50 @@
+use protox::{
+    Compiler,
+    file::{ChainFileResolver, DescriptorSetFileResolver, GoogleFileResolver, IncludeFileResolver},
+};
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    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(())
+}

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;
+}

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;
+}