diff --git a/rust/src/runtime/active_summary.rs b/rust/src/runtime/active_summary.rs index 53d60f7ae..73f9fbf25 100644 --- a/rust/src/runtime/active_summary.rs +++ b/rust/src/runtime/active_summary.rs @@ -13,6 +13,11 @@ use super::{ use chrono::{DateTime, Utc}; use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; +// One aggregate scan for each source table, including the empty-table case. +const PROJECTION_CENSUS_SQL: &str = "SELECT waits.*,tasks.*,(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$1 AND status='received') AS received_signals,(SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$2) AS failures FROM (SELECT COUNT(CASE WHEN status='open' AND kind NOT IN ('condition','signal','timer','activity','child','update') THEN 1 END) AS other_waits,COUNT(CASE WHEN (status='open' OR (status='resolved' AND source_status='timed_out')) AND kind IN ('condition','signal') THEN 1 END) AS external_waits,COUNT(CASE WHEN kind='timer' AND status='open' THEN 1 END) AS timers,COUNT(CASE WHEN kind='activity' AND status='open' THEN 1 END) AS activities,COUNT(CASE WHEN kind='child' THEN 1 END) AS children,COUNT(CASE WHEN kind='update' THEN 1 END) AS updates FROM workflow_run_waits WHERE workflow_run_id=$3) waits CROSS JOIN (SELECT COUNT(CASE WHEN task_type='workflow' AND repair_count>0 THEN 1 END) AS repairs,COUNT(CASE WHEN task_type='workflow' AND status='failed' THEN 1 END) AS failed_tasks FROM workflow_tasks WHERE workflow_run_id=$4) tasks"; +#[cfg(test)] +const PREVIOUS_CENSUS_SQL: &str = "SELECT (SELECT COUNT(*) FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open' AND kind NOT IN ('condition','signal','timer','activity','child','update')) AS other_waits,(SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$2 AND task_type='workflow' AND repair_count>0) AS repairs,(SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$3 AND task_type='workflow' AND status='failed') AS failed_tasks,(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$4 AND status='received') AS received_signals,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$5 AND (w.status='open' OR (w.status='resolved' AND w.source_status='timed_out')) AND w.kind IN ('condition','signal')) AS external_waits,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$6 AND w.kind='timer' AND w.status='open') AS timers,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$7 AND w.kind='activity' AND w.status='open') AS activities,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$8 AND w.kind='child') AS children,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$9 AND w.kind='update') AS updates,(SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$10) AS failures"; + pub(super) struct ActiveSummary { pub(super) state: String, pub(super) reason: String, @@ -27,6 +32,7 @@ pub(super) struct ActiveSummary { } /// Each zero count skips only the corresponding read, never its validation. +#[cfg_attr(test, derive(Debug, PartialEq, Eq))] pub(super) struct ProjectionCensus { other_waits: i64, repairs: i64, @@ -63,9 +69,8 @@ where usize: ColumnIndex, { pub(super) async fn projection_census(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result { - let row = Self::query("SELECT (SELECT COUNT(*) FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open' AND kind NOT IN ('condition','signal','timer','activity','child','update')) AS other_waits,(SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$2 AND task_type='workflow' AND repair_count>0) AS repairs,(SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$3 AND task_type='workflow' AND status='failed') AS failed_tasks,(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$4 AND status='received') AS received_signals,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$5 AND (w.status='open' OR (w.status='resolved' AND w.source_status='timed_out')) AND w.kind IN ('condition','signal')) AS external_waits,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$6 AND w.kind='timer' AND w.status='open') AS timers,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$7 AND w.kind='activity' AND w.status='open') AS activities,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$8 AND w.kind='child') AS children,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$9 AND w.kind='update') AS updates,(SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$10) AS failures") - .bind(run_id).bind(run_id).bind(run_id).bind(run_id).bind(run_id) - .bind(run_id).bind(run_id).bind(run_id).bind(run_id).bind(run_id) + let row = Self::query(PROJECTION_CENSUS_SQL) + .bind(run_id).bind(run_id).bind(run_id).bind(run_id) .fetch_one(&mut **tx).await?; Ok(ProjectionCensus { other_waits: DB::number(&row,"other_waits")?, repairs: DB::number(&row,"repairs")?, @@ -76,6 +81,90 @@ where }) } + #[cfg(test)] + async fn census_values(tx: &mut Transaction<'_,DB>, run_id: &str, sql: &'static str, binds: usize) -> [i64;10] { + let mut query = Self::query(sql); + for _ in 0..binds { query = query.bind(run_id); } + let row = query.fetch_one(&mut **tx).await.unwrap(); + ["other_waits","repairs","failed_tasks","received_signals","external_waits","timers","activities","children","updates","failures"] + .map(|name| DB::number(&row,name).unwrap()) + } + + #[cfg(test)] + async fn census_scans(tx: &mut Transaction<'_,DB>, run_id: &str, sql: &'static str, binds: usize) -> (usize,usize) { + let prefix = match DB::NAME {"PostgreSQL"=>"EXPLAIN (FORMAT JSON) ","SQLite"=>"EXPLAIN QUERY PLAN ",_=>"EXPLAIN "}; + let mut query = sqlx::query(sqlx::AssertSqlSafe(format!("{prefix}{}",DB::statement(sql)))); + for _ in 0..binds { query = query.bind(run_id); } + let rows = query.fetch_all(&mut **tx).await.unwrap(); + if DB::NAME == "PostgreSQL" { + fn visit(value: &serde_json::Value, counts: &mut (usize,usize)) { + if value["Relation Name"] == "workflow_run_waits" { counts.0 += 1; } + if value["Relation Name"] == "workflow_tasks" { counts.1 += 1; } + match value { + serde_json::Value::Object(map) => { for child in map.values() { visit(child,counts); } } + serde_json::Value::Array(items) => { for child in items { visit(child,counts); } } + _ => {} + } + } + let mut counts = (0,0); + visit(&DB::document_row(&rows[0],"QUERY PLAN").unwrap(),&mut counts); + return counts; + } + let mut counts = (0,0); + for row in rows { + let table = if DB::NAME == "SQLite" { + let detail = DB::string(&row,"detail").unwrap(); + let mut parts = detail.split_whitespace(); + match parts.next() { + Some("SCAN" | "SEARCH") => parts.next().map(str::to_owned), + _ => None, + } + } else { DB::optional_string(&row,"table").unwrap() }; + match table.as_deref() { + Some("workflow_run_waits" | "w") => counts.0 += 1, + Some("workflow_tasks") => counts.1 += 1, + _ => {} + } + } + counts + } + + #[cfg(test)] + pub(super) async fn qualify_projection_census_passes(&self, run_id: &str, instance_id: &str) { + let mut tx = self.begin().await.unwrap(); + for absent in [run_id,"missing-census-run"] { + assert_eq!(Self::census_values(&mut tx,absent,PROJECTION_CENSUS_SQL,4).await, + Self::census_values(&mut tx,absent,PREVIOUS_CENSUS_SQL,10).await); + } + let mut position = 0_i64; + for kind in ["other","condition","signal","timer","activity","child","update"] { + for status in ["open","resolved","cancelled"] { + for source in [None,Some("timed_out"),Some("completed")] { + Self::query("INSERT INTO workflow_run_waits(id,workflow_run_id,workflow_instance_id,wait_id,position,kind,status,source_status) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)") + .bind(super::store::id()).bind(run_id).bind(instance_id).bind(format!("census-{position}")) + .bind(position).bind(kind).bind(status).bind(source).execute(&mut *tx).await.unwrap(); + position += 1; + } + } + } + assert_eq!(Self::census_values(&mut tx,run_id,PROJECTION_CENSUS_SQL,4).await, + Self::census_values(&mut tx,run_id,PREVIOUS_CENSUS_SQL,10).await); + for kind in ["workflow","activity"] { + for status in ["ready","leased","failed","completed"] { + for repairs in [0_i64,1,2] { + Self::query("INSERT INTO workflow_tasks(id,workflow_run_id,namespace,queue,task_type,status,repair_count) VALUES ($1,$2,'default','census',$3,$4,$5)") + .bind(super::store::id()).bind(run_id).bind(kind).bind(status).bind(repairs) + .execute(&mut *tx).await.unwrap(); + } + } + } + assert_eq!(Self::census_values(&mut tx,run_id,PROJECTION_CENSUS_SQL,4).await, + Self::census_values(&mut tx,run_id,PREVIOUS_CENSUS_SQL,10).await); + assert_eq!(Self::census_scans(&mut tx,run_id,PREVIOUS_CENSUS_SQL,10).await,(6,2)); + assert_eq!(Self::census_scans(&mut tx,run_id,PROJECTION_CENSUS_SQL,4).await,(1,1)); + tx.rollback().await.unwrap(); + } + #[cfg(test)] pub(super) async fn projection_census_reference(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result { // The old projector always performed these reads. Keep that path diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 03b027f55..55d87bef2 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -92,6 +92,7 @@ pub(super) trait Backend: Database { // Waits whose stored position differs from their chronological ordinal // over every retained wait of the run, as Workflow #941's projector stores. const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,EXTRACT(EPOCH FROM opened_at) AS seconds FROM workflow_run_waits WHERE workflow_run_id=$1), ordinals AS (SELECT id,position,ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN seconds<0 THEN 2*FLOOR(seconds)-seconds ELSE seconds END)*1000),wait_id)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE position IS DISTINCT FROM ordinal ORDER BY ordinal LIMIT 100"; + const SUMMARY_UPSERT_SQL: &str = super::terminal_summary::SUMMARY_UPSERT_SQL; const SUMMARY_HISTORY_SQL: &str = "SELECT id,sequence,event_type,CAST(payload AS TEXT) AS payload_json FROM workflow_history_events WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT 16"; const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND payload->>'failure_id'=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1"; const OBSERVATION_ATTRIBUTES: &str = "SELECT \"key\" AS attribute_key,type,value_string,value_keyword,value_int,value_float,CASE WHEN value_bool IS NULL THEN NULL WHEN value_bool THEN 1 ELSE 0 END AS value_bool,value_datetime,value_keyword_list FROM workflow_search_attributes WHERE workflow_run_id=$1 AND workflow_instance_id=$2 AND \"key\" IN ($3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22) ORDER BY \"key\" LIMIT 20"; @@ -126,6 +127,7 @@ pub(super) trait Backend: Database { const QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND value::jsonb->>'namespace'=$1 ORDER BY \"key\" LIMIT 65"; const QUERY_CACHE_GET: &str = "SELECT value,expiration FROM dw_query_cache WHERE \"key\"=$1"; + const QUERY_CACHE_GET_PAIR: &str = "SELECT 0 AS cache_part,value,expiration FROM dw_query_cache WHERE \"key\"=$1 UNION ALL SELECT 1 AS cache_part,value,expiration FROM dw_query_cache WHERE \"key\"=$2"; const QUERY_CACHE_INSERT: &str = "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3)"; const QUERY_CACHE_UPDATE: &str = @@ -623,6 +625,7 @@ impl Backend for MySql { // The run-and-sequence index keeps MySQL from sorting whole rows: a // filesort of one large payload exceeds the default sort buffer and // fails the transaction ("Out of sort memory"). + const SUMMARY_UPSERT_SQL: &str = super::terminal_summary::summary_projection_sql!("REPLACE", ""); const SUMMARY_HISTORY_SQL: &str = "SELECT id,sequence,event_type,CAST(payload AS CHAR) AS payload_json FROM workflow_history_events FORCE INDEX (workflow_history_events_workflow_run_id_sequence_unique) WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT 16"; const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=? AND CAST(JSON_UNQUOTE(JSON_EXTRACT(payload,'$.failure_id')) AS BINARY)=CAST(? AS BINARY) AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1"; const OBSERVATION_ATTRIBUTES: &str = "SELECT `key` AS attribute_key,type,value_string,value_keyword,value_int,value_float,CASE WHEN value_bool IS NULL THEN NULL WHEN value_bool THEN 1 ELSE 0 END AS value_bool,value_datetime,value_keyword_list FROM workflow_search_attributes WHERE workflow_run_id=$1 AND workflow_instance_id=$2 AND `key` IN ($3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22) ORDER BY `key` LIMIT 20"; @@ -650,6 +653,7 @@ impl Backend for MySql { type EncodedDocument = Json; const QUERY_CACHE_GET: &str = "SELECT value,expiration FROM dw_query_cache WHERE `key`=$1"; + const QUERY_CACHE_GET_PAIR: &str = "SELECT 0 AS cache_part,value,expiration FROM dw_query_cache WHERE `key`=$1 UNION ALL SELECT 1 AS cache_part,value,expiration FROM dw_query_cache WHERE `key`=$2"; const QUERY_CACHE_INSERT: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3)"; const QUERY_CACHE_UPDATE: &str = diff --git a/rust/src/runtime/carried_polls.rs b/rust/src/runtime/carried_polls.rs index 34f0ff921..5614a0d9d 100644 --- a/rust/src/runtime/carried_polls.rs +++ b/rust/src/runtime/carried_polls.rs @@ -165,9 +165,10 @@ store_methods! { &self, tx: &mut super::transaction::Transaction<'_, DB>, poll: &ActivityPoll, completion_owner: &str, ) -> Result { let owner = super::text(&poll.body, "worker_id")?; - if owner != completion_owner || !self.worker_poll_fence_holds_in(tx, &poll.fence, owner).await? { + if owner != completion_owner { return Ok(Outcome::Declined); } + let Some(registration) = self.fenced_claim_registration(tx,&poll.fence,owner).await? else { return Ok(Outcome::Declined); - } + }; // Validation work uses the workflow slot too. An unverifiable lease // deadline remains protected, rather than being treated as free. let at = super::store::now(); @@ -179,7 +180,7 @@ store_methods! { return Ok(Outcome::Declined); } let saved = tx.poll_savepoint().await?; - match self.claim_in(tx, &poll.body, "activity", &poll.protocol, None, None).await { + match self.claim_with_registration(tx, &poll.body, "activity", &poll.protocol, None, None, Some(registration)).await { Ok(task) => { tx.release_poll_claim().await?; Ok(task.map_or(Outcome::Empty, Outcome::Leased)) @@ -213,9 +214,10 @@ store_methods! { &self, tx: &mut super::transaction::Transaction<'_, DB>, poll: &WorkflowPoll, completion_owner: &str, ) -> Result { let owner = super::text(&poll.body, "worker_id")?; - if owner != completion_owner || !self.worker_poll_fence_holds_in(tx, &poll.fence, owner).await? { + if owner != completion_owner { return Ok(Outcome::Declined); } + let Some(registration) = self.fenced_claim_registration(tx,&poll.fence,owner).await? else { return Ok(Outcome::Declined); - } + }; let occupied = Self::scalar("SELECT COUNT(*) FROM workflow_tasks WHERE namespace=$1 AND lease_owner=$2 AND task_type='activity' AND status='leased' AND (lease_expires_at IS NULL OR lease_expires_at>$3)") .bind(self.namespace.as_ref()).bind(owner).bind(DB::bind_time(super::store::now())).fetch_one(&mut **tx).await?; if occupied > 0 { return Ok(Outcome::Declined); } @@ -226,7 +228,7 @@ store_methods! { if let Some(answer) = self.replay_workflow_poll_in(tx, &poll.request, owner, super::text(&poll.body, "task_queue")?).await? { return Ok(if answer["task"].is_object() { Outcome::Leased(answer["task"].clone()) } else { Outcome::Empty }); } - let task = self.claim_in(tx, &poll.body, "workflow", &poll.protocol, poll.page, Some(&poll.request)).await?; + let task = self.claim_with_registration(tx, &poll.body, "workflow", &poll.protocol, poll.page, Some(&poll.request), Some(registration)).await?; if let Some(task) = task { Self::remember_workflow_poll(tx, &poll.request, &json!({"task":task,"poll_status":"leased"})).await?; Ok(Outcome::Leased(task)) diff --git a/rust/src/runtime/commit_handoff_tests.rs b/rust/src/runtime/commit_handoff_tests.rs index fbd0003b0..63292965b 100644 --- a/rust/src/runtime/commit_handoff_tests.rs +++ b/rust/src/runtime/commit_handoff_tests.rs @@ -159,7 +159,7 @@ store_methods! { Default::default(), None, None, )).await; assert_eq!(result.unwrap()["recorded"], true); - assert_eq!(statements.count("SELECT sequence FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1"), 1, + assert_eq!(statements.count(super::store::AUTHORED_RUN_SQL), 1, "scope validation and command application share one authored-sequence scan"); let rows = Self::query("SELECT sequence,event_type FROM workflow_history_events WHERE workflow_run_id=$1 ORDER BY sequence") .bind(started["run_id"].as_str().unwrap()).fetch_all(&self.pool).await.unwrap(); diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 92bd2f8ae..7d0b5fa02 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1197,6 +1197,9 @@ mod tests { async fn qualification_poll_cache_single_statement_writes() { let db = Database::new().await; let store = super::super::store::Store::new(MySqlStorage::open(db.options.clone()).await.unwrap().into_pool()); + store.qualify_completion_waiting_status().await; + store.qualify_authored_run_sequences().await; + store.qualify_transaction_run_tags().await; store.qualify_query_poll_cache_writes().await; store.qualify_workflow_poll_cache_writes().await; store.qualify_summary_projection_bytes().await; @@ -1226,6 +1229,7 @@ mod tests { store.qualify_workflow_poll_transactions().await; store.qualify_first_candidate_batch().await; store.qualify_claim_history_sequence().await; + store.qualify_fenced_registration_reuse().await; store.qualify_query_poll_transactions().await; store.close().await; db.remove().await; diff --git a/rust/src/runtime/poll_transaction_tests.rs b/rust/src/runtime/poll_transaction_tests.rs index 8166571a3..20ed68650 100644 --- a/rust/src/runtime/poll_transaction_tests.rs +++ b/rust/src/runtime/poll_transaction_tests.rs @@ -44,8 +44,14 @@ store_methods! { let name = format!("counted-workflow-{multiplexed}"); self.count_start(&name).await; self.count_reset(); - let claimed = self.count_poll(&name, multiplexed).await; + let (claimed, reads) = super::sql_counts::measure(self.count_poll(&name, multiplexed)).await; assert!(claimed["task"].is_object()); + assert_eq!(reads.count(super::worker_poll_fence::FENCE_REGISTRATION),1); + // The workflow claim reuses the row its in-transaction fence read. + // A multiplexed probe tries the validation kind first here, and + // claim_update_validation_in keeps its own registration read. + assert_eq!(reads.count("SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2"),usize::from(multiplexed), + "the workflow claim must reuse the registration its in-transaction fence validated"); assert_eq!(self.count_get(), 1, "one replay/claim probe, multiplexed={multiplexed}"); self.count_reset(); let replayed = self.count_poll(&name, multiplexed).await; @@ -227,6 +233,37 @@ store_methods! { .bind(run_id).execute(&self.pool).await.unwrap(); } + pub(super) async fn qualify_fenced_registration_reuse(&self) { + let registration = json!({"worker_id":"fence-reuse","task_queue":"fence-reuse","runtime":"php", + "supported_workflow_types":["fence-reuse"],"supported_activity_types":["fence-reuse"]}); + self.register(registration.clone(),&QueryTaskConfig::default(),Some("1.20")).await.unwrap(); + let super::worker_poll_fence::PollAdmission::Fenced(fence) = self.admit_worker_poll("fence-reuse","fence-reuse").await.unwrap() + else { panic!("fresh registration must admit its original poll"); }; + let mut tx = self.begin().await.unwrap(); + let (row,reads) = super::sql_counts::measure(self.fenced_claim_registration(&mut tx,&fence,"fence-reuse")).await; + let row = row.unwrap().unwrap(); + assert_eq!(DB::string(&row,"worker_id").unwrap(),"fence-reuse"); + assert_eq!(DB::string(&row,"namespace").unwrap(),self.namespace.as_ref()); + assert_eq!(reads.total(),1); + assert_eq!(reads.count(super::worker_poll_fence::FENCE_REGISTRATION),1); + tx.rollback().await.unwrap(); + // No process or clock change is needed: native re-registration rotates + // the authority token even inside the same stored second. + self.register(registration,&QueryTaskConfig::default(),Some("1.20")).await.unwrap(); + let mut tx = self.begin().await.unwrap(); + assert!(self.fenced_claim_registration(&mut tx,&fence,"fence-reuse").await.unwrap().is_none()); + tx.rollback().await.unwrap(); + // An unfenced internal claim still performs its original registration + // read. The optional optimization cannot bypass that fallback. + let body = json!({"worker_id":"fence-reuse","task_queue":"fence-reuse"}); + let mut tx = self.begin().await.unwrap(); + let (claimed,reads) = super::sql_counts::measure(self.claim_in(&mut tx,&body,"activity","1.20",None,None)).await; + assert!(claimed.unwrap().is_none()); + assert_eq!(reads.count("SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2"),1); + assert_eq!(reads.count(super::worker_poll_fence::FENCE_REGISTRATION),0); + tx.rollback().await.unwrap(); + } + pub(super) async fn qualify_query_poll_transactions(&self) { self.count_worker().await; let workflow_id = "counted-query-workflow"; @@ -294,6 +331,7 @@ async fn sqlite_workflow_and_query_poll_transaction_counts() { store.qualify_workflow_poll_transactions().await; store.qualify_first_candidate_batch().await; store.qualify_claim_history_sequence().await; + store.qualify_fenced_registration_reuse().await; store.qualify_query_poll_transactions().await; store.close().await; } diff --git a/rust/src/runtime/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index 946d6e1a8..5170f33e5 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -1,15 +1,208 @@ -//! The actual cache writers prepare and execute exactly one statement per row. +//! Compare stored projections and count the actual cache/projection writes. use super::store::Store; use serde_json::json; use std::collections::BTreeMap; store_methods! { + pub(super) async fn qualify_completion_waiting_status(&self) { + let input = crate::codec::ValueCodec::new().unwrap() + .encode(&crate::codec::Value::Array(vec![])).unwrap(); + self.register(json!({"worker_id":"waiting-status-worker","task_queue":"waiting-status", + "runtime":"php","supported_workflow_types":["waiting-status"],"supported_activity_types":["waiting-status"]}), + &super::QueryTaskConfig::default(),Some("1.20")).await.unwrap(); + let cases = [ + ("terminal",vec![json!({"type":"complete_workflow","result":{"codec":"avro","blob":input}})],false,"completed",0_usize), + ("activity",vec![json!({"type":"schedule_activity","activity_type":"waiting-status", + "arguments":{"codec":"avro","blob":input}})],false,"waiting",0), + // A completion needs at least one command. A side effect changes + // no run status itself, so only the final write can settle it. + ("side-effect",vec![json!({"type":"record_side_effect","result":{"codec":"avro","blob":input}})],false,"waiting",1), + ("already-waiting",vec![json!({"type":"record_side_effect","result":{"codec":"avro","blob":input}})],true,"waiting",0), + ]; + for (name,commands,waiting,expected,updates) in cases { + let run = self.start(json!({"workflow_id":format!("waiting-status-{name}"), + "workflow_type":"waiting-status","task_queue":"waiting-status","input":{"codec":"avro","blob":input}}), + BTreeMap::new(),None,None).await.unwrap(); + let run_id = run["run_id"].as_str().unwrap(); + let task = self.poll(json!({"worker_id":"waiting-status-worker","task_queue":"waiting-status","timeout_seconds":0}), + "workflow","1.20").await.unwrap()["task"].clone(); + assert_eq!(task["run_id"],run_id); + if waiting { + Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1") + .bind(run_id).execute(&self.pool).await.unwrap(); + } + let (answer,count) = super::sql_counts::measure(self.complete_or_checkpoint( + task["task_id"].as_str().unwrap(),json!({"lease_owner":"waiting-status-worker", + "workflow_task_attempt":task["workflow_task_attempt"],"commands":commands}), + std::sync::Arc::new(tokio::sync::Semaphore::new(1)),BTreeMap::new(),None, + super::carried_polls::Config {protocol:Some("1.20"),pages:Default::default(),default_wait:30}, + Default::default(),None,None)).await; + let answer = answer.unwrap(); + assert_eq!(answer["recorded"],true); + assert_eq!(answer["run_status"],expected); + assert_eq!(count.count("UPDATE workflow_runs SET status='waiting' WHERE id=$1 AND status IN ('pending','running')"),updates, + "only pending/running needs the final waiting write: {name}"); + let row = Self::query("SELECT status FROM workflow_runs WHERE id=$1").bind(run_id) + .fetch_one(&self.pool).await.unwrap(); + assert_eq!(DB::string(&row,"status").unwrap(),expected); + } + } + + pub(super) async fn qualify_authored_run_sequences(&self) { + let input = crate::codec::ValueCodec::new().unwrap() + .encode(&crate::codec::Value::Array(vec![])).unwrap(); + let run = self.start(json!({"workflow_id":"authored-counter-control","workflow_type":"authored-counter", + "task_queue":"authored-counter","input":{"codec":"avro","blob":input}}), BTreeMap::new(), None, None).await.unwrap(); + let foreign = self.start(json!({"workflow_id":"authored-counter-foreign","workflow_type":"authored-counter", + "task_queue":"authored-counter","input":{"codec":"avro","blob":input}}), BTreeMap::new(), None, None).await.unwrap(); + let run_id = run["run_id"].as_str().unwrap(); + // Physical counters, latest logical history, scope spans and redrive + // all contribute independently. Control command order contributes none. + let cases = [ + (vec![], vec![], vec![], None::, 1_i64, vec![], 0_i64), + (vec![2_i64,7,3], vec![4_i64], vec![6_i64], None, 1, vec![], 7), + (vec![2], vec![8,4], vec![6], None, 1, vec![], 8), + (vec![2], vec![4], vec![6,9], None, 1, vec![], 9), + (vec![0], vec![0], vec![0], Some(0), 1, vec![], 0), + (vec![], vec![], vec![], Some(37), 1, vec![], 37), + (vec![], vec![], vec![], None, 1, + vec![("SignalWaitOpened", json!({"sequence":99})), ("ConditionWaitOpened", json!({"sequence":11}))], 11), + (vec![], vec![], vec![], None, 1, + vec![("VersionMarkerRecorded", json!({"sequence":12})), ("MemoUpserted", json!({"sequence":13})), + ("SideEffectRecorded", json!({"sequence":14})), ("SearchAttributesUpserted", json!({"sequence":15}))], 15), + (vec![], vec![], vec![], None, 1, + vec![("CancellationScopeOpened", json!({"sequence":17,"sequence_span":100})), + ("CancellationScopeDelivered", json!({"sequence":20,"sequence_span":4}))], 23), + (vec![], vec![], vec![], None, 1, + vec![("CancellationScopeDelivered", json!({"sequence":i64::MAX-1,"sequence_span":4}))], i64::MAX-1), + (vec![], vec![], vec![], None, 2, vec![], 39), + ]; + for (activities,timers,children,cancellation,run_number,events,expected) in cases { + let mut tx = self.begin().await.unwrap(); + Self::query("UPDATE workflow_runs SET run_number=$1,cancellation_delivery_sequence=$2,last_command_sequence=1000000 WHERE id=$3") + .bind(run_number).bind(DB::optional_integer(cancellation)).bind(run_id).execute(&mut *tx).await.unwrap(); + if run_number > 1 { + Self::query("UPDATE workflow_history_events SET payload=$1 WHERE workflow_run_id=$2 AND event_type='WorkflowStarted'") + .bind(DB::document(&json!({"recovery_kind":"redrive","resume_step_sequence":40}))) + .bind(run_id).execute(&mut *tx).await.unwrap(); + } + for (value, owner) in activities.into_iter().map(|n| (n,run_id)) + .chain([(1000000,foreign["run_id"].as_str().unwrap())]) { + Self::query("INSERT INTO activity_executions(id,workflow_run_id,sequence,activity_class,activity_type,status) VALUES ($1,$2,$3,'counter','counter','completed')") + .bind(super::store::id()).bind(owner).bind(value).execute(&mut *tx).await.unwrap(); + } + for (value,owner) in timers.into_iter().map(|n| (n,run_id)) + .chain([(1000000,foreign["run_id"].as_str().unwrap())]) { + Self::query("INSERT INTO workflow_run_timers(id,workflow_run_id,sequence,status,delay_seconds,fire_at) VALUES ($1,$2,$3,'fired',1,$4)") + .bind(super::store::id()).bind(owner).bind(value).bind(DB::bind_time(super::store::now())) + .execute(&mut *tx).await.unwrap(); + } + for (value,owner,instance) in children.into_iter().map(|n| (n,run_id,run["workflow_id"].as_str().unwrap())) + .chain([(1000000,foreign["run_id"].as_str().unwrap(),foreign["workflow_id"].as_str().unwrap())]) { + Self::query("INSERT INTO workflow_child_calls(parent_workflow_run_id,parent_workflow_instance_id,sequence,child_workflow_type,child_workflow_class,scheduled_at) VALUES ($1,$2,$3,'counter','counter',$4)") + .bind(owner).bind(instance).bind(value).bind(DB::bind_time(super::store::now())) + .execute(&mut *tx).await.unwrap(); + } + // These are stored-reader controls, not admission examples for + // the wait/scope writers or their independent projections. + // The physical column width differs by backend (INT4 on PostgreSQL). + let first = DB::number(&Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(&mut *tx).await.unwrap(),"last_history_sequence").unwrap(); + let count = i64::try_from(events.len()).unwrap(); + for (index,(kind,payload)) in events.into_iter().enumerate() { + Self::query("INSERT INTO workflow_history_events(id,workflow_run_id,sequence,event_type,payload,recorded_at) VALUES ($1,$2,$3,$4,$5,$6)") + .bind(super::store::id()).bind(run_id).bind(first+i64::try_from(index).unwrap()+1) + .bind(kind).bind(DB::document(&payload)).bind(DB::bind_time(super::store::now())) + .execute(&mut *tx).await.unwrap(); + } + Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2") + .bind(first+count).bind(run_id).execute(&mut *tx).await.unwrap(); + let (actual, fast) = super::sql_counts::measure(Self::authored_sequence(&mut tx,run_id)).await; + let (previous, old) = super::sql_counts::measure(Self::authored_sequence_reference(&mut tx,run_id)).await; + assert_eq!(actual.unwrap(),expected); + assert_eq!(previous.unwrap(),expected); + assert_eq!(fast.count(super::store::AUTHORED_RUN_SQL),1); + assert_eq!(old.total(),fast.total()+3,"four physical/run reads become one, history reads remain"); + tx.rollback().await.unwrap(); + } + let mut tx = self.begin().await.unwrap(); + let actual = Self::authored_sequence(&mut tx,"missing-counter-run").await.unwrap_err(); + let previous = Self::authored_sequence_reference(&mut tx,"missing-counter-run").await.unwrap_err(); + assert_eq!(actual.to_string(),previous.to_string()); + tx.rollback().await.unwrap(); + } + + pub(super) async fn qualify_transaction_run_tags(&self) { + let input = crate::codec::ValueCodec::new().unwrap() + .encode(&crate::codec::Value::Array(vec![])).unwrap(); + let run = self.start(json!({"workflow_id":"run-tags","workflow_type":"run-tags", + "task_queue":"run-tags","input":{"codec":"avro","blob":input}}),BTreeMap::new(),None,None).await.unwrap(); + let foreign = self.start(json!({"workflow_id":"run-tags-other","workflow_type":"run-tags", + "task_queue":"run-tags","input":{"codec":"avro","blob":input}}),BTreeMap::new(),None,None).await.unwrap(); + let run_id = run["run_id"].as_str().unwrap(); + Self::query("UPDATE workflow_runs SET priority=7,fairness_key=' raw λ ',fairness_weight=3,compatibility=' build ' WHERE id=$1") + .bind(run_id).execute(&self.pool).await.unwrap(); + let read = "SELECT namespace,priority,fairness_key,fairness_weight,compatibility FROM workflow_runs WHERE id=$1"; + let mut tx = self.begin().await.unwrap(); + Self::active_run(&mut tx,run_id).await.unwrap(); + let (task,count) = super::sql_counts::measure(Self::create_task(&mut tx,run_id,"run-tags","workflow",json!({"own":"payload"}))).await; + let task = task.unwrap(); + assert_eq!(count.count(read),0,"the authoritative active-run row supplies its tags"); + assert_eq!(count.total(),1,"only the original INSERT remains"); + let row = Self::query("SELECT namespace,priority,fairness_key,fairness_weight,compatibility,payload FROM workflow_tasks WHERE id=$1") + .bind(&task).fetch_one(&mut *tx).await.unwrap(); + assert_eq!(DB::string(&row,"namespace").unwrap(),self.namespace.as_ref()); + assert_eq!(DB::number(&row,"priority").unwrap(),7); + assert_eq!(DB::optional_string(&row,"fairness_key").unwrap().as_deref(),Some(" raw λ ")); + assert_eq!(DB::number(&row,"fairness_weight").unwrap(),3); + assert_eq!(DB::optional_string(&row,"compatibility").unwrap().as_deref(),Some(" build ")); + assert_eq!(DB::document_row(&row,"payload").unwrap(),json!({"own":"payload"})); + let saved = tx.poll_savepoint().await.unwrap(); + Self::query("UPDATE workflow_runs SET priority=19 WHERE id=$1").bind(run_id).execute(&mut *tx).await.unwrap(); + Self::active_run(&mut tx,run_id).await.unwrap(); + assert_eq!(tx.run_tags[run_id].priority,19); + let discarded = Self::create_task(&mut tx,run_id,"run-tags","activity",json!({})).await.unwrap(); + tx.rollback_poll_claim(saved).await.unwrap(); + assert!(tx.run_tags.is_empty(),"a rolled-back view cannot remain cached"); + assert!(Self::query("SELECT id FROM workflow_tasks WHERE id=$1").bind(discarded) + .fetch_optional(&mut *tx).await.unwrap().is_none()); + let (task,count) = super::sql_counts::measure(Self::create_task(&mut tx,run_id,"run-tags","activity",json!({}))).await; + let task = task.unwrap(); + assert_eq!(count.count(read),1,"rollback must force the original row read"); + let row = Self::query("SELECT priority FROM workflow_tasks WHERE id=$1").bind(task) + .fetch_one(&mut *tx).await.unwrap(); + assert_eq!(DB::number(&row,"priority").unwrap(),7); + let (_,count) = super::sql_counts::measure(Self::create_task(&mut tx,foreign["run_id"].as_str().unwrap(),"run-tags","workflow",json!({}))).await; + assert_eq!(count.count(read),1,"another run cannot consume these tags"); + tx.rollback().await.unwrap(); + Self::query("UPDATE workflow_runs SET priority=13 WHERE id=$1").bind(run_id).execute(&self.pool).await.unwrap(); + let mut tx = self.begin().await.unwrap(); + let (task,count) = super::sql_counts::measure(Self::create_task(&mut tx,run_id,"run-tags","workflow",json!({}))).await; + let task = task.unwrap(); + assert_eq!(count.count(read),1,"a fresh transaction starts with no cached view"); + let row = Self::query("SELECT priority FROM workflow_tasks WHERE id=$1").bind(task) + .fetch_one(&mut *tx).await.unwrap(); + assert_eq!(DB::number(&row,"priority").unwrap(),13); + assert!(Self::create_task(&mut tx,"missing-run-tags","run-tags","workflow",json!({})).await.is_err()); + tx.rollback().await.unwrap(); + if DB::PHP_CONNECTION == "sqlite" { + let mut tx = self.begin().await.unwrap(); + Self::active_run(&mut tx,run_id).await.unwrap(); + Self::query("UPDATE workflow_runs SET priority='not-an-integer' WHERE id=$1") + .bind(run_id).execute(&mut *tx).await.unwrap(); + Self::active_run(&mut tx,run_id).await.unwrap(); + assert!(!tx.run_tags.contains_key(run_id),"a failed seed must remove an earlier cached value"); + assert!(Self::create_task(&mut tx,run_id,"run-tags","workflow",json!({})).await.is_err()); + tx.rollback().await.unwrap(); + } + } pub(super) async fn qualify_summary_projection_bytes(&self) { let input = crate::codec::ValueCodec::new().unwrap() .encode(&crate::codec::Value::Array(vec![])).unwrap(); let run = self.start(json!({"workflow_id":"projection-bytes","workflow_type":"projection", "task_queue":"projection","input":{"codec":"avro","blob":input}}), BTreeMap::new(), None, None).await.unwrap(); let run_id = run["run_id"].as_str().unwrap(); + self.qualify_projection_census_passes(run_id,run["workflow_id"].as_str().unwrap()).await; // Keep transport classification far from a live clock boundary while // the two reference paths run one after another. Self::query("UPDATE workflow_tasks SET available_at='2035-01-01 00:00:00' WHERE workflow_run_id=$1") @@ -18,6 +211,21 @@ store_methods! { let (fast, old) = Self::compare_projection_bytes(&mut tx, run_id).await.unwrap(); assert!(fast < old, "an ordinary ready run must issue fewer statements: {fast}/{old}"); tx.rollback().await.unwrap(); + // Both insert and conflict paths compare every stored column with the + // exact predecessor. Poison old nullable fields to prove they clear. + for remove in [false,true] { + let mut tx = self.begin().await.unwrap(); + if remove { + Self::query("DELETE FROM workflow_run_summaries WHERE id=$1") + .bind(run_id).execute(&mut *tx).await.unwrap(); + } else { + Self::query("UPDATE workflow_run_summaries SET wait_kind='stale',wait_reason='stale',open_wait_id='stale',resume_source_kind='stale',resume_source_id='stale',next_task_id='stale',next_task_type='stale',next_task_status='stale',next_task_at='2000-01-01 00:00:00',next_task_lease_expires_at='2000-01-01 00:00:00' WHERE id=$1") + .bind(run_id).execute(&mut *tx).await.unwrap(); + } + Self::compare_projection_bytes(&mut tx, run_id).await.unwrap(); + tx.rollback().await.unwrap(); + } + self.qualify_summary_projection_absence(run_id,run["workflow_id"].as_str().unwrap()).await; // Every column is compared on the same database/transaction. These // are stored-row classifier controls, not producer/admission claims. for sql in [ @@ -89,6 +297,9 @@ store_methods! { async fn sqlite_poll_cache_writes_use_one_statement_and_rollback() { let dir = tempfile::tempdir().unwrap(); let store = Store::::new(super::sqlite::open(dir.path().join("writes.sqlite").to_str().unwrap()).await.unwrap()); + store.qualify_completion_waiting_status().await; + store.qualify_authored_run_sequences().await; + store.qualify_transaction_run_tags().await; store.qualify_query_poll_cache_writes().await; store.qualify_workflow_poll_cache_writes().await; store.qualify_summary_projection_bytes().await; diff --git a/rust/src/runtime/postgres.rs b/rust/src/runtime/postgres.rs index 804a33504..4165471f7 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -717,6 +717,9 @@ mod tests { async fn qualification_poll_cache_single_statement_writes() { let db = Database::new().await; let store = super::super::store::Store::new(PostgresStorage::open(db.options.clone()).await.unwrap().into_pool()); + store.qualify_completion_waiting_status().await; + store.qualify_authored_run_sequences().await; + store.qualify_transaction_run_tags().await; store.qualify_query_poll_cache_writes().await; store.qualify_workflow_poll_cache_writes().await; store.qualify_summary_projection_bytes().await; @@ -746,6 +749,7 @@ mod tests { store.qualify_workflow_poll_transactions().await; store.qualify_first_candidate_batch().await; store.qualify_claim_history_sequence().await; + store.qualify_fenced_registration_reuse().await; store.qualify_query_poll_transactions().await; store.close().await; db.remove().await; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 484103f2e..0bffe8a42 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1,5 +1,5 @@ //! One durable execution state machine for typed database adapters. -use super::transaction::Transaction; +use super::transaction::{RunTags, Transaction}; use super::{Result, backend::Backend, envelope, refuse, side_effect_result, text, wire}; use axum::http::StatusCode; use chrono::{DateTime, Utc}; @@ -9,6 +9,11 @@ use std::{sync::Arc, time::Duration}; use tokio::sync::{Mutex, Notify, Semaphore}; use ulid::Ulid; +// Each subquery keeps its original ordered lookup. The run and three +// physical authored counters share one statement; history-based scope, +// wait and redrive authority is still read by authored_sequence. +pub(super) const AUTHORED_RUN_SQL: &str = "SELECT r.cancellation_delivery_sequence,r.run_number,(SELECT sequence FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1) AS authored_activity_sequence,(SELECT sequence FROM workflow_run_timers WHERE workflow_run_id=$2 ORDER BY sequence DESC LIMIT 1) AS authored_timer_sequence,(SELECT sequence FROM workflow_child_calls WHERE parent_workflow_run_id=$3 ORDER BY sequence DESC LIMIT 1) AS authored_child_sequence FROM workflow_runs r WHERE r.id=$4"; + pub(super) struct PreparedControlArguments { pub(super) arguments: String, pub(super) values: Vec, @@ -1311,6 +1316,64 @@ where pub(super) async fn authored_sequence( tx: &mut Transaction<'_, DB>, run_id: &str, + ) -> Result { + // Public control command order and deterministic authored call order + // are different counters in PHP. Signals must never shift a replayed + // activity/timer/wait's authored position. + let run = Self::query(AUTHORED_RUN_SQL) + .bind(run_id).bind(run_id).bind(run_id).bind(run_id) + .fetch_one(&mut **tx).await?; + // Each counter is its column's own nullable integer, not a COALESCE + // that MariaDB would type as DECIMAL; no row counts as zero. + let activities = DB::optional_number(&run, "authored_activity_sequence")?.unwrap_or(0); + let timers = DB::optional_number(&run, "authored_timer_sequence")?.unwrap_or(0); + let children = DB::optional_number(&run, "authored_child_sequence")?.unwrap_or(0); + let wait = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('SignalWaitOpened','ConditionWaitOpened','VersionMarkerRecorded','MemoUpserted','SideEffectRecorded','SearchAttributesUpserted') ORDER BY sequence DESC LIMIT 1") + .bind(run_id).fetch_optional(&mut **tx).await?; + let wait_sequence = wait + .map(|row| DB::document_row(&row, "payload")) + .transpose()? + .and_then(|value| value["sequence"].as_i64()) + .unwrap_or(0); + // `nextDurableCommandSequence`: a scope opening consumes its sequence, + // and a delivered scope boundary its whole span, as a delivered run + // request does. + let mut scoped = 0; + for row in Self::query("SELECT event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('CancellationScopeOpened','CancellationScopeDelivered')") + .bind(run_id).fetch_all(&mut **tx).await? + { + let payload = DB::document_row(&row, "payload")?; + if let Some(sequence) = payload["sequence"].as_i64() { + let span = payload["sequence_span"].as_i64().filter(|span| *span > 1 && DB::string(&row, "event_type").is_ok_and(|kind| kind == "CancellationScopeDelivered")); + scoped = scoped.max(span.and_then(|span| sequence.checked_add(span - 1)).unwrap_or(sequence)); + } + } + // A redriven run replays the activities before its resume step from + // reused completions, which have no execution rows. + let mut reused = 0; + if DB::number(&run, "run_number")? > 1 { + 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).fetch_optional(&mut **tx).await?; + if let Some(started) = started { + let payload = DB::document_row(&started, "payload")?; + if payload["recovery_kind"] == "redrive" { + reused = payload["resume_step_sequence"].as_i64().unwrap_or(1) - 1; + } + } + } + Ok(activities + .max(timers) + .max(wait_sequence) + .max(children) + .max(scoped) + .max(reused) + .max(DB::optional_number(&run, "cancellation_delivery_sequence")?.unwrap_or(0))) + } + + #[cfg(test)] + pub(super) async fn authored_sequence_reference( + tx: &mut Transaction<'_, DB>, + run_id: &str, ) -> Result { // Public control command order and deterministic authored call order // are different counters in PHP. Signals must never shift a replayed @@ -2218,12 +2281,12 @@ where fence: Option<&super::worker_poll_fence::WorkerPollFence>, ) -> Result> { let mut tx = self.begin().await?; - if let Some(fence) = fence - && !self.worker_poll_fence_holds_in(&mut tx, fence, text(body, "worker_id")?).await? - { - return Err(super::worker_poll_fence::stale_claim()); - } - let claimed = self.claim_in(&mut tx, body, kind, protocol_version, page, poll_request).await?; + let registration = match fence { + Some(fence) => Some(self.fenced_claim_registration(&mut tx,fence,text(body,"worker_id")?).await? + .ok_or_else(super::worker_poll_fence::stale_claim)?), + None => None, + }; + let claimed = self.claim_with_registration(&mut tx, body, kind, protocol_version, page, poll_request, registration).await?; tx.commit().await?; Ok(claimed) } @@ -2247,14 +2310,15 @@ where tx.commit().await?; return Ok(FirstProbe::Replayed(answer)); } - if let Some(fence) = fence - && !self.worker_poll_fence_holds_in(&mut tx, fence, worker_id).await? - { - tx.commit().await?; - return Ok(FirstProbe::Claimed(Err(super::worker_poll_fence::stale_claim()))); - } + let registration = if let Some(fence) = fence { + let Some(row) = self.fenced_claim_registration(&mut tx,fence,worker_id).await? else { + tx.commit().await?; + return Ok(FirstProbe::Claimed(Err(super::worker_poll_fence::stale_claim()))); + }; + Some(row) + } else { None }; let saved = tx.poll_savepoint().await?; - let claimed = match self.claim_in(&mut tx, body, kind, protocol_version, page, Some(request)).await { + let claimed = match self.claim_with_registration(&mut tx, body, kind, protocol_version, page, Some(request), registration).await { Ok(claimed) => { tx.release_poll_claim().await?; Ok(claimed) @@ -2279,17 +2343,34 @@ where protocol_version: &str, page: Option, poll_request: Option<&super::workflow_poll_requests::WorkflowPollRequest>, + ) -> Result> { + self.claim_with_registration(tx,body,kind,protocol_version,page,poll_request,None).await + } + + #[allow(clippy::too_many_arguments)] + pub(super) async fn claim_with_registration( + &self, + tx: &mut super::transaction::Transaction<'_, DB>, + body: &Value, + kind: &str, + protocol_version: &str, + page: Option, + poll_request: Option<&super::workflow_poll_requests::WorkflowPollRequest>, + registration: Option, ) -> Result> { let worker_id = text(body, "worker_id")?; let queue = text(body, "task_queue")?; - let worker = Self::query( - "SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2", - ) - .bind(worker_id) - .bind(self.namespace.as_ref()) - .fetch_optional(&mut **tx) - .await? - .ok_or_else(|| refuse(StatusCode::CONFLICT, "worker_not_registered"))?; + let worker = match registration { + Some(row) => row, + None => Self::query( + "SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2", + ) + .bind(worker_id) + .bind(self.namespace.as_ref()) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(|| refuse(StatusCode::CONFLICT, "worker_not_registered"))?, + }; if DB::string(&worker, "task_queue")? != queue { return Err(refuse(StatusCode::CONFLICT, "task_queue_mismatch")); } @@ -2301,8 +2382,8 @@ where if let Some(request) = poll_request && !request.replayed_in(tx) { - Self::bind_workflow_poll(tx, request).await?; - if let Some(answer) = self.deliverable_workflow_lease(tx, request, worker_id, queue).await? + let cached = Self::bind_workflow_poll(tx, request).await?; + if let Some(answer) = self.deliverable_workflow_lease(tx, request, worker_id, queue, cached).await? && answer["task"]["task_kind"] != "update_validation" { return Ok(Some(answer["task"].clone())); @@ -2423,6 +2504,7 @@ where .bind(&run_id) .fetch_one(&mut **tx) .await?; + tx.remember_run_tags(&run_id,&run); let cancellation = Self::cancellation_request(&run)?; if cancellation.is_some() && (DB::optional_instant(&run, "cancellation_deadline_at")? @@ -3494,15 +3576,18 @@ where Self::record_completion(&mut tx, task_id, &json!({COMPLETION_RECEIPT: {"request": body, "created_task_ids": created}})).await?; // PHP's executor leaves every run a completed task did not close // waiting, even when the task recorded only markers or side effects. - Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1 AND status IN ('pending','running')") - .bind(&run_id) - .execute(&mut *tx) - .await?; let current = Self::query("SELECT status FROM workflow_runs WHERE id=$1") .bind(&run_id) .fetch_one(&mut *tx) .await?; - let run_status = DB::string(¤t, "status")?; + let mut run_status = DB::string(¤t, "status")?; + if matches!(run_status.as_str(), "pending" | "running") { + Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1 AND status IN ('pending','running')") + .bind(&run_id) + .execute(&mut *tx) + .await?; + run_status = "waiting".into(); + } if let Some(worker) = &worker { Self::attribute_worker_terminal(&mut tx, &run_id, task_id, worker).await?; } @@ -4587,9 +4672,19 @@ where "cancellation_deadline_expired", )); } + tx.remember_run_tags(run_id,&run); Ok(run) } + async fn run_tags(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result { + if let Some(tags) = tx.run_tags.get(run_id) { return Ok(tags.clone()); } + let row = Self::query("SELECT namespace,priority,fairness_key,fairness_weight,compatibility FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(&mut **tx).await?; + let tags = RunTags::from_row::(&row)?; + tx.run_tags.insert(run_id.to_owned(),tags.clone()); + Ok(tags) + } + pub(super) async fn create_task( tx: &mut Transaction<'_, DB>, run_id: &str, @@ -4613,19 +4708,16 @@ where tx.summary_changed(run_id); let task_id = id(); // A task carries its run's priority and fairness tags, as in PHP. - let run = Self::query("SELECT namespace,priority,fairness_key,fairness_weight,compatibility FROM workflow_runs WHERE id=$1") - .bind(run_id) - .fetch_one(&mut **tx) - .await?; - let namespace = DB::string(&run, "namespace")?; + let tags = Self::run_tags(tx,run_id).await?; + let namespace = &tags.namespace; let created = now(); let dispatched_at = dispatched.then_some(created); - let fairness_key = DB::optional_string(&run, "fairness_key")?; + let fairness_key = tags.fairness_key.as_deref(); Self::query("INSERT INTO workflow_tasks(id,workflow_run_id,task_type,status,payload,queue,available_at,namespace,created_at,updated_at,last_dispatched_at,last_dispatch_attempt_at,priority,fairness_key,fairness_weight,compatibility,ready_origin_at) VALUES ($1,$2,$3,'ready',$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)") - .bind(&task_id).bind(run_id).bind(kind).bind(DB::document(&payload)).bind(queue).bind(DB::bind_time(created)).bind(&namespace) + .bind(&task_id).bind(run_id).bind(kind).bind(DB::document(&payload)).bind(queue).bind(DB::bind_time(created)).bind(namespace) .bind(DB::bind_time(created)).bind(DB::bind_time(created)).bind(DB::optional_time(dispatched_at)).bind(DB::optional_time(dispatched_at)) - .bind(DB::number(&run, "priority")?).bind(fairness_key.as_deref()).bind(DB::number(&run, "fairness_weight")?) - .bind(DB::optional_string(&run, "compatibility")?.as_deref()) + .bind(tags.priority).bind(fairness_key).bind(tags.fairness_weight) + .bind(tags.compatibility.as_deref()) .bind(DB::optional_time(matches!(kind, "workflow" | "activity").then_some(created))) .execute(&mut **tx).await?; Ok(task_id) diff --git a/rust/src/runtime/terminal_summary.rs b/rust/src/runtime/terminal_summary.rs index 01619357d..5ffb4b37b 100644 --- a/rust/src/runtime/terminal_summary.rs +++ b/rust/src/runtime/terminal_summary.rs @@ -12,11 +12,26 @@ use serde_json::Value; use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; use std::collections::BTreeSet; +// MySQL's summary table has only the id primary key as a unique key. +// Every projected column is supplied; replacing/updating a row clears stale +// nullable wait and next-task fields just as the old DELETE + INSERT did. +macro_rules! summary_projection_sql { + ($verb:literal, $suffix:literal) => { + concat!($verb, " INTO workflow_run_summaries(id,workflow_instance_id,run_number,is_current_run,engine_source,projection_schema_version,class,workflow_type,namespace,compatibility,declared_entry_mode,declared_contract_source,business_key,visibility_labels,status,status_bucket,closed_reason,connection,queue,started_at,sort_timestamp,sort_key,closed_at,archived_at,archive_command_id,archive_reason,duration_ms,liveness_state,liveness_reason,repair_blocked_reason,repair_attention,task_problem,exception_count,history_event_count,history_size_bytes,history_fan_out,continue_as_new_recommended,history_budget_pressure,created_at,updated_at,wait_kind,wait_reason,wait_started_at,wait_deadline_at,open_wait_id,resume_source_kind,resume_source_id,next_task_at,next_task_id,next_task_type,next_task_status,next_task_lease_expires_at) SELECT r.id,r.workflow_instance_id,r.run_number,CASE WHEN i.current_run_id=r.id THEN TRUE ELSE FALSE END,'v2',2,r.workflow_class,r.workflow_type,COALESCE(r.namespace,i.namespace),r.compatibility,$1,$2,COALESCE(r.business_key,i.business_key),COALESCE(r.visibility_labels,i.visibility_labels),r.status,CASE WHEN r.status='completed' THEN 'completed' WHEN r.status IN ('failed','cancelled','terminated') THEN 'failed' ELSE 'running' END,r.closed_reason,r.connection,r.queue,r.started_at,$3,$4,r.closed_at,r.archived_at,r.archive_command_id,r.archive_reason,CASE WHEN r.started_at IS NULL OR r.closed_at IS NULL THEN NULL ELSE $5 END,$6,$7,CASE WHEN i.current_run_id=r.id THEN $8 ELSE 'selected_run_not_current' END,CASE WHEN $9=1 THEN TRUE ELSE FALSE END,CASE WHEN $10=1 THEN TRUE ELSE FALSE END,$11,$12,$13,$14,CASE WHEN $15='continue_as_new_recommended' THEN TRUE ELSE FALSE END,$16,r.created_at,COALESCE(r.closed_at,r.last_progress_at,r.updated_at),$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28 FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE r.id=$29", $suffix) + }; +} +pub(super) use summary_projection_sql; +pub(super) const SUMMARY_UPSERT_SQL: &str = summary_projection_sql!("INSERT", " ON CONFLICT(id) DO UPDATE SET workflow_instance_id=excluded.workflow_instance_id,run_number=excluded.run_number,is_current_run=excluded.is_current_run,engine_source=excluded.engine_source,projection_schema_version=excluded.projection_schema_version,class=excluded.class,workflow_type=excluded.workflow_type,namespace=excluded.namespace,compatibility=excluded.compatibility,declared_entry_mode=excluded.declared_entry_mode,declared_contract_source=excluded.declared_contract_source,business_key=excluded.business_key,visibility_labels=excluded.visibility_labels,status=excluded.status,status_bucket=excluded.status_bucket,closed_reason=excluded.closed_reason,connection=excluded.connection,queue=excluded.queue,started_at=excluded.started_at,sort_timestamp=excluded.sort_timestamp,sort_key=excluded.sort_key,closed_at=excluded.closed_at,archived_at=excluded.archived_at,archive_command_id=excluded.archive_command_id,archive_reason=excluded.archive_reason,duration_ms=excluded.duration_ms,liveness_state=excluded.liveness_state,liveness_reason=excluded.liveness_reason,repair_blocked_reason=excluded.repair_blocked_reason,repair_attention=excluded.repair_attention,task_problem=excluded.task_problem,exception_count=excluded.exception_count,history_event_count=excluded.history_event_count,history_size_bytes=excluded.history_size_bytes,history_fan_out=excluded.history_fan_out,continue_as_new_recommended=excluded.continue_as_new_recommended,history_budget_pressure=excluded.history_budget_pressure,created_at=excluded.created_at,updated_at=excluded.updated_at,wait_kind=excluded.wait_kind,wait_reason=excluded.wait_reason,wait_started_at=excluded.wait_started_at,wait_deadline_at=excluded.wait_deadline_at,open_wait_id=excluded.open_wait_id,resume_source_kind=excluded.resume_source_kind,resume_source_id=excluded.resume_source_id,next_task_at=excluded.next_task_at,next_task_id=excluded.next_task_id,next_task_type=excluded.next_task_type,next_task_status=excluded.next_task_status,next_task_lease_expires_at=excluded.next_task_lease_expires_at"); +#[cfg(test)] +const SUMMARY_INSERT_SQL: &str = summary_projection_sql!("INSERT", ""); + #[derive(Clone, Copy)] enum ProjectionMode { Current, #[cfg(test)] Reference, + #[cfg(test)] + DeleteInsert, } impl Store @@ -45,8 +60,11 @@ where ) -> Result<()> { let run = Self::query("SELECT * FROM workflow_runs WHERE id=$1") .bind(run_id) - .fetch_one(&mut **tx) + .fetch_optional(&mut **tx) .await?; + let Some(run) = run else { + return Self::delete_run_summary(tx, run_id).await; + }; let status = DB::string(&run, "status")?; let closed = terminal(&status); // Retention and v1 projections need their separately qualified @@ -58,14 +76,15 @@ where if DB::optional_instant(&run, "details_pruned_at")?.is_some() || (import_source.is_some() && !embedded_import) { return Ok(()); } - // A state outside the qualified producer must not retain an earlier - // healthy snapshot. Reads still never create or repair this row. - Self::query("DELETE FROM workflow_run_summaries WHERE id=$1") - .bind(run_id) - .execute(&mut **tx) - .await?; + // Test-only predecessors retain their original delete-before-read. + // Production deletes only a missing/unqualified projection. + let upsert = matches!(mode, ProjectionMode::Current); + if !upsert { + Self::delete_run_summary(tx, run_id).await?; + } let last = DB::number(&run, "last_history_sequence")?; let Some(history) = Self::summary_history(tx, run_id, closed, last).await? else { + if upsert { Self::delete_run_summary(tx, run_id).await?; } return Ok(()); }; let SummaryHistory { @@ -86,11 +105,14 @@ where ProjectionMode::Current => Self::projection_census(tx, run_id).await?, #[cfg(test)] ProjectionMode::Reference => Self::projection_census_reference(tx, run_id).await?, + #[cfg(test)] + ProjectionMode::DeleteInsert => Self::projection_census(tx, run_id).await?, }) }; let active = if closed { None } else { if !active_history_supported { + if upsert { Self::delete_run_summary(tx, run_id).await?; } return Ok(()); } match Self::active_task_summary( @@ -106,7 +128,10 @@ where .await? { Some(summary) => Some(summary), - None => return Ok(()), + None => { + if upsert { Self::delete_run_summary(tx, run_id).await?; } + return Ok(()); + } } }; let mut after = String::new(); @@ -161,9 +186,12 @@ where Self::write_active_summary_fields(tx, run_id, &active).await?; } } - ProjectionMode::Current => { + _ => { let wait = active.as_ref().and_then(|summary| summary.wait.as_ref()); - Self::query("INSERT INTO workflow_run_summaries(id,workflow_instance_id,run_number,is_current_run,engine_source,projection_schema_version,class,workflow_type,namespace,compatibility,declared_entry_mode,declared_contract_source,business_key,visibility_labels,status,status_bucket,closed_reason,connection,queue,started_at,sort_timestamp,sort_key,closed_at,archived_at,archive_command_id,archive_reason,duration_ms,liveness_state,liveness_reason,repair_blocked_reason,repair_attention,task_problem,exception_count,history_event_count,history_size_bytes,history_fan_out,continue_as_new_recommended,history_budget_pressure,created_at,updated_at,wait_kind,wait_reason,wait_started_at,wait_deadline_at,open_wait_id,resume_source_kind,resume_source_id,next_task_at,next_task_id,next_task_type,next_task_status,next_task_lease_expires_at) SELECT r.id,r.workflow_instance_id,r.run_number,CASE WHEN i.current_run_id=r.id THEN TRUE ELSE FALSE END,'v2',2,r.workflow_class,r.workflow_type,COALESCE(r.namespace,i.namespace),r.compatibility,$1,$2,COALESCE(r.business_key,i.business_key),COALESCE(r.visibility_labels,i.visibility_labels),r.status,CASE WHEN r.status='completed' THEN 'completed' WHEN r.status IN ('failed','cancelled','terminated') THEN 'failed' ELSE 'running' END,r.closed_reason,r.connection,r.queue,r.started_at,$3,$4,r.closed_at,r.archived_at,r.archive_command_id,r.archive_reason,CASE WHEN r.started_at IS NULL OR r.closed_at IS NULL THEN NULL ELSE $5 END,$6,$7,CASE WHEN i.current_run_id=r.id THEN $8 ELSE 'selected_run_not_current' END,CASE WHEN $9=1 THEN TRUE ELSE FALSE END,CASE WHEN $10=1 THEN TRUE ELSE FALSE END,$11,$12,$13,$14,CASE WHEN $15='continue_as_new_recommended' THEN TRUE ELSE FALSE END,$16,r.created_at,COALESCE(r.closed_at,r.last_progress_at,r.updated_at),$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28 FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE r.id=$29") + let sql = DB::SUMMARY_UPSERT_SQL; + #[cfg(test)] + let sql = if upsert { sql } else { SUMMARY_INSERT_SQL }; + let written = Self::query(sql) .bind(contract["entry_mode"].as_str()).bind(contract["source"].as_str()).bind(DB::optional_time(sort_time)).bind(sort_key.as_deref()).bind(duration).bind(state).bind(reason).bind(repair_blocked) .bind(i64::from(active.as_ref().is_some_and(|summary| summary.repair_blocked == Some("unsupported_history")))) .bind(i64::from(active.as_ref().is_some_and(|summary| summary.problem))) @@ -177,6 +205,11 @@ where .bind(active.as_ref().and_then(|summary| summary.task_status.as_deref())) .bind(DB::optional_time(active.as_ref().and_then(|summary| summary.lease_expires_at))) .bind(run_id).execute(&mut **tx).await?; + if upsert && DB::affected(written) == 0 { + // The joined instance disappeared or is missing. Do not + // retain an older snapshot when INSERT ... SELECT is empty. + Self::delete_run_summary(tx, run_id).await?; + } } } if embedded_import { @@ -188,6 +221,12 @@ where } Ok(()) } + async fn delete_run_summary(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result<()> { + Self::query("DELETE FROM workflow_run_summaries WHERE id=$1") + .bind(run_id).execute(&mut **tx).await?; + Ok(()) + } + #[cfg(test)] async fn projection_snapshot(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result> { let row = Self::query("SELECT * FROM workflow_run_summaries WHERE id=$1") @@ -216,6 +255,18 @@ where let (fast, fast_count) = super::sql_counts::measure(Self::project_run_summary(tx, run_id)).await; fast?; let expected = Self::projection_snapshot(tx, run_id).await?; + let (replaced, replaced_count) = super::sql_counts::measure( + Self::project_run_summary_mode(tx, run_id, ProjectionMode::DeleteInsert), + ).await; + replaced?; + assert_eq!(Self::projection_snapshot(tx, run_id).await?, expected, + "upsert and the exact DELETE + INSERT writer must store every column identically"); + if fast_count.count(DB::SUMMARY_UPSERT_SQL) == 1 && expected.is_some() { + assert_eq!(fast_count.count("DELETE FROM workflow_run_summaries WHERE id=$1"), 0); + assert_eq!(replaced_count.count("DELETE FROM workflow_run_summaries WHERE id=$1"), 1); + assert_eq!(replaced_count.total(), fast_count.total() + 1, + "the writer alone must save exactly one statement"); + } let (old, old_count) = super::sql_counts::measure(Self::project_run_summary_mode(tx, run_id, ProjectionMode::Reference)).await; old?; let actual = Self::projection_snapshot(tx, run_id).await?; @@ -223,6 +274,62 @@ where Ok((fast_count.total(),old_count.total())) } + #[cfg(test)] + pub(super) async fn qualify_summary_projection_absence(&self, run_id: &str, workflow_id: &str) { + let delete = "DELETE FROM workflow_run_summaries WHERE id=$1"; + // Retention and unqualified imports keep their previous projection. + // They must not pass through either the writer or invalid-state delete. + for sql in [ + "UPDATE workflow_runs SET details_pruned_at='2035-01-01 00:00:00' WHERE id=$1", + "UPDATE workflow_runs SET import_source='unqualified' WHERE id=$1", + ] { + let mut tx = self.begin().await.unwrap(); + let before = Self::projection_snapshot(&mut tx,run_id).await.unwrap(); + assert!(before.is_some()); + Self::query(sql).bind(run_id).execute(&mut *tx).await.unwrap(); + let (result,counts) = super::sql_counts::measure(Self::project_run_summary(&mut tx,run_id)).await; + result.unwrap(); + assert_eq!(Self::projection_snapshot(&mut tx,run_id).await.unwrap(),before); + assert_eq!(counts.count(delete),0); + assert_eq!(counts.count(DB::SUMMARY_UPSERT_SQL),0); + tx.rollback().await.unwrap(); + } + for missing in ["history","instance","run"] { + let mut tx = self.begin().await.unwrap(); + let before = Self::projection_snapshot(&mut tx,run_id).await.unwrap(); + assert!(before.is_some(),"seed a real stored row before {missing} removal"); + match missing { + "history" => { + Self::query("DELETE FROM workflow_history_events WHERE workflow_run_id=$1 AND sequence=1") + .bind(run_id).execute(&mut *tx).await.unwrap(); + } + "instance" => { + Self::query("DELETE FROM workflow_instances WHERE id=$1") + .bind(workflow_id).execute(&mut *tx).await.unwrap(); + assert_eq!(Self::scalar("SELECT COUNT(*) FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(&mut *tx).await.unwrap(),1); + } + "run" => { + Self::query("DELETE FROM workflow_runs WHERE id=$1") + .bind(run_id).execute(&mut *tx).await.unwrap(); + } + _ => unreachable!(), + } + assert_eq!(Self::projection_snapshot(&mut tx,run_id).await.unwrap(),before, + "the fixture must leave a stale row for the writer to remove"); + let (result,counts) = super::sql_counts::measure(Self::project_run_summary(&mut tx,run_id)).await; + result.unwrap(); + assert!(Self::projection_snapshot(&mut tx,run_id).await.unwrap().is_none(),"{missing}"); + assert_eq!(counts.count(delete),1,"{missing}"); + assert_eq!(counts.count(DB::SUMMARY_UPSERT_SQL),usize::from(missing=="instance"),"{missing}"); + tx.rollback().await.unwrap(); + let mut restored = self.begin().await.unwrap(); + assert_eq!(Self::projection_snapshot(&mut restored,run_id).await.unwrap(),before, + "rolling back {missing} cleanup must restore the entire previous row"); + restored.rollback().await.unwrap(); + } + } + #[cfg(test)] pub(super) async fn qualify_summary_history_pages(&self) { let input = crate::codec::ValueCodec::new().unwrap().encode(&crate::codec::Value::Array(vec![])).unwrap(); diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs index 81ca90456..6249b4071 100644 --- a/rust/src/runtime/transaction.rs +++ b/rust/src/runtime/transaction.rs @@ -2,11 +2,29 @@ use super::{Result, backend::Backend, store::Store}; use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; use std::{ - collections::BTreeSet, + collections::{BTreeMap, BTreeSet}, ops::{Deref, DerefMut}, }; use tokio::sync::OwnedMutexGuard; +#[derive(Clone)] +pub(super) struct RunTags { + pub(super) namespace: String, + pub(super) priority: i64, + pub(super) fairness_key: Option, + pub(super) fairness_weight: i64, + pub(super) compatibility: Option, +} + +impl RunTags { + pub(super) fn from_row(row: &DB::Row) -> Result { + Ok(Self { namespace: DB::string(row,"namespace")?, priority: DB::number(row,"priority")?, + fairness_key: DB::optional_string(row,"fairness_key")?, + fairness_weight: DB::number(row,"fairness_weight")?, + compatibility: DB::optional_string(row,"compatibility")? }) + } +} + pub(super) struct PollSavepoint { summaries: BTreeSet, workflow_poll_replayed: Option, @@ -14,6 +32,7 @@ pub(super) struct PollSavepoint { pub(super) struct Transaction<'a, DB: Backend> { inner: sqlx::Transaction<'a, DB>, + pub(super) run_tags: BTreeMap, summaries: BTreeSet, pub(super) workflow_poll_replayed: Option, pub(super) history_budget: super::history_budget::Config, @@ -35,6 +54,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> { ) -> Self { Self { inner, + run_tags: BTreeMap::new(), summaries: BTreeSet::new(), workflow_poll_replayed: None, history_budget, @@ -46,6 +66,16 @@ impl<'a, DB: Backend> Transaction<'a, DB> { } } + /// Cache only immutable task tags from an authoritative row this + /// transaction already read. A malformed tag is decoded at the original + /// task-creation boundary if that path needs it, not on unrelated reads. + pub(super) fn remember_run_tags(&mut self, run_id: &str, row: &DB::Row) { + self.run_tags.remove(run_id); + if let Ok(tags) = RunTags::from_row::(row) { + self.run_tags.insert(run_id.to_owned(),tags); + } + } + pub(super) fn summary_changed(&mut self, run_id: &str) { self.summaries.insert(run_id.to_owned()); } @@ -120,6 +150,7 @@ where self.summaries = saved.summaries; self.workflow_poll_replayed = saved.workflow_poll_replayed; self.scope_history = None; + self.run_tags.clear(); self.release_poll_claim().await } diff --git a/rust/src/runtime/update_validation.rs b/rust/src/runtime/update_validation.rs index b6bec0632..39dc91456 100644 --- a/rust/src/runtime/update_validation.rs +++ b/rust/src/runtime/update_validation.rs @@ -1067,18 +1067,22 @@ where ) -> Result> { let worker_id = text(body, "worker_id")?; let task_queue = text(body, "task_queue")?; - if let Some(fence) = fence - && !self.worker_poll_fence_holds_in(tx, fence, worker_id).await? - { - return Ok(Some(Probe::Stale)); + let mut registration = None; + if let Some(fence) = fence { + if kinds.contains(&"workflow") { + registration = self.fenced_claim_registration(tx,fence,worker_id).await?; + if registration.is_none() { return Ok(Some(Probe::Stale)); } + } else if !self.worker_poll_fence_holds_in(tx,fence,worker_id).await? { + return Ok(Some(Probe::Stale)); + } } // A concurrent duplicate may have leased since this poll began: the // binding row, locked first, makes it wait for that one's answer. if let Some(request) = request && !request.replayed_in(tx) { - Self::bind_workflow_poll(tx, request).await?; - if let Some(answer) = self.deliverable_workflow_lease(tx, request, worker_id, task_queue).await? { + let cached = Self::bind_workflow_poll(tx, request).await?; + if let Some(answer) = self.deliverable_workflow_lease(tx, request, worker_id, task_queue, cached).await? { return Ok(Some(Probe::Replayed(answer))); } } @@ -1088,7 +1092,7 @@ where order.sort_by_key(|kind| *kind != first); for kind in order { let leased = if kind == "workflow" { - self.claim_in(tx, body, "workflow", protocol_version, page, request).await?.map(|task| { + self.claim_with_registration(tx, body, "workflow", protocol_version, page, request, registration.take()).await?.map(|task| { Leased::Workflow(json!({"task":task,"poll_status":"leased","server_capabilities":super::store::capabilities()})) }) } else if let Some(claim) = self.claim_update_validation_in(tx, worker_id, task_queue).await? { diff --git a/rust/src/runtime/worker_poll_fence.rs b/rust/src/runtime/worker_poll_fence.rs index 6828ab3f5..f1a3ae8f7 100644 --- a/rust/src/runtime/worker_poll_fence.rs +++ b/rust/src/runtime/worker_poll_fence.rs @@ -33,7 +33,15 @@ use std::collections::BTreeMap; pub(super) const STALE: &str = "stale_worker_registration"; /// The registration a fence reads. -const FENCE_ROW: &str = "SELECT status,task_queue,build_id,supported_workflow_types,workflow_definition_fingerprints,workflow_command_contracts,supported_activity_types,capabilities,process_metrics,updated_at,last_heartbeat_at,(SELECT dw_worker_registration_incarnations.token FROM dw_worker_registration_incarnations WHERE dw_worker_registration_incarnations.namespace=workflow_worker_registrations.namespace AND dw_worker_registration_incarnations.worker_id=workflow_worker_registrations.worker_id AND dw_worker_registration_incarnations.status='active' ORDER BY dw_worker_registration_incarnations.created_at DESC LIMIT 1) AS incarnation_token FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2"; +macro_rules! fence_registration_sql { + ($fields:literal) => { + concat!("SELECT ", $fields, ",(SELECT dw_worker_registration_incarnations.token FROM dw_worker_registration_incarnations WHERE dw_worker_registration_incarnations.namespace=workflow_worker_registrations.namespace AND dw_worker_registration_incarnations.worker_id=workflow_worker_registrations.worker_id AND dw_worker_registration_incarnations.status='active' ORDER BY dw_worker_registration_incarnations.created_at DESC LIMIT 1) AS incarnation_token FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2") + }; +} +const FENCE_ROW: &str = fence_registration_sql!("status,task_queue,build_id,supported_workflow_types,workflow_definition_fingerprints,workflow_command_contracts,supported_activity_types,capabilities,process_metrics,updated_at,last_heartbeat_at"); +// Only a claim's in-transaction check reads all registration columns. Idle +// probes and admission keep their original narrower snapshot query. +pub(super) const FENCE_REGISTRATION: &str = fence_registration_sql!("workflow_worker_registrations.*"); /// How a poll is admitted. pub(super) enum PollAdmission { @@ -235,6 +243,19 @@ store_methods! { fence.holds::(row.as_ref(), self.worker_stale_after_seconds) } + /// Return only the current row checked inside this transaction. An + /// admission snapshot from outside the transaction is never a claim row. + pub(super) async fn fenced_claim_registration( + &self, + tx: &mut super::transaction::Transaction<'_, DB>, + fence: &WorkerPollFence, + worker_id: &str, + ) -> Result> { + let row = Self::query(FENCE_REGISTRATION).bind(self.namespace.as_ref()).bind(worker_id) + .fetch_optional(&mut **tx).await?; + if fence.holds::(row.as_ref(),self.worker_stale_after_seconds)? { Ok(row) } else { Ok(None) } + } + /// `isCurrentForUpdate` inside a claim transaction, which transitions /// take one at a time. pub(super) async fn worker_poll_fence_holds_in( diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs index f3f2447bf..988a176f8 100644 --- a/rust/src/runtime/workflow_poll_requests.rs +++ b/rust/src/runtime/workflow_poll_requests.rs @@ -156,16 +156,33 @@ store_methods! { Ok(answer) } + async fn workflow_poll_pair_rows( + tx: &mut Transaction<'_, DB>, keys: &WorkflowPollKeys, + ) -> Result<(Option, Option)> { + let rows = Self::query(DB::QUERY_CACHE_GET_PAIR).bind(&keys.binding).bind(&keys.result) + .fetch_all(&mut **tx).await?; + let (mut binding, mut result) = (None, None); + for row in rows { + match DB::number(&row, "cache_part")? { + 0 => binding = Some(row), + 1 => result = Some(row), + _ => return Err(super::refuse(StatusCode::INTERNAL_SERVER_ERROR, "invalid_poll_receipt")), + } + } + Ok((binding, result)) + } + async fn cached_workflow_poll_read_only_in( &self, read: &mut Transaction<'_, DB>, request: &WorkflowPollRequest, keys: &WorkflowPollKeys, worker_id: &str, queue: &str, leases_only: bool, ) -> Result> { let at = epoch(); - let Some(binding) = Self::query(DB::QUERY_CACHE_GET).bind(&keys.binding).fetch_optional(&mut **read).await? else { return Ok(None); }; + let (binding, result) = Self::workflow_poll_pair_rows(read, keys).await?; + let Some(binding) = binding else { return Ok(None); }; if DB::number(&binding, "expiration")? <= at { return Ok(None); } let bound: Value = serde_json::from_str(&DB::string(&binding, "value")?)?; if bound["task_kinds"] != json!(request.kinds) { return Err(kinds_conflict(request, bound["task_kinds"].clone())); } - let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(&keys.result).fetch_optional(&mut **read).await? else { return Ok(None); }; + let Some(row) = result else { return Ok(None); }; if DB::number(&row, "expiration")? <= at { return Ok(None); } let mut result: Value = serde_json::from_str(&DB::string(&row, "value")?)?; validate_cached_answer(&result)?; @@ -233,8 +250,8 @@ store_methods! { // Recovering an expired binding cannot start a new drain window // beyond the retained answer's original absolute expiry. legacy.expiration_cap = Some(expiration); - Self::bind_workflow_poll(tx, &legacy).await?; - if let Some(answer) = self.deliverable_workflow_poll(tx, &legacy, worker_id, queue).await? { return Ok(Some(answer)); } + let cached = Self::bind_workflow_poll(tx, &legacy).await?; + if let Some(answer) = self.deliverable_workflow_poll(tx, &legacy, worker_id, queue, cached).await? { return Ok(Some(answer)); } } Ok(None) } @@ -269,8 +286,8 @@ store_methods! { return Ok(Some(answer)); } self.sweep_workflow_poll_rows(tx).await?; - Self::bind_workflow_poll(tx, request).await?; - let replay = self.deliverable_workflow_poll(tx, request, worker_id, queue).await?; + let cached = Self::bind_workflow_poll(tx, request).await?; + let replay = self.deliverable_workflow_poll(tx, request, worker_id, queue, cached).await?; if replay.is_none() { Self::workflow_poll_normalization_ready(tx, request).await?; } tx.workflow_poll_replayed = Some(request.keys.binding.clone()); Ok(replay) @@ -280,7 +297,7 @@ store_methods! { /// absent or expired one is written for the pending window. The binding /// row stays locked to the end of the transaction, so a duplicate's /// claim waits and then sees this one's answer. - pub(super) async fn bind_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest) -> Result<()> { + pub(super) async fn bind_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest) -> Result> { let binding = json!({"task_kinds": request.kinds}); let expiration = (epoch() + request.pending_ttl).min(request.expiration_cap.unwrap_or(i64::MAX)); Self::query(DB::QUERY_CACHE_LOCK) @@ -289,11 +306,17 @@ store_methods! { .bind(expiration) .execute(&mut **tx) .await?; - match Self::workflow_poll_row(tx, &request.keys.binding).await? { - Some(bound) if bound["task_kinds"] != binding["task_kinds"] => Err(kinds_conflict(request, bound["task_kinds"].clone())), - Some(_) => Ok(()), - None => Self::put_workflow_poll_row_until(tx, &request.keys.binding, &binding, expiration).await, + let (row, result) = Self::workflow_poll_pair_rows(tx, &request.keys).await?; + if let Some(bound) = Self::workflow_poll_value(tx, &request.keys.binding, row).await? { + if bound["task_kinds"] != binding["task_kinds"] { + return Err(kinds_conflict(request, bound["task_kinds"].clone())); + } + } else { + Self::put_workflow_poll_row_until(tx, &request.keys.binding, &binding, expiration).await?; } + // The result remains an undecoded database row until the kind binding + // has succeeded. A malformed result cannot mask a kinds conflict. + Ok(result) } /// The cached answer if its task can still be delivered; a stale answer @@ -306,8 +329,9 @@ store_methods! { request: &WorkflowPollRequest, worker_id: &str, queue: &str, + cached: Option, ) -> Result> { - let Some(mut result) = Self::workflow_poll_row(tx, &request.keys.result).await? else { + let Some(mut result) = Self::workflow_poll_value(tx, &request.keys.result, cached).await? else { return Ok(None); }; validate_cached_answer(&result)?; @@ -340,8 +364,9 @@ store_methods! { request: &WorkflowPollRequest, worker_id: &str, queue: &str, + cached: Option, ) -> Result> { - let answer = self.deliverable_workflow_poll(tx, request, worker_id, queue).await?.filter(|answer| answer["task"].is_object()); + let answer = self.deliverable_workflow_poll(tx, request, worker_id, queue, cached).await?.filter(|answer| answer["task"].is_object()); if answer.is_none() { Self::workflow_poll_normalization_ready(tx, request).await?; } Ok(answer) } @@ -431,8 +456,8 @@ store_methods! { tx.commit().await?; return Ok(Some(answer)); } - Self::bind_workflow_poll(&mut tx, request).await?; - let lease = self.deliverable_workflow_lease(&mut tx, request, worker_id, queue).await?; + let cached = Self::bind_workflow_poll(&mut tx, request).await?; + let lease = self.deliverable_workflow_lease(&mut tx, request, worker_id, queue, cached).await?; if lease.is_none() { Self::remember_workflow_poll(&mut tx, request, &json!({"task": null, "poll_status": poll_status})).await?; } @@ -440,10 +465,14 @@ store_methods! { Ok(lease) } + #[cfg(test)] pub(super) async fn workflow_poll_row(tx: &mut Transaction<'_, DB>, key: &str) -> Result> { - let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&mut **tx).await? else { - return Ok(None); - }; + let row = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&mut **tx).await?; + Self::workflow_poll_value(tx, key, row).await + } + + async fn workflow_poll_value(tx: &mut Transaction<'_, DB>, key: &str, row: Option) -> Result> { + let Some(row) = row else { return Ok(None); }; if DB::number(&row, "expiration")? <= epoch() { Self::query(DB::QUERY_CACHE_DELETE).bind(key).execute(&mut **tx).await?; return Ok(None); @@ -506,6 +535,114 @@ store_methods! { } tx.rollback().await.unwrap(); assert!(Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&self.pool).await.unwrap().is_none()); + self.qualify_workflow_poll_pair_reads().await; + } + + #[cfg(test)] + async fn qualify_workflow_poll_pair_reads(&self) { + let keys = WorkflowPollKeys {binding:"dw-native-workflow-poll:read-binding".into(), + result:"dw-native-workflow-poll:read-result".into()}; + for existing in [0,1,2,3] { + let mut tx = self.begin().await.unwrap(); + for (bit,key,value,expiration) in [ + (1,&keys.binding,"{\"task_kinds\":[\"workflow\"]}",1_i64), + (2,&keys.result,"opaque \" λ not JSON",2_000_000_000_i64), + ] { + if existing & bit != 0 { + Self::query(DB::QUERY_CACHE_PUT).bind(key).bind(value).bind(expiration) + .execute(&mut *tx).await.unwrap(); + } + } + let (pair,counts) = super::sql_counts::measure(Self::workflow_poll_pair_rows(&mut tx,&keys)).await; + let (binding,result) = pair.unwrap(); + assert_eq!(counts.count(DB::QUERY_CACHE_GET_PAIR),1); + assert_eq!(counts.total(),1); + let (reference,counts) = super::sql_counts::measure(async { + let binding = Self::query(DB::QUERY_CACHE_GET).bind(&keys.binding).fetch_optional(&mut *tx).await?; + let result = Self::query(DB::QUERY_CACHE_GET).bind(&keys.result).fetch_optional(&mut *tx).await?; + Ok::<_,super::RuntimeError>((binding,result)) + }).await; + assert_eq!(counts.count(DB::QUERY_CACHE_GET),2); + assert_eq!(counts.total(),2); + let (old_binding,old_result) = reference.unwrap(); + for (actual,expected) in [(binding,old_binding),(result,old_result)] { + let bytes = |row: Option| row.map(|row| + (DB::string(&row,"value").unwrap(),DB::number(&row,"expiration").unwrap())); + assert_eq!(bytes(actual),bytes(expected)); + } + tx.rollback().await.unwrap(); + } + self.qualify_workflow_poll_pair_decoding(&keys).await; + } + + #[cfg(test)] + async fn qualify_workflow_poll_pair_decoding(&self, keys: &WorkflowPollKeys) { + let request = WorkflowPollRequest {id:"paired-reader".into(),keys:keys.clone(),pending_ttl:35, + kinds:vec!["workflow"],legacy_keys:vec![],expiration_cap:None}; + let response = json!({"task":null,"poll_status":"empty"}); + for (conflict,expired) in [(false,false),(true,false),(false,true),(true,true)] { + let mut tx = self.begin().await.unwrap(); + let kinds = if conflict {vec!["update_validation"]} else {request.kinds.clone()}; + Self::put_workflow_poll_row_until(&mut tx,&keys.binding,&json!({"task_kinds":kinds}),2_000_000_000_i64).await.unwrap(); + Self::query(DB::QUERY_CACHE_PUT).bind(&keys.result) + .bind(if conflict {"malformed JSON must not hide the kind conflict".into()} else {response.to_string()}) + .bind(if expired {1_i64} else {2_000_000_000_i64}).execute(&mut *tx).await.unwrap(); + let (read,counts) = super::sql_counts::measure( + self.cached_workflow_poll_read_only_in(&mut tx,&request,keys,"reader","queue",false), + ).await; + assert_eq!(counts.total(),1); + assert_eq!(counts.count(DB::QUERY_CACHE_GET_PAIR),1); + if conflict { + match read.unwrap_err() { + super::RuntimeError::Protocol {response,..} => assert_eq!(response["reason"],"poll_request_task_kinds_conflict"), + other => panic!("a kind conflict must precede result decoding: {other:?}"), + } + let error = Self::bind_workflow_poll(&mut tx,&request).await.err().expect("kind binding must refuse before result parsing"); + assert!(matches!(error,super::RuntimeError::Protocol {..})); + } else { + assert_eq!(read.unwrap(),(!expired).then_some(response.clone())); + // Snapshot reads leave expired rows alone, as before. The + // locked transition deletes them when interpreting its row. + assert!(Self::query(DB::QUERY_CACHE_GET).bind(&keys.result).fetch_optional(&mut *tx).await.unwrap().is_some()); + let (answer,counts) = super::sql_counts::measure(async { + let cached = Self::bind_workflow_poll(&mut tx,&request).await?; + self.deliverable_workflow_poll(&mut tx,&request,"reader","queue",cached).await + }).await; + assert_eq!(answer.unwrap(),(!expired).then_some(response.clone())); + assert_eq!(counts.count(DB::QUERY_CACHE_GET_PAIR),1); + assert_eq!(counts.count(DB::QUERY_CACHE_GET),0); + assert_eq!(counts.count(DB::QUERY_CACHE_DELETE),usize::from(expired)); + assert_eq!(counts.total(),2+usize::from(expired)); + assert_eq!(Self::query(DB::QUERY_CACHE_GET).bind(&keys.result).fetch_optional(&mut *tx).await.unwrap().is_none(),expired); + } + tx.rollback().await.unwrap(); + } + // Missing/expired bindings suppress a malformed result. With a + // matching live binding, that same result remains a visible error. + for state in [0,1,2] { + let mut tx = self.begin().await.unwrap(); + if state != 0 { + Self::put_workflow_poll_row_until(&mut tx,&keys.binding,&json!({"task_kinds":request.kinds}), + if state == 1 {1} else {2_000_000_000_i64}).await.unwrap(); + } + Self::query(DB::QUERY_CACHE_PUT).bind(&keys.result).bind("malformed result") + .bind(2_000_000_000_i64).execute(&mut *tx).await.unwrap(); + let read = self.cached_workflow_poll_read_only_in(&mut tx,&request,keys,"reader","queue",false).await; + if state == 2 { assert!(read.is_err()); } else { assert_eq!(read.unwrap(),None); } + tx.rollback().await.unwrap(); + } + // An expired binding is still deleted and replaced. The original + // result expiry/value are not renewed by reading the pair. + let mut tx = self.begin().await.unwrap(); + Self::put_workflow_poll_row_until(&mut tx,&keys.binding,&json!({"task_kinds":["old"]}),1).await.unwrap(); + Self::put_workflow_poll_row_until(&mut tx,&keys.result,&response,2_000_000_000_i64).await.unwrap(); + let (cached,counts) = super::sql_counts::measure(Self::bind_workflow_poll(&mut tx,&request)).await; + assert_eq!(counts.count(DB::QUERY_CACHE_DELETE),1); + assert_eq!(counts.count(DB::QUERY_CACHE_PUT),1); + let row = cached.unwrap().unwrap(); + assert_eq!(DB::string(&row,"value").unwrap(),response.to_string()); + assert_eq!(DB::number(&row,"expiration").unwrap(),2_000_000_000_i64); + tx.rollback().await.unwrap(); } // Workers send a fresh request ID with most polls, so retire a bounded