[Brook] Added Redis plain mode and the store's record and batch limits.

crates/sdk-core/src/core_tests/external_streams.rs

@@ -487,6 +487,32 @@ async fn a_store_that_refuses_the_append_fails_the_task_and_sends_no_completion(
     assert_eq!(store.calls(), ["append"], "a refusal is not sent again");
 }
 
+#[tokio::test]
+async fn a_record_above_the_stores_record_limit_fails_the_task() {
+    let store = Arc::new(RecordingStore {
+        inner: MemoryStore::with_capabilities(temporalio_streams::Capabilities {
+            max_record_bytes: 64,
+            ..MemoryStore::new().capabilities()
+        }),
+        ..Default::default()
+    });
+    let (worker, failed) = worker_failing_with(Some(store.clone()), &["record limit of 64"]);
+    let first = worker.poll_workflow_activation().await.unwrap();
+    complete(
+        &worker,
+        &first.run_id,
+        vec![
+            output_commit_command(records(&["a"])),
+            start_timer_cmd(1, Duration::from_secs(10)),
+        ],
+    )
+    .await;
+    failed_within_deadline(&failed).await;
+    worker.shutdown().await;
+    worker.finalize_shutdown().await;
+    assert!(store.appends().is_empty());
+}
+
 #[tokio::test]
 async fn a_live_commit_on_a_worker_without_a_store_fails_the_task() {
     let (worker, failed) = worker_failing_with(None, &["has no stream store"]);

crates/sdk-core/src/worker/workflow/managed_run.rs

@@ -953,7 +953,12 @@ impl ManagedRun {
             self.stream_attempt = attempt;
             self.committed_by_topic.clear();
         }
-        let topics = output_by_topic(commit.records, &mut self.committed_by_topic)?;
+        let max_batch_bytes = self.max_batch_bytes();
+        let topics = output_by_topic(
+            commit.records,
+            &mut self.committed_by_topic,
+            max_batch_bytes,
+        )?;
         self.stream_sessions
             .entry(attempt)
             .or_default()
@@ -969,6 +974,14 @@ impl ManagedRun {
         }))
     }
 
