From ccb6a6ea3b829440f687ddd8b0e091c8f9611658 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 09:38:02 +0000 Subject: [PATCH 01/13] Write complete run summaries in one backend-specific upsert (cherry picked from commit 779118f7eccb0ffbf38a4b8df21f18c599ac59da) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 2 + rust/src/runtime/poll_write_tests.rs | 17 +++- rust/src/runtime/terminal_summary.rs | 127 ++++++++++++++++++++++++--- 3 files changed, 135 insertions(+), 11 deletions(-) diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 03b027f55..b5ecdc943 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"; @@ -623,6 +624,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"; diff --git a/rust/src/runtime/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index 946d6e1a8..1d721ca0b 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -1,4 +1,4 @@ -//! 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; @@ -18,6 +18,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 [ 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(); From fc3c561f33480616c839dc95cc5e0968a261912a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 09:42:52 +0000 Subject: [PATCH 02/13] Store a workflow poll binding and answer with one two-row upsert (cherry picked from commit 958820d4bf5417c0bdd6f5bca2ebfab78e20527b) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 2 + rust/src/runtime/workflow_poll_requests.rs | 85 +++++++++++++++++++++- 2 files changed, 85 insertions(+), 2 deletions(-) diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index b5ecdc943..06f613476 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -132,6 +132,7 @@ pub(super) trait Backend: Database { const QUERY_CACHE_UPDATE: &str = "UPDATE dw_query_cache SET value=$1,expiration=$2 WHERE \"key\"=$3"; const QUERY_CACHE_PUT: &str = "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3) ON CONFLICT (\"key\") DO UPDATE SET value=excluded.value,expiration=excluded.expiration"; + const QUERY_CACHE_PUT_PAIR: &str = "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3),($4,$5,$6) ON CONFLICT (\"key\") DO UPDATE SET value=excluded.value,expiration=excluded.expiration"; const QUERY_CACHE_DELETE: &str = "DELETE FROM dw_query_cache WHERE \"key\"=$1"; const QUERY_CACHE_DELETE_EXPIRED: &str = "DELETE FROM dw_query_cache WHERE \"key\"=$1 AND expiration<=$2"; // One primary-key page of the whole cache table, for the one-time @@ -657,6 +658,7 @@ impl Backend for MySql { const QUERY_CACHE_UPDATE: &str = "UPDATE dw_query_cache SET value=$1,expiration=$2 WHERE `key`=$3"; const QUERY_CACHE_PUT: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3) ON DUPLICATE KEY UPDATE value=VALUES(value),expiration=VALUES(expiration)"; + const QUERY_CACHE_PUT_PAIR: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3),($4,$5,$6) ON DUPLICATE KEY UPDATE value=VALUES(value),expiration=VALUES(expiration)"; const QUERY_CACHE_DELETE: &str = "DELETE FROM dw_query_cache WHERE `key`=$1"; const QUERY_CACHE_DELETE_EXPIRED: &str = "DELETE FROM dw_query_cache WHERE `key`=$1 AND expiration<=$2"; const QUERY_CACHE_PAGE: &str = "SELECT `key`,value,expiration FROM dw_query_cache WHERE `key`>$1 ORDER BY `key` LIMIT 64"; diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs index f3f2447bf..31c344b78 100644 --- a/rust/src/runtime/workflow_poll_requests.rs +++ b/rust/src/runtime/workflow_poll_requests.rs @@ -400,9 +400,11 @@ store_methods! { /// binding replays for the whole live lease. pub(super) async fn remember_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest, response: &Value) -> Result<()> { let ttl = result_ttl(response, request.pending_ttl); - Self::put_workflow_poll_row(tx, &request.keys.binding, &json!({"task_kinds": request.kinds}), ttl).await?; + let binding_expiration = epoch() + ttl; let kept = if response["task"].is_object() && task_kind(response) == "workflow" { 3600 } else { ttl }; - Self::put_workflow_poll_row(tx, &request.keys.result, response, kept).await + let result_expiration = epoch() + kept; + Self::put_workflow_poll_pair_until(tx, &request.keys, &json!({"task_kinds": request.kinds}), + binding_expiration, response, result_expiration).await } /// Whether a concurrent duplicate already left an answer to replay. @@ -461,6 +463,19 @@ store_methods! { Ok(()) } + async fn put_workflow_poll_pair_until( + tx: &mut Transaction<'_, DB>, keys: &WorkflowPollKeys, binding: &Value, binding_expiration: i64, + result: &Value, result_expiration: i64, + ) -> Result<()> { + // Keep both JSON values, independent expirations and original keys. + // This changes only the two writes after the existing locked claim. + Self::query(DB::QUERY_CACHE_PUT_PAIR) + .bind(&keys.binding).bind(binding.to_string()).bind(binding_expiration) + .bind(&keys.result).bind(result.to_string()).bind(result_expiration) + .execute(&mut **tx).await?; + Ok(()) + } + #[cfg(test)] pub(super) async fn expire_workflow_poll_binding_for_test(&self, request: &WorkflowPollRequest) { Self::query(DB::QUERY_CACHE_PUT).bind(&request.keys.binding) @@ -506,6 +521,72 @@ 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_writes().await; + } + + #[cfg(test)] + async fn qualify_workflow_poll_pair_writes(&self) { + let keys = WorkflowPollKeys {binding:"dw-native-workflow-poll:pair-binding".into(), + result:"dw-native-workflow-poll:pair-result".into()}; + let binding = json!({"task_kinds":["workflow"]}); + for existing in [0,1,2] { + let mut tx = self.begin().await.unwrap(); + for key in [&keys.binding,&keys.result].into_iter().take(existing) { + Self::put_workflow_poll_row_until(&mut tx,key,&json!({"old":true}),1).await.unwrap(); + } + for result in [json!({"task":null,"poll_status":"empty"}),json!({"task":{"text":"escaped \\\" λ"},"poll_status":"leased"})] { + let (written,counts) = super::sql_counts::measure( + Self::put_workflow_poll_pair_until(&mut tx,&keys,&binding,1,&result,4_102_444_800), + ).await; + written.unwrap(); + assert_eq!(counts.total(),1); + assert_eq!(counts.count(DB::QUERY_CACHE_PUT_PAIR),1); + let mut actual = Vec::new(); + for key in [&keys.binding,&keys.result] { + let row = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap(); + actual.push((DB::string(&row,"value").unwrap(),DB::number(&row,"expiration").unwrap())); + } + let (old,counts) = super::sql_counts::measure(async { + Self::put_workflow_poll_row_until(&mut tx,&keys.binding,&binding,1).await?; + Self::put_workflow_poll_row_until(&mut tx,&keys.result,&result,4_102_444_800).await + }).await; + old.unwrap(); + assert_eq!(counts.total(),2); + assert_eq!(counts.count(DB::QUERY_CACHE_PUT),2); + for (key,expected) in [&keys.binding,&keys.result].into_iter().zip(actual) { + let row = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap(); + assert_eq!((DB::string(&row,"value").unwrap(),DB::number(&row,"expiration").unwrap()),expected); + } + } + tx.rollback().await.unwrap(); + for key in [&keys.binding,&keys.result] { + assert!(Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&self.pool).await.unwrap().is_none()); + } + } + let request = WorkflowPollRequest {id:"pair-writer".into(),keys:keys.clone(),pending_ttl:35, + kinds:normalized(&["workflow","update_validation"]),legacy_keys:vec![],expiration_cap:None}; + // Exercise the actual caller as well as the writer, including the + // different retention of a workflow answer whose lease has elapsed. + // These are cache-writer vectors, not admitted protocol deliveries. + for (response,ttl,kept) in [ + (json!({"task":null,"poll_status":"empty"}),35,35), + (json!({"task":{"task_id":"workflow","lease_expires_at":"2000-01-01T00:00:00Z"},"poll_status":"leased"}),35,3600), + (json!({"task":{"task_kind":"update_validation","update_validation_task_id":"validation","lease_expires_at":"2000-01-01T00:00:00Z"},"poll_status":"leased"}),35,35), + ] { + let mut tx = self.begin().await.unwrap(); + let before = epoch(); + let (result,counts) = super::sql_counts::measure(Self::remember_workflow_poll(&mut tx,&request,&response)).await; + result.unwrap(); + let after = epoch(); + assert_eq!(counts.total(),1); + assert_eq!(counts.count(DB::QUERY_CACHE_PUT_PAIR),1); + for (key,value,retained) in [(&keys.binding,json!({"task_kinds":request.kinds}),ttl),(&keys.result,response,kept)] { + let row = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap(); + assert_eq!(DB::string(&row,"value").unwrap(),value.to_string()); + assert!((before+retained..=after+retained).contains(&DB::number(&row,"expiration").unwrap())); + } + tx.rollback().await.unwrap(); + } } // Workers send a fresh request ID with most polls, so retire a bounded From 9042342017c02a568171bcfb24c8d737bfe7e43e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:00:20 +0000 Subject: [PATCH 03/13] Prepare one aggregate pass per summary census table (cherry picked from commit f86e49fef5c9870e94df14eb22a4eaffbe59b498) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/active_summary.rs | 95 +++++++++++++++++++++++++++- rust/src/runtime/poll_write_tests.rs | 1 + 2 files changed, 93 insertions(+), 3 deletions(-) diff --git a/rust/src/runtime/active_summary.rs b/rust/src/runtime/active_summary.rs index 53d60f7ae..7c93297e9 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=$3 AND status='received') AS received_signals,(SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$4) 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=$1) 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=$2) 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/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index 1d721ca0b..eb185cd77 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -10,6 +10,7 @@ store_methods! { 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") From bba5c857742f6680a7c735cd662917b7972a26ce Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:53:01 +0000 Subject: [PATCH 04/13] Skip the final waiting update when the run is already settled (cherry picked from commit 13bb465a2a55a67517c916dd4bca3b3343e802a5) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/mysql.rs | 1 + rust/src/runtime/poll_write_tests.rs | 42 ++++++++++++++++++++++++++++ rust/src/runtime/postgres.rs | 1 + rust/src/runtime/store.rs | 13 +++++---- 4 files changed, 52 insertions(+), 5 deletions(-) diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 92bd2f8ae..462c33552 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1197,6 +1197,7 @@ 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_query_poll_cache_writes().await; store.qualify_workflow_poll_cache_writes().await; store.qualify_summary_projection_bytes().await; diff --git a/rust/src/runtime/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index eb185cd77..7f1f87b84 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -4,6 +4,47 @@ 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), + ("empty",vec![],false,"waiting",1), + ("already-waiting",vec![],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_summary_projection_bytes(&self) { let input = crate::codec::ValueCodec::new().unwrap() .encode(&crate::codec::Value::Array(vec![])).unwrap(); @@ -105,6 +146,7 @@ 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_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..5c5ebfcfb 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -717,6 +717,7 @@ 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_query_poll_cache_writes().await; store.qualify_workflow_poll_cache_writes().await; store.qualify_summary_projection_bytes().await; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 484103f2e..9842ed89d 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -3494,15 +3494,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?; } From a861a750ad685a5fb8104247ba77a0d79c4e82d6 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 11:53:26 +0000 Subject: [PATCH 05/13] Prepare single-statement workflow poll cache reads (cherry picked from commit 141ea82026c3e88dbec9f5b006bca02569fac9dc) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 2 + rust/src/runtime/store.rs | 4 +- rust/src/runtime/update_validation.rs | 4 +- rust/src/runtime/workflow_poll_requests.rs | 173 ++++++++++++++++++--- 4 files changed, 161 insertions(+), 22 deletions(-) diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 06f613476..9d33f33dd 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -127,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 = @@ -653,6 +654,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/store.rs b/rust/src/runtime/store.rs index 9842ed89d..ce032a98a 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -2301,8 +2301,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())); diff --git a/rust/src/runtime/update_validation.rs b/rust/src/runtime/update_validation.rs index b6bec0632..e2b2bdea7 100644 --- a/rust/src/runtime/update_validation.rs +++ b/rust/src/runtime/update_validation.rs @@ -1077,8 +1077,8 @@ where 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))); } } diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs index 31c344b78..8d55005fa 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) } @@ -433,8 +458,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?; } @@ -442,10 +467,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); @@ -522,6 +551,7 @@ 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_writes().await; + self.qualify_workflow_poll_pair_reads().await; } #[cfg(test)] @@ -589,6 +619,113 @@ store_methods! { } } + #[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 // batch of expired rows once per configured interval. pub(super) async fn sweep_workflow_poll_rows(&self, tx: &mut Transaction<'_, DB>) -> Result<()> { From 357d740c650288629ef55043bcf35e3d00a24d57 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:41:03 +0000 Subject: [PATCH 06/13] Read physical authored counters with the run row (cherry picked from commit 225f0fefa9574ad651c5a3586d2b7df620ff5718) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/mysql.rs | 1 + rust/src/runtime/poll_write_tests.rs | 74 ++++++++++++++++++++++++++++ rust/src/runtime/postgres.rs | 1 + rust/src/runtime/store.rs | 61 +++++++++++++++++++++++ 4 files changed, 137 insertions(+) diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 462c33552..08eecb76e 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1198,6 +1198,7 @@ mod tests { 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_query_poll_cache_writes().await; store.qualify_workflow_poll_cache_writes().await; store.qualify_summary_projection_bytes().await; diff --git a/rust/src/runtime/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index 7f1f87b84..411536b25 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -45,6 +45,79 @@ store_methods! { 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(); + } + for (kind,payload) in events { + Self::append(&mut tx,run_id,kind,payload,None,None).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_summary_projection_bytes(&self) { let input = crate::codec::ValueCodec::new().unwrap() .encode(&crate::codec::Value::Array(vec![])).unwrap(); @@ -147,6 +220,7 @@ 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_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 5c5ebfcfb..efbec56b4 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -718,6 +718,7 @@ mod tests { 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_query_poll_cache_writes().await; store.qualify_workflow_poll_cache_writes().await; store.qualify_summary_projection_bytes().await; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index ce032a98a..7fa1c1bc5 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -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,COALESCE((SELECT sequence FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1),0) AS authored_activity_sequence,COALESCE((SELECT sequence FROM workflow_run_timers WHERE workflow_run_id=$2 ORDER BY sequence DESC LIMIT 1),0) AS authored_timer_sequence,COALESCE((SELECT sequence FROM workflow_child_calls WHERE parent_workflow_run_id=$3 ORDER BY sequence DESC LIMIT 1),0) 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,62 @@ 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?; + let activities = DB::number(&run, "authored_activity_sequence")?; + let timers = DB::number(&run, "authored_timer_sequence")?; + let children = DB::number(&run, "authored_child_sequence")?; + 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 From 15a1e2ed3c1b403e4474ea2dec4b43b4057c218b Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 13:19:26 +0000 Subject: [PATCH 07/13] Keep authored sequence fixtures at the stored read boundary (cherry picked from commit 833d1225f439205d710258070502e156cbed1afa) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/poll_write_tests.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/rust/src/runtime/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index 411536b25..7b595335d 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -101,9 +101,19 @@ store_methods! { .bind(owner).bind(instance).bind(value).bind(DB::bind_time(super::store::now())) .execute(&mut *tx).await.unwrap(); } - for (kind,payload) in events { - Self::append(&mut tx,run_id,kind,payload,None,None).await.unwrap(); + // These are stored-reader controls, not admission examples for + // the wait/scope writers or their independent projections. + let first = Self::scalar("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(&mut *tx).await.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); From 8de8f46b2ea81697e7c67ccace587f703e672c5e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 13:02:43 +0000 Subject: [PATCH 08/13] Reuse the registration checked by a claim transaction fence (cherry picked from commit 038440ca1056d9e5088d5191fe9c10aaf7ce418d) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/carried_polls.rs | 14 ++--- rust/src/runtime/mysql.rs | 1 + rust/src/runtime/poll_transaction_tests.rs | 37 ++++++++++++- rust/src/runtime/postgres.rs | 1 + rust/src/runtime/store.rs | 60 ++++++++++++++-------- rust/src/runtime/update_validation.rs | 14 +++-- rust/src/runtime/worker_poll_fence.rs | 23 ++++++++- 7 files changed, 116 insertions(+), 34 deletions(-) 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/mysql.rs b/rust/src/runtime/mysql.rs index 08eecb76e..2cbeaff0d 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1228,6 +1228,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..759750c9c 100644 --- a/rust/src/runtime/poll_transaction_tests.rs +++ b/rust/src/runtime/poll_transaction_tests.rs @@ -44,8 +44,11 @@ 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); + assert_eq!(reads.count("SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2"),0, + "the 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 +230,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 +328,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/postgres.rs b/rust/src/runtime/postgres.rs index efbec56b4..6258255c8 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -748,6 +748,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 7fa1c1bc5..78c6293d7 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -2279,12 +2279,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) } @@ -2308,14 +2308,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) @@ -2340,17 +2341,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")); } diff --git a/rust/src/runtime/update_validation.rs b/rust/src/runtime/update_validation.rs index e2b2bdea7..39dc91456 100644 --- a/rust/src/runtime/update_validation.rs +++ b/rust/src/runtime/update_validation.rs @@ -1067,10 +1067,14 @@ 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. @@ -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( From acf35c84a98d21817f72c5a42b35dcd752372c71 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 7 Nov 2026 13:14:20 +0000 Subject: [PATCH 09/13] Reuse transaction-local immutable task tags and clear them on rollback (cherry picked from commit 4b8ac18a85dd3032cfbd394b20ff925d2686e9d2) Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/mysql.rs | 1 + rust/src/runtime/poll_write_tests.rs | 66 ++++++++++++++++++++++++++++ rust/src/runtime/postgres.rs | 1 + rust/src/runtime/store.rs | 28 +++++++----- rust/src/runtime/transaction.rs | 33 +++++++++++++- 5 files changed, 118 insertions(+), 11 deletions(-) diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 2cbeaff0d..7d0b5fa02 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1199,6 +1199,7 @@ mod tests { 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; diff --git a/rust/src/runtime/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index 7b595335d..6c508b259 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -128,6 +128,71 @@ store_methods! { 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(); @@ -231,6 +296,7 @@ async fn sqlite_poll_cache_writes_use_one_statement_and_rollback() { 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 6258255c8..4165471f7 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -719,6 +719,7 @@ mod tests { 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; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 78c6293d7..acf622152 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}; @@ -2502,6 +2502,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")? @@ -4669,9 +4670,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, @@ -4695,19 +4706,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/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 } From e65b6cab940fe75a0b849f5dd416f060c66e78fc Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 8 Nov 2026 06:17:30 +0000 Subject: [PATCH 10/13] Keep the original two sequential workflow poll cache writes Revert the paired binding/answer upsert (c916c34b, prepared as 958820d4). remember_workflow_poll again writes the binding and then the result, each with its own epoch() + ttl origin, exactly as on main and as PHP's two sequential puts. A single pair write took both origins before either write, which can shift the result's expiry by the binding write's duration, and that retention change is not accepted. The unused QUERY_CACHE_PUT_PAIR statements, put_workflow_poll_pair_until and its test go with it. The batch-writer checklist item is deferred; the paired read and the other optimizations stay. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 2 - rust/src/runtime/workflow_poll_requests.rs | 85 +--------------------- 2 files changed, 2 insertions(+), 85 deletions(-) diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 9d33f33dd..55d87bef2 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -133,7 +133,6 @@ pub(super) trait Backend: Database { const QUERY_CACHE_UPDATE: &str = "UPDATE dw_query_cache SET value=$1,expiration=$2 WHERE \"key\"=$3"; const QUERY_CACHE_PUT: &str = "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3) ON CONFLICT (\"key\") DO UPDATE SET value=excluded.value,expiration=excluded.expiration"; - const QUERY_CACHE_PUT_PAIR: &str = "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3),($4,$5,$6) ON CONFLICT (\"key\") DO UPDATE SET value=excluded.value,expiration=excluded.expiration"; const QUERY_CACHE_DELETE: &str = "DELETE FROM dw_query_cache WHERE \"key\"=$1"; const QUERY_CACHE_DELETE_EXPIRED: &str = "DELETE FROM dw_query_cache WHERE \"key\"=$1 AND expiration<=$2"; // One primary-key page of the whole cache table, for the one-time @@ -660,7 +659,6 @@ impl Backend for MySql { const QUERY_CACHE_UPDATE: &str = "UPDATE dw_query_cache SET value=$1,expiration=$2 WHERE `key`=$3"; const QUERY_CACHE_PUT: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3) ON DUPLICATE KEY UPDATE value=VALUES(value),expiration=VALUES(expiration)"; - const QUERY_CACHE_PUT_PAIR: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3),($4,$5,$6) ON DUPLICATE KEY UPDATE value=VALUES(value),expiration=VALUES(expiration)"; const QUERY_CACHE_DELETE: &str = "DELETE FROM dw_query_cache WHERE `key`=$1"; const QUERY_CACHE_DELETE_EXPIRED: &str = "DELETE FROM dw_query_cache WHERE `key`=$1 AND expiration<=$2"; const QUERY_CACHE_PAGE: &str = "SELECT `key`,value,expiration FROM dw_query_cache WHERE `key`>$1 ORDER BY `key` LIMIT 64"; diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs index 8d55005fa..988a176f8 100644 --- a/rust/src/runtime/workflow_poll_requests.rs +++ b/rust/src/runtime/workflow_poll_requests.rs @@ -425,11 +425,9 @@ store_methods! { /// binding replays for the whole live lease. pub(super) async fn remember_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest, response: &Value) -> Result<()> { let ttl = result_ttl(response, request.pending_ttl); - let binding_expiration = epoch() + ttl; + Self::put_workflow_poll_row(tx, &request.keys.binding, &json!({"task_kinds": request.kinds}), ttl).await?; let kept = if response["task"].is_object() && task_kind(response) == "workflow" { 3600 } else { ttl }; - let result_expiration = epoch() + kept; - Self::put_workflow_poll_pair_until(tx, &request.keys, &json!({"task_kinds": request.kinds}), - binding_expiration, response, result_expiration).await + Self::put_workflow_poll_row(tx, &request.keys.result, response, kept).await } /// Whether a concurrent duplicate already left an answer to replay. @@ -492,19 +490,6 @@ store_methods! { Ok(()) } - async fn put_workflow_poll_pair_until( - tx: &mut Transaction<'_, DB>, keys: &WorkflowPollKeys, binding: &Value, binding_expiration: i64, - result: &Value, result_expiration: i64, - ) -> Result<()> { - // Keep both JSON values, independent expirations and original keys. - // This changes only the two writes after the existing locked claim. - Self::query(DB::QUERY_CACHE_PUT_PAIR) - .bind(&keys.binding).bind(binding.to_string()).bind(binding_expiration) - .bind(&keys.result).bind(result.to_string()).bind(result_expiration) - .execute(&mut **tx).await?; - Ok(()) - } - #[cfg(test)] pub(super) async fn expire_workflow_poll_binding_for_test(&self, request: &WorkflowPollRequest) { Self::query(DB::QUERY_CACHE_PUT).bind(&request.keys.binding) @@ -550,75 +535,9 @@ 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_writes().await; self.qualify_workflow_poll_pair_reads().await; } - #[cfg(test)] - async fn qualify_workflow_poll_pair_writes(&self) { - let keys = WorkflowPollKeys {binding:"dw-native-workflow-poll:pair-binding".into(), - result:"dw-native-workflow-poll:pair-result".into()}; - let binding = json!({"task_kinds":["workflow"]}); - for existing in [0,1,2] { - let mut tx = self.begin().await.unwrap(); - for key in [&keys.binding,&keys.result].into_iter().take(existing) { - Self::put_workflow_poll_row_until(&mut tx,key,&json!({"old":true}),1).await.unwrap(); - } - for result in [json!({"task":null,"poll_status":"empty"}),json!({"task":{"text":"escaped \\\" λ"},"poll_status":"leased"})] { - let (written,counts) = super::sql_counts::measure( - Self::put_workflow_poll_pair_until(&mut tx,&keys,&binding,1,&result,4_102_444_800), - ).await; - written.unwrap(); - assert_eq!(counts.total(),1); - assert_eq!(counts.count(DB::QUERY_CACHE_PUT_PAIR),1); - let mut actual = Vec::new(); - for key in [&keys.binding,&keys.result] { - let row = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap(); - actual.push((DB::string(&row,"value").unwrap(),DB::number(&row,"expiration").unwrap())); - } - let (old,counts) = super::sql_counts::measure(async { - Self::put_workflow_poll_row_until(&mut tx,&keys.binding,&binding,1).await?; - Self::put_workflow_poll_row_until(&mut tx,&keys.result,&result,4_102_444_800).await - }).await; - old.unwrap(); - assert_eq!(counts.total(),2); - assert_eq!(counts.count(DB::QUERY_CACHE_PUT),2); - for (key,expected) in [&keys.binding,&keys.result].into_iter().zip(actual) { - let row = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap(); - assert_eq!((DB::string(&row,"value").unwrap(),DB::number(&row,"expiration").unwrap()),expected); - } - } - tx.rollback().await.unwrap(); - for key in [&keys.binding,&keys.result] { - assert!(Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&self.pool).await.unwrap().is_none()); - } - } - let request = WorkflowPollRequest {id:"pair-writer".into(),keys:keys.clone(),pending_ttl:35, - kinds:normalized(&["workflow","update_validation"]),legacy_keys:vec![],expiration_cap:None}; - // Exercise the actual caller as well as the writer, including the - // different retention of a workflow answer whose lease has elapsed. - // These are cache-writer vectors, not admitted protocol deliveries. - for (response,ttl,kept) in [ - (json!({"task":null,"poll_status":"empty"}),35,35), - (json!({"task":{"task_id":"workflow","lease_expires_at":"2000-01-01T00:00:00Z"},"poll_status":"leased"}),35,3600), - (json!({"task":{"task_kind":"update_validation","update_validation_task_id":"validation","lease_expires_at":"2000-01-01T00:00:00Z"},"poll_status":"leased"}),35,35), - ] { - let mut tx = self.begin().await.unwrap(); - let before = epoch(); - let (result,counts) = super::sql_counts::measure(Self::remember_workflow_poll(&mut tx,&request,&response)).await; - result.unwrap(); - let after = epoch(); - assert_eq!(counts.total(),1); - assert_eq!(counts.count(DB::QUERY_CACHE_PUT_PAIR),1); - for (key,value,retained) in [(&keys.binding,json!({"task_kinds":request.kinds}),ttl),(&keys.result,response,kept)] { - let row = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap(); - assert_eq!(DB::string(&row,"value").unwrap(),value.to_string()); - assert!((before+retained..=after+retained).contains(&DB::number(&row,"expiration").unwrap())); - } - tx.rollback().await.unwrap(); - } - } - #[cfg(test)] async fn qualify_workflow_poll_pair_reads(&self) { let keys = WorkflowPollKeys {binding:"dw-native-workflow-poll:read-binding".into(), From e1e0866252f228f790c2db93b81dff2cfb6ef8a8 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 8 Nov 2026 06:22:40 +0000 Subject: [PATCH 11/13] Number the census statement's placeholders in text order for MySQL The MySQL adapter rewrites $N to ? and requires the positions to appear in order. PROJECTION_CENSUS_SQL read $3,$4,$1,$2 left to right, so every MySQL and MariaDB summary projection panicked in backend.rs. All four bindings are the same run ID, so renumbering them in text order changes no answer. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/active_summary.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust/src/runtime/active_summary.rs b/rust/src/runtime/active_summary.rs index 7c93297e9..73f9fbf25 100644 --- a/rust/src/runtime/active_summary.rs +++ b/rust/src/runtime/active_summary.rs @@ -14,7 +14,7 @@ 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=$3 AND status='received') AS received_signals,(SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$4) 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=$1) 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=$2) tasks"; +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"; From 0b14c1530045a215e512bcc94ab34ec5ada32deb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 8 Nov 2026 06:23:04 +0000 Subject: [PATCH 12/13] Correct three prepared test expectations found by the first compile - The commit handoff control counts the combined authored-run statement, the one authored-sequence scan scope validation and command application share. - A multiplexed probe tries the validation kind first in this fixture, and that claim keeps its own registration read; only the workflow claim reuses the fenced row. A plain poll still reads it zero times. - Completions need at least one command, so the waiting-status cases use a side effect, which changes no run status itself, instead of an empty list. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/commit_handoff_tests.rs | 2 +- rust/src/runtime/poll_transaction_tests.rs | 7 +++++-- rust/src/runtime/poll_write_tests.rs | 6 ++++-- 3 files changed, 10 insertions(+), 5 deletions(-) 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/poll_transaction_tests.rs b/rust/src/runtime/poll_transaction_tests.rs index 759750c9c..20ed68650 100644 --- a/rust/src/runtime/poll_transaction_tests.rs +++ b/rust/src/runtime/poll_transaction_tests.rs @@ -47,8 +47,11 @@ store_methods! { 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); - assert_eq!(reads.count("SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2"),0, - "the claim must reuse the registration its in-transaction fence validated"); + // 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; diff --git a/rust/src/runtime/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index 6c508b259..8e3793432 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -14,8 +14,10 @@ store_methods! { ("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), - ("empty",vec![],false,"waiting",1), - ("already-waiting",vec![],true,"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}"), From 50bdde972fae0914e1a821fa4a864dbf07eb552b Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 8 Nov 2026 06:55:54 +0000 Subject: [PATCH 13/13] Decode authored counters in each backend's native integer type MariaDB types COALESCE((SELECT sequence ...),0) as DECIMAL, so every authored-sequence read failed to decode there. Select each physical counter as its column's own nullable integer and count a missing row as zero in Rust. The authored-counters control reads last_history_sequence through DB::number, since that column is INT4 on PostgreSQL. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/poll_write_tests.rs | 5 +++-- rust/src/runtime/store.rs | 10 ++++++---- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/rust/src/runtime/poll_write_tests.rs b/rust/src/runtime/poll_write_tests.rs index 8e3793432..5170f33e5 100644 --- a/rust/src/runtime/poll_write_tests.rs +++ b/rust/src/runtime/poll_write_tests.rs @@ -105,8 +105,9 @@ store_methods! { } // These are stored-reader controls, not admission examples for // the wait/scope writers or their independent projections. - let first = Self::scalar("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") - .bind(run_id).fetch_one(&mut *tx).await.unwrap(); + // 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)") diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index acf622152..0bffe8a42 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -12,7 +12,7 @@ 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,COALESCE((SELECT sequence FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1),0) AS authored_activity_sequence,COALESCE((SELECT sequence FROM workflow_run_timers WHERE workflow_run_id=$2 ORDER BY sequence DESC LIMIT 1),0) AS authored_timer_sequence,COALESCE((SELECT sequence FROM workflow_child_calls WHERE parent_workflow_run_id=$3 ORDER BY sequence DESC LIMIT 1),0) AS authored_child_sequence FROM workflow_runs r WHERE r.id=$4"; +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, @@ -1323,9 +1323,11 @@ where let run = Self::query(AUTHORED_RUN_SQL) .bind(run_id).bind(run_id).bind(run_id).bind(run_id) .fetch_one(&mut **tx).await?; - let activities = DB::number(&run, "authored_activity_sequence")?; - let timers = DB::number(&run, "authored_timer_sequence")?; - let children = DB::number(&run, "authored_child_sequence")?; + // 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