diff --git a/rust/src/runtime/definition_routing.rs b/rust/src/runtime/definition_routing.rs index 2406206e6..4d454f7d2 100644 --- a/rust/src/runtime/definition_routing.rs +++ b/rust/src/runtime/definition_routing.rs @@ -57,14 +57,18 @@ where run: &DB::Row, task_queue: Option<&str>, required_build_id: Option, + started: Option<&Value>, ) -> Result { - let started = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='WorkflowStarted' ORDER BY sequence LIMIT 1") - .bind(DB::string(run, "id")?) - .fetch_optional(&mut **tx) - .await? - .map(|row| DB::document_row(&row, "payload")) - .transpose()? - .unwrap_or(Value::Null); + // A caller that already read the start payload in this transaction + // passes it; otherwise it is read here. + let read; + let started = match started { + Some(started) => started, + None => { + read = Self::started_payload(&mut **tx, &DB::string(run, "id")?).await?; + &read + } + }; Ok(DefinitionContext { worker_pinned: started["workflow_definition_fingerprint_source"] == "worker", recorded: trimmed(started["workflow_definition_fingerprint"].as_str()), @@ -210,6 +214,7 @@ where tx: &mut Transaction<'_, DB>, run: &DB::Row, task_queue: Option<&str>, + started: Option<&Value>, ) -> Result { let run_id = DB::string(run, "id")?; let task = Self::query("SELECT queue,compatibility FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status='ready' AND (available_at IS NULL OR available_at<=$2) ORDER BY id") @@ -226,7 +231,7 @@ where ), None => (task_queue.map(str::to_owned), None), }; - let context = Self::definition_context(tx, run, queue.as_deref(), required).await?; + let context = Self::definition_context(tx, run, queue.as_deref(), required, started).await?; let (mut matching, mut changed, mut missing, mut unsafe_replacement) = (0, 0, 0, 0); let mut examples = Vec::new(); let peers = self.definition_peers(tx, run, &context).await?; @@ -296,7 +301,7 @@ where } let mut example = json!({"workflow_id": DB::string(run, "workflow_instance_id")?, "run_id": DB::string(run, "id")?}); if let (Some(example), Value::Object(inspected)) = - (example.as_object_mut(), self.definition_inspect(tx, run, Some(task_queue)).await?) + (example.as_object_mut(), self.definition_inspect(tx, run, Some(task_queue), None).await?) { example.extend(inspected); } diff --git a/rust/src/runtime/describe_snapshot_tests.rs b/rust/src/runtime/describe_snapshot_tests.rs new file mode 100644 index 000000000..9217e0411 --- /dev/null +++ b/rust/src/runtime/describe_snapshot_tests.rs @@ -0,0 +1,135 @@ +//! Hold one description after its run row while another connection commits +//! a real termination. The returned run and command journal must agree. +use super::cancellation::TerminalCommand; +use crate::codec::{Value as Payload, ValueCodec}; +use serde_json::{Value, json}; +use sqlx::Database; +use std::{sync::{Arc, atomic::{AtomicUsize, Ordering}}, time::Duration}; +use tokio::sync::Notify; + +#[derive(Default)] +struct Gate { reached: Notify, release: Notify } + +tokio::task_local! { static AT_RUN_ROW: Arc; } + +pub(super) async fn after_run_row() { + if let Ok(gate) = AT_RUN_ROW.try_with(Arc::clone) { + gate.reached.notify_one(); + gate.release.notified().await; + } +} + +/// One eager connection, with every subsequent checkout counted. A nested +/// pool read while the description owns it would time out, not pass silently. +pub(super) fn reader_pool( + options: sqlx::pool::PoolOptions, checkouts: Arc, +) -> sqlx::pool::PoolOptions { + options.max_connections(1).min_connections(1) + .acquire_timeout(Duration::from_secs(2)) + .before_acquire(move |_, _| { + checkouts.fetch_add(1, Ordering::Relaxed); + Box::pin(async { Ok(true) }) + }) +} + +fn input() -> Value { + json!({"codec":"avro","blob":ValueCodec::new().unwrap().encode(&Payload::Array(vec![])).unwrap()}) +} + +store_methods! { + /// The reader has one pool connection; the writer has an independent pool + /// on the same database. `ordinary` is the READ COMMITTED counterexample. + pub(super) async fn qualify_describe_snapshot( + &self, writer: &Self, checkouts: &AtomicUsize, ordinary: bool, + ) { + let workflow_id = format!("describe-snapshot-{}",ulid::Ulid::new()); + let started = writer.start(json!({"workflow_id":workflow_id,"workflow_type":"snapshot","input":input()}), + super::memos::initial(json!({"sample":"before"})).unwrap(),None,None).await.unwrap(); + let run_id = started["run_id"].as_str().unwrap(); + let before = self.describe_with(&workflow_id,None,Some(60)).await.unwrap(); + let gate = Arc::new(Gate::default()); + checkouts.store(0,Ordering::Relaxed); + let read = AT_RUN_ROW.scope(gate.clone(),async { + if ordinary { + let mut tx = self.begin_read().await?; + let answer = self.describe_in(&mut tx,&workflow_id,None,Some(60)).await?; + tx.commit().await?; + Ok::<_,super::RuntimeError>(answer) + } else { + self.describe_with(&workflow_id,None,Some(60)).await + } + }); + let write = async { + gate.reached.notified().await; + let (status,_) = writer.close_workflow(TerminalCommand::Terminate,&workflow_id,None,None,None,None).await.unwrap(); + assert!(status.is_success()); + gate.release.notify_one(); + }; + let ((answer,reads),()) = tokio::time::timeout(Duration::from_secs(10),async { + tokio::join!(super::sql_counts::measure(read),write) + }).await.expect("description and concurrent termination must finish"); + let answer = answer.unwrap(); + assert_eq!(checkouts.load(Ordering::Relaxed),1,"one checkout for the whole description"); + assert_eq!(reads.count(super::run_describe::STARTED_PAYLOAD_SQL),1,"one shared start-payload read"); + assert_eq!(answer["status"],before["status"],"run row was read before the termination"); + if ordinary { + assert_ne!(answer["commands"],before["commands"],"READ COMMITTED exposes the later journal and mixes generations"); + } else { + assert_eq!(answer,before,"every section must come from the pre-termination snapshot"); + } + self.assert_describe_connection_returned().await; + let fresh = self.describe_with(&workflow_id,Some(run_id),Some(60)).await.unwrap(); + assert_eq!(fresh["status"],"terminated"); + assert_eq!(fresh["commands"].as_array().unwrap().len(),before["commands"].as_array().unwrap().len()+1); + + // Both not-found paths return the only connection, including an + // explicitly selected run outside the instance's namespace/identity. + for (workflow,run) in [("missing-instance",None),(workflow_id.as_str(),Some("missing-run"))] { + assert!(self.describe_with(workflow,run,Some(60)).await.is_err()); + self.assert_describe_connection_returned().await; + } + // Cancellation at an actual suspended read must also release it. + let gate = Arc::new(Gate::default()); + let mut reading = Box::pin(AT_RUN_ROW.scope(gate.clone(),self.describe_with(&workflow_id,None,Some(60)))); + tokio::time::timeout(Duration::from_secs(5),async { + tokio::select! { + _ = gate.reached.notified() => {}, + result = &mut reading => panic!("description did not pause: {result:?}"), + } + }).await.unwrap(); + assert!(self.pool.try_acquire().is_none(),"description holds the one connection until cancellation"); + drop(reading); + self.assert_describe_connection_returned().await; + + // A stored-payload decoding failure occurs after the transaction has + // read real rows, unlike the early not-found refusals above. + Self::query("UPDATE workflow_memos SET portable_value=$1,portable_value_sequence=upserted_at_sequence WHERE workflow_run_id=$2") + .bind(DB::document(&json!({"codec":"avro","blob":"invalid"}))).bind(run_id) + .execute(&writer.pool).await.unwrap(); + assert!(matches!(self.describe_with(&workflow_id,None,Some(60)).await, + Err(super::RuntimeError::Refused { reason:"unsupported_stored_memo", .. }))); + self.assert_describe_connection_returned().await; + } + + async fn assert_describe_connection_returned(&self) { + tokio::time::timeout(Duration::from_secs(3),async { + let mut connection = self.pool.acquire().await.unwrap(); + Self::query("SELECT 1").fetch_one(&mut *connection).await.unwrap(); + }).await.expect("description must return its connection on success, error and cancellation"); + } +} + +#[tokio::test] +async fn sqlite_describe_is_coherent_and_returns_one_connection() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("describe.sqlite"); + let writer = super::store::Store::::new(super::sqlite::open(path.to_str().unwrap()).await.unwrap()); + let checkouts = Arc::new(AtomicUsize::new(0)); + let pool = reader_pool(sqlx::sqlite::SqlitePoolOptions::new(),checkouts.clone()) + .connect_with(sqlx::sqlite::SqliteConnectOptions::new().filename(path).read_only(true) + .busy_timeout(Duration::from_secs(5))).await.unwrap(); + let reader = super::store::Store::new(pool); + reader.qualify_describe_snapshot(&writer,&checkouts,false).await; + reader.close().await; + writer.close().await; +} diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 3ce56bf9f..f0af11bf8 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -52,6 +52,8 @@ mod command_context; mod command_requests; mod continuations; mod definition_routing; +#[cfg(test)] +mod describe_snapshot_tests; mod control_plane; mod cooperative; mod cross_kind_poll; diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 935019a7a..cc850609c 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1235,6 +1235,34 @@ mod tests { db.remove().await; } + #[tokio::test] + #[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] + async fn qualification_describe_snapshot_under_repeatable_read_and_read_committed() { + use super::super::{describe_snapshot_tests::reader_pool, store::Store}; + use std::sync::{Arc, atomic::AtomicUsize}; + let db = Database::new().await; + let writer = Store::new(MySqlStorage::open(db.options.clone()).await.unwrap().into_pool()); + for level in ["SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ", + "SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED"] { + let checkouts = Arc::new(AtomicUsize::new(0)); + let pool = reader_pool(runtime_pool_options(),checkouts.clone()) + .connect_with(db.options.clone()).await.unwrap(); + pool.acquire().await.unwrap().execute(level).await.unwrap(); + let reader = Store::new(pool); + if level.ends_with("READ COMMITTED") { + reader.qualify_describe_snapshot(&writer,&checkouts,true).await; + } + reader.qualify_describe_snapshot(&writer,&checkouts,false).await; + if level.ends_with("READ COMMITTED") { + // The one-transaction override must leave the session intact. + reader.qualify_describe_snapshot(&writer,&checkouts,true).await; + } + reader.close().await; + } + writer.close().await; + db.remove().await; + } + #[tokio::test] #[ignore = "requires the existing disposable mysql source suite"] async fn qualification_transition_queue_preserves_reads_fifo_and_timeout_admission() { diff --git a/rust/src/runtime/postgres.rs b/rust/src/runtime/postgres.rs index 4156fc099..f3d913688 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -755,6 +755,23 @@ mod tests { db.remove().await; } + #[tokio::test] + #[ignore = "requires disposable PostgreSQL; run qualification_ explicitly"] + async fn qualification_describe_is_coherent_and_returns_one_connection() { + use super::super::{describe_snapshot_tests::reader_pool, store::Store}; + use std::sync::{Arc, atomic::AtomicUsize}; + let db = Database::new().await; + let writer = Store::new(PostgresStorage::open(db.options.clone()).await.unwrap().into_pool()); + let checkouts = Arc::new(AtomicUsize::new(0)); + let pool = reader_pool(PgPoolOptions::new(),checkouts.clone()) + .connect_with(db.options.clone()).await.unwrap(); + let reader = Store::new(pool); + reader.qualify_describe_snapshot(&writer,&checkouts,false).await; + reader.close().await; + writer.close().await; + db.remove().await; + } + #[tokio::test] #[ignore = "requires the existing disposable postgres source suite"] async fn qualification_empty_query_poll_is_lock_free_only_when_proven() { diff --git a/rust/src/runtime/run_describe.rs b/rust/src/runtime/run_describe.rs index d76a21f2f..62ef6ae24 100644 --- a/rust/src/runtime/run_describe.rs +++ b/rust/src/runtime/run_describe.rs @@ -11,12 +11,17 @@ use super::{ backend::Backend, store::{Store, now}, task_queues::json_time, + transaction::Transaction, }; use chrono::{DateTime, Utc}; use serde_json::{Map, Value, json}; use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; use std::sync::OnceLock; +/// A run's `WorkflowStarted` payload. +pub(super) const STARTED_PAYLOAD_SQL: &str = + "SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='WorkflowStarted' ORDER BY sequence LIMIT 1"; + /// `server.limits.max_payload_bytes`, which describe reports and uses to /// omit oversized previews. fn max_payload_bytes() -> i64 { @@ -76,21 +81,34 @@ where /// The registered query route for a run row, with workers whose /// heartbeat is newer than `stale_after` seconds. pub(super) async fn query_route(&self, run: &DB::Row, stale_after: i64) -> Result { + let mut connection = self.pool.acquire().await?; + let started = Self::started_payload(&mut connection, &DB::string(run, "id")?).await?; + self.query_route_in(&mut connection, run, &started, stale_after).await + } + + /// The run's `WorkflowStarted` payload, or null when it has none. + pub(super) async fn started_payload(connection: &mut DB::Connection, run_id: &str) -> Result { + Self::query(STARTED_PAYLOAD_SQL) + .bind(run_id) + .fetch_optional(&mut *connection) + .await? + .map(|row| DB::document_row(&row, "payload")) + .transpose() + .map(|payload| payload.unwrap_or(Value::Null)) + } + + /// `query_route` on a given connection with the run's start payload. + pub(super) async fn query_route_in( + &self, connection: &mut DB::Connection, run: &DB::Row, started: &Value, stale_after: i64, + ) -> Result { let task_queue = super::query_route::trimmed(DB::optional_string(run, "queue")?.as_deref()).unwrap_or("default").to_owned(); let workflow_type = DB::optional_string(run, "workflow_type")?; let compatibility = DB::optional_string(run, "compatibility")?; - let started = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='WorkflowStarted' ORDER BY sequence LIMIT 1") - .bind(DB::string(run, "id")?) - .fetch_optional(&self.pool) - .await? - .map(|row| DB::document_row(&row, "payload")) - .transpose()? - .unwrap_or(Value::Null); let fresh_after = now() - chrono::Duration::seconds(stale_after.max(1)); let rows = Self::query("SELECT capabilities,supported_workflow_types,build_id,workflow_definition_fingerprints,last_heartbeat_at FROM workflow_worker_registrations WHERE namespace=$1 AND task_queue=$2 AND status='active' ORDER BY worker_id") .bind(self.namespace.as_ref()) .bind(task_queue.as_str()) - .fetch_all(&self.pool) + .fetch_all(&mut *connection) .await?; let mut workers = Vec::with_capacity(rows.len()); for row in rows { @@ -109,16 +127,15 @@ where &task_queue, super::query_route::trimmed(workflow_type.as_deref()), super::query_route::trimmed(compatibility.as_deref()), - &started, + started, &workers, )) } /// `WorkflowRun::typedSearchAttributes`, sorted by key; PHP renders an /// empty set as `[]`. - async fn typed_search_attributes(&self, run_id: &str) -> Result { - let mut connection = self.pool.acquire().await?; - let rows = Self::typed_rows(&mut connection, run_id).await?; + async fn typed_search_attributes(connection: &mut DB::Connection, run_id: &str) -> Result { + let rows = Self::typed_rows(connection, run_id).await?; Ok(if rows.is_empty() { json!([]) } else { @@ -186,7 +203,7 @@ where ..Default::default() }; rows.summary = next().into_iter().next(); - rows.search_attributes = self.typed_search_attributes(run_id).await?; + rows.search_attributes = Self::typed_search_attributes(&mut *self.pool.acquire().await?, run_id).await?; if let Some(summary) = &rows.summary { rows.task_problem_badge = super::run_diagnostics::task_problem_badge( summary["task_problem"].as_bool().unwrap_or(false), @@ -198,10 +215,10 @@ where } /// `runUpdateDiagnostics`: the run's updates in command order. - async fn update_diagnostics(&self, run_id: &str) -> Result> { + async fn update_diagnostics(connection: &mut DB::Connection, run_id: &str) -> Result> { let rows = Self::query("SELECT id,workflow_command_id,workflow_instance_id,workflow_run_id,target_scope,requested_workflow_run_id,resolved_workflow_run_id,update_name,status,outcome,command_sequence,workflow_sequence,rejection_reason,validation_errors,failure_id,failure_message,accepted_at,applied_at,rejected_at,closed_at FROM workflow_updates WHERE workflow_run_id=$1 ORDER BY command_sequence,created_at,id") .bind(run_id) - .fetch_all(&self.pool) + .fetch_all(&mut *connection) .await?; let mut updates = Vec::with_capacity(rows.len()); for row in rows { @@ -227,9 +244,15 @@ where /// A run's compatibility marker and whether the live fleet supports it, /// as describe and history report them. pub(super) async fn compatibility_fields(&self, run: &DB::Row) -> Result> { + let mut connection = self.pool.acquire().await?; + self.compatibility_fields_in(&mut connection, run).await + } + + /// `compatibility_fields` on a given connection. + pub(super) async fn compatibility_fields_in(&self, connection: &mut DB::Connection, run: &DB::Row) -> Result> { let compatibility = DB::optional_string(run, "compatibility")?; let queue = DB::optional_string(run, "queue")?; - let snapshots = self.fleet_snapshots(now()).await?; + let snapshots = self.fleet_snapshots_in(connection, now()).await?; let (supported, fleet_reason) = super::run_diagnostics::fleet_supports( &snapshots, compatibility.as_deref().map(str::trim).filter(|value| !value.is_empty()), @@ -244,8 +267,10 @@ where } /// The fields `formatRun` adds for a described run row. + /// Every read goes through the caller's one read snapshot (Server #1075). pub(super) async fn describe_fields( &self, + tx: &mut Transaction<'_, DB>, run: &DB::Row, current: bool, terminal: bool, @@ -255,27 +280,22 @@ where ) -> Result> { let run_id = DB::string(run, "id")?; let status = DB::string(run, "status")?; - let started = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='WorkflowStarted' ORDER BY sequence LIMIT 1") - .bind(run_id.as_str()) - .fetch_optional(&self.pool) - .await? - .map(|row| DB::document_row(&row, "payload")) - .transpose()? - .unwrap_or(Value::Null); + // One start-payload read serves the description and the query route. + let started = Self::started_payload(&mut **tx, &run_id).await?; let summary = Self::query("SELECT wait_kind,wait_reason FROM workflow_run_summaries WHERE id=$1") .bind(run_id.as_str()) - .fetch_optional(&self.pool) + .fetch_optional(&mut **tx) .await?; let (wait_kind, wait_reason) = match summary { Some(row) => (DB::optional_string(&row, "wait_kind")?, DB::optional_string(&row, "wait_reason")?), None => (None, None), }; - let compatibility = self.compatibility_fields(run).await?; + let compatibility = self.compatibility_fields_in(&mut **tx, run).await?; // `canServeQuery`: failed, cancelled and terminated runs refuse // queries; others need a registered route. let query_servable = current && (!terminal || status == "completed") - && self.query_route(run, stale_after).await?.servable; + && self.query_route_in(&mut **tx, run, &started, stale_after).await?.servable; let archived = DB::optional_instant(run, "archived_at")?.is_some(); let (input_omitted, output_omitted) = (oversized(input), oversized(output)); let mut fields = Map::new(); @@ -293,10 +313,7 @@ where let routing = if DB::optional_string(run, "import_source")?.as_deref() == Some(super::waterline_import::IMPORT_SOURCE) { Value::Null } else { - let mut tx = self.begin_read().await?; - let routing = self.definition_inspect(&mut tx, run, None).await?; - tx.commit().await?; - routing + self.definition_inspect(tx, run, None, Some(&started)).await? }; fields.insert("definition_routing".into(), routing); fields.insert("payload_previews".into(), json!({"input_omitted":input_omitted,"output_omitted":output_omitted, @@ -304,14 +321,12 @@ where fields.insert("last_progress_at".into(), iso8601(DB::optional_instant(run, "last_progress_at")?)); fields.insert("wait_kind".into(), json!(wait_kind)); fields.insert("wait_reason".into(), json!(wait_reason)); - fields.insert("search_attributes".into(), self.typed_search_attributes(&run_id).await?); + fields.insert("search_attributes".into(), Self::typed_search_attributes(&mut **tx, &run_id).await?); fields.insert("actions".into(), actions(current, terminal, archived, query_servable)); fields.insert("updates_scope".into(), json!("selected_run")); - fields.insert("updates".into(), json!(self.update_diagnostics(&run_id).await?)); + fields.insert("updates".into(), json!(Self::update_diagnostics(&mut **tx, &run_id).await?)); // `terminalFailurePayload`: a failed run's failure, last (Server #967). - let mut tx = self.begin_read().await?; - let failure = Self::terminal_failure_describe(&mut tx, &run_id, DB::string(run, "status")?.as_str()).await?; - tx.commit().await?; + let failure = Self::terminal_failure_describe(tx, &run_id, DB::string(run, "status")?.as_str()).await?; for (key, value) in failure.into_iter().flatten() { fields.insert(key.into(), value); } diff --git a/rust/src/runtime/run_diagnostics.rs b/rust/src/runtime/run_diagnostics.rs index 03c4bca78..d76bd7bba 100644 --- a/rust/src/runtime/run_diagnostics.rs +++ b/rust/src/runtime/run_diagnostics.rs @@ -428,8 +428,16 @@ where /// `WorkflowCompatibilityFleet::detailsForNamespace` from the /// registrations whose heartbeat is within the fleet window. pub(super) async fn fleet_snapshots(&self, at: DateTime) -> Result, Vec, DateTime)>> { + let mut connection = self.pool.acquire().await?; + self.fleet_snapshots_in(&mut connection, at).await + } + + /// `fleet_snapshots` on a given connection. + pub(super) async fn fleet_snapshots_in( + &self, connection: &mut DB::Connection, at: DateTime, + ) -> Result, Vec, DateTime)>> { let rows = Self::query("SELECT worker_id,task_queue,build_id,last_heartbeat_at FROM workflow_worker_registrations WHERE namespace=$1 AND last_heartbeat_at IS NOT NULL") - .bind(self.namespace.as_ref()).fetch_all(&self.pool).await?; + .bind(self.namespace.as_ref()).fetch_all(&mut *connection).await?; let mut snapshots = Vec::new(); for row in rows { let Some(heartbeat) = DB::optional_instant(&row, "last_heartbeat_at")? else { continue }; @@ -838,7 +846,7 @@ where // `WorkflowRunDiagnostics`: the definition route of the next ready // task, and a warning when no eligible worker can take it. let mut tx = self.begin_read().await?; - let routing = self.definition_inspect(&mut tx, &run, None).await?; + let routing = self.definition_inspect(&mut tx, &run, None, None).await?; tx.commit().await?; if routing["blocked"] == true { findings.push(json!({"severity": "warning", "code": "workflow_definition_routing_blocked", diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 52adb626c..2ebe789c4 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -793,9 +793,15 @@ where /// A run's decoded memo map as describe and the run list render it. pub(super) async fn run_memo(&self, run_id: &str) -> Result { + let mut connection = self.pool.acquire().await?; + self.run_memo_in(&mut connection, run_id).await + } + + /// `run_memo` on a given connection. + pub(super) async fn run_memo_in(&self, connection: &mut DB::Connection, run_id: &str) -> Result { let memo_rows = Self::query(DB::MEMO_READ) .bind(run_id) - .fetch_all(&self.pool) + .fetch_all(&mut *connection) .await?; let mut stored = Vec::with_capacity(memo_rows.len()); for memo in memo_rows { @@ -824,12 +830,27 @@ where /// The run description; with a worker stale window it also carries the /// fields PHP's `formatRun` adds (actions, fleet, wait and updates). + /// + /// The whole description reads one bounded read-only snapshot on one + /// connection (Server #1075), even below session REPEATABLE READ. Errors + /// and cancellation drop the transaction and return its connection. pub(crate) async fn describe_with(&self, workflow_id: &str, run_id: Option<&str>, stale_after: Option) -> Result { + let mut tx = self.begin_repeatable_read().await?; + let described = self.describe_in(&mut tx, workflow_id, run_id, stale_after).await?; + tx.commit().await?; + Ok(described) + } + + pub(super) async fn describe_in( + &self, tx: &mut Transaction<'_, DB>, workflow_id: &str, run_id: Option<&str>, stale_after: Option, + ) -> Result { let row = Self::query("SELECT r.*,i.execution_timeout_seconds,i.current_run_id,(SELECT parent_workflow_run_id FROM workflow_links l WHERE l.child_workflow_run_id=r.id AND l.link_type='continue_as_new' LIMIT 1) AS continued_from_run_id,(SELECT COUNT(*) FROM workflow_runs counted WHERE counted.workflow_instance_id=i.id) AS run_count FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE i.id=$1 AND r.id=COALESCE($2,i.current_run_id) AND r.namespace=$3") - .bind(workflow_id).bind(run_id).bind(self.namespace.as_ref()).fetch_optional(&self.pool).await? + .bind(workflow_id).bind(run_id).bind(self.namespace.as_ref()).fetch_optional(&mut **tx).await? .ok_or_else(|| refuse(StatusCode::NOT_FOUND, if run_id.is_some() {"run_not_found"} else {"instance_not_found"}))?; + #[cfg(test)] + super::describe_snapshot_tests::after_run_row().await; let output = DB::optional_string(&row, "output")?; - let memo = self.run_memo(&DB::string(&row, "id")?).await?; + let memo = self.run_memo_in(&mut **tx, &DB::string(&row, "id")?).await?; let status = DB::string(&row, "status")?; let terminal = matches!( status.as_str(), @@ -842,7 +863,7 @@ where // Only a stored external reference is read, to tell whether its preview // is omitted; inline payloads stay unread. let command_rows = Self::query(DB::RUN_COMMANDS_SQL) - .bind(DB::string(&row, "id")?).bind(workflow_id).fetch_all(&self.pool).await?; + .bind(DB::string(&row, "id")?).bind(workflow_id).fetch_all(&mut **tx).await?; let mut commands = Vec::with_capacity(command_rows.len()); let mut reason_codec = None; for command in command_rows { @@ -913,7 +934,7 @@ where "run_deadline_at": super::run_describe::iso8601(DB::optional_instant(&row,"run_deadline_at")?)}); if let Some(stale_after) = stale_after { let fields = self - .describe_fields(&row, current, terminal, &described["input_envelope"], &described["output_envelope"], stale_after) + .describe_fields(tx, &row, current, terminal, &described["input_envelope"], &described["output_envelope"], stale_after) .await?; described.as_object_mut().expect("description object").extend(fields); } @@ -921,10 +942,10 @@ where // description, every action off, the stored codec and progress time, // and the projection's metadata. if DB::optional_string(&row, "import_source")?.as_deref() == Some(super::waterline_import::IMPORT_SOURCE) { - let metadata = self.projection_metadata(&DB::string(&row, "id")?, DB::optional_instant(&row, "imported_at")?, + let metadata = Self::projection_metadata_in(&mut **tx, &DB::string(&row, "id")?, DB::optional_instant(&row, "imported_at")?, DB::optional_string(&row, "import_dedupe_key")?).await?; let failure = if stale_after.is_some() && described["status"] == "failed" { - self.projected_failure(&DB::string(&row, "id")?).await? + Self::projected_failure_in(&mut **tx, &DB::string(&row, "id")?).await? } else { Value::Null }; @@ -2526,7 +2547,7 @@ where let required = DB::optional_string(&task, "compatibility")? .map(|build| build.trim().to_owned()) .filter(|build| !build.is_empty()); - let context = Self::definition_context(tx, &run, Some(queue), required).await?; + let context = Self::definition_context(tx, &run, Some(queue), required, None).await?; if !self.definition_allows(tx, &run, &worker, &context).await?.0 { continue; } diff --git a/rust/src/runtime/task_repair.rs b/rust/src/runtime/task_repair.rs index f35a00d56..fb361e1eb 100644 --- a/rust/src/runtime/task_repair.rs +++ b/rust/src/runtime/task_repair.rs @@ -114,7 +114,7 @@ where .bind(workflow_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?, }; if let Some(run) = run { - response["definition_routing"] = self.definition_inspect(&mut tx, &run, None).await?; + response["definition_routing"] = self.definition_inspect(&mut tx, &run, None, None).await?; } tx.commit().await?; Ok((status, response)) diff --git a/rust/src/runtime/waterline_import.rs b/rust/src/runtime/waterline_import.rs index d631d3679..ff8f9256e 100644 --- a/rust/src/runtime/waterline_import.rs +++ b/rust/src/runtime/waterline_import.rs @@ -921,10 +921,10 @@ store_methods! { /// `terminalFailurePayload` for a failed projection, which has no failure /// snapshots: its activity failure summaries and their last message, or /// null when it has none. - pub(crate) async fn projected_failure(&self, run_id: &str) -> Result { + pub(crate) async fn projected_failure_in(connection: &mut DB::Connection, run_id: &str) -> Result { let rows = Self::query("SELECT sequence,event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityFailed','ActivityTimedOut') ORDER BY sequence") .bind(run_id) - .fetch_all(&self.pool) + .fetch_all(&mut *connection) .await?; let mut summaries = Vec::with_capacity(rows.len()); for row in &rows { @@ -958,9 +958,17 @@ store_methods! { /// `LegacyV1Projection::metadataForRun` for a projected run: the started /// event's projection context and the projection's own description. pub(crate) async fn projection_metadata(&self, run_id: &str, imported_at: Option>, dedupe_key: Option) -> Result { + let mut connection = self.pool.acquire().await?; + Self::projection_metadata_in(&mut connection, run_id, imported_at, dedupe_key).await + } + + /// `projection_metadata` on a given connection. + pub(crate) async fn projection_metadata_in( + connection: &mut DB::Connection, run_id: &str, imported_at: Option>, dedupe_key: Option, + ) -> Result { let started = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 ORDER BY sequence LIMIT 1") .bind(run_id) - .fetch_optional(&self.pool) + .fetch_optional(&mut *connection) .await? .map(|row| DB::document_row(&row, "payload")) .transpose()?