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 8bf971cc28..e0bfeb85e2 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/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 { diff --git a/crates/sdk-core-c-bridge/Cargo.toml b/crates/sdk-core-c-bridge/Cargo.toml index 23540da9dd..725819c014 100644 --- a/crates/sdk-core-c-bridge/Cargo.toml +++ b/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" diff --git a/crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h b/crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h index 119f1027f7..32d74e47f1 100644 --- a/crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h +++ b/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. diff --git a/crates/sdk-core-c-bridge/src/lib.rs b/crates/sdk-core-c-bridge/src/lib.rs index d96da6ca24..806fe78e56 100644 --- a/crates/sdk-core-c-bridge/src/lib.rs +++ b/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; diff --git a/crates/sdk-core-c-bridge/src/streams.rs b/crates/sdk-core-c-bridge/src/streams.rs new file mode 100644 index 0000000000..95b289c6fa --- /dev/null +++ b/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, +} + +/// 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) }; + }); +} diff --git a/crates/sdk-core-c-bridge/src/tests/context.rs b/crates/sdk-core-c-bridge/src/tests/context.rs index fe06c0606b..da82c6c03a 100644 --- a/crates/sdk-core-c-bridge/src/tests/context.rs +++ b/crates/sdk-core-c-bridge/src/tests/context.rs @@ -8,6 +8,10 @@ use crate::{ Runtime, RuntimeInfoArray, RuntimeOptions, RuntimeOrFail, temporal_core_byte_array_free, temporal_core_runtime_free, temporal_core_runtime_new, }, + streams::{ + StreamStore, temporal_core_stream_store_call, temporal_core_stream_store_free, + temporal_core_stream_store_new, + }, testing::{ DevServerOptions, EphemeralServer, TestServerOptions, temporal_core_ephemeral_server_free, temporal_core_ephemeral_server_shutdown, temporal_core_ephemeral_server_start_dev_server, @@ -449,6 +453,69 @@ impl Context { self.wait_for_operation_result().map(|r| *r) } + /// Builds a stream store from the serialized `config` on this context's client. + pub fn stream_store_new(self: &Arc, config: Vec) -> anyhow::Result { + let (runtime, client) = { + let mut guard = self.wait_for_available()?; + let client = guard.client; + if client.is_null() { + return Err(anyhow!("Client is null")); + } + guard.operation_state = ContextOperationState::InProgress; + (guard.runtime, client) + }; + let config_ref: ByteArrayRef = config.as_slice().into(); + let user_data = Box::into_raw(Box::new(CallbackUserData { + data: runtime, + context: Arc::downgrade(self), + _allocations: Box::new(config), + })) as *mut libc::c_void; + temporal_core_stream_store_new(client, config_ref, user_data, stream_store_new_callback); + self.wait_for_operation_result().map(|r| *r) + } + + /// Makes one stream call, answering with the response or the serialized failure. + pub fn stream_store_call( + self: &Arc, + store: &StreamStorePtr, + rpc: &str, + request: Vec, + ) -> anyhow::Result, Vec>> { + self.stream_store_call_with_token(store, rpc, request, std::ptr::null()) + } + + /// Makes a stream call that `cancellation_token`, when not null, can cancel. + pub fn stream_store_call_with_token( + self: &Arc, + store: &StreamStorePtr, + rpc: &str, + request: Vec, + cancellation_token: *const crate::CancellationToken, + ) -> anyhow::Result, Vec>> { + let runtime = { + let mut guard = self.wait_for_available()?; + guard.operation_state = ContextOperationState::InProgress; + guard.runtime + }; + let rpc = rpc.to_string(); + let rpc_ref: ByteArrayRef = rpc.as_str().into(); + let request_ref: ByteArrayRef = request.as_slice().into(); + let user_data = Box::into_raw(Box::new(CallbackUserData { + data: runtime, + context: Arc::downgrade(self), + _allocations: Box::new((rpc, request)), + })) as *mut libc::c_void; + temporal_core_stream_store_call( + store.0, + rpc_ref, + request_ref, + cancellation_token, + user_data, + stream_store_call_callback, + ); + self.wait_for_operation_result().map(|r| *r) + } + fn wait_while( &self, condition: impl FnMut(&mut InnerContext) -> bool, @@ -710,3 +777,70 @@ impl From<&HashMap>> for GrpcMetadataHolder { } } } + +/// A stream store the test owns, freed when dropped. +#[derive(Debug)] +pub struct StreamStorePtr(*mut StreamStore); + +unsafe impl Send for StreamStorePtr {} + +impl Drop for StreamStorePtr { + fn drop(&mut self) { + temporal_core_stream_store_free(self.0); + } +} + +extern "C" fn stream_store_new_callback( + user_data: *mut libc::c_void, + mut store: *mut StreamStore, + mut fail: *const ByteArray, +) { + let user_data = unsafe { Box::from_raw(user_data as *mut CallbackUserData<(), Context>) }; + if let Some(context) = user_data.context.upgrade() { + let _ = context.complete_operation_catch_unwind(|guard| { + if let Some(fail) = byte_array_to_string(guard.runtime, std::mem::take(&mut fail)) { + ContextOperationState::CallbackError(anyhow!("Stream store failed: {fail}")) + } else { + let store = StreamStorePtr(std::mem::take(&mut store)); + ContextOperationState::CallbackOk(Some(Box::new(store) as _)) + } + }); + } + if !fail.is_null() { + temporal_core_byte_array_free(std::ptr::null_mut(), fail); + } + if !store.is_null() { + temporal_core_stream_store_free(store); + } +} + +extern "C" fn stream_store_call_callback( + user_data: *mut libc::c_void, + mut success: *const ByteArray, + mut failure: *const ByteArray, +) { + let user_data = unsafe { Box::from_raw(user_data as *mut CallbackUserData<(), Context>) }; + if let Some(context) = user_data.context.upgrade() { + let _ = context.complete_operation_catch_unwind(|guard| { + let success = byte_array_to_vec(guard.runtime, std::mem::take(&mut success)); + let failure = byte_array_to_vec(guard.runtime, std::mem::take(&mut failure)); + let result: Result, Vec> = match (success, failure) { + (Some(success), None) => Ok(success), + (None, Some(failure)) => Err(failure), + other => { + return ContextOperationState::CallbackError(anyhow!( + "A stream call must answer with exactly one of success and failure, got \ + {other:?}" + )); + } + }; + ContextOperationState::CallbackOk(Some(Box::new(result) as _)) + }); + } + if !success.is_null() { + temporal_core_byte_array_free(std::ptr::null_mut(), success); + } + if !failure.is_null() { + temporal_core_byte_array_free(std::ptr::null_mut(), failure); + } +} diff --git a/crates/sdk-core-c-bridge/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs index c4554d5001..ba49907a18 100644 --- a/crates/sdk-core-c-bridge/src/tests/mod.rs +++ b/crates/sdk-core-c-bridge/src/tests/mod.rs @@ -463,3 +463,216 @@ fn assert_binary_header_value( let decoded = decode_result.ok().unwrap(); assert_eq!(expected_value, decoded); } + +/// A server where every Workflow's latest run is `run-1`, the first of its chain, still running. +unsafe extern "C" fn streams_callback_override( + req: *mut ClientGrpcOverrideRequest, + _user_data: *mut libc::c_void, +) { + use temporalio_common::protos::temporal::api::{ + common::v1::WorkflowExecution, enums::v1::WorkflowExecutionStatus, + workflow::v1::WorkflowExecutionInfo, + workflowservice::v1::DescribeWorkflowExecutionResponse, + }; + let resp_raw = match temporal_core_client_grpc_override_request_rpc(req).to_str() { + "GetSystemInfo" => GetSystemInfoResponse::default().encode_to_vec(), + "DescribeWorkflowExecution" => DescribeWorkflowExecutionResponse { + workflow_execution_info: Some(WorkflowExecutionInfo { + execution: Some(WorkflowExecution { + workflow_id: "wf".to_string(), + run_id: "run-1".to_string(), + }), + first_run_id: "run-1".to_string(), + status: WorkflowExecutionStatus::Running as i32, + ..Default::default() + }), + ..Default::default() + } + .encode_to_vec(), + other => panic!("unexpected RPC {other}"), + }; + let resp = ClientGrpcOverrideResponse { + status_code: 0, + headers: ByteArrayRefArray::empty(), + success_proto: resp_raw.as_slice().into(), + fail_message: ByteArrayRef::empty(), + fail_details: ByteArrayRef::empty(), + }; + temporal_core_client_grpc_override_request_respond(req, resp); + let _ = resp_raw; +} + +fn memory_store_config() -> Vec { + use temporalio_sdk_core::streams::proto::{ + MemoryStoreConfig, StreamStoreConfig, stream_store_config, + }; + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + } + .encode_to_vec() +} + +#[test] +fn test_stream_store_appends_and_reads_through_the_c_bridge() { + use temporalio_common::protos::temporal::api::common::v1::Payload; + use temporalio_sdk_core::streams::proto::{ + AppendRecord, AppendRequest, AppendResponse, NamedProducer, ReadRequest, ReadResponse, + RetireProducerRequest, StreamAddress, StreamOwnerKind, StreamRecordKind, append_request, + read_record, retire_producer_request, + }; + Context::with(|context| { + context.runtime_new().unwrap(); + context + .client_connect_with_override( + Box::new(default_connection_options("127.0.0.1:4567")), + Some(streams_callback_override), + std::ptr::null_mut(), + ) + .unwrap(); + let store = context.stream_store_new(memory_store_config()).unwrap(); + let address = StreamAddress { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: "wf".to_string(), + run_id: String::new(), + topic: "out".to_string(), + }; + + let appended = context + .stream_store_call( + &store, + "Append", + AppendRequest { + stream: Some(address.clone()), + producer: Some(append_request::Producer::Named(NamedProducer { + producer_id: "p".to_string(), + attempt: 1, + })), + sequence: 1, + digest: vec![1; 32], + records: vec![AppendRecord { + kind: StreamRecordKind::Data as i32, + body: Some(Payload { + data: b"encrypted".to_vec(), + ..Default::default() + }), + content_hash: vec![7; 32], + }], + } + .encode_to_vec(), + ) + .unwrap() + .unwrap(); + let appended = AppendResponse::decode(appended.as_slice()).unwrap(); + + let read = context + .stream_store_call( + &store, + "Read", + ReadRequest { + stream: Some(address.clone()), + ..Default::default() + } + .encode_to_vec(), + ) + .unwrap() + .unwrap(); + let read = ReadResponse::decode(read.as_slice()).unwrap(); + assert_eq!(read.cursor, appended.last_cursor); + let [record] = read.records.as_slice() else { + panic!("expected one record, got {:?}", read.records); + }; + let Some(read_record::Record::Stored(stored)) = &record.record else { + panic!("expected a stored record, got {record:?}"); + }; + assert_eq!(stored.body.as_ref().unwrap().data, b"encrypted"); + + // The producer's last append returned, so its dedupe state may go. + let retired = context + .stream_store_call( + &store, + "RetireProducer", + RetireProducerRequest { + stream: Some(address), + producer: Some(retire_producer_request::Producer::Named(NamedProducer { + producer_id: "p".to_string(), + attempt: 1, + })), + } + .encode_to_vec(), + ) + .unwrap() + .unwrap(); + assert!(retired.is_empty()); + }); +} + +#[test] +fn test_a_cancelled_stream_read_says_it_only_stopped_reading() { + use temporalio_sdk_core::streams::proto::{ + ReadRequest, StreamAddress, StreamFailure, StreamOwnerKind, + }; + Context::with(|context| { + context.runtime_new().unwrap(); + context + .client_connect_with_override( + Box::new(default_connection_options("127.0.0.1:4567")), + Some(streams_callback_override), + std::ptr::null_mut(), + ) + .unwrap(); + let store = context.stream_store_new(memory_store_config()).unwrap(); + let token = crate::temporal_core_cancellation_token_new(); + crate::temporal_core_cancellation_token_cancel(token); + let read = ReadRequest { + stream: Some(StreamAddress { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: "wf".to_string(), + run_id: String::new(), + topic: "out".to_string(), + }), + after: "$end".to_string(), + wait: Some(std::time::Duration::from_secs(30).try_into().unwrap()), + ..Default::default() + }; + let failure = context + .stream_store_call_with_token(&store, "Read", read.encode_to_vec(), token) + .unwrap() + .unwrap_err(); + crate::temporal_core_cancellation_token_free(token); + let failure = StreamFailure::decode(failure.as_slice()).unwrap(); + assert_eq!(failure.message, "Cancelled before the read answered"); + }); +} + +#[test] +fn test_stream_store_failures_cross_the_c_bridge_as_stream_failures() { + use temporalio_sdk_core::streams::proto::{StreamFailure, StreamFailureKind}; + Context::with(|context| { + context.runtime_new().unwrap(); + context + .client_connect_with_override( + Box::new(default_connection_options("127.0.0.1:4567")), + Some(streams_callback_override), + std::ptr::null_mut(), + ) + .unwrap(); + let bad_config = context.stream_store_new(vec![0xff, 0xff]).unwrap_err(); + assert!( + bad_config + .to_string() + .contains("Invalid stream store config"), + "{bad_config}" + ); + + let store = context.stream_store_new(memory_store_config()).unwrap(); + let failure = context + .stream_store_call(&store, "Truncate", vec![]) + .unwrap() + .unwrap_err(); + let failure = StreamFailure::decode(failure.as_slice()).unwrap(); + assert_eq!(failure.kind(), StreamFailureKind::Unsupported); + assert!(failure.message.contains("Truncate")); + }); +} diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 2fb14def13..d7faf8a18e 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -33,6 +33,19 @@ relevant information. ## Unreleased +### Added +* Experimental: behind the new `streams` feature, `streams::connect_stream_service` builds one + stream store from a `StreamStoreConfig` (Redis or memory) and serves lang's stream calls in + process: append, read, latest, close and owner delete, as serialized `coresdk.streams` + requests. +* Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the + stream service, built with the Redis store. +* Experimental: the stream service's `RetireProducer` call drops one producer attempt's dedupe + state on a stream, which the Redis store keeps as one field of the topic's meta hash until + retention. A producer that writes once per short-lived attempt calls it after the attempt's + last append returned, so the meta hash tracks the attempts in flight. Calling it earlier lets a + retried batch land twice. + ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting nondeterminism while retaining compatibility with histories recorded before these checks. diff --git a/crates/sdk-core/src/lib.rs b/crates/sdk-core/src/lib.rs index e7608a603e..dccb420231 100644 --- a/crates/sdk-core/src/lib.rs +++ b/crates/sdk-core/src/lib.rs @@ -26,6 +26,8 @@ mod pollers; mod protosext; pub mod replay; pub(crate) mod retry_logic; +#[cfg(feature = "streams")] +pub mod streams; pub mod telemetry; mod worker; diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs new file mode 100644 index 0000000000..c36fb3e472 --- /dev/null +++ b/crates/sdk-core/src/streams.rs @@ -0,0 +1,181 @@ +//! Streams for lang: one process-wide stream service over the configured store, asking the +//! client's server about streams' owners. + +use prost::Message; +use std::sync::Arc; +use temporalio_client::Connection; +use temporalio_common::protos::google::rpc::Status as RpcStatus; +use temporalio_common::protos::temporal::api::{ + common::v1::WorkflowExecution, + workflowservice::v1::{DescribeWorkflowExecutionRequest, DescribeWorkflowExecutionResponse}, +}; +use temporalio_streams::{OwnerClient, OwnerDescription, OwnerError}; +pub use temporalio_streams::{StreamError, StreamResult, StreamService, StreamStore, proto}; +use tonic::IntoRequest; + +/// Connects to the store `config` names. The service asks `connection`'s server about streams' +/// owners. +pub async fn connect_stream_service( + config: proto::StreamStoreConfig, + connection: Connection, +) -> StreamResult> { + let owner = Arc::new(ConnectionOwnerClient { connection }); + Ok(Arc::new(StreamService::connect(config, owner).await?)) +} + +/// Asks a Temporal server about owners through a client connection. +struct ConnectionOwnerClient { + connection: Connection, +} + +#[async_trait::async_trait] +impl OwnerClient for ConnectionOwnerClient { + async fn describe( + &self, + namespace: &str, + workflow_id: &str, + run_id: &str, + ) -> Result { + let response = self + .connection + .workflow_service() + .describe_workflow_execution( + DescribeWorkflowExecutionRequest { + namespace: namespace.to_string(), + execution: Some(WorkflowExecution { + workflow_id: workflow_id.to_string(), + run_id: run_id.to_string(), + }), + } + .into_request(), + ) + .await + .map_err(owner_error)?; + owner_description(response.into_inner()) + } +} + +fn owner_error(status: tonic::Status) -> OwnerError { + // A missing namespace says nothing about the Workflow. A failover or a namespace being + // set up answers it for a while, and a chain must not close on it. + if status.code() == tonic::Code::NotFound && !namespace_not_found(&status) { + OwnerError::NotFound(status.message().to_string()) + } else { + OwnerError::Failed(status.to_string()) + } +} + +fn namespace_not_found(status: &tonic::Status) -> bool { + RpcStatus::decode(status.details()).is_ok_and(|details| { + details.details.iter().any(|detail| { + detail + .type_url + .ends_with("/temporal.api.errordetails.v1.NamespaceNotFoundFailure") + }) + }) +} + +fn owner_description( + response: DescribeWorkflowExecutionResponse, +) -> Result { + let info = response + .workflow_execution_info + .ok_or_else(|| OwnerError::Failed("the server described no execution".to_string()))?; + let run_id = info + .execution + .as_ref() + .map(|execution| execution.run_id.clone()) + .unwrap_or_default(); + // A run that started its chain may leave the first run id unset. + let first_run_id = if info.first_run_id.is_empty() { + run_id.clone() + } else { + info.first_run_id.clone() + }; + Ok(OwnerDescription { + status: info.status(), + run_id, + first_run_id, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use temporalio_common::protos::temporal::api::{ + enums::v1::WorkflowExecutionStatus, workflow::v1::WorkflowExecutionInfo, + }; + + fn described(run_id: &str, first_run_id: &str) -> DescribeWorkflowExecutionResponse { + DescribeWorkflowExecutionResponse { + workflow_execution_info: Some(WorkflowExecutionInfo { + execution: Some(WorkflowExecution { + workflow_id: "wf".to_string(), + run_id: run_id.to_string(), + }), + first_run_id: first_run_id.to_string(), + status: WorkflowExecutionStatus::ContinuedAsNew as i32, + ..Default::default() + }), + ..Default::default() + } + } + + #[test] + fn a_description_names_the_run_its_chain_and_its_status() { + assert_eq!( + owner_description(described("run-2", "run-1")).unwrap(), + OwnerDescription { + run_id: "run-2".to_string(), + first_run_id: "run-1".to_string(), + status: WorkflowExecutionStatus::ContinuedAsNew, + } + ); + } + + #[test] + fn a_run_without_a_first_run_id_starts_its_chain() { + assert_eq!( + owner_description(described("run-1", "")) + .unwrap() + .first_run_id, + "run-1" + ); + } + + #[test] + fn a_missing_namespace_is_a_failure_not_a_gone_owner() { + let details = RpcStatus { + code: tonic::Code::NotFound as i32, + message: "namespace ns not found".to_string(), + details: vec![prost_types::Any { + type_url: + "type.googleapis.com/temporal.api.errordetails.v1.NamespaceNotFoundFailure" + .to_string(), + value: Vec::new(), + }], + }; + let status = tonic::Status::with_details( + tonic::Code::NotFound, + "namespace ns not found", + details.encode_to_vec().into(), + ); + assert!(matches!(owner_error(status), OwnerError::Failed(_))); + } + + #[test] + fn not_found_says_the_owner_is_gone_and_anything_else_that_the_server_failed() { + assert!(matches!( + owner_error(tonic::Status::not_found("no such workflow")), + OwnerError::NotFound(message) if message == "no such workflow" + )); + assert!(matches!( + owner_error(tonic::Status::unavailable("down")), + OwnerError::Failed(_) + )); + assert!(matches!( + owner_description(DescribeWorkflowExecutionResponse::default()), + Err(OwnerError::Failed(_)) + )); + } +} diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index 44e2128292..08193a4177 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -12,6 +12,14 @@ count, its first and last entry ids, and lang's digest of it in hex. The script new batch once the chain or the topic is closed, in the same step as the write, so no batch lands after a close returned. Only that topic check differs from the SDKs' script. +**Retired producers.** A field stays until a write's trim finds it older than the retention, so +a topic written by many short-lived producers holds one field per producer attempt of the last +retention window, about 235 bytes each. `RetireProducer` deletes one attempt's field with +`HDEL` and touches nothing else. A missing field or meta is no failure. Call it only once the +attempt's last append returned: a retry of an earlier batch would find no field to compare with +and land a second time. Readers lose nothing, since a store compares a batch with its own +attempt's field only and readers find a newer attempt from the records. + **Retention.** Every script that writes a log trims it with `XTRIM MINID` to the retention and slides the expiry of the log and its meta, so a stream dies a retention after its last write and nobody cleans up when the Workflow closes. The meta outlives its log by a 30-day grace, as diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 036b851db0..d316fecb6f 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -8,10 +8,11 @@ use crate::{ check_append_digest, mint_cursor, owner::{OwnerClient, OwnerDescription, OwnerError}, proto::{ - AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, DeleteOwnerRequest, - DeleteOwnerResponse, LatestRequest, LatestResponse, ReadRequest, ReadResponse, - StoreAppendRequest, StoreLatestRequest, StreamAddress, StreamOwnerKind, - append_request::Producer, + ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, + DeleteOwnerRequest, DeleteOwnerResponse, LatestRequest, LatestResponse, NamedProducer, + ReadRequest, ReadResponse, RetireProducerRequest, RetireProducerResponse, + StoreAppendRequest, StoreLatestRequest, StreamAddress, StreamOwnerKind, append_request, + retire_producer_request, }, reader::{ReadState, read_with}, stored_append_record, stream_hash, @@ -181,26 +182,14 @@ impl Streams { pub async fn append(&self, request: AppendRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; let (producer_id, attempt) = match &request.producer { - Some(Producer::Named(named)) => { - if named.producer_id.is_empty() { - return Err(StreamError::invalid_argument("a producer needs an id")); - } - (named.producer_id.clone(), named.attempt) - } - Some(Producer::Activity(activity)) => { - (activity_producer_id(activity), i64::from(activity.attempt)) - } + Some(append_request::Producer::Named(named)) => producer(Some(named), None)?, + Some(append_request::Producer::Activity(activity)) => producer(None, Some(activity))?, None => { return Err(StreamError::invalid_argument( "an append needs its producer", )); } }; - if attempt < 1 { - return Err(StreamError::invalid_argument(format!( - "a producer's attempt starts at one, not {attempt}" - ))); - } if request.sequence < 1 { return Err(StreamError::invalid_argument(format!( "a producer's sequence starts at one, not {}", @@ -214,11 +203,7 @@ impl Streams { } check_append_digest(&request.digest)?; let chain = self - .producer_chain(ProducerKey { - owner: owner_key(stream), - producer_id: producer_id.clone(), - attempt, - }) + .producer_chain(producer_key(owner_key(stream), &producer_id, attempt)) .await?; let records = request .records @@ -389,6 +374,40 @@ impl Streams { Ok(CloseResponse {}) } + /// Drops one producer attempt's dedupe state on a stream, on the chain that attempt wrote + /// on. It refuses nothing on a closed stream, and a stream or attempt with no state answers + /// the same. + /// + /// Only safe once the attempt's last append returned: a retry of an earlier batch would land + /// a second time. Nothing else is lost. A store compares a batch with its own attempt's + /// state only, and readers find a newer attempt from the records. + pub async fn retire_producer( + &self, + request: RetireProducerRequest, + ) -> StreamResult { + let stream = address(request.stream.as_ref())?; + let (producer_id, attempt) = match &request.producer { + Some(retire_producer_request::Producer::Named(named)) => producer(Some(named), None)?, + Some(retire_producer_request::Producer::Activity(activity)) => { + producer(None, Some(activity))? + } + None => { + return Err(StreamError::invalid_argument("a retire needs its producer")); + } + }; + let key = producer_key(owner_key(stream), &producer_id, attempt); + // The chain the attempt's appends resolved, even when a newer chain took the Workflow + // id since, so the state goes from where it was written. + let chain = match self.producers.get(&key) { + Some(checked) => chain_of_key(&key.owner, &checked.first_run_id), + None => self.chain(stream).await?, + }; + self.store + .retire_producer(&chain, &stream.topic, &key.producer_id, attempt) + .await?; + Ok(RetireProducerResponse {}) + } + /// Deletes every stream of one owner, across all its chains. It asks Temporal nothing, so it /// works after the owner is gone. Delete only after the chain closed and its producers /// stopped, since a late append writes a fresh stream with no dedupe state and no close mark. @@ -593,6 +612,59 @@ fn address(stream: Option<&StreamAddress>) -> StreamResult<&StreamAddress> { } /// Refuses a topic no store keys reliably, by the rules Core's Workflow path also applies. +/// The producer id and attempt a named producer or an Activity attempt writes as. +fn producer( + named: Option<&NamedProducer>, + activity: Option<&ActivityProducer>, +) -> StreamResult<(String, i64)> { + let (producer_id, attempt) = match (named, activity) { + (Some(named), _) => { + if named.producer_id.is_empty() { + return Err(StreamError::invalid_argument("a producer needs an id")); + } + // A Workflow run writes its own output as `@`. A named producer with that id + // would share the run's dedupe session and could withdraw its records. + if named.producer_id.starts_with('@') { + return Err(StreamError::invalid_argument(format!( + "a named producer's id can't start with `@`, which names a Workflow run's \ + own output, not {:?}", + named.producer_id + ))); + } + (named.producer_id.clone(), named.attempt) + } + (None, Some(activity)) => (activity_producer_id(activity), i64::from(activity.attempt)), + (None, None) => return Err(StreamError::invalid_argument("a call needs its producer")), + }; + if attempt < 1 { + return Err(StreamError::invalid_argument(format!( + "a producer's attempt starts at one, not {attempt}" + ))); + } + Ok((producer_id, attempt)) +} + +/// The key a producer attempt's chain is remembered by. A Workflow writes through a new local +/// producer per flush, `__temporal__@`, and every flush of a run shares the +/// run's chain. So they share one entry, and a busy run can't push the long-lived producers out. +fn producer_key(owner: OwnerKey, producer_id: &str, attempt: i64) -> ProducerKey { + match producer_id + .strip_prefix("__temporal_") + .and_then(|rest| rest.rsplit_once('@')) + { + Some((_, run_id)) => ProducerKey { + owner, + producer_id: format!("__temporal_@{run_id}"), + attempt: 0, + }, + None => ProducerKey { + owner, + producer_id: producer_id.to_string(), + attempt, + }, + } +} + fn check_topic(topic: &str) -> StreamResult<()> { temporalio_common::streams::check_topic(topic).map_err(StreamError::invalid_argument) } @@ -674,7 +746,7 @@ mod tests { MemoryStore, proto::{ ActivityProducer, AppendRecord, NamedProducer, StoreReadRequest, StreamFailureKind, - StreamRecordKind, + StreamRecordKind, append_request::Producer, }, }; use sha2::{Digest, Sha256}; @@ -1113,6 +1185,41 @@ mod tests { assert!(chain_closed(&setup.store, "run-1").await); } + #[tokio::test] + async fn a_workflows_local_producers_share_one_memory_entry() { + let setup = setup(StreamsOptions { + remembered: 2, + ..StreamsOptions::default() + }); + setup + .streams + .append(append("long-lived", 1, "1")) + .await + .unwrap(); + for flush in 1..=20 { + let producer = format!("__temporal_local_{flush}@run-1"); + setup + .streams + .append(append(&producer, 1, "x")) + .await + .unwrap(); + } + let key = producer_key(owner_key(&stream("out")), "long-lived", 1); + assert!(setup.streams.producers.get(&key).is_some()); + } + + #[tokio::test] + async fn a_named_producer_can_not_write_as_a_workflow_run() { + let setup = setup(StreamsOptions::default()); + let error = setup + .streams + .append(append("@run-1", 1, "1")) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::InvalidArgument); + assert!(error.message.contains("@run-1"), "{error}"); + } + #[tokio::test(start_paused = true)] async fn one_not_found_from_the_owner_check_closes_nothing() { let setup = setup(StreamsOptions::default()); @@ -1266,6 +1373,51 @@ mod tests { setup.streams.append(other).await.unwrap(); } + #[tokio::test] + async fn a_retire_drops_the_attempts_state_on_the_chain_it_wrote() { + let setup = setup(StreamsOptions::default()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup + .streams + .close(CloseRequest { + stream: Some(stream("out")), + }) + .await + .unwrap(); + let before = setup.owners.describes(); + let retire = |producer: Option| RetireProducerRequest { + stream: Some(stream("out")), + producer, + }; + let named = |attempt| { + Some(retire_producer_request::Producer::Named(NamedProducer { + producer_id: "p".to_string(), + attempt, + })) + }; + // A closed stream refuses no retire, and a repeat answers the same. + for _ in 0..2 { + setup + .streams + .retire_producer(retire(named(1))) + .await + .unwrap(); + } + // The append resolved the chain, so the retire asks Temporal nothing. + assert_eq!(setup.owners.describes(), before); + // With the state gone, the repeat is a new batch, which the closed stream refuses. + assert_eq!( + kind(setup.streams.append(append("p", 1, "1")).await), + StreamFailureKind::Closed + ); + for bad in [retire(None), retire(named(0))] { + assert_eq!( + kind(setup.streams.retire_producer(bad).await), + StreamFailureKind::InvalidArgument + ); + } + } + fn quick() -> StreamsOptions { StreamsOptions { owner_check_min: Duration::from_millis(20), diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 42fc7bd763..02a1dceea1 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -17,6 +17,7 @@ mod reader; mod record; #[cfg(feature = "redis")] mod redis; +mod service; mod store; #[cfg(feature = "redis")] @@ -32,6 +33,7 @@ pub use reader::{ pub use record::{ WORKFLOW_OWNER_KIND, activity_producer_id, check_append_digest, stored_append_record, }; +pub use service::StreamService; pub use store::StreamStore; /// The protos stores and lang exchange. diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index a190c3bfe7..d92ed331e5 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -361,6 +361,23 @@ impl StreamStore for MemoryStore { Ok(()) } + async fn retire_producer( + &self, + chain: &ChainId, + topic: &str, + producer_id: &str, + attempt: i64, + ) -> StreamResult<()> { + let mut chains = self.chains.lock().unwrap(); + if let Some(topic) = chains + .get_mut(&chain_key(chain)) + .and_then(|chain| chain.topics.get_mut(topic)) + { + topic.held.remove(&(producer_id.to_string(), attempt)); + } + Ok(()) + } + async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { if request.owner_kind != StreamOwnerKind::Workflow as i32 { return Err(StreamError::unsupported(format!( diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index 48bf3ad6da..c70ea74921 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -460,6 +460,10 @@ mod tests { unreachable!() } + async fn retire_producer(&self, _: &ChainId, _: &str, _: &str, _: i64) -> StreamResult<()> { + unreachable!() + } + async fn delete_owner(&self, _: DeleteOwnerRequest) -> StreamResult { unreachable!() } diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 9a1cd0db7a..2c02485e24 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -618,6 +618,23 @@ impl StreamStore for RedisStore { .mapped(Call::Write) } + async fn retire_producer( + &self, + chain: &ChainId, + topic: &str, + producer_id: &str, + attempt: i64, + ) -> StreamResult<()> { + // One field of one key, so a plain `HDEL` is atomic, and it creates nothing when the meta + // is gone. + redis::cmd("HDEL") + .arg(ChainKeys::new(&self.prefix, chain).meta(topic)) + .arg(session_field(producer_id, attempt)) + .query_async::<()>(&mut self.shared.clone()) + .await + .mapped(Call::Write) + } + async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { if request.owner_kind != StreamOwnerKind::Workflow as i32 { return Err(StreamError::unsupported(format!( diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs new file mode 100644 index 0000000000..e3a695ac49 --- /dev/null +++ b/crates/streams/src/service.rs @@ -0,0 +1,470 @@ +//! The stream calls lang makes, served in process. + +use crate::{ + OwnerClient, StreamError, StreamResult, StreamStore, Streams, StreamsOptions, + proto::{ + AppendRequest, CloseRequest, DeleteOwnerRequest, LatestRequest, ReadRequest, + RetireProducerRequest, StreamFailure, StreamStoreConfig, stream_store_config, + }, +}; +use prost::Message; +use std::{sync::Arc, time::Duration}; + +/// How long a store keeps records when its configuration doesn't say. +const DEFAULT_RETENTION: Duration = Duration::from_secs(7 * 24 * 60 * 60); +/// The most records one read answers with, as the stream service contract allows. +const MAX_READ_RECORDS: u32 = 1000; +/// The longest one read waits, as the stream service contract allows. A longer wait would hold a +/// store connection that other reads need. +const MAX_READ_WAIT: Duration = Duration::from_secs(60); + +/// One process's stream store and the calls lang makes on it. +/// +/// Lang builds one from a [StreamStoreConfig] and sends its stream calls to +/// [StreamService::call] as serialized `coresdk.streams` requests. +pub struct StreamService { + store: Arc, + streams: Streams, +} + +impl std::fmt::Debug for StreamService { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StreamService") + .field("store", &self.store.name()) + .finish() + } +} + +impl StreamService { + /// Connects to the store `config` names, asking `owner` about streams' owners. + pub async fn connect( + config: StreamStoreConfig, + owner: Arc, + ) -> StreamResult { + let (store, retention): (Arc, Duration) = match config.store { + Some(stream_store_config::Store::Memory(_)) => { + (Arc::new(crate::MemoryStore::new()), DEFAULT_RETENTION) + } + Some(stream_store_config::Store::Redis(redis)) => connect_redis(redis).await?, + None => { + return Err(StreamError::refused( + "the stream store config names no store", + )); + } + }; + Ok(Self::new( + store, + owner, + StreamsOptions::for_retention(retention), + )) + } + + /// Serves calls on `store`. + pub fn new( + store: Arc, + owner: Arc, + options: StreamsOptions, + ) -> Self { + let streams = Streams::new(store.clone(), owner, options); + Self { store, streams } + } + + /// Serves one call: `rpc` is a `StreamService` method name and `request` its serialized + /// request. Answers with the serialized response, or the failure the call ended with. + pub async fn call(&self, rpc: &str, request: &[u8]) -> Result, StreamFailure> { + let response = match rpc { + "Append" => self + .streams + .append(decode::(rpc, request)?) + .await + .map(|r| r.encode_to_vec()), + "Read" => self + .streams + .read(capped(decode::(rpc, request)?)) + .await + .map(|r| r.encode_to_vec()), + "Latest" => self + .streams + .latest(decode::(rpc, request)?) + .await + .map(|r| r.encode_to_vec()), + "Close" => self + .streams + .close(decode::(rpc, request)?) + .await + .map(|r| r.encode_to_vec()), + "DeleteOwner" => self + .streams + .delete_owner(decode::(rpc, request)?) + .await + .map(|r| r.encode_to_vec()), + "RetireProducer" => self + .streams + .retire_producer(decode::(rpc, request)?) + .await + .map(|r| r.encode_to_vec()), + other => Err(StreamError::unsupported(format!( + "the stream service has no call {other:?}" + ))), + }; + response.map_err(StreamFailure::from) + } +} + +/// `request` with its record count and wait cut to what Core allows. +fn capped(mut request: ReadRequest) -> ReadRequest { + request.max_records = request.max_records.min(MAX_READ_RECORDS); + let too_long = request + .wait + .and_then(|wait| Duration::try_from(wait).ok()) + .is_some_and(|wait| wait > MAX_READ_WAIT); + if too_long { + request.wait = MAX_READ_WAIT.try_into().ok(); + } + request +} + +fn decode(rpc: &str, request: &[u8]) -> Result { + T::decode(request).map_err(|error| { + StreamError::invalid_argument(format!("the {rpc} request did not decode ({error})")).into() + }) +} + +#[cfg(feature = "redis")] +async fn connect_redis( + config: crate::proto::RedisStoreConfig, +) -> StreamResult<(Arc, Duration)> { + let mut options = crate::RedisStoreOptions::new(String::new()); + options.urls = config.urls; + options.cluster = config.cluster; + if !config.key_prefix.is_empty() { + options.key_prefix = config.key_prefix; + } + if let Some(retention) = config.retention { + options.retention = duration(retention.seconds, retention.nanos, "retention")?; + } + if config.blocking_reads_per_node > 0 { + options.blocking_reads_per_node = config.blocking_reads_per_node as usize; + } + if let Some(timeout) = config.response_timeout { + options.response_timeout = duration(timeout.seconds, timeout.nanos, "response timeout")?; + } + let retention = options.retention; + let store = crate::RedisStore::connect(options).await?; + Ok((Arc::new(store), retention)) +} + +#[cfg(not(feature = "redis"))] +async fn connect_redis( + _: crate::proto::RedisStoreConfig, +) -> StreamResult<(Arc, Duration)> { + Err(StreamError::unsupported( + "this build has no Redis store; build temporalio-streams with its redis feature", + )) +} + +#[cfg(feature = "redis")] +fn duration(seconds: i64, nanos: i32, name: &str) -> StreamResult { + match (u64::try_from(seconds), u32::try_from(nanos)) { + (Ok(seconds), Ok(nanos)) if seconds > 0 || nanos > 0 => Ok(Duration::new(seconds, nanos)), + _ => Err(StreamError::refused(format!( + "the Redis store's {name} must be positive" + ))), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + OwnerDescription, OwnerError, + proto::{ + AppendRecord, AppendResponse, LatestResponse, MemoryStoreConfig, NamedProducer, + ReadResponse, StreamAddress, StreamFailureKind, StreamOwnerKind, StreamRecordKind, + append_request, read_record, + }, + }; + use temporalio_common::protos::temporal::api::{ + common::v1::Payload, enums::v1::WorkflowExecutionStatus, + }; + + /// Every Workflow's latest run is `run-1`, the first of its chain, and still running. + struct Running; + + #[async_trait::async_trait] + impl OwnerClient for Running { + async fn describe( + &self, + _: &str, + _: &str, + _: &str, + ) -> Result { + Ok(OwnerDescription { + run_id: "run-1".to_string(), + first_run_id: "run-1".to_string(), + status: WorkflowExecutionStatus::Running, + }) + } + } + + async fn service() -> StreamService { + StreamService::connect( + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + }, + Arc::new(Running), + ) + .await + .unwrap() + } + + fn address() -> StreamAddress { + StreamAddress { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: "wf".to_string(), + run_id: String::new(), + topic: "out".to_string(), + } + } + + fn append(sequence: i64, values: &[&str]) -> Vec { + AppendRequest { + stream: Some(address()), + producer: Some(append_request::Producer::Named(NamedProducer { + producer_id: "p".to_string(), + attempt: 1, + })), + sequence, + digest: vec![sequence as u8; 32], + records: values + .iter() + .map(|value| AppendRecord { + kind: StreamRecordKind::Data as i32, + body: Some(Payload { + data: value.as_bytes().to_vec(), + ..Default::default() + }), + content_hash: vec![value.as_bytes()[0]; 32], + }) + .collect(), + } + .encode_to_vec() + } + + #[tokio::test] + async fn an_append_reads_back_through_the_service() { + let service = service().await; + let appended = AppendResponse::decode( + service + .call("Append", &append(1, &["a", "b"])) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + let read = ReadResponse::decode( + service + .call( + "Read", + &ReadRequest { + stream: Some(address()), + ..Default::default() + } + .encode_to_vec(), + ) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + let bodies: Vec<_> = read + .records + .iter() + .map(|r| match &r.record { + Some(read_record::Record::Stored(stored)) => stored.body.clone().unwrap().data, + other => panic!("expected a stored record, got {other:?}"), + }) + .collect(); + assert_eq!(bodies, [b"a".to_vec(), b"b".to_vec()]); + assert_eq!(read.cursor, appended.last_cursor); + + let latest = LatestResponse::decode( + service + .call( + "Latest", + &LatestRequest { + stream: Some(address()), + } + .encode_to_vec(), + ) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + assert_eq!(latest.cursor, appended.last_cursor); + } + + fn read(request: ReadRequest) -> Vec { + ReadRequest { + stream: Some(address()), + ..request + } + .encode_to_vec() + } + + #[tokio::test] + async fn a_read_answers_at_most_a_thousand_records() { + let service = service().await; + let values: Vec = (0..1001).map(|i| format!("v{i}")).collect(); + let values: Vec<&str> = values.iter().map(String::as_str).collect(); + service.call("Append", &append(1, &values)).await.unwrap(); + let answer = service + .call( + "Read", + &read(ReadRequest { + max_records: 1_000_000, + ..Default::default() + }), + ) + .await + .unwrap(); + let read = ReadResponse::decode(answer.as_slice()).unwrap(); + assert_eq!(read.records.len(), 1000); + } + + #[tokio::test(start_paused = true)] + async fn a_read_waits_at_most_a_minute() { + let service = service().await; + let started = tokio::time::Instant::now(); + service + .call( + "Read", + &read(ReadRequest { + after: "$end".to_string(), + wait: Some(Duration::from_secs(3600).try_into().unwrap()), + ..Default::default() + }), + ) + .await + .unwrap(); + assert!( + started.elapsed() <= Duration::from_secs(61), + "{:?}", + started.elapsed() + ); + } + + #[tokio::test] + async fn a_failure_crosses_with_its_kind() { + let service = service().await; + service + .call("Append", &append(1, &["a", "b"])) + .await + .unwrap(); + let failure = service + .call("Append", &append(2, &["c"])) + .await + .unwrap_err(); + assert_eq!( + failure.kind(), + StreamFailureKind::ProducerStale, + "{failure:?}" + ); + } + + #[tokio::test] + async fn an_unknown_call_or_a_bad_request_is_refused() { + let service = service().await; + let unknown = service.call("Truncate", &[]).await.unwrap_err(); + assert_eq!(unknown.kind(), StreamFailureKind::Unsupported); + assert!(unknown.message.contains("Truncate")); + let garbled = service.call("Append", &[0xff, 0xff]).await.unwrap_err(); + assert_eq!(garbled.kind(), StreamFailureKind::InvalidArgument); + assert!(garbled.message.contains("did not decode")); + } + + #[tokio::test] + async fn a_retired_producer_writes_its_batch_again() { + let service = service().await; + let first = AppendResponse::decode( + service + .call("Append", &append(1, &["a"])) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + let retire = crate::proto::RetireProducerRequest { + stream: Some(address()), + producer: Some(crate::proto::retire_producer_request::Producer::Named( + NamedProducer { + producer_id: "p".to_string(), + attempt: 1, + }, + )), + }; + let answer = service + .call("RetireProducer", &retire.encode_to_vec()) + .await + .unwrap(); + assert!(answer.is_empty()); + let again = AppendResponse::decode( + service + .call("Append", &append(1, &["a"])) + .await + .unwrap() + .as_slice(), + ) + .unwrap(); + assert_ne!(again.first_cursor, first.first_cursor); + } + + #[tokio::test] + async fn a_delete_is_checked_before_it_reaches_the_store() { + let service = service().await; + let other_owner = crate::proto::DeleteOwnerRequest { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Unspecified as i32, + workflow_id: "wf".to_string(), + }; + let failure = service + .call("DeleteOwner", &other_owner.encode_to_vec()) + .await + .unwrap_err(); + assert_eq!(failure.kind(), StreamFailureKind::Unsupported); + let no_id = crate::proto::DeleteOwnerRequest { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: String::new(), + }; + let failure = service + .call("DeleteOwner", &no_id.encode_to_vec()) + .await + .unwrap_err(); + assert_eq!(failure.kind(), StreamFailureKind::InvalidArgument); + } + + #[tokio::test] + async fn a_config_that_names_no_store_is_refused() { + let error = StreamService::connect(StreamStoreConfig { store: None }, Arc::new(Running)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Refused); + } + + #[cfg(not(feature = "redis"))] + #[tokio::test] + async fn redis_needs_the_redis_feature() { + let error = StreamService::connect( + StreamStoreConfig { + store: Some(stream_store_config::Store::Redis(Default::default())), + }, + Arc::new(Running), + ) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Unsupported); + } +} diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 36bfcd1dae..69b7a2d9f8 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -75,6 +75,19 @@ pub trait StreamStore: Send + Sync { /// Marks one topic of a chain closed, like [StreamStore::close_chain] for that topic alone. async fn close_topic(&self, chain: &ChainId, topic: &str) -> StreamResult<()>; + /// Drops the newest-batch record [StreamStore::append] keeps for one producer attempt on a + /// topic, and nothing else. A missing attempt or topic is no failure. + /// + /// Call it only once the attempt's last append returned. A retry of an earlier batch would + /// then find no record to compare with and land a second time. + async fn retire_producer( + &self, + chain: &ChainId, + topic: &str, + producer_id: &str, + attempt: i64, + ) -> StreamResult<()>; + /// Deletes every stream of one owner, across all its chains. async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult; } diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 391f533e21..cd54a18c2d 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -1122,6 +1122,45 @@ mod cases { } } + pub(crate) async fn retiring_a_producer_drops_only_its_dedupe_state(case: &dyn Case) { + let store = case.store(); + let chain = case.chain(); + let first = Producer::new(case, &chain, OUT, "p", 1) + .append(&["1"]) + .await; + let kept = Producer::new(case, &chain, OUT, "q", 1) + .append(&["2"]) + .await; + store.retire_producer(&chain, OUT, "p", 1).await.unwrap(); + // A repeat, an attempt that never wrote, a topic and a chain that hold nothing. + store.retire_producer(&chain, OUT, "p", 1).await.unwrap(); + store.retire_producer(&chain, OUT, "p", 2).await.unwrap(); + store.retire_producer(&chain, OTHER, "p", 1).await.unwrap(); + store + .retire_producer(&case.chain(), OUT, "p", 1) + .await + .unwrap(); + assert_eq!( + values(&read(case, &chain, OUT, BEGINNING).await.unwrap()), + ["1", "2"] + ); + // The other producer still dedupes. + let repeat = Producer::new(case, &chain, OUT, "q", 1) + .append(&["2"]) + .await; + assert_eq!(repeat, kept); + // The retired attempt no longer does, which is why only an attempt whose last append + // returned may be retired. + let again = Producer::new(case, &chain, OUT, "p", 1) + .append(&["1"]) + .await; + assert_ne!(again.first_position, first.first_position); + assert_eq!( + values(&read(case, &chain, OUT, BEGINNING).await.unwrap()), + ["1", "2", "1"] + ); + } + pub(crate) async fn deleting_an_owner_drops_every_chain_of_it(case: &dyn Case) { let store = case.store(); let chain = case.chain(); @@ -1204,6 +1243,7 @@ mod memory { a_closed_topic_leaves_the_others_open, a_closed_topic_answers_a_repeat_and_ends_reads, a_refused_append_leaves_no_trace_of_its_topic, + retiring_a_producer_drops_only_its_dedupe_state, deleting_an_owner_drops_every_chain_of_it, ); } @@ -1245,6 +1285,7 @@ macro_rules! redis_cases { a_closed_chain_refuses_new_batches_but_answers_a_repeat, a_closed_topic_leaves_the_others_open, a_closed_topic_answers_a_repeat_and_ends_reads, + retiring_a_producer_drops_only_its_dedupe_state, ); }; } diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 94fb45ddae..3ff150e9a6 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -187,6 +187,46 @@ async fn a_batch_is_one_script_and_keeps_one_high_water_field() { assert_eq!(parts[4], "05"); } +#[tokio::test] +async fn retiring_a_producer_deletes_its_session_field_only() { + let Some(mut setup) = setup().await else { + return; + }; + let chain = chain(); + for producer in ["p", "q"] { + setup + .store + .append(append(&chain, producer, 1, 1, b"\x01")) + .await + .unwrap(); + } + let meta = format!("{}:meta", log_key(&setup.prefix, &chain, "events")); + setup + .store + .retire_producer(&chain, "events", "p", 1) + .await + .unwrap(); + let fields: HashMap = setup.raw.hgetall(&meta).await.unwrap(); + assert!(!fields.contains_key("hw:1:p:1"), "{fields:?}"); + assert!(fields.contains_key("hw:1:q:1"), "{fields:?}"); + assert_eq!(fields["added"], "2"); + // A topic Redis holds nothing for keeps holding nothing. + setup + .store + .retire_producer(&chain, "unknown", "p", 1) + .await + .unwrap(); + let exists: bool = setup + .raw + .exists(format!( + "{}:meta", + log_key(&setup.prefix, &chain, "unknown") + )) + .await + .unwrap(); + assert!(!exists); +} + #[tokio::test] async fn appends_answer_as_the_python_sdk_recorded() { // The outcomes the Python provider's script gave, run again through the Core store.