From 05cf9161bb5d1ea197666751f70223df217190fc Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 19:32:09 -0700 Subject: [PATCH 1/3] Retired a producer attempt's dedupe session on request. --- .../temporal/sdk/core/streams/streams.proto | 15 ++ crates/sdk-core-c-bridge/src/tests/mod.rs | 23 ++- crates/sdk-core/CHANGELOG.md | 5 + crates/streams/docs/redis.md | 8 ++ crates/streams/src/handle.rs | 133 +++++++++++++++--- crates/streams/src/memory.rs | 17 +++ crates/streams/src/reader.rs | 4 + crates/streams/src/redis/mod.rs | 17 +++ crates/streams/src/service.rs | 45 +++++- crates/streams/src/store.rs | 13 ++ crates/streams/tests/conformance.rs | 41 ++++++ crates/streams/tests/redis_store.rs | 40 ++++++ 12 files changed, 338 insertions(+), 23 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 a6fd6e8e6e..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,6 +24,9 @@ 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 @@ -210,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/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs index c5ffb56509..ba49907a18 100644 --- a/crates/sdk-core-c-bridge/src/tests/mod.rs +++ b/crates/sdk-core-c-bridge/src/tests/mod.rs @@ -517,7 +517,8 @@ 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, - StreamAddress, StreamOwnerKind, StreamRecordKind, append_request, read_record, + RetireProducerRequest, StreamAddress, StreamOwnerKind, StreamRecordKind, append_request, + read_record, retire_producer_request, }; Context::with(|context| { context.runtime_new().unwrap(); @@ -569,7 +570,7 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() { &store, "Read", ReadRequest { - stream: Some(address), + stream: Some(address.clone()), ..Default::default() } .encode_to_vec(), @@ -585,6 +586,24 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() { 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()); }); } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 9aedcb07cd..d7faf8a18e 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -40,6 +40,11 @@ relevant information. 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 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 0a4c75d4db..7d9bdda936 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, @@ -164,26 +165,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 {}", @@ -372,6 +361,44 @@ 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 = ProducerKey { + owner: 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. @@ -520,6 +547,29 @@ 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")); + } + (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)) +} + fn check_topic(topic: &str) -> StreamResult<()> { temporalio_common::streams::check_topic(topic).map_err(StreamError::invalid_argument) } @@ -593,7 +643,7 @@ mod tests { MemoryStore, proto::{ ActivityProducer, AppendRecord, NamedProducer, StoreReadRequest, StreamFailureKind, - StreamRecordKind, + StreamRecordKind, append_request::Producer, }, }; use sha2::{Digest, Sha256}; @@ -1119,6 +1169,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/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 b9823b1674..c5ff5b6d0e 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -414,6 +414,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 4bea9b3b5b..bf2e709091 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -568,6 +568,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 index f4689e3189..e3a695ac49 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -3,8 +3,8 @@ use crate::{ OwnerClient, StreamError, StreamResult, StreamStore, Streams, StreamsOptions, proto::{ - AppendRequest, CloseRequest, DeleteOwnerRequest, LatestRequest, ReadRequest, StreamFailure, - StreamStoreConfig, stream_store_config, + AppendRequest, CloseRequest, DeleteOwnerRequest, LatestRequest, ReadRequest, + RetireProducerRequest, StreamFailure, StreamStoreConfig, stream_store_config, }, }; use prost::Message; @@ -98,6 +98,11 @@ impl StreamService { .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:?}" ))), @@ -380,6 +385,42 @@ mod tests { 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; 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 a8011bff60..1370659856 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 D1b's script gave the Python provider, run again through the Core store. From 0284f07b3788f81b328e089533ce14900b672a0c Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 03:29:42 -0700 Subject: [PATCH 2/3] Refused a named producer whose id starts with `@`. That id names a Workflow run's own output. A client writing as it shared the run's dedupe session. --- crates/streams/src/handle.rs | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index c7b897a16a..11dcdcc6a5 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -630,6 +630,15 @@ fn producer( 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)), @@ -1163,6 +1172,18 @@ mod tests { assert!(chain_closed(&setup.store, "run-1").await); } + #[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()); From af44165f660e1dc4cb49ec6d92befc4f4c08aaaf Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 03:30:01 -0700 Subject: [PATCH 3/3] Remembered a Workflow's local producers as one entry per run. Each flush wrote as a new producer, so a busy run filled the memory and pushed out the long-lived producers' chain checks. --- crates/streams/src/handle.rs | 56 +++++++++++++++++++++++++++++------- 1 file changed, 46 insertions(+), 10 deletions(-) diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 11dcdcc6a5..d316fecb6f 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -203,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 @@ -399,11 +395,7 @@ impl Streams { return Err(StreamError::invalid_argument("a retire needs its producer")); } }; - let key = ProducerKey { - owner: owner_key(stream), - producer_id, - attempt, - }; + 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) { @@ -652,6 +644,27 @@ fn producer( 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) } @@ -1172,6 +1185,29 @@ 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());