From c60802aaa908434b69320260f75584fcf3efaaac Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 12 Nov 2026 10:14:21 +0000 Subject: [PATCH] Write a workflow poll's binding and answer in one two-row upsert (#1063) remember_workflow_poll wrote the binding row and then the answer row with two sequential upserts inside the claim transaction, after bind_workflow_poll already holds the binding's lock. Both now go in one QUERY_CACHE_PUT_PAIR statement (ON CONFLICT on PostgreSQL/SQLite, ON DUPLICATE KEY UPDATE on MySQL/MariaDB; the key is the table's only unique key), dropping one locked round trip per remembered poll answer. Keys, JSON values and the two retention rules are unchanged: the binding keeps result_ttl, a workflow lease's answer keeps 3600 s and any other answer keeps the binding's ttl. Each expiry is still the clock read just before the statement that writes the row plus that row's ttl; one statement now writes both rows, so they share that one reading. The sequential writer read the answer's clock after the binding's round trip, so its answer could expire up to one whole second later when that round trip crossed a second boundary. The pair can only leave the answer at the earlier of those values, which the sequential writer also produces whenever no boundary is crossed, and the binding still never outlives the answer. The common backend qualification (SQLite, PostgreSQL, MySQL) compares the pair statement with the two sequential upserts for every existing-row combination, checks a neighbouring key is untouched, that a rolled-back transition restores committed rows exactly or leaves none, and that the caller issues one statement whose two expiries differ by exactly the two ttls' difference. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 4 + rust/src/runtime/workflow_poll_requests.rs | 130 ++++++++++++++++++++- 2 files changed, 132 insertions(+), 2 deletions(-) diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 890fa42c2..4afa44c6a 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -151,6 +151,9 @@ 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"; + // A workflow poll's binding and answer rows, upserted together. Each row + // keeps its own key, value and expiration; only the round trip is shared. + 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 @@ -683,6 +686,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 54a165ebf..4f987191f 100644 --- a/rust/src/runtime/workflow_poll_requests.rs +++ b/rust/src/runtime/workflow_poll_requests.rs @@ -424,11 +424,19 @@ store_methods! { /// binding with PHP's result expiry. A workflow lease's answer is kept /// while its task can be delivered (the row decides), as PHP's lease /// binding replays for the whole live lease. + /// + /// Both rows go in one two-row upsert, after the claim's binding lock. + /// Each expiry is still the clock read just before the statement that + /// writes the row plus that row's own ttl, as each of PHP's two puts + /// reads its own clock; one statement now writes both, so they share + /// one reading. The answer can expire no later than two sequential + /// writes would leave it, and the binding still never outlives it. 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 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 origin = epoch(); + Self::put_workflow_poll_pair_until(tx, &request.keys, &json!({"task_kinds": request.kinds}), origin + ttl, + response, origin + kept).await } /// Whether a concurrent duplicate already left an answer to replay. @@ -481,10 +489,24 @@ store_methods! { Ok(Some(serde_json::from_str(&DB::string(&row, "value")?)?)) } + #[cfg(test)] async fn put_workflow_poll_row(tx: &mut Transaction<'_, DB>, key: &str, value: &Value, ttl: i64) -> Result<()> { Self::put_workflow_poll_row_until(tx, key, value, epoch() + ttl).await } + /// The binding and answer rows of one poll request in one statement. + /// Keys, JSON values and both expirations are exactly the caller's. + async fn put_workflow_poll_pair_until( + tx: &mut Transaction<'_, DB>, keys: &WorkflowPollKeys, binding: &Value, binding_expiration: i64, + result: &Value, result_expiration: i64, + ) -> Result<()> { + 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(()) + } + async fn put_workflow_poll_row_until(tx: &mut Transaction<'_, DB>, key: &str, value: &Value, expiration: i64) -> Result<()> { Self::query(DB::QUERY_CACHE_PUT).bind(key).bind(value.to_string()) .bind(expiration).execute(&mut **tx).await?; @@ -536,9 +558,113 @@ 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","update_validation"]}); + let stored = |row: DB::Row| (DB::string(&row,"value").unwrap(),DB::number(&row,"expiration").unwrap()); + // The pair statement stores exactly the rows the two sequential + // upserts store for the same values and expirations, whichever of + // the two rows already exists, and touches no other key. + for existing in [0,1,2,3] { + let mut tx = self.begin().await.unwrap(); + Self::put_workflow_poll_row_until(&mut tx,"dw-native-workflow-poll:pair-neighbour",&json!({"kept":true}),7).await.unwrap(); + for (bit,key) in [(1,&keys.binding),(2,&keys.result)] { + if existing & bit != 0 { + Self::put_workflow_poll_row_until(&mut tx,key,&json!({"old":bit}),1).await.unwrap(); + } + } + for (result,binding_expiration,result_expiration) in [ + (json!({"task":null,"poll_status":"empty"}),40_i64,41_i64), + (json!({"task":{"text":"escaped \\\" λ"},"poll_status":"leased"}),4_102_444_700,4_102_444_800), + ] { + let (written,counts) = super::sql_counts::measure(Self::put_workflow_poll_pair_until( + &mut tx,&keys,&binding,binding_expiration,&result,result_expiration)).await; + written.unwrap(); + assert_eq!(counts.total(),1); + assert_eq!(counts.count(DB::QUERY_CACHE_PUT_PAIR),1); + let mut paired = Vec::new(); + for key in [&keys.binding,&keys.result] { + paired.push(stored(Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap())); + } + assert_eq!(paired,vec![(binding.to_string(),binding_expiration),(result.to_string(),result_expiration)]); + let (sequential,counts) = super::sql_counts::measure(async { + Self::put_workflow_poll_row_until(&mut tx,&keys.binding,&binding,binding_expiration).await?; + Self::put_workflow_poll_row_until(&mut tx,&keys.result,&result,result_expiration).await + }).await; + sequential.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(paired) { + assert_eq!(stored(Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap()),expected); + } + } + let neighbour = Self::query(DB::QUERY_CACHE_GET).bind("dw-native-workflow-poll:pair-neighbour").fetch_one(&mut *tx).await.unwrap(); + assert_eq!(stored(neighbour),(json!({"kept":true}).to_string(),7)); + tx.rollback().await.unwrap(); + } + // Rolling back the transition restores both committed rows, value + // and expiry, or leaves no row where there was none. + for existing in [false,true] { + if existing { + for (key,value,expiration) in [(&keys.binding,"{\"task_kinds\":[\"old\"]}",11_i64),(&keys.result,"{\"task\":null,\"poll_status\":\"old\"}",12)] { + Self::query(DB::QUERY_CACHE_PUT).bind(key).bind(value).bind(expiration).execute(&self.pool).await.unwrap(); + } + } + let mut before = Vec::new(); + for key in [&keys.binding,&keys.result] { + before.push(Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&self.pool).await.unwrap().map(stored)); + } + let mut tx = self.begin().await.unwrap(); + Self::put_workflow_poll_pair_until(&mut tx,&keys,&binding,2_000_000_000,&json!({"task":null,"poll_status":"empty"}),2_000_000_001).await.unwrap(); + assert!(Self::query(DB::QUERY_CACHE_GET).bind(&keys.result).fetch_optional(&mut *tx).await.unwrap().is_some()); + tx.rollback().await.unwrap(); + for (key,expected) in [&keys.binding,&keys.result].into_iter().zip(before) { + let after = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&self.pool).await.unwrap().map(stored); + assert_eq!(after,expected); + assert_eq!(after.is_some(),existing); + } + for key in [&keys.binding,&keys.result] { + Self::query(DB::QUERY_CACHE_DELETE).bind(key).execute(&self.pool).await.unwrap(); + } + } + // The caller: one statement; each row is its own ttl past one clock + // reading taken inside the call, so the answer's expiry exceeds the + // binding's by exactly the difference of the two ttls. A workflow + // lease's answer is kept for an hour, any other answer only as long + // as its binding. These are cache-writer vectors, not admitted + // protocol deliveries. + let request = WorkflowPollRequest {id:"pair-writer".into(),keys:keys.clone(),pending_ttl:35, + kinds:normalized(&["workflow","update_validation"]),legacy_keys:vec![],expiration_cap:None}; + 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); + let mut expirations = Vec::new(); + for (key,value,retained) in [(&keys.binding,json!({"task_kinds":request.kinds}),ttl),(&keys.result,response,kept)] { + let (stored_value,expiration) = stored(Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_one(&mut *tx).await.unwrap()); + assert_eq!(stored_value,value.to_string()); + assert!((before+retained..=after+retained).contains(&expiration)); + expirations.push(expiration); + } + assert_eq!(expirations[1]-expirations[0],kept-ttl); + 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(),