From eb35eef59b709ab34f7a47338c729f3a15d83d49 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 9 Nov 2026 05:35:21 +0000 Subject: [PATCH] Describe workflow runs through one repeatable read snapshot Reuse the description connection for memo, command, summary, routing, search, update, failure and projection reads. Share the native start payload with query and definition routing. Force the existing repeatable snapshot primitive even on MySQL/MariaDB READ COMMITTED sessions. Add real concurrent-termination controls, a READ COMMITTED counterexample, one-checkout/start-read counts, and error/cancellation release controls. First compile and execution are deferred to the normal hosted source train. Refs #1075. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: Codex --- rust/src/runtime/definition_routing.rs | 23 ++-- rust/src/runtime/describe_snapshot_tests.rs | 135 ++++++++++++++++++++ rust/src/runtime/mod.rs | 2 + rust/src/runtime/mysql.rs | 28 ++++ rust/src/runtime/postgres.rs | 17 +++ rust/src/runtime/run_describe.rs | 85 +++++++----- rust/src/runtime/run_diagnostics.rs | 12 +- rust/src/runtime/store.rs | 37 ++++-- rust/src/runtime/task_repair.rs | 2 +- rust/src/runtime/waterline_import.rs | 14 +- 10 files changed, 297 insertions(+), 58 deletions(-) create mode 100644 rust/src/runtime/describe_snapshot_tests.rs 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()?