[Brook] Added the DescribeStore call and the store config for the fallbacks.

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 {

crates/sdk-core-c-bridge/src/tests/mod.rs

@@ -507,7 +507,9 @@ fn memory_store_config() -> Vec<u8> {
         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<u8> {
 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);
     });
 }
 

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<dyn OwnerClient>,
     ) -> StreamResult<Self> {
         let (store, retention): (Arc<dyn StreamStore>, 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::<RetireProducerRequest>(rpc, request)?)
                 .await
                 .map(|r| r.encode_to_vec()),
+            "DescribeStore" => {
+                decode::<DescribeStoreRequest>(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;

crates/streams/src/store.rs

@@ -109,6 +109,59 @@ impl Capabilities {
     }
 }
 
+impl From<Capabilities> 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()