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 0c9b198b7..050ac393a 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -20,9 +20,10 @@ import "temporal/api/stream/v1/message.proto"; 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/Cargo.toml b/crates/streams/Cargo.toml index a43408885..089c207c7 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -11,9 +11,14 @@ repository = "https://github.com/temporalio/sdk-rust" keywords = ["temporal", "workflow", "streams"] categories = ["development-tools"] +[features] +# The test helpers, for the crates of other Stores. +test-support = [] + [dependencies] prost = { workspace = true } prost-types = { workspace = true } +sha2 = "0.10" thiserror = { workspace = true } tonic = { workspace = true } @@ -22,5 +27,8 @@ path = "../common" version = "~1.1.0" default-features = false +[dev-dependencies] +serde_json = { workspace = true } + [lints] workspace = true diff --git a/crates/streams/README.md b/crates/streams/README.md index 562816beb..c53d50b60 100644 --- a/crates/streams/README.md +++ b/crates/streams/README.md @@ -4,7 +4,7 @@ Experimental. The Stream runtime that the Temporal SDKs link into their bridges. the runtime per client, as a plain object, and calls it with a method name and proto bytes. A Workflow makes each Stream call from a Local Activity. Core has no Stream code. -## Protos +## Protos and shared vectors The crate defines no protos. `temporalio-protos` generates them, and `temporalio_streams::proto` re-exports them, so a value needs no conversion between the crates. @@ -14,3 +14,8 @@ re-exports them, so a value needs no conversion between the crates. `crates/protos/protos/api_upstream`. - `crates/protos/protos/local/temporal/sdk/core/streams/streams.proto`, in the package `coresdk.streams`, has the crate's own calls, such as the writer calls. +- `testdata/*.json` are the shared vectors. They cover Stream ids, Producer ids, Store keys and + meta fields, offsets, epoch keys, the withdrawal rule and the batch digest of a dedupe entry. + `testdata/gen_vectors.py` writes them from a second implementation, and the crate's tests read + each file. The Python SDK also reads `ids.json` and `withdrawals.json`. `digests.json` is the + golden of the crate's batch digest. diff --git a/crates/streams/src/digest.rs b/crates/streams/src/digest.rs new file mode 100644 index 000000000..ef448b357 --- /dev/null +++ b/crates/streams/src/digest.rs @@ -0,0 +1,44 @@ +//! The content digest of a batch. Each dedupe entry stores it, so that a resend with the same +//! `sequence` and other content fails instead of being dropped without an error. The Store compares +//! the digest that this function gives, and `testdata/digests.json` has its expected values. + +use crate::proto::AppendRecord; +use sha2::{Digest, Sha256}; + +/// The length of a [batch_digest]. +pub const DIGEST_LEN: usize = 16; + +/// The first 16 bytes of a SHA-256 over the Records in order. Each part has its length before it, +/// so that two different batches never hash the same bytes. The metadata is hashed in key order, +/// since a map has no order on the wire. +pub fn batch_digest(records: &[AppendRecord]) -> [u8; DIGEST_LEN] { + let mut hasher = Sha256::new(); + for record in records { + // A Record with no body and one with an empty body are two different stored Records. + let Some(body) = &record.body else { + hasher.update([0]); + continue; + }; + hasher.update([1]); + let mut metadata: Vec<_> = body.metadata.iter().collect(); + metadata.sort(); + hasher.update((metadata.len() as u64).to_be_bytes()); + for (key, value) in metadata { + part(&mut hasher, key.as_bytes()); + part(&mut hasher, value); + } + part(&mut hasher, &body.data); + hasher.update((body.external_payloads.len() as u64).to_be_bytes()); + for external in &body.external_payloads { + hasher.update(external.size_bytes.to_be_bytes()); + } + } + let mut digest = [0; DIGEST_LEN]; + digest.copy_from_slice(&hasher.finalize()[..DIGEST_LEN]); + digest +} + +fn part(hasher: &mut Sha256, bytes: &[u8]) { + hasher.update((bytes.len() as u64).to_be_bytes()); + hasher.update(bytes); +} diff --git a/crates/streams/src/epoch.rs b/crates/streams/src/epoch.rs new file mode 100644 index 000000000..15b1ca6a2 --- /dev/null +++ b/crates/streams/src/epoch.rs @@ -0,0 +1,250 @@ +//! Producer epochs, their order, and the withdrawal rule of a `Superseded` item. +//! +//! A Store keeps an epoch as a key of fixed-width big-endian fields, in the order of comparison. +//! A byte comparison of two keys then orders the epochs field by field. `testdata/epochs.json` has +//! the shared vectors of the keys, in hex. + +use crate::{ + StreamError, StreamResult, + proto::{ + Producer, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer::Kind, + stream_record::Item, + }, +}; +use prost_types::Timestamp; +use std::cmp::Ordering; + +/// The kind of a Producer, which is the case of its epoch. The stored field names and keys hold +/// its number, so each kind keeps its number. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ProducerKind { + /// A Workflow run. + Workflow = 1, + /// A Local Activity. + LocalActivity = 2, + /// An Activity of a Workflow. + Activity = 3, + /// A client. + Client = 4, + /// A Standalone Activity. + StandaloneActivity = 5, +} + +impl From<&Kind> for ProducerKind { + fn from(kind: &Kind) -> Self { + match kind { + Kind::Workflow(_) => Self::Workflow, + Kind::LocalActivity(_) => Self::LocalActivity, + Kind::Activity(_) => Self::Activity, + Kind::StandaloneActivity(_) => Self::StandaloneActivity, + Kind::Client(_) => Self::Client, + } + } +} + +/// Refuses a Producer with no kind case. +impl TryFrom<&Producer> for ProducerKind { + type Error = StreamError; + + fn try_from(producer: &Producer) -> StreamResult { + kind_of(producer).ok_or_else(|| StreamError::invalid("The Producer has no kind")) + } +} + +/// 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]. +pub const TASK_KEY_LEN: usize = 24; +/// The length of an [epoch_key]. +pub const EPOCH_KEY_LEN: usize = 40; + +// A negative field is clamped at zero, since its bytes would outrank every real epoch. +fn be64(value: i64) -> [u8; 8] { + value.max(0).to_be_bytes() +} + +fn be32(value: i32) -> [u8; 4] { + value.max(0).to_be_bytes() +} + +fn time_fields(time: Option<&Timestamp>) -> (i64, i32) { + time.map_or((0, 0), |time| (time.seconds, time.nanos)) +} + +/// The key of a Workflow Task attempt. It is the started event id, the attempt, and the started +/// seconds and nanos, in 8, 4, 8 and 4 bytes. The epoch keys of the run and of the task's Local +/// Activities start with it. +pub fn task_key(task: &WorkflowTaskEpoch) -> [u8; TASK_KEY_LEN] { + let (seconds, nanos) = time_fields(task.started_time.as_ref()); + let mut key = [0; TASK_KEY_LEN]; + key[..8].copy_from_slice(&be64(task.started_event_id)); + key[8..12].copy_from_slice(&be32(task.attempt)); + key[12..20].copy_from_slice(&be64(seconds)); + key[20..].copy_from_slice(&be32(nanos)); + key +} + +/// 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 a Producer +/// with no case is all zeros. +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)) => { + 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.epoch.map_or(0, |e| e.attempt)), + Some(Kind::Workflow(_)) | None => (None, 0), + }; + 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(kind))); + key[24..32].copy_from_slice(&be64(seconds)); + key[32..36].copy_from_slice(&be32(nanos)); + key[36..].copy_from_slice(&be32(attempt)); + key +} + +/// 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(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: Option<&Kind>, b: Option<&Kind>) -> Ordering { + epoch_key(a).cmp(&epoch_key(b)) +} + +fn compare_tasks(a: &WorkflowTaskEpoch, b: &WorkflowTaskEpoch) -> Ordering { + task_key(a).cmp(&task_key(b)) +} + +/// Whether a `Superseded` item withdraws an earlier Record. The case of the epoch of the item's +/// writer picks the rule. +/// +/// For an Activity or a Local Activity, it withdraws that Producer's Records with a lower epoch. +/// For a Workflow task t with the previous completed task c, it withdraws the Records of the run +/// and of its Local Activities whose task is in `failed_tasks` or strictly between c and t, and +/// those at c or at t with a lower epoch. The item of a retry run withdraws every Record of the +/// previous run. A client's newer epoch fences the earlier ones but withdraws nothing, since a +/// restarted client's earlier Records are real messages, not a failed attempt's. +pub fn withdraws(superseded: &StreamSuperseded, record: &StreamRecord) -> bool { + let Some(writer) = superseded.writer.as_ref() else { + return false; + }; + let Some(Item::Data(data)) = &record.item else { + return false; + }; + let Some(producer) = data.producer.as_ref() else { + return false; + }; + 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 (new, old) = (withdrawing.kind.as_ref(), producer.kind.as_ref()); + let (Some(new_kind), Some(old_kind)) = (new, old) else { + return false; + }; + match ProducerKind::from(new_kind) { + ProducerKind::Client => return false, + ProducerKind::Workflow => {} + same => { + return ProducerKind::from(old_kind) == same + && producer.id == withdrawing.id + && compare_epochs(old, new) == Ordering::Less; + } + } + if !of_run(producer, withdrawing) { + return false; + } + 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(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_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 kind_of(producer) { + Some(ProducerKind::Workflow) => producer.id == withdrawing.id, + Some(ProducerKind::LocalActivity) => { + producer_run_id(producer) == producer_run_id(withdrawing) + } + _ => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + 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 = 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 new file mode 100644 index 000000000..45eb1e77b --- /dev/null +++ b/crates/streams/src/ids.rs @@ -0,0 +1,311 @@ +//! Stream ids, Producer ids and the Redis key layout. Every SDK must build them the same way, and +//! `testdata/ids.json` has the shared vectors that check it. +//! +//! ```text +//! :{:}:meta the create key, the fence, dedupe and close state +//! :{:}:log the log, one entry per Record +//! :{:}:producers each Producer's last append, to prune its fields +//! ``` +//! +//! The part in braces is the Redis Cluster hash tag, so that the keys are in one slot for one +//! 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, producer_run_id}, + proto::Producer, +}; + +/// The longest Stream id, in bytes. +pub const MAX_STREAM_ID_BYTES: usize = 4096; + +/// The key prefix when the backend sets none. +pub const DEFAULT_KEY_PREFIX: &str = "temporal-streams"; + +/// Percent-encodes every byte but the unreserved ones, as Python's `quote(text, safe="")` does. +pub fn encode_part(text: &str) -> String { + let mut out = String::with_capacity(text.len()); + for byte in text.bytes() { + if byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'.' | b'-' | b'~') { + out.push(byte as char); + } else { + out.push_str(&format!("%{byte:02X}")); + } + } + out +} + +/// Decodes the text of [encode_part]. Returns `None` for text that [encode_part] can't make. +pub fn decode_part(text: &str) -> Option { + let mut bytes = Vec::with_capacity(text.len()); + let mut rest = text.as_bytes(); + while let Some((&byte, tail)) = rest.split_first() { + if byte == b'%' { + let hex = std::str::from_utf8(tail.get(..2)?).ok()?; + bytes.push(u8::from_str_radix(hex, 16).ok()?); + rest = &tail[2..]; + } else { + bytes.push(byte); + rest = tail; + } + } + String::from_utf8(bytes).ok() +} + +fn join(parts: &[&str]) -> String { + parts + .iter() + .map(|part| encode_part(part)) + .collect::>() + .join(":") +} + +/// The id of a Stream that a Workflow creates with a name, `::`. +pub fn scoped_stream_id(workflow_id: &str, first_run_id: &str, name: &str) -> String { + join(&[workflow_id, first_run_id, name]) +} + +/// Refuses an empty or too long Stream id. +pub fn check_stream_id(id: &str) -> StreamResult<()> { + if id.is_empty() { + return Err(StreamError::invalid("The Stream id is empty")); + } + if id.len() > MAX_STREAM_ID_BYTES { + return Err(StreamError::invalid(format!( + "The Stream id has {} bytes, more than {MAX_STREAM_ID_BYTES}", + id.len() + ))); + } + Ok(()) +} + +/// The Producer id of a Workflow run, `wf:`. +pub fn workflow_producer_id(run_id: &str) -> String { + format!("wf:{}", join(&[run_id])) +} + +/// The Producer id of a Local Activity call, `la::`. +pub fn local_activity_producer_id(run_id: &str, call_key: &str) -> String { + format!("la:{}", join(&[run_id, call_key])) +} + +/// The Producer id of an Activity, `act:::`. +pub fn activity_producer_id(run_id: &str, activity_id: &str, schedule_key: &str) -> String { + format!("act:{}", join(&[run_id, activity_id, schedule_key])) +} + +/// The Producer id of a Standalone Activity, `sa::`. +pub fn standalone_activity_producer_id(activity_id: &str, activity_run_id: &str) -> String { + format!("sa:{}", join(&[activity_id, activity_run_id])) +} + +/// The Producer id of a client process, `client::`. +pub fn client_producer_id(identity: &str, process_key: &str) -> String { + format!("client:{}", join(&[identity, process_key])) +} + +/// The Producer id that a client names itself, `client:`. The caller keeps the name across +/// restarts and raises the epoch on each one, so that a new incarnation fences the earlier one. +pub fn named_client_producer_id(name: &str) -> String { + format!("client:{}", join(&[name])) +} + +/// The id prefix of a Producer kind. +pub(crate) fn kind_prefix(kind: ProducerKind) -> &'static str { + match kind { + ProducerKind::Workflow => "wf:", + ProducerKind::LocalActivity => "la:", + ProducerKind::Activity => "act:", + ProducerKind::StandaloneActivity => "sa:", + ProducerKind::Client => "client:", + } +} + +/// Checks that a Producer's epoch has a case, that its id has the prefix of that kind, and that the +/// run in the id is its `run_id`. The kind prefix stops a Producer from using the id of another +/// kind. +pub fn check_producer(producer: &Producer) -> StreamResult<()> { + let kind = ProducerKind::try_from(producer)?; + let prefix = kind_prefix(kind); + let Some(rest) = producer.id.strip_prefix(prefix) else { + return Err(StreamError::invalid(format!( + "The Producer id {:?} doesn't start with {prefix:?}, as its kind requires", + producer.id + ))); + }; + let parts: Vec<_> = rest.split(':').collect(); + let expected = match kind { + ProducerKind::Workflow => 1, + ProducerKind::LocalActivity | ProducerKind::StandaloneActivity => 2, + // A named client id has one part. A process client id has two. + ProducerKind::Client if parts.len() == 1 => 1, + ProducerKind::Client => 2, + ProducerKind::Activity => 3, + }; + // The meta field names use `|` as a separator, so a part must be in the form of `encode_part`. + let encoded = |part: &&str| decode_part(part).is_some_and(|text| encode_part(&text) == *part); + if parts.len() != expected || !parts.iter().all(encoded) { + return Err(StreamError::invalid(format!( + "The Producer id {:?} must have {expected} encoded parts after {prefix:?}", + producer.id + ))); + } + let runs = matches!( + kind, + ProducerKind::Workflow | ProducerKind::LocalActivity | ProducerKind::Activity + ); + 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) + ))); + } + if kind == ProducerKind::Client && !producer_run_id(producer).is_empty() { + return Err(StreamError::invalid("A client Producer has no run id")); + } + Ok(()) +} + +/// The keys of one Stream. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StreamKeys { + base: String, +} + +impl StreamKeys { + /// The keys of `stream_id` in `namespace`, under `prefix`. + pub fn new(prefix: &str, namespace: &str, stream_id: &str) -> Self { + Self { + base: format!( + "{}:{{{}}}", + encode_part(prefix), + join(&[namespace, stream_id]) + ), + } + } + + /// The hash with the format version, the fence, dedupe and close state. + pub fn meta(&self) -> String { + format!("{}:meta", self.base) + } + + /// The log. + pub fn log(&self) -> String { + format!("{}:log", self.base) + } + + /// The sorted set of each Producer's last append time. An append finds in it the Producers + /// whose last append is older than the Retention. + pub fn producers(&self) -> String { + format!("{}:producers", self.base) + } + + /// The key that a close adds one entry to. A close writes no log entry, so a waiting read also + /// waits on this key, and the close wakes it. + pub fn closed(&self) -> String { + format!("{}:closed", self.base) + } +} + +#[cfg(test)] +mod tests { + use super::*; + 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(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(), + kind: Some(kind), + } + } + + #[test] + fn a_part_round_trips_and_bad_text_does_not_decode() { + for text in ["plain", "a:b", "é {x}%", ""] { + assert_eq!(decode_part(&encode_part(text)).as_deref(), Some(text)); + } + assert_eq!(decode_part("%zz"), None); + assert_eq!(decode_part("%4"), None); + } + + #[test] + fn the_kind_prefix_and_the_run_must_match_the_producer() { + let run = "run:1"; + let ok = [ + producer(ProducerKind::Workflow, &workflow_producer_id(run), run), + producer( + ProducerKind::LocalActivity, + &local_activity_producer_id(run, "k"), + run, + ), + producer( + ProducerKind::Activity, + &activity_producer_id(run, "a:1", "s"), + run, + ), + producer( + ProducerKind::StandaloneActivity, + &standalone_activity_producer_id("a", "r"), + "", + ), + producer( + ProducerKind::Client, + &client_producer_id("me@host", "p"), + "", + ), + ]; + for producer in &ok { + check_producer(producer).unwrap(); + } + let bad = [ + producer(ProducerKind::Client, &workflow_producer_id(run), run), + producer(ProducerKind::Workflow, &workflow_producer_id(run), "other"), + producer(ProducerKind::Activity, "act:run%3A1:a", run), + Producer { + kind: None, + ..producer(ProducerKind::Workflow, "wf:x", "x") + }, + producer(ProducerKind::Client, "client:a|b", ""), + producer(ProducerKind::Client, "client:run%3a1", ""), + ]; + for producer in &bad { + assert_eq!( + check_producer(producer).unwrap_err().reason, + crate::Reason::Invalid, + "{producer:?}" + ); + } + } + + #[test] + fn a_stream_id_is_bounded() { + check_stream_id("a").unwrap(); + check_stream_id(&"a".repeat(MAX_STREAM_ID_BYTES)).unwrap(); + assert!(check_stream_id("").is_err()); + assert!(check_stream_id(&"a".repeat(MAX_STREAM_ID_BYTES + 1)).is_err()); + } +} diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index a211e817e..fcc86adbc 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -6,7 +6,13 @@ //! client and calls it with an operation name and proto bytes. Core has no hook for it. A Workflow //! makes each call from a Local Activity. +pub mod digest; +pub mod epoch; mod error; +pub mod ids; +pub mod offset; +#[cfg(any(test, feature = "test-support"))] +pub mod testing; pub use error::{Reason, StreamError, StreamResult}; diff --git a/crates/streams/src/offset.rs b/crates/streams/src/offset.rs new file mode 100644 index 000000000..607d7bfb9 --- /dev/null +++ b/crates/streams/src/offset.rs @@ -0,0 +1,48 @@ +//! Offsets. An offset is the Redis entry id `-` packed as `ms << 20 | seq` in an `i64`. +//! It is ordered and unique in one Stream, and sparse. Zero is never an offset. + +use crate::{StreamError, StreamResult}; + +/// The number of bits of the entry id's sequence part. +pub const SEQ_BITS: u32 = 20; + +const SEQ_MASK: i64 = (1 << SEQ_BITS) - 1; +/// The largest millisecond part that fits, which is past the year 2200. +const MAX_MS: u64 = (i64::MAX >> SEQ_BITS) as u64; + +/// Packs a Redis entry id. +pub fn offset_of_entry_id(id: &str) -> StreamResult { + let bad = || StreamError::storage_corrupt(format!("The Store gave the entry id {id:?}")); + let (ms, seq) = id.split_once('-').ok_or_else(bad)?; + let ms: u64 = ms.parse().map_err(|_| bad())?; + let seq: u64 = seq.parse().map_err(|_| bad())?; + if ms > MAX_MS || seq > SEQ_MASK as u64 { + return Err(StreamError::storage_corrupt(format!( + "The entry id {id:?} doesn't fit in an offset" + ))); + } + Ok(((ms as i64) << SEQ_BITS) | seq as i64) +} + +/// Unpacks an offset into its Redis entry id. The offset must be above zero. +pub fn entry_id_of_offset(offset: i64) -> String { + debug_assert!(offset > 0); + format!("{}-{}", offset >> SEQ_BITS, offset & SEQ_MASK) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_entry_id_that_doesnt_fit_is_refused() { + for id in ["1-1048576", &format!("{}-0", MAX_MS + 1), "x"] { + let error = offset_of_entry_id(id).unwrap_err(); + assert_eq!(error.reason, crate::Reason::StorageCorrupt, "{id}"); + } + assert_eq!( + offset_of_entry_id(&format!("{MAX_MS}-{SEQ_MASK}")).unwrap(), + i64::MAX + ); + } +} diff --git a/crates/streams/src/testing/fields.rs b/crates/streams/src/testing/fields.rs new file mode 100644 index 000000000..f05c80d6d --- /dev/null +++ b/crates/streams/src/testing/fields.rs @@ -0,0 +1,53 @@ +//! The fields of the `:meta` hash. The field names hold encoded ids only, so a `|` in a field name +//! is always a separator. + +use crate::ids::encode_part; + +/// The format version of the Stream. +pub const VERSION: &str = "v"; + +/// The newest epoch key of a Producer, the fence. +pub fn fence(kind: i32, producer_id: &str) -> String { + format!("e|{kind}|{producer_id}") +} + +/// The last batch of a Producer's newest epoch, as +/// `||||`. These are the dedupe key, the first and last +/// entry ids of the batch, and the content digest that a resend must match. +pub fn dedupe(kind: i32, producer_id: &str) -> String { + format!("d|{kind}|{producer_id}") +} + +/// The newest epoch key of a Producer that still has Records in the Stream. +pub fn holds(kind: i32, producer_id: &str) -> String { + format!("h|{kind}|{producer_id}") +} + +/// Set when a Producer of the run wrote a Record to the Stream. +pub fn run_wrote(run_id: &str) -> String { + format!("run|{}", encode_part(run_id)) +} + +/// Set when a Workflow retry withdrew the run. +pub fn withdrawn_run(run_id: &str) -> String { + format!("wr|{}", encode_part(run_id)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_fields_match_the_shared_vectors() { + let case: serde_json::Value = + serde_json::from_str(include_str!("../../testdata/fields.json")).unwrap(); + let text = |field: &str| case[field].as_str().unwrap(); + let kind = case["kind"].as_i64().unwrap() as i32; + let (producer, run) = (text("producer_id"), text("run_id")); + assert_eq!(fence(kind, producer), text("fence")); + assert_eq!(dedupe(kind, producer), text("dedupe")); + assert_eq!(holds(kind, producer), text("holds")); + assert_eq!(run_wrote(run), text("run_wrote")); + assert_eq!(withdrawn_run(run), text("withdrawn_run")); + } +} diff --git a/crates/streams/src/testing/mod.rs b/crates/streams/src/testing/mod.rs new file mode 100644 index 000000000..f8feafa3b --- /dev/null +++ b/crates/streams/src/testing/mod.rs @@ -0,0 +1,3 @@ +//! Test helpers that the crates of other Stores share. + +pub mod fields; diff --git a/crates/streams/testdata/digests.json b/crates/streams/testdata/digests.json new file mode 100644 index 000000000..d870b506a --- /dev/null +++ b/crates/streams/testdata/digests.json @@ -0,0 +1,92 @@ +[ + { + "name": "empty batch", + "records": [], + "digest": "e3b0c44298fc1c149afbf4c8996fb924" + }, + { + "name": "one record, no body", + "records": [ + {} + ], + "digest": "6e340b9cffb37a989ca544e6bb780a2c" + }, + { + "name": "one record, an empty body", + "records": [ + { + "data": "" + } + ], + "digest": "6bb0f5246ac6ee3fa6a6879c512f393f" + }, + { + "name": "one record", + "records": [ + { + "data": "68656c6c6f" + } + ], + "digest": "00cedb65fe85721791cf031124cc4ba2" + }, + { + "name": "metadata in key order", + "records": [ + { + "metadata": { + "encoding": "6a736f6e2f706c61696e", + "a": "00ff" + }, + "data": "7b2278223a317d" + } + ], + "digest": "eea3c073d22bc6bd16ffd5c5a18a2900" + }, + { + "name": "two records", + "records": [ + { + "data": "01" + }, + { + "data": "0203" + } + ], + "digest": "b75e10295b9753dbbc5774e454c9a7fb" + }, + { + "name": "external payloads", + "records": [ + { + "data": "01", + "external_payloads": [ + 10, + 20 + ] + } + ], + "digest": "381530dbc9c2fb0d5ddefbc99e72c6e1" + }, + { + "name": "metadata key and value split one way", + "records": [ + { + "metadata": { + "ab": "63" + } + } + ], + "digest": "be5bd276e8b11daeb753e3d3b6d6f93d" + }, + { + "name": "metadata key and value split another way", + "records": [ + { + "metadata": { + "a": "6263" + } + } + ], + "digest": "ddc7d1a81ecebb12548d3adae06622b8" + } +] diff --git a/crates/streams/testdata/epochs.json b/crates/streams/testdata/epochs.json new file mode 100644 index 000000000..fdc62c58c --- /dev/null +++ b/crates/streams/testdata/epochs.json @@ -0,0 +1,55 @@ +[ + { + "kind": "workflow", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 2, + "task_started_time": [ + 1760000000, + 5 + ] + }, + "key": "0000000000000008000000020000000068e778000000000500000000000000000000000000000000", + "task_key": "0000000000000008000000020000000068e7780000000005" + }, + { + "kind": "local_activity", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 1760000000, + 5 + ], + "attempt": 3 + }, + "key": "0000000000000008000000010000000068e778000000000500000000000000000000000000000003", + "task_key": "0000000000000008000000010000000068e7780000000005" + }, + { + "kind": "activity", + "epoch": { + "attempt_scheduled_time": [ + 1760000001, + 999999999 + ], + "attempt": 2 + }, + "key": "0000000000000000000000000000000000000000000000000000000068e778013b9ac9ff00000002", + "task_key": "000000000000000000000000000000000000000000000000" + }, + { + "kind": "client", + "epoch": { + "attempt": 1 + }, + "key": "00000000000000000000000000000000000000000000000000000000000000000000000000000001", + "task_key": "000000000000000000000000000000000000000000000000" + }, + { + "kind": "empty", + "epoch": {}, + "key": "00000000000000000000000000000000000000000000000000000000000000000000000000000000", + "task_key": "000000000000000000000000000000000000000000000000" + } +] diff --git a/crates/streams/testdata/fields.json b/crates/streams/testdata/fields.json new file mode 100644 index 000000000..93517dbde --- /dev/null +++ b/crates/streams/testdata/fields.json @@ -0,0 +1,10 @@ +{ + "kind": 3, + "producer_id": "act:run%3A1:a:s", + "run_id": "run:1", + "fence": "e|3|act:run%3A1:a:s", + "dedupe": "d|3|act:run%3A1:a:s", + "holds": "h|3|act:run%3A1:a:s", + "run_wrote": "run|run%3A1", + "withdrawn_run": "wr|run%3A1" +} diff --git a/crates/streams/testdata/gen_vectors.py b/crates/streams/testdata/gen_vectors.py new file mode 100644 index 000000000..5e181a1bb --- /dev/null +++ b/crates/streams/testdata/gen_vectors.py @@ -0,0 +1,402 @@ +"""Writes the shared test vectors of the Stream runtime. + +This is a second implementation of the layout, in Python, so the crate's tests check the Rust +code against it. The crate's README says which side reads each file. Run it from this directory: + + python3 gen_vectors.py +""" + +import hashlib +import json +from urllib.parse import quote + +SEQ_BITS = 20 + + +def part(text): + return quote(text, safe="") + + +def join(*parts): + return ":".join(part(p) for p in parts) + + +def keys(prefix, namespace, stream_id): + base = f"{part(prefix)}:{{{join(namespace, stream_id)}}}" + return { + "meta": f"{base}:meta", + "log": f"{base}:log", + "producers": f"{base}:producers", + "closed": f"{base}:closed", + } + + +def be64(value): + return max(value, 0).to_bytes(8, "big") + + +def be32(value): + return max(value, 0).to_bytes(4, "big") + + +def task_key(started_event_id=0, attempt=0, started_time=(0, 0)): + """24 bytes, which compare as bytes in the order of the fields.""" + return be64(started_event_id) + be32(attempt) + be64(started_time[0]) + be32(started_time[1]) + + +def epoch_key(epoch): + """40 bytes, the task key and then the scheduled time and the attempt.""" + scheduled = epoch.get("attempt_scheduled_time", (0, 0)) + task = task_key( + epoch.get("task_started_event_id", 0), + epoch.get("task_attempt", 0), + tuple(epoch.get("task_started_time", (0, 0))), + ) + return task + be64(scheduled[0]) + be32(scheduled[1]) + be32(epoch.get("attempt", 0)) + + +def ids(): + cases = [] + for prefix, namespace, workflow_id, first_run_id, name in [ + ("temporal-streams", "default", "agent-1", "run-1", "turn-3"), + ("p:x", "ns{1}", "a:b/c", "r%1", "é name"), + ]: + stream_id = join(workflow_id, first_run_id, name) + cases.append( + { + "prefix": prefix, + "namespace": namespace, + "workflow_id": workflow_id, + "first_run_id": first_run_id, + "name": name, + "stream_id": stream_id, + **keys(prefix, namespace, stream_id), + } + ) + producers = { + "workflow": {"run_id": "run:1", "id": "wf:" + join("run:1")}, + "local_activity": { + "run_id": "run:1", + "call_key": "k-1", + "id": "la:" + join("run:1", "k-1"), + }, + "activity": { + "run_id": "run:1", + "activity_id": "a/1", + "schedule_key": "s-1", + "id": "act:" + join("run:1", "a/1", "s-1"), + }, + "standalone_activity": { + "activity_id": "sa 1", + "activity_run_id": "ar-1", + "id": "sa:" + join("sa 1", "ar-1"), + }, + "client": { + "identity": "1@host", + "process_key": "p-1", + "id": "client:" + join("1@host", "p-1"), + }, + "named_client": {"name": "ingest:eu/1", "id": "client:" + join("ingest:eu/1")}, + } + return {"streams": cases, "producers": producers} + + +def offsets(): + cases = [] + for entry_id in ["0-1", "1-0", "1700000000000-0", "1700000000000-7", "1760000000123-1048575"]: + ms, seq = (int(x) for x in entry_id.split("-")) + cases.append({"entry_id": entry_id, "offset": (ms << SEQ_BITS) | seq}) + return cases + + +def epochs(): + cases = [ + {"kind": "workflow", "epoch": {"task_started_event_id": 8, "task_attempt": 2, + "task_started_time": [1760000000, 5]}}, + {"kind": "local_activity", "epoch": {"task_started_event_id": 8, "task_attempt": 1, + "task_started_time": [1760000000, 5], + "attempt": 3}}, + {"kind": "activity", "epoch": {"attempt_scheduled_time": [1760000001, 999999999], + "attempt": 2}}, + {"kind": "client", "epoch": {"attempt": 1}}, + {"kind": "empty", "epoch": {}}, + ] + for case in cases: + case["key"] = epoch_key(case["epoch"]).hex() + epoch = case["epoch"] + case["task_key"] = task_key( + epoch.get("task_started_event_id", 0), + epoch.get("task_attempt", 0), + tuple(epoch.get("task_started_time", (0, 0))), + ).hex() + return cases + + +def fields(): + run = "run:1" + producer = "act:" + join(run, "a", "s") + return { + "kind": 3, + "producer_id": producer, + "run_id": run, + "fence": f"e|3|{producer}", + "dedupe": f"d|3|{producer}", + "holds": f"h|3|{producer}", + "run_wrote": f"run|{part(run)}", + "withdrawn_run": f"wr|{part(run)}", + } + + +def digest(records): + """The content digest of a batch, which a dedupe entry keeps, as the first 16 bytes of the + SHA-256.""" + out = hashlib.sha256() + + def part(data): + out.update(len(data).to_bytes(8, "big")) + out.update(data) + + for record in records: + # A record with none of these keys has no body. + if not {"metadata", "data", "external_payloads"} & record.keys(): + out.update(b"\x00") + continue + out.update(b"\x01") + metadata = record.get("metadata", {}) + metadata = sorted((key.encode(), bytes.fromhex(value)) for key, value in metadata.items()) + out.update(len(metadata).to_bytes(8, "big")) + for key, value in metadata: + part(key) + part(value) + part(bytes.fromhex(record.get("data", ""))) + external = record.get("external_payloads", []) + out.update(len(external).to_bytes(8, "big")) + for size in external: + out.update(size.to_bytes(8, "big", signed=True)) + return out.digest()[:16].hex() + + +def digests(): + """Batches as hex bytes, with the hex digest that the crate's `batch_digest` must compute.""" + cases = [ + {"name": "empty batch", "records": []}, + {"name": "one record, no body", "records": [{}]}, + {"name": "one record, an empty body", "records": [{"data": ""}]}, + {"name": "one record", "records": [{"data": b"hello".hex()}]}, + { + "name": "metadata in key order", + "records": [ + { + "metadata": {"encoding": b"json/plain".hex(), "a": "00ff"}, + "data": b'{"x":1}'.hex(), + } + ], + }, + {"name": "two records", "records": [{"data": "01"}, {"data": "0203"}]}, + {"name": "external payloads", "records": [{"data": "01", "external_payloads": [10, 20]}]}, + # The length prefixes give these two different digests, though their joined bytes are equal. + { + "name": "metadata key and value split one way", + "records": [{"metadata": {"ab": "63"}}], + }, + { + "name": "metadata key and value split another way", + "records": [{"metadata": {"a": "6263"}}], + }, + ] + for case in cases: + case["digest"] = digest(case["records"]) + assert cases[-1]["digest"] != cases[-2]["digest"] + assert cases[1]["digest"] != cases[2]["digest"] + return cases + + +def withdraws(superseded, record): + """The reader rule of a Superseded item, as the API proposal states it.""" + producer = record["producer"] + if superseded.get("withdraw_run") and producer.get("run_id") == superseded["withdraw_run"]: + return True + # A client's newer epoch only fences. Its earlier Records are real messages. + if superseded["producer_kind"] == "client": + return False + # An item with no epoch, such as a create's, withdraws only a run. + if "epoch" not in superseded or "epoch" not in producer: + return False + old, new = producer["epoch"], superseded["epoch"] + if superseded["producer_kind"] != "workflow": + return ( + producer["kind"] == superseded["producer_kind"] + and producer["id"] == superseded["producer_id"] + and epoch_key(old) < epoch_key(new) + ) + if producer["kind"] == "workflow": + if producer["id"] != superseded["producer_id"]: + return False + elif producer["kind"] == "local_activity": + if producer.get("run_id", "") != superseded["run_id"]: + return False + else: + return False + + def task(epoch): + return ( + epoch.get("task_started_event_id", 0), + task_key( + epoch.get("task_started_event_id", 0), + epoch.get("task_attempt", 0), + tuple(epoch.get("task_started_time", (0, 0))), + ), + ) + + started, old_key = task(old) + t_started, t_key = task(new) + c = superseded.get("previous_completed_task") + c_started = c["started_event_id"] if c else 0 + if started in superseded.get("failed_tasks", []): + return True + if c_started < started < t_started: + return True + if started == t_started and old_key < t_key: + return True + if c and started == c_started: + c_key = task_key(c_started, c.get("attempt", 0), tuple(c.get("started_time", (0, 0)))) + if old_key < c_key: + return True + return False + + +def wf(run, started, attempt=1, time=(100, 0)): + return {"kind": "workflow", "id": "wf:" + part(run), "run_id": run, + "epoch": {"task_started_event_id": started, "task_attempt": attempt, + "task_started_time": list(time)}} + + +def withdrawal_examples(): + """Appendix E of the API proposal, one case per Record that a Superseded item meets.""" + r1 = "r1" + cases = [] + + def case(name, superseded, producer, expected): + record = {"producer": producer} + assert withdraws(superseded, record) == expected, name + cases.append({"name": name, "superseded": superseded, "record": record, + "withdraws": expected}) + + def sup(task, c=None, failed=(), withdraw_run=None, run=r1): + s = {"producer_kind": "workflow", "producer_id": "wf:" + part(run), "run_id": run, + "epoch": task["epoch"], "failed_tasks": list(failed)} + if c: + s["previous_completed_task"] = c + if withdraw_run: + s["withdraw_run"] = withdraw_run + return s + + def completed(started, attempt=1, time=(100, 0)): + return {"started_event_id": started, "attempt": attempt, "started_time": list(time)} + + # 1. A retried task writes nothing. B at [5] failed, [8] wrote nothing, C at [11], c = [8]. + s = sup(wf(r1, 11), completed(8), failed=[5]) + case("e1 B at the failed task 5 goes", s, wf(r1, 5), True) + case("e1 a Record at the completed task 8 stays", s, wf(r1, 8), False) + case("e1 a Record before c stays", s, wf(r1, 3), False) + # 2. A refused completion. B at [5] refused, B' at [8] with failed [5]. + s = sup(wf(r1, 8), completed(2), failed=[5]) + case("e2 B goes", s, wf(r1, 5), True) + case("e2 an older completed Record stays", s, wf(r1, 2), False) + # 3. A transient retry. D at [8] attempt 2, D' at [8] attempt 3. + s = sup(wf(r1, 8, attempt=3), completed(5)) + case("e3 D at a lower attempt of t goes", s, wf(r1, 8, attempt=2), True) + case("e3 a Record of t itself stays", s, wf(r1, 8, attempt=3), False) + # 4. A dropped speculative task. U at [9] T1, the next task at [9] T2. + s = sup(wf(r1, 9, time=(200, 0)), completed(6)) + case("e4 U at an earlier started time goes", s, wf(r1, 9, time=(150, 0)), True) + # 5. A refused transient attempt. c = [5], D at [11] refused, D' at [14]. + s = sup(wf(r1, 14), completed(5)) + case("e5 D between c and t goes", s, wf(r1, 11, attempt=2), True) + case("e5 a Record at c stays", s, wf(r1, 5), False) + # c is the third attempt of task [8], as Core sends it. The transient attempts 1 and 2 have + # no failed event, so only c's attempt withdraws them. + s = sup(wf(r1, 14), completed(8, attempt=3, time=(300, 0))) + case("c attempt 3, attempt 2 at c goes", s, wf(r1, 8, attempt=2, time=(200, 0)), True) + case("c attempt 3, c itself stays", s, wf(r1, 8, attempt=3, time=(300, 0)), False) + # A c with only its started id, from a Core that sends no attempt, withdraws nothing at c. + s = sup(wf(r1, 14), {"started_event_id": 8}) + case("c with no attempt keeps every Record at c", s, wf(r1, 8, attempt=1, time=(200, 0)), + False) + # A Local Activity of a withdrawn task is withdrawn too. + la = {"kind": "local_activity", "id": "la:" + join(r1, "k"), "run_id": r1, + "epoch": {"task_started_event_id": 5, "task_attempt": 1, + "task_started_time": [100, 0], "attempt": 2}} + case("e1 a Local Activity of the failed task goes", sup(wf(r1, 11), completed(8), [5]), la, + True) + other_run = wf("r9", 5) + case("another run's Record stays", sup(wf(r1, 11), completed(8), [5]), other_run, False) + # 6. A Workflow retry. r2's first batch withdraws r1 and its Activity. + act = {"kind": "activity", "id": "act:" + join(r1, "a", "s"), "run_id": r1, + "epoch": {"attempt_scheduled_time": [100, 0], "attempt": 1}} + s = sup(wf("r2", 3), None, withdraw_run=r1, run="r2") + case("e6 r1's Record goes", s, wf(r1, 20), True) + case("e6 r1's Activity Record goes", s, act, True) + case("e6 r2's own earlier Record of t stays", s, wf("r2", 3), False) + # An Activity retry withdraws only its own earlier attempt. + s = {"producer_kind": "activity", "producer_id": act["id"], "run_id": r1, + "epoch": {"attempt_scheduled_time": [130, 0], "attempt": 2}} + case("an Activity retry withdraws attempt 1", s, act, True) + other = dict(act, id="act:" + join(r1, "b", "s")) + case("an Activity retry keeps another Activity", s, other, False) + # A client's newer epoch fences its earlier one and withdraws nothing. + named = {"kind": "client", "id": "client:" + join("ingest"), "epoch": {"attempt": 1}} + s = {"producer_kind": "client", "producer_id": named["id"], "epoch": {"attempt": 2}} + case("a client's newer epoch keeps its earlier Records", s, named, False) + + def la_at(started, task_attempt=1, attempt=1, run=r1, call="k"): + return {"kind": "local_activity", "id": "la:" + join(run, call), "run_id": run, + "epoch": {"task_started_event_id": started, "task_attempt": task_attempt, + "task_started_time": [100, 0], "attempt": attempt}} + + # A retry run's create writes an item with no epoch. It withdraws only the previous run, even + # when it names failed tasks. + s = {"producer_kind": "workflow", "producer_id": "wf:" + part("r2"), "run_id": "r2", + "withdraw_run": r1, "failed_tasks": [5]} + case("no epoch, r1's Record goes", s, wf(r1, 5), True) + case("no epoch, r1's Local Activity Record goes", s, la_at(5), True) + case("no epoch, r2's Local Activity Record at a failed task stays", s, la_at(5, run="r2"), + False) + case("no epoch, r2's Record stays", s, wf("r2", 5), False) + # A Local Activity's newer epoch withdraws only its own lower epochs. + s = {"producer_kind": "local_activity", "producer_id": "la:" + join(r1, "k"), "run_id": r1, + "epoch": la_at(8, attempt=3)["epoch"]} + case("a Local Activity's newer epoch withdraws a lower attempt", s, la_at(8, attempt=2), True) + case("a Local Activity's newer epoch withdraws an earlier task", s, la_at(5, attempt=4), True) + case("a Local Activity's newer epoch keeps its own epoch", s, la_at(8, attempt=3), False) + case("a Local Activity's newer epoch keeps another call", s, la_at(5, call="j"), False) + case("a Local Activity's item keeps the run's Workflow Record", s, wf(r1, 5), False) + # Local Activity Records under a Workflow item with c = [8], t = [14] attempt 2, failed [5]. + s = sup(wf(r1, 14, attempt=2), completed(8), failed=[5]) + case("a Local Activity before c stays", s, la_at(3), False) + case("a Local Activity at c stays", s, la_at(8), False) + case("a Local Activity between c and t goes", s, la_at(11), True) + case("a Local Activity of a lower attempt of t goes", s, la_at(14, task_attempt=1), True) + case("a Local Activity of t itself stays", s, la_at(14, task_attempt=2), False) + case("a Local Activity of another run stays", s, la_at(11, run="r9"), False) + no_epoch = {"kind": "local_activity", "id": "la:" + join(r1, "k"), "run_id": r1} + case("a Local Activity Record with no epoch stays", s, no_epoch, False) + return cases + + +def main(): + for name, value in [ + ("ids.json", ids()), + ("offsets.json", offsets()), + ("epochs.json", epochs()), + ("fields.json", fields()), + ("withdrawals.json", withdrawal_examples()), + ("digests.json", digests()), + ]: + with open(name, "w") as out: + json.dump(value, out, indent=2, ensure_ascii=False) + out.write("\n") + + +if __name__ == "__main__": + main() diff --git a/crates/streams/testdata/ids.json b/crates/streams/testdata/ids.json new file mode 100644 index 000000000..c872c4989 --- /dev/null +++ b/crates/streams/testdata/ids.json @@ -0,0 +1,59 @@ +{ + "streams": [ + { + "prefix": "temporal-streams", + "namespace": "default", + "workflow_id": "agent-1", + "first_run_id": "run-1", + "name": "turn-3", + "stream_id": "agent-1:run-1:turn-3", + "meta": "temporal-streams:{default:agent-1%3Arun-1%3Aturn-3}:meta", + "log": "temporal-streams:{default:agent-1%3Arun-1%3Aturn-3}:log", + "producers": "temporal-streams:{default:agent-1%3Arun-1%3Aturn-3}:producers", + "closed": "temporal-streams:{default:agent-1%3Arun-1%3Aturn-3}:closed" + }, + { + "prefix": "p:x", + "namespace": "ns{1}", + "workflow_id": "a:b/c", + "first_run_id": "r%1", + "name": "é name", + "stream_id": "a%3Ab%2Fc:r%251:%C3%A9%20name", + "meta": "p%3Ax:{ns%7B1%7D:a%253Ab%252Fc%3Ar%25251%3A%25C3%25A9%2520name}:meta", + "log": "p%3Ax:{ns%7B1%7D:a%253Ab%252Fc%3Ar%25251%3A%25C3%25A9%2520name}:log", + "producers": "p%3Ax:{ns%7B1%7D:a%253Ab%252Fc%3Ar%25251%3A%25C3%25A9%2520name}:producers", + "closed": "p%3Ax:{ns%7B1%7D:a%253Ab%252Fc%3Ar%25251%3A%25C3%25A9%2520name}:closed" + } + ], + "producers": { + "workflow": { + "run_id": "run:1", + "id": "wf:run%3A1" + }, + "local_activity": { + "run_id": "run:1", + "call_key": "k-1", + "id": "la:run%3A1:k-1" + }, + "activity": { + "run_id": "run:1", + "activity_id": "a/1", + "schedule_key": "s-1", + "id": "act:run%3A1:a%2F1:s-1" + }, + "standalone_activity": { + "activity_id": "sa 1", + "activity_run_id": "ar-1", + "id": "sa:sa%201:ar-1" + }, + "client": { + "identity": "1@host", + "process_key": "p-1", + "id": "client:1%40host:p-1" + }, + "named_client": { + "name": "ingest:eu/1", + "id": "client:ingest%3Aeu%2F1" + } + } +} diff --git a/crates/streams/testdata/offsets.json b/crates/streams/testdata/offsets.json new file mode 100644 index 000000000..c2f529e70 --- /dev/null +++ b/crates/streams/testdata/offsets.json @@ -0,0 +1,22 @@ +[ + { + "entry_id": "0-1", + "offset": 1 + }, + { + "entry_id": "1-0", + "offset": 1048576 + }, + { + "entry_id": "1700000000000-0", + "offset": 1782579200000000000 + }, + { + "entry_id": "1700000000000-7", + "offset": 1782579200000000007 + }, + { + "entry_id": "1760000000123-1048575", + "offset": 1845493760130023423 + } +] diff --git a/crates/streams/testdata/withdrawals.json b/crates/streams/testdata/withdrawals.json new file mode 100644 index 000000000..fdf395bf4 --- /dev/null +++ b/crates/streams/testdata/withdrawals.json @@ -0,0 +1,1390 @@ +[ + { + "name": "e1 B at the failed task 5 goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 11, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": true + }, + { + "name": "e1 a Record at the completed task 8 stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 11, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "e1 a Record before c stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 11, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 3, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "e2 B goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 2, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": true + }, + { + "name": "e2 an older completed Record stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 2, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 2, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "e3 D at a lower attempt of t goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 3, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "previous_completed_task": { + "started_event_id": 5, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": true + }, + { + "name": "e3 a Record of t itself stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 3, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "previous_completed_task": { + "started_event_id": 5, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 3, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "e4 U at an earlier started time goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 9, + "task_attempt": 1, + "task_started_time": [ + 200, + 0 + ] + }, + "failed_tasks": [], + "previous_completed_task": { + "started_event_id": 6, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 9, + "task_attempt": 1, + "task_started_time": [ + 150, + 0 + ] + } + } + }, + "withdraws": true + }, + { + "name": "e5 D between c and t goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "previous_completed_task": { + "started_event_id": 5, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 11, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": true + }, + { + "name": "e5 a Record at c stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "previous_completed_task": { + "started_event_id": 5, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "c attempt 3, attempt 2 at c goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 3, + "started_time": [ + 300, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 2, + "task_started_time": [ + 200, + 0 + ] + } + } + }, + "withdraws": true + }, + { + "name": "c attempt 3, c itself stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 3, + "started_time": [ + 300, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 3, + "task_started_time": [ + 300, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "c with no attempt keeps every Record at c", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "previous_completed_task": { + "started_event_id": 8 + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 200, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "e1 a Local Activity of the failed task goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 11, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 2 + } + } + }, + "withdraws": true + }, + { + "name": "another run's Record stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 11, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r9", + "run_id": "r9", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "e6 r1's Record goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r2", + "run_id": "r2", + "epoch": { + "task_started_event_id": 3, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "withdraw_run": "r1" + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 20, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": true + }, + { + "name": "e6 r1's Activity Record goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r2", + "run_id": "r2", + "epoch": { + "task_started_event_id": 3, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "withdraw_run": "r1" + }, + "record": { + "producer": { + "kind": "activity", + "id": "act:r1:a:s", + "run_id": "r1", + "epoch": { + "attempt_scheduled_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": true + }, + { + "name": "e6 r2's own earlier Record of t stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r2", + "run_id": "r2", + "epoch": { + "task_started_event_id": 3, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [], + "withdraw_run": "r1" + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r2", + "run_id": "r2", + "epoch": { + "task_started_event_id": 3, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "an Activity retry withdraws attempt 1", + "superseded": { + "producer_kind": "activity", + "producer_id": "act:r1:a:s", + "run_id": "r1", + "epoch": { + "attempt_scheduled_time": [ + 130, + 0 + ], + "attempt": 2 + } + }, + "record": { + "producer": { + "kind": "activity", + "id": "act:r1:a:s", + "run_id": "r1", + "epoch": { + "attempt_scheduled_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": true + }, + { + "name": "an Activity retry keeps another Activity", + "superseded": { + "producer_kind": "activity", + "producer_id": "act:r1:a:s", + "run_id": "r1", + "epoch": { + "attempt_scheduled_time": [ + 130, + 0 + ], + "attempt": 2 + } + }, + "record": { + "producer": { + "kind": "activity", + "id": "act:r1:b:s", + "run_id": "r1", + "epoch": { + "attempt_scheduled_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": false + }, + { + "name": "a client's newer epoch keeps its earlier Records", + "superseded": { + "producer_kind": "client", + "producer_id": "client:ingest", + "epoch": { + "attempt": 2 + } + }, + "record": { + "producer": { + "kind": "client", + "id": "client:ingest", + "epoch": { + "attempt": 1 + } + } + }, + "withdraws": false + }, + { + "name": "no epoch, r1's Record goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r2", + "run_id": "r2", + "withdraw_run": "r1", + "failed_tasks": [ + 5 + ] + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": true + }, + { + "name": "no epoch, r1's Local Activity Record goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r2", + "run_id": "r2", + "withdraw_run": "r1", + "failed_tasks": [ + 5 + ] + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": true + }, + { + "name": "no epoch, r2's Local Activity Record at a failed task stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r2", + "run_id": "r2", + "withdraw_run": "r1", + "failed_tasks": [ + 5 + ] + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r2:k", + "run_id": "r2", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": false + }, + { + "name": "no epoch, r2's Record stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r2", + "run_id": "r2", + "withdraw_run": "r1", + "failed_tasks": [ + 5 + ] + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r2", + "run_id": "r2", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "a Local Activity's newer epoch withdraws a lower attempt", + "superseded": { + "producer_kind": "local_activity", + "producer_id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 3 + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 2 + } + } + }, + "withdraws": true + }, + { + "name": "a Local Activity's newer epoch withdraws an earlier task", + "superseded": { + "producer_kind": "local_activity", + "producer_id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 3 + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 4 + } + } + }, + "withdraws": true + }, + { + "name": "a Local Activity's newer epoch keeps its own epoch", + "superseded": { + "producer_kind": "local_activity", + "producer_id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 3 + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 3 + } + } + }, + "withdraws": false + }, + { + "name": "a Local Activity's newer epoch keeps another call", + "superseded": { + "producer_kind": "local_activity", + "producer_id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 3 + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:j", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": false + }, + { + "name": "a Local Activity's item keeps the run's Workflow Record", + "superseded": { + "producer_kind": "local_activity", + "producer_id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 3 + } + }, + "record": { + "producer": { + "kind": "workflow", + "id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 5, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ] + } + } + }, + "withdraws": false + }, + { + "name": "a Local Activity before c stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 3, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": false + }, + { + "name": "a Local Activity at c stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 8, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": false + }, + { + "name": "a Local Activity between c and t goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 11, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": true + }, + { + "name": "a Local Activity of a lower attempt of t goes", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": true + }, + { + "name": "a Local Activity of t itself stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": false + }, + { + "name": "a Local Activity of another run stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r9:k", + "run_id": "r9", + "epoch": { + "task_started_event_id": 11, + "task_attempt": 1, + "task_started_time": [ + 100, + 0 + ], + "attempt": 1 + } + } + }, + "withdraws": false + }, + { + "name": "a Local Activity Record with no epoch stays", + "superseded": { + "producer_kind": "workflow", + "producer_id": "wf:r1", + "run_id": "r1", + "epoch": { + "task_started_event_id": 14, + "task_attempt": 2, + "task_started_time": [ + 100, + 0 + ] + }, + "failed_tasks": [ + 5 + ], + "previous_completed_task": { + "started_event_id": 8, + "attempt": 1, + "started_time": [ + 100, + 0 + ] + } + }, + "record": { + "producer": { + "kind": "local_activity", + "id": "la:r1:k", + "run_id": "r1" + } + }, + "withdraws": false + } +] diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs new file mode 100644 index 000000000..d824561a4 --- /dev/null +++ b/crates/streams/tests/vectors.rs @@ -0,0 +1,269 @@ +//! The shared test vectors, which `testdata/gen_vectors.py` writes from a second implementation. +//! The crate's README says which other side reads each file. + +use prost_types::Timestamp; +use serde_json::Value; +use temporalio_common::protos::temporal::api::common::v1::{ + Payload, payload::ExternalPayloadDetails, +}; +use temporalio_streams::{ + digest::batch_digest, + epoch::{epoch_key, task_key, task_of, withdraws}, + ids::{self, StreamKeys}, + offset::{entry_id_of_offset, offset_of_entry_id}, + proto::{ + ActivityEpoch, ActivityProducer, AppendRecord, ClientEpoch, ClientProducer, + LocalActivityEpoch, LocalActivityProducer, Producer, RunWithdrawal, StreamData, + StreamRecord, StreamSuperseded, StreamWriter, WorkflowProducer, WorkflowTaskEpoch, + producer::Kind, stream_record::Item, + }, +}; + +fn load(name: &str) -> Value { + let path = format!("{}/testdata/{name}", env!("CARGO_MANIFEST_DIR")); + serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap() +} + +fn text<'a>(value: &'a Value, field: &str) -> &'a str { + value[field].as_str().unwrap() +} + +fn time(value: &Value) -> Option { + let pair = value.as_array()?; + Some(Timestamp { + seconds: pair[0].as_i64().unwrap(), + nanos: pair[1].as_i64().unwrap() as i32, + }) +} + +fn number(value: &Value, field: &str) -> i64 { + value[field].as_i64().unwrap_or(0) +} + +/// 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 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(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), + "empty" => None, + other => panic!("unknown kind {other}"), + } +} + +#[test] +fn stream_ids_and_keys_match() { + let vectors = load("ids.json"); + for case in vectors["streams"].as_array().unwrap() { + let id = ids::scoped_stream_id( + text(case, "workflow_id"), + text(case, "first_run_id"), + text(case, "name"), + ); + assert_eq!(id, text(case, "stream_id")); + let keys = StreamKeys::new(text(case, "prefix"), text(case, "namespace"), &id); + assert_eq!(keys.meta(), text(case, "meta")); + assert_eq!(keys.log(), text(case, "log")); + assert_eq!(keys.producers(), text(case, "producers")); + assert_eq!(keys.closed(), text(case, "closed")); + } +} + +#[test] +fn producer_ids_match() { + let producers = &load("ids.json")["producers"]; + let id = |name: &str| text(&producers[name], "id"); + let field = |name: &str, field: &str| text(&producers[name], field); + assert_eq!( + ids::workflow_producer_id(field("workflow", "run_id")), + id("workflow") + ); + assert_eq!( + ids::local_activity_producer_id( + field("local_activity", "run_id"), + field("local_activity", "call_key") + ), + id("local_activity") + ); + assert_eq!( + ids::activity_producer_id( + field("activity", "run_id"), + field("activity", "activity_id"), + field("activity", "schedule_key") + ), + id("activity") + ); + assert_eq!( + ids::standalone_activity_producer_id( + field("standalone_activity", "activity_id"), + field("standalone_activity", "activity_run_id") + ), + id("standalone_activity") + ); + assert_eq!( + ids::client_producer_id(field("client", "identity"), field("client", "process_key")), + id("client") + ); + assert_eq!( + ids::named_client_producer_id(field("named_client", "name")), + id("named_client") + ); +} + +#[test] +fn offsets_match() { + for case in load("offsets.json").as_array().unwrap() { + let offset = number(case, "offset"); + assert_eq!(offset_of_entry_id(text(case, "entry_id")).unwrap(), offset); + assert_eq!(entry_id_of_offset(offset), text(case, "entry_id")); + } +} + +#[test] +fn epoch_keys_match() { + for case in load("epochs.json").as_array().unwrap() { + 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(kind.as_ref())).to_vec(), + hex(text(case, "task_key")) + ); + } +} + +#[test] +fn the_appendix_e_withdrawals_match() { + let cases = load("withdrawals.json"); + 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(), + 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), + 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(), + }), + }), + }; + let p = &case["record"]["producer"]; + let record = StreamRecord { + item: Some(Item::Data(StreamData { + producer: Some(Producer { + id: text(p, "id").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() + })), + ..Default::default() + }; + assert_eq!( + withdraws(&superseded, &record), + case["withdraws"].as_bool().unwrap(), + "{}", + text(case, "name") + ); + } +} + +fn hex(text: &str) -> Vec { + (0..text.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&text[i..i + 2], 16).unwrap()) + .collect() +} + +#[test] +fn batch_digests_match() { + for case in load("digests.json").as_array().unwrap() { + let records: Vec<_> = case["records"] + .as_array() + .unwrap() + .iter() + .map(|r| AppendRecord { + // A record with none of these keys has no body. + body: ["metadata", "data", "external_payloads"] + .iter() + .any(|key| r.get(key).is_some()) + .then(|| Payload { + metadata: r["metadata"] + .as_object() + .into_iter() + .flatten() + .map(|(key, value)| (key.clone(), hex(value.as_str().unwrap()))) + .collect(), + data: hex(r["data"].as_str().unwrap_or_default()), + external_payloads: r["external_payloads"] + .as_array() + .into_iter() + .flatten() + .map(|size| ExternalPayloadDetails { + size_bytes: size.as_i64().unwrap(), + }) + .collect(), + }), + }) + .collect(); + assert_eq!( + batch_digest(&records).to_vec(), + hex(text(case, "digest")), + "{}", + text(case, "name") + ); + } +}