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..115091078 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -10,6 +10,19 @@ import "temporal/api/stream/v1/message.proto"; // Lang sends them through the bridge's call on the runtime object, by the method name in each // comment. One generic call keeps Stream logic out of every bridge. +// The settings of the memory Stream runtime, for tests. Each field has the number of the same +// field in the config of the other Stores, such as Redis, so a backend can send that config here +// too. +message MemoryStreamConfig { + // The default Retention of a new Stream. Empty is 7 days. + google.protobuf.Duration retention = 4; + // How long an open Workflow-owned Stream stays after its last write. Empty is 90 days. + google.protobuf.Duration safety_ttl = 5; + // The backend name that lang registered this runtime under. The runtime sets it on each + // StreamRef that it builds, such as the one in a create response. + string backend = 10; +} + // Method `OpenWriter`, asynchronous. A writer for an Activity, a Local Activity or a client // Producer, which keeps its epoch and `sequence` order. A writer of a Local Activity, or of an // Activity attempt above 1, first sends an empty batch, so that the new epoch withdraws the earlier @@ -20,9 +33,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 eefe30a42..4c623c40c 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -36,6 +36,8 @@ default-features = false futures-util = { version = "0.3", default-features = false, features = ["std"] } serde_json = { workspace = true } tokio = { version = "1.47", features = ["macros", "rt-multi-thread", "time"] } +# The integration tests call the memory Store's test hooks, which test-support gates. +temporalio-streams = { path = ".", features = ["test-support"] } [lints] workspace = true diff --git a/crates/streams/README.md b/crates/streams/README.md index 634a43b22..7a6d20b5e 100644 --- a/crates/streams/README.md +++ b/crates/streams/README.md @@ -21,6 +21,15 @@ re-exports them, so a value needs no conversion between the crates. `fields.json` through the shared Store suite. `digests.json` is the golden of the crate's batch digest. +## The runtime + +`StreamRuntime` serves the Stream service on one Store. A Store crate builds it with +`StreamRuntime::new` and a `StoreFactory`, which connects the Store on the runtime's first call +that needs it. So a bridge can build the runtime with no async runtime, and a Workflow that never +publishes never connects. `MemoryStreamRuntime` builds it on the memory Store, from a +`MemoryStreamConfig`. That message has the field numbers of the same fields in the Redis config, +so a bridge can send the memory runtime the config it builds for Redis. + ## Stores and the shared suites The memory Store is for tests. It keeps the key layout of the Redis Store and applies the same rules 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/lib.rs b/crates/streams/src/lib.rs index 8c87e1fbf..e85e36dca 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -11,12 +11,17 @@ pub mod epoch; mod error; pub mod ids; pub mod memory; +mod metrics; pub mod offset; +mod record; +mod runtime; +mod service; pub mod store; #[cfg(any(test, feature = "test-support"))] pub mod testing; pub use error::{Reason, StreamError, StreamResult}; +pub use runtime::{MemoryStreamRuntime, StoreFactory, StreamRuntime}; /// The protos that lang and the crate exchange. `temporalio-protos` generates them, so a value /// needs no conversion between the crates. diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 52804d291..4cd315d1f 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -578,7 +578,7 @@ 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 withdrawal = &batch.withdrawal; let (safety, tombstone) = (millis(self.safety_ttl), millis(TOMBSTONE)); let field = format!("p|{}|{}", kind as i32, producer.id); @@ -721,7 +721,7 @@ 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 field = format!("p|{}|{}", kind as i32, producer.id); let tombstone = millis(TOMBSTONE); let chain = owner_chain(stream_ref); diff --git a/crates/streams/src/metrics.rs b/crates/streams/src/metrics.rs new file mode 100644 index 000000000..b894762e7 --- /dev/null +++ b/crates/streams/src/metrics.rs @@ -0,0 +1,50 @@ +//! The crate's metrics. They use the meter that lang gives, or a no-op meter. + +use temporalio_common::telemetry::metrics::{ + Counter, MetricAttributes, MetricParameters, TemporalMeter, +}; + +#[derive(Clone)] +pub(crate) struct Metrics { + attributes: MetricAttributes, + fenced: Counter, + superseded: Counter, +} + +fn parameters( + name: &'static str, + description: &'static str, + unit: &'static str, +) -> MetricParameters { + MetricParameters { + name: name.into(), + description: description.into(), + unit: unit.into(), + } +} + +impl Metrics { + pub(crate) fn new(meter: &TemporalMeter) -> Self { + Self { + attributes: meter.get_default_attributes().clone(), + fenced: meter.counter(parameters( + "stream_fenced", + "Batches that the Store refused for a lower epoch or a withdrawn run", + "", + )), + superseded: meter.counter(parameters( + "stream_superseded", + "Superseded flags that the Store set", + "", + )), + } + } + + pub(crate) fn fenced(&self) { + self.fenced.add(1, &self.attributes); + } + + pub(crate) fn superseded(&self) { + self.superseded.add(1, &self.attributes); + } +} diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs new file mode 100644 index 000000000..2277adf9b --- /dev/null +++ b/crates/streams/src/record.rs @@ -0,0 +1,117 @@ +//! The stored Records and the `Superseded` facts of a batch. + +use crate::{ + Reason, StreamError, StreamResult, + proto::{ + AppendRecord, Producer, RunWithdrawal, StreamData, StreamRecord, StreamSuperseded, + StreamWriter, stream_record::Item, + }, + store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES}, +}; +use prost::Message; + +/// One stored data Record. +pub(crate) fn data_record(producer: &Producer, record: AppendRecord) -> Vec { + StreamRecord { + item: Some(Item::Data(StreamData { + producer: Some(producer.clone()), + body: record.body, + })), + ..Default::default() + } + .encode_to_vec() +} + +/// The stored Records of one batch from outside Workflow code. A batch over the Store's limits +/// fails, since the crate doesn't split a batch it didn't build. +pub(crate) fn data_records( + producer: &Producer, + records: Vec, +) -> StreamResult>> { + if records.len() > MAX_BATCH_RECORDS { + return Err(StreamError::new( + Reason::TooLarge, + format!( + "The batch has {} Records, more than {MAX_BATCH_RECORDS}", + records.len() + ), + )); + } + let records: Vec<_> = records + .into_iter() + .map(|record| data_record(producer, record)) + .collect(); + if let Some(big) = records + .iter() + .find(|record| record.len() > MAX_RECORD_BYTES) + { + return Err(StreamError::new( + Reason::TooLarge, + format!( + "A Record has {} bytes, more than {MAX_RECORD_BYTES}", + big.len() + ), + )); + } + let total: usize = records.iter().map(Vec::len).sum(); + if total > MAX_BATCH_BYTES { + return Err(StreamError::new( + Reason::TooLarge, + format!("The batch has {total} bytes, more than {MAX_BATCH_BYTES}"), + )); + } + Ok(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()), + withdrawal, + }), + } + .encode_to_vec() +} + +#[cfg(test)] +mod tests { + use super::*; + use temporalio_common::protos::temporal::api::common::v1::Payload; + + #[test] + fn a_batch_over_the_limits_fails() { + let producer = Producer::default(); + let record = |bytes: usize| AppendRecord { + body: Some(Payload { + data: vec![0; bytes], + ..Default::default() + }), + }; + let error = data_records(&producer, vec![record(MAX_RECORD_BYTES)]).unwrap_err(); + assert_eq!(error.reason, Reason::TooLarge); + let many = (0..4) + .map(|_| record(MAX_RECORD_BYTES / 2 + 1)) + .collect::>(); + let many = [many.clone(), many].concat(); + assert_eq!( + data_records(&producer, many).unwrap_err().reason, + Reason::TooLarge + ); + assert_eq!(data_records(&producer, vec![record(10)]).unwrap().len(), 1); + let small = vec![record(1); MAX_BATCH_RECORDS + 1]; + assert_eq!( + data_records(&producer, small).unwrap_err().reason, + Reason::TooLarge + ); + let small = vec![record(1); MAX_BATCH_RECORDS]; + assert_eq!( + data_records(&producer, small).unwrap().len(), + MAX_BATCH_RECORDS + ); + } +} diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs new file mode 100644 index 000000000..08d7e6a61 --- /dev/null +++ b/crates/streams/src/runtime.rs @@ -0,0 +1,230 @@ +//! The Stream runtime, a plain object that lang's bridge builds per client and calls by name. + +use crate::{ + Reason, StreamError, StreamResult, + memory::MemoryStore, + metrics::Metrics, + proto::{AppendRecordsRequest, CloseStreamRequest, DescribeStreamRequest, MemoryStreamConfig}, + service::StreamService, + store::{DEFAULT_RETENTION, DEFAULT_SAFETY_TTL, Store}, +}; +use prost::Message; +use std::{sync::Arc, time::Duration}; +use temporalio_common::telemetry::metrics::TemporalMeter; +use tokio::sync::OnceCell; + +/// Connects the Store of a runtime. +/// +/// The runtime connects on its first call that needs the Store. So a bridge can build the runtime +/// with no async runtime, and a Workflow that never publishes never connects. A failed connect is +/// tried again on the next call. +#[async_trait::async_trait] +pub trait StoreFactory: Send + Sync + 'static { + /// Connects a Store. + async fn connect(&self) -> StreamResult>; +} + +/// A Store that is already connected. +struct Connected(Arc); + +#[async_trait::async_trait] +impl StoreFactory for Connected { + async fn connect(&self) -> StreamResult> { + Ok(self.0.clone()) + } +} + +/// The state that the runtime's calls share. +pub(crate) struct Shared { + name: &'static str, + backend: String, + metrics: Metrics, + factory: Arc, + service: OnceCell>, +} + +impl Shared { + /// The service, which connects its Store on its first use. + pub(crate) async fn service(&self) -> StreamResult<&Arc> { + self.service + .get_or_try_init(|| async { + let store = self.factory.connect().await?; + Ok(Arc::new(StreamService { + store, + metrics: self.metrics.clone(), + backend: self.backend.clone(), + })) + }) + .await + } +} + +/// The Stream runtime on one Store. +/// +/// A bridge builds one for each client with a Stream backend, and that client's Workers share it. +/// A Store crate, such as `temporalio-streams-redis`, builds it with its own Store. +pub struct StreamRuntime { + shared: Arc, +} + +impl std::fmt::Debug for StreamRuntime { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StreamRuntime") + .field("name", &self.shared.name) + .field("backend", &self.shared.backend) + .finish_non_exhaustive() + } +} + +impl StreamRuntime { + /// Builds a runtime that connects its Store with `factory` on its first call that needs it. + /// `name` identifies the runtime in logs. `backend` is the name that lang registered the + /// runtime under, which the refs that the runtime builds carry. The runtime records metrics + /// on `meter`, lang's runtime meter, if set. + pub fn new( + name: &'static str, + backend: impl Into, + factory: Arc, + meter: Option, + ) -> Arc { + let meter = meter.unwrap_or_else(TemporalMeter::no_op); + let shared = Arc::new(Shared { + name, + backend: backend.into(), + metrics: Metrics::new(&meter), + factory, + service: OnceCell::new(), + }); + Arc::new(Self { shared }) + } + + /// Builds a runtime on a Store that is already connected. + pub fn with_store( + name: &'static str, + backend: impl Into, + store: Arc, + meter: Option, + ) -> Arc { + Self::new(name, backend, Arc::new(Connected(store)), meter) + } + + /// A name that identifies the runtime in logs. + pub fn name(&self) -> &str { + self.shared.name + } + + /// Runs one method on encoded input and returns the encoded output. The bridge raises an error + /// with [StreamError::code] and [StreamError::details]. + pub async fn call(&self, method: &str, input: &[u8]) -> StreamResult> { + let service = self.shared.service().await?; + let output = match method { + "CreateStream" => service.create_stream(decode(input)?).await?.encode_to_vec(), + "AppendRecords" => service + .append_records(decode::(input)?) + .await? + .encode_to_vec(), + "CloseStream" => service + .close_stream(decode::(input)?) + .await? + .encode_to_vec(), + "DescribeStream" => service + .describe_stream(decode::(input)?) + .await? + .encode_to_vec(), + _ => { + return Err(StreamError::new( + Reason::Unsupported, + format!("Unknown method {method}"), + )); + } + }; + Ok(output) + } +} + +/// The name that [StreamRuntime::name] gives for a memory runtime. +const MEMORY_RUNTIME_NAME: &str = "temporalio-streams-memory"; + +/// The Stream runtime on the memory Store, for tests. It serves the same calls with the same +/// rules as every other Store, and it keeps its Streams in this process only. +#[derive(Debug)] +pub struct MemoryStreamRuntime; + +impl MemoryStreamRuntime { + /// Builds a runtime from an encoded [MemoryStreamConfig]. Its field numbers are those of the + /// same fields in the config of another Store, such as Redis, so a backend can pass that config + /// too. + pub fn from_config_bytes( + config: &[u8], + meter: Option, + ) -> StreamResult> { + let config = MemoryStreamConfig::decode(config) + .map_err(|error| StreamError::invalid(format!("The config is bad: {error}")))?; + let memory = Arc::new(MemoryStore::new( + duration(config.retention).unwrap_or(DEFAULT_RETENTION), + duration(config.safety_ttl).unwrap_or(DEFAULT_SAFETY_TTL), + )); + Ok(StreamRuntime::with_store( + MEMORY_RUNTIME_NAME, + config.backend, + memory, + meter, + )) + } + + /// Builds a runtime on any Store, for tests of a Store that lacks some capabilities. + #[doc(hidden)] + pub fn with_store(store: Arc, meter: Option) -> Arc { + StreamRuntime::with_store(MEMORY_RUNTIME_NAME, String::new(), store, meter) + } +} + +fn duration(value: Option) -> Option { + let value = value?; + let duration = Duration::new(value.seconds.max(0) as u64, value.nanos.max(0) as u32); + (!duration.is_zero()).then_some(duration) +} + +pub(crate) fn decode(input: &[u8]) -> StreamResult { + M::decode(input).map_err(|error| StreamError::invalid(format!("The request is bad: {error}"))) +} + +#[cfg(test)] +mod tests { + use super::*; + + const METHODS: [&str; 4] = [ + "CreateStream", + "AppendRecords", + "CloseStream", + "DescribeStream", + ]; + + fn runtime() -> Arc { + let config = MemoryStreamConfig::default().encode_to_vec(); + MemoryStreamRuntime::from_config_bytes(&config, None).unwrap() + } + + // A call connects the Store before it decodes its request, so the runtime is a memory one. A + // typo in a method name would make it unknown, which is unsupported, so the loop also checks + // the names that a bridge calls. + #[tokio::test] + async fn a_bad_request_is_invalid_and_an_unknown_method_is_unsupported() { + let runtime = runtime(); + for method in METHODS { + let error = runtime.call(method, &[0xff]).await.unwrap_err(); + assert_eq!(error.reason, Reason::Invalid, "{method}"); + } + let error = runtime.call("Nope", &[]).await.unwrap_err(); + assert_eq!( + (error.reason, error.message.as_str()), + (Reason::Unsupported, "Unknown method Nope") + ); + } + + #[test] + fn a_bad_memory_config_is_invalid() { + let error = MemoryStreamRuntime::from_config_bytes(&[0xff], None).unwrap_err(); + assert_eq!(error.reason, Reason::Invalid); + } +} diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs new file mode 100644 index 000000000..4fb03054b --- /dev/null +++ b/crates/streams/src/service.rs @@ -0,0 +1,331 @@ +//! The Stream service operations on the Store, for Activities, clients and the Local Activity that +//! runs each Workflow call. + +use crate::{ + Reason, StreamError, StreamResult, + digest::batch_digest, + epoch::{ProducerKind, producer_run_id, task_of, writer_withdraw_run}, + ids::{check_producer, check_stream_id}, + metrics::Metrics, + proto::{ + AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, CloseStreamResponse, + CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, + Producer, StreamIdConflictPolicy, StreamRef, StreamStatus, + }, + record::{data_records, superseded}, + store::{ + Batch, Capabilities, ConflictPolicy, Create, CreateOutcome, MAX_SEQUENCE, MAX_TTL, + MIN_RETENTION, Owner, Store, Withdrawal, owner_chain, + }, +}; +use std::{sync::Arc, time::Duration}; + +/// The Stream service on one Store. +pub(crate) struct StreamService { + pub(crate) store: Arc, + pub(crate) metrics: Metrics, + /// The backend name for the refs the service builds. + pub(crate) backend: String, +} + +fn duration(value: Option) -> Option { + let value = value?; + let duration = Duration::new(value.seconds.max(0) as u64, value.nanos.max(0) as u32); + (!duration.is_zero()).then_some(duration) +} + +/// The Retention of a create, `None` for the Store default. A Store computes expiry times from it, +/// so a value past [MAX_TTL] would overflow them after the create's first write. +fn retention(value: Option) -> StreamResult> { + let Some(value) = value else { + return Ok(None); + }; + let out_of_range = || { + StreamError::invalid(format!( + "The Retention must be 0 for the Store default, or from {} ms to {} s", + MIN_RETENTION.as_millis(), + MAX_TTL.as_secs() + )) + }; + let (Ok(seconds), Ok(nanos)) = (u64::try_from(value.seconds), u32::try_from(value.nanos)) + else { + return Err(out_of_range()); + }; + if nanos >= 1_000_000_000 || Duration::new(seconds, nanos) > MAX_TTL { + return Err(out_of_range()); + } + let retention = duration(Some(value)); + if retention.is_some_and(|retention| retention < MIN_RETENTION) { + return Err(out_of_range()); + } + Ok(retention) +} + +/// The conflict policy of a create. Unspecified is FAIL, as for a Workflow start. +fn conflict_policy(value: i32) -> StreamResult { + match StreamIdConflictPolicy::try_from(value) { + Ok(StreamIdConflictPolicy::Unspecified | StreamIdConflictPolicy::Fail) => { + Ok(ConflictPolicy::Fail) + } + Ok(StreamIdConflictPolicy::UseExisting) => Ok(ConflictPolicy::UseExisting), + Err(_) => Err(StreamError::invalid(format!( + "The conflict policy {value} is unknown" + ))), + } +} + +fn proto_duration(value: Duration) -> prost_types::Duration { + prost_types::Duration { + seconds: value.as_secs() as i64, + nanos: value.subsec_nanos() as i32, + } +} + +fn stream_id(stream_ref: Option<&StreamRef>) -> StreamResult<&str> { + let id = stream_ref + .map(|stream_ref| stream_ref.id.as_str()) + .unwrap_or_default(); + check_stream_id(id)?; + Ok(id) +} + +/// The `Superseded` facts of a create that withdraws a retry run's previous run. It has no epoch, +/// so a Reader withdraws only that run's Records. +fn run_superseded(owner: Option<&Producer>, withdraw_run: &str) -> Vec { + let producer = Producer { + kind: None, + ..owner.cloned().unwrap_or_default() + }; + superseded(&producer, withdraw_run) +} + +/// 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"))?; + check_producer(producer)?; + let kind = ProducerKind::try_from(producer)?; + if kind == ProducerKind::Workflow { + return Err(StreamError::invalid( + "A Workflow writes with the Producer of its Local Activity", + )); + } + // Without its task, a re-dispatch in a later task has the same epoch as the first dispatch. + 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", + )); + } + Ok(producer) +} + +/// 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> { + check_producer(owner)?; + if ProducerKind::try_from(owner)? != ProducerKind::Workflow { + return Err(StreamError::invalid( + "The owner of a create is a Workflow Producer", + )); + } + if owner_chain(stream_ref).is_empty() { + return Err(StreamError::invalid( + "A create with an owner needs a ref whose owner is the Workflow's run chain", + )); + } + Ok(owner) +} + +/// Checks the Producer of a close. A Workflow's close Activity closes as the run's Workflow +/// Producer, so a Workflow Producer may close. +fn close_producer(producer: Option<&Producer>) -> StreamResult<&Producer> { + let producer = producer.ok_or_else(|| StreamError::invalid("The request has no Producer"))?; + check_producer(producer)?; + Ok(producer) +} + +impl StreamService { + pub(crate) async fn create_stream( + &self, + request: CreateStreamRequest, + ) -> StreamResult { + stream_id(request.stream_ref.as_ref())?; + let stream = request.stream_ref.clone().unwrap_or_default(); + let retention = retention(request.retention)?; + let conflict = conflict_policy(request.id_conflict_policy)?; + // A Workflow's call from outside Workflow code, such as from a Local Activity, sets the + // owner run. Then the Stream has the safety TTL of a Workflow-owned Stream. + let owner = match &request.owner { + Some(owner) => { + create_owner(owner, &stream)?; + self.store + .capabilities() + .require(Capabilities::SAFETY_TTL)?; + Owner::Workflow + } + None => Owner::Client, + }; + if !request.withdraw_run.is_empty() { + // The `Superseded` item names its Producer, which here can only be the owner. Any other + // caller could withdraw and fence an unrelated run, as on the append. + let Some(owner) = &request.owner else { + return Err(StreamError::invalid( + "A create with withdraw_run needs the owner run", + )); + }; + 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)?; + } + let bytes = run_superseded(request.owner.as_ref(), &request.withdraw_run); + let create = Create { + retention, + owner, + conflict, + request_id: &request.request_id, + withdrawal: Withdrawal { + withdraw_run: &request.withdraw_run, + superseded: &bytes, + tasks: None, + }, + }; + let created = self + .store + .create(&request.namespace, &stream, &create) + .await?; + if created.flagged { + self.metrics.superseded(); + } + Ok(CreateStreamResponse { + r#ref: Some(StreamRef { + id: stream.id, + owner: stream.owner.filter(|_| owner == Owner::Workflow), + backend: self.backend.clone(), + }), + created: created.outcome == CreateOutcome::Created, + }) + } + + pub(crate) async fn append_records( + &self, + request: AppendRecordsRequest, + ) -> StreamResult { + stream_id(request.r#ref.as_ref())?; + let stream = request.r#ref.clone().unwrap_or_default(); + let writer = request.writer.unwrap_or_default(); + let producer = outside_producer(writer.producer.as_ref())?; + let kind = ProducerKind::try_from(producer)?; + if !(0..=MAX_SEQUENCE).contains(&request.sequence) { + return Err(StreamError::invalid( + "The sequence must be from 1 to 2^53 - 1, or zero for the next one", + )); + } + // Without the fence and the flag, a copy or a stale batch would reach Readers. + self.store + .capabilities() + .require(Capabilities::PRODUCER_FENCE.union(Capabilities::WITHDRAWAL))?; + if request.sequence == 0 { + // A call with no writer state sends no sequence. Only the Store can number the batch in + // the same step as its dedupe check. + self.store + .capabilities() + .require(Capabilities::NEXT_SEQUENCE)?; + } + 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 || 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)?; + } + let digest = batch_digest(&request.records); + let records = data_records(producer, request.records)?; + let bytes = superseded(producer, withdraw_run); + let batch = Batch { + producer, + sequence: request.sequence, + records: &records, + digest, + withdrawal: Withdrawal { + withdraw_run, + superseded: &bytes, + tasks: None, + }, + }; + let appended = self.store.append(&request.namespace, &stream, &batch).await; + let appended = self.count(appended)?; + let (first_offset, last_offset) = appended.offsets.unwrap_or_default(); + Ok(AppendRecordsResponse { + first_offset, + last_offset, + }) + } + + /// Counts a fence or a flag of an append on the metrics. + pub(crate) fn count(&self, result: StreamResult) -> StreamResult { + match &result { + Ok(value) if value.flagged() => self.metrics.superseded(), + Err(error) if error.reason == Reason::Fenced => self.metrics.fenced(), + _ => {} + } + result + } + + pub(crate) async fn close_stream( + &self, + request: CloseStreamRequest, + ) -> StreamResult { + stream_id(request.r#ref.as_ref())?; + let stream = request.r#ref.clone().unwrap_or_default(); + let writer = request.writer.unwrap_or_default(); + let producer = close_producer(writer.producer.as_ref())?; + self.store.capabilities().require(Capabilities::CLOSE)?; + self.store + .close(&request.namespace, &stream, producer) + .await?; + Ok(CloseStreamResponse {}) + } + + pub(crate) async fn describe_stream( + &self, + request: DescribeStreamRequest, + ) -> StreamResult { + let id = stream_id(request.r#ref.as_ref())?; + let (state, _) = self.store.read(&request.namespace, id, 0, 0).await?; + if !state.exists { + return Err(StreamError::new( + Reason::NotFound, + "Nobody created the Stream", + )); + } + // A closed Stream that expired has lost its log, and every read fails. + let status = if state.expired() { + StreamStatus::Expired + } else if state.closed { + StreamStatus::Closed + } else { + StreamStatus::Open + }; + Ok(DescribeStreamResponse { + status: status as i32, + last_offset: state.newest, + retention: Some(proto_duration(state.retention)), + }) + } +} + +/// A Store answer that may have withdrawn Records. +pub(crate) trait Flagged { + fn flagged(&self) -> bool; +} + +impl Flagged for crate::store::Appended { + fn flagged(&self) -> bool { + self.flagged + } +} diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 7a0e8d1ac..825976ca4 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -24,6 +24,9 @@ pub const FORMAT_VERSION: u32 = 1; pub const MAX_RECORD_BYTES: usize = 1536 * 1024; /// The maximum size of a stored append, in bytes. pub const MAX_BATCH_BYTES: usize = 4 * 1024 * 1024; +/// The maximum number of Records in an append. Many small Records fit under [MAX_BATCH_BYTES], but +/// they keep a Store busy for long. +pub const MAX_BATCH_RECORDS: usize = 1000; /// The highest batch `sequence`. A Lua number is exact for integers only up to this value. pub const MAX_SEQUENCE: i64 = (1 << 53) - 1; /// How long the meta key stays after the log key expires, so that a read can tell an expired Stream @@ -32,6 +35,9 @@ pub const TOMBSTONE: Duration = Duration::from_secs(30 * 24 * 60 * 60); /// The longest Retention or safety TTL, which is the largest proto Duration. Its ms plus the time /// of a write stay exact in a Lua number. pub const MAX_TTL: Duration = Duration::from_secs(315_576_000_000); +/// The shortest Retention. A Store keeps whole ms, so a shorter one would expire the Stream at its +/// next write. +pub const MIN_RETENTION: Duration = Duration::from_millis(1); /// The owner run chain of a ref as `:`, each part encoded, or empty /// for another owner. A Store keeps it on a Workflow-owned Stream and checks creates and closes on diff --git a/crates/streams/src/testing/mod.rs b/crates/streams/src/testing/mod.rs index 0b2aa498d..350ab7964 100644 --- a/crates/streams/src/testing/mod.rs +++ b/crates/streams/src/testing/mod.rs @@ -4,15 +4,19 @@ mod support; pub mod fields; +pub mod service; pub mod store; pub use support::Needs; use crate::{ + MemoryStreamRuntime, StreamRuntime, memory::MemoryStore, + proto::MemoryStreamConfig, store::{DEFAULT_RETENTION, DEFAULT_SAFETY_TTL, Store}, }; use futures_util::{FutureExt, future::join_all}; +use prost::Message; use std::{future::Future, panic::AssertUnwindSafe, pin::Pin, sync::Arc}; /// The state under a Store's calls, which some cases read or change. @@ -37,6 +41,8 @@ pub trait TestBackend: Send + Sync { fn gives(&self) -> Needs; /// A Store for one case, and the state under it. async fn connect(&self) -> (Arc, Arc); + /// A runtime on the backend, whose refs carry `backend`. + fn runtime(&self, backend: &str) -> Arc; } /// The memory Store, with a new Store for each case. @@ -53,6 +59,15 @@ impl TestBackend for MemoryBackend { let store = Arc::new(MemoryStore::new(DEFAULT_RETENTION, DEFAULT_SAFETY_TTL)); (store.clone(), store) } + + // Through the config bytes, as a bridge builds it. + fn runtime(&self, backend: &str) -> Arc { + let config = MemoryStreamConfig { + backend: backend.to_string(), + ..Default::default() + }; + MemoryStreamRuntime::from_config_bytes(&config.encode_to_vec(), None).unwrap() + } } #[async_trait::async_trait] @@ -133,4 +148,9 @@ mod tests { async fn the_store_suite_passes_on_the_memory_store() { run(vec![Arc::new(MemoryBackend)], store::CASES).await; } + + #[tokio::test] + async fn the_service_suite_passes_on_the_memory_store() { + run(vec![Arc::new(MemoryBackend)], service::CASES).await; + } } diff --git a/crates/streams/src/testing/service.rs b/crates/streams/src/testing/service.rs new file mode 100644 index 000000000..6eea3828b --- /dev/null +++ b/crates/streams/src/testing/service.rs @@ -0,0 +1,493 @@ +//! The Stream service operations through the runtime's byte call, as lang sends them. +//! +//! Every Store runs the suite through its own runtime. + +use super::{case, support::Needs}; + +use crate::{ + StreamError, StreamRuntime, + ids::{activity_producer_id, client_producer_id}, + proto::{ + ActivityEpoch, ActivityProducer, AppendRecord, AppendRecordsRequest, AppendRecordsResponse, + ClientEpoch, ClientProducer, CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, + CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, LocalActivityEpoch, + LocalActivityProducer, Producer, StreamErrorDetails, StreamIdConflictPolicy, StreamRef, + StreamStatus, StreamWriter, WorkflowProducer, WorkflowTaskEpoch, producer::Kind, + }, +}; +use prost::Message; +use std::{sync::Arc, time::Duration}; +use temporalio_common::protos::temporal::api::common::v1::Payload; +use tonic::Code; + +struct Case { + ext: Arc, + namespace: String, + stream: String, +} + +static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + +fn unique() -> String { + let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + format!("{now}-{n}") +} + +/// A fresh Stream on each backend that gives `needs`. +async fn cases_for(needs: Needs) -> Vec { + super::support::runtimes(needs) + .await + .into_iter() + .map(|runtime| { + let unique = unique(); + Case { + ext: runtime.ext, + namespace: runtime.namespace.unwrap_or(format!("ns-{unique}")), + stream: format!("s-{unique}"), + } + }) + .collect() +} + +async fn call( + ext: &StreamRuntime, + method: &str, + request: Req, +) -> Result { + let bytes = ext.call(method, &request.encode_to_vec()).await?; + Ok(Resp::decode(bytes.as_slice()).unwrap()) +} + +fn reason(error: &StreamError) -> String { + let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap(); + assert_eq!(details.reason(), error.reason.proto()); + error.reason.token().to_string() +} + +/// The writer of a write by `producer`, with no withdrawal facts. +fn writer_for(producer: Producer) -> Option { + Some(StreamWriter { + producer: Some(producer), + ..Default::default() + }) +} + +fn client(attempt: i32) -> Producer { + Producer { + id: client_producer_id("me", "p"), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), + } +} + +/// The Producer of run `run` in its first Workflow Task, as its close Activity sends it. +fn workflow(run: &str) -> Producer { + Producer { + id: crate::ids::workflow_producer_id(run), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: run.to_string(), + epoch: Some(WorkflowTaskEpoch::default()), + })), + } +} + +/// The fields of a create that uses an existing Stream. +fn use_existing() -> CreateStreamRequest { + CreateStreamRequest { + id_conflict_policy: StreamIdConflictPolicy::UseExisting as i32, + ..Default::default() + } +} + +fn record(text: &str) -> AppendRecord { + AppendRecord { + body: Some(Payload { + data: text.as_bytes().to_vec(), + ..Default::default() + }), + } +} + +fn millis(ms: u64) -> Option { + let duration = Duration::from_millis(ms); + Some(prost_types::Duration { + seconds: duration.as_secs() as i64, + nanos: duration.subsec_nanos() as i32, + }) +} + +impl Case { + fn stream_ref(&self) -> Option { + Some(StreamRef { + id: self.stream.clone(), + ..Default::default() + }) + } + + async fn create( + &self, + retention: Option, + ) -> Result { + self.create_with(CreateStreamRequest { + retention, + ..Default::default() + }) + .await + } + + /// Creates the Stream with the fields of `request`, in the case's namespace and ref. + async fn create_with( + &self, + request: CreateStreamRequest, + ) -> Result { + let request = CreateStreamRequest { + namespace: self.namespace.clone(), + stream_ref: self.stream_ref(), + ..request + }; + call(&self.ext, "CreateStream", request).await + } + + async fn close(&self, producer: Producer) -> Result { + let close = CloseStreamRequest { + namespace: self.namespace.clone(), + r#ref: self.stream_ref(), + writer: writer_for(producer), + ..Default::default() + }; + call(&self.ext, "CloseStream", close).await + } + + async fn append( + &self, + producer: Producer, + sequence: i64, + records: Vec, + ) -> Result { + call( + &self.ext, + "AppendRecords", + AppendRecordsRequest { + namespace: self.namespace.clone(), + r#ref: self.stream_ref(), + writer: writer_for(producer), + sequence, + records, + ..Default::default() + }, + ) + .await + } +} + +/// A create that uses a Stream that exists returns it with `created` false, whatever its +/// Retention. A Retention outside a proto Duration's range, or under 1 ms, fails. +async fn a_create_returns_the_stream_that_exists_whatever_its_retention() { + for case in cases_for(Needs::CALLS).await { + for (seconds, nanos) in [(i64::MAX, 0), (-1, 0), (0, 999_999)] { + let retention = Some(prost_types::Duration { seconds, nanos }); + let error = case.create(retention).await.unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } + assert!(case.create(None).await.unwrap().created); + for retention in [None, millis(1000)] { + let request = CreateStreamRequest { + retention, + ..use_existing() + }; + assert!(!case.create_with(request).await.unwrap().created); + } + } +} + +/// A create of an open Stream fails as `already_exists`, unless it uses the Stream or it is a +/// retry of the create that made the Stream. Such a retry has the same request id, and it gets +/// `created` true with either policy. A create needs a ref with an id. +async fn a_create_of_an_open_stream_fails_unless_it_uses_it_or_retries() { + for case in cases_for(Needs::CALLS).await { + let first = CreateStreamRequest { + request_id: "create-1".to_string(), + ..Default::default() + }; + assert!(case.create_with(first.clone()).await.unwrap().created); + let other = CreateStreamRequest { + request_id: "create-2".to_string(), + ..Default::default() + }; + for request in [other.clone(), CreateStreamRequest::default()] { + let error = case.create_with(request).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::AlreadyExists, "stream_already_exists") + ); + } + let used = CreateStreamRequest { + request_id: other.request_id, + ..use_existing() + }; + assert!(!case.create_with(used).await.unwrap().created); + let retry = CreateStreamRequest { + request_id: first.request_id.clone(), + ..use_existing() + }; + for request in [first, retry] { + assert!(case.create_with(request).await.unwrap().created); + } + for stream_ref in [None, Some(StreamRef::default())] { + let request = CreateStreamRequest { + namespace: case.namespace.clone(), + stream_ref, + ..Default::default() + }; + let error = call::<_, CreateStreamResponse>(&case.ext, "CreateStream", request) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } + } +} + +/// A create of a closed Stream fails as closed, before its conflict policy and before a retry of +/// the create that made the Stream. +async fn a_create_of_a_closed_stream_fails() { + for case in cases_for(Needs::CALLS).await { + let first = CreateStreamRequest { + request_id: "create-1".to_string(), + ..Default::default() + }; + case.create_with(first.clone()).await.unwrap(); + case.close(client(1)).await.unwrap(); + for request in [first, CreateStreamRequest::default()] { + let error = case.create_with(request).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::FailedPrecondition, "stream_closed") + ); + } + } +} + +/// A close writes no Record, so the describe's last offset stays at the newest Record. A second +/// close of the closed Stream succeeds, and a later append fails as closed. +async fn a_close_writes_no_record_and_a_second_close_succeeds() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + let written = case.append(client(1), 1, vec![record("a")]).await.unwrap(); + case.close(client(1)).await.unwrap(); + case.close(client(1)).await.unwrap(); + let error = case + .append(client(1), 2, vec![record("b")]) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_closed"); + let describe = DescribeStreamRequest { + namespace: case.namespace.clone(), + r#ref: case.stream_ref(), + }; + let described: DescribeStreamResponse = + call(&case.ext, "DescribeStream", describe).await.unwrap(); + assert_eq!(described.status, StreamStatus::Closed as i32); + assert_eq!(described.last_offset, written.last_offset); + } +} + +/// A Stream with no owner takes a close from a Workflow or a client. The close of an Activity +/// fails as `not_owner`. +async fn a_workflow_or_a_client_closes_a_stream_with_no_owner() { + for case in cases_for(Needs::UNCHECKED).await { + case.create(None).await.unwrap(); + let run = format!("{}-run", case.stream); + let activity = Producer { + id: activity_producer_id(&run, "a", "s"), + kind: Some(Kind::Activity(ActivityProducer { + run_id: run.clone(), + epoch: Some(ActivityEpoch { + attempt: 1, + ..Default::default() + }), + })), + }; + let error = case.close(activity).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::FailedPrecondition, "stream_not_owner") + ); + case.close(workflow(&run)).await.unwrap(); + } +} + +/// A create of a Stream whose Retention passed fails as expired while its Tombstone is there, not +/// as `already_exists`. +async fn a_create_of_an_expired_stream_fails() { + for case in cases_for(Needs::SHORT_TIMES).await { + case.create(millis(300)).await.unwrap(); + tokio::time::sleep(Duration::from_millis(400)).await; + let error = case.create(None).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::NotFound, "stream_expired") + ); + } +} + +/// A create with an owner needs a Workflow Producer and a ref of its run chain, and the answer's +/// ref has that chain. A create with another owner fails as `not_owner`, not as +/// `already_exists`, and no run withdraws itself. +async fn a_create_checks_its_owner_and_its_ref() { + for case in cases_for(Needs::UNCHECKED).await { + let run = format!("{}-run", case.stream); + let workflow = workflow(&run); + let task = WorkflowTaskEpoch { + started_event_id: 3, + ..Default::default() + }; + let local_activity = Producer { + id: crate::ids::local_activity_producer_id(&run, "call-1"), + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: run.clone(), + epoch: Some(LocalActivityEpoch { + task: Some(task), + attempt: 1, + }), + })), + }; + let chain = |workflow_id: &str| StreamRef { + id: case.stream.clone(), + owner: super::support::workflow_execution(workflow_id, "first-run"), + ..Default::default() + }; + let create = |owner: Option<&Producer>, stream_ref: StreamRef, withdraw_run: &str| { + CreateStreamRequest { + namespace: case.namespace.clone(), + owner: owner.cloned(), + stream_ref: Some(stream_ref), + withdraw_run: withdraw_run.to_string(), + ..Default::default() + } + }; + for request in [ + create(Some(&local_activity), chain("agent"), ""), + create(Some(&workflow), case.stream_ref().unwrap(), ""), + create( + Some(&workflow), + StreamRef { + id: String::new(), + ..chain("agent") + }, + "", + ), + create(Some(&workflow), chain("agent"), &run), + ] { + let error = call::<_, CreateStreamResponse>(&case.ext, "CreateStream", request) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } + let created: CreateStreamResponse = call( + &case.ext, + "CreateStream", + create(Some(&workflow), chain("agent"), ""), + ) + .await + .unwrap(); + assert_eq!(created.r#ref.unwrap().owner, chain("agent").owner); + for request in [ + create(Some(&workflow), chain("another-agent"), ""), + create(None, case.stream_ref().unwrap(), ""), + ] { + let error = call::<_, CreateStreamResponse>(&case.ext, "CreateStream", request) + .await + .unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::FailedPrecondition, "stream_not_owner") + ); + } + } +} + +/// A Producer's epoch has a case, its id must match that kind, a Workflow appends only through a +/// Local Activity, and a Local Activity's epoch names the Workflow Task that dispatched it. +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 = 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()), + (Code::InvalidArgument, "stream_invalid") + ); + } + let error = case.append(workflow("r1"), 1, vec![]).await.unwrap_err(); + assert_eq!(error.code(), Code::InvalidArgument); + let no_task = Producer { + id: crate::ids::local_activity_producer_id("r1", "call-1"), + 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")]) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } +} + +/// An append with no sequence takes the next one of its Producer's epoch, as a call with no writer +/// state sends it. Nothing dedupes it, so its retry writes the batch again. A negative sequence +/// fails. +async fn an_append_with_no_sequence_takes_the_next_one_of_its_epoch() { + for case in cases_for(Needs::CALLS).await { + case.create(None).await.unwrap(); + let first = case.append(client(1), 0, vec![record("a")]).await.unwrap(); + let retry = case.append(client(1), 0, vec![record("a")]).await.unwrap(); + assert!( + retry.first_offset > first.last_offset, + "{first:?} {retry:?}" + ); + let resend = case.append(client(1), 2, vec![record("a")]).await.unwrap(); + assert_eq!(resend, retry); + let error = case + .append(client(1), -1, vec![record("b")]) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } +} + +/// An unknown method fails with UNIMPLEMENTED. +async fn an_unknown_method_is_unimplemented() { + for case in cases_for(Needs::CALLS).await { + let error = case.ext.call("Nope", &[]).await.unwrap_err(); + assert_eq!(error.code(), Code::Unimplemented); + } +} + +/// The cases of the suite, in file order. +pub const CASES: &[super::Case] = &[ + case!(a_create_returns_the_stream_that_exists_whatever_its_retention), + case!(a_create_of_an_open_stream_fails_unless_it_uses_it_or_retries), + case!(a_create_of_a_closed_stream_fails), + case!(a_close_writes_no_record_and_a_second_close_succeeds), + case!(a_workflow_or_a_client_closes_a_stream_with_no_owner), + case!(a_create_of_an_expired_stream_fails), + case!(a_create_checks_its_owner_and_its_ref), + case!(a_producer_must_match_its_kind_and_not_be_a_workflow), + case!(an_append_with_no_sequence_takes_the_next_one_of_its_epoch), + case!(an_unknown_method_is_unimplemented), +]; diff --git a/crates/streams/src/testing/store.rs b/crates/streams/src/testing/store.rs index 29ade82d1..dc9288ec1 100644 --- a/crates/streams/src/testing/store.rs +++ b/crates/streams/src/testing/store.rs @@ -17,8 +17,8 @@ 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}, }; @@ -29,26 +29,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 }), + })), } } @@ -718,10 +717,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 +743,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()); } } @@ -856,7 +856,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), diff --git a/crates/streams/src/testing/support.rs b/crates/streams/src/testing/support.rs index c9386579b..fcaa45fec 100644 --- a/crates/streams/src/testing/support.rs +++ b/crates/streams/src/testing/support.rs @@ -5,14 +5,14 @@ use super::{RawKeys, backends}; use crate::{ - StreamResult, + StreamResult, StreamRuntime, digest::batch_digest, 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(), + }), + })), } } } @@ -185,11 +185,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() @@ -360,3 +363,38 @@ impl Target { .await } } + +/// One runtime under test, the namespace its cases use, and what its backend gives. +pub(crate) struct Runtime { + pub(crate) ext: Arc, + /// The namespace of a backend that has fixed ones, such as the server. `None` lets the case + /// make one of its own. + pub(crate) namespace: Option, + pub(crate) gives: Needs, +} + +/// A runtime on each backend under test that gives `needs`. +pub(crate) async fn runtimes(needs: Needs) -> Vec { + backends() + .into_iter() + .filter(|backend| backend.gives().gives(needs)) + .map(|backend| Runtime { + ext: backend.runtime(""), + namespace: None, + gives: backend.gives(), + }) + .collect() +} + +/// A runtime on each backend under test. +pub(crate) fn stream_runtimes() -> Vec> { + named_runtimes("") +} + +/// A runtime on each backend under test, whose refs carry `backend`. +pub(crate) fn named_runtimes(backend: &str) -> Vec> { + backends() + .into_iter() + .map(|test_backend| test_backend.runtime(backend)) + .collect() +} diff --git a/crates/streams/tests/capabilities.rs b/crates/streams/tests/capabilities.rs new file mode 100644 index 000000000..34212ccc9 --- /dev/null +++ b/crates/streams/tests/capabilities.rs @@ -0,0 +1,295 @@ +//! A Store that lacks a capability refuses every call that needs it, by name, and writes nothing. +//! The tests use the memory Store with one capability masked. + +use prost::Message; +use std::{ + sync::Arc, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; +use temporalio_common::protos::temporal::api::{common::v1::Execution, enums::v1::ExecutionType}; +use temporalio_streams::{ + MemoryStreamRuntime, StreamError, StreamResult, StreamRuntime, + ids::client_producer_id, + memory::MemoryStore, + proto::{ + 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}, +}; +use tokio::time::Instant; +use tonic::Code; + +/// A memory Store that lacks the capabilities in `without`, as a Store with fewer would. +struct Masked { + inner: Arc, + without: Capabilities, +} + +#[async_trait::async_trait] +impl Store for Masked { + fn capabilities(&self) -> Capabilities { + self.inner.capabilities().without(self.without) + } + + async fn create( + &self, + namespace: &str, + stream_ref: &StreamRef, + create: &Create<'_>, + ) -> StreamResult { + self.inner.create(namespace, stream_ref, create).await + } + + async fn append( + &self, + namespace: &str, + stream_ref: &StreamRef, + batch: &Batch<'_>, + ) -> StreamResult { + self.inner.append(namespace, stream_ref, batch).await + } + + async fn read( + &self, + namespace: &str, + stream_id: &str, + from: i64, + count: usize, + ) -> StreamResult<(StreamState, Vec)> { + self.inner.read(namespace, stream_id, from, count).await + } + + async fn wait_after( + &self, + namespace: &str, + stream_id: &str, + after: i64, + deadline: Instant, + ) -> StreamResult { + self.inner + .wait_after(namespace, stream_id, after, deadline) + .await + } + + // The masked calls still work underneath, so a call that skips the check would succeed, + // and the tests would see it. + async fn close( + &self, + namespace: &str, + stream_ref: &StreamRef, + producer: &Producer, + ) -> StreamResult { + self.inner.close(namespace, stream_ref, producer).await + } +} + +const NAMESPACE: &str = "ns"; + +struct Case { + ext: Arc, + memory: Arc, + run_id: String, + stream: String, +} + +fn case(without: Capabilities) -> Case { + let memory = Arc::new(MemoryStore::new( + Duration::from_secs(3600), + Duration::from_secs(7200), + )); + let store = Arc::new(Masked { + inner: memory.clone(), + without, + }); + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let run_id = format!("run-{unique}"); + Case { + ext: MemoryStreamRuntime::with_store(store, None), + memory, + stream: format!("wf:{run_id}"), + run_id, + } +} + +fn client() -> Producer { + Producer { + id: client_producer_id("me", "p"), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: 1 }), + })), + } +} + +/// The writer of a client's write, with no withdrawal facts. +fn client_writer() -> Option { + Some(StreamWriter { + producer: Some(client()), + ..Default::default() + }) +} + +/// The capability an error names, after it checked the code and the reason. +fn refused_capability(error: &StreamError) -> String { + assert_eq!(error.code(), Code::Unimplemented, "{error:?}"); + let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap(); + assert_eq!(details.reason(), StreamErrorReason::Unsupported); + error + .message + .rsplit(' ') + .next() + .unwrap_or_default() + .to_string() +} + +impl Case { + fn stream_ref(&self) -> Option { + Some(StreamRef { + id: self.stream.clone(), + ..Default::default() + }) + } + + async fn call(&self, method: &str, request: Req) -> Result, StreamError> { + self.ext.call(method, &request.encode_to_vec()).await + } + + async fn create(&self) { + let request = CreateStreamRequest { + namespace: NAMESPACE.to_string(), + stream_ref: self.stream_ref(), + ..Default::default() + }; + self.call("CreateStream", request).await.unwrap(); + } + + fn log_len(&self) -> usize { + self.memory.test_log_len(NAMESPACE, &self.stream) + } + + fn created(&self) -> bool { + self.memory + .test_meta(NAMESPACE, &self.stream, b"v") + .is_some() + } + + /// A create from a Workflow's call outside Workflow code, which names the owning run. + async fn create_owned(&self, withdraw_run: &str) -> Result, StreamError> { + let request = CreateStreamRequest { + namespace: NAMESPACE.to_string(), + withdraw_run: withdraw_run.to_string(), + owner: Some(Producer { + id: temporalio_streams::ids::workflow_producer_id(&self.run_id), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: self.run_id.clone(), + epoch: Some(WorkflowTaskEpoch::default()), + })), + }), + stream_ref: Some(StreamRef { + id: self.stream.clone(), + owner: Some(Execution { + r#type: ExecutionType::Workflow as i32, + business_id: "agent".to_string(), + run_id: self.run_id.clone(), + }), + ..Default::default() + }), + ..Default::default() + }; + self.call("CreateStream", request).await + } +} + +#[tokio::test] +async fn an_append_without_the_fence_or_the_flag_is_refused_with_nothing_written() { + for (capability, name) in [ + (Capabilities::PRODUCER_FENCE, "producer_fence"), + (Capabilities::WITHDRAWAL, "withdrawal"), + ] { + let case = case(capability); + case.create().await; + let request = AppendRecordsRequest { + namespace: NAMESPACE.to_string(), + r#ref: case.stream_ref(), + writer: client_writer(), + sequence: 1, + records: vec![AppendRecord::default()], + ..Default::default() + }; + let error = case.call("AppendRecords", request).await.unwrap_err(); + assert_eq!(refused_capability(&error), name); + assert_eq!(case.log_len(), 0); + } +} + +#[tokio::test] +async fn an_append_with_no_sequence_without_the_capability_is_refused_with_nothing_written() { + let case = case(Capabilities::NEXT_SEQUENCE); + case.create().await; + let request = AppendRecordsRequest { + namespace: NAMESPACE.to_string(), + r#ref: case.stream_ref(), + writer: client_writer(), + records: vec![AppendRecord::default()], + ..Default::default() + }; + let error = case + .call("AppendRecords", request.clone()) + .await + .unwrap_err(); + assert_eq!(refused_capability(&error), "next_sequence"); + assert_eq!(case.log_len(), 0); + // A batch with its own sequence needs no number from the Store. + let numbered = AppendRecordsRequest { + sequence: 1, + ..request + }; + case.call("AppendRecords", numbered).await.unwrap(); + assert_eq!(case.log_len(), 1); +} + +#[tokio::test] +async fn a_close_without_the_capability_is_refused_and_the_stream_stays_open() { + let case = case(Capabilities::CLOSE); + case.create().await; + let request = CloseStreamRequest { + namespace: NAMESPACE.to_string(), + r#ref: case.stream_ref(), + writer: client_writer(), + ..Default::default() + }; + let error = case.call("CloseStream", request).await.unwrap_err(); + assert_eq!(refused_capability(&error), "close"); + assert_eq!(case.log_len(), 0); + let describe = DescribeStreamRequest { + namespace: NAMESPACE.to_string(), + r#ref: case.stream_ref(), + }; + let bytes = case.call("DescribeStream", describe).await.unwrap(); + let described = DescribeStreamResponse::decode(bytes.as_slice()).unwrap(); + assert_eq!(described.status, StreamStatus::Open as i32); +} + +#[tokio::test] +async fn a_workflow_create_without_the_safety_ttl_is_refused_before_it_writes() { + let case = case(Capabilities::SAFETY_TTL); + let error = case.create_owned("").await.unwrap_err(); + assert_eq!(refused_capability(&error), "safety_ttl"); + assert!(!case.created()); +} + +#[tokio::test] +async fn a_retry_run_without_the_run_fence_is_refused_before_it_writes() { + let case = case(Capabilities::RUN_FENCE); + let error = case.create_owned("previous-run").await.unwrap_err(); + assert_eq!(refused_capability(&error), "run_fence"); + assert!(!case.created()); + // A run that isn't a retry needs no run fence. + case.create_owned("").await.unwrap(); + assert!(case.created()); +} 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() })),