From 8be48fae6e8f87af77675bc873eb13f491b9f8dd Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 11 Nov 2026 02:36:56 -0700 Subject: [PATCH 1/2] Added the Stream runtime API with each of its calls stubbed. A bridge calls the runtime's typed methods or their byte form, and holds each writer as an object. Each byte call decodes its request, and every call fails with Unsupported, so the API comes before the Store and the logic of the calls. --- .../temporal/sdk/core/streams/streams.proto | 19 ++- crates/sdk-core/CHANGELOG.md | 4 + crates/streams/Cargo.toml | 3 + crates/streams/src/dispatch.rs | 92 ++++++++++++ crates/streams/src/lib.rs | 6 + crates/streams/src/runtime.rs | 142 ++++++++++++++++++ crates/streams/src/store.rs | 5 + crates/streams/src/writer.rs | 30 ++++ 8 files changed, 299 insertions(+), 2 deletions(-) create mode 100644 crates/streams/src/dispatch.rs create mode 100644 crates/streams/src/runtime.rs create mode 100644 crates/streams/src/store.rs create mode 100644 crates/streams/src/writer.rs 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 2666ef584..d3026dfab 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -6,8 +6,23 @@ option ruby_package = "Temporalio::Internal::Bridge::Api::Streams"; import "google/protobuf/duration.proto"; import "temporal/api/stream/v1/message.proto"; -// The writer messages that `temporalio-streams` adds to the five Stream service operations. A -// bridge passes them as bytes to the runtime's byte calls, so it decodes no proto. +// The writer messages that `temporalio-streams` adds to the five Stream service operations, and +// the config of its memory runtime. A bridge passes them as bytes to the runtime's byte calls, so +// it decodes no proto. + +// The settings of the memory Stream runtime, for tests. Each field has the number of the same +// field in the config of the other Stores, such as Redis, so a backend can send that config here +// too. +message MemoryStreamConfig { + // The default Retention of a new Stream. Empty is 7 days. + google.protobuf.Duration retention = 4; + // How long an open Stream stays after its last write. Empty keeps an open Stream until its + // close. + google.protobuf.Duration last_write_ttl = 5; + // The backend name that lang registered this runtime under. The runtime sets it on each + // StreamRef that it builds, such as the one in a create response. + string backend = 10; +} // The open of a writer, through the runtime's `open_writer`. A writer is for an Activity, a Local // Activity or a client Producer, and it keeps its epoch and `sequence` order. A writer of a Local diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 88177f6ff..a41383654 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -33,6 +33,10 @@ relevant information. ## Unreleased +### Added +* A new experimental crate, `temporalio-streams`, has the Stream runtime that the SDKs link + into their bridges. Core itself has no Stream code. + ## [0.11.0] - 2026-11-09 ### Added diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index 3bebf24f2..5ac179228 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -22,5 +22,8 @@ path = "../common" version = "~1.2.0" default-features = false +[dev-dependencies] +tokio = { version = "1.47", features = ["macros", "rt-multi-thread", "time"] } + [lints] workspace = true diff --git a/crates/streams/src/dispatch.rs b/crates/streams/src/dispatch.rs new file mode 100644 index 000000000..ca91d9365 --- /dev/null +++ b/crates/streams/src/dispatch.rs @@ -0,0 +1,92 @@ +//! The byte form of the runtime's and the writer's calls, for bridges. A bridge passes the encoded +//! request and gets the encoded response, so it decodes no proto. + +use crate::{ + Reason, StreamError, StreamResult, StreamRuntime, StreamWriter, proto::WriteRecordsRequest, +}; +use prost::Message; + +pub(crate) fn decode(input: &[u8]) -> StreamResult { + M::decode(input).map_err(|error| StreamError::invalid(format!("The request is bad: {error}"))) +} + +impl StreamRuntime { + /// Runs one of the five Stream service operations on encoded input, by its name, and returns + /// the encoded output. The bridge raises an error with [StreamError::code] and + /// [StreamError::details]. + pub async fn call(&self, method: &str, input: &[u8]) -> StreamResult> { + let output = match method { + "CreateStream" => self.create_stream(decode(input)?).await?.encode_to_vec(), + "AppendRecords" => self.append_records(decode(input)?).await?.encode_to_vec(), + "CloseStream" => self.close_stream(decode(input)?).await?.encode_to_vec(), + "ReadRecords" => self.read_records(decode(input)?).await?.encode_to_vec(), + "DescribeStream" => self.describe_stream(decode(input)?).await?.encode_to_vec(), + _ => { + return Err(StreamError::new( + Reason::Unsupported, + format!("Unknown method {method}"), + )); + } + }; + Ok(output) + } + + /// Opens a writer from an encoded `OpenWriterRequest`. + pub async fn open_writer_bytes(&self, input: &[u8]) -> StreamResult { + self.open_writer(decode(input)?).await + } +} + +impl StreamWriter { + /// Writes an encoded `WriteRecordsRequest`, and returns the encoded `WriteRecordsResponse`. + pub async fn write_bytes(&self, input: &[u8]) -> StreamResult> { + let request = decode::(input)?; + Ok(self.write(request).await?.encode_to_vec()) + } + + /// Closes the writer, and returns the encoded `CloseWriterResponse`. + pub async fn close_bytes(&self) -> StreamResult> { + Ok(self.close().await?.encode_to_vec()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::store::Store; + use std::sync::Arc; + + struct NoStore; + + impl Store for NoStore {} + + const METHODS: [&str; 5] = [ + "CreateStream", + "AppendRecords", + "CloseStream", + "ReadRecords", + "DescribeStream", + ]; + + fn runtime() -> Arc { + StreamRuntime::new("test", "", Arc::new(NoStore), None) + } + + // Each method decodes its request before anything else. A typo in a method name would make it + // unknown, which is unsupported, so the loop also checks the names that a bridge calls. + #[tokio::test] + async fn bad_bytes_are_invalid_and_an_unknown_method_is_unsupported() { + let runtime = runtime(); + for method in METHODS { + let error = runtime.call(method, &[0xff]).await.unwrap_err(); + assert_eq!(error.reason, Reason::Invalid, "{method}"); + } + let error = runtime.open_writer_bytes(&[0xff]).await.unwrap_err(); + assert_eq!(error.reason, Reason::Invalid); + let error = runtime.call("Nope", &[]).await.unwrap_err(); + assert_eq!( + (error.reason, error.message.as_str()), + (Reason::Unsupported, "Unknown method Nope") + ); + } +} diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index a3810d654..704b59b9f 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -6,9 +6,15 @@ //! client and calls its typed methods, or its byte form with an operation name and proto bytes. //! Core has no hook for it. A Workflow makes each call from a Local Activity. +mod dispatch; mod error; +mod runtime; +pub mod store; +mod writer; pub use error::{Reason, StreamError, StreamResult}; +pub use runtime::{MemoryStreamRuntime, StreamRuntime}; +pub use writer::StreamWriter; /// The protos that lang and the crate exchange. `temporalio-protos` generates them, so a value /// needs no conversion between the crates. diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs new file mode 100644 index 000000000..bbdcbc998 --- /dev/null +++ b/crates/streams/src/runtime.rs @@ -0,0 +1,142 @@ +//! The Stream runtime, a plain object that lang's bridge builds per client. + +use crate::{ + Reason, StreamError, StreamResult, StreamWriter, + proto::{ + AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, CloseStreamResponse, + CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, + MemoryStreamConfig, OpenWriterRequest, ReadRecordsRequest, ReadRecordsResponse, + }, + store::Store, +}; +use prost::Message; +use std::sync::Arc; +use temporalio_common::telemetry::metrics::TemporalMeter; + +/// The Stream runtime on one Store. It serves the five Stream service operations and the writers. +/// +/// A bridge builds one for each client with a Stream backend, and that client's Workers share it. +/// A Store crate, such as `temporalio-streams-redis`, builds it with its own Store. +pub struct StreamRuntime { + name: &'static str, + backend: String, +} + +impl std::fmt::Debug for StreamRuntime { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StreamRuntime") + .field("name", &self.name) + .field("backend", &self.backend) + .finish_non_exhaustive() + } +} + +pub(crate) fn not_implemented() -> StreamError { + StreamError::new(Reason::Unsupported, "Not implemented") +} + +impl StreamRuntime { + /// Builds a runtime on `store`. `name` identifies the runtime in logs. `backend` is the name + /// that lang registered the runtime under, and `meter` is lang's runtime meter, if set. No + /// call of this runtime uses the Store or records a metric, so the runtime keeps neither. + pub fn new( + name: &'static str, + backend: impl Into, + store: Arc, + meter: Option, + ) -> Arc { + let _ = (store, meter); + Arc::new(Self { + name, + backend: backend.into(), + }) + } + + /// A name that identifies the runtime in logs. + pub fn name(&self) -> &str { + self.name + } + + /// Creates a Stream, or gets it with USE_EXISTING. + pub async fn create_stream( + &self, + request: CreateStreamRequest, + ) -> StreamResult { + let _ = request; + Err(not_implemented()) + } + + /// Appends one batch as one Producer. + pub async fn append_records( + &self, + request: AppendRecordsRequest, + ) -> StreamResult { + let _ = request; + Err(not_implemented()) + } + + /// Closes a Stream. + pub async fn close_stream( + &self, + request: CloseStreamRequest, + ) -> StreamResult { + let _ = request; + Err(not_implemented()) + } + + /// Reads one page of items from an offset. + pub async fn read_records( + &self, + request: ReadRecordsRequest, + ) -> StreamResult { + let _ = request; + Err(not_implemented()) + } + + /// Gets the status, the newest offset and the Retention of a Stream. + pub async fn describe_stream( + &self, + request: DescribeStreamRequest, + ) -> StreamResult { + let _ = request; + Err(not_implemented()) + } + + /// Opens a writer for an Activity, a Local Activity or a client Producer. The bridge holds the + /// writer, and dropping it releases it. + pub async fn open_writer(&self, request: OpenWriterRequest) -> StreamResult { + let _ = request; + Err(not_implemented()) + } +} + +/// The Stream runtime for tests. +#[derive(Debug)] +pub struct MemoryStreamRuntime; + +impl MemoryStreamRuntime { + /// Takes an encoded [MemoryStreamConfig]. Its field numbers are those of the same fields in + /// the config of another Store, such as Redis, so a backend can pass that config too. A bad + /// config fails with [Reason::Invalid]. The crate has no memory Store, so a good config fails + /// with [Reason::Unsupported]. + pub fn from_config_bytes( + config: &[u8], + meter: Option, + ) -> StreamResult> { + MemoryStreamConfig::decode(config) + .map_err(|error| StreamError::invalid(format!("The config is bad: {error}")))?; + let _ = meter; + Err(not_implemented()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_bad_memory_config_is_invalid() { + let error = MemoryStreamRuntime::from_config_bytes(&[0xff], None).unwrap_err(); + assert_eq!(error.reason, Reason::Invalid); + } +} diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs new file mode 100644 index 000000000..cfadb83e4 --- /dev/null +++ b/crates/streams/src/store.rs @@ -0,0 +1,5 @@ +//! The Store under a Stream runtime. + +/// Keeps the Records and the state of Streams. A [crate::StreamRuntime] serves the Stream service +/// on one Store, which a bridge or a Store crate, such as `temporalio-streams-redis`, gives it. +pub trait Store: Send + Sync + 'static {} diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs new file mode 100644 index 000000000..2f3165a64 --- /dev/null +++ b/crates/streams/src/writer.rs @@ -0,0 +1,30 @@ +//! Writers for Activities and clients. The bridge holds each writer as an object, and dropping it +//! releases it. + +use crate::{ + StreamResult, + proto::{CloseWriterResponse, WriteRecordsRequest, WriteRecordsResponse}, +}; + +/// Has no value, so no writer exists until the runtime opens one. +#[derive(Debug)] +enum Never {} + +/// A writer of one Producer epoch on one Stream, for an Activity, a Local Activity or a client. +/// The bridge holds it, and dropping it releases it. +#[derive(Debug)] +pub struct StreamWriter(Never); + +impl StreamWriter { + /// Appends Records at once and answers with the offsets of the first and the last one. + pub async fn write(&self, request: WriteRecordsRequest) -> StreamResult { + let _ = request; + match self.0 {} + } + + /// Closes the writer. It waits for a write in flight, then answers with the offset of the + /// newest Record that the writer wrote. It doesn't close the Stream. + pub async fn close(&self) -> StreamResult { + match self.0 {} + } +} From 4c8120ba8ed478fba18c8b3af73a6ed814a92256 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Thu, 12 Nov 2026 02:33:01 -0700 Subject: [PATCH 2/2] Followed the Stream API to done-only reads and no client epoch. The read and describe docs of the stubs named items and the Retention, which the API removed. --- crates/streams/src/runtime.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs index bbdcbc998..6d4776b5f 100644 --- a/crates/streams/src/runtime.rs +++ b/crates/streams/src/runtime.rs @@ -84,7 +84,7 @@ impl StreamRuntime { Err(not_implemented()) } - /// Reads one page of items from an offset. + /// Reads one page of Records after an offset. pub async fn read_records( &self, request: ReadRecordsRequest, @@ -93,7 +93,7 @@ impl StreamRuntime { Err(not_implemented()) } - /// Gets the status, the newest offset and the Retention of a Stream. + /// Gets the status and the newest offset of a Stream. pub async fn describe_stream( &self, request: DescribeStreamRequest,