From 36d68cb160736b7d4934a0c286b9d4052a9b09ce Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 12:24:33 -0700 Subject: [PATCH] Added the DescribeStore call, the memory mask and the Redis plain setting. Lang reads the store's capabilities and limits after connect. The memory store config takes a capability mask and the Redis config takes plain, so lang tests can run every fallback. --- .../temporal/sdk/core/streams/streams.proto | 52 +++++++++++++- crates/sdk-core-c-bridge/src/tests/mod.rs | 18 +++-- crates/streams/src/service.rs | 69 +++++++++++++++++-- crates/streams/src/store.rs | 53 ++++++++++++++ 4 files changed, 181 insertions(+), 11 deletions(-) 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 ab48869af7..192c9e1dbb 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -27,6 +27,8 @@ service StreamService { // 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); + // What the store can do beyond the log every store gives, and its limits. + rpc DescribeStore(DescribeStoreRequest) returns (DescribeStoreResponse); } // Which store holds a process's streams, and how to reach it. Lang builds one store from it and @@ -59,11 +61,18 @@ message RedisStoreConfig { 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; + // Runs with no Lua, so the store keeps no producer state and no last position, and Core takes + // their fallbacks. For tests of the fallbacks on a real store. + bool plain = 10; } // 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 {} +message MemoryStoreConfig { + // A mask. Only the capabilities it sets stay, and its nonzero limits replace the store's. + // Unset keeps every capability, so tests and users can try the weaker guarantees. + StoreCapabilities capabilities = 1; +} // What owns a stream. The owner decides when the stream closes. enum StreamOwnerKind { @@ -230,6 +239,47 @@ message RetireProducerRequest { // A producer attempt with no dedupe state left, or a stream that holds nothing, answers the same. message RetireProducerResponse {} +message DescribeStoreRequest {} + +message DescribeStoreResponse { + // The short name the store mints into its cursors, such as `redis` or `memory`. + string name = 1; + StoreCapabilities capabilities = 2; +} + +// What a store can do beyond the log every store gives. Core takes a fallback for each +// capability the store lacks, with the weaker guarantee the fallback names. +message StoreCapabilities { + // The append checks the producer attempt's dedupe entry and the close flags in the same step. + bool producer_state = 1; + // The append can fail unless the log's next position is the expected one. + bool conditional_tail = 2; + // The append can fail unless the stream's fence token is the expected one. + bool conditional_fence = 3; + // The store knows a topic's newest position after retention dropped every record. + bool kept_last = 4; + // A holder can keep records from retention. + bool holds = 5; + // One consumer can read every append in order. + bool change_feed = 6; + // A log per topic of each chain costs nothing much. + bool cheap_logs = 7; + // The store deletes every stream of one owner. + bool delete_owner = 8; + // The store lists the topics of one chain. + bool topic_listing = 9; + // A read can wait for the next record. + bool long_poll = 10; + // The next record's position is always the last one plus one. + bool dense_positions = 11; + // How long the store keeps records with no hold, when a positions-only reader can count on it. + google.protobuf.Duration replay_horizon = 12; + // The largest record the store takes, with its envelope. + uint64 max_record_bytes = 13; + // The most record bytes one append takes. + uint64 max_batch_bytes = 14; +} + // 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/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs index 95c3cabf12..b50e74beb0 100644 --- a/crates/sdk-core-c-bridge/src/tests/mod.rs +++ b/crates/sdk-core-c-bridge/src/tests/mod.rs @@ -507,7 +507,9 @@ fn memory_store_config() -> Vec { MemoryStoreConfig, StreamStoreConfig, stream_store_config, }; StreamStoreConfig { - store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + store: Some(stream_store_config::Store::Memory( + MemoryStoreConfig::default(), + )), ..Default::default() } .encode_to_vec() @@ -517,9 +519,9 @@ fn memory_store_config() -> Vec { 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, + AppendRecord, AppendRequest, AppendResponse, DescribeStoreResponse, NamedProducer, + ReadRequest, ReadResponse, RetireProducerRequest, StreamAddress, StreamOwnerKind, + StreamRecordKind, append_request, read_record, retire_producer_request, }; Context::with(|context| { context.runtime_new().unwrap(); @@ -605,6 +607,14 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() { .unwrap() .unwrap(); assert!(retired.is_empty()); + + let described = context + .stream_store_call(&store, "DescribeStore", Vec::new()) + .unwrap() + .unwrap(); + let described = DescribeStoreResponse::decode(described.as_slice()).unwrap(); + assert_eq!(described.name, "memory"); + assert!(described.capabilities.unwrap().producer_state); }); } diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 55ffdfe23f..685b389a2f 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -3,8 +3,9 @@ use crate::{ OwnerClient, StreamError, StreamResult, StreamStore, Streams, StreamsOptions, proto::{ - AppendRequest, CloseRequest, DeleteOwnerRequest, LatestRequest, ReadRequest, - RetireProducerRequest, StreamFailure, StreamStoreConfig, stream_store_config, + AppendRequest, CloseRequest, DeleteOwnerRequest, DescribeStoreRequest, + DescribeStoreResponse, LatestRequest, ReadRequest, RetireProducerRequest, StreamFailure, + StreamStoreConfig, stream_store_config, }, }; use prost::Message; @@ -42,8 +43,12 @@ impl StreamService { 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::Memory(memory)) => { + let store = match &memory.capabilities { + Some(mask) => crate::MemoryStore::with_capabilities(mask.into()), + None => crate::MemoryStore::new(), + }; + (Arc::new(store), DEFAULT_RETENTION) } Some(stream_store_config::Store::Redis(redis)) => connect_redis(redis).await?, None => { @@ -110,6 +115,14 @@ impl StreamService { .retire_producer(decode::(rpc, request)?) .await .map(|r| r.encode_to_vec()), + "DescribeStore" => { + decode::(rpc, request)?; + Ok(DescribeStoreResponse { + name: self.store.name().to_string(), + capabilities: Some(self.store.capabilities().into()), + } + .encode_to_vec()) + } other => Err(StreamError::unsupported(format!( "the stream service has no call {other:?}" ))), @@ -156,6 +169,7 @@ async fn connect_redis( if let Some(timeout) = config.response_timeout { options.response_timeout = duration(timeout.seconds, timeout.nanos, "response timeout")?; } + options.plain = config.plain; let retention = options.retention; let store = crate::RedisStore::connect(options).await?; Ok((Arc::new(store), retention)) @@ -217,7 +231,9 @@ mod tests { async fn service() -> StreamService { StreamService::connect( StreamStoreConfig { - store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + store: Some(stream_store_config::Store::Memory( + MemoryStoreConfig::default(), + )), ..Default::default() }, Arc::new(Running), @@ -454,6 +470,45 @@ mod tests { assert_eq!(failure.kind(), StreamFailureKind::InvalidArgument); } + #[tokio::test] + async fn describe_answers_the_stores_capabilities_and_limits() { + use crate::proto::StoreCapabilities; + let describe = async |service: &StreamService| { + let answer = service + .call("DescribeStore", &DescribeStoreRequest {}.encode_to_vec()) + .await + .unwrap(); + DescribeStoreResponse::decode(answer.as_slice()).unwrap() + }; + let full = describe(&service().await).await; + assert_eq!(full.name, "memory"); + let capabilities = full.capabilities.unwrap(); + assert!(capabilities.producer_state && capabilities.kept_last); + assert!(capabilities.dense_positions && !capabilities.holds); + assert_eq!(capabilities.max_record_bytes, 1536 * 1024); + assert_eq!(capabilities.max_batch_bytes, 4 << 20); + let masked = StreamService::connect( + StreamStoreConfig { + store: Some(stream_store_config::Store::Memory(MemoryStoreConfig { + capabilities: Some(StoreCapabilities { + long_poll: true, + max_record_bytes: 1024, + ..Default::default() + }), + })), + ..Default::default() + }, + Arc::new(Running), + ) + .await + .unwrap(); + let capabilities = describe(&masked).await.capabilities.unwrap(); + assert!(!capabilities.producer_state && !capabilities.kept_last); + assert!(capabilities.long_poll); + assert_eq!(capabilities.max_record_bytes, 1024); + assert_eq!(capabilities.max_batch_bytes, 4 << 20); + } + #[tokio::test] async fn a_config_that_names_no_store_is_refused() { let error = StreamService::connect(StreamStoreConfig::default(), Arc::new(Running)) @@ -465,7 +520,9 @@ mod tests { #[tokio::test] async fn a_negative_owner_start_grace_is_refused() { let mut config = StreamStoreConfig { - store: Some(stream_store_config::Store::Memory(MemoryStoreConfig {})), + store: Some(stream_store_config::Store::Memory( + MemoryStoreConfig::default(), + )), owner_start_grace: Some(Default::default()), }; config.owner_start_grace.as_mut().unwrap().seconds = -1; diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 1678fb5b48..478c3be397 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -109,6 +109,59 @@ impl Capabilities { } } +impl From for crate::proto::StoreCapabilities { + fn from(capabilities: Capabilities) -> Self { + Self { + producer_state: capabilities.producer_state, + conditional_tail: capabilities.conditional.tail, + conditional_fence: capabilities.conditional.fence, + kept_last: capabilities.kept_last, + holds: capabilities.holds, + change_feed: capabilities.change_feed, + cheap_logs: capabilities.cheap_logs, + delete_owner: capabilities.delete_owner, + topic_listing: capabilities.topic_listing, + long_poll: capabilities.long_poll, + dense_positions: capabilities.dense_positions, + replay_horizon: capabilities + .replay_horizon + .and_then(|horizon| horizon.try_into().ok()), + max_record_bytes: capabilities.max_record_bytes as u64, + max_batch_bytes: capabilities.max_batch_bytes as u64, + } + } +} + +impl From<&crate::proto::StoreCapabilities> for Capabilities { + /// A mask from its proto. A zero limit keeps the default. + fn from(mask: &crate::proto::StoreCapabilities) -> Self { + let limit = |bytes: u64, default: usize| match bytes { + 0 => default, + bytes => usize::try_from(bytes).unwrap_or(usize::MAX), + }; + Self { + producer_state: mask.producer_state, + conditional: ConditionForms { + tail: mask.conditional_tail, + fence: mask.conditional_fence, + }, + kept_last: mask.kept_last, + holds: mask.holds, + change_feed: mask.change_feed, + cheap_logs: mask.cheap_logs, + delete_owner: mask.delete_owner, + topic_listing: mask.topic_listing, + long_poll: mask.long_poll, + dense_positions: mask.dense_positions, + replay_horizon: mask + .replay_horizon + .and_then(|horizon| Duration::try_from(horizon).ok()), + max_record_bytes: limit(mask.max_record_bytes, MAX_RECORD_BYTES), + max_batch_bytes: limit(mask.max_batch_bytes, MAX_BATCH_BYTES), + } + } +} + impl Default for Capabilities { fn default() -> Self { Self::log_only()