[Brook] Added the CLOSE and TOPIC records and stopped readers from closing.

crates/protos/protos/local/temporal/sdk/streams/v1/message.proto

@@ -54,4 +54,13 @@ enum StreamRecordKind {
     // A reader synthesizes SUPERSEDED with this value and never stores it, so
     // no stored kind may take it.
     reserved 3;
+    // Kept for links between streams.
+    reserved 4;
+    // Ends `topic` for every reader, or every topic of the chain when `topic`
+    // is empty. Readers drop every record after the first one. On a store
+    // without close flags, producers and the owner's Core write it to close.
+    STREAM_RECORD_KIND_CLOSE = 5;
+    // Names `topic` in the chain's control log `_chain`, so a closer finds the
+    // chain's topics on a store that can't list them. Topic readers skip it.
+    STREAM_RECORD_KIND_TOPIC = 6;
 }

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

@@ -453,7 +453,13 @@ async fn an_append_that_may_not_have_landed_is_sent_again() {
     commit_in_two_tasks(worker, &["a"], &[]).await;
     assert_eq!(
         store.calls(),
-        ["append", "append", "append", "retire_producer"]
+        [
+            "append",
+            "append",
+            "append",
+            "retire_producer",
+            "close_chain"
+        ]
     );
     assert_eq!(store.read(TOPIC).await, ["encrypted a"]);
 }
@@ -650,7 +656,71 @@ async fn replay_makes_no_store_call() {
     let fired = worker.poll_workflow_activation().await.unwrap();
     worker.complete_execution(&fired.run_id).await;
     worker.drain_pollers_and_shutdown().await;
-    assert!(store.calls().is_empty(), "{:?}", store.calls());
+    // The live task that completes the run closes its chain, and nothing else reaches the store.
+    assert_eq!(store.calls(), ["close_chain"]);
+}
+
+/// The owner's Core closes the chain once the server accepts a task that ends it, and leaves a
+/// chain that can go on to the producers' owner checks.
+#[rstest::rstest]
+#[case::completed(CompleteWorkflowExecution::default().into(), false, false, true)]
+#[case::failed(fail_workflow(), false, false, true)]
+#[case::canceled(cancel_workflow(), false, false, true)]
+#[case::failed_with_retries(fail_workflow(), true, false, false)]
+#[case::completed_with_cron(CompleteWorkflowExecution::default().into(), false, true, false)]
+#[case::continued_as_new(continue_as_new(), false, false, false)]
+#[tokio::test]
+async fn the_owners_core_closes_a_chain_the_run_ended(
+    #[case] command: workflow_command::Variant,
+    #[case] retries: bool,
+    #[case] cron: bool,
+    #[case] closes: bool,
+) {
+    use temporalio_common::protos::temporal::api::common::v1::RetryPolicy;
+    let store = Arc::new(RecordingStore::default());
+    let mut t = TestHistoryBuilder::default();
+    t.add(WorkflowExecutionStartedEventAttributes {
+        retry_policy: retries.then(RetryPolicy::default),
+        cron_schedule: if cron {
+            "@hourly".to_string()
+        } else {
+            String::new()
+        },
+        ..default_wes_attribs()
+    });
+    t.add_workflow_task_scheduled_and_started();
+    let worker = worker_appending_to(
+        t,
+        vec![ResponseType::AllHistory],
+        store.clone(),
+        Default::default(),
+    );
+    let activation = worker.poll_workflow_activation().await.unwrap();
+    complete(&worker, &activation.run_id, vec![command]).await;
+    worker.drain_pollers_and_shutdown().await;
+    let expected: &[&str] = if closes { &["close_chain"] } else { &[] };
+    assert_eq!(store.calls(), expected);
+}
+
+fn fail_workflow() -> workflow_command::Variant {
+    workflow_command::Variant::FailWorkflowExecution(
+        temporalio_common::protos::coresdk::workflow_commands::FailWorkflowExecution {
+            failure: Some(Failure::default()),
+        },
+    )
+}
+
+fn cancel_workflow() -> workflow_command::Variant {
+    workflow_command::Variant::CancelWorkflowExecution(Default::default())
+}
+
+fn continue_as_new() -> workflow_command::Variant {
+    workflow_command::Variant::ContinueAsNewWorkflowExecution(
+        temporalio_common::protos::coresdk::workflow_commands::ContinueAsNewWorkflowExecution {
+            workflow_type: "next".to_string(),
+            ..Default::default()
+        },
+    )
 }
 
 #[tokio::test]

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

