diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index a98b687c8..92a57f77e 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -9,17 +9,18 @@ use crate::{ owner::{OwnerClient, OwnerDescription, OwnerError}, proto::{ AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, LatestRequest, - LatestResponse, ReadRequest, ReadResponse, StoreAppendRequest, StoreLatestRequest, - StreamAddress, StreamOwnerKind, append_request::Producer, + LatestResponse, PromoteOutcome, ReadRequest, ReadResponse, StageRef, StoreAppendRequest, + StoreLatestRequest, StreamAddress, StreamOwnerKind, append_request::Producer, }, reader::{ReadState, read_with}, + repair::{Decision, decide_token}, stored_append_record, stream_hash, }; use prost::Message; use std::{ collections::HashMap, sync::{Arc, Mutex}, - time::Duration, + time::{Duration, SystemTime}, }; use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; use tokio::time::Instant; @@ -27,6 +28,17 @@ use tokio::time::Instant; /// The longest a producer writes without asking whether its owner's chain ended. pub const OWNER_RECHECK: Duration = Duration::from_secs(60); +/// Whether a repair takes a stage History couldn't decide lately as still undecided. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Recent { + Trust, + /// The owner ended or the stream closed, so History may hold the answer now. + Ask, +} + +/// How long a stage History couldn't decide is taken as still undecided without asking again. +const UNDECIDED_FOR: Duration = Duration::from_secs(5); + /// How the stream layer behaves. #[derive(Debug, Clone, PartialEq, Eq)] pub struct StreamsOptions { @@ -127,6 +139,9 @@ pub struct Streams { options: StreamsOptions, producers: Remembered, chains: Remembered, + /// Stages History couldn't decide lately, by token. Each new read would otherwise walk the + /// staging run's History back to the floor again. + undecided: Remembered, } impl std::fmt::Debug for Streams { @@ -149,6 +164,7 @@ impl Streams { store, owner, producers: Remembered::new(options.remembered), + undecided: Remembered::new(options.remembered), chains: Remembered::new(options.remembered), options, } @@ -249,7 +265,8 @@ impl Streams { pub async fn read(&self, request: ReadRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; let mut state = ReadState::decode_from(&request.state)?; - if state.first_run_id.is_empty() { + let starting = state.first_run_id.is_empty(); + if starting { state.first_run_id = self.chain(stream).await?.first_run_id; } let target = ReadTarget { @@ -257,6 +274,9 @@ impl Streams { topic: stream.topic.clone(), stream_hash: self.hash(stream), }; + if starting { + self.repair(&target.chain, Recent::Trust).await; + } let wait = request .wait .and_then(|wait| Duration::try_from(wait).ok()) @@ -290,13 +310,18 @@ impl Streams { break page; } if closed && !state.owner_ended { - // One more pass after learning of the close, as after learning the owner ended. + // One more pass after learning of the close, as after learning the owner ended, + // with output a Worker committed but never promoted. state.owner_ended = true; + self.repair(&target.chain, Recent::Ask).await; continue; } if state.owner_ended { // A close doesn't prove the chain's output is visible, since a promotion still // lands after it. The read ends once nothing of the chain is pending. + if awaiting_stage { + self.repair(&target.chain, Recent::Trust).await; + } if self.store.pending_stages(&target.chain).await?.is_empty() { page.done = true; break page; @@ -313,8 +338,10 @@ impl Streams { .clamp(millis(minimum), millis(self.options.owner_check_max)); state.next_check_ms = unix_ms() + state.check_interval_ms; // One more pass after learning the owner ended, so a record that landed - // between the read and the describe is delivered. + // between the read and the describe is delivered, and so is output a Worker + // committed but never promoted. if state.owner_ended { + self.repair(&target.chain, Recent::Ask).await; continue; } } @@ -328,6 +355,119 @@ impl Streams { }) } + /// Settles the stages a Worker left behind on `chain`, as History decides them. Best effort: + /// a stage History has not decided stays for the next reader, and a failure only logs. + async fn repair(&self, chain: &ChainId, recent: Recent) { + if let Err(error) = self.try_repair(chain, recent).await { + tracing::warn!( + workflow_id = %chain.workflow_id, + "Could not settle the pending stream stages of the Workflow: {error}" + ); + } + } + + async fn try_repair(&self, chain: &ChainId, recent: Recent) -> StreamResult<()> { + let mut pending = self.store.pending_stages(chain).await?; + let mut started: HashMap> = HashMap::new(); + for stage in &pending { + if started.contains_key(&stage.run_id) { + continue; + } + let start = match self + .owner + .describe(&chain.namespace, &chain.workflow_id, &stage.run_id) + .await + { + Ok(described) => Some(described.start_time.unwrap_or(SystemTime::UNIX_EPOCH)), + Err(OwnerError::NotFound(_)) => None, + Err(error) => return Err(describe_failed(&error)), + }; + started.insert(stage.run_id.clone(), start); + } + // Promotions append, so they go in commit order: by run, then by floor within the run. + pending.sort_by(|a, b| { + (started[&a.run_id], &a.run_id, a.history_floor_event_id).cmp(&( + started[&b.run_id], + &b.run_id, + b.history_floor_event_id, + )) + }); + for stage in pending { + if started[&stage.run_id].is_none() { + // Retention may have removed the run, or a replica behind a failover may not have + // it yet. Aborting could drop committed output, so the stage waits for its expiry + // and holds back nothing. + tracing::debug!( + token = %stage.token, + run_id = %stage.run_id, + "Leaving a stream stage whose run is not in History" + ); + continue; + } + if recent == Recent::Trust + && self + .undecided + .get(&stage.token) + .is_some_and(|checked| checked.at.elapsed() < UNDECIDED_FOR) + { + return Ok(()); + } + let events = match self + .owner + .history_after( + &chain.namespace, + &chain.workflow_id, + &stage.run_id, + stage.history_floor_event_id, + ) + .await + { + Ok(events) => events, + Err(OwnerError::NotFound(_)) => continue, + Err(error) => return Err(describe_failed(&error)), + }; + let reference = StageRef { + chain: Some(chain.clone()), + token: stage.token.clone(), + topics: stage.topics, + }; + // A reader never holds an attempt in flight, so the same-floor rule always holds. + let settled = + match decide_token(&events, &stage.token, stage.history_floor_event_id, true) { + Decision::Promote => self.store.promote(&reference).await.map(|promoted| { + if promoted.outcome == PromoteOutcome::Lost as i32 { + tracing::warn!( + token = %stage.token, + workflow_id = %chain.workflow_id, + "Stream output the Workflow committed was dropped by retention \ + before it was promoted" + ); + } + }), + Decision::Abort => self.store.abort(&reference).await, + // A stage History has not decided holds back the ones after it. + Decision::Unknown => { + self.undecided.put(stage.token.clone(), "", UNDECIDED_FOR); + return Ok(()); + } + }; + match settled { + Ok(()) => {} + // A stage the store refuses for good would hold back every later stage on every + // read, so it is left to its expiry and the repair goes on. + Err(error) if error.kind == crate::proto::StreamFailureKind::Refused => { + tracing::warn!( + token = %stage.token, + workflow_id = %chain.workflow_id, + "Leaving a stream stage the store refuses to settle: {error}" + ) + } + Err(error) => return Err(error), + } + } + Ok(()) + } + /// Whether the owner a read follows ended. Following the chain, a run that continued as /// new is not the end, and a new chain on the Workflow id is. async fn owner_ended(&self, stream: &StreamAddress, chain: &ChainId) -> StreamResult { @@ -586,26 +726,40 @@ mod tests { StreamRecordKind, }, }; + use crate::{ + proto::{StagedBatch, StagedRecord}, + repair::tests::{event, marker}, + stored_output_record, + }; use sha2::{Digest, Sha256}; use std::sync::atomic::{AtomicUsize, Ordering}; use temporalio_common::protos::temporal::api::{ - common::v1::Payload, enums::v1::WorkflowExecutionStatus, + common::v1::Payload, + enums::v1::{EventType, WorkflowExecutionStatus}, + history::v1::HistoryEvent, }; - /// Temporal as the tests need it: the latest run of each Workflow id, and pinned runs. + /// Temporal as the tests need it: the latest run of each Workflow id, pinned runs, and the + /// History of each run. #[derive(Default)] struct Owners { latest: Mutex>>, runs: Mutex>, + histories: Mutex>>, + failing_runs: Mutex>, + described_runs: Mutex>, describes: AtomicUsize, + history_reads: AtomicUsize, } impl Owners { fn run(&self, workflow_id: &str, run_id: &str, first_run_id: &str) { + let started = self.runs.lock().unwrap().len() as u64; let described = OwnerDescription { run_id: run_id.to_string(), first_run_id: first_run_id.to_string(), status: WorkflowExecutionStatus::Running, + start_time: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(started)), }; self.runs .lock() @@ -645,6 +799,16 @@ mod tests { ) -> Result { self.describes.fetch_add(1, Ordering::Relaxed); if !run_id.is_empty() { + self.described_runs.lock().unwrap().push(run_id.to_string()); + if self + .failing_runs + .lock() + .unwrap() + .iter() + .any(|run| run == run_id) + { + return Err(OwnerError::Failed("unavailable".to_string())); + } return self .runs .lock() @@ -660,6 +824,25 @@ mod tests { .cloned() .unwrap_or_else(|| Err(OwnerError::NotFound(workflow_id.to_string()))) } + + async fn history_after( + &self, + _namespace: &str, + _workflow_id: &str, + run_id: &str, + floor: i64, + ) -> Result, OwnerError> { + self.history_reads.fetch_add(1, Ordering::Relaxed); + let histories = self.histories.lock().unwrap(); + let events = histories + .get(run_id) + .ok_or_else(|| OwnerError::NotFound(run_id.to_string()))?; + Ok(events + .iter() + .filter(|event| event.event_id > floor) + .cloned() + .collect()) + } } struct Setup { @@ -1027,6 +1210,7 @@ mod tests { run_id: String::new(), first_run_id: String::new(), status, + start_time: None, } .chain_ended() }; @@ -1341,6 +1525,238 @@ mod tests { assert_eq!(setup.owners.describes(), before); } + /// A stage of `run_id` at `floor` that a Worker left behind, holding `value` on `out`. + async fn left_behind(store: &MemoryStore, run_id: &str, token: &str, floor: i64, value: &str) { + let record = data(value); + store + .stage(StagedBatch { + chain: Some(chain("run-1")), + run_id: run_id.to_string(), + token: token.to_string(), + history_floor_event_id: floor, + records: vec![StagedRecord { + topic: "out".to_string(), + record: stored_output_record( + "out", + record.kind, + record.body, + &record.content_hash, + run_id, + ) + .unwrap() + .encode_to_vec(), + }], + }) + .await + .unwrap(); + } + + fn history(owners: &Owners, run_id: &str, events: Vec) { + owners + .histories + .lock() + .unwrap() + .insert(run_id.to_string(), events); + } + + fn values(page: &ReadResponse) -> Vec { + page.records + .iter() + .filter_map(|record| match record.record.as_ref() { + Some(crate::proto::read_record::Record::Stored(stored)) => { + Some(String::from_utf8(stored.body.clone().unwrap_or_default().data).unwrap()) + } + _ => None, + }) + .collect() + } + + async fn pending(store: &MemoryStore) -> Vec { + store + .pending_stages(&chain("run-1")) + .await + .unwrap() + .into_iter() + .map(|stage| stage.token) + .collect() + } + + #[tokio::test] + async fn a_reader_promotes_output_a_stopped_worker_committed() { + // The run finished, so no Worker will ever replay it, and only a reader finds the stage. + let setup = setup(quick()); + left_behind(&setup.store, "run-1", "t1", 3, "1").await; + history( + &setup.owners, + "run-1", + vec![ + event(4, EventType::WorkflowTaskCompleted), + marker(5, "t1", 3), + ], + ); + let page = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(values(&page), ["1"]); + assert!(pending(&setup.store).await.is_empty()); + // A second reader finds nothing left to repair and sees the record once. + let again = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(values(&again), ["1"]); + } + + #[tokio::test] + async fn rule_24_1_a_reader_aborts_a_stage_a_later_attempt_superseded() { + // A transient attempt's failure is not in History, but the attempt that completed + // carries every commit made at that floor. + let setup = setup(quick()); + left_behind(&setup.store, "run-1", "orphan", 3, "0").await; + left_behind(&setup.store, "run-1", "t1", 3, "1").await; + history( + &setup.owners, + "run-1", + vec![ + event(4, EventType::WorkflowTaskCompleted), + marker(5, "t1", 3), + ], + ); + let page = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(values(&page), ["1"]); + assert!(pending(&setup.store).await.is_empty()); + } + + #[tokio::test] + async fn rule_24_3_a_reader_repairs_pending_stages_in_commit_order() { + let setup = setup(quick()); + setup.owners.run("wf", "run-2", "run-1"); + // A store keeps no order a reader can rely on, so the stages are left in reverse. + left_behind(&setup.store, "run-2", "t4", 3, "4").await; + left_behind(&setup.store, "run-1", "t3", 9, "3").await; + left_behind(&setup.store, "run-1", "t2", 6, "2").await; + left_behind(&setup.store, "run-1", "t1", 3, "1").await; + history( + &setup.owners, + "run-1", + vec![marker(5, "t1", 3), marker(8, "t2", 6), marker(11, "t3", 9)], + ); + history(&setup.owners, "run-2", vec![marker(5, "t4", 3)]); + let page = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(values(&page), ["1", "2", "3", "4"]); + // One describe per run, however many stages it left. + let mut described = setup.owners.described_runs.lock().unwrap().clone(); + described.sort(); + assert_eq!(described, ["run-1", "run-2"]); + } + + #[tokio::test] + async fn rule_24_3_a_stage_history_has_not_decided_holds_back_the_ones_after_it() { + let setup = setup(quick()); + left_behind(&setup.store, "run-1", "t1", 3, "1").await; + left_behind(&setup.store, "run-1", "t2", 6, "2").await; + // A Local Activity may hold the first task open, so neither may land yet. + history( + &setup.owners, + "run-1", + vec![ + event(4, EventType::WorkflowTaskCompleted), + marker(8, "t2", 6), + ], + ); + let page = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert!(page.records.is_empty()); + assert_eq!(pending(&setup.store).await.len(), 2); + } + + #[tokio::test] + async fn rule_24_4_a_reader_repairs_past_a_stage_whose_run_is_gone() { + let setup = setup(quick()); + left_behind(&setup.store, "gone", "lost", 3, "0").await; + left_behind(&setup.store, "run-1", "t1", 3, "1").await; + history(&setup.owners, "run-1", vec![marker(5, "t1", 3)]); + let page = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(values(&page), ["1"]); + // Not found can mean a replica behind a failover, so the stage waits for its expiry. + assert_eq!(pending(&setup.store).await, ["lost"]); + } + + #[tokio::test] + async fn a_failed_repair_leaves_the_read_running() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + left_behind(&setup.store, "run-x", "t1", 3, "0").await; + setup + .owners + .failing_runs + .lock() + .unwrap() + .push("run-x".to_string()); + let page = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(values(&page), ["1"]); + assert_eq!(pending(&setup.store).await, ["t1"]); + } + + #[tokio::test] + async fn a_read_repairs_again_when_its_owner_ends() { + let setup = setup(quick()); + left_behind(&setup.store, "run-1", "t1", 3, "1").await; + history( + &setup.owners, + "run-1", + vec![event(4, EventType::WorkflowTaskStarted)], + ); + let first = setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert!(first.records.is_empty()); + // The Worker committed and stopped before promoting, then the run finished. + history( + &setup.owners, + "run-1", + vec![ + marker(5, "t1", 3), + event(9, EventType::WorkflowExecutionCompleted), + ], + ); + setup.owners.end("wf", WorkflowExecutionStatus::Completed); + let page = setup + .streams + .read(read_after( + &first.cursor, + &first.state, + Duration::from_secs(5), + )) + .await + .unwrap(); + assert_eq!(values(&page), ["1"]); + } + /// The digest the Python SDK's Redis provider took over `[{"n": 1}, {"n": 2}]` from producer /// `p`, attempt 1, sequence 1. const PYTHON_DIGEST: &str = "0499768af1856bc6fd19cf6cf90e28e971099f1680e8347d70bdb4b65273f30f"; @@ -1399,4 +1815,122 @@ mod tests { } assert_eq!(setup.owners.describes(), 0); } + + /// A memory store that refuses for good to promote one stage. + struct Refusing { + inner: MemoryStore, + token: String, + } + + #[async_trait::async_trait] + impl StreamStore for Refusing { + fn name(&self) -> &str { + self.inner.name() + } + async fn append( + &self, + r: crate::proto::StoreAppendRequest, + ) -> StreamResult { + self.inner.append(r).await + } + async fn read( + &self, + r: crate::proto::StoreReadRequest, + ) -> StreamResult { + self.inner.read(r).await + } + async fn latest( + &self, + r: crate::proto::StoreLatestRequest, + ) -> StreamResult { + self.inner.latest(r).await + } + async fn record_at(&self, c: &ChainId, t: &str, p: &str) -> StreamResult>> { + self.inner.record_at(c, t, p).await + } + async fn stage(&self, b: StagedBatch) -> StreamResult<()> { + self.inner.stage(b).await + } + async fn promote(&self, s: &StageRef) -> StreamResult { + if s.token == self.token { + return Err(StreamError::refused( + "the stage names a topic the store can't key", + )); + } + self.inner.promote(s).await + } + async fn abort(&self, s: &StageRef) -> StreamResult<()> { + self.inner.abort(s).await + } + async fn close_chain(&self, c: &ChainId) -> StreamResult<()> { + self.inner.close_chain(c).await + } + async fn close_topic(&self, c: &ChainId, t: &str) -> StreamResult<()> { + self.inner.close_topic(c, t).await + } + async fn pending_stages( + &self, + c: &ChainId, + ) -> StreamResult> { + self.inner.pending_stages(c).await + } + async fn delete_owner( + &self, + r: crate::proto::DeleteOwnerRequest, + ) -> StreamResult { + self.inner.delete_owner(r).await + } + } + + #[tokio::test] + async fn rule_12_2_repair_goes_past_a_stage_the_store_refuses_for_good() { + let store = Arc::new(Refusing { + inner: MemoryStore::new(), + token: "bad".to_string(), + }); + let owners = Arc::new(Owners::default()); + owners.run("wf", "run-1", "run-1"); + let streams = Streams::new(store.clone(), owners.clone(), quick()); + left_behind(&store.inner, "run-1", "bad", 3, "0").await; + left_behind(&store.inner, "run-1", "t1", 6, "1").await; + history( + &owners, + "run-1", + vec![marker(5, "bad", 3), marker(8, "t1", 6)], + ); + let page = streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + assert_eq!(values(&page), ["1"]); + let left: Vec<_> = store + .inner + .pending_stages(&chain("run-1")) + .await + .unwrap() + .into_iter() + .map(|stage| stage.token) + .collect(); + assert_eq!(left, ["bad"]); + } + + #[tokio::test] + async fn rule_12_3_an_undecided_stage_is_not_read_again_by_every_new_reader() { + // A long Local Activity holds the task open, so History can't decide the stage yet. + let setup = setup(quick()); + left_behind(&setup.store, "run-1", "t1", 3, "1").await; + history( + &setup.owners, + "run-1", + vec![event(4, EventType::WorkflowTaskStarted)], + ); + for _ in 0..3 { + setup + .streams + .read(read_after("", &[], Duration::ZERO)) + .await + .unwrap(); + } + assert_eq!(setup.owners.history_reads.load(Ordering::Relaxed), 1); + } } diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 918c32c6a..544ea0f8d 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -18,6 +18,7 @@ mod reader; mod record; #[cfg(feature = "redis")] mod redis; +mod repair; mod store; #[cfg(feature = "redis")] @@ -32,6 +33,7 @@ pub use record::{ WORKFLOW_OWNER_KIND, activity_producer_id, check_append_digest, stored_append_record, stored_output_record, }; +pub use repair::{Decision, decide_token}; pub use store::StreamStore; /// The protos stores and lang exchange. diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 220268ad9..369173735 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -368,7 +368,7 @@ impl StreamStore for MemoryStore { .iter() .find(|record| !stage.topics.contains(&record.topic)) { - return Err(StreamError::storage(format!( + return Err(StreamError::refused(format!( "the stage holds topic {:?}, which the promotion did not name", stray.topic ))); diff --git a/crates/streams/src/owner.rs b/crates/streams/src/owner.rs index 37bd22ec4..04d5cd553 100644 --- a/crates/streams/src/owner.rs +++ b/crates/streams/src/owner.rs @@ -4,7 +4,10 @@ //! keyed by, to refuse a producer once the chain ended, and to end a read. Core implements this //! over its client, and tests use a fake. -use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; +use std::time::SystemTime; +use temporalio_common::protos::temporal::api::{ + enums::v1::WorkflowExecutionStatus, history::v1::HistoryEvent, +}; /// One run of a Workflow, as Temporal describes it. #[derive(Debug, Clone, PartialEq, Eq)] @@ -15,6 +18,8 @@ pub struct OwnerDescription { pub first_run_id: String, /// How the run stands. pub status: WorkflowExecutionStatus, + /// When the run started. Runs of one chain commit in the order they started. + pub start_time: Option, } impl OwnerDescription { @@ -54,4 +59,14 @@ pub trait OwnerClient: Send + Sync { workflow_id: &str, run_id: &str, ) -> Result; + + /// Run `run_id`'s History events after event `floor`, in order. Implementations read from + /// the end, since the events a repair needs are the newest. + async fn history_after( + &self, + namespace: &str, + workflow_id: &str, + run_id: &str, + floor: i64, + ) -> Result, OwnerError>; } diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index a515416b0..46be20617 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -529,7 +529,7 @@ impl StreamStore for RedisStore { .invoke_async(&mut self.shared.clone()) .await .map_err(|error| match error.code() { - Some("STREAMS_TOPIC") => StreamError::storage(error.detail().unwrap_or_default()), + Some("STREAMS_TOPIC") => StreamError::refused(error.detail().unwrap_or_default()), _ => errors::stream_error(&error, Call::Write), })?; let (outcome, records) = match added { diff --git a/crates/streams/src/repair.rs b/crates/streams/src/repair.rs new file mode 100644 index 000000000..12ba55c17 --- /dev/null +++ b/crates/streams/src/repair.rs @@ -0,0 +1,202 @@ +//! Settling stages a Worker left behind, as History decides them. +//! +//! A Worker that stops after a Workflow Task's commit but before the promotion leaves the stage +//! pending, and a run that finished is never replayed to promote it. A reader looks for such +//! stages and reads the staging run's History to promote or abort each one. + +use temporalio_common::protos::{ + constants::EXTERNAL_STREAM_MARKER_NAME, + coresdk::external_data::{ExternalOutputStreamManifest, extract_external_stream_marker_data}, + temporal::api::{ + enums::v1::EventType, + history::v1::{HistoryEvent, history_event::Attributes}, + }, +}; + +/// What History says about one stage. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Decision { + /// History holds the marker that names the stage, so its Workflow Task committed. + Promote, + /// The stage's Workflow Task never committed, and nothing can commit it now. + Abort, + /// History doesn't tell yet, for example while a Local Activity holds the task open. + Unknown, +} + +fn output(event: &HistoryEvent) -> Option { + let Some(Attributes::MarkerRecordedEventAttributes(marker)) = &event.attributes else { + return None; + }; + if marker.marker_name != EXTERNAL_STREAM_MARKER_NAME { + return None; + } + extract_external_stream_marker_data(&marker.details)?.output +} + +fn run_closed(event: &HistoryEvent) -> bool { + matches!( + event.event_type(), + EventType::WorkflowExecutionCompleted + | EventType::WorkflowExecutionFailed + | EventType::WorkflowExecutionTimedOut + | EventType::WorkflowExecutionCanceled + | EventType::WorkflowExecutionTerminated + | EventType::WorkflowExecutionContinuedAsNew + ) +} + +/// What History says about the stage `token`. +/// +/// `events` are the staging run's events after the stage's History floor, in order. A marker +/// that names the token proves the commit. Otherwise a closed run, or a failed or timed out +/// result for the Workflow Task after the floor, proves the completion was dropped. Anything +/// else is not known yet. +/// +/// `survived_eviction` turns on the same-floor rule. Several commits of one task attempt share a +/// floor, but never straddle an eviction, and a failed attempt always ends in one, since Core +/// evicts a run whose Workflow Task failed. So a stage from before an eviction whose floor +/// another stage committed at belongs to a failed attempt, even a transient one History never +/// records. A Worker sets it for a stage held across an eviction, since its own attempt may still +/// be committing. A reader sets it always, since it never holds an attempt in flight. +pub fn decide_token( + events: &[HistoryEvent], + token: &str, + history_floor_event_id: i64, + survived_eviction: bool, +) -> Decision { + let outputs: Vec<_> = events.iter().filter_map(output).collect(); + if outputs.iter().any(|output| output.stage_token == token) { + return Decision::Promote; + } + // The run is over and never committed this stage, so nothing can. + if events.iter().any(run_closed) { + return Decision::Abort; + } + if survived_eviction + && outputs + .iter() + .any(|output| output.history_floor_event_id == history_floor_event_id) + { + return Decision::Abort; + } + let result = events.iter().find(|event| { + event.event_id > history_floor_event_id + && matches!( + event.event_type(), + EventType::WorkflowTaskCompleted + | EventType::WorkflowTaskFailed + | EventType::WorkflowTaskTimedOut + ) + }); + match result.map(HistoryEvent::event_type) { + Some(EventType::WorkflowTaskFailed | EventType::WorkflowTaskTimedOut) => Decision::Abort, + _ => Decision::Unknown, + } +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + use temporalio_common::protos::{ + coresdk::external_data::{ExternalStreamMarkerData, build_external_stream_marker_details}, + temporal::api::history::v1::MarkerRecordedEventAttributes, + }; + + pub(crate) fn event(event_id: i64, event_type: EventType) -> HistoryEvent { + HistoryEvent { + event_id, + event_type: event_type as i32, + ..Default::default() + } + } + + pub(crate) fn marker(event_id: i64, token: &str, floor: i64) -> HistoryEvent { + let data = ExternalStreamMarkerData { + schema_version: 1, + output: Some(ExternalOutputStreamManifest { + stage_token: token.to_string(), + history_floor_event_id: floor, + ..Default::default() + }), + ..Default::default() + }; + HistoryEvent { + event_id, + event_type: EventType::MarkerRecorded as i32, + attributes: Some(Attributes::MarkerRecordedEventAttributes( + MarkerRecordedEventAttributes { + marker_name: EXTERNAL_STREAM_MARKER_NAME.to_string(), + details: build_external_stream_marker_details(&data), + ..Default::default() + }, + )), + ..Default::default() + } + } + + #[test] + fn a_marker_naming_the_stage_proves_the_commit() { + let events = [ + event(4, EventType::WorkflowTaskCompleted), + marker(5, "t1", 3), + ]; + assert_eq!(decide_token(&events, "t1", 3, false), Decision::Promote); + assert_eq!(decide_token(&events, "t2", 3, false), Decision::Unknown); + } + + #[test] + fn a_marker_of_another_kind_proves_nothing() { + let mut other = marker(5, "t1", 3); + if let Some(Attributes::MarkerRecordedEventAttributes(attributes)) = &mut other.attributes { + attributes.marker_name = "core_local_activity".to_string(); + } + assert_eq!(decide_token(&[other], "t1", 3, false), Decision::Unknown); + } + + #[test] + fn a_closed_run_that_never_committed_the_stage_aborts_it() { + for closed in [ + EventType::WorkflowExecutionCompleted, + EventType::WorkflowExecutionFailed, + EventType::WorkflowExecutionTimedOut, + EventType::WorkflowExecutionCanceled, + EventType::WorkflowExecutionTerminated, + EventType::WorkflowExecutionContinuedAsNew, + ] { + let events = [event(9, closed)]; + assert_eq!(decide_token(&events, "t1", 3, false), Decision::Abort); + } + } + + #[test] + fn the_task_result_after_the_floor_decides_a_dropped_completion() { + for failed in [ + EventType::WorkflowTaskFailed, + EventType::WorkflowTaskTimedOut, + ] { + let events = [event(4, EventType::WorkflowTaskStarted), event(5, failed)]; + assert_eq!(decide_token(&events, "t1", 3, false), Decision::Abort); + } + // A completion without the marker may still hold it in a later event, for example after + // a Local Activity, so it decides nothing. + let completed = [event(5, EventType::WorkflowTaskCompleted)]; + assert_eq!(decide_token(&completed, "t1", 3, false), Decision::Unknown); + // A result at or before the floor belongs to an earlier task. + let earlier = [event(3, EventType::WorkflowTaskFailed)]; + assert_eq!(decide_token(&earlier, "t1", 3, false), Decision::Unknown); + } + + #[test] + fn rule_24_1_a_commit_at_the_same_floor_aborts_a_stage_from_before_an_eviction() { + let events = [ + event(5, EventType::WorkflowTaskCompleted), + marker(6, "other", 3), + ]; + assert_eq!(decide_token(&events, "t1", 3, true), Decision::Abort); + // Without an eviction between them, both stages may belong to one attempt. + assert_eq!(decide_token(&events, "t1", 3, false), Decision::Unknown); + // Another floor says nothing about this stage. + assert_eq!(decide_token(&events, "t1", 4, true), Decision::Unknown); + } +} diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 95bcfe245..940df18a4 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -417,10 +417,13 @@ fn stage_ref(chain: &ChainId, token: &str, topics: &[&str]) -> StageRef { } } -use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; +use temporalio_common::protos::temporal::api::{ + enums::v1::WorkflowExecutionStatus, history::v1::HistoryEvent, +}; -/// Temporal for one chain whose Workflow is still running, for cases that go through [Streams]. -struct RunningChain(ChainId); +/// Temporal for one chain whose Workflow is still running, for cases that go through [Streams], +/// with the History its first run holds. +struct RunningChain(ChainId, std::sync::Mutex>); #[async_trait::async_trait] impl OwnerClient for RunningChain { @@ -434,20 +437,40 @@ impl OwnerClient for RunningChain { run_id: self.0.first_run_id.clone(), first_run_id: self.0.first_run_id.clone(), status: WorkflowExecutionStatus::Running, + start_time: None, }) } + + async fn history_after( + &self, + _namespace: &str, + _workflow_id: &str, + _run_id: &str, + floor: i64, + ) -> Result, OwnerError> { + Ok(self + .1 + .lock() + .unwrap() + .iter() + .filter(|event| event.event_id > floor) + .cloned() + .collect()) + } } -fn streams_over(case: &dyn Case, chain: &ChainId) -> Streams { - Streams::new( +fn streams_over(case: &dyn Case, chain: &ChainId) -> (Streams, Arc) { + let owner = Arc::new(RunningChain(chain.clone(), Default::default())); + let streams = Streams::new( case.store(), - Arc::new(RunningChain(chain.clone())), + owner.clone(), StreamsOptions { owner_check_min: Duration::from_millis(20), owner_check_max: Duration::from_millis(80), ..StreamsOptions::default() }, - ) + ); + (streams, owner) } fn stream_read(chain: &ChainId, after: &str, state: &[u8], wait: Duration) -> ReadRequest { @@ -466,6 +489,42 @@ fn stream_read(chain: &ChainId, after: &str, state: &[u8], wait: Duration) -> Re } } +/// The marker a Workflow Task writes when it commits the stage `token` made at `floor`. +fn marker(event_id: i64, token: &str, floor: i64) -> HistoryEvent { + use temporalio_common::protos::{ + constants::EXTERNAL_STREAM_MARKER_NAME, + coresdk::external_data::{ + ExternalOutputStreamManifest, ExternalStreamMarkerData, + build_external_stream_marker_details, + }, + temporal::api::{ + enums::v1::EventType, + history::v1::{MarkerRecordedEventAttributes, history_event::Attributes}, + }, + }; + let data = ExternalStreamMarkerData { + schema_version: 1, + output: Some(ExternalOutputStreamManifest { + stage_token: token.to_string(), + history_floor_event_id: floor, + ..Default::default() + }), + ..Default::default() + }; + HistoryEvent { + event_id, + event_type: EventType::MarkerRecorded as i32, + attributes: Some(Attributes::MarkerRecordedEventAttributes( + MarkerRecordedEventAttributes { + marker_name: EXTERNAL_STREAM_MARKER_NAME.to_string(), + details: build_external_stream_marker_details(&data), + ..Default::default() + }, + )), + ..Default::default() + } +} + fn kind(result: StreamResult) -> StreamFailureKind { result.unwrap_err().kind } @@ -1167,7 +1226,7 @@ mod cases { .await .unwrap(); store.close_chain(&chain).await.unwrap(); - let streams = streams_over(case, &chain); + let (streams, _) = streams_over(case, &chain); let first = streams .read(stream_read( &chain, @@ -1197,6 +1256,40 @@ mod cases { panic!("the read never ended, and delivered {delivered:?}"); } + pub(crate) async fn rule_12_1_a_read_that_ends_on_a_close_repairs_the_chain(case: &dyn Case) { + // A Worker that stopped after the final commit leaves its stage pending, and only the + // read that sees the close is there to promote it. + let store = case.store(); + let chain = case.chain(); + let (streams, owner) = streams_over(case, &chain); + let first = streams + .read(stream_read(&chain, BEGINNING, &[], Duration::ZERO)) + .await + .unwrap(); + assert!(first.records.is_empty() && !first.done); + store + .stage(staged(&chain, &chain.first_run_id, "t1", &[(OUT, "last")])) + .await + .unwrap(); + owner.1.lock().unwrap().push(marker(5, "t1", 3)); + store.close_chain(&chain).await.unwrap(); + let mut request = stream_read(&chain, &first.cursor, &first.state, Duration::from_secs(5)); + let mut delivered = Vec::new(); + for _ in 0..5 { + let page = tokio::time::timeout(Duration::from_secs(10), streams.read(request.clone())) + .await + .expect("the read waited for a stage nothing would promote") + .unwrap(); + delivered.extend(values(&page)); + if page.done { + assert_eq!(delivered, ["last"]); + return; + } + request = stream_read(&chain, &page.cursor, &page.state, Duration::from_secs(5)); + } + panic!("the read never ended, and delivered {delivered:?}"); + } + pub(crate) async fn an_aborted_stage_never_lands(case: &dyn Case) { let store = case.store(); let chain = case.chain(); @@ -1225,6 +1318,8 @@ mod cases { .await .unwrap_err(); assert!(error.message.contains(OTHER), "{error}"); + // A retry names the same topics, so the store refuses it for good. + assert_eq!(error.kind, StreamFailureKind::Refused); assert_eq!(latest(case, &chain, OUT).await, ""); assert_eq!(store.pending_stages(&chain).await.unwrap().len(), 1); } @@ -1376,6 +1471,7 @@ mod memory { a_repeated_stage_holds_its_records_once, an_aborted_stage_never_lands, rule_11_1_a_read_ends_only_after_the_chains_last_stage_lands, + rule_12_1_a_read_that_ends_on_a_close_repairs_the_chain, a_promotion_naming_too_few_topics_writes_nothing, a_promotion_lands_on_a_closed_chain, rule_19_3_a_reset_run_writes_on_and_keeps_the_chain_open, @@ -1424,6 +1520,7 @@ macro_rules! redis_cases { a_repeated_stage_holds_its_records_once, an_aborted_stage_never_lands, rule_11_1_a_read_ends_only_after_the_chains_last_stage_lands, + rule_12_1_a_read_that_ends_on_a_close_repairs_the_chain, a_promotion_naming_too_few_topics_writes_nothing, a_promotion_lands_on_a_closed_chain, rule_19_3_a_reset_run_writes_on_and_keeps_the_chain_open, diff --git a/crates/streams/tests/cross_impl.rs b/crates/streams/tests/cross_impl.rs new file mode 100644 index 000000000..07b180dcf --- /dev/null +++ b/crates/streams/tests/cross_impl.rs @@ -0,0 +1,510 @@ +//! The Core store and the Python SDK's Redis provider on one Redis, each reading and writing +//! what the other wrote. +//! +//! Set `STREAMS_REDIS_URL`, and `STREAMS_BROOK_PY` to a Python SDK checkout whose Redis provider +//! is the one before the store moved into Core. The Python side runs `tests/cross_impl/brook.py` +//! through `uv`. + +#![cfg(feature = "redis")] + +use prost::Message; +use serde_json::{Value, json}; +use std::{ + path::PathBuf, + process::Command, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; +use temporalio_common::protos::temporal::api::{ + common::v1::Payload, enums::v1::WorkflowExecutionStatus, history::v1::HistoryEvent, +}; +use temporalio_streams::{ + OwnerClient, OwnerDescription, OwnerError, ReadTarget, RedisStore, RedisStoreOptions, + StreamStore, Streams, StreamsOptions, mint_cursor, + proto::{ + AppendRecord, AppendRequest, ChainId, NamedProducer, ReadRequest, StageRef, StagedBatch, + StagedRecord, StoreAppendRequest, StreamAddress, StreamFailureKind, StreamOwnerKind, + StreamRecordKind, append_request::Producer, read_record, + }, + read_page, stored_append_record, stored_output_record, stream_hash, +}; + +static IDS: AtomicUsize = AtomicUsize::new(0); + +const NAMESPACE: &str = "ns"; +const FIRST_RUN: &str = "run-1"; + +struct Both { + url: String, + checkout: PathBuf, + prefix: String, + workflow_id: String, + retention_ms: u64, + core: RedisStore, +} + +async fn pair(retention: Duration) -> Option { + let (Ok(url), Ok(checkout)) = ( + std::env::var("STREAMS_REDIS_URL"), + std::env::var("STREAMS_BROOK_PY"), + ) else { + eprintln!( + "set STREAMS_REDIS_URL and STREAMS_BROOK_PY to run the cross-implementation test" + ); + return None; + }; + let id = IDS.fetch_add(1, Ordering::Relaxed); + let prefix = format!("cross-{}-{id}", std::process::id()); + let mut options = RedisStoreOptions::new(url.clone()); + options.key_prefix = prefix.clone(); + options.retention = retention; + Some(Both { + core: RedisStore::connect(options).await.unwrap(), + url, + checkout: checkout.into(), + prefix, + workflow_id: format!("wf-{id}"), + retention_ms: retention.as_millis() as u64, + }) +} + +impl Both { + /// Runs one operation of the Python provider and answers its JSON. + fn brook(&self, op: Value) -> Value { + let helper = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/cross_impl/brook.py"); + let mut op = op; + let fields = op.as_object_mut().unwrap(); + fields.insert("url".into(), json!(self.url)); + fields.insert("prefix".into(), json!(self.prefix)); + fields.insert("namespace".into(), json!(NAMESPACE)); + fields.insert("workflow_id".into(), json!(self.workflow_id)); + fields.insert("first_run".into(), json!(FIRST_RUN)); + fields.insert("retention_ms".into(), json!(self.retention_ms)); + let output = Command::new("uv") + .args(["run", "--no-sync", "python"]) + .arg(helper) + .arg(op.to_string()) + .current_dir(&self.checkout) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + serde_json::from_str(stdout.lines().last().unwrap()).unwrap() + } + + fn chain(&self) -> ChainId { + ChainId { + namespace: NAMESPACE.to_string(), + workflow_id: self.workflow_id.clone(), + first_run_id: FIRST_RUN.to_string(), + } + } + + fn target(&self, topic: &str) -> ReadTarget { + ReadTarget { + chain: self.chain(), + topic: topic.to_string(), + stream_hash: stream_hash(NAMESPACE, "workflow", &self.workflow_id, topic), + } + } + + fn cursor(&self, topic: &str, position: &str) -> String { + mint_cursor("redis", &self.target(topic).stream_hash, position) + } + + /// Appends JSON values as producer `p` the way lang hands them to Core. + async fn core_append( + &self, + sequence: i64, + values: &[Value], + digest: &[u8], + ) -> (String, String) { + let records: Vec<_> = values + .iter() + .enumerate() + .map(|(index, value)| { + stored_append_record( + "events", + "p", + 1, + sequence + index as i64, + &json_record(value), + ) + .unwrap() + .encode_to_vec() + }) + .collect(); + let landed = self + .core + .append(StoreAppendRequest { + chain: Some(self.chain()), + topic: "events".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: digest.to_vec(), + records, + }) + .await + .unwrap(); + (landed.first_position, landed.last_position) + } + + /// Reads a topic from `after` with the Core reader, as JSON values. + async fn core_read(&self, topic: &str, after: &str) -> Result, StreamFailureKind> { + let page = read_page( + &self.core, + &self.target(topic), + &ReadRequest { + after: after.to_string(), + ..Default::default() + }, + ) + .await + .map_err(|error| error.kind)?; + Ok(page + .records + .iter() + .filter_map(|record| match record.record.as_ref()? { + read_record::Record::Stored(stored) => { + let body = stored.body.as_ref()?; + serde_json::from_slice(&body.data).ok() + } + read_record::Record::Superseded(_) => None, + }) + .collect()) + } +} + +fn json_record(value: &Value) -> AppendRecord { + let payload = Payload { + metadata: [("encoding".to_string(), b"json/plain".to_vec())].into(), + data: value.to_string().into_bytes(), + ..Default::default() + }; + use sha2::{Digest, Sha256}; + AppendRecord { + kind: StreamRecordKind::Data as i32, + content_hash: Sha256::digest(payload.encode_to_vec()).to_vec(), + body: Some(payload), + } +} + +fn values(records: &Value) -> Vec { + records + .as_array() + .unwrap_or_else(|| panic!("not records: {records}")) + .iter() + .map(|record| record["value"].clone()) + .collect() +} + +fn hex(text: &str) -> Vec { + (0..text.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&text[i..i + 2], 16).unwrap()) + .collect() +} + +const WEEK: Duration = Duration::from_secs(7 * 24 * 60 * 60); +/// Long enough that starting the Python side never outlasts it. +const SHORT: Duration = Duration::from_secs(2); + +#[tokio::test] +async fn what_python_wrote_core_reads_and_resumes() { + let Some(both) = pair(WEEK).await else { + return; + }; + let first = + both.brook(json!({"op": "append", "producer": "p", "values": [{"n": 1}, {"n": 2}]})); + both.brook(json!({"op": "append", "producer": "p", "sequence": 3, "values": [{"n": 3}]})); + assert_eq!( + both.core_read("events", "").await.unwrap(), + [json!({"n": 1}), json!({"n": 2}), json!({"n": 3})] + ); + // A cursor the Python provider minted resumes on the Core store. + let cursor = first["cursor"].as_str().unwrap(); + assert_eq!( + both.core_read("events", cursor).await.unwrap(), + [json!({"n": 3})] + ); + let page = read_page(&both.core, &both.target("events"), &ReadRequest::default()) + .await + .unwrap(); + let Some(read_record::Record::Stored(stored)) = &page.records[2].record else { + panic!("no record"); + }; + assert_eq!( + (stored.producer_id.as_str(), stored.attempt, stored.sequence), + ("p", 1, 3) + ); +} + +#[tokio::test] +async fn what_core_wrote_python_reads_and_resumes() { + let Some(both) = pair(WEEK).await else { + return; + }; + let (_, first) = both.core_append(1, &[json!({"n": 1})], b"\x01").await; + both.core_append(2, &[json!({"n": 2})], b"\x02").await; + let read = both.brook(json!({"op": "read", "count": 2})); + assert_eq!(values(&read), [json!({"n": 1}), json!({"n": 2})]); + assert_eq!(read[0]["producer_id"], "p"); + // A cursor the Core store minted resumes on the Python provider. + let after = + both.brook(json!({"op": "read", "count": 1, "after": both.cursor("events", &first)})); + assert_eq!(values(&after), [json!({"n": 2})]); +} + +/// Temporal as the stream layer asks it: one Workflow, running in its first run. +struct Running; + +#[async_trait::async_trait] +impl OwnerClient for Running { + async fn describe( + &self, + _namespace: &str, + _workflow_id: &str, + _run_id: &str, + ) -> Result { + Ok(OwnerDescription { + run_id: FIRST_RUN.to_string(), + first_run_id: FIRST_RUN.to_string(), + status: WorkflowExecutionStatus::Running, + start_time: None, + }) + } + + async fn history_after( + &self, + _namespace: &str, + _workflow_id: &str, + _run_id: &str, + _floor: i64, + ) -> Result, OwnerError> { + Ok(Vec::new()) + } +} + +impl Both { + /// The stream layer over a store on the same keys, as lang reaches it. + async fn streams(&self) -> Streams { + let mut options = RedisStoreOptions::new(self.url.clone()); + options.key_prefix = self.prefix.clone(); + let store = RedisStore::connect(options).await.unwrap(); + Streams::new( + Arc::new(store), + Arc::new(Running), + StreamsOptions::default(), + ) + } + + fn append_request(&self, values: &[Value], digest: Vec) -> AppendRequest { + AppendRequest { + stream: Some(StreamAddress { + namespace: NAMESPACE.to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: self.workflow_id.clone(), + run_id: String::new(), + topic: "events".to_string(), + }), + producer: Some(Producer::Named(NamedProducer { + producer_id: "p".to_string(), + attempt: 1, + })), + sequence: 1, + records: values.iter().map(json_record).collect(), + digest, + } + } +} + +#[tokio::test] +async fn a_batch_python_wrote_dedupes_through_core_with_its_digest() { + // Lang takes the digest and the store keeps it as given, so a retry that carries the same + // digest is recognized whichever SDK wrote first. + let Some(both) = pair(WEEK).await else { + return; + }; + let streams = both.streams().await; + let landed = both.brook(json!({"op": "append", "producer": "p", "values": [{"n": 1}]})); + let digest = hex(landed["digest"].as_str().unwrap()); + let retry = streams + .append(both.append_request(&[json!({"n": 1})], digest)) + .await + .unwrap(); + assert_eq!(retry.last_cursor, landed["cursor"].as_str().unwrap()); + let error = streams + .append(both.append_request(&[json!({"n": 1})], vec![7; 32])) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::ProducerDivergent); + assert_eq!(both.core_read("events", "").await.unwrap().len(), 1); +} + +#[tokio::test] +async fn a_batch_core_wrote_dedupes_through_python() { + let Some(both) = pair(WEEK).await else { + return; + }; + let streams = both.streams().await; + let digest = both.brook(json!({"op": "digest", "producer": "p", "values": [{"n": 1}]})); + let landed = streams + .append(both.append_request(&[json!({"n": 1})], hex(digest.as_str().unwrap()))) + .await + .unwrap(); + let retry = both.brook(json!({"op": "append", "producer": "p", "values": [{"n": 1}]})); + assert_eq!(retry["cursor"].as_str().unwrap(), landed.last_cursor); + assert_eq!(both.core_read("events", "").await.unwrap().len(), 1); +} + +#[tokio::test] +async fn either_side_sees_a_chain_the_other_closed() { + let Some(both) = pair(WEEK).await else { + return; + }; + both.core_append(1, &[json!({"n": 1})], b"\x01").await; + both.core.close_chain(&both.chain()).await.unwrap(); + let refused = both.brook(json!({"op": "append", "producer": "q", "values": [{"n": 2}]})); + assert_eq!(refused["error"], "StreamClosedError", "{refused}"); + + let other = pair(WEEK).await.unwrap(); + other.brook(json!({"op": "close"})); + let error = other + .core + .append(StoreAppendRequest { + chain: Some(other.chain()), + topic: "events".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence: 1, + digest: b"\x01".to_vec(), + records: vec![b"\x1a\x06events".to_vec()], + }) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Closed); +} + +#[tokio::test] +async fn a_stage_one_side_wrote_the_other_promotes() { + let Some(both) = pair(WEEK).await else { + return; + }; + let token = both.brook(json!({ + "op": "stage", + "run_id": FIRST_RUN, + "floor": 3, + "records": [ + {"topic": "events", "value": {"n": 1}}, + {"topic": "other", "value": {"n": 2}}, + ], + })); + let token = token.as_str().unwrap(); + let pending = both.core.pending_stages(&both.chain()).await.unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!( + ( + pending[0].token.as_str(), + pending[0].run_id.as_str(), + pending[0].history_floor_event_id, + pending[0].topics.clone(), + ), + ( + token, + FIRST_RUN, + 3, + vec!["events".to_string(), "other".to_string()] + ) + ); + let promoted = both + .core + .promote(&StageRef { + chain: Some(both.chain()), + token: token.to_string(), + topics: pending[0].topics.clone(), + }) + .await + .unwrap(); + assert_eq!(promoted.records, 2); + let read = both.brook(json!({"op": "read", "count": 1, "topic": "other"})); + assert_eq!(values(&read), [json!({"n": 2})]); + assert_eq!(read[0]["run_id"], FIRST_RUN); + + let record = json_record(&json!({"n": 3})); + both.core + .stage(StagedBatch { + chain: Some(both.chain()), + run_id: FIRST_RUN.to_string(), + token: "core-token".to_string(), + history_floor_event_id: 7, + records: vec![StagedRecord { + topic: "events".to_string(), + record: stored_output_record( + "events", + record.kind, + record.body, + &record.content_hash, + FIRST_RUN, + ) + .unwrap() + .encode_to_vec(), + }], + }) + .await + .unwrap(); + both.brook(json!({"op": "promote", "token": "core-token", "topics": ["events"]})); + assert_eq!( + both.core_read("events", "").await.unwrap(), + [json!({"n": 1}), json!({"n": 3})] + ); + assert!( + both.core + .pending_stages(&both.chain()) + .await + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn a_trim_core_made_expires_a_python_cursor() { + let Some(both) = pair(SHORT).await else { + return; + }; + both.brook(json!({"op": "append", "producer": "p", "values": [{"n": 1}, {"n": 2}]})); + let delivered = both.brook(json!({"op": "read", "count": 1})); + tokio::time::sleep(SHORT + Duration::from_millis(200)).await; + // This append trims both records, and the reader never got the second. + both.core_append(3, &[json!({"n": 3})], b"\x03").await; + let expired = both.brook(json!({ + "op": "read", + "count": 1, + "after": delivered[0]["cursor"], + })); + assert_eq!(expired["error"], "StreamExpiredError", "{expired}"); +} + +#[tokio::test] +async fn a_trim_python_made_expires_a_core_cursor() { + let Some(both) = pair(SHORT).await else { + return; + }; + let (first, _) = both + .core_append(1, &[json!({"n": 1}), json!({"n": 2})], b"\x01") + .await; + tokio::time::sleep(SHORT + Duration::from_millis(200)).await; + both.brook(json!({"op": "append", "producer": "q", "values": [{"n": 3}]})); + assert_eq!( + both.core_read("events", &both.cursor("events", &first)) + .await, + Err(StreamFailureKind::Expired) + ); +} diff --git a/crates/streams/tests/cross_impl/brook.py b/crates/streams/tests/cross_impl/brook.py new file mode 100644 index 000000000..0b1243b4e --- /dev/null +++ b/crates/streams/tests/cross_impl/brook.py @@ -0,0 +1,178 @@ +"""Drives the Python SDK's Redis provider for the cross-implementation test. + +Run from a Python SDK checkout with ``uv run --no-sync python brook.py ''``. The argument +names one operation on one stream, and the answer is one JSON line. The provider talks to Redis +as in production. A stand-in for the Temporal client describes one running Workflow, so no +server is needed. +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from datetime import timedelta +from types import SimpleNamespace +from typing import Any + +import temporalio.converter +from temporalio.client import WorkflowExecutionStatus +from temporalio.contrib.streams import Cursor, RecordKind, StreamError, StreamRef +from temporalio.contrib.streams._body import content_fingerprint +from temporalio.contrib.streams._output import StagedBatch, StageRef +from temporalio.contrib.streams._wire import to_wire +from temporalio.contrib.streams.redis import RedisStreams + + +class Owner: + """A Temporal client that knows one running Workflow, whose first run is ``first_run``.""" + + def __init__(self, namespace: str, first_run: str) -> None: + self.namespace = namespace + self.data_converter = temporalio.converter.DataConverter.default + self._first_run = first_run + + def get_workflow_handle(self, workflow_id: str, run_id: str | None = None) -> Any: + async def describe() -> Any: + info = SimpleNamespace(first_run_id=self._first_run) + return SimpleNamespace( + status=WorkflowExecutionStatus.RUNNING, + run_id=run_id or self._first_run, + raw_description=SimpleNamespace(workflow_execution_info=info), + ) + + return SimpleNamespace(describe=describe) + + +def record(item: Any) -> dict[str, Any]: + return { + "cursor": item.cursor.token, + "kind": item.kind.name, + "value": item.value, + "producer_id": item.producer_id, + "attempt": item.attempt, + "sequence": item.sequence, + "run_id": item.run_id, + "stale": item.stale, + } + + +async def run(op: dict[str, Any]) -> Any: + provider = RedisStreams( + op["url"], + key_prefix=op["prefix"], + retention=timedelta(milliseconds=op.get("retention_ms", 7 * 24 * 3600 * 1000)), + poll_interval=timedelta(milliseconds=50), + ) + owner = Owner(op["namespace"], op["first_run"]) + ref = StreamRef.for_workflow(op["workflow_id"]) + stream = provider.get_stream_handle(owner, ref) # type: ignore[arg-type] + topic = op.get("topic", "events") + try: + if op["op"] == "append": + producer = stream.producer( + topic=topic, producer_id=op["producer"], attempt=op.get("attempt", 1) + ) + producer._sequence = op.get("sequence", 1) + converter = owner.data_converter.payload_converter + wires = [ + to_wire( + converter, + topic=topic, + kind=RecordKind.DATA, + value=value, + producer_id=op["producer"], + attempt=op.get("attempt", 1), + sequence=producer._sequence + index, + ) + for index, value in enumerate(op["values"]) + ] + digest = content_fingerprint(wires).hex() + cursor = await producer.append(*op["values"]) + return {"cursor": cursor.token, "digest": digest} + if op["op"] == "digest": + converter = owner.data_converter.payload_converter + wires = [ + to_wire( + converter, + topic=topic, + kind=RecordKind.DATA, + value=value, + producer_id=op["producer"], + attempt=op.get("attempt", 1), + sequence=op.get("sequence", 1) + index, + ) + for index, value in enumerate(op["values"]) + ] + return content_fingerprint(wires).hex() + if op["op"] == "read": + records = stream.read(topic=topic, after=Cursor(op.get("after", ""))) + out = [] + try: + while len(out) < op["count"]: + out.append(record(await asyncio.wait_for(records.__anext__(), 5))) + except asyncio.TimeoutError: + return { + "error": "Timeout", + "message": f"got {len(out)} records", + "cursor": None, + } + finally: + await records.aclose() + return out + if op["op"] == "latest": + return (await stream.latest(topic=topic)).token + if op["op"] == "stage": + converter = owner.data_converter.payload_converter + records = [ + to_wire( + converter, + topic=item["topic"], + kind=RecordKind.DATA, + value=item["value"], + run_id=op["run_id"], + ) + for item in op["records"] + ] + return await provider._stage( + StagedBatch( + op["namespace"], + op["workflow_id"], + op["first_run"], + op["run_id"], + records, + history_floor_event_id=op["floor"], + ) + ) + if op["op"] == "promote": + await provider._promote( + StageRef( + op["namespace"], + op["workflow_id"], + op["first_run"], + op["token"], + tuple(op["topics"]), + ) + ) + return None + if op["op"] == "close": + await provider._mark_closed( + provider._chain_keys( + op["namespace"], op["workflow_id"], op["first_run"] + ) + ) + return None + raise ValueError(f"no operation {op['op']!r}") + except StreamError as error: + cursor = getattr(error, "cursor", None) + return { + "error": type(error).__name__, + "message": str(error), + "cursor": cursor.token if cursor is not None else None, + } + finally: + await provider.close() + + +if __name__ == "__main__": + print(json.dumps(asyncio.run(run(json.loads(sys.argv[1]))))) diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 228cf14be..554195dcd 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -925,8 +925,22 @@ impl temporalio_streams::OwnerClient for OneOwner { run_id: "run-1".to_string(), first_run_id: "run-1".to_string(), status: *self.status.lock().unwrap(), + start_time: None, }) } + + async fn history_after( + &self, + _namespace: &str, + _workflow_id: &str, + _run_id: &str, + _floor: i64, + ) -> Result< + Vec, + temporalio_streams::OwnerError, + > { + Ok(Vec::new()) + } } #[tokio::test]