diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto index 63bca5f6c..7385028a1 100644 --- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -29,27 +29,6 @@ message StreamRef { string backend = 3; } -// Experimental. Might change in the future. -// The ordered attempt of a Producer. A higher epoch supersedes the Records of a lower epoch of the -// same Producer. The case is the kind of the Producer. It decides how the Store orders the epochs, -// and which earlier Records a higher epoch withdraws. -message ProducerEpoch { - oneof kind { - // A Workflow run, which writes in its Workflow Task. A higher epoch withdraws the Records - // of the run's Workflow Tasks that never completed, also those of their Local Activities. - WorkflowTaskEpoch workflow = 1; - // A Local Activity. A higher epoch withdraws the Producer's lower epochs. - LocalActivityEpoch local_activity = 2; - // An Activity of a Workflow. A higher epoch withdraws the Producer's lower epochs. - ActivityEpoch activity = 3; - // A Standalone Activity. A higher epoch withdraws the Producer's lower epochs. - ActivityEpoch standalone_activity = 4; - // A client. A higher epoch only fences the lower ones, since a restarted client's earlier - // Records are real messages, not the Records of a failed attempt. - ClientEpoch client = 5; - } -} - // Experimental. Might change in the future. // One Workflow Task attempt of a run. It orders by started_event_id, then attempt, then // started_time. @@ -86,32 +65,78 @@ message ClientEpoch { } // Experimental. Might change in the future. -// The code that wrote a batch. Its id stays the same across attempts, and its epoch orders them. +// A Workflow run Producer. It writes in its Workflow Task. A higher epoch withdraws the Records of +// the run's Workflow Tasks that never completed, with those of their Local Activities. +message WorkflowProducer { + // The first execution run id of the run chain. + string run_id = 1; + WorkflowTaskEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// A Local Activity Producer. A higher epoch withdraws the Producer's lower epochs. +message LocalActivityProducer { + string run_id = 1; + LocalActivityEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// An Activity Producer, for an Activity of a Workflow or a Standalone Activity. A higher epoch +// withdraws the Producer's lower epochs. +message ActivityProducer { + string run_id = 1; + ActivityEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// A client Producer. A higher epoch only fences the lower ones, since a restarted client's earlier +// Records are real messages, not the Records of a failed attempt. A client has no run. +message ClientProducer { + ClientEpoch epoch = 1; +} + +// Experimental. Might change in the future. +// The code that wrote a batch. Its id stays the same across attempts, and the kind's case orders +// its epochs and picks which earlier Records a higher epoch withdraws. A run-based kind carries its +// run, so a client can't. message Producer { // Stable across attempts, such as `wf:` for the Workflow Producer of a run. string id = 1; - // Required. Its case gives the kind of the Producer. - ProducerEpoch epoch = 2; - // The run of the Producer, so nobody parses the id. Empty for a client. - string run_id = 3; + // Required. The case is the kind of the Producer. + oneof kind { + WorkflowProducer workflow = 2; + LocalActivityProducer local_activity = 3; + ActivityProducer activity = 4; + ActivityProducer standalone_activity = 5; + ClientProducer client = 6; + } } // Experimental. Might change in the future. // The Producer of a write, and which earlier Records the write withdraws. An append, a close and // a writer carry the same fields, so the Store orders and withdraws each write the same way. message StreamWriter { - // Required on a write. Its epoch's case picks the withdrawal rule. + // Required on a write. Its Producer's kind picks the withdrawal rule. Producer producer = 1; - // Workflow and Local Activity only. The run's last completed Workflow Task, from a Local - // Activity's start. Its started_event_id is 0 before the first. The write withdraws the run's - // Local Activity Records whose Workflow Task never completed. - WorkflowTaskEpoch previous_completed_task = 2; - // Workflow and Local Activity only. The started event ids of the run's failed or timed out - // Workflow Tasks. The write withdraws the run's Local Activity Records of these tasks. - repeated int64 failed_tasks = 3; - // Retry run only. The previous run, on the retry run's first write. Every Record whose - // Producer carries it is withdrawn. - string withdraw_run = 4; + // Set only by a Workflow or Local Activity write, since only a run withdraws orphaned Local + // Activity Records or a previous run. Unset for an Activity or client write. + RunWithdrawal withdrawal = 2; +} + +// Experimental. Might change in the future. +// What a Workflow or Local Activity write withdraws from its run. One write can carry both the task +// facts and a run to withdraw, so these are fields of one message and not a oneof. +message RunWithdrawal { + // The run's last completed Workflow Task, from a Local Activity's start. Its started_event_id + // is 0 before the first. The write withdraws the run's Local Activity Records whose Workflow + // Task never completed. + WorkflowTaskEpoch previous_completed_task = 1; + // The started event ids of the run's failed or timed out Workflow Tasks. The write withdraws + // the run's Local Activity Records of these tasks. + repeated int64 failed_tasks = 2; + // Retry run only. The previous run, on the retry run's first write. Every Record whose Producer + // carries it is withdrawn. + string withdraw_run = 3; } // Experimental. Might change in the future. @@ -125,8 +150,7 @@ message AppendRecord { // batch of a higher epoch, or for a retry run's first write, and a read builds a Superseded item // from them. A reader drops the earlier Records that the facts match. message StreamSuperseded { - // The writer of the batch that withdraws. Its Producer's epoch is empty when only - // withdraw_run applies. + // The writer of the batch that withdraws, with its withdrawal facts. StreamWriter writer = 1; } diff --git a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto index cdd61456e..463f91194 100644 --- a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto +++ b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto @@ -75,7 +75,7 @@ message AppendRecordsRequest { int64 sequence = 4; repeated temporal.api.stream.v1.AppendRecord records = 5; // Only read by the server, on a Workflow Producer's batch. The batch withdraws no Workflow - // Task and keeps no task facts, but the writer's withdraw_run still applies. + // Task and keeps no task facts, but the writer's withdrawal withdraw_run still applies. bool defer_withdrawal = 6; // Only read by the server. The server checks the Producer against the Activity's current // attempt, so nobody else writes as that Activity. 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 fcec138bd..115091078 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -33,9 +33,10 @@ message MemoryStreamConfig { message OpenWriterRequest { string namespace = 1; temporal.api.stream.v1.StreamRef ref = 2; - // Each batch sends it as AppendRecords does. For a Local Activity it has the run's task facts, - // so the writes withdraw the run's orphaned Local Activity Records. Its withdraw_run must be - // empty, since only a retry run's first append withdraws a run, and a writer sends many. + // Each batch sends it as AppendRecords does. For a Local Activity its withdrawal has the run's + // task facts, so the writes withdraw the run's orphaned Local Activity Records. The withdrawal's + // withdraw_run must be empty, since only a retry run's first append withdraws a run, and a + // writer sends many. temporal.api.stream.v1.StreamWriter writer = 3; // A write sends at most this many Records in one batch. Zero is 1000. int32 max_batch_size = 4; diff --git a/crates/streams/src/epoch.rs b/crates/streams/src/epoch.rs index 697d66ef6..15b1ca6a2 100644 --- a/crates/streams/src/epoch.rs +++ b/crates/streams/src/epoch.rs @@ -7,8 +7,8 @@ use crate::{ StreamError, StreamResult, proto::{ - Producer, ProducerEpoch, StreamRecord, StreamSuperseded, WorkflowTaskEpoch, - producer_epoch::Kind, stream_record::Item, + Producer, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer::Kind, + stream_record::Item, }, }; use prost_types::Timestamp; @@ -42,22 +42,52 @@ impl From<&Kind> for ProducerKind { } } -/// Refuses a Producer with no epoch, or with an epoch that has no case. +/// Refuses a Producer with no kind case. impl TryFrom<&Producer> for ProducerKind { type Error = StreamError; fn try_from(producer: &Producer) -> StreamResult { - let epoch = producer - .epoch - .as_ref() - .ok_or_else(|| StreamError::invalid("The Producer has no epoch"))?; - kind(epoch).ok_or_else(|| StreamError::invalid("The Producer's epoch has no kind")) + kind_of(producer).ok_or_else(|| StreamError::invalid("The Producer has no kind")) } } -/// The kind of an epoch's case, `None` when the epoch has no case. -fn kind(epoch: &ProducerEpoch) -> Option { - epoch.kind.as_ref().map(ProducerKind::from) +/// The kind case of a Producer, `None` when it has none. +pub fn kind_of(producer: &Producer) -> Option { + producer.kind.as_ref().map(ProducerKind::from) +} + +/// The run of a Producer. Empty for a client, which has no run. +pub fn producer_run_id(producer: &Producer) -> &str { + match &producer.kind { + Some(Kind::Workflow(w)) => &w.run_id, + Some(Kind::LocalActivity(l)) => &l.run_id, + Some(Kind::Activity(a) | Kind::StandaloneActivity(a)) => &a.run_id, + Some(Kind::Client(_)) | None => "", + } +} + +/// The run that a writer withdraws, empty for none. +pub fn writer_withdraw_run(writer: &StreamWriter) -> &str { + writer + .withdrawal + .as_ref() + .map_or("", |w| w.withdraw_run.as_str()) +} + +/// The started event ids of the run's failed Workflow Tasks that a writer reports. +pub fn writer_failed_tasks(writer: &StreamWriter) -> &[i64] { + writer + .withdrawal + .as_ref() + .map_or(&[], |w| w.failed_tasks.as_slice()) +} + +/// The run's last completed Workflow Task that a writer reports, `None` for none. +pub fn writer_previous_completed_task(writer: &StreamWriter) -> Option<&WorkflowTaskEpoch> { + writer + .withdrawal + .as_ref() + .and_then(|w| w.previous_completed_task.as_ref()) } /// The length of a [task_key]. @@ -92,20 +122,24 @@ pub fn task_key(task: &WorkflowTaskEpoch) -> [u8; TASK_KEY_LEN] { } /// The key of an epoch. It is the [task_of] key, then the attempt's scheduled seconds and nanos -/// and the attempt, in 8, 4 and 4 bytes. A field that the case doesn't have is zero, and an epoch +/// and the attempt, in 8, 4 and 4 bytes. A field that the case doesn't have is zero, and a Producer /// with no case is all zeros. -pub fn epoch_key(epoch: &ProducerEpoch) -> [u8; EPOCH_KEY_LEN] { - let (scheduled, attempt) = match &epoch.kind { - Some(Kind::LocalActivity(local)) => (None, local.attempt), +pub fn epoch_key(kind: Option<&Kind>) -> [u8; EPOCH_KEY_LEN] { + let (scheduled, attempt) = match kind { + Some(Kind::LocalActivity(local)) => (None, local.epoch.map_or(0, |e| e.attempt)), Some(Kind::Activity(activity) | Kind::StandaloneActivity(activity)) => { - (activity.attempt_scheduled_time.as_ref(), activity.attempt) + let epoch = activity.epoch; + ( + epoch.and_then(|e| e.attempt_scheduled_time), + epoch.map_or(0, |e| e.attempt), + ) } - Some(Kind::Client(client)) => (None, client.attempt), + Some(Kind::Client(client)) => (None, client.epoch.map_or(0, |e| e.attempt)), Some(Kind::Workflow(_)) | None => (None, 0), }; - let (seconds, nanos) = time_fields(scheduled); + let (seconds, nanos) = time_fields(scheduled.as_ref()); let mut key = [0; EPOCH_KEY_LEN]; - key[..TASK_KEY_LEN].copy_from_slice(&task_key(&task_of(epoch))); + key[..TASK_KEY_LEN].copy_from_slice(&task_key(&task_of(kind))); key[24..32].copy_from_slice(&be64(seconds)); key[32..36].copy_from_slice(&be32(nanos)); key[36..].copy_from_slice(&be32(attempt)); @@ -114,16 +148,16 @@ pub fn epoch_key(epoch: &ProducerEpoch) -> [u8; EPOCH_KEY_LEN] { /// The Workflow Task of an epoch. It is a Workflow's own task, or the task that dispatched a Local /// Activity, and zero for the other kinds. -pub fn task_of(epoch: &ProducerEpoch) -> WorkflowTaskEpoch { - match &epoch.kind { - Some(Kind::Workflow(task)) => *task, - Some(Kind::LocalActivity(local)) => local.task.unwrap_or_default(), +pub fn task_of(kind: Option<&Kind>) -> WorkflowTaskEpoch { + match kind { + Some(Kind::Workflow(w)) => w.epoch.unwrap_or_default(), + Some(Kind::LocalActivity(local)) => local.epoch.and_then(|e| e.task).unwrap_or_default(), _ => WorkflowTaskEpoch::default(), } } /// Orders two epochs field by field. -pub fn compare_epochs(a: &ProducerEpoch, b: &ProducerEpoch) -> Ordering { +pub fn compare_epochs(a: Option<&Kind>, b: Option<&Kind>) -> Ordering { epoch_key(a).cmp(&epoch_key(b)) } @@ -150,48 +184,52 @@ pub fn withdraws(superseded: &StreamSuperseded, record: &StreamRecord) -> bool { let Some(producer) = data.producer.as_ref() else { return false; }; - if !writer.withdraw_run.is_empty() && producer.run_id == writer.withdraw_run { + let withdraw_run = writer_withdraw_run(writer); + if !withdraw_run.is_empty() && producer_run_id(producer) == withdraw_run { return true; } let Some(withdrawing) = writer.producer.as_ref() else { return false; }; - let (Some(epoch), Some(old)) = (withdrawing.epoch.as_ref(), producer.epoch.as_ref()) else { + let (new, old) = (withdrawing.kind.as_ref(), producer.kind.as_ref()); + let (Some(new_kind), Some(old_kind)) = (new, old) else { return false; }; - match kind(epoch) { - None | Some(ProducerKind::Client) => return false, - Some(ProducerKind::Workflow) => {} + match ProducerKind::from(new_kind) { + ProducerKind::Client => return false, + ProducerKind::Workflow => {} same => { - return kind(old) == same + return ProducerKind::from(old_kind) == same && producer.id == withdrawing.id - && compare_epochs(old, epoch) == Ordering::Less; + && compare_epochs(old, new) == Ordering::Less; } } if !of_run(producer, withdrawing) { return false; } - let (task, old) = (task_of(epoch), task_of(old)); - let started = old.started_event_id; - if writer.failed_tasks.contains(&started) { + let (task, old_task) = (task_of(new), task_of(old)); + let started = old_task.started_event_id; + if writer_failed_tasks(writer).contains(&started) { return true; } - let completed = writer.previous_completed_task.as_ref(); + let completed = writer_previous_completed_task(writer); let after = completed.map_or(0, |c| c.started_event_id); if after < started && started < task.started_event_id { return true; } let lower_at = |at: &WorkflowTaskEpoch| { - started == at.started_event_id && compare_tasks(&old, at) == Ordering::Less + started == at.started_event_id && compare_tasks(&old_task, at) == Ordering::Less }; lower_at(&task) || completed.is_some_and(lower_at) } /// Whether a Producer is the Workflow run that withdraws, or one of its Local Activities. fn of_run(producer: &Producer, withdrawing: &Producer) -> bool { - match producer.epoch.as_ref().and_then(kind) { + match kind_of(producer) { Some(ProducerKind::Workflow) => producer.id == withdrawing.id, - Some(ProducerKind::LocalActivity) => producer.run_id == withdrawing.run_id, + Some(ProducerKind::LocalActivity) => { + producer_run_id(producer) == producer_run_id(withdrawing) + } _ => false, } } @@ -199,17 +237,14 @@ fn of_run(producer: &Producer, withdrawing: &Producer) -> bool { #[cfg(test)] mod tests { use super::*; - use crate::proto::ClientEpoch; + use crate::proto::{ClientEpoch, ClientProducer, producer::Kind}; // Without the clamp, the bytes of a negative field would outrank every real epoch. #[test] fn a_negative_field_compares_as_zero() { - let negative = ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt: -1 })), - }; - assert_eq!( - compare_epochs(&negative, &ProducerEpoch::default()), - Ordering::Equal - ); + let negative = Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: -1 }), + }); + assert_eq!(compare_epochs(Some(&negative), None), Ordering::Equal); } } diff --git a/crates/streams/src/ids.rs b/crates/streams/src/ids.rs index d50cf47f3..45eb1e77b 100644 --- a/crates/streams/src/ids.rs +++ b/crates/streams/src/ids.rs @@ -11,7 +11,11 @@ //! script. Each part is percent-encoded, so a `:` or a brace in an id can't make two Streams share //! a key. -use crate::{StreamError, StreamResult, epoch::ProducerKind, proto::Producer}; +use crate::{ + StreamError, StreamResult, + epoch::{ProducerKind, producer_run_id}, + proto::Producer, +}; /// The longest Stream id, in bytes. pub const MAX_STREAM_ID_BYTES: usize = 4096; @@ -151,13 +155,14 @@ pub fn check_producer(producer: &Producer) -> StreamResult<()> { kind, ProducerKind::Workflow | ProducerKind::LocalActivity | ProducerKind::Activity ); - if runs && decode_part(parts[0]).as_deref() != Some(producer.run_id.as_str()) { + if runs && decode_part(parts[0]).as_deref() != Some(producer_run_id(producer)) { return Err(StreamError::invalid(format!( "The Producer id {:?} names another run than its run id {:?}", - producer.id, producer.run_id + producer.id, + producer_run_id(producer) ))); } - if kind == ProducerKind::Client && !producer.run_id.is_empty() { + if kind == ProducerKind::Client && !producer_run_id(producer).is_empty() { return Err(StreamError::invalid("A client Producer has no run id")); } Ok(()) @@ -207,20 +212,34 @@ impl StreamKeys { #[cfg(test)] mod tests { use super::*; - use crate::proto::{ProducerEpoch, producer_epoch::Kind}; + use crate::proto::{ + ActivityProducer, ClientProducer, LocalActivityProducer, WorkflowProducer, producer::Kind, + }; fn producer(kind: ProducerKind, id: &str, run_id: &str) -> Producer { + let run_id = run_id.to_string(); let kind = match kind { - ProducerKind::Workflow => Kind::Workflow(Default::default()), - ProducerKind::LocalActivity => Kind::LocalActivity(Default::default()), - ProducerKind::Activity => Kind::Activity(Default::default()), - ProducerKind::StandaloneActivity => Kind::StandaloneActivity(Default::default()), - ProducerKind::Client => Kind::Client(Default::default()), + ProducerKind::Workflow => Kind::Workflow(WorkflowProducer { + run_id, + ..Default::default() + }), + ProducerKind::LocalActivity => Kind::LocalActivity(LocalActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::Activity => Kind::Activity(ActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::StandaloneActivity => Kind::StandaloneActivity(ActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::Client => Kind::Client(ClientProducer::default()), }; Producer { id: id.to_string(), - epoch: Some(ProducerEpoch { kind: Some(kind) }), - run_id: run_id.to_string(), + kind: Some(kind), } } @@ -267,14 +286,9 @@ mod tests { producer(ProducerKind::Workflow, &workflow_producer_id(run), "other"), producer(ProducerKind::Activity, "act:run%3A1:a", run), Producer { - epoch: None, - ..producer(ProducerKind::Workflow, "wf:x", "x") - }, - Producer { - epoch: Some(ProducerEpoch::default()), + kind: None, ..producer(ProducerKind::Workflow, "wf:x", "x") }, - producer(ProducerKind::Client, "client:a:b", "run"), producer(ProducerKind::Client, "client:a|b", ""), producer(ProducerKind::Client, "client:run%3a1", ""), ]; diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 52804d291..de076e762 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -5,10 +5,12 @@ use crate::{ Reason, StreamError, StreamResult, digest::DIGEST_LEN, - epoch::{EPOCH_KEY_LEN, ProducerKind, epoch_key}, - ids::StreamKeys, + epoch::{ + EPOCH_KEY_LEN, ProducerKind, TASK_KEY_LEN, epoch_key, producer_run_id, task_key, task_of, + }, + ids::{StreamKeys, encode_part}, offset::{entry_id_of_offset, offset_of_entry_id}, - proto::{Producer, StreamRef}, + proto::{Producer, StreamRef, WorkflowTaskEpoch}, store::{ Appended, Batch, Capabilities, Closed, ConflictPolicy, Create, CreateOutcome, Created, Entry, FORMAT_VERSION, MAX_SEQUENCE, Owner, Store, StreamState, TOMBSTONE, owner_chain, @@ -189,6 +191,82 @@ fn entry_id(text: &str) -> StreamResult { )) } +type TaskKey = [u8; TASK_KEY_LEN]; + +/// The started event id of a task key, its first 8 bytes. +fn started(key: &TaskKey) -> i64 { + let mut id = [0; 8]; + id.copy_from_slice(&key[..8]); + i64::from_be_bytes(id) +} + +/// Whether the facts of task `t` show that task `key` never completed. An orphan item withdraws the +/// Records of such a task. +fn never_completed(key: &TaskKey, t: &TaskKey, c: Option<&TaskKey>, failed: &[i64]) -> bool { + let (s, ts) = (started(key), started(t)); + let cs = c.map_or(0, started); + failed.contains(&s) + || (cs < s && s < ts) + || (s == ts && key < t) + || c.is_some_and(|c| s == cs && key < c) +} + +/// The orphan rule over a run's open Local Activity tasks, as `orphan_tasks` in `lib.lua`. +fn orphan_tasks( + open: &[TaskKey], + t: &TaskKey, + c: Option<&TaskKey>, + failed: &[i64], +) -> (Vec, Vec) { + let (mut withdrawn, mut kept) = (vec![], vec![]); + for key in open { + if never_completed(key, t, c, failed) { + withdrawn.push(*key); + } else if c.is_none_or(|c| key > c) { + kept.push(*key); + } + } + (withdrawn, kept) +} + +/// The task key of the Workflow Task `[started_event_id]`, with no attempt and no time. +fn started_key(started_event_id: i64) -> TaskKey { + task_key(&WorkflowTaskEpoch { + started_event_id, + ..Default::default() + }) +} + +/// The task keys of a field that keeps them back to back, such as `oa|` or `tf|`. +fn task_keys(value: &[u8]) -> StreamResult> { + let (keys, rest) = value.as_chunks::(); + if !rest.is_empty() { + return Err(refuse( + Reason::StorageCorrupt, + "a field of task keys ends in a partial key", + )); + } + Ok(keys.to_vec()) +} + +/// Whether the facts that `tf|` keeps show that task `key` never completed. The zero key in +/// the completed place works as no completed task, since no task key is below it. +fn fenced_by_facts(facts: &[TaskKey], key: &TaskKey) -> StreamResult { + let [t, c, failed @ ..] = facts else { + return Err(refuse( + Reason::StorageCorrupt, + "the run's task facts have fewer than two task keys", + )); + }; + let failed: Vec = failed.iter().map(started).collect(); + Ok(never_completed(key, t, Some(c), &failed)) +} + +/// The `ow||` field, set when a `Superseded` item withdrew the run's task. +fn withdrawn_field(run: &str, task: &TaskKey) -> Vec { + [format!("ow|{run}|").as_bytes(), task].concat() +} + impl Keys { fn get(&self, field: impl AsRef<[u8]>) -> Option<&[u8]> { self.meta.get(field.as_ref()).map(Vec::as_slice) @@ -361,6 +439,14 @@ impl Keys { } } + fn withdraw_run(&mut self, run: &str) -> bool { + if run.is_empty() || self.get(format!("wr|{run}")).is_some() { + return false; + } + self.set(format!("wr|{run}"), "1"); + self.get(format!("run|{run}")).is_some() + } + /// Refuses `count` entries whose ids would not fit an offset. The ids go on from the newest id /// the Stream gave, as `check_ids_fit` in `lib.lua`. fn check_ids_fit(&self, now: i64, count: u64) -> StreamResult<()> { @@ -497,6 +583,14 @@ impl MemoryStore { } } +fn encode_run(run_id: &str) -> String { + if run_id.is_empty() { + String::new() + } else { + encode_part(run_id) + } +} + fn offsets(first: &str, last: &str) -> StreamResult> { if first.is_empty() || last.is_empty() { return Ok(None); @@ -510,7 +604,7 @@ fn offsets(first: &str, last: &str) -> StreamResult> { #[async_trait::async_trait] impl Store for MemoryStore { fn capabilities(&self) -> Capabilities { - Capabilities::ALL.without(Capabilities::RUN_FENCE) + Capabilities::ALL } async fn create( @@ -528,6 +622,8 @@ impl Store for MemoryStore { Owner::Client => "client", }; let (safety, tombstone) = (millis(self.safety_ttl), millis(TOMBSTONE)); + let withdrawal = &create.withdrawal; + let withdrawn_run = encode_run(withdrawal.withdraw_run); let chain = if owner == "wf" { owner_chain(stream_ref) } else { @@ -544,8 +640,9 @@ impl Store for MemoryStore { if !request_id.is_empty() { keys.set("rid", request_id); } + keys.withdraw_run(&withdrawn_run); keys.refresh(now, safety, tombstone); - return Ok((CreateOutcome::Created, retention)); + return Ok((CreateOutcome::Created, retention, false)); } keys.check_live(now)?; keys.check_identity(owner, &chain)?; @@ -554,18 +651,38 @@ impl Store for MemoryStore { } // A retry of the create that made the Stream gets its answer again. if !request_id.is_empty() && keys.get("rid") == Some(request_id.as_bytes()) { - return Ok((CreateOutcome::Created, keys.number("ret"))); + return Ok((CreateOutcome::Created, keys.number("ret"), false)); } if create.conflict == ConflictPolicy::Fail { return Err(refuse(Reason::AlreadyExists, "the Stream exists")); } - Ok((CreateOutcome::Exists, keys.number("ret"))) + if !withdrawn_run.is_empty() { + keys.check_ids_fit(now, 1)?; + } + let mut flagged = false; + if keys.withdraw_run(&withdrawn_run) { + keys.trim(now); + let id = keys.add( + now, + Fields { + record: None, + superseded: Some(withdrawal.superseded.to_vec()), + }, + ); + keys.set("newest", id); + keys.refresh(now, safety, tombstone); + flagged = true; + } + Ok((CreateOutcome::Exists, keys.number("ret"), flagged)) }); - let (outcome, stored) = created?; + let (outcome, stored, flagged) = created?; + if flagged { + self.appended.notify_waiters(); + } Ok(Created { outcome, retention: Duration::from_millis(stored.max(0) as u64), - flagged: false, + flagged, }) } @@ -578,11 +695,23 @@ impl Store for MemoryStore { let stream_id = stream_ref.id.as_str(); let producer = batch.producer; let kind = ProducerKind::try_from(producer)?; - let epoch = epoch_key(&producer.epoch.unwrap_or_default()); + let epoch = epoch_key(producer.kind.as_ref()); + let run = encode_run(producer_run_id(producer)); let withdrawal = &batch.withdrawal; + let withdrawn_run = encode_run(withdrawal.withdraw_run); let (safety, tombstone) = (millis(self.safety_ttl), millis(TOMBSTONE)); let field = format!("p|{}|{}", kind as i32, producer.id); let member = format!("{}|{}", kind as i32, producer.id); + // The Workflow Task of a Local Activity of a run, which the run's task facts can withdraw. + let task = (kind == ProducerKind::LocalActivity && !run.is_empty()) + .then(|| task_key(&task_of(producer.kind.as_ref()))); + let facts = withdrawal.tasks.filter(|_| task.is_some()); + let completed = facts.and_then(|facts| facts.completed).map(task_key); + let open_field = format!("oa|{run}"); + // The task facts of the run's highest task, as task keys back to back. These are its own + // key, the completed task's key or the zero key before the first, and the key of each + // failed task's started event id. + let facts_field = format!("tf|{run}"); let appended = self.with(namespace, stream_id, |keys, now| { // The format version comes first, since a newer format can change the Producer field. keys.check_live(now)?; @@ -632,25 +761,80 @@ impl Store for MemoryStore { if keys.get("closed").is_some() { return Err(refuse(Reason::Closed, "the Stream is closed")); } + if !run.is_empty() && keys.get(format!("wr|{run}")).is_some() { + return Err(refuse( + Reason::Fenced, + "a Workflow retry withdrew the run of the Producer", + )); + } if stored.is_some_and(|stored| epoch < stored.fence) { return Err(refuse( Reason::Fenced, "the epoch is lower than the newest one of the Producer", )); } + if let Some(task) = task { + if keys.get(withdrawn_field(&run, &task)).is_some() { + return Err(refuse( + Reason::Fenced, + "a Superseded item withdrew the Workflow Task of the call", + )); + } + // A task can fail before its call writes, so no orphan item withdrew it. + let newest = keys.get(&facts_field).map(task_keys).transpose()?; + if let Some(newest) = newest + && fenced_by_facts(&newest, &task)? + { + return Err(refuse( + Reason::Fenced, + "the run's task facts show that the Workflow Task of the call never \ + completed", + )); + } + } let records = batch.records.len() as u64; - keys.check_ids_fit(now, records + 1)?; + keys.check_ids_fit(now, records + 2)?; // A client's newer epoch only fences, as in `append.lua`. - let flagged = kind != ProducerKind::Client + let newer = kind != ProducerKind::Client && stored .and_then(|stored| stored.held) .is_some_and(|held| held < epoch); + let mut flagged = newer; + if keys.withdraw_run(&withdrawn_run) { + flagged = true; + } + let open = match task { + Some(_) => keys + .get(&open_field) + .map(task_keys) + .transpose()? + .unwrap_or_default(), + None => vec![], + }; + let (withdrawn, mut open) = match (task, facts) { + (Some(task), Some(facts)) => { + orphan_tasks(&open, &task, completed.as_ref(), facts.failed_tasks) + } + _ => (vec![], open), + }; keys.trim(now); keys.prune(now); let at = keys.next_ms(now); let (mut first, mut last) = (String::new(), String::new()); + // The orphan item has an entry of its own, as in `append.lua`. + let mut orphan = String::new(); + if let (false, Some(facts)) = (withdrawn.is_empty(), facts) { + orphan = keys.add( + now, + Fields { + record: None, + superseded: Some(facts.superseded.to_vec()), + }, + ); + last = orphan.clone(); + } if records == 0 && flagged { last = keys.add( now, @@ -675,9 +859,13 @@ impl Store for MemoryStore { } } + if first.is_empty() { + first = orphan; + } + // Only a newer epoch's item withdraws the held epoch's Records. let held = if records > 0 { Some(epoch) - } else if flagged { + } else if newer { None } else { stored.and_then(|stored| stored.held) @@ -695,12 +883,49 @@ impl Store for MemoryStore { digest: batch.digest, }; keys.set(field.as_str(), value.encode()); + if !run.is_empty() && records > 0 { + keys.set(format!("run|{run}"), "1"); + } + if let Some(task) = task { + // A write with no task facts, as from an older Core, goes in `oa` too, so that a + // later write's facts can withdraw it. + let added = records > 0 && !open.contains(&task); + if added { + open.push(task); + } + if facts.is_some() || added { + if open.is_empty() { + keys.remove(&open_field); + } else { + keys.set(open_field.as_str(), open.concat()); + } + } + // One field per task, so the fence reads one field, as in `append.lua`. + for key in &withdrawn { + keys.set(withdrawn_field(&run, key), "1"); + } + } + if let (Some(task), Some(facts)) = (task, facts) { + // A long Local Activity of an earlier task writes with older facts. + let newest = keys.get(&facts_field).map(task_keys).transpose()?; + let highest = newest.as_ref().and_then(|newest| newest.first()); + if highest.is_none_or(|highest| *highest < task) { + let completed = completed.unwrap_or_else(|| started_key(0)); + let failed = facts.failed_tasks.iter().map(|&failed| started_key(failed)); + let value: Vec = [task, completed] + .into_iter() + .chain(failed) + .flatten() + .collect(); + keys.set(facts_field.as_str(), value); + } + } if !last.is_empty() { keys.set("newest", last.clone()); } keys.producers.insert(member.clone(), at); keys.refresh(now, safety, tombstone); - Ok((false, first, last, flagged)) + Ok((false, first, last, flagged || !withdrawn.is_empty())) }); let (resend, first, last, flagged) = appended?; if !resend && !last.is_empty() { @@ -721,7 +946,8 @@ impl Store for MemoryStore { ) -> StreamResult { let stream_id = stream_ref.id.as_str(); let kind = ProducerKind::try_from(producer)?; - let epoch = epoch_key(&producer.epoch.unwrap_or_default()); + let epoch = epoch_key(producer.kind.as_ref()); + let run = encode_run(producer_run_id(producer)); let field = format!("p|{}|{}", kind as i32, producer.id); let tombstone = millis(TOMBSTONE); let chain = owner_chain(stream_ref); @@ -731,6 +957,12 @@ impl Store for MemoryStore { if keys.get("closed").is_some() { return Ok(true); } + if !run.is_empty() && keys.get(format!("wr|{run}")).is_some() { + return Err(refuse( + Reason::Fenced, + "a Workflow retry withdrew the run of the Producer", + )); + } let stored = keys.get(&field).map(ProducerField::try_from).transpose()?; if stored.is_some_and(|stored| epoch < stored.fence) { return Err(refuse( diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index 85d32e1f4..2277adf9b 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -3,8 +3,8 @@ use crate::{ Reason, StreamError, StreamResult, proto::{ - AppendRecord, Producer, StreamData, StreamRecord, StreamSuperseded, StreamWriter, - stream_record::Item, + AppendRecord, Producer, RunWithdrawal, StreamData, StreamRecord, StreamSuperseded, + StreamWriter, stream_record::Item, }, store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES}, }; @@ -65,11 +65,14 @@ pub(crate) fn data_records( /// The facts a `Superseded` item of this batch gives the Reader. pub(crate) fn superseded(producer: &Producer, withdraw_run: &str) -> Vec { + let withdrawal = (!withdraw_run.is_empty()).then(|| RunWithdrawal { + withdraw_run: withdraw_run.to_string(), + ..Default::default() + }); StreamSuperseded { writer: Some(StreamWriter { producer: Some(producer.clone()), - withdraw_run: withdraw_run.to_string(), - ..Default::default() + withdrawal, }), } .encode_to_vec() diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 7899b9992..b074cbe53 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -4,19 +4,23 @@ use crate::{ Reason, StreamError, StreamResult, digest::batch_digest, - epoch::{ProducerKind, task_of, withdraws}, + epoch::{ + ProducerKind, producer_run_id, task_of, withdraws, writer_failed_tasks, + writer_previous_completed_task, writer_withdraw_run, + }, ids::{check_producer, check_stream_id}, metrics::Metrics, proto::{ AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, - Producer, ReadRecordsRequest, ReadRecordsResponse, StreamIdConflictPolicy, StreamReadStart, - StreamRecord, StreamRef, StreamStatus, StreamSuperseded, stream_record::Item, + Producer, ReadRecordsRequest, ReadRecordsResponse, RunWithdrawal, StreamIdConflictPolicy, + StreamReadStart, StreamRecord, StreamRef, StreamStatus, StreamSuperseded, StreamWriter, + WorkflowProducer, WorkflowTaskEpoch, producer::Kind, stream_record::Item, }, record::{data_records, superseded}, store::{ Batch, Capabilities, ConflictPolicy, Create, CreateOutcome, FORMAT_VERSION, MAX_SEQUENCE, - MAX_TTL, MIN_RETENTION, Owner, Store, StreamState, Withdrawal, owner_chain, + MAX_TTL, MIN_RETENTION, Owner, Store, StreamState, TaskFacts, Withdrawal, owner_chain, }, }; use prost::Message; @@ -107,12 +111,41 @@ fn stream_id(stream_ref: Option<&StreamRef>) -> StreamResult<&str> { /// so a Reader withdraws only that run's Records. fn run_superseded(owner: Option<&Producer>, withdraw_run: &str) -> Vec { let producer = Producer { - epoch: None, + kind: None, ..owner.cloned().unwrap_or_default() }; superseded(&producer, withdraw_run) } +/// The `Superseded` facts that withdraw a run's orphaned Local Activity Records. Their writer is +/// the run's Workflow Producer in the batch's task, so that a Reader applies the Workflow rule. +fn orphan_superseded( + producer: &Producer, + completed: Option<&WorkflowTaskEpoch>, + failed_tasks: &[i64], +) -> Vec { + let run_id = producer_run_id(producer); + let task = task_of(producer.kind.as_ref()); + let workflow = Producer { + id: crate::ids::workflow_producer_id(run_id), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: run_id.to_string(), + epoch: Some(task), + })), + }; + StreamSuperseded { + writer: Some(StreamWriter { + producer: Some(workflow), + withdrawal: Some(RunWithdrawal { + previous_completed_task: completed.copied(), + failed_tasks: failed_tasks.to_vec(), + withdraw_run: String::new(), + }), + }), + } + .encode_to_vec() +} + /// Checks a Producer that writes outside Workflow code. fn outside_producer(producer: Option<&Producer>) -> StreamResult<&Producer> { let producer = producer.ok_or_else(|| StreamError::invalid("The request has no Producer"))?; @@ -124,9 +157,7 @@ fn outside_producer(producer: Option<&Producer>) -> StreamResult<&Producer> { )); } // Without its task, a re-dispatch in a later task has the same epoch as the first dispatch. - let task = producer - .epoch - .map_or(0, |epoch| task_of(&epoch).started_event_id); + let task = task_of(producer.kind.as_ref()).started_event_id; if kind == ProducerKind::LocalActivity && task <= 0 { return Err(StreamError::invalid( "A Local Activity Producer's epoch must name the Workflow Task that dispatched it", @@ -135,6 +166,33 @@ fn outside_producer(producer: Option<&Producer>) -> StreamResult<&Producer> { Ok(producer) } +/// Refuses task facts that a Store would drop or misread. Only a Local Activity has its run's task +/// facts, and a failed task is a started event id, which is never zero or negative. +fn check_task_facts(kind: ProducerKind, writer: &StreamWriter) -> StreamResult<()> { + let completed = writer_previous_completed_task(writer); + if completed.is_none() && writer_failed_tasks(writer).is_empty() { + return Ok(()); + } + if kind != ProducerKind::LocalActivity { + return Err(StreamError::invalid( + "Only a Local Activity sends its run's task facts", + )); + } + if completed.is_none() { + return Err(StreamError::invalid( + "Failed tasks need the run's last completed task", + )); + } + if completed.is_some_and(|c| c.started_event_id < 0) + || writer_failed_tasks(writer).iter().any(|&id| id <= 0) + { + return Err(StreamError::invalid( + "A task's started event id must be positive", + )); + } + Ok(()) +} + /// Checks the owner of a create, which must be a Workflow run with a ref of its run chain. A /// Store stores that chain and checks every later create and close against it. fn create_owner<'a>(owner: &'a Producer, stream_ref: &StreamRef) -> StreamResult<&'a Producer> { @@ -189,7 +247,7 @@ impl StreamService { "A create with withdraw_run needs the owner run", )); }; - if request.withdraw_run == owner.run_id { + if request.withdraw_run == producer_run_id(owner) { return Err(StreamError::invalid("A run doesn't withdraw itself")); } self.store.capabilities().require(Capabilities::RUN_FENCE)?; @@ -248,28 +306,40 @@ impl StreamService { .capabilities() .require(Capabilities::NEXT_SEQUENCE)?; } - if !writer.withdraw_run.is_empty() { + let withdraw_run = writer_withdraw_run(&writer); + if !withdraw_run.is_empty() { // Only a retry run's Local Activity withdraws its previous run, as only the owner's // create does. Any other Producer could withdraw and fence an unrelated run. - if kind != ProducerKind::LocalActivity || writer.withdraw_run == producer.run_id { + if kind != ProducerKind::LocalActivity || withdraw_run == producer_run_id(producer) { return Err(StreamError::invalid( "Only a Local Activity of a later run withdraws a run", )); } self.store.capabilities().require(Capabilities::RUN_FENCE)?; } + check_task_facts(kind, &writer)?; let digest = batch_digest(&request.records); let records = data_records(producer, request.records)?; - let bytes = superseded(producer, &writer.withdraw_run); + let bytes = superseded(producer, withdraw_run); + // A Local Activity sends its run's task facts, so that the batch withdraws the run's + // orphaned Local Activity Records. These are the Records of the tasks that never completed. + let completed = writer_previous_completed_task(&writer).filter(|c| c.started_event_id > 0); + let orphans = (kind == ProducerKind::LocalActivity + && writer_previous_completed_task(&writer).is_some()) + .then(|| orphan_superseded(producer, completed, writer_failed_tasks(&writer))); let batch = Batch { producer, sequence: request.sequence, records: &records, digest, withdrawal: Withdrawal { - withdraw_run: &writer.withdraw_run, + withdraw_run, superseded: &bytes, - tasks: None, + tasks: orphans.as_ref().map(|orphans| TaskFacts { + completed, + failed_tasks: writer_failed_tasks(&writer), + superseded: orphans, + }), }, }; let appended = self.store.append(&request.namespace, &stream, &batch).await; diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 825976ca4..7fb71ee0e 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -122,13 +122,18 @@ pub struct Withdrawal<'a> { /// The encoded `Superseded` facts, which the log keeps when the batch withdraws something. pub superseded: &'a [u8], /// The run's task facts of a Local Activity batch, which withdraw the run's orphaned Local - /// Activity Records. + /// Activity Records and fence their later writes. pub tasks: Option>, } /// The run's Workflow Task facts that a Local Activity batch gives. A task of the run after /// `completed` and before the batch's own task, or one in `failed_tasks`, never completed, so /// its Local Activity Records are orphans. +/// +/// A Store keeps the task of each Local Activity batch with Records, also of one with no facts, so +/// that a later batch's facts withdraw it. It also keeps the facts of the run's highest task and +/// fences a write whose task they show never completed, since a task can fail before its call +/// writes. #[derive(Debug, Clone, Copy, Default)] pub struct TaskFacts<'a> { /// The run's last completed Workflow Task, `None` before the first one. diff --git a/crates/streams/src/testing/fields.rs b/crates/streams/src/testing/fields.rs index 5d1620166..720d70543 100644 --- a/crates/streams/src/testing/fields.rs +++ b/crates/streams/src/testing/fields.rs @@ -1,7 +1,7 @@ -//! The fields of the `:meta` hash. The field names hold encoded ids only, so a `|` in a field name -//! is always a separator. +//! The fields of the `:meta` hash. The field names hold encoded ids, so a `|` in a field name is a +//! separator. Only an `ow|` name ends in a binary task key, after its last separator. -use crate::ids::encode_part; +use crate::{epoch::TASK_KEY_LEN, ids::encode_part}; /// The format version of the Stream. pub const VERSION: &str = "v"; @@ -31,6 +31,24 @@ pub fn withdrawn_run(run_id: &str) -> String { format!("wr|{}", encode_part(run_id)) } +/// The task keys of the run's Local Activity writes that no task facts showed completed yet, back +/// to back. +pub fn open_tasks(run_id: &str) -> String { + format!("oa|{}", encode_part(run_id)) +} + +/// Set when a `Superseded` item withdrew the run's Workflow Task with the key `task_key`. +pub fn withdrawn_task(run_id: &str, task_key: &[u8; TASK_KEY_LEN]) -> Vec { + [format!("ow|{}|", encode_part(run_id)).as_bytes(), task_key].concat() +} + +/// The task facts of the run's highest task, as task keys back to back. These are its own key, the +/// completed task's key or the zero key before the first, and the key of each failed task's +/// started event id. +pub fn task_facts(run_id: &str) -> String { + format!("tf|{}", encode_part(run_id)) +} + #[cfg(test)] mod tests { use super::*; @@ -45,5 +63,12 @@ mod tests { assert_eq!(producer(kind, producer_id), text("producer")); assert_eq!(run_wrote(run), text("run_wrote")); assert_eq!(withdrawn_run(run), text("withdrawn_run")); + assert_eq!(open_tasks(run), text("open_tasks")); + let task_key = crate::testing::support::hex(text("task_key")); + assert_eq!( + withdrawn_task(run, &task_key.as_slice().try_into().unwrap()), + [text("withdrawn_task_prefix").as_bytes(), &task_key].concat() + ); + assert_eq!(task_facts(run), text("task_facts")); } } diff --git a/crates/streams/src/testing/service.rs b/crates/streams/src/testing/service.rs index 9678705d6..7cc54d9fe 100644 --- a/crates/streams/src/testing/service.rs +++ b/crates/streams/src/testing/service.rs @@ -6,15 +6,16 @@ use super::{case, support::Needs}; use crate::{ StreamError, StreamRuntime, - epoch::withdraws, + epoch::{withdraws, writer_withdraw_run}, ids::{activity_producer_id, client_producer_id, named_client_producer_id}, proto::{ - ActivityEpoch, AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch, - CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, CreateStreamResponse, - DescribeStreamRequest, DescribeStreamResponse, LocalActivityEpoch, Producer, ProducerEpoch, - ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, StreamIdConflictPolicy, - StreamReadStart, StreamRecord, StreamRef, StreamStatus, StreamSuperseded, StreamWriter, - WorkflowTaskEpoch, producer_epoch::Kind, stream_record::Item, + ActivityEpoch, ActivityProducer, AppendRecord, AppendRecordsRequest, AppendRecordsResponse, + ClientEpoch, ClientProducer, CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, + CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, LocalActivityEpoch, + LocalActivityProducer, Producer, ReadRecordsRequest, ReadRecordsResponse, RunWithdrawal, + StreamErrorDetails, StreamIdConflictPolicy, StreamReadStart, StreamRecord, StreamRef, + StreamStatus, StreamSuperseded, StreamWriter, WorkflowProducer, WorkflowTaskEpoch, + producer::Kind, stream_record::Item, }, }; use prost::Message; @@ -88,10 +89,9 @@ fn writer_for(producer: Producer) -> Option { fn client(attempt: i32) -> Producer { Producer { id: client_producer_id("me", "p"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), } } @@ -108,16 +108,16 @@ fn named(name: &str, attempt: i32) -> Producer { fn activity(attempt: i32, scheduled: i64) -> Producer { Producer { id: activity_producer_id("r1", "a", "s"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch { + kind: Some(Kind::Activity(ActivityProducer { + run_id: "r1".to_string(), + epoch: Some(ActivityEpoch { attempt_scheduled_time: Some(Timestamp { seconds: scheduled, nanos: 0, }), attempt, - })), - }), - run_id: "r1".to_string(), + }), + })), } } @@ -125,10 +125,10 @@ fn activity(attempt: i32, scheduled: i64) -> Producer { fn workflow(run: &str) -> Producer { Producer { id: crate::ids::workflow_producer_id(run), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())), - }), - run_id: run.to_string(), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: run.to_string(), + epoch: Some(WorkflowTaskEpoch::default()), + })), } } @@ -388,13 +388,13 @@ async fn a_workflow_or_a_client_closes_a_stream_with_no_owner() { let run = format!("{}-run", case.stream); let activity = Producer { id: activity_producer_id(&run, "a", "s"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch { + kind: Some(Kind::Activity(ActivityProducer { + run_id: run.clone(), + epoch: Some(ActivityEpoch { attempt: 1, ..Default::default() - })), - }), - run_id: run.clone(), + }), + })), }; let error = case.close(activity).await.unwrap_err(); assert_eq!( @@ -432,13 +432,13 @@ async fn a_create_checks_its_owner_and_its_ref() { }; let local_activity = Producer { id: crate::ids::local_activity_producer_id(&run, "call-1"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: run.clone(), + epoch: Some(LocalActivityEpoch { task: Some(task), attempt: 1, - })), - }), - ..workflow.clone() + }), + })), }; let chain = |workflow_id: &str| StreamRef { id: case.stream.clone(), @@ -500,11 +500,12 @@ async fn a_create_checks_its_owner_and_its_ref() { async fn a_producer_must_match_its_kind_and_not_be_a_workflow() { for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); - let activity = ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch::default())), - }; - for epoch in [Some(activity), None, Some(ProducerEpoch::default())] { - let wrong = Producer { epoch, ..client(1) }; + let activity = Kind::Activity(ActivityProducer { + run_id: String::new(), + epoch: Some(ActivityEpoch::default()), + }); + for kind in [Some(activity), None] { + let wrong = Producer { kind, ..client(1) }; let error = case.append(wrong, 1, vec![record("x")]).await.unwrap_err(); assert_eq!( (error.code(), reason(&error).as_str()), @@ -515,13 +516,13 @@ async fn a_producer_must_match_its_kind_and_not_be_a_workflow() { assert_eq!(error.code(), Code::InvalidArgument); let no_task = Producer { id: crate::ids::local_activity_producer_id("r1", "call-1"), - run_id: "r1".to_string(), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: "r1".to_string(), + epoch: Some(LocalActivityEpoch { task: None, attempt: 1, - })), - }), + }), + })), }; let error = case .append(no_task, 1, vec![record("x")]) @@ -1014,6 +1015,11 @@ async fn a_create_answer_names_the_backend_and_the_store_never_reads_it() { } } +/// The Local Activity Producer of a Workflow call in run `run`. +fn local_activity(run: &str) -> Producer { + user_local_activity(run, 3, 100) +} + /// The case's ref with the owner of a made-up run chain, which a create with an owner needs. fn owned_ref(case: &Case) -> Option { Some(StreamRef { @@ -1023,6 +1029,126 @@ fn owned_ref(case: &Case) -> Option { }) } +/// A retry run's Local Activity calls withdraw the previous run. +async fn a_retry_run_s_local_activity_calls_withdraw_the_previous_run() { + for case in cases_for(Needs::UNCHECKED).await { + let (previous, retry) = (format!("{}-r1", case.stream), format!("{}-r2", case.stream)); + let create = |run: &str, withdraw_run: &str| CreateStreamRequest { + namespace: case.namespace.clone(), + withdraw_run: withdraw_run.to_string(), + owner: Some(workflow(run)), + stream_ref: owned_ref(&case), + ..use_existing() + }; + let append = |run: &str, withdraw_run: &str, text: &str| AppendRecordsRequest { + namespace: case.namespace.clone(), + r#ref: case.stream_ref(), + writer: Some(StreamWriter { + producer: Some(local_activity(run)), + withdrawal: Some(RunWithdrawal { + withdraw_run: withdraw_run.to_string(), + ..Default::default() + }), + }), + sequence: 1, + records: vec![record(text)], + ..Default::default() + }; + let _: CreateStreamResponse = call(&case.ext, "CreateStream", create(&previous, "")) + .await + .unwrap(); + let _: AppendRecordsResponse = + call(&case.ext, "AppendRecords", append(&previous, "", "old")) + .await + .unwrap(); + // The retry run's first create withdraws the previous run. + let _: CreateStreamResponse = call(&case.ext, "CreateStream", create(&retry, &previous)) + .await + .unwrap(); + let first: AppendRecordsResponse = + call(&case.ext, "AppendRecords", append(&retry, &previous, "new")) + .await + .unwrap(); + // A retried Local Activity sends the same batch and gets its first answer. + let again: AppendRecordsResponse = + call(&case.ext, "AppendRecords", append(&retry, &previous, "new")) + .await + .unwrap(); + assert_eq!( + (again.first_offset, again.last_offset), + (first.first_offset, first.last_offset) + ); + assert!(first.last_offset > 0); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["old", "", "new"]); + let superseded = superseded_of(&read.records[1]); + let writer = superseded.writer.as_ref().unwrap(); + assert_eq!(writer_withdraw_run(writer), previous); + // The item's writer has its Producer, which the WIT requires. A create's item has no kind, + // so a Reader can't read it as an epoch, and the run stays in the id. + let producer = writer.producer.as_ref().unwrap(); + assert_eq!(producer.id, crate::ids::workflow_producer_id(&retry)); + assert!(producer.kind.is_none()); + let error = call::<_, CreateStreamResponse>( + &case.ext, + "CreateStream", + CreateStreamRequest { + owner: None, + ..create(&retry, &previous) + }, + ) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + assert!(withdraws(superseded, &read.records[0])); + assert!(!withdraws(superseded, &read.records[2])); + } +} + +/// Only a Local Activity of a later run withdraws a run on append. A client, an Activity or the +/// run's own Local Activity that sends a run to withdraw fails with `stream_invalid` and withdraws +/// nothing. +async fn only_a_later_run_s_local_activity_withdraws_a_run_on_append() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-1", 3, 2, "workflow output"), + ) + .await + .unwrap(); + for producer in [client(1), activity(1, 10), local_activity(&run)] { + let error = call::<_, AppendRecordsResponse>( + &case.ext, + "AppendRecords", + AppendRecordsRequest { + namespace: case.namespace.clone(), + r#ref: case.stream_ref(), + writer: Some(StreamWriter { + producer: Some(producer), + withdrawal: Some(RunWithdrawal { + withdraw_run: run.clone(), + ..Default::default() + }), + }), + sequence: 1, + records: vec![record("other")], + ..Default::default() + }, + ) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["workflow output"]); + let next = local_activity_write(&case, &run, "call-2", 4, 3, "next"); + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", next).await.unwrap(); + } +} + /// The user Local Activity `call-1` of `run`, dispatched by the Workflow Task `[started]`. A rerun /// in a later task keeps the Producer id and has a newer dispatching task. fn user_local_activity(run: &str, started: i64, seconds: i64) -> Producer { @@ -1033,13 +1159,13 @@ fn user_local_activity(run: &str, started: i64, seconds: i64) -> Producer { }; Producer { id: crate::ids::local_activity_producer_id(run, "call-1"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: run.to_string(), + epoch: Some(LocalActivityEpoch { task: Some(task), attempt: 1, - })), - }), - run_id: run.to_string(), + }), + })), } } @@ -1105,6 +1231,553 @@ async fn a_rerun_local_activity_supersedes_its_first_attempt() { } } +/// A write of the Local Activity `call` of `run`, dispatched by task `[started]`, with the run's +/// last completed task `[completed]`, or 0 before the first. +fn local_activity_write( + case: &Case, + run: &str, + call: &str, + started: i64, + completed: i64, + text: &str, +) -> AppendRecordsRequest { + let mut producer = user_local_activity(run, started, 100 + started); + producer.id = crate::ids::local_activity_producer_id(run, call); + AppendRecordsRequest { + namespace: case.namespace.clone(), + r#ref: case.stream_ref(), + writer: Some(StreamWriter { + producer: Some(producer), + withdrawal: Some(RunWithdrawal { + previous_completed_task: Some(WorkflowTaskEpoch { + started_event_id: completed, + ..Default::default() + }), + ..Default::default() + }), + }), + sequence: 1, + records: vec![record(text)], + ..Default::default() + } +} + +/// The writer of a write, to change its Producer. +fn writer_mut(request: &mut AppendRecordsRequest) -> &mut StreamWriter { + request.writer.get_or_insert_with(Default::default) +} + +/// The withdrawal facts of a write, to change its task facts. +fn withdrawal_mut(request: &mut AppendRecordsRequest) -> &mut RunWithdrawal { + writer_mut(request) + .withdrawal + .get_or_insert_with(Default::default) +} + +/// The close Activity closes a Workflow's Stream as the run's Workflow Producer, which the service +/// refuses on an append. +async fn the_close_activity_closes_a_workflow_stream_as_its_workflow_producer() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + let chain = StreamRef { + id: case.stream.clone(), + owner: super::support::workflow_execution(&format!("{}-wf", case.stream), "first-run"), + ..Default::default() + }; + let _: CreateStreamResponse = call( + &case.ext, + "CreateStream", + CreateStreamRequest { + namespace: case.namespace.clone(), + owner: Some(workflow(&run)), + stream_ref: Some(chain.clone()), + ..Default::default() + }, + ) + .await + .unwrap(); + let close = CloseStreamRequest { + namespace: case.namespace.clone(), + r#ref: Some(chain), + writer: writer_for(workflow(&run)), + ..Default::default() + }; + let _: CloseStreamResponse = call(&case.ext, "CloseStream", close).await.unwrap(); + assert_eq!( + case.describe().await.unwrap().status, + StreamStatus::Closed as i32 + ); + } +} + +/// A Local Activity of a task that never completed is withdrawn by the run's next write. +async fn a_local_activity_of_a_task_that_never_completed_is_withdrawn_by_the_next_write() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + // Task [2] completed. Task [3] runs a Local Activity that writes, then task [3] fails. + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-1", 2, 0, "done"), + ) + .await + .unwrap(); + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-2", 3, 2, "orphan"), + ) + .await + .unwrap(); + // Task [4] takes another path and writes once. + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-3", 4, 2, "other"), + ) + .await + .unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["done", "orphan", "", "other"]); + let superseded = superseded_of(&read.records[2]); + let kept: Vec<_> = [0, 1, 3] + .iter() + .filter(|&&i| !withdraws(superseded, &read.records[i])) + .map(|&i| texts(&read)[i].clone()) + .collect(); + assert_eq!(kept, ["done", "other"]); + // A late write of the orphaned call is fenced. + let late = call::<_, AppendRecordsResponse>( + &case.ext, + "AppendRecords", + AppendRecordsRequest { + sequence: 2, + ..local_activity_write(&case, &run, "call-2", 3, 2, "late") + }, + ) + .await + .unwrap_err(); + assert_eq!(reason(&late), "stream_fenced"); + // A later write that knows task [4] completed withdraws nothing more. + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-4", 7, 4, "later"), + ) + .await + .unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!( + texts(&read), + ["done", "orphan", "", "other", "later"] + ); + } +} + +/// A failed task before the last completed one is withdrawn by its id. +async fn a_failed_task_before_the_last_completed_one_is_withdrawn_by_its_id() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-1", 3, 2, "orphan"), + ) + .await + .unwrap(); + // Task [3] failed and task [4] completed with no write here. Task [5] knows both. + let mut next = local_activity_write(&case, &run, "call-2", 5, 4, "next"); + withdrawal_mut(&mut next).failed_tasks = vec![3]; + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", next).await.unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["orphan", "", "next"]); + let superseded = superseded_of(&read.records[1]); + assert!(withdraws(superseded, &read.records[0])); + assert!(!withdraws(superseded, &read.records[2])); + } +} + +/// A call of a task that failed before the call wrote is fenced on its first write, by the task +/// facts of the run's highest task. A write of a lower task keeps those facts. +async fn a_first_write_of_a_task_that_never_completed_is_fenced() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-1", 2, 0, "done"), + ) + .await + .unwrap(); + // Task [3] runs call-2 and fails before call-2 writes. Task [4] knows that [3] failed. + let mut retry = local_activity_write(&case, &run, "call-3", 4, 2, "retry"); + withdrawal_mut(&mut retry).failed_tasks = vec![3]; + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", retry).await.unwrap(); + // Call-1 of the completed task [2] still runs and writes with its older facts. + let more = AppendRecordsRequest { + sequence: 2, + ..local_activity_write(&case, &run, "call-1", 2, 0, "more") + }; + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", more).await.unwrap(); + let late = call::<_, AppendRecordsResponse>( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-2", 3, 2, "late"), + ) + .await + .unwrap_err(); + assert_eq!(reason(&late), "stream_fenced"); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["done", "retry", "more"]); + } +} + +/// Task facts that a Store would drop or misread fail before any write. Only a Local Activity sends +/// them, failed tasks need the last completed task, and a started event id is positive. +async fn a_write_with_bad_task_facts_fails() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let facts = |failed: Vec| { + let mut request = local_activity_write(&case, &run, "call-1", 4, 2, "bad"); + withdrawal_mut(&mut request).failed_tasks = failed; + request + }; + let mut no_completed = facts(vec![3]); + withdrawal_mut(&mut no_completed).previous_completed_task = None; + let mut by_client = facts(vec![]); + writer_mut(&mut by_client).producer = Some(client(1)); + let mut by_activity = facts(vec![3]); + writer_mut(&mut by_activity).producer = Some(activity(1, 100)); + for bad in [ + facts(vec![0]), + facts(vec![-5]), + no_completed, + by_client, + by_activity, + ] { + let error = call::<_, AppendRecordsResponse>(&case.ext, "AppendRecords", bad) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert!(texts(&read).is_empty()); + } +} + +/// A Local Activity write with no task facts, as from an older Core, is withdrawn by a later write +/// whose facts show that its task never completed. +async fn a_write_with_no_task_facts_is_withdrawn_by_a_later_write() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let mut orphan = local_activity_write(&case, &run, "call-1", 3, 2, "orphan"); + withdrawal_mut(&mut orphan).previous_completed_task = None; + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", orphan).await.unwrap(); + let mut next = local_activity_write(&case, &run, "call-2", 5, 4, "next"); + withdrawal_mut(&mut next).failed_tasks = vec![3]; + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", next).await.unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["orphan", "", "next"]); + let superseded = superseded_of(&read.records[1]); + assert!(withdraws(superseded, &read.records[0])); + assert!(!withdraws(superseded, &read.records[2])); + } +} + +/// A rerun of a call in a later task withdraws the failed task's Records with two items. The +/// orphan item has an entry of its own, and the newer epoch's item comes with the first Record. +async fn a_batch_writes_an_orphan_item_and_an_epoch_item() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-1", 3, 2, "orphan"), + ) + .await + .unwrap(); + let mut rerun = local_activity_write(&case, &run, "call-1", 5, 4, "rerun"); + withdrawal_mut(&mut rerun).failed_tasks = vec![3]; + let answer: AppendRecordsResponse = call(&case.ext, "AppendRecords", rerun).await.unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!( + texts(&read), + ["orphan", "", "", "rerun"] + ); + let (orphans, epoch) = ( + superseded_of(&read.records[1]), + superseded_of(&read.records[2]), + ); + let kind = |item: &StreamSuperseded| -> Option { + Some( + item.writer + .as_ref()? + .producer + .as_ref()? + .kind + .as_ref()? + .clone(), + ) + }; + assert!(matches!(kind(orphans), Some(Kind::Workflow(_)))); + assert!(matches!(kind(epoch), Some(Kind::LocalActivity(_)))); + assert!(read.records[1].offset < answer.first_offset); + assert_eq!(read.records[2].offset, answer.first_offset); + assert_eq!(read.records[3].offset, answer.last_offset); + for item in [orphans, epoch] { + assert!(withdraws(item, &read.records[0])); + assert!(!withdraws(item, &read.records[3])); + } + } +} + +/// A resend of an empty batch that withdrew orphans gets the offsets of its orphan item again, and +/// adds no second item. +async fn a_resend_of_an_orphan_batch_withdraws_nothing_again() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let _: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + local_activity_write(&case, &run, "call-1", 3, 2, "orphan"), + ) + .await + .unwrap(); + // The writer of call-2 in task [4] opens with an empty batch. + let open = AppendRecordsRequest { + records: vec![], + ..local_activity_write(&case, &run, "call-2", 4, 2, "") + }; + let first: AppendRecordsResponse = call(&case.ext, "AppendRecords", open.clone()) + .await + .unwrap(); + let again: AppendRecordsResponse = call(&case.ext, "AppendRecords", open).await.unwrap(); + assert!(first.first_offset > 0); + assert_eq!( + (again.first_offset, again.last_offset), + (first.first_offset, first.last_offset) + ); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["orphan", ""]); + assert_eq!(read.records[1].offset, first.last_offset); + } +} + +/// The Workflow Task `[started_event_id]` in its attempt `attempt`, started at `seconds`. +fn task(started_event_id: i64, attempt: i32, seconds: i64) -> WorkflowTaskEpoch { + WorkflowTaskEpoch { + started_event_id, + attempt, + started_time: Some(Timestamp { seconds, nanos: 0 }), + } +} + +/// A write of the Local Activity `call` of `run`, dispatched by `task`, with the run's last +/// completed task, if the write sends it. +fn task_write( + case: &Case, + run: &str, + call: &str, + task: WorkflowTaskEpoch, + completed: Option, + text: &str, +) -> AppendRecordsRequest { + let producer = Producer { + id: crate::ids::local_activity_producer_id(run, call), + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: run.to_string(), + epoch: Some(LocalActivityEpoch { + task: Some(task), + attempt: 1, + }), + })), + }; + AppendRecordsRequest { + namespace: case.namespace.clone(), + r#ref: case.stream_ref(), + writer: Some(StreamWriter { + producer: Some(producer), + withdrawal: Some(RunWithdrawal { + previous_completed_task: completed, + ..Default::default() + }), + }), + sequence: 1, + records: vec![record(text)], + ..Default::default() + } +} + +/// A Local Activity of a transient attempt of a task that later completed is withdrawn by the +/// next write, which names the completed attempt. A Record of the completed attempt stays. +async fn a_local_activity_of_a_lower_attempt_of_the_completed_task_is_withdrawn() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let write = + |call: &str, + task: WorkflowTaskEpoch, + completed: Option, + text: &str| task_write(&case, &run, call, task, completed, text); + // Attempt 2 of task [9] writes, then fails with no event in History. + let c = Some(task(3, 1, 100)); + let orphan = write("call-1", task(9, 2, 200), c, "orphan"); + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", orphan).await.unwrap(); + // Attempt 3 completes. Its write sends no task facts, so it withdraws nothing itself. + let kept = write("call-2", task(9, 3, 300), None, "kept"); + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", kept).await.unwrap(); + let later = write("call-3", task(12, 1, 400), Some(task(9, 3, 300)), "later"); + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", later).await.unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["orphan", "kept", "", "later"]); + let superseded = superseded_of(&read.records[2]); + let kept: Vec<_> = [0, 1, 3] + .iter() + .filter(|&&i| !withdraws(superseded, &read.records[i])) + .map(|&i| texts(&read)[i].clone()) + .collect(); + assert_eq!(kept, ["kept", "later"]); + } +} + +/// A write sends the facts of its own task, attempt 3 of `[9]`. It withdraws the Records of +/// attempt 2 of `[9]`, which can't have completed, and keeps its own. +async fn a_write_withdraws_a_lower_attempt_of_its_own_task() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + case.create(None).await.unwrap(); + let completed = Some(task(3, 1, 100)); + let orphan = task_write(&case, &run, "call-1", task(9, 2, 200), completed, "orphan"); + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", orphan).await.unwrap(); + let own = task_write(&case, &run, "call-2", task(9, 3, 300), completed, "own"); + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", own).await.unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), ["orphan", "", "own"]); + let superseded = superseded_of(&read.records[1]); + assert!(withdraws(superseded, &read.records[0])); + assert!(!withdraws(superseded, &read.records[2])); + } +} + +/// A run that a retry run withdrew can't close the Stream, and the retry run can. +async fn a_withdrawn_run_cant_close_the_stream() { + for case in cases_for(Needs::UNCHECKED).await { + let (previous, retry) = (format!("{}-r1", case.stream), format!("{}-r2", case.stream)); + let create = |run: &str, withdraw_run: &str| CreateStreamRequest { + namespace: case.namespace.clone(), + withdraw_run: withdraw_run.to_string(), + owner: Some(workflow(run)), + stream_ref: owned_ref(&case), + ..use_existing() + }; + let close = |run: &str| CloseStreamRequest { + namespace: case.namespace.clone(), + r#ref: owned_ref(&case), + writer: writer_for(workflow(run)), + ..Default::default() + }; + let _: CreateStreamResponse = call(&case.ext, "CreateStream", create(&previous, "")) + .await + .unwrap(); + let _: CreateStreamResponse = call(&case.ext, "CreateStream", create(&retry, &previous)) + .await + .unwrap(); + let error = call::<_, CloseStreamResponse>(&case.ext, "CloseStream", close(&previous)) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_fenced"); + let _: CloseStreamResponse = call(&case.ext, "CloseStream", close(&retry)).await.unwrap(); + } +} + +/// A retry run's create of a new Stream fences the previous run, whose later write fails. +async fn a_retry_run_s_create_of_a_new_stream_fences_the_previous_run() { + for case in cases_for(Needs::UNCHECKED).await { + let (previous, retry) = (format!("{}-r1", case.stream), format!("{}-r2", case.stream)); + let created: CreateStreamResponse = call( + &case.ext, + "CreateStream", + CreateStreamRequest { + namespace: case.namespace.clone(), + withdraw_run: previous.clone(), + owner: Some(workflow(&retry)), + stream_ref: owned_ref(&case), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(created.created); + let append = |run: &str| AppendRecordsRequest { + namespace: case.namespace.clone(), + r#ref: case.stream_ref(), + writer: writer_for(local_activity(run)), + sequence: 1, + records: vec![record(run)], + ..Default::default() + }; + let error = call::<_, AppendRecordsResponse>(&case.ext, "AppendRecords", append(&previous)) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_fenced"); + let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", append(&retry)) + .await + .unwrap(); + let read = case.read(ReadRecordsRequest::default()).await.unwrap(); + assert_eq!(texts(&read), [retry]); + } +} + +/// A retry run's create after the log expired keeps the gap, so a read from an offset that the +/// log took fails as expired. +async fn a_retry_run_s_create_after_the_log_expired_keeps_the_gap() { + for case in cases_for(Needs::UNCHECKED.and(Needs::SHORT_TIMES)).await { + let (previous, retry) = (format!("{}-r1", case.stream), format!("{}-r2", case.stream)); + let create = |run: &str, withdraw_run: &str| CreateStreamRequest { + namespace: case.namespace.clone(), + retention: millis(300), + withdraw_run: withdraw_run.to_string(), + owner: Some(workflow(run)), + stream_ref: owned_ref(&case), + ..use_existing() + }; + let _: CreateStreamResponse = call(&case.ext, "CreateStream", create(&previous, "")) + .await + .unwrap(); + let first: AppendRecordsResponse = call( + &case.ext, + "AppendRecords", + AppendRecordsRequest { + namespace: case.namespace.clone(), + r#ref: case.stream_ref(), + writer: writer_for(local_activity(&previous)), + sequence: 1, + records: vec![record("a"), record("b")], + ..Default::default() + }, + ) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(400)).await; + let _: CreateStreamResponse = call(&case.ext, "CreateStream", create(&retry, &previous)) + .await + .unwrap(); + let behind = ReadRecordsRequest { + from_offset: first.first_offset + 1, + ..Default::default() + }; + let error = case.read(behind).await.unwrap_err(); + assert_eq!(reason(&error), "stream_expired"); + } +} + /// A read of a closed Stream gives `done` only with its last item. With no item left, a read /// answers at once with `done`, even with a long wait. async fn a_read_of_a_closed_stream_with_no_item_left_answers_at_once() { @@ -1280,7 +1953,22 @@ pub const CASES: &[super::Case] = &[ case!(a_close_after_the_log_expired_keeps_the_gap), case!(a_ref_with_an_owner_and_one_without_reach_the_same_stream), case!(a_create_answer_names_the_backend_and_the_store_never_reads_it), + case!(a_retry_run_s_local_activity_calls_withdraw_the_previous_run), + case!(only_a_later_run_s_local_activity_withdraws_a_run_on_append), case!(a_rerun_local_activity_supersedes_its_first_attempt), + case!(the_close_activity_closes_a_workflow_stream_as_its_workflow_producer), + case!(a_local_activity_of_a_task_that_never_completed_is_withdrawn_by_the_next_write), + case!(a_failed_task_before_the_last_completed_one_is_withdrawn_by_its_id), + case!(a_first_write_of_a_task_that_never_completed_is_fenced), + case!(a_write_with_bad_task_facts_fails), + case!(a_write_with_no_task_facts_is_withdrawn_by_a_later_write), + case!(a_batch_writes_an_orphan_item_and_an_epoch_item), + case!(a_resend_of_an_orphan_batch_withdraws_nothing_again), + case!(a_local_activity_of_a_lower_attempt_of_the_completed_task_is_withdrawn), + case!(a_write_withdraws_a_lower_attempt_of_its_own_task), + case!(a_withdrawn_run_cant_close_the_stream), + case!(a_retry_run_s_create_of_a_new_stream_fences_the_previous_run), + case!(a_retry_run_s_create_after_the_log_expired_keeps_the_gap), case!(a_read_of_a_closed_stream_with_no_item_left_answers_at_once), case!(a_waiting_read_wakes_with_done_on_a_close), case!(a_last_read_gives_the_newest_records_oldest_first_without_withdrawn_ones), diff --git a/crates/streams/src/testing/store.rs b/crates/streams/src/testing/store.rs index 29ade82d1..9221007d4 100644 --- a/crates/streams/src/testing/store.rs +++ b/crates/streams/src/testing/store.rs @@ -5,7 +5,8 @@ use super::{ case, fields, support::{ - Needs, Run, Target, digest, record, records_of, stores_for, superseded, workflow_execution, + Needs, Run, Target, digest, hex, record, records_of, stores_for, superseded, + workflow_execution, }, }; use crate::{ @@ -17,10 +18,10 @@ use crate::{ memory::{EntryId, ProducerField}, offset::{entry_id_of_offset, offset_of_entry_id}, proto::{ - ActivityEpoch, ClientEpoch, Producer, ProducerEpoch, StreamRef, WorkflowTaskEpoch, - producer_epoch::Kind, + ActivityEpoch, ActivityProducer, ClientEpoch, ClientProducer, Producer, StreamRef, + WorkflowProducer, WorkflowTaskEpoch, producer::Kind, }, - store::{Batch, ConflictPolicy, Create, CreateOutcome, Created, Owner, Withdrawal}, + store::{Batch, ConflictPolicy, Create, CreateOutcome, Created, Owner, TaskFacts, Withdrawal}, }; use futures_util::future::join_all; use prost_types::Timestamp; @@ -29,26 +30,25 @@ use std::time::Duration; fn activity(run: &str, attempt: i32, scheduled: i64) -> Producer { Producer { id: activity_producer_id(run, "a", "s"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch { + kind: Some(Kind::Activity(ActivityProducer { + run_id: run.to_string(), + epoch: Some(ActivityEpoch { attempt_scheduled_time: Some(Timestamp { seconds: scheduled, nanos: 0, }), attempt, - })), - }), - run_id: run.to_string(), + }), + })), } } fn client(attempt: i32) -> Producer { Producer { id: client_producer_id("me", "p1"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), } } @@ -362,6 +362,61 @@ async fn an_empty_batch_of_a_retry_withdraws_the_earlier_attempt() { } } +/// A retry run's first write withdraws the previous run and fences its Producers. +async fn a_retry_run_withdraws_and_fences_the_previous_run() { + for target in stores_for(Needs::UNCHECKED).await { + target.create(Owner::Workflow).await; + let (r1, r2) = (Run::new("r1"), Run::new("r2")); + target + .batch(&r1.local_activity("k", 3, 1, 1), 1, &["A"]) + .await + .unwrap(); + append(&target, &activity("r1", 1, 10), 1, &["A'"]) + .await + .unwrap(); + let b = target + .batch_retry(&r2.local_activity("k", 3, 1, 1), "r1", &["B"]) + .await + .unwrap(); + assert!(b.flagged); + assert_eq!(target.visible().await, ["B"]); + let late = append(&target, &activity("r1", 1, 10), 2, &["A''"]).await; + assert_eq!(late.unwrap_err().reason, Reason::Fenced); + let late = target + .batch(&r1.local_activity("j", 6, 1, 1), 1, &["A3"]) + .await; + assert_eq!(late.unwrap_err().reason, Reason::Fenced); + // A second withdrawal of the same run withdraws nothing more. + let again = target + .batch_retry(&r2.local_activity("j", 6, 1, 1), "r1", &["B2"]) + .await + .unwrap(); + assert!(!again.flagged); + } +} + +/// An empty batch that withdraws a run keeps its Producer's held epoch, whose Records it doesn't +/// withdraw. So the Producer's next epoch still withdraws them. +async fn an_empty_batch_that_withdraws_a_run_keeps_the_held_epoch() { + for target in stores_for(Needs::UNCHECKED).await { + target.create(Owner::Workflow).await; + let (r1, r2) = (Run::new("r1"), Run::new("r2")); + target + .batch(&r1.local_activity("k", 3, 1, 1), 1, &["A"]) + .await + .unwrap(); + let first = r2.local_activity("k", 3, 1, 1); + target.batch(&first, 1, &["B1"]).await.unwrap(); + target.write(&first, 2, "r1", &[]).await.unwrap(); + assert_eq!(target.visible().await, ["B1"]); + target + .batch(&r2.local_activity("k", 3, 1, 2), 1, &["B2"]) + .await + .unwrap(); + assert_eq!(target.visible().await, ["B2"]); + } +} + /// An open Workflow Stream's meta keeps the safety TTL, and its log keeps the Retention, after /// which a read hides every Record. Its close gives the meta the Retention too. async fn the_ttl_follows_the_owner_and_the_close() { @@ -718,10 +773,10 @@ async fn a_stale_producer_cant_close_the_stream() { fn workflow(run: &str) -> Producer { Producer { id: workflow_producer_id(run), - run_id: run.to_string(), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())), - }), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: run.to_string(), + epoch: Some(WorkflowTaskEpoch::default()), + })), } } @@ -744,13 +799,14 @@ async fn a_workflow_or_a_client_closes_a_stream_with_no_owner() { async fn a_producer_with_no_epoch_case_is_refused() { for target in stores_for(Needs::CALLS).await { target.create(Owner::Client).await; - for epoch in [None, Some(ProducerEpoch::default())] { - let producer = Producer { epoch, ..client(1) }; - let error = append(&target, &producer, 1, &["a"]).await.unwrap_err(); - assert_eq!(error.reason, Reason::Invalid); - let error = target.close(&producer).await.unwrap_err(); - assert_eq!(error.reason, Reason::Invalid); - } + let producer = Producer { + kind: None, + ..client(1) + }; + let error = append(&target, &producer, 1, &["a"]).await.unwrap_err(); + assert_eq!(error.reason, Reason::Invalid); + let error = target.close(&producer).await.unwrap_err(); + assert_eq!(error.reason, Reason::Invalid); assert!(target.texts().await.is_empty()); } } @@ -833,12 +889,77 @@ async fn every_store_keeps_the_meta_fields_of_the_shared_vectors() { assert_eq!(producer.id, vectors["producer_id"].as_str().unwrap()); let field = fields::producer(kind, &producer.id); assert_eq!(field, vectors["producer"].as_str().unwrap()); - for field in [field, fields::VERSION.to_string()] { + for field in [ + field, + vectors["run_wrote"].as_str().unwrap().to_string(), + fields::VERSION.to_string(), + ] { assert!(target.meta(&field).await.is_some(), "{field}"); } + // Task [3] fails, and a write of task [5] with the completed task [4] withdraws it. + let run = Run::new("run:1"); + task_batch(&target, &run.local_activity("k", 3, 1, 1), 2, &[], &["b"]) + .await + .unwrap(); + task_batch(&target, &run.local_activity("j", 5, 1, 1), 4, &[3], &["c"]) + .await + .unwrap(); + let text = |field: &str| vectors[field].as_str().unwrap(); + let withdrawn = [ + text("withdrawn_task_prefix").as_bytes(), + &hex(text("task_key")), + ] + .concat(); + assert!(target.meta(text("open_tasks")).await.is_some()); + assert!(target.meta(withdrawn).await.is_some()); + assert_eq!( + target.meta(text("task_facts")).await, + Some(hex(text("task_facts_value"))) + ); } } +/// A batch of a Local Activity with its run's task facts, the run's completed task +/// `[completed]` and its failed tasks. +async fn task_batch( + target: &Target, + producer: &Producer, + completed: i64, + failed_tasks: &[i64], + texts: &[&str], +) -> Result { + let records: Vec<_> = texts.iter().map(|text| record(producer, text)).collect(); + let completed = WorkflowTaskEpoch { + started_event_id: completed, + ..Default::default() + }; + // The Store keeps the `Superseded` facts as sent, so these do for both items. + let bytes = superseded(producer, &Withdrawal::default()); + let withdrawal = Withdrawal { + withdraw_run: "", + superseded: &bytes, + tasks: Some(TaskFacts { + completed: Some(&completed), + failed_tasks, + superseded: &bytes, + }), + }; + target + .store + .append( + &target.namespace, + &target.stream_ref(), + &Batch { + producer, + sequence: 1, + records: &records, + digest: digest(texts), + withdrawal, + }, + ) + .await +} + /// The entry ids of an append's offsets. fn entry_ids(offsets: Option<(i64, i64)>) -> Option<(EntryId, EntryId)> { let id = |offset: i64| { @@ -856,7 +977,7 @@ async fn a_producer_field_has_the_shared_layout() { target.create(Owner::Workflow).await; let kind = ProducerKind::Activity as i32; let field = |producer: &Producer, held: Option<&Producer>, offsets, texts: &[&str]| { - let epoch = |producer: &Producer| epoch_key(&producer.epoch.unwrap_or_default()); + let epoch = |producer: &Producer| epoch_key(producer.kind.as_ref()); ProducerField { fence: epoch(producer), held: held.map(epoch), @@ -920,6 +1041,8 @@ pub const CASES: &[super::Case] = &[ case!(an_expired_stream_refuses_writes_and_its_create), case!(an_activity_retry_flags_its_first_batch_only), case!(an_empty_batch_of_a_retry_withdraws_the_earlier_attempt), + case!(a_retry_run_withdraws_and_fences_the_previous_run), + case!(an_empty_batch_that_withdraws_a_run_keeps_the_held_epoch), case!(the_ttl_follows_the_owner_and_the_close), case!(an_append_trims_by_retention_and_keeps_the_watermark), case!(offsets_go_on_after_the_log_expires), diff --git a/crates/streams/src/testing/support.rs b/crates/streams/src/testing/support.rs index a018402b9..3e035a932 100644 --- a/crates/streams/src/testing/support.rs +++ b/crates/streams/src/testing/support.rs @@ -10,9 +10,9 @@ use crate::{ epoch::withdraws, ids::local_activity_producer_id, proto::{ - AppendRecord, LocalActivityEpoch, Producer, ProducerEpoch, StreamData, StreamRecord, - StreamRef, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer_epoch::Kind, - stream_record::Item, + AppendRecord, LocalActivityEpoch, LocalActivityProducer, Producer, RunWithdrawal, + StreamData, StreamRecord, StreamRef, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, + producer::Kind, stream_record::Item, }, store::{Appended, Batch, Closed, Create, Created, Entry, Owner, Store, Withdrawal}, }; @@ -132,13 +132,13 @@ impl Run { }; Producer { id: local_activity_producer_id(&self.id, call), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: self.id.clone(), + epoch: Some(LocalActivityEpoch { task: Some(task), attempt, - })), - }), - run_id: self.id.clone(), + }), + })), } } } @@ -168,6 +168,14 @@ pub(crate) fn record(producer: &Producer, text: &str) -> Vec { .encode_to_vec() } +/// The bytes of a hex text, as the vectors give binary values. +pub(crate) fn hex(text: &str) -> Vec { + (0..text.len()) + .step_by(2) + .map(|at| u8::from_str_radix(&text[at..at + 2], 16).unwrap()) + .collect() +} + /// The digest of the batch that [record] makes of each text. pub(crate) fn digest(texts: &[&str]) -> [u8; crate::digest::DIGEST_LEN] { let records: Vec<_> = texts @@ -185,11 +193,14 @@ pub(crate) fn digest(texts: &[&str]) -> [u8; crate::digest::DIGEST_LEN] { } pub(crate) fn superseded(producer: &Producer, withdrawal: &Withdrawal<'_>) -> Vec { + let run = (!withdrawal.withdraw_run.is_empty()).then(|| RunWithdrawal { + withdraw_run: withdrawal.withdraw_run.to_string(), + ..Default::default() + }); StreamSuperseded { writer: Some(StreamWriter { producer: Some(producer.clone()), - withdraw_run: withdrawal.withdraw_run.to_string(), - ..Default::default() + withdrawal: run, }), } .encode_to_vec() @@ -315,7 +326,18 @@ impl Target { self.write(producer, sequence, "", texts).await } - async fn write( + /// A retry run's first batch, which withdraws `previous`. + pub(crate) async fn batch_retry( + &self, + producer: &Producer, + previous: &str, + texts: &[&str], + ) -> StreamResult { + self.write(producer, 1, previous, texts).await + } + + /// A batch that withdraws `withdraw_run`, or no run when it is empty. + pub(crate) async fn write( &self, producer: &Producer, sequence: i64, diff --git a/crates/streams/testdata/fields.json b/crates/streams/testdata/fields.json index 8b9b269cd..036632040 100644 --- a/crates/streams/testdata/fields.json +++ b/crates/streams/testdata/fields.json @@ -26,5 +26,10 @@ "digest": "e3b0c44298fc1c149afbf4c8996fb924", "value": "000000000000000000000000000000000000000000000000000000000068e778020000000000000003000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000000000000000000000000000000e3b0c44298fc1c149afbf4c8996fb924" } - ] + ], + "task_key": "000000000000000300000001000000000000006400000000", + "open_tasks": "oa|run%3A1", + "withdrawn_task_prefix": "ow|run%3A1|", + "task_facts": "tf|run%3A1", + "task_facts_value": "000000000000000500000001000000000000006400000000000000000000000400000000000000000000000000000000000000000000000300000000000000000000000000000000" } diff --git a/crates/streams/testdata/gen_vectors.py b/crates/streams/testdata/gen_vectors.py index 97ce51084..ab49e64f7 100644 --- a/crates/streams/testdata/gen_vectors.py +++ b/crates/streams/testdata/gen_vectors.py @@ -174,6 +174,9 @@ def producer_values(): def fields(): run = "run:1" producer = "act:" + join(run, "a", "s") + # Task [3] fails, and a write of task [5] with the completed task [4] withdraws it. + withdrawn = task_key(3, 1, (100, 0)) + task, completed = task_key(5, 1, (100, 0)), task_key(4) return { "kind": 3, "producer_id": producer, @@ -182,6 +185,12 @@ def fields(): "run_wrote": f"run|{part(run)}", "withdrawn_run": f"wr|{part(run)}", "producer_values": producer_values(), + # The task keys are binary, so the vectors give them in hex after the field name's prefix. + "task_key": withdrawn.hex(), + "open_tasks": f"oa|{part(run)}", + "withdrawn_task_prefix": f"ow|{part(run)}|", + "task_facts": f"tf|{part(run)}", + "task_facts_value": (task + completed + task_key(3)).hex(), } diff --git a/crates/streams/tests/capabilities.rs b/crates/streams/tests/capabilities.rs index 2b312f1da..34212ccc9 100644 --- a/crates/streams/tests/capabilities.rs +++ b/crates/streams/tests/capabilities.rs @@ -12,10 +12,10 @@ use temporalio_streams::{ ids::client_producer_id, memory::MemoryStore, proto::{ - AppendRecord, AppendRecordsRequest, ClientEpoch, CloseStreamRequest, CreateStreamRequest, - DescribeStreamRequest, DescribeStreamResponse, Producer, ProducerEpoch, StreamErrorDetails, - StreamErrorReason, StreamRef, StreamStatus, StreamWriter, WorkflowTaskEpoch, - producer_epoch::Kind, + AppendRecord, AppendRecordsRequest, ClientEpoch, ClientProducer, CloseStreamRequest, + CreateStreamRequest, DescribeStreamRequest, DescribeStreamResponse, Producer, + StreamErrorDetails, StreamErrorReason, StreamRef, StreamStatus, StreamWriter, + WorkflowProducer, WorkflowTaskEpoch, producer::Kind, }, store::{Appended, Batch, Capabilities, Closed, Create, Created, Entry, Store, StreamState}, }; @@ -120,10 +120,9 @@ fn case(without: Capabilities) -> Case { fn client() -> Producer { Producer { id: client_producer_id("me", "p"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt: 1 })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: 1 }), + })), } } @@ -186,10 +185,10 @@ impl Case { withdraw_run: withdraw_run.to_string(), owner: Some(Producer { id: temporalio_streams::ids::workflow_producer_id(&self.run_id), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())), - }), - run_id: self.run_id.clone(), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: self.run_id.clone(), + epoch: Some(WorkflowTaskEpoch::default()), + })), }), stream_ref: Some(StreamRef { id: self.stream.clone(), diff --git a/crates/streams/tests/scans.rs b/crates/streams/tests/scans.rs index bcc5ea044..e2e5f54c4 100644 --- a/crates/streams/tests/scans.rs +++ b/crates/streams/tests/scans.rs @@ -14,10 +14,10 @@ use temporalio_streams::{ ids::client_producer_id, memory::MemoryStore, proto::{ - AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch, - CreateStreamRequest, CreateStreamResponse, Producer, ProducerEpoch, ReadRecordsRequest, + AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch, ClientProducer, + CreateStreamRequest, CreateStreamResponse, Producer, ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, StreamErrorReason, StreamRef, StreamWriter, - producer_epoch::Kind, + producer::Kind, }, store::{Appended, Batch, Capabilities, Create, Created, Entry, Store, StreamState}, }; @@ -129,10 +129,9 @@ async fn create(ext: &StreamRuntime) { fn client() -> Producer { Producer { id: client_producer_id("me", "p"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt: 1 })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: 1 }), + })), } } diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs index 853a4e96c..d824561a4 100644 --- a/crates/streams/tests/vectors.rs +++ b/crates/streams/tests/vectors.rs @@ -12,9 +12,10 @@ use temporalio_streams::{ ids::{self, StreamKeys}, offset::{entry_id_of_offset, offset_of_entry_id}, proto::{ - ActivityEpoch, AppendRecord, ClientEpoch, LocalActivityEpoch, Producer, ProducerEpoch, - StreamData, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, - producer_epoch::Kind, stream_record::Item, + ActivityEpoch, ActivityProducer, AppendRecord, ClientEpoch, ClientProducer, + LocalActivityEpoch, LocalActivityProducer, Producer, RunWithdrawal, StreamData, + StreamRecord, StreamSuperseded, StreamWriter, WorkflowProducer, WorkflowTaskEpoch, + producer::Kind, stream_record::Item, }, }; @@ -39,31 +40,43 @@ fn number(value: &Value, field: &str) -> i64 { value[field].as_i64().unwrap_or(0) } -/// The epoch of a kind, from the flat fields of the vectors. The case takes the fields of its kind. -fn epoch(kind: &str, value: &Value) -> ProducerEpoch { +/// The Producer kind of a vector, from its flat fields and its run. The case takes the fields of +/// its kind, and a run-based kind carries the run. `empty` has no case. +fn kind_of(kind: &str, value: &Value, run_id: &str) -> Option { let task = WorkflowTaskEpoch { started_event_id: number(value, "task_started_event_id"), attempt: number(value, "task_attempt") as i32, started_time: time(&value["task_started_time"]), }; let attempt = number(value, "attempt") as i32; - let activity = ActivityEpoch { - attempt_scheduled_time: time(&value["attempt_scheduled_time"]), - attempt, - }; - let kind = match kind { - "workflow" => Some(Kind::Workflow(task)), - "local_activity" => Some(Kind::LocalActivity(LocalActivityEpoch { - task: Some(task), + let run_id = run_id.to_string(); + let activity = ActivityProducer { + run_id: run_id.clone(), + epoch: Some(ActivityEpoch { + attempt_scheduled_time: time(&value["attempt_scheduled_time"]), attempt, + }), + }; + match kind { + "workflow" => Some(Kind::Workflow(WorkflowProducer { + run_id, + epoch: Some(task), + })), + "local_activity" => Some(Kind::LocalActivity(LocalActivityProducer { + run_id, + epoch: Some(LocalActivityEpoch { + task: Some(task), + attempt, + }), })), "activity" => Some(Kind::Activity(activity)), "standalone_activity" => Some(Kind::StandaloneActivity(activity)), - "client" => Some(Kind::Client(ClientEpoch { attempt })), + "client" => Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), "empty" => None, other => panic!("unknown kind {other}"), - }; - ProducerEpoch { kind } + } } #[test] @@ -137,10 +150,10 @@ fn offsets_match() { #[test] fn epoch_keys_match() { for case in load("epochs.json").as_array().unwrap() { - let epoch = epoch(text(case, "kind"), &case["epoch"]); - assert_eq!(epoch_key(&epoch).to_vec(), hex(text(case, "key"))); + let kind = kind_of(text(case, "kind"), &case["epoch"], ""); + assert_eq!(epoch_key(kind.as_ref()).to_vec(), hex(text(case, "key"))); assert_eq!( - task_key(&task_of(&epoch)).to_vec(), + task_key(&task_of(kind.as_ref())).to_vec(), hex(text(case, "task_key")) ); } @@ -152,29 +165,34 @@ fn the_appendix_e_withdrawals_match() { assert!(cases.as_array().unwrap().len() >= 35); for case in cases.as_array().unwrap() { let s = &case["superseded"]; + // A create's item has no epoch, and a Reader must not read it as an empty one. let withdrawing = Producer { id: text(s, "producer_id").to_string(), - // A create's item has no epoch, and a Reader must not read it as an empty one. - epoch: s - .get("epoch") - .map(|value| epoch(text(s, "producer_kind"), value)), - run_id: s["run_id"].as_str().unwrap_or_default().to_string(), + kind: s.get("epoch").and_then(|value| { + kind_of( + text(s, "producer_kind"), + value, + s["run_id"].as_str().unwrap_or_default(), + ) + }), }; let superseded = StreamSuperseded { writer: Some(StreamWriter { producer: Some(withdrawing), - previous_completed_task: s.get("previous_completed_task").map(|c| { - WorkflowTaskEpoch { - started_event_id: number(c, "started_event_id"), - attempt: number(c, "attempt") as i32, - started_time: time(&c["started_time"]), - } + withdrawal: Some(RunWithdrawal { + previous_completed_task: s.get("previous_completed_task").map(|c| { + WorkflowTaskEpoch { + started_event_id: number(c, "started_event_id"), + attempt: number(c, "attempt") as i32, + started_time: time(&c["started_time"]), + } + }), + failed_tasks: s["failed_tasks"] + .as_array() + .map(|list| list.iter().map(|v| v.as_i64().unwrap()).collect()) + .unwrap_or_default(), + withdraw_run: s["withdraw_run"].as_str().unwrap_or_default().to_string(), }), - failed_tasks: s["failed_tasks"] - .as_array() - .map(|list| list.iter().map(|v| v.as_i64().unwrap()).collect()) - .unwrap_or_default(), - withdraw_run: s["withdraw_run"].as_str().unwrap_or_default().to_string(), }), }; let p = &case["record"]["producer"]; @@ -182,8 +200,13 @@ fn the_appendix_e_withdrawals_match() { item: Some(Item::Data(StreamData { producer: Some(Producer { id: text(p, "id").to_string(), - epoch: p.get("epoch").map(|value| epoch(text(p, "kind"), value)), - run_id: p["run_id"].as_str().unwrap_or_default().to_string(), + kind: p.get("epoch").and_then(|value| { + kind_of( + text(p, "kind"), + value, + p["run_id"].as_str().unwrap_or_default(), + ) + }), }), ..Default::default() })),