From b78e1615d782b68e8b5f78314b25948738ddb670 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 12:09:46 -0700 Subject: [PATCH] Added the CLOSE and TOPIC records and stopped readers from closing. On a store without close flags, producers and the owner's Core write a CLOSE record, and every reader ends at the first one. The owner's Core closes the chain once the server accepts the task that ends it. A reader that sees its owner end only ends its own read. --- .../temporal/sdk/streams/v1/message.proto | 9 + .../src/core_tests/external_streams.rs | 74 +++- .../src/worker/workflow/driven_workflow.rs | 1 + .../src/worker/workflow/managed_run.rs | 36 +- crates/sdk-core/src/worker/workflow/mod.rs | 2 + .../src/worker/workflow/stream_output.rs | 15 +- crates/streams/src/close.rs | 243 +++++++++++++ crates/streams/src/handle.rs | 179 +++++++--- crates/streams/src/lib.rs | 6 +- crates/streams/src/reader.rs | 111 +++++- crates/streams/src/record.rs | 23 ++ crates/streams/testdata/reader_rules.json | 323 ++++++++++++++++-- 12 files changed, 923 insertions(+), 99 deletions(-) create mode 100644 crates/streams/src/close.rs diff --git a/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto b/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto index 3bf4af50ff..43be0c0410 100644 --- a/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto +++ b/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; } diff --git a/crates/sdk-core/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs index 863ae29517..7ed5159010 100644 --- a/crates/sdk-core/src/core_tests/external_streams.rs +++ b/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] diff --git a/crates/sdk-core/src/worker/workflow/driven_workflow.rs b/crates/sdk-core/src/worker/workflow/driven_workflow.rs index c9a2b3ed0b..f9871cdbbf 100644 --- a/crates/sdk-core/src/worker/workflow/driven_workflow.rs +++ b/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(), diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs index 40b04a8580..8c0898de94 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/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> { + /// 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> { let current = self.wfm.machines.stream_producer_attempt(); let kept = self.stream_sessions.split_off(¤t); 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 { diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index 80f3c166ed..f68227d88a 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -1828,6 +1828,8 @@ struct WorkflowStartedInfo { memo: Option, search_attrs: Option, retry_policy: Option, + /// 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 diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs index d7332e22c0..add9d4d797 100644 --- a/crates/sdk-core/src/worker/workflow/stream_output.rs +++ b/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>, + /// 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, } /// 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"); + } }); } diff --git a/crates/streams/src/close.rs b/crates/streams/src/close.rs new file mode 100644 index 0000000000..535a8dab08 --- /dev/null +++ b/crates/streams/src/close.rs @@ -0,0 +1,243 @@ +//! Closing streams and naming topics on every store. +//! +//! A store with producer state keeps close flags that its append checks. On other stores a close +//! is a `CLOSE` record, and readers drop every record after the first one. Readers never call +//! this. Producers and the owner's Core do. + +use crate::{ + StreamResult, StreamStore, close_record, + proto::{ + ChainId, StoreAppendRequest, StoreAppendResponse, StoreReadRequest, StreamRecord, + StreamRecordKind, + }, + topic_record, +}; +use prost::Message; +use sha2::{Digest, Sha256}; + +/// The chain's control log, where producers name their topics on a store that can't list them. +pub const CHAIN_LOG: &str = "_chain"; + +const CHAIN_LOG_PAGE: u32 = 1_000; + +/// Closes one topic of `chain` for every reader, as `producer_id` at `attempt`. With close flags +/// the store refuses later appends. Without them, the `CLOSE` record ends every read there, and +/// a later append lands where no reader delivers it. +pub async fn close_topic( + store: &dyn StreamStore, + chain: &ChainId, + topic: &str, + producer_id: &str, + attempt: i64, +) -> StreamResult<()> { + if store.capabilities().producer_state { + return store.close_topic(chain, topic).await; + } + write_one( + store, + chain, + topic, + close_record(topic, producer_id, attempt), + ) + .await +} + +/// Closes every topic of `chain`, as [close_topic] closes one. Without close flags, it finds the +/// topics by listing them, or from the `TOPIC` records in [CHAIN_LOG]. +pub async fn close_chain( + store: &dyn StreamStore, + chain: &ChainId, + producer_id: &str, + attempt: i64, +) -> StreamResult<()> { + if store.capabilities().producer_state { + return store.close_chain(chain).await; + } + for topic in chain_topics(store, chain).await? { + close_topic(store, chain, &topic, producer_id, attempt).await?; + } + Ok(()) +} + +/// Appends one batch of a producer attempt. On a store that can neither keep close flags nor list +/// topics, the attempt's first batch on a topic names the topic in [CHAIN_LOG] first. A crash +/// between the two leaves a name with no record, and a close of an empty topic does no harm. +pub async fn append_batch( + store: &dyn StreamStore, + request: StoreAppendRequest, +) -> StreamResult { + let capabilities = store.capabilities(); + if !capabilities.producer_state && !capabilities.topic_listing && request.sequence == 1 { + let chain = request.chain.clone().unwrap_or_default(); + let named = topic_record(&request.topic, &request.producer_id, request.attempt); + write_one(store, &chain, CHAIN_LOG, named).await?; + } + store.append(request).await +} + +/// The topics of `chain` that took a record. +async fn chain_topics(store: &dyn StreamStore, chain: &ChainId) -> StreamResult> { + if store.capabilities().topic_listing { + return store.list_topics(chain).await; + } + let mut topics: Vec = Vec::new(); + let mut after = String::new(); + loop { + let page = store + .read(StoreReadRequest { + chain: Some(chain.clone()), + topic: CHAIN_LOG.to_string(), + after_position: after.clone(), + wait: None, + max_records: CHAIN_LOG_PAGE, + }) + .await?; + let Some(last) = page.records.last() else { + break; + }; + after = last.position.clone(); + for stored in &page.records { + let Ok(record) = StreamRecord::decode(&*stored.record) else { + continue; + }; + if record.kind == StreamRecordKind::Topic as i32 && !topics.contains(&record.topic) { + topics.push(record.topic); + } + } + } + Ok(topics) +} + +/// Appends one control record. Its sequence is zero, so no reader takes it for a copy. +async fn write_one( + store: &dyn StreamStore, + chain: &ChainId, + topic: &str, + record: StreamRecord, +) -> StreamResult<()> { + let bytes = record.encode_to_vec(); + store + .append(StoreAppendRequest { + chain: Some(chain.clone()), + topic: topic.to_string(), + producer_id: record.producer_id, + attempt: record.attempt, + sequence: 0, + digest: Sha256::digest(&bytes).to_vec(), + records: vec![bytes], + }) + .await + .map(drop) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{Capabilities, MemoryStore, StreamStore}; + + fn chain() -> ChainId { + ChainId { + namespace: "ns".to_string(), + workflow_id: "wf".to_string(), + first_run_id: "run-1".to_string(), + } + } + + fn batch(topic: &str, sequence: i64) -> StoreAppendRequest { + StoreAppendRequest { + chain: Some(chain()), + topic: topic.to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: vec![sequence as u8], + records: vec![ + StreamRecord { + topic: topic.to_string(), + kind: StreamRecordKind::Data as i32, + producer_id: "p".to_string(), + attempt: 1, + sequence, + ..Default::default() + } + .encode_to_vec(), + ], + } + } + + /// Each record of a topic as `(kind, producer)`. + async fn kinds(store: &dyn StreamStore, topic: &str) -> Vec<(i32, String)> { + let page = store + .read(StoreReadRequest { + chain: Some(chain()), + topic: topic.to_string(), + max_records: 100, + ..Default::default() + }) + .await + .unwrap(); + page.records + .iter() + .map(|stored| { + let record = StreamRecord::decode(&*stored.record).unwrap(); + (record.kind, record.producer_id) + }) + .collect() + } + + const DATA: i32 = StreamRecordKind::Data as i32; + const CLOSE: i32 = StreamRecordKind::Close as i32; + const TOPIC: i32 = StreamRecordKind::Topic as i32; + + fn without(mask: impl FnOnce(&mut Capabilities)) -> MemoryStore { + let mut capabilities = MemoryStore::new().capabilities(); + mask(&mut capabilities); + MemoryStore::with_capabilities(capabilities) + } + + #[tokio::test] + async fn with_close_flags_a_close_sets_them_and_writes_no_record() { + let store = MemoryStore::new(); + append_batch(&store, batch("a", 1)).await.unwrap(); + close_chain(&store, &chain(), "@run-1", 7).await.unwrap(); + assert_eq!(kinds(&store, "a").await, [(DATA, "p".to_string())]); + let refused = store.append(batch("a", 2)).await.unwrap_err(); + assert_eq!(refused.kind, crate::proto::StreamFailureKind::Closed); + } + + #[tokio::test] + async fn without_close_flags_a_chain_close_writes_a_close_on_each_listed_topic() { + let store = without(|c| c.producer_state = false); + append_batch(&store, batch("a", 1)).await.unwrap(); + append_batch(&store, batch("b", 1)).await.unwrap(); + close_chain(&store, &chain(), "@run-1", 7).await.unwrap(); + for topic in ["a", "b"] { + assert_eq!( + kinds(&store, topic).await, + [(DATA, "p".to_string()), (CLOSE, "@run-1".to_string())] + ); + } + assert!(kinds(&store, CHAIN_LOG).await.is_empty()); + } + + #[tokio::test] + async fn without_listing_a_session_names_its_topic_once_and_a_close_finds_it() { + let store = without(|c| { + c.producer_state = false; + c.topic_listing = false; + }); + append_batch(&store, batch("a", 1)).await.unwrap(); + append_batch(&store, batch("a", 2)).await.unwrap(); + append_batch(&store, batch("b", 1)).await.unwrap(); + assert_eq!( + kinds(&store, CHAIN_LOG).await, + [(TOPIC, "p".to_string()), (TOPIC, "p".to_string())] + ); + close_chain(&store, &chain(), "@run-1", 7).await.unwrap(); + assert_eq!(kinds(&store, "a").await.last().unwrap().0, CLOSE); + assert_eq!(kinds(&store, "b").await.last().unwrap().0, CLOSE); + // A close of a topic no record reached does no harm. + close_topic(&store, &chain(), "quiet", "", 0).await.unwrap(); + assert_eq!(kinds(&store, "quiet").await, [(CLOSE, String::new())]); + } +} diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 557aea3634..678bc91412 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -5,7 +5,7 @@ use crate::{ BEGINNING, END, ReadTarget, StreamError, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, - activity_producer_id, check_append_digest, mint_cursor, + activity_producer_id, append_batch, check_append_digest, close_topic, mint_cursor, owner::{OwnerClient, OwnerDescription, OwnerError}, proto::{ ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, @@ -138,6 +138,15 @@ impl Remembered { } } +/// The producer attempt a call writes as, and the topic it writes, so a close at its owner check +/// reaches the topic on a store without close flags. +#[derive(Debug, Clone, Copy)] +struct Writer<'a> { + topic: &'a str, + producer_id: &'a str, + attempt: i64, +} + /// The stream layer over one store. pub struct Streams { store: Arc, @@ -183,9 +192,10 @@ impl Streams { /// Appends one batch for one producer attempt. /// /// The first batch for an owner, and the first after [StreamsOptions::owner_recheck], asks - /// whether the owner's chain ended. All producers of the owner share that check. If it did, or Temporal no longer holds the owner, - /// or a new chain took the Workflow id, the chain is marked closed and the batch is refused - /// as closed. + /// whether the owner's chain ended. All producers of the owner share that check. If it did, + /// or Temporal no longer holds the owner, or a new chain took the Workflow id, the producer + /// closes the chain, or its own topic on a store without close flags, and the batch is + /// refused as closed. pub async fn append(&self, request: AppendRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; let (producer_id, attempt) = match &request.producer { @@ -209,8 +219,16 @@ impl Streams { )); } check_append_digest(&request.digest)?; + let writer = Writer { + topic: &stream.topic, + producer_id: &producer_id, + attempt, + }; let chain = self - .producer_chain(producer_key(owner_key(stream), &producer_id, attempt)) + .producer_chain( + producer_key(owner_key(stream), &producer_id, attempt), + writer, + ) .await?; let records = request .records @@ -227,9 +245,9 @@ impl Streams { .map(|stored| stored.encode_to_vec()) }) .collect::>>()?; - let landed = self - .store - .append(StoreAppendRequest { + let landed = append_batch( + &*self.store, + StoreAppendRequest { chain: Some(chain), topic: stream.topic.clone(), producer_id, @@ -239,8 +257,9 @@ impl Streams { // store keeps it as given. digest: request.digest.clone(), records, - }) - .await?; + }, + ) + .await?; Ok(AppendResponse { first_cursor: self.cursor(stream, &landed.first_position), last_cursor: self.cursor(stream, &landed.last_position), @@ -377,25 +396,19 @@ impl Streams { } /// Whether the owner a read follows ended. Following the chain, a run that continued as - /// new is not the end, and a new chain on the Workflow id is. + /// new is not the end, and a new chain on the Workflow id is. A reader only ends its own read, + /// so it needs no write access. Producers and the owner's Core close the chain. async fn owner_ended(&self, stream: &StreamAddress, chain: &ChainId) -> StreamResult { let key = owner_key(stream); match self.describe(&key, &key.run_id).await { // The chain came from this Workflow, so History has since dropped it, or a failover - // hides it for a while. A reader only ends its own read, and producers close the - // chain. + // hides it for a while. Err(OwnerError::NotFound(_)) => Ok(self.gone_for_good(chain)), Err(error) => Err(describe_failed(&error)), Ok(described) if !key.run_id.is_empty() => { Ok(described.status != WorkflowExecutionStatus::Running) } - Ok(latest) if !latest.chain_ended() && latest.first_run_id == chain.first_run_id => { - Ok(false) - } - Ok(_) => { - self.store.close_chain(chain).await?; - Ok(true) - } + Ok(latest) => Ok(latest.chain_ended() || latest.first_run_id != chain.first_run_id), } } @@ -419,11 +432,12 @@ impl Streams { }) } - /// Closes one stream, so producers' new batches are refused and reads end. + /// Closes one stream, so reads end. With close flags, producers' new batches are refused. + /// Without them, a batch that lands after the close is one no reader delivers. pub async fn close(&self, request: CloseRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; let chain = self.chain(stream).await?; - self.store.close_topic(&chain, &stream.topic).await?; + close_topic(&*self.store, &chain, &stream.topic, "", 0).await?; Ok(CloseResponse {}) } @@ -484,13 +498,13 @@ impl Streams { /// The chain a producer writes on. A producer attempt stays on the chain it began on, and /// is refused once a new chain took the Workflow id. - async fn producer_chain(&self, key: ProducerKey) -> StreamResult { - let chain = self.owner_chain(&key.owner).await?; + async fn producer_chain(&self, key: ProducerKey, writer: Writer<'_>) -> StreamResult { + let chain = self.owner_chain(&key.owner, writer).await?; if let Some(began) = self.producers.get(&key) && began.first_run_id != chain.first_run_id { let began = chain_of_key(&key.owner, &began.first_run_id); - return Err(self.closed(&began, "has closed").await); + return Err(self.closed(&began, "has closed", writer).await); } self.producers .put(key, &chain.first_run_id, self.options.owner_recheck); @@ -499,7 +513,7 @@ impl Streams { /// The chain an owner writes on now, asking about the owner when no producer of it asked /// within the recheck interval. - async fn owner_chain(&self, owner: &OwnerKey) -> StreamResult { + async fn owner_chain(&self, owner: &OwnerKey, writer: Writer<'_>) -> StreamResult { let recheck = self.options.owner_recheck; let known = self.owners.get(owner); if let Some(checked) = &known @@ -513,18 +527,18 @@ impl Streams { if let (Err(OwnerError::NotFound(_)), Some(checked)) = (&latest, &known) { // The chain came from this Workflow, so History has since dropped it. let gone = chain_of_key(owner, &checked.first_run_id); - return Err(self.gone(&gone).await); + return Err(self.gone(&gone, writer).await); } let latest = latest.map_err(not_found)?; let chain = chain_of_key(owner, &first_run_id(owner, &latest)?); if latest.chain_ended() { - return Err(self.closed(&chain, "has closed").await); + return Err(self.closed(&chain, "has closed", writer).await); } chain } else if let Some(checked) = known { // A pinned run's chain never changes, so only its end is left to check. let chain = chain_of_key(owner, &checked.first_run_id); - self.refuse_if_ended(&chain).await?; + self.refuse_if_ended(&chain, writer).await?; chain } else { let pinned = self @@ -534,7 +548,7 @@ impl Streams { let chain = chain_of_key(owner, &first_run_id(owner, &pinned)?); // A running run is its chain's latest, so it answers the end check too. if pinned.status != WorkflowExecutionStatus::Running { - self.refuse_if_ended(&chain).await?; + self.refuse_if_ended(&chain, writer).await?; } chain }; @@ -542,9 +556,9 @@ impl Streams { Ok(chain) } - /// Marks the chain closed and refuses when its latest run ended, when Temporal no longer - /// holds it, or when a new chain took its Workflow id. - async fn refuse_if_ended(&self, chain: &ChainId) -> StreamResult<()> { + /// Closes the chain and refuses when its latest run ended, when Temporal no longer holds it, + /// or when a new chain took its Workflow id. + async fn refuse_if_ended(&self, chain: &ChainId, writer: Writer<'_>) -> StreamResult<()> { let key = OwnerKey { namespace: chain.namespace.clone(), workflow_id: chain.workflow_id.clone(), @@ -552,10 +566,10 @@ impl Streams { }; match self.describe(&key, "").await { // The chain came from this Workflow, so History has since dropped it. - Err(OwnerError::NotFound(_)) => Err(self.gone(chain).await), + Err(OwnerError::NotFound(_)) => Err(self.gone(chain, writer).await), Err(error) => Err(describe_failed(&error)), Ok(latest) if latest.chain_ended() || latest.first_run_id != chain.first_run_id => { - Err(self.closed(chain, "has closed").await) + Err(self.closed(chain, "has closed", writer).await) } Ok(_) => Ok(()), } @@ -564,9 +578,9 @@ impl Streams { /// Closes a chain whose owner Temporal answered is gone, but only on the second such answer /// at least [StreamsOptions::owner_recheck] after the first. The first answer fails the call /// as a storage failure, which the caller retries. - async fn gone(&self, chain: &ChainId) -> StreamError { + async fn gone(&self, chain: &ChainId, writer: Writer<'_>) -> StreamError { if self.gone_for_good(chain) { - self.closed(chain, "is gone from History").await + self.closed(chain, "is gone from History", writer).await } else { missing(chain) } @@ -590,8 +604,16 @@ impl Streams { } } - async fn closed(&self, chain: &ChainId, how: &str) -> StreamError { - if let Err(error) = self.store.close_chain(chain).await { + /// Closes what a producer that learned its owner ended can reach: the chain with close + /// flags, else its own topic. + async fn closed(&self, chain: &ChainId, how: &str, writer: Writer<'_>) -> StreamError { + let closed = if self.store.capabilities().producer_state { + self.store.close_chain(chain).await + } else { + let (topic, producer_id) = (writer.topic, writer.producer_id); + close_topic(&*self.store, chain, topic, producer_id, writer.attempt).await + }; + if let Err(error) = closed { return error; } StreamError::closed(format!( @@ -804,7 +826,7 @@ mod tests { MemoryStore, proto::{ ActivityProducer, AppendRecord, NamedProducer, StoreReadRequest, StreamFailureKind, - StreamRecordKind, append_request::Producer, + StreamRecord, StreamRecordKind, append_request::Producer, }, }; use sha2::{Digest, Sha256}; @@ -890,7 +912,19 @@ mod tests { } fn setup(options: StreamsOptions) -> Setup { - let store = Arc::new(MemoryStore::new()); + setup_on(MemoryStore::new(), options) + } + + /// A store with no close flags and no dedupe entries. + fn without_producer_state() -> MemoryStore { + MemoryStore::with_capabilities(crate::Capabilities { + producer_state: false, + ..MemoryStore::new().capabilities() + }) + } + + fn setup_on(store: MemoryStore, options: StreamsOptions) -> Setup { + let store = Arc::new(store); let owners = Arc::new(Owners::default()); owners.run("wf", "run-1", "run-1"); let streams = Streams::new(store.clone(), owners.clone(), options); @@ -1431,6 +1465,61 @@ mod tests { setup.streams.append(other).await.unwrap(); } + async fn last_kind(store: &MemoryStore, topic: &str) -> (i32, String) { + let page = store + .read(StoreReadRequest { + chain: Some(chain("run-1")), + topic: topic.to_string(), + max_records: 100, + ..Default::default() + }) + .await + .unwrap(); + let record = StreamRecord::decode(&*page.records.last().unwrap().record).unwrap(); + (record.kind, record.producer_id) + } + + #[tokio::test] + async fn without_close_flags_a_close_is_a_record_that_ends_every_read() { + let setup = setup_on(without_producer_state(), quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup + .streams + .close(CloseRequest { + stream: Some(stream("out")), + }) + .await + .unwrap(); + assert_eq!( + last_kind(&setup.store, "out").await, + (StreamRecordKind::Close as i32, String::new()) + ); + // The store can't refuse it, so it lands where no reader delivers it, and so does a + // repeat of the batch that landed before the close. + setup.streams.append(append("p", 2, "2")).await.unwrap(); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let before = setup.owners.describes(); + let (delivered, _) = read_to_end(&setup.streams, Duration::from_secs(5)).await; + assert_eq!(delivered, 1); + assert_eq!(setup.owners.describes(), before); + } + + #[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()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup.owners.end("wf", WorkflowExecutionStatus::Terminated); + tokio::time::advance(OWNER_RECHECK + Duration::from_secs(1)).await; + assert_eq!( + kind(setup.streams.append(append("p", 2, "2")).await), + StreamFailureKind::Closed + ); + assert_eq!( + last_kind(&setup.store, "out").await, + (StreamRecordKind::Close as i32, "p".to_string()) + ); + } + #[tokio::test] async fn a_retire_drops_the_attempts_state_on_the_chain_it_wrote() { let setup = setup(StreamsOptions::default()); @@ -1604,15 +1693,15 @@ mod tests { } #[tokio::test] - async fn a_read_ends_when_its_owner_chain_ends_and_marks_it_closed() { + async fn a_read_ends_when_its_owner_chain_ends_and_writes_nothing() { let setup = setup(quick()); setup.streams.append(append("p", 1, "1")).await.unwrap(); setup.owners.end("wf", WorkflowExecutionStatus::Completed); let (delivered, end) = read_to_end(&setup.streams, Duration::from_secs(5)).await; assert_eq!(delivered, 1); assert!(end.records.is_empty()); - // The read saw the chain end, and says so to later producers. - assert!(chain_closed(&setup.store, "run-1").await); + // A reader needs no write access, so the producers' owner check closes the chain. + assert!(!chain_closed(&setup.store, "run-1").await); } #[tokio::test] @@ -1694,7 +1783,7 @@ mod tests { request = read_after(&page.cursor, &page.state, Duration::from_secs(5)); } assert!(done); - assert!(chain_closed(&setup.store, "run-1").await); + assert!(!chain_closed(&setup.store, "run-1").await); } #[tokio::test] diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index ef29393049..2d0f5166a7 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -8,6 +8,7 @@ //! records, and the error every call fails with. Lang reaches a store through Core's //! `StreamService`, and Core's Worker appends a Workflow's own output through the same trait. +mod close; mod cursor; mod error; mod handle; @@ -22,6 +23,7 @@ mod store; #[cfg(feature = "redis")] pub use self::redis::{RedisStore, RedisStoreOptions}; +pub use close::{CHAIN_LOG, append_batch, close_chain, close_topic}; pub use cursor::{BEGINNING, END, cursor_position, mint_cursor, stream_hash}; pub use error::{StreamError, StreamResult}; pub use handle::{OWNER_RECHECK, OWNER_START_GRACE, Streams, StreamsOptions}; @@ -31,8 +33,8 @@ 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, stored_append_record, - workflow_batch_digest, workflow_producer_id, + WORKFLOW_OWNER_KIND, activity_producer_id, check_append_digest, 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/reader.rs b/crates/streams/src/reader.rs index ff93d39f3a..fdea7dfb09 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -100,6 +100,9 @@ pub(crate) struct ReadState { /// found one. The owner start grace counts from it. #[prost(uint64, tag = "11")] pub(crate) owner_missing_since_ms: u64, + /// Set once the read reached the first `CLOSE` record, so it ends there. + #[prost(bool, tag = "12")] + closed: bool, } /// What a read knows of one producer. @@ -253,6 +256,10 @@ impl ReadState { /// copy. A read that starts from a cursor without state knows only the cursor record's sequence, /// so it can deliver a copy whose first copy came before the cursor. /// +/// The first `CLOSE` record of the topic, or of the chain, ends the read on every store. The read +/// answers the records before it, and later calls answer none. A `TOPIC` record is not the topic +/// reader's, so the read skips it. The read writes nothing. +/// /// A record that does not parse fails the call as `RECORD` with its cursor, so the caller resumes /// past it on purpose. Skipping it would lose data no one hears of. The records before it in the /// same page are returned first, and the next call fails. @@ -268,13 +275,21 @@ pub async fn read_page( } /// [read_page] with the state already decoded, for a caller that keeps fields of its own in it. -/// The answer's `state` is left empty. Also answers whether the store marked the stream closed. +/// The answer's `state` is left empty. Also answers whether the store marked the stream closed, +/// or the read reached a `CLOSE` record. pub(crate) async fn read_with( store: &dyn StreamStore, target: &ReadTarget, request: &ReadRequest, state: &mut ReadState, ) -> StreamResult<(ReadResponse, bool)> { + if state.closed { + let response = ReadResponse { + cursor: request.after.clone(), + ..Default::default() + }; + return Ok((response, true)); + } let from_end = request.after == END; let position = if from_end { store @@ -332,7 +347,7 @@ pub(crate) async fn read_with( }, }) .await?; - let closed = page.closed; + let mut closed = page.closed; let mut records = Vec::with_capacity(page.records.len()); let mut previous = start; for stored in page.records { @@ -347,6 +362,21 @@ pub(crate) async fn read_with( } Err(_) => break, }; + match StreamRecordKind::try_from(record.kind) { + // The cursor stays before the close, so a read that resumes there ends at it too. + Ok(StreamRecordKind::Close) + if record.topic.is_empty() || record.topic == target.topic => + { + state.closed = true; + closed = true; + break; + } + Ok(StreamRecordKind::Close | StreamRecordKind::Topic) => { + previous = cursor; + continue; + } + _ => {} + } let stale = match state.observe(&record) { Observed::Copy => { previous = cursor; @@ -387,7 +417,7 @@ pub(crate) async fn read_with( )) } -/// A stored record as a reader delivers it. Fails on bytes that are no record, and on a kind no +/// A stored record as the read takes it. Fails on bytes that are no record, and on a kind no /// writer stores, such as the `SUPERSEDED` that only readers synthesize. /// /// Empty bytes are no record either. Every stored record names its topic, so it is never empty, @@ -398,7 +428,12 @@ fn parse(bytes: &[u8]) -> Result { } let mut record = StreamRecord::decode(bytes).map_err(|error| error.to_string())?; match StreamRecordKind::try_from(record.kind) { - Ok(StreamRecordKind::Data | StreamRecordKind::Finish) => {} + Ok( + StreamRecordKind::Data + | StreamRecordKind::Finish + | StreamRecordKind::Close + | StreamRecordKind::Topic, + ) => {} Ok(StreamRecordKind::Unspecified) => record.kind = StreamRecordKind::Data as i32, Err(_) => return Err(format!("no writer stores a record of kind {}", record.kind)), } @@ -692,6 +727,64 @@ mod tests { assert_eq!(values(&earlier), ["2"]); } + fn control(kind: StreamRecordKind, topic: &str) -> Vec { + StreamRecord { + topic: topic.to_string(), + kind: kind as i32, + producer_id: "closer".to_string(), + attempt: 1, + ..Default::default() + } + .encode_to_vec() + } + + #[tokio::test] + async fn the_first_close_ends_the_read_and_every_later_call() { + let log = Log::default(); + log.write("p", 1, b"1"); + log.push(control(StreamRecordKind::Topic, "out")); + log.write("p", 1, b"2"); + log.push(control(StreamRecordKind::Close, "out")); + log.write("p", 1, b"3"); + log.push(control(StreamRecordKind::Close, "out")); + let mut state = ReadState::default(); + let request = ReadRequest { + after: BEGINNING.to_string(), + ..Default::default() + }; + let (page, closed) = read_with(&log, &target(), &request, &mut state) + .await + .unwrap(); + assert!(closed); + assert_eq!(values(&page), ["1", "2"]); + // The cursor stays before the close, so a read that resumes there ends at it too. + assert_eq!(page.cursor, cursor("3")); + let lookups = log.lookups.load(Ordering::Relaxed); + let next = ReadRequest { + after: page.cursor.clone(), + ..Default::default() + }; + let (again, closed) = read_with(&log, &target(), &next, &mut state).await.unwrap(); + assert!(closed && again.records.is_empty()); + assert_eq!(log.lookups.load(Ordering::Relaxed), lookups); + let resumed = read(&log, &page.cursor, &[]).await.unwrap(); + assert!(resumed.records.is_empty()); + let state = ReadState::decode_from(&resumed.state).unwrap(); + assert!(state.closed); + } + + #[tokio::test] + async fn a_chain_close_ends_the_read_and_another_topics_close_does_not() { + let log = Log::default(); + log.write("p", 1, b"1"); + log.push(control(StreamRecordKind::Close, "other")); + log.write("p", 1, b"2"); + log.push(control(StreamRecordKind::Close, "")); + log.write("p", 1, b"3"); + let read = read(&log, BEGINNING, &[]).await.unwrap(); + assert_eq!(values(&read), ["1", "2"]); + } + fn values(response: &ReadResponse) -> Vec { response .records @@ -732,7 +825,15 @@ mod tests { Some(index) => cursor(&(index + 1).to_string()), None => BEGINNING.to_string(), }; - let page = read(&log, &after, &[]).await.unwrap(); + let mut state = ReadState::default(); + let request = ReadRequest { + after, + ..Default::default() + }; + let (page, closed) = read_with(&log, &target(), &request, &mut state) + .await + .unwrap(); + assert_eq!(closed, case["closed"].as_bool().unwrap_or(false), "{label}"); let delivered: Vec = page .records .iter() diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index 1d01d0b958..bcbec50b42 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -145,6 +145,29 @@ fn stored_record( Ok(stored) } +/// 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 { + topic: topic.to_string(), + kind: StreamRecordKind::Close as i32, + producer_id: producer_id.to_string(), + attempt, + ..Default::default() + } +} + +/// A `TOPIC` record that `producer_id` at `attempt` writes to the chain's control log before its +/// first record on `topic`. +pub fn topic_record(topic: &str, producer_id: &str, attempt: i64) -> StreamRecord { + StreamRecord { + topic: topic.to_string(), + kind: StreamRecordKind::Topic as i32, + producer_id: producer_id.to_string(), + attempt, + ..Default::default() + } +} + fn metadata_payload(data: &[u8]) -> Payload { Payload { metadata: [("encoding".to_string(), METADATA_ENCODING.to_vec())].into(), diff --git a/crates/streams/testdata/reader_rules.json b/crates/streams/testdata/reader_rules.json index ff47cf47ca..afbbf9a11a 100644 --- a/crates/streams/testdata/reader_rules.json +++ b/crates/streams/testdata/reader_rules.json @@ -1,83 +1,332 @@ { - "comment": "The rules a reader applies on every store. Each case lists the stored records of one topic in log order, and the indexes of the records a read from the start delivers (or a read that resumes after the record at index resume_after, with no state). kind defaults to 1 (DATA) and topic to \"out\".", + "comment": "The rules a reader applies on every store. Each case lists the stored records of one topic in log order, and the indexes of the records a read from the start delivers (or a read that resumes after the record at index resume_after, with no state). kind defaults to 1 (DATA), 5 is CLOSE and 6 is TOPIC. topic defaults to \"out\". closed says whether the read ended at a CLOSE.", "cases": [ { "label": "a copy after a lost answer is dropped", "records": [ - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "q", "attempt": 1, "sequence": 1}, - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "p", "attempt": 1, "sequence": 2} + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "q", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 2 + } ], - "delivered": [0, 1, 3] + "delivered": [ + 0, + 1, + 3 + ] }, { "label": "a copy of a whole batch is dropped", "records": [ - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "p", "attempt": 1, "sequence": 2}, - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "p", "attempt": 1, "sequence": 2}, - {"producer": "p", "attempt": 1, "sequence": 3} + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 2 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 2 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 3 + } ], - "delivered": [0, 1, 4] + "delivered": [ + 0, + 1, + 4 + ] }, { "label": "a record with no sequence is never a copy", "records": [ - {"producer": "p", "attempt": 1, "sequence": 0}, - {"producer": "p", "attempt": 1, "sequence": 0} + { + "producer": "p", + "attempt": 1, + "sequence": 0 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 0 + } ], - "delivered": [0, 1] + "delivered": [ + 0, + 1 + ] }, { "label": "a gap in the sequence is no copy", "records": [ - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "p", "attempt": 1, "sequence": 4}, - {"producer": "p", "attempt": 1, "sequence": 3} + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 4 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 3 + } ], - "delivered": [0, 1] + "delivered": [ + 0, + 1 + ] }, { "label": "a copy of an older attempt is delivered as stale", "records": [ - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "p", "attempt": 2, "sequence": 1}, - {"producer": "p", "attempt": 1, "sequence": 1} + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 2, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 1 + } ], - "delivered": [0, 1, 2], - "stale": [2] + "delivered": [ + 0, + 1, + 2 + ], + "stale": [ + 2 + ] }, { "label": "producers count their sequences apart", "records": [ - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "q", "attempt": 1, "sequence": 1}, - {"producer": "q", "attempt": 1, "sequence": 1} + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "q", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "q", + "attempt": 1, + "sequence": 1 + } ], - "delivered": [0, 1] + "delivered": [ + 0, + 1 + ] }, { "label": "a read from a cursor knows the cursor record's sequence", "records": [ - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "p", "attempt": 1, "sequence": 2}, - {"producer": "p", "attempt": 1, "sequence": 2}, - {"producer": "p", "attempt": 1, "sequence": 3} + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 2 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 2 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 3 + } ], "resume_after": 1, - "delivered": [3] + "delivered": [ + 3 + ] }, { "label": "a read from a cursor can deliver a copy whose first copy came before it", "records": [ - {"producer": "p", "attempt": 1, "sequence": 1}, - {"producer": "q", "attempt": 1, "sequence": 1}, - {"producer": "p", "attempt": 1, "sequence": 1} + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "q", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 1 + } ], "resume_after": 1, - "delivered": [2] + "delivered": [ + 2 + ] + }, + { + "label": "the first close ends the topic", + "records": [ + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "c", + "attempt": 1, + "sequence": 0, + "kind": 5 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 2 + }, + { + "producer": "d", + "attempt": 1, + "sequence": 0, + "kind": 5 + } + ], + "delivered": [ + 0 + ], + "closed": true + }, + { + "label": "a close with no topic ends the chain's topics", + "records": [ + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "c", + "attempt": 1, + "sequence": 0, + "kind": 5, + "topic": "" + }, + { + "producer": "p", + "attempt": 1, + "sequence": 2 + } + ], + "delivered": [ + 0 + ], + "closed": true + }, + { + "label": "another topic's close and a topic name are skipped", + "records": [ + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "c", + "attempt": 1, + "sequence": 0, + "kind": 5, + "topic": "other" + }, + { + "producer": "p", + "attempt": 1, + "sequence": 0, + "kind": 6 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 2 + } + ], + "delivered": [ + 0, + 3 + ] + }, + { + "label": "a copy of a landed batch after a close is not delivered", + "records": [ + { + "producer": "p", + "attempt": 1, + "sequence": 1 + }, + { + "producer": "c", + "attempt": 1, + "sequence": 0, + "kind": 5 + }, + { + "producer": "p", + "attempt": 1, + "sequence": 1 + } + ], + "delivered": [ + 0 + ], + "closed": true } ] }