Answer a proven-empty non-waiting query poll without the transition lock (#1057)

rust/src/runtime/backend.rs

@@ -80,6 +80,10 @@ pub(super) trait Backend: Database {
     // Multi-statement reads that must observe one committed state. SQLite's
     // deferred read transaction already keeps one WAL snapshot.
     const READ_SNAPSHOT_BEGIN: &str = "BEGIN";
+    // A read whose correctness needs one fixed snapshot whatever the server's
+    // configured isolation (Server #1057). SQLite and PostgreSQL's ordinary
+    // snapshot already is one; MySQL overrides it.
+    const REPEATABLE_READ_SNAPSHOT_BEGIN: &str = Self::READ_SNAPSHOT_BEGIN;
     const COMMIT_HANDOFF: bool = false;
     const HISTORY_SEQUENCE_MAX: i64 = i64::MAX;
     const HISTORY_ADVANCE_RETURNS_ROW: bool = true;
@@ -607,6 +611,12 @@ impl Backend for MySql {
         "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE namespace=$3 AND ", super::retention::reference!("payload_blob"), ")"
     );
     const READ_SNAPSHOT_BEGIN: &str = "START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY";
+    // WITH CONSISTENT SNAPSHOT keeps the session's isolation and is ignored
+    // outside REPEATABLE READ. SET TRANSACTION without SESSION or GLOBAL
+    // applies to the next transaction only (MySQL and MariaDB), so this one
+    // snapshot is fixed and the pooled session keeps its own level.
+    const REPEATABLE_READ_SNAPSHOT_BEGIN: &str =
+        "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY";
     const ACTIVITY_METADATA_HISTORY: &str = "SELECT h.sequence,h.event_type,CASE WHEN page.projection_bytes<=8388608 THEN h.payload ELSE NULL END AS payload,h.workflow_task_id,h.recorded_at,page.projection_count FROM (SELECT id,sequence,COUNT(*) OVER () AS projection_count,SUM(payload_bytes) OVER (ORDER BY sequence) AS projection_bytes FROM (SELECT id,sequence,octet_length(CAST(payload AS CHAR CHARACTER SET utf8mb4)) AS payload_bytes FROM workflow_history_events FORCE INDEX (workflow_history_events_workflow_run_id_sequence_unique) WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled')) AS sizes ORDER BY sequence LIMIT 10001) AS page JOIN workflow_history_events h ON h.id=page.id ORDER BY page.sequence";
     const PHP_CONNECTION: &str = "mysql";
     const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,TIMESTAMPDIFF(MICROSECOND,'1970-01-01 00:00:00',opened_at) AS micros FROM workflow_run_waits WHERE workflow_run_id=?), ordinals AS (SELECT id,position,CAST(ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN micros<0 THEN 2*FLOOR(micros/1000000)*1000000-micros ELSE micros END)/1000),wait_id) AS SIGNED)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE NOT (position <=> ordinal) ORDER BY ordinal LIMIT 100";
