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..6d4776b5f --- /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 Records after an offset. + pub async fn read_records( + &self, + request: ReadRecordsRequest, + ) -> StreamResult { + let _ = request; + Err(not_implemented()) + } + + /// Gets the status and the newest offset 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 {} + } +}