+    fn max_batch_bytes(&self) -> usize {
+        #[cfg(feature = "streams")]
+        if let Some(store) = &self.config.stream_store {
+            return store.capabilities().max_batch_bytes;
+        }
+        super::stream_output::MAX_APPEND_BYTES
+    }
+
     fn has_stream_store(&self) -> bool {
         #[cfg(feature = "streams")]
         {

crates/sdk-core/src/worker/workflow/stream_output.rs

@@ -51,19 +51,25 @@ pub(crate) struct RetiredSessions {
     pub(crate) chain_end: Option<i64>,
 }
 
-/// The most record bytes one append carries. A larger commit goes as several appends per topic, so
-/// no activation is too large to append and lang never fails a task for its size.
+/// The most record bytes one append carries when the store declares no limit. A larger commit goes
+/// as several appends per topic, so no activation is too large to append and lang never fails a
+/// task for its size.
 pub(crate) const MAX_APPEND_BYTES: usize = 4 << 20;
 
+/// What the store adds to each record besides its body, topic and hash, counted high: the
+/// producer id, the attempt, the sequence, the hash's metadata entry and the completed tasks.
+const ENVELOPE_BYTES: usize = 256;
+
 /// Splits `records` into appends per topic, in order of first publish, each of at most
-/// [MAX_APPEND_BYTES] past its first record, and counts them into `committed`, the records the
+/// `max_batch_bytes` past its first record, and counts them into `committed`, the records the
 /// attempt committed to each topic so far.
 ///
 /// Fails the Workflow Task when a record is one no writer stores. Each failure names what to
 /// fix, since the task fails the same way on every retry.
 pub(super) fn output_by_topic(
     records: Vec<OutputRecord>,
     committed: &mut HashMap<String, i64>,
+    max_batch_bytes: usize,
 ) -> Result<Vec<TopicOutput>, WFMachinesError> {
     for (index, record) in records.iter().enumerate() {
         check_record(record).map_err(|reason| {
@@ -87,10 +93,9 @@ pub(super) fn output_by_topic(
     for topic in topics {
         let mut bytes = 0;
         for record in topic.records {
-            let size = record.body.as_ref().map_or(0, prost::Message::encoded_len)
-                + record.content_hash.len();
+            let size = prost::Message::encoded_len(&record) + ENVELOPE_BYTES;
             let full = appends.last().is_some_and(|last: &TopicOutput| {
-                last.topic == topic.topic && bytes + size > MAX_APPEND_BYTES
+                last.topic == topic.topic && bytes + size > max_batch_bytes
             });
             if full || appends.last().is_none_or(|last| last.topic != topic.topic) {
                 appends.push(TopicOutput {
@@ -224,6 +229,7 @@ impl OutputStore {
                     })
                 })
                 .collect::<Result<Vec<_>, _>>()?;
+            temporalio_streams::check_record_limit(&records, &self.store.capabilities())?;
             let request = StoreAppendRequest {
                 chain: Some(self.chain(&batch.workflow_id, &batch.first_run_id)),
                 topic: topic.topic.clone(),
@@ -278,8 +284,14 @@ impl OutputStore {
         let chain = self.chain(&retired.workflow_id, &retired.first_run_id);
         let producer_id = temporalio_streams::workflow_producer_id(&retired.run_id);
         let store = self.store.clone();
+        // A store without producer state keeps no sessions to drop.
+        let sessions = if store.capabilities().producer_state {
+            retired.sessions
+        } else {
+            Default::default()
+        };
         self.tasks.spawn(async move {
-            for (attempt, topics) in &retired.sessions {
+            for (attempt, topics) in &sessions {
                 for topic in topics {
                     if let Err(error) = store
                         .retire_producer(&chain, topic, &producer_id, *attempt)
@@ -343,6 +355,7 @@ mod tests {
         let appends = output_by_topic(
             vec![big("a", 1), big("a", 2), big("a", 3), data("b", 4)],
             &mut committed,
+            MAX_APPEND_BYTES,
         )
         .unwrap();
         let shape: Vec<(&str, i64, usize)> = appends
@@ -353,20 +366,31 @@ mod tests {
         assert_eq!(committed["a"], 3);
     }
 
+    #[test]
+    fn a_commit_splits_by_the_stores_batch_limit() {
+        let mut committed = HashMap::new();
+        let records = (0..4).map(|n| data("a", n)).collect();
+        let appends = output_by_topic(records, &mut committed, 700).unwrap();
+        let sizes: Vec<usize> = appends.iter().map(|append| append.records.len()).collect();
+        assert_eq!(sizes, [2, 2]);
+        assert_eq!((appends[1].sequence, committed["a"]), (3, 4));
+    }
+
     #[test]
     fn topics_keep_their_order_of_first_publish_and_sequences_go_on() {
         let mut committed = HashMap::new();
         let first = output_by_topic(
             vec![data("b", 1), data("a", 2), finish("b")],
             &mut committed,
+            MAX_APPEND_BYTES,
         )
         .unwrap();
         let shape: Vec<_> = first
             .iter()
             .map(|t| (t.topic.as_str(), t.sequence, t.records.len()))
             .collect();
         assert_eq!(shape, [("b", 1, 2), ("a", 1, 1)]);
-        let second = output_by_topic(vec![data("a", 3)], &mut committed).unwrap();
+        let second = output_by_topic(vec![data("a", 3)], &mut committed, MAX_APPEND_BYTES).unwrap();
         assert_eq!((second[0].topic.as_str(), second[0].sequence), ("a", 2));
     }
 
@@ -390,9 +414,13 @@ mod tests {
             (finish_with_body, "record 1 is FINISH with a body"),
             (unspecified, "record 1 has kind 0"),
         ] {
-            let message = output_by_topic(vec![data("t", 9), record], &mut HashMap::new())
-                .unwrap_err()
-                .to_string();
+            let message = output_by_topic(
+                vec![data("t", 9), record],
+                &mut HashMap::new(),
+                MAX_APPEND_BYTES,
+            )
+            .unwrap_err()
+            .to_string();
             assert!(message.contains(expected), "{expected}: {message}");
         }
     }

crates/streams/docs/redis.md

@@ -47,6 +47,13 @@ so on a cluster the user must exist on every node. Deleting an owner's streams a
 `+scan`. `+config|get` only reads
 `maxmemory-policy`, and the store goes on when a server refuses it.
 
+**Plain mode.** `RedisStoreOptions::plain` runs the store with no Lua. An append is a
+`MULTI` of `XADD`s that trim by age, a read is `XREAD` alone, and no meta and no close flag is
+kept. So the store declares neither producer state nor a kept last position, and Core takes
+their fallbacks: readers drop copies and end at the first `CLOSE` record, and a read decides
+expiry from the record at its cursor. It exists so tests run those fallbacks on a real store.
+Listing a chain's topics scans every primary, as a delete does, and needs `+scan`.
+
 **Tests.** The Redis tests run when `STREAMS_REDIS_URL` names a standalone server and
 `STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both. The
 conformance suite runs on each, and once more as a user with exactly the ACL above. The

crates/streams/src/handle.rs

@@ -5,7 +5,8 @@
 
 use crate::{
     BEGINNING, END, ReadTarget, StreamError, StreamResult, StreamStore, WORKFLOW_OWNER_KIND,
-    activity_producer_id, append_batch, check_append_digest, close_topic, mint_cursor,
+    activity_producer_id, append_batch, check_append_digest, check_store_limits, close_topic,
+    mint_cursor,
     owner::{OwnerClient, OwnerDescription, OwnerError},
     proto::{
         ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse,
@@ -245,6 +246,7 @@ impl Streams {
                 .map(|stored| stored.encode_to_vec())
             })
             .collect::<StreamResult<Vec<_>>>()?;
+        check_store_limits(&records, &self.store.capabilities())?;
         let landed = append_batch(
             &*self.store,
             StoreAppendRequest {
@@ -447,7 +449,8 @@ impl Streams {
     ///
     /// 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.
+    /// state only, and readers find a newer attempt from the records. A store without producer
+    /// state keeps none, so the call does nothing there.
     pub async fn retire_producer(
         &self,
         request: RetireProducerRequest,
@@ -462,6 +465,9 @@ impl Streams {
                 return Err(StreamError::invalid_argument("a retire needs its producer"));
             }
         };
+        if !self.store.capabilities().producer_state {
+            return Ok(RetireProducerResponse {});
+        }
         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.
@@ -1504,6 +1510,45 @@ mod tests {
         assert_eq!(setup.owners.describes(), before);
     }
 
+    #[tokio::test]
+    async fn a_batch_above_the_stores_limits_is_refused_before_it_lands() {
+        let store = MemoryStore::with_capabilities(crate::Capabilities {
+            max_record_bytes: 200,
+            max_batch_bytes: 300,
+            ..MemoryStore::new().capabilities()
+        });
+        let setup = setup_on(store, StreamsOptions::default());
+        let error = setup
+            .streams
+            .append(append("p", 1, &"x".repeat(300)))
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::InvalidArgument);
+        assert!(error.message.contains("record limit of 200"), "{error}");
+        let mut batch = append("p", 1, "small");
+        batch.records = vec![data("a"), data("b"), data("c")];
+        let error = setup.streams.append(batch).await.unwrap_err();
+        assert!(error.message.contains("batch limit of 300"), "{error}");
+        setup.streams.append(append("p", 1, "fits")).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn without_producer_state_a_retire_does_nothing() {
+        let setup = setup_on(without_producer_state(), StreamsOptions::default());
+        setup.streams.append(append("p", 1, "1")).await.unwrap();
+        setup
+            .streams
+            .retire_producer(RetireProducerRequest {
+                stream: Some(stream("out")),
+                producer: Some(retire_producer_request::Producer::Named(NamedProducer {
+                    producer_id: "p".to_string(),
+                    attempt: 1,
+                })),
+            })
+            .await
+            .unwrap();
+    }
+
     #[tokio::test(start_paused = true)]
     async fn without_close_flags_a_producer_that_sees_its_owner_end_closes_its_topic() {
         let setup = setup_on(without_producer_state(), StreamsOptions::default());

crates/streams/src/lib.rs

@@ -33,8 +33,9 @@ pub use reader::{
     DEFAULT_MAX_RECORDS, MAX_STATE_BYTES, MAX_TRACKED_PRODUCERS, ReadTarget, read_page,
 };
 pub use record::{
-    WORKFLOW_OWNER_KIND, activity_producer_id, check_append_digest, close_record,
-    stored_append_record, topic_record, workflow_batch_digest, workflow_producer_id,
+    WORKFLOW_OWNER_KIND, activity_producer_id, check_append_digest, check_record_limit,
+    check_store_limits, close_record, stored_append_record, topic_record, workflow_batch_digest,
+    workflow_producer_id,
 };
 pub use service::StreamService;
 pub use store::{Capabilities, ConditionForms, MAX_BATCH_BYTES, MAX_RECORD_BYTES, StreamStore};

crates/streams/src/memory.rs

@@ -125,12 +125,17 @@ impl MemoryStore {
         Self::default()
     }
 
-    /// An empty store with only the capabilities both it and `mask` have. Without producer
-    /// state it keeps no dedupe entries and no close flags, without a kept last position its
-    /// reads never fail as expired or not found, and without a long poll its reads never wait.
+    /// An empty store with only the capabilities both it and `mask` have, and the limits of
+    /// `mask`. Without producer state it keeps no dedupe entries and no close flags, without a
+    /// kept last position its reads never fail as expired or not found, and without a long poll
+    /// its reads never wait.
     pub fn with_capabilities(mask: Capabilities) -> Self {
         Self {
-            capabilities: FULL.masked(&mask),
+            capabilities: Capabilities {
+                max_record_bytes: mask.max_record_bytes,
+                max_batch_bytes: mask.max_batch_bytes,
+                ..FULL.masked(&mask)
+            },
             ..Self::default()
         }
     }

crates/streams/src/record.rs

@@ -1,7 +1,7 @@
 //! Building the records a store keeps, and naming who wrote them.
 
 use crate::{
-    StreamError, StreamResult,
+    Capabilities, StreamError, StreamResult,
     proto::{ActivityProducer, AppendRecord, StreamRecord, StreamRecordKind},
 };
 use temporalio_common::{
@@ -145,6 +145,34 @@ fn stored_record(
     Ok(stored)
 }
 
+/// Refuses a batch of serialized records with a record above the store's record limit.
+pub fn check_record_limit(records: &[Vec<u8>], capabilities: &Capabilities) -> StreamResult<()> {
+    let limit = capabilities.max_record_bytes;
+    if let Some((index, record)) = records.iter().enumerate().find(|(_, r)| r.len() > limit) {
+        return Err(StreamError::invalid_argument(format!(
+            "record {index} takes {} bytes in the store, above the store's record limit of \
+             {limit} bytes",
+            record.len()
+        )));
+    }
+    Ok(())
+}
+
+/// Refuses a batch above the store's record or batch limit. Core doesn't split a producer's
+/// batch, since lang numbered its records as one sequence range.
+pub fn check_store_limits(records: &[Vec<u8>], capabilities: &Capabilities) -> StreamResult<()> {
+    check_record_limit(records, capabilities)?;
+    let bytes: usize = records.iter().map(Vec::len).sum();
+    let limit = capabilities.max_batch_bytes;
+    if bytes > limit {
+        return Err(StreamError::invalid_argument(format!(
+            "the batch takes {bytes} bytes in the store, above the store's batch limit of {limit} \
+             bytes"
+        )));
+    }
+    Ok(())
+}
+
 /// A `CLOSE` record that `producer_id` at `attempt` writes to end `topic` for every reader.
 pub fn close_record(topic: &str, producer_id: &str, attempt: i64) -> StreamRecord {
     StreamRecord {