@@ -794,6 +804,20 @@ mod timestamp_tests {
     use chrono::SecondsFormat;
     use serde_json::json;
 
+    /// Server #1057: the lock-free empty query poll needs one fixed snapshot.
+    /// SQLite and PostgreSQL's ordinary snapshot is one; MySQL's is one only
+    /// at REPEATABLE READ, so its begin sets that level for the next
+    /// transaction alone, never the session or the server.
+    #[test]
+    fn repeatable_snapshot_begins_fix_one_snapshot_per_backend() {
+        assert_eq!(<sqlx::Sqlite as Backend>::REPEATABLE_READ_SNAPSHOT_BEGIN, <sqlx::Sqlite as Backend>::READ_SNAPSHOT_BEGIN);
+        assert_eq!(<sqlx::Postgres as Backend>::REPEATABLE_READ_SNAPSHOT_BEGIN, "BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY");
+        let mysql = <sqlx::MySql as Backend>::REPEATABLE_READ_SNAPSHOT_BEGIN;
+        assert_eq!(mysql, "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY");
+        assert!(!mysql.contains("SESSION") && !mysql.contains("GLOBAL"));
+        assert!(mysql.ends_with(<sqlx::MySql as Backend>::READ_SNAPSHOT_BEGIN));
+    }
+
     #[test]
     fn mysql_statement_converts_numbered_positions_and_keeps_json_paths() {
         let sql = "SELECT COUNT(*) FROM s WHERE namespace=$1 AND at>=$2 AND t IN (SELECT jt.t FROM JSON_TABLE($3,'$[*]' COLUMNS(t VARCHAR(255) PATH '$')) jt)";

rust/src/runtime/mysql.rs

@@ -1392,6 +1392,64 @@ mod tests {
         database.remove().await;
     }
 
+    /// Server #1057: a session below REPEATABLE READ. The ordinary snapshot
+    /// begin keeps that level, so a later commit becomes visible inside it;
+    /// the lock-free poll's begin fixes one snapshot for its transaction only.
+    #[tokio::test]
+    #[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"]
+    async fn qualification_repeatable_snapshot_is_fixed_under_a_read_committed_session() {
+        use super::super::backend::Backend;
+        let database = Database::new().await;
+        let mut writer = database.connection().await;
+        sqlx::query("CREATE TABLE snapshot_probe (id BIGINT PRIMARY KEY)")
+            .execute(&mut writer)
+            .await
+            .unwrap();
+        let mut reader = database.connection().await;
+        sqlx::query("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")
+            .execute(&mut reader)
+            .await
+            .unwrap();
+        let count = "SELECT COUNT(*) FROM snapshot_probe";
+        // Returns (first read, read after a concurrent commit) inside one
+        // transaction begun with `begin`.
+        async fn reads(reader: &mut MySqlConnection, writer: &mut MySqlConnection, begin: &'static str, id: i64) -> (i64, i64) {
+            let mut read = reader.begin_with(begin).await.unwrap();
+            let count = "SELECT COUNT(*) FROM snapshot_probe";
+            let before: i64 = sqlx::query_scalar(count).fetch_one(&mut *read).await.unwrap();
+            sqlx::query("INSERT INTO snapshot_probe VALUES (?)").bind(id).execute(&mut *writer).await.unwrap();
+            let during: i64 = sqlx::query_scalar(count).fetch_one(&mut *read).await.unwrap();
+            read.rollback().await.unwrap();
+            (before, during)
+        }
+        // The predecessor's premise fails on this session: no fixed snapshot.
+        assert_eq!(reads(&mut reader, &mut writer, MySql::READ_SNAPSHOT_BEGIN, 1).await, (0, 1));
+        // The successor's begin fixes one, and it stays read-only.
+        let mut read = reader.begin_with(MySql::REPEATABLE_READ_SNAPSHOT_BEGIN).await.unwrap();
+        let before: i64 = sqlx::query_scalar(count).fetch_one(&mut *read).await.unwrap();
+        sqlx::query("INSERT INTO snapshot_probe VALUES (2)").execute(&mut writer).await.unwrap();
+        let during: i64 = sqlx::query_scalar(count).fetch_one(&mut *read).await.unwrap();
+        assert_eq!((before, during), (1, 1), "the later commit stays invisible to the fixed snapshot");
+        assert!(sqlx::query("INSERT INTO snapshot_probe VALUES (3)").execute(&mut *read).await.is_err());
+        read.rollback().await.unwrap();
+        // The level was for that transaction only: the session is still
+        // READ COMMITTED, so the ordinary begin sees later commits again.
+        // (Rows 1 and 2 exist; the refused read-only insert added nothing.)
+        assert_eq!(reads(&mut reader, &mut writer, MySql::READ_SNAPSHOT_BEGIN, 4).await, (2, 3));
+        drop((reader, writer));
+        database.remove().await;
+    }
+
+    #[tokio::test]
+    #[ignore = "requires the existing disposable mysql source suite"]
+    async fn qualification_empty_query_poll_is_lock_free_only_when_proven() {
+        let db = Database::new().await;
+        let mut store = super::super::store::Store::new(MySqlStorage::open(db.options.clone()).await.unwrap().into_pool());
+        store.qualify_empty_query_poll().await;
+        store.close().await;
+        db.remove().await;
+    }
+
     #[tokio::test]
     #[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"]
     async fn qualification_concurrent_bootstrap_and_reopen() {

rust/src/runtime/poll_sweeps.rs

@@ -54,9 +54,23 @@ impl Sweeps {
         self.due_at(kind, interval, Instant::now())
     }
 
+    /// Whether `due` would run the sweep now, without reserving it. A
+    /// lock-free reader uses this to leave a due sweep to the locked path.
+    pub(super) fn is_due(&self, kind: Kind, interval: Duration) -> bool {
+        self.is_due_at(kind, interval, Instant::now())
+    }
+
+    fn is_due_at(&self, kind: Kind, interval: Duration, at: Instant) -> bool {
+        let last = self.slot(kind).lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        last.is_none_or(|last| at.saturating_duration_since(last) >= interval)
+    }
+
+    fn slot(&self, kind: Kind) -> &Mutex<Option<Instant>> {
+        match kind { Kind::Workflow => &self.workflow, Kind::Query => &self.query, Kind::Receipts => &self.receipts }
+    }
+
     fn due_at(&self, kind: Kind, interval: Duration, at: Instant) -> bool {
-        let slot = match kind { Kind::Workflow => &self.workflow, Kind::Query => &self.query, Kind::Receipts => &self.receipts };
-        let mut last = slot.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut last = self.slot(kind).lock().unwrap_or_else(std::sync::PoisonError::into_inner);
         if last.is_some_and(|last| at.saturating_duration_since(last) < interval) { return false; }
         *last = Some(at);
         true
@@ -108,4 +122,21 @@ mod tests {
         assert!(!sweeps.due_at(Kind::Query, interval, start + Duration::from_millis(72)));
         assert!(sweeps.due_at(Kind::Query, interval, start + interval));
     }
+
+    #[test]
+    fn a_due_peek_never_reserves_or_postpones_the_sweep() {
+        let sweeps = Sweeps::default();
+        let interval = Duration::from_millis(73);
+        let start = Instant::now();
+        // Never run: due, and peeking any number of times leaves it due.
+        for _ in 0..3 {
+            assert!(sweeps.is_due_at(Kind::Query, interval, start));
+        }
+        assert!(sweeps.due_at(Kind::Query, interval, start), "the locked caller still takes it");
+        assert!(!sweeps.is_due_at(Kind::Query, interval, start + Duration::from_millis(72)));
+        assert!(sweeps.is_due_at(Kind::Query, interval, start + interval));
+        assert!(sweeps.is_due_at(Kind::Query, interval, start + interval));
+        assert!(sweeps.due_at(Kind::Query, interval, start + interval), "a peek did not move the reservation");
+        assert!(sweeps.is_due_at(Kind::Workflow, interval, start), "kinds stay independent");
+    }
 }

rust/src/runtime/postgres.rs

@@ -751,6 +751,16 @@ mod tests {
         db.remove().await;
     }
 
+    #[tokio::test]
+    #[ignore = "requires the existing disposable postgres source suite"]
+    async fn qualification_empty_query_poll_is_lock_free_only_when_proven() {
+        let db = Database::new().await;
+        let mut store = super::super::store::Store::new(PostgresStorage::open(db.options.clone()).await.unwrap().into_pool());
+        store.qualify_empty_query_poll().await;
+        store.close().await;
+        db.remove().await;
+    }
+
     #[tokio::test]
     #[ignore = "requires the existing disposable postgres source suite"]
     async fn qualification_transition_queue_preserves_reads_fifo_and_timeout_admission() {

rust/src/runtime/queries.rs

@@ -37,6 +37,9 @@ pub(super) const RESULT_BYTES: usize = 256 * 1024;
 
 #[path = "query_poll_requests.rs"]
 mod poll_requests;
+#[cfg(test)]
+#[path = "query_empty_poll_tests.rs"]
+mod empty_poll_tests;
 
 enum QueryClaim {
     Task(Value),
@@ -408,6 +411,43 @@ where
         format!("{OPEN_PREFIX}{digest}")
     }
 
+    /// Whether, read in `tx`, a locked claim in this namespace would find no
+    /// open query task and write no query bookkeeping (Server #1057): the
+    /// 3.0.1 cutover is done, the receipt sweep has no minute to retire, and
+    /// the namespace's open set is absent or exactly empty. A task joins the
+    /// open set in the transaction that writes it, so a snapshot that cannot
+    /// see the membership cannot see the task either. A row this reader does
+    /// not recognize exactly is not evidence of an empty namespace; the
+    /// caller then takes the locked path, which keeps its own handling.
+    pub(super) async fn query_bookkeeping_idle(tx: &mut Transaction<'_, DB>, namespace: &str) -> Result<bool> {
+        let Some(Some(cutover)) = Self::cache_row(tx, CUTOVER_MARK).await? else {
+            return Ok(false);
+        };
+        if cutover["done"] != true {
+            return Ok(false);
+        }
+        // `sweep_query_receipts` writes when its mark is missing or a whole
+        // minute after the mark has passed.
+        let Some(Some(mark)) = Self::cache_row(tx, EXPIRY_MARK).await? else {
+            return Ok(false);
+        };
+        if mark["minute"].as_i64().and_then(|minute| minute.checked_add(1)).is_none_or(|next| next < now().div_euclid(60)) {
+            return Ok(false);
+        }
+        Ok(match Self::cache_row(tx, &Self::open_set_key(namespace)).await? {
+            None => true,
+            Some(set) => set == Some(json!({"namespace": namespace, "ids": []})),
+        })
+    }
+
+    /// A bookkeeping row as stored: absent, or its value if it parses.
+    async fn cache_row(tx: &mut Transaction<'_, DB>, key: &str) -> Result<Option<Option<Value>>> {
+        let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&mut **tx).await? else {
+            return Ok(None);
+        };
+        Ok(Some(serde_json::from_str(&DB::string(&row, "value")?).ok()))
+    }
+
     /// The namespace's listed open task ids: unique canonical ULIDs, at most
     /// `MAX_ENTRIES`. A row naming another namespace, or none, lists nothing.
     async fn open_set(tx: &mut Transaction<'_, DB>, namespace: &str) -> Result<Vec<String>> {