feat: add temporalio-streams, the Stream runtime that SDK bridges link

Cargo.toml

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

crates/sdk-core/CHANGELOG.md

@@ -36,6 +36,12 @@ relevant information.
 
 ## Unreleased
 
+### Added
+* [lang-internal] Added the experimental `temporalio-streams` crate, the Stream runtime that a
+  language bridge links. It answers the five operations of the Stream service on a Stream Store,
+  through `StreamRuntime::call` with an operation name and the encoded request. Core's worker has no
+  Stream code.
+
 ### Fixed
 * A Nexus operation cancelled with the `ABANDON` cancellation type that later completes, fails, or
   times out no longer fails workflow tasks with a nondeterminism error.

crates/streams/Cargo.toml

@@ -0,0 +1,43 @@
+[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"]
+
+[features]
+# The shared suites, for the crates of other Stores.
+test-support = ["dep:futures-util", "dep:serde_json"]
+
+[dependencies]
+async-trait = "0.1"
+futures-util = { version = "0.3", default-features = false, features = ["std"], optional = true }
+parking_lot = "0.12"
+prost = { workspace = true }
+prost-types = { workspace = true }
+serde_json = { workspace = true, optional = true }
+sha2 = "0.10"
+thiserror = { workspace = true }
+tokio = { version = "1.47", default-features = false, features = ["rt", "sync", "time"] }
+tonic = { workspace = true }
+
+[dependencies.temporalio-common]
+path = "../common"
+version = "~1.2.0"
+default-features = false
+
+[dev-dependencies]
+futures-util = { version = "0.3", default-features = false, features = ["std"] }
+serde_json = { workspace = true }
+tokio = { version = "1.47", features = ["macros", "rt-multi-thread", "time"] }
+# The integration tests call the memory Store's test hooks, which test-support gates.
+temporalio-streams = { path = ".", features = ["test-support"] }
+
+[lints]
+workspace = true

crates/streams/README.md

@@ -0,0 +1,44 @@
+# temporalio-streams
+
+Experimental. The Stream runtime that the Temporal SDKs link into their bridges. It answers the five
+operations of the Stream service on a Store. A bridge builds the runtime per client, as a plain
+object, and calls `StreamRuntime::call` with an operation name and the encoded request. The call
+gives back the encoded response. Core has no Stream code.
+
+## Protos and shared vectors
+
+The crate defines no protos. `temporalio-protos` generates them, and `temporalio_streams::proto`
+re-exports them, so a value needs no conversion between the crates.
+
+- `temporal/api/stream/v1/message.proto`, `temporal/api/streamservice/v1/request_response.proto`
+  and `temporal/api/enums/v1/stream.proto` are the Stream protos of the API. They are in
+  `crates/protos/protos/api_upstream`.
+- `testdata/*.json` are the shared vectors. They cover the Store keys of a Stream id, the meta
+  field of a Producer, offsets, epoch keys and the batch digest of a dedupe entry.
+  `examples/gen_vectors.rs` writes them from a second implementation, and the crate's tests read
+  each file. Run it with `cargo run -p temporalio-streams --example gen_vectors`. It shares no
+  code with the crate. Each Store reads `fields.json` through the shared Store suite.
+  `digests.json` is the golden of the crate's batch digest.
+
+A Stream id and a Producer id are opaque text. The caller builds both. The runtime refuses an
+empty id of either kind, and one over 4096 bytes. A Producer id names one Producer in one Stream.
+A Store keeps the fence of a Producer by kind and id, and an epoch has no run. So the id of a
+run-based kind must hold its run id. A caller that uses one id in two runs gets `FENCED` in the
+second run when that run's epoch is lower.
+
+## The runtime
+
+`StreamRuntime` serves the Stream service on one Store, with one typed method for each operation.
+`StreamRuntime::call` is the byte form of the same methods. It takes `CreateStream`,
+`AppendRecords`, `CloseStream`, `ReadRecords` or `DescribeStream`, and any other name fails with
+`Reason::Unsupported`. A Store crate builds the runtime with `StreamRuntime::new`, the name that
+lang registered the Store under, and its Store. A `store::LazyStore` connects its inner Store on
+the first call that needs it. So a bridge can build the runtime with no async runtime.
+`MemoryStreamRuntime::new` builds it on the memory Store, from a `MemoryStreamOptions`.
+
+## Stores and the shared suites
+
+The memory Store is for tests. It keeps the key layout of the Redis Store and applies the same rules
+in the same order. The `test-support` feature exports the shared suites in `testing`. A Store crate
+implements `testing::TestStore` for its Store and runs each suite with `testing::run`, so that
+every Store keeps one set of rules. This crate runs them on the memory Store in its unit tests.