@@ -53,6 +53,7 @@ impl DrivenWorkflow {
             memo: attribs.memo.clone(),
             search_attrs: attribs.search_attributes.clone(),
             retry_policy: attribs.retry_policy.clone(),
+            cron: !attribs.cron_schedule.is_empty(),
         };
         self.send_job(
             start_workflow_from_attribs(attribs, workflow_id, randomness_seed, start_time).into(),

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

@@ -998,23 +998,43 @@ impl ManagedRun {
     }
 
     /// The store sessions of attempts below the current one. Once the server accepts the
-    /// current task, no earlier attempt of the run writes again.
-    fn sessions_to_retire(&mut self) -> Option<Box<RetiredSessions>> {
+    /// current task, no earlier attempt of the run writes again. With `ends_chain`, the
+    /// current attempt also closes the chain.
+    fn sessions_to_retire(&mut self, ends_chain: bool) -> Option<Box<RetiredSessions>> {
         let current = self.wfm.machines.stream_producer_attempt();
         let kept = self.stream_sessions.split_off(&current);
         let sessions = mem::replace(&mut self.stream_sessions, kept);
-        if sessions.is_empty() {
+        let machines = &self.wfm.machines;
+        if sessions.is_empty() && (!ends_chain || machines.first_execution_run_id.is_empty()) {
             return None;
         }
-        let machines = &self.wfm.machines;
         Some(Box::new(RetiredSessions {
             workflow_id: machines.workflow_id.clone(),
             run_id: machines.run_id.clone(),
             first_run_id: machines.first_execution_run_id.clone(),
             sessions,
+            chain_end: ends_chain.then_some(current),
         }))
     }
 
+    /// Whether a task with `commands` ends the run's chain, so the owner's Core closes its
+    /// streams. A failure with a retry policy and a run with a cron schedule can go on in the
+    /// same chain, so they leave the close to the producers' owner checks.
+    fn ends_chain(&self, commands: &[super::ProtoCommand]) -> bool {
+        if !self.has_stream_store() {
+            return false;
+        }
+        let started = self.wfm.machines.get_started_info();
+        let cron = started.is_some_and(|info| info.cron);
+        let retries = started.is_some_and(|info| info.retry_policy.is_some());
+        commands.iter().any(|command| match command.attributes {
+            Some(super::Attributes::CompleteWorkflowExecutionCommandAttributes(_)) => !cron,
+            Some(super::Attributes::FailWorkflowExecutionCommandAttributes(_)) => !cron && !retries,
+            Some(super::Attributes::CancelWorkflowExecutionCommandAttributes(_)) => true,
+            _ => false,
+        })
+    }
+
     fn _local_resolution(
         &mut self,
         res: LocalResolution,
@@ -1407,9 +1427,13 @@ impl ManagedRun {
         };
         let replaying = machines_wft_response.replaying;
         let mut retired = None;
-        if matches!(outcome, ActivationCompleteOutcome::ReportWFTSuccess(_)) {
+        if let ActivationCompleteOutcome::ReportWFTSuccess(report) = &outcome {
             self.wfm.wft_completion_reported();
-            retired = self.sessions_to_retire();
+            let ends_chain = match &report.action {
+                ActivationAction::WftComplete { commands, .. } => self.ends_chain(commands),
+                _ => false,
+            };
+            retired = self.sessions_to_retire(ends_chain);
         }
         FulfillableActivationComplete {
             result: ActivationCompleteResult {

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

@@ -1828,6 +1828,8 @@ struct WorkflowStartedInfo {
     memo: Option<Memo>,
     search_attrs: Option<SearchAttributes>,
     retry_policy: Option<RetryPolicy>,
+    /// A cron schedule starts the chain's next run when this one completes or fails.
+    cron: bool,
 }
 
 /// Wraps outgoing activation job protos with some internal details core might care about

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

@@ -46,6 +46,9 @@ pub(crate) struct RetiredSessions {
     pub(crate) run_id: String,
     pub(crate) first_run_id: String,
     pub(crate) sessions: BTreeMap<i64, BTreeSet<String>>,
+    /// The run's attempt, when the task ends the run's chain. The chain closes once the server
+    /// accepts the task, since a command the server doesn't take can still keep the run open.
+    pub(crate) chain_end: Option<i64>,
 }
 
 /// The most record bytes one append carries. A larger commit goes as several appends per topic, so
@@ -239,7 +242,7 @@ impl OutputStore {
             let mut delay = APPEND_RETRY_FIRST;
             let mut tries = 1;
             loop {
-                match self.store.append(request.clone()).await {
+                match temporalio_streams::append_batch(&*self.store, request.clone()).await {
                     Ok(_) => break,
                     Err(error) if error.kind == StreamFailureKind::Closed => {
                         if !dropped.contains(&topic.topic) {
@@ -268,7 +271,9 @@ impl OutputStore {
 
     /// Drops the store's dedupe sessions of attempts that will not write again, in the
     /// background, so the next activation doesn't wait a store round trip per session. A
-    /// failure only leaves a session in the store until retention, so it is logged.
+    /// failure only leaves a session in the store until retention, so it is logged. Then closes
+    /// the chain when the run ended it. A failed close leaves each reader to end on the owner's
+    /// status and each producer at its owner check.
     pub(crate) fn retire(&self, retired: RetiredSessions) {
         let chain = self.chain(&retired.workflow_id, &retired.first_run_id);
         let producer_id = temporalio_streams::workflow_producer_id(&retired.run_id);
@@ -284,6 +289,12 @@ impl OutputStore {
                     }
                 }
             }
+            if let Some(attempt) = retired.chain_end
+                && let Err(error) =
+                    temporalio_streams::close_chain(&*store, &chain, &producer_id, attempt).await
+            {
+                warn!(%error, "Could not close the streams of a run that ended its chain");
+            }
         });
     }