From c2051c5ee7cc34c3c96bab92f0f3972d425aa994 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 12:18:33 -0700 Subject: [PATCH 1/2] Added Redis plain mode and the store's record and batch limits. Plain mode runs Redis with no Lua, so the producer state and kept last position fallbacks run on a real store. Core takes the record and batch limits from the store, splits Workflow output by them and refuses a record or a producer batch above them. --- .../src/core_tests/external_streams.rs | 26 +++ .../src/worker/workflow/managed_run.rs | 15 +- .../src/worker/workflow/stream_output.rs | 42 +++-- crates/streams/docs/redis.md | 7 + crates/streams/src/handle.rs | 26 ++- crates/streams/src/lib.rs | 5 +- crates/streams/src/memory.rs | 13 +- crates/streams/src/record.rs | 30 +++- crates/streams/src/redis/mod.rs | 155 ++++++++++++++++++ 9 files changed, 300 insertions(+), 19 deletions(-) diff --git a/crates/sdk-core/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs index 7ed5159010..642998563a 100644 --- a/crates/sdk-core/src/core_tests/external_streams.rs +++ b/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"]); diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs index 8c0898de94..caa8d46bb4 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/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")] { diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index add9d4d797..0145dba786 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/crates/sdk-core/src/worker/workflow/stream_output.rs @@ -51,12 +51,17 @@ pub(crate) struct RetiredSessions { pub(crate) chain_end: Option, } -/// 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 @@ -64,6 +69,7 @@ pub(crate) const MAX_APPEND_BYTES: usize = 4 << 20; pub(super) fn output_by_topic( records: Vec, committed: &mut HashMap, + max_batch_bytes: usize, ) -> Result, 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::, _>>()?; + 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(), @@ -343,6 +349,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,12 +360,23 @@ 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 = 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 @@ -366,7 +384,7 @@ mod tests { .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 +408,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}"); } } diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index 08193a4177..b00c428a9f 100644 --- a/crates/streams/docs/redis.md +++ b/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 diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 678bc91412..9a4a8bb2f1 100644 --- a/crates/streams/src/handle.rs +++ b/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::>>()?; + check_store_limits(&records, &self.store.capabilities())?; let landed = append_batch( &*self.store, StoreAppendRequest { @@ -1504,6 +1506,28 @@ 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(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()); diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 2d0f5166a7..7d85bd9851 100644 --- a/crates/streams/src/lib.rs +++ b/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}; diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index d6e262ae9e..fe9720875f 100644 --- a/crates/streams/src/memory.rs +++ b/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() } } diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index bcbec50b42..60b988f097 100644 --- a/crates/streams/src/record.rs +++ b/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], 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], 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 { diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index a5d7037833..b8125584e0 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -67,6 +67,10 @@ pub struct RedisStoreOptions { pub blocking_reads_per_node: usize, /// How long a short call waits for its reply. pub response_timeout: Duration, + /// Runs with no Lua: no dedupe entries, no close flags and no topic meta. Core then takes + /// the fallbacks of producer state and the kept last position, as on a store that has + /// only a log. For tests of those fallbacks on a real store. + pub plain: bool, } impl RedisStoreOptions { @@ -79,6 +83,7 @@ impl RedisStoreOptions { retention: Duration::from_secs(7 * 24 * 60 * 60), blocking_reads_per_node: 256, response_timeout: Duration::from_secs(5), + plain: false, } } } @@ -90,6 +95,7 @@ pub struct RedisStore { prefix: String, retention_ms: u64, grace_ms: u64, + plain: bool, } impl std::fmt::Debug for RedisStore { @@ -97,6 +103,7 @@ impl std::fmt::Debug for RedisStore { f.debug_struct("RedisStore") .field("prefix", &self.prefix) .field("retention_ms", &self.retention_ms) + .field("plain", &self.plain) .finish_non_exhaustive() } } @@ -147,9 +154,98 @@ impl RedisStore { prefix: options.key_prefix, retention_ms, grace_ms: TOMBSTONE_GRACE.as_millis() as u64, + plain: options.plain, }) } + /// Appends the batch with `XADD`s in one transaction, each trimming by age as it lands. + async fn append_plain( + &self, + request: &StoreAppendRequest, + ) -> StreamResult { + let log = self.keys(request.chain.as_ref())?.log(&request.topic); + // The client's clock, since no script can read the server's in the same step. A skew + // of seconds moves the trim by seconds, against a retention of days. + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |since| since.as_millis() as u64); + let floor = format!("{}-0", now_ms.saturating_sub(self.retention_ms)); + let mut pipe = redis::pipe(); + pipe.atomic(); + for record in &request.records { + pipe.cmd("XADD") + .arg(&log) + .arg("MINID") + .arg(&floor) + .arg("*") + .arg(RECORD_FIELD) + .arg(record.as_slice()); + } + pipe.cmd("PEXPIRE") + .arg(&log) + .arg(self.retention_ms) + .ignore(); + let ids: Vec = pipe + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Append)?; + let (Some(first), Some(last)) = (ids.first(), ids.last()) else { + return Err(StreamError::outcome_unknown( + "Redis answered the append with no stream ids", + )); + }; + Ok(StoreAppendResponse { + first_position: first.clone(), + last_position: last.clone(), + }) + } + + /// Reads with `XREAD` alone. With no topic meta, the read can't tell lost records apart, and + /// Core decides expiry from the record at the cursor. + async fn read_plain(&self, request: &StoreReadRequest) -> StreamResult { + let keys = self.keys(request.chain.as_ref())?; + let topic = request.topic.as_str(); + if !request.after_position.is_empty() { + entry(&request.after_position)?; + } + let after = match request.after_position.as_str() { + "" => "0-0", + position => position, + }; + let wait = wait_of(request); + let deadline = Instant::now() + wait; + let xread = Self::xread(&keys, topic, after, request.max_records, None); + let reply: Value = xread + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Read)?; + let mut found = records(reply)?; + if found.is_empty() && !wait.is_zero() { + let xread = Self::xread(&keys, topic, after, request.max_records, Some(wait)); + if let Some(reply) = self + .blocking + .query(&keys.log(topic), &xread, deadline) + .await + .mapped(Call::Read)? + { + found = records(reply)?; + } + } + Ok(StoreReadResponse { + records: found, + closed: false, + }) + } + + fn producer_state(&self) -> StreamResult<()> { + if self.plain { + return Err(StreamError::unsupported( + "the Redis store in plain mode keeps no producer state", + )); + } + Ok(()) + } + fn keys(&self, chain: Option<&ChainId>) -> StreamResult { let chain = chain .ok_or_else(|| StreamError::refused("a store call needs the chain it is about"))?; @@ -451,6 +547,13 @@ impl StreamStore for RedisStore { } fn capabilities(&self) -> Capabilities { + if self.plain { + return Capabilities { + producer_state: false, + kept_last: false, + ..CAPABILITIES + }; + } CAPABILITIES } @@ -461,6 +564,9 @@ impl StreamStore for RedisStore { if request.digest.is_empty() { return Err(StreamError::refused("an append needs the batch's digest")); } + if self.plain { + return self.append_plain(&request).await; + } let keys = self.keys(request.chain.as_ref())?; let mut script = scripts::APPEND_SCRIPT.prepare_invoke(); script @@ -486,6 +592,9 @@ impl StreamStore for RedisStore { } async fn read(&self, request: StoreReadRequest) -> StreamResult { + if self.plain { + return self.read_plain(&request).await; + } let keys = self.keys(request.chain.as_ref())?; let topic = request.topic.as_str(); let position = match request.after_position.as_str() { @@ -587,6 +696,9 @@ impl StreamStore for RedisStore { } async fn newest_ever(&self, chain: &ChainId, topic: &str) -> StreamResult { + if self.plain { + return Ok(String::new()); + } let last: Option = redis::cmd("HGET") .arg(ChainKeys::new(&self.prefix, chain).meta(topic)) .arg("last") @@ -602,6 +714,9 @@ impl StreamStore for RedisStore { topic: &str, mark: &str, ) -> StreamResult> { + if self.plain { + return Ok(None); + } let trimmed: Option = redis::cmd("HGET") .arg(ChainKeys::new(&self.prefix, chain).meta(topic)) .arg("trimmed") @@ -620,6 +735,7 @@ impl StreamStore for RedisStore { } async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { + self.producer_state()?; let keys = ChainKeys::new(&self.prefix, chain); redis::pipe() .atomic() @@ -631,6 +747,7 @@ impl StreamStore for RedisStore { } async fn close_topic(&self, chain: &ChainId, topic: &str) -> StreamResult<()> { + self.producer_state()?; let meta = ChainKeys::new(&self.prefix, chain).meta(topic); // The meta is the topic's tombstone, so the mark lives as long as the topic is known. redis::pipe() @@ -649,6 +766,7 @@ impl StreamStore for RedisStore { producer_id: &str, attempt: i64, ) -> StreamResult<()> { + self.producer_state()?; // One field of one key, so a plain `HDEL` is atomic, and it creates nothing when the meta // is gone. redis::cmd("HDEL") @@ -874,6 +992,43 @@ mod tests { assert_eq!(store.blocking.in_use(), 0); } + #[tokio::test] + async fn in_plain_mode_a_repeat_lands_again_and_no_close_flag_is_kept() { + let Ok(url) = std::env::var("STREAMS_REDIS_URL") else { + return; + }; + let mut options = RedisStoreOptions::new(url); + options.key_prefix = format!("plain-{}", std::process::id()); + options.plain = true; + let store = RedisStore::connect(options).await.unwrap(); + let capabilities = store.capabilities(); + assert!(!capabilities.producer_state && !capabilities.kept_last); + assert!(capabilities.long_poll && capabilities.topic_listing); + let run = format!("plain-{}", std::process::id()); + let first = store.append(append(&run, 1)).await.unwrap(); + let again = store.append(append(&run, 1)).await.unwrap(); + assert!(entry(&again.first_position).unwrap() > entry(&first.last_position).unwrap()); + let read = store + .read(waiting_read(&run, Duration::ZERO)) + .await + .unwrap(); + assert_eq!(read.records.len(), 2); + assert!(!read.closed); + assert_eq!(store.list_topics(&chain(&run)).await.unwrap(), ["out"]); + for error in [ + store.close_chain(&chain(&run)).await.unwrap_err(), + store.close_topic(&chain(&run), "out").await.unwrap_err(), + ] { + assert_eq!(error.kind, StreamFailureKind::Unsupported); + } + // A read after a position the log never held answers what follows, as a plain log does. + let missing = StoreReadRequest { + after_position: "1-0".to_string(), + ..waiting_read(&format!("{run}-missing"), Duration::ZERO) + }; + assert!(store.read(missing).await.unwrap().records.is_empty()); + } + #[tokio::test] async fn a_read_that_finds_the_pool_full_answers_empty_at_its_deadline() { let Some(store) = store(1).await else { From 4c265169988b3fe58eb71b7d4230f71e883abff4 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 12:21:11 -0700 Subject: [PATCH 2/2] Made a retire a no-op on stores without producer state. --- .../src/worker/workflow/stream_output.rs | 8 ++++++- crates/streams/src/handle.rs | 23 ++++++++++++++++++- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index 0145dba786..0ab774f560 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/crates/sdk-core/src/worker/workflow/stream_output.rs @@ -284,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) diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 9a4a8bb2f1..2f0ae53963 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -449,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, @@ -464,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. @@ -1528,6 +1532,23 @@ mod tests { 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());