diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 4a387e4ac..b5d8a1c37 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1225,6 +1225,7 @@ mod tests { let store = super::super::store::Store::new(MySqlStorage::open(db.options.clone()).await.unwrap().into_pool()); store.qualify_workflow_poll_transactions().await; store.qualify_first_candidate_batch().await; + store.qualify_claim_history_sequence().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 6f0214ade..8166571a3 100644 --- a/rust/src/runtime/poll_transaction_tests.rs +++ b/rust/src/runtime/poll_transaction_tests.rs @@ -4,6 +4,8 @@ use crate::codec::{Value as Payload, ValueCodec}; use serde_json::{Value, json}; use std::{collections::BTreeMap, sync::{Arc, atomic::Ordering}, time::Duration}; use tokio::sync::Semaphore; +#[cfg(debug_assertions)] +use futures_util::FutureExt; fn input() -> Value { json!({"codec":"avro","blob":ValueCodec::new().unwrap().encode(&Payload::Array(vec![])).unwrap()}) @@ -184,6 +186,47 @@ store_methods! { } } + pub(super) async fn qualify_claim_history_sequence(&self) { + let started = self.count_start("claimed-history-total").await; + let run_id = started["run_id"].as_str().unwrap(); + let mut tx = self.begin().await.unwrap(); + for sequence in 1..=2 { + Self::append(&mut tx,run_id,"SideEffectRecorded",json!({"sequence":sequence,"result":null}),None,None) + .await.unwrap(); + } + let run = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(&mut *tx).await.unwrap(); + let last = DB::number(&run,"last_history_sequence").unwrap(); + assert_eq!(last,4); + for (after,size) in [(0,1),(1,2),(0,1000),(last,1)] { + let (reference, reference_count) = super::sql_counts::measure( + Self::history_page_connection(&mut *tx,run_id,after,size), + ).await; + let (known, known_count) = super::sql_counts::measure( + Self::history_page_connection_with_total(&mut *tx,run_id,after,size,Some(last)), + ).await; + assert_eq!(known.unwrap(),reference.unwrap(),"after={after}, size={size}"); + let query = "SELECT last_history_sequence FROM workflow_runs WHERE id=$1"; + assert_eq!(reference_count.count(query),1); + assert_eq!(known_count.count(query),usize::from(cfg!(debug_assertions)), + "only debug builds re-read the claimed history total"); + } + #[cfg(debug_assertions)] + { + Self::append(&mut tx,run_id,"SideEffectRecorded",json!({"sequence":3,"result":null}),None,None) + .await.unwrap(); + let refused = std::panic::AssertUnwindSafe( + Self::history_page_connection_with_total(&mut *tx,run_id,0,1,Some(last)), + ).catch_unwind().await.expect_err("an intervening history write must trip the debug assertion"); + let message = refused.downcast_ref::().map(String::as_str) + .or_else(|| refused.downcast_ref::<&str>().copied()).unwrap_or(""); + assert!(message.contains("claimed run history sequence changed before its first page"),"{message}"); + } + tx.rollback().await.unwrap(); + Self::query("UPDATE workflow_tasks SET status='completed' WHERE workflow_run_id=$1") + .bind(run_id).execute(&self.pool).await.unwrap(); + } + pub(super) async fn qualify_query_poll_transactions(&self) { self.count_worker().await; let workflow_id = "counted-query-workflow"; @@ -250,6 +293,7 @@ async fn sqlite_workflow_and_query_poll_transaction_counts() { let store = Store::::new(super::sqlite::open(dir.path().join("counts.sqlite").to_str().unwrap()).await.unwrap()); store.qualify_workflow_poll_transactions().await; store.qualify_first_candidate_batch().await; + store.qualify_claim_history_sequence().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 ff601d923..e9544ee46 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -745,6 +745,7 @@ mod tests { let store = super::super::store::Store::new(PostgresStorage::open(db.options.clone()).await.unwrap().into_pool()); store.qualify_workflow_poll_transactions().await; store.qualify_first_candidate_batch().await; + store.qualify_claim_history_sequence().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 f4a7963f9..8c99338f5 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -2700,15 +2700,16 @@ where } } } + let last = DB::number(&run, "last_history_sequence")?; let (mut events, next) = match page { - Some(size) => Self::history_page_connection(&mut **tx, &run_id, 0, size).await?, - None => (Self::whole_history(&mut **tx, &run_id).await?, None), + Some(size) => Self::history_page_connection_with_total(&mut **tx, &run_id, 0, size, Some(last)).await?, + None => (Self::whole_history(&mut **tx, &run_id, last).await?, None), }; Self::with_signal_sequences(&mut **tx, &run_id, &mut events).await?; let budget = Self::workflow_history_budget( tx, &run_id, - DB::number(&run, "last_history_sequence")?, + last, ) .await?; budget.write_task(&mut claim); @@ -4977,11 +4978,13 @@ where Ok(()) } - async fn whole_history(connection: &mut DB::Connection, run_id: &str) -> Result> { + async fn whole_history(connection: &mut DB::Connection, run_id: &str, last: i64) -> Result> { let mut events = Vec::new(); let mut after = 0; loop { - let (page, next) = Self::history_page_connection(&mut *connection, run_id, after, 1000).await?; + let (page, next) = Self::history_page_connection_with_total( + &mut *connection, run_id, after, 1000, (after == 0).then_some(last), + ).await?; if let Some(last) = page.last() { after = last["sequence"].as_i64().unwrap_or(after); } @@ -4997,6 +5000,16 @@ where run_id: &str, after_sequence: i64, page_size: i64, + ) -> Result<(Vec, Option)> { + Self::history_page_connection_with_total(connection, run_id, after_sequence, page_size, None).await + } + + pub(super) async fn history_page_connection_with_total( + connection: &mut DB::Connection, + run_id: &str, + after_sequence: i64, + page_size: i64, + known_total: Option, ) -> Result<(Vec, Option)> { use base64::{Engine, engine::general_purpose::STANDARD}; // Bound serialized history bytes before hydrating JSON. SQLite executes @@ -5007,11 +5020,20 @@ where .bind(page_size) .fetch_all(&mut *connection) .await?; - let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") - .bind(run_id) - .fetch_one(connection) - .await?; - let total = DB::number(&row, "last_history_sequence")?; + // A workflow claim already read this run under the transition lock. + // No intervening history write is allowed before its first page. + let observed_total = if known_total.is_none() || cfg!(debug_assertions) { + let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(connection).await?; + Some(DB::number(&row, "last_history_sequence")?) + } else { None }; + let total = match known_total { + Some(total) => { + debug_assert_eq!(observed_total, Some(total), "claimed run history sequence changed before its first page"); + total + } + None => observed_total.expect("an unknown history total was read"), + }; let last = match rows.last() { Some(row) => DB::number(row, "sequence")?, None => after_sequence,