[Brook] Served the store service through the C bridge and retired sessions.

crates/protos/protos/local/temporal/sdk/core/streams/streams.proto

@@ -24,8 +24,42 @@ service StreamService {
   rpc Close(CloseRequest) returns (CloseResponse);
   // Deletes every stream of one owner.
   rpc DeleteOwner(DeleteOwnerRequest) returns (DeleteOwnerResponse);
+  // Drops the dedupe state of one producer attempt on one stream. Safe only once the attempt's
+  // last append returned, since a retry of that append would then land twice.
+  rpc RetireProducer(RetireProducerRequest) returns (RetireProducerResponse);
 }
 
+// Which store holds a process's streams, and how to reach it. Lang builds one store from it and
+// sends its stream calls there.
+message StreamStoreConfig {
+  oneof store {
+    RedisStoreConfig redis = 1;
+    MemoryStoreConfig memory = 2;
+  }
+}
+
+// Streams in a Redis or Redis Cluster that Core connects to itself.
+message RedisStoreConfig {
+  // One `redis://` or `rediss://` URL, or the seed nodes of a cluster. Credentials and TLS come
+  // from the URL.
+  repeated string urls = 1;
+  bool cluster = 2;
+  // Prepended to every key, so streams can share a Redis with other data and an ACL can scope
+  // them. Empty uses `temporal-streams`.
+  string key_prefix = 3;
+  // How long a stream keeps a record. Unset keeps seven days.
+  google.protobuf.Duration retention = 4;
+  // How many reads may wait on one node at once, each on a connection of its own. Zero uses the
+  // default.
+  uint32 blocking_reads_per_node = 5;
+  // How long a short call waits for its reply. Unset uses the default.
+  google.protobuf.Duration response_timeout = 6;
+}
+
+// Streams in this process's memory, for tests and local development. Nothing survives the
+// process, and each connect makes a store of its own that shares no streams with another.
+message MemoryStoreConfig {}
+
 // What owns a stream. The owner decides when the stream closes.
 enum StreamOwnerKind {
   STREAM_OWNER_KIND_UNSPECIFIED = 0;
@@ -179,6 +213,18 @@ message DeleteOwnerResponse {
   uint64 deleted = 1;
 }
 
+message RetireProducerRequest {
+  StreamAddress stream = 1;
+  // The producer attempt whose dedupe state goes. Nothing it wrote is touched.
+  oneof producer {
+    NamedProducer named = 2;
+    ActivityProducer activity = 3;
+  }
+}
+
+// A producer attempt with no dedupe state left, or a stream that holds nothing, answers the same.
+message RetireProducerResponse {}
+
 // One class per way a stream call can fail, so every lang maps the same failure to the same
 // error.
 enum StreamFailureKind {

crates/sdk-core-c-bridge/Cargo.toml

@@ -47,7 +47,12 @@ features = ["experimental"]
 [dependencies.temporalio-sdk-core]
 path = "../sdk-core"
 version = "=0.10.0"
-features = ["ephemeral-server", "otel"]
+features = ["ephemeral-server", "otel", "streams"]
+
+[dependencies.temporalio-streams]
+path = "../streams"
+version = "0.1"
+features = ["redis"]
 
 [dependencies.temporalio-common]
 path = "../common"

crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h

@@ -108,6 +108,11 @@ typedef struct TemporalCoreRuntime TemporalCoreRuntime;
 
 typedef struct TemporalCoreSlotReserveCompletionCtx TemporalCoreSlotReserveCompletionCtx;
 
+/**
+ * One process's stream store, serving the stream calls lang makes.
+ */
+typedef struct TemporalCoreStreamStore TemporalCoreStreamStore;
+
 typedef struct TemporalCoreWorker TemporalCoreWorker;
 
 typedef struct TemporalCoreWorkerReplayPusher TemporalCoreWorkerReplayPusher;
@@ -547,6 +552,22 @@ typedef struct TemporalCoreRuntimeOptions {
   bool disable_environment_info;
 } TemporalCoreRuntimeOptions;
 
+/**
+ * If success or fail are not null, they must be manually freed when done.
+ */
+typedef void (*TemporalCoreStreamStoreNewCallback)(void *user_data,
+                                                   struct TemporalCoreStreamStore *success,
+                                                   const struct TemporalCoreByteArray *fail);
+
+/**
+ * If success or failure are not null, they must be manually freed when done. Exactly one is
+ * set. Success is the serialized response, and failure a serialized
+ * `coresdk.streams.StreamFailure`.
+ */
+typedef void (*TemporalCoreStreamStoreCallCallback)(void *user_data,
+                                                    const struct TemporalCoreByteArray *success,
+                                                    const struct TemporalCoreByteArray *failure);
+
 typedef struct TemporalCoreTestServerOptions {
   /**
    * Empty means default behavior
@@ -1073,6 +1094,30 @@ uint64_t temporal_core_forwarded_log_timestamp_millis(const struct TemporalCoreF
 
 struct TemporalCoreByteArrayRef temporal_core_forwarded_log_fields_json(const struct TemporalCoreForwardedLog *log);
 
+/**
+ * Connects to the store `config` names, a serialized `coresdk.streams.StreamStoreConfig`. The
+ * store asks `client`'s server about streams' owners. Client must live as long as the store, and
+ * config and user data must live through the callback.
+ */
+void temporal_core_stream_store_new(struct TemporalCoreConnection *client,
+                                    struct TemporalCoreByteArrayRef config,
+                                    void *user_data,
+                                    TemporalCoreStreamStoreNewCallback callback);
+
+void temporal_core_stream_store_free(struct TemporalCoreStreamStore *store);
+
+/**
+ * Makes one stream call. `rpc` names a `coresdk.streams.StreamService` method and `request` is
+ * its serialized request. Cancelling the token drops the call, which a read uses to stop
+ * waiting. Store, request and user data must live through the callback.
+ */
+void temporal_core_stream_store_call(struct TemporalCoreStreamStore *store,
+                                     struct TemporalCoreByteArrayRef rpc,
+                                     struct TemporalCoreByteArrayRef request,
+                                     const struct TemporalCoreCancellationToken *cancellation_token,
+                                     void *user_data,
+                                     TemporalCoreStreamStoreCallCallback callback);
+
 /**
  * Runtime must live as long as server. Options and user data must live through
  * callback.

crates/sdk-core-c-bridge/src/lib.rs

@@ -13,6 +13,7 @@ pub mod envconfig;
 pub mod metric;
 pub mod random;
 pub mod runtime;
+pub mod streams;
 pub mod testing;
 pub mod worker;
 

crates/sdk-core-c-bridge/src/streams.rs

@@ -0,0 +1,132 @@
+use crate::{
+    ByteArray, ByteArrayRef, CancellationToken, UserDataHandle, client::Connection,
+    runtime::Runtime,
+};
+use prost::Message;
+use std::sync::Arc;
+use temporalio_sdk_core::streams::{StreamService, connect_stream_service, proto};
+
+/// One process's stream store, serving the stream calls lang makes.
+pub struct StreamStore {
+    pub(crate) runtime: Runtime,
+    pub(crate) service: Arc<StreamService>,
+}
+
+/// If success or fail are not null, they must be manually freed when done.
+pub type StreamStoreNewCallback = unsafe extern "C" fn(
+    user_data: *mut libc::c_void,
+    success: *mut StreamStore,
+    fail: *const ByteArray,
+);
+
+/// Connects to the store `config` names, a serialized `coresdk.streams.StreamStoreConfig`. The
+/// store asks `client`'s server about streams' owners. Client must live as long as the store, and
+/// config and user data must live through the callback.
+#[unsafe(no_mangle)]
+pub extern "C" fn temporal_core_stream_store_new(
+    client: *mut Connection,
+    config: ByteArrayRef,
+    user_data: *mut libc::c_void,
+    callback: StreamStoreNewCallback,
+) {
+    let client = unsafe { &*client };
+    let mut runtime = client.runtime.clone();
+    let config = match proto::StreamStoreConfig::decode(config.to_slice()) {
+        Ok(config) => config,
+        Err(err) => {
+            let fail = runtime.alloc_utf8(&format!("Invalid stream store config: {err}"));
+            unsafe { callback(user_data, std::ptr::null_mut(), fail.into_raw()) };
+            return;
+        }
+    };
+    let connection = client.core.clone();
+    let user_data = UserDataHandle(user_data);
+    client.runtime.core.tokio_handle().spawn(async move {
+        match connect_stream_service(config, connection).await {
+            Ok(service) => {
+                let store = Box::into_raw(Box::new(StreamStore {
+                    runtime: runtime.clone(),
+                    service,
+                }));
+                unsafe { callback(user_data.into(), store, std::ptr::null()) };
+            }
+            Err(err) => {
+                let fail =
+                    ByteArray::from_utf8(format!("Could not connect the stream store: {err}"));
+                unsafe { callback(user_data.into(), std::ptr::null_mut(), fail.into_raw()) };
+            }
+        }
+    });
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn temporal_core_stream_store_free(store: *mut StreamStore) {
+    unsafe {
+        let _ = Box::from_raw(store);
+    }
+}
+
+/// The failure a call answers with when its caller cancelled it. A write may or may not have
+/// landed, but a read only stopped, so it gets a message that says so.
+fn cancelled(rpc: &str) -> proto::StreamFailure {
+    let message = match rpc {
+        "Read" | "Latest" => "Cancelled before the read answered",
+        _ => "Cancelled",
+    };
+    proto::StreamFailure {
+        kind: proto::StreamFailureKind::OutcomeUnknown as i32,
+        message: message.to_string(),
+        cursor: String::new(),
+    }
+}
+
+/// If success or failure are not null, they must be manually freed when done. Exactly one is
+/// set. Success is the serialized response, and failure a serialized
+/// `coresdk.streams.StreamFailure`.
+pub type StreamStoreCallCallback = unsafe extern "C" fn(
+    user_data: *mut libc::c_void,
+    success: *const ByteArray,
+    failure: *const ByteArray,
+);
+
+/// Makes one stream call. `rpc` names a `coresdk.streams.StreamService` method and `request` is
+/// its serialized request. Cancelling the token drops the call, which a read uses to stop
+/// waiting. Store, request and user data must live through the callback.
+#[unsafe(no_mangle)]
+pub extern "C" fn temporal_core_stream_store_call(
+    store: *mut StreamStore,
+    rpc: ByteArrayRef,
+    request: ByteArrayRef,
+    cancellation_token: *const CancellationToken,
+    user_data: *mut libc::c_void,
+    callback: StreamStoreCallCallback,
+) {
+    let store = unsafe { &*store };
+    let rpc = rpc.to_string();
+    let request = request.to_vec();
+    let cancel_token = unsafe { cancellation_token.as_ref() }.map(|v| v.token.clone());
+    let service = store.service.clone();
+    let user_data = UserDataHandle(user_data);
+    store.runtime.core.tokio_handle().spawn(async move {
+        let call = service.call(&rpc, &request);
+        let result = if let Some(cancel_token) = cancel_token {
+            tokio::select! {
+                _ = cancel_token.cancelled() => Err(cancelled(&rpc)),
+                result = call => result,
+            }
+        } else {
+            call.await
+        };
+        let (success, failure) = match result {
+            Ok(response) => (
+                ByteArray::from_vec(response).into_raw(),
+                std::ptr::null_mut(),
+            ),
+            Err(failure) => (
+                std::ptr::null_mut(),
+                ByteArray::from_vec(failure.encode_to_vec()).into_raw(),
+            ),
+        };
+        unsafe { callback(user_data.into(), success, failure) };
+    });
+}