Reuse the claimed history total for the first history page

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;

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::<String>().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::<sqlx::Sqlite>::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;
 }

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;

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<Vec<Value>> {
+    async fn whole_history(connection: &mut DB::Connection, run_id: &str, last: i64) -> Result<Vec<Value>> {
         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<Value>, Option<String>)> {
+        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<i64>,
     ) -> Result<(Vec<Value>, Option<String>)> {
         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,