[Brook] Retired a producer attempt's dedupe session on request.

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 {

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());
     });
 }
 

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

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

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,
@@ -181,26 +182,14 @@ impl Streams {
     pub async fn append(&self, request: AppendRequest) -> StreamResult<AppendResponse> {
         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 {}",
@@ -214,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
@@ -389,6 +374,40 @@ 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<RetireProducerResponse> {
+        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 = 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) {
+            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.
@@ -593,6 +612,59 @@ 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"));
+            }
+            // A Workflow run writes its own output as `@<run id>`. 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)),
+        (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))
+}
+
+/// The key a producer attempt's chain is remembered by. A Workflow writes through a new local
+/// producer per flush, `__temporal_<kind>_<n>@<run id>`, 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)
 }
@@ -674,7 +746,7 @@ mod tests {
         MemoryStore,
         proto::{
             ActivityProducer, AppendRecord, NamedProducer, StoreReadRequest, StreamFailureKind,
-            StreamRecordKind,
+            StreamRecordKind, append_request::Producer,
         },
     };
     use sha2::{Digest, Sha256};
@@ -1113,6 +1185,41 @@ 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());
+        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());
@@ -1266,6 +1373,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<retire_producer_request::Producer>| 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),

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<DeleteOwnerResponse> {
         if request.owner_kind != StreamOwnerKind::Workflow as i32 {
             return Err(StreamError::unsupported(format!(

crates/streams/src/reader.rs

@@ -460,6 +460,10 @@ mod tests {
             unreachable!()
         }
 
+        async fn retire_producer(&self, _: &ChainId, _: &str, _: &str, _: i64) -> StreamResult<()> {
+            unreachable!()
+        }
+
         async fn delete_owner(&self, _: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
             unreachable!()
         }