diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 7161edcdc..03b027f55 100644 --- a/rust/src/runtime/backend.rs +++ b/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!(::REPEATABLE_READ_SNAPSHOT_BEGIN, ::READ_SNAPSHOT_BEGIN); + assert_eq!(::REPEATABLE_READ_SNAPSHOT_BEGIN, "BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY"); + let mysql = ::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(::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)"; diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index b5d8a1c37..92bd2f8ae 100644 --- a/rust/src/runtime/mysql.rs +++ b/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() { diff --git a/rust/src/runtime/poll_sweeps.rs b/rust/src/runtime/poll_sweeps.rs index b5b27618d..7b0715aba 100644 --- a/rust/src/runtime/poll_sweeps.rs +++ b/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> { + 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"); + } } diff --git a/rust/src/runtime/postgres.rs b/rust/src/runtime/postgres.rs index e9544ee46..804a33504 100644 --- a/rust/src/runtime/postgres.rs +++ b/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() { diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index 915ebd5ed..3592b6a4a 100644 --- a/rust/src/runtime/queries.rs +++ b/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 { + 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>> { + 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> { diff --git a/rust/src/runtime/query_empty_poll_tests.rs b/rust/src/runtime/query_empty_poll_tests.rs new file mode 100644 index 000000000..b730f1184 --- /dev/null +++ b/rust/src/runtime/query_empty_poll_tests.rs @@ -0,0 +1,351 @@ +//! Server #1057: a new non-waiting query poll is answered "empty" from one +//! read snapshot, without the transition lock, only while every fact of that +//! answer holds at once. Each other state takes the maintained locked path. +// `now` is the query cache's clock: epoch seconds, as the bookkeeping rows use. +use super::{CUTOVER_MARK, EXPIRY_MARK, now, read_arguments}; +use crate::codec::{Value as Payload, ValueCodec}; +use crate::runtime::{QueryTaskConfig, RuntimeError, store::Store, worker_poll_fence::PollAdmission}; +use serde_json::{Value, json}; +use std::{collections::BTreeMap, sync::{Arc, atomic::Ordering}, time::{Duration, Instant}}; +use tokio::sync::{Semaphore, oneshot}; + +const WORKER: &str = "empty-worker"; +const QUEUE: &str = "empty-queue"; +const WORKFLOW: &str = "empty-workflow"; + +fn input() -> Value { + json!({"codec":"avro","blob":ValueCodec::new().unwrap().encode(&Payload::Array(vec![])).unwrap()}) +} + +fn body(id: &str) -> Value { + json!({"worker_id":WORKER,"task_queue":QUEUE,"poll_request_id":id,"timeout_seconds":0}) +} + +fn config() -> QueryTaskConfig { + QueryTaskConfig { timeout_seconds: 10, ..Default::default() } +} + +store_methods! { + /// A query-capable worker and a closed run it can answer queries for. + async fn empty_fixture(&self) { + self.register(json!({"worker_id":WORKER,"task_queue":QUEUE,"runtime":"php", + "supported_workflow_types":["probe"],"supported_activity_types":[],"capabilities":["query_tasks"], + "workflow_command_contracts":{"probe":{"queries":["probe"],"query_contracts":[{"name":"probe","parameters":[]}]}}}), + &QueryTaskConfig::default(), Some("1.20")).await.unwrap(); + self.start(json!({"workflow_id":WORKFLOW,"workflow_type":"probe","task_queue":QUEUE,"input":input()}), + BTreeMap::new(), None, None).await.unwrap(); + let task = self.poll(json!({"worker_id":WORKER,"task_queue":QUEUE,"timeout_seconds":0}), "workflow", "1.20").await.unwrap(); + self.complete_or_checkpoint(task["task"]["task_id"].as_str().unwrap(), json!({"lease_owner":WORKER, + "workflow_task_attempt":task["task"]["workflow_task_attempt"], + "commands":[{"type":"complete_workflow","result":input()}]}), + Arc::new(Semaphore::new(1)), BTreeMap::new(), None, + crate::runtime::carried_polls::Config { protocol: Some("1.20"), pages: Default::default(), default_wait: 30 }, + Default::default(), None, None).await.unwrap(); + // One locked poll finishes the cutover, starts the receipt sweep's + // mark and reserves the poll-row sweep. + let warm = self.poll_query(body("warm"), &config()).await.unwrap(); + assert!(warm["task"].is_null(), "{warm}"); + self.pin_expiry_mark().await; + } + + /// Keep the receipt sweep idle across a minute boundary during a test. + async fn pin_expiry_mark(&self) { + let mut tx = self.begin().await.unwrap(); + Self::put_cache_document(&mut tx, EXPIRY_MARK, &json!({"minute": now().div_euclid(60) + 1000})).await.unwrap(); + tx.commit().await.unwrap(); + } + + async fn put_raw(&self, key: &str, raw: &str) { + let mut tx = self.begin().await.unwrap(); + Self::query(DB::QUERY_CACHE_PUT).bind(key.to_owned()).bind(raw.to_owned()).bind(i64::MAX / 2) + .execute(&mut *tx).await.unwrap(); + tx.commit().await.unwrap(); + } + + async fn delete_raw(&self, key: &str) { + let mut tx = self.begin().await.unwrap(); + Self::query(DB::QUERY_CACHE_DELETE).bind(key.to_owned()).execute(&mut *tx).await.unwrap(); + tx.commit().await.unwrap(); + } + + async fn raw(&self, key: &str) -> Option { + let row = Self::query(DB::QUERY_CACHE_GET).bind(key.to_owned()).fetch_optional(&self.pool).await.unwrap()?; + serde_json::from_str(&DB::string(&row, "value").unwrap()).ok() + } + + /// Every cache row, to show an answer wrote nothing. + async fn cache_rows(&self) -> Vec<(String, String, i64)> { + let mut rows = vec![]; + let mut after = String::new(); + loop { + let page = Self::query(DB::QUERY_CACHE_PAGE).bind(after.clone()).fetch_all(&self.pool).await.unwrap(); + for row in &page { + rows.push((DB::string(row, "key").unwrap(), DB::string(row, "value").unwrap(), DB::number(row, "expiration").unwrap())); + } + let last = rows.last().map(|(key, ..)| key.clone()); + match last { + Some(key) if page.len() == 64 => after = key, + _ => return rows, + } + } + } + + /// One public non-waiting poll and the transition locks it took. + async fn locks_for(&self, id: &str) -> (crate::runtime::Result, usize) { + self.test_transitions.store(0, Ordering::Relaxed); + let answer = self.poll_query(body(id), &config()).await; + (answer, self.test_transitions.load(Ordering::Relaxed)) + } + + async fn assert_lock_free_empty(&self, id: &str, why: &str) { + let before = self.cache_rows().await; + let (answer, locks) = self.locks_for(id).await; + let answer = answer.unwrap(); + assert_eq!((answer["task"].clone(), answer["poll_status"].clone()), (Value::Null, json!("empty")), "{why}: {answer}"); + assert_eq!(locks, 0, "{why}: the proven empty answer takes no transition lock"); + assert_eq!(self.cache_rows().await, before, "{why}: the proven empty answer writes nothing"); + } + + async fn assert_forced_lock(&self, id: &str, why: &str) -> Value { + let (answer, locks) = self.locks_for(id).await; + let answer = answer.unwrap_or_else(|error| panic!("{why}: {error}")); + assert!(locks >= 1, "{why}: an unproven fact takes the locked path"); + answer + } + + /// The proof inputs `poll_query` computes before it reaches the peek. + async fn proof_inputs(&self, id: &str) -> (String, Option, super::poll_requests::PollKeys) { + let proof = Self::query_worker_proof(&self.pool, self.namespace.as_ref(), WORKER, &config()).await.unwrap().unwrap(); + let fence = match self.admit_worker_poll(WORKER, QUEUE).await.unwrap() { + PollAdmission::Fenced(fence) => Some(fence), + _ => None, + }; + (proof, fence, self.poll_keys(&body(id), id).await.unwrap()) + } + + async fn peek(&self, id: &str, proof: Option<&str>) -> bool { + let (real, fence, keys) = self.proof_inputs(id).await; + self.peek_empty_poll(&body(id), &config(), proof.unwrap_or(&real), fence.as_ref(), &keys, std::future::ready(())).await.unwrap() + } + + async fn pending_query_visible(&self) -> bool { + // Task rows sort after the poll rows, so read every page. + self.cache_rows().await.iter().any(|(_, value, _)| { + serde_json::from_str::(value) + .is_ok_and(|value| value["query_name"] == "probe" && value["status"] == "pending") + }) + } + + async fn wait_for_pending_query(&self) { + tokio::time::timeout(Duration::from_secs(5), async { + while !self.pending_query_visible().await { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }).await.expect("the racing query commits"); + } + + async fn submit_query(&self) -> crate::runtime::Result<(axum::http::StatusCode, Value)> { + let blob = input()["blob"].as_str().unwrap().to_owned(); + let values = read_arguments(Arc::new(Semaphore::new(1)), blob.clone()).await.unwrap(); + self.query_workflow(WORKFLOW, None, "probe", blob, values, &config(), None).await + } + + async fn answer(&self, claimed: &Value) { + self.finish_query(claimed["task"]["query_task_id"].as_str().unwrap(), json!({"lease_owner":WORKER, + "query_task_attempt":claimed["task"]["query_task_attempt"],"result":7}), false, &config()).await.unwrap(); + } + + async fn qualify_lock_free_empty_answer(&self) { + // A retry of a remembered request replays it under the lock. + assert!(self.assert_forced_lock("warm", "remembered request").await["task"].is_null()); + self.assert_lock_free_empty("fresh-1", "proven empty").await; + // Nothing was remembered, so its retry polls again, also lock-free. + self.assert_lock_free_empty("fresh-1", "retry of an unremembered empty answer").await; + // A registration that no longer matches the poll's proof. + assert!(!self.peek("fresh-2", Some("not-this-registration")).await); + assert!(self.peek("fresh-2", None).await); + } + + async fn qualify_bookkeeping_forces_lock(&self) { + let namespace = self.namespace.to_string(); + let open = Self::open_set_key(&namespace); + // The receipt sweep has whole minutes to retire: locked, and it runs. + self.put_raw(EXPIRY_MARK, &json!({"minute": now().div_euclid(60) - 5}).to_string()).await; + self.assert_forced_lock("mark-behind", "receipt sweep due").await; + let mark = self.raw(EXPIRY_MARK).await.unwrap()["minute"].as_i64().unwrap(); + assert!(mark + 1 >= now().div_euclid(60), "the locked path advanced the mark: {mark}"); + self.delete_raw(EXPIRY_MARK).await; + self.assert_forced_lock("mark-missing", "receipt sweep never started").await; + assert!(self.raw(EXPIRY_MARK).await.is_some(), "the locked path started the mark"); + self.pin_expiry_mark().await; + // An unfinished cutover: locked, and the locked path finishes it. + self.put_raw(CUTOVER_MARK, &json!({"after":"~","done":false}).to_string()).await; + self.assert_forced_lock("cutover", "cutover not done").await; + assert_eq!(self.raw(CUTOVER_MARK).await.unwrap()["done"], true); + self.put_raw(CUTOVER_MARK, "not json").await; + assert!(!self.peek("cutover-malformed", None).await, "an unreadable cutover marker proves nothing"); + self.put_raw(CUTOVER_MARK, &json!({"after":"~","done":true}).to_string()).await; + // Open-set rows this reader does not recognize exactly. The locked + // path keeps its own reading of each. + for (raw, why) in [ + ("not json".to_owned(), "unreadable open set"), + (json!({"namespace":"another","ids":[]}).to_string(), "another namespace's open set"), + (json!({"namespace":namespace,"ids":[],"extra":true}).to_string(), "unexpected open-set field"), + (json!({"namespace":namespace,"ids":["not-a-ulid"]}).to_string(), "non-canonical member"), + (json!({"namespace":namespace,"ids":"none"}).to_string(), "ids not a list"), + ] { + self.put_raw(&open, &raw).await; + let answer = self.assert_forced_lock(&format!("open-{why}"), why).await; + assert!(answer["task"].is_null(), "{why}: {answer}"); + } + // Exactly empty, or absent, is proven. + self.put_raw(&open, &json!({"namespace":namespace,"ids":[]}).to_string()).await; + self.assert_lock_free_empty("open-empty", "exactly empty open set").await; + self.delete_raw(&open).await; + self.assert_lock_free_empty("open-absent", "absent open set").await; + } + + async fn qualify_request_rows_force_lock(&self) { + let (_, _, keys) = self.proof_inputs("own").await; + // This request already has a remembered answer or a pending leader. + self.put_test_poll_row(&keys, "result", Some("own"), &json!({"query_task_id":null,"poll_status":"empty"}).to_string(), 60).await; + assert!(!self.peek("own", None).await, "remembered request"); + let (_, _, keys) = self.proof_inputs("own-pending").await; + self.put_test_poll_row(&keys, "pending", Some("own-pending"), &json!({"pending":true}).to_string(), 60).await; + assert!(!self.peek("own-pending", None).await, "pending request"); + // The worker's current request is another one that has not answered: + // it may be about to wait, and only the locked path ends it. + self.put_test_poll_row(&keys, "current", None, &json!({"poll_request_id":"other"}).to_string(), 60).await; + self.assert_forced_lock("after-unanswered", "current request without an answer").await; + self.put_test_poll_row(&keys, "current", None, &json!({"poll_request_id":"other"}).to_string(), 60).await; + self.put_test_poll_row(&keys, "result", Some("other"), &json!({"query_task_id":null,"poll_status":"empty"}).to_string(), 60).await; + self.put_test_poll_row(&keys, "pending", Some("other"), &json!({"pending":true}).to_string(), 60).await; + assert!(!self.peek("after-pending", None).await, "current request still pending"); + // An expired row is unproven, not absent: the locked path deletes it. + self.put_test_poll_row(&keys, "pending", Some("other"), &json!({"pending":true}).to_string(), -1).await; + assert!(!self.peek("after-expired-pending", None).await, "an expired row is unproven"); + // A current row this reader cannot read exactly. The locked path + // overwrites it as before, without a new refusal. + for (raw, why) in [("not json", "unreadable current row"), ("\"other\"", "current row not an object"), + (r#"{"poll_request_id":"other","extra":1}"#, "unexpected current field"), (r#"{"poll_request_id":7}"#, "current id not text")] { + self.put_test_poll_row(&keys, "current", None, raw, 60).await; + let answer = self.assert_forced_lock(&format!("current-{why}"), why).await; + assert!(answer["task"].is_null(), "{why}: {answer}"); + } + // The current row names this very request. + self.put_test_poll_row(&keys, "current", None, &json!({"poll_request_id":"self-named"}).to_string(), 60).await; + assert!(!self.peek("self-named", None).await, "current names this request"); + } + + /// A newer request still ends an older waiting one: it is never answered + /// lock-free while that request is pending. + async fn qualify_waiting_request_is_superseded(&self) { + let (_, _, keys) = self.proof_inputs("waiting").await; + let waiting = async { + let started = Instant::now(); + let answer = self.poll_query(json!({"worker_id":WORKER,"task_queue":QUEUE,"poll_request_id":"waiting","timeout_seconds":5}), &config()).await.unwrap(); + (answer, started.elapsed()) + }; + let newer = async { + tokio::time::timeout(Duration::from_secs(3), async { + while self.test_poll_row(&keys, "pending", Some("waiting")).await.is_none() { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }).await.expect("the older request starts waiting"); + assert!(!self.peek("newer", None).await, "a pending current request forces the lock"); + let newer = self.poll_query(body("newer"), &config()).await.unwrap(); + assert!(newer["task"].is_null(), "{newer}"); + }; + let ((answer, elapsed), ()) = tokio::join!(waiting, newer); + assert_eq!(answer["poll_status"], "empty", "{answer}"); + assert!(elapsed < Duration::from_secs(4), "the newer request ended the older wait: {elapsed:?}"); + } + + async fn qualify_registration_mismatch_keeps_its_refusal(&self) { + let other = json!({"worker_id":WORKER,"task_queue":"another-queue","poll_request_id":"wrong-queue","timeout_seconds":0}); + match self.poll_query(other, &config()).await { + Err(RuntimeError::Refused { reason: "task_queue_mismatch", .. }) => {} + answer => panic!("a queue the registration does not hold keeps the locked refusal: {answer:?}"), + } + } + + /// Every #1057 control on a fresh store of any backend: SQLite here, and + /// the MySQL/MariaDB and PostgreSQL qualification suites. + pub(in crate::runtime) async fn qualify_empty_query_poll(&mut self) { + // The poll-row sweep stays reserved for the test once warmed. + self.poll_sweep_config.interval = Duration::from_secs(3600); + self.empty_fixture().await; + self.qualify_lock_free_empty_answer().await; + self.qualify_bookkeeping_forces_lock().await; + self.qualify_request_rows_force_lock().await; + self.qualify_waiting_request_is_superseded().await; + self.qualify_registration_mismatch_keeps_its_refusal().await; + self.qualify_query_arrival_race().await; + // A due poll-row sweep is left to the locked path, which takes it. + self.poll_sweeps = Arc::default(); + self.assert_forced_lock("sweep-due", "poll-row sweep due").await; + self.assert_lock_free_empty("after-sweep", "sweep reserved again").await; + } + + /// The race the issue requires, in both commit orders, ordered by + /// barriers rather than sleeps. + async fn qualify_query_arrival_race(&self) { + // Order 1: the query commits after the snapshot read its first row. + // The snapshot cannot see it, so "empty" is an answer from before it + // existed, and nothing written hides it from the next poll. + let (proof, fence, keys) = self.proof_inputs("race-before").await; + let (taken_tx, taken_rx) = oneshot::channel::<()>(); + let (go_tx, go_rx) = oneshot::channel::<()>(); + let hook = async move { + taken_tx.send(()).unwrap(); + go_rx.await.unwrap(); + }; + let (race_body, race_config) = (body("race-before"), config()); + let peek = self.peek_empty_poll(&race_body, &race_config, &proof, fence.as_ref(), &keys, hook); + let racer = async move { + taken_rx.await.unwrap(); + let query = self.submit_query(); + let release = async move { + self.wait_for_pending_query().await; + go_tx.send(()).unwrap(); + }; + tokio::join!(query, release).0 + }; + let worker = async move { + let proven = peek.await.unwrap(); + assert!(proven, "the snapshot predates the query, so empty is proven"); + assert!(self.pending_query_visible().await, "the empty answer lost nothing"); + let (claimed, locks) = self.locks_for("race-next").await; + let claimed = claimed.unwrap(); + assert!(claimed["task"].is_object(), "the next poll leases the query: {claimed}"); + assert!(locks >= 1, "a visible query is claimed under the lock"); + self.answer(&claimed).await; + }; + let (answer, ()) = tokio::join!(racer, worker); + let (status, answer) = answer.unwrap(); + assert_eq!((status, answer["result"].clone()), (axum::http::StatusCode::OK, json!(7))); + + // Order 2: the query commits before the snapshot. The open set shows + // it, so the poll takes the lock and leases it at once. + let query = self.submit_query(); + let worker = async { + self.wait_for_pending_query().await; + assert!(!self.peek("race-after", None).await, "a visible query is never answered empty lock-free"); + let (claimed, locks) = self.locks_for("race-after").await; + let claimed = claimed.unwrap(); + assert!(claimed["task"].is_object(), "{claimed}"); + assert!(locks >= 1, "a visible query is claimed under the lock"); + self.answer(&claimed).await; + }; + let (answer, ()) = tokio::join!(query, worker); + assert_eq!(answer.unwrap().1["result"], 7); + } +} + +#[tokio::test] +async fn sqlite_empty_query_poll_takes_no_lock_only_when_proven() { + let dir = tempfile::tempdir().unwrap(); + let mut store = Store::::new(crate::runtime::sqlite::open(dir.path().join("empty.sqlite").to_str().unwrap()).await.unwrap()); + store.qualify_empty_query_poll().await; + store.close().await; +} diff --git a/rust/src/runtime/query_poll_requests.rs b/rust/src/runtime/query_poll_requests.rs index bc82c785b..e79d393b6 100644 --- a/rust/src/runtime/query_poll_requests.rs +++ b/rust/src/runtime/query_poll_requests.rs @@ -16,6 +16,28 @@ pub(super) struct PollKeys { pending: String, result: String, current: String, + /// The namespace, queue, build ID and worker these keys cover. + scope: Value, +} + +impl PollKeys { + // MySQL and MariaDB keys are varchar(64): keep 128 bits of the digest. + fn key(scope: &Value, kind: &str, id: Option<&str>) -> String { + let material = json!([kind, scope[0], scope[1], scope[2], scope[3], id]); + let digest: String = Sha256::digest(material.to_string().as_bytes())[..16] + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + format!("{POLL_PREFIX}{digest}") + } +} + +/// A poll row as a lock-free reader sees it. Only a live row it can parse +/// counts as present; an expired or unreadable one proves nothing. +enum Peek { + Absent, + Live(Value), + Unproven, } fn pending_ttl(config: &QueryTaskConfig) -> i64 { @@ -53,20 +75,13 @@ where .map(|row| DB::optional_string(&row, "build_id")) .transpose()? .flatten(); - // MySQL and MariaDB keys are varchar(64): keep 128 bits of the digest. - let key = |kind: &str, id: Option<&str>| { - let material = json!([kind, self.namespace.as_ref(), queue, build, owner, id]); - let digest: String = Sha256::digest(material.to_string().as_bytes())[..16] - .iter() - .map(|byte| format!("{byte:02x}")) - .collect(); - format!("{POLL_PREFIX}{digest}") - }; + let scope = json!([self.namespace.as_ref(), queue, build, owner]); Ok(PollKeys { request_id: request_id.to_owned(), - pending: key("pending", Some(request_id)), - result: key("result", Some(request_id)), - current: key("current", None), + pending: PollKeys::key(&scope, "pending", Some(request_id)), + result: PollKeys::key(&scope, "result", Some(request_id)), + current: PollKeys::key(&scope, "current", None), + scope, }) } @@ -99,6 +114,23 @@ where Ok(()) } + /// Writes one poll row of this worker's scope exactly as given, for + /// controls that need a row the maintained paths would not write. + #[cfg(test)] + pub(super) async fn put_test_poll_row(&self, keys: &PollKeys, kind: &str, request: Option<&str>, raw: &str, ttl: i64) { + let mut tx = self.begin().await.unwrap(); + Self::query(DB::QUERY_CACHE_PUT).bind(PollKeys::key(&keys.scope, kind, request)).bind(raw.to_owned()) + .bind(now() + ttl).execute(&mut *tx).await.unwrap(); + tx.commit().await.unwrap(); + } + + #[cfg(test)] + pub(super) async fn test_poll_row(&self, keys: &PollKeys, kind: &str, request: Option<&str>) -> Option { + let row = Self::query(DB::QUERY_CACHE_GET).bind(PollKeys::key(&keys.scope, kind, request)) + .fetch_optional(&self.pool).await.unwrap()?; + Some(serde_json::from_str(&DB::string(&row, "value").unwrap()).unwrap()) + } + #[cfg(test)] pub(in crate::runtime) async fn qualify_query_poll_cache_writes(&self) { let mut tx = self.begin().await.unwrap(); @@ -365,6 +397,126 @@ where Ok(Some(response)) } + /// A poll row as `peek_empty_poll` reads it: never deleted, and only a + /// live row that parses is present. Storage faults are returned. + async fn peek_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(Peek::Absent); + }; + if DB::number(&row, "expiration")? <= now() { + return Ok(Peek::Unproven); + } + Ok(serde_json::from_str(&DB::string(&row, "value")?).map_or(Peek::Unproven, Peek::Live)) + } + + /// Server #1057: whether a new non-waiting poll would claim nothing and + /// write nothing a later answer depends on, proven from one read + /// snapshot without the transition lock. The locked path would make this + /// request current, find no task and remember "empty". Without those rows + /// a retry of this request polls again, as a poll a moment later would. + /// + /// Every fact holds at the snapshot's one instant: + /// - nothing is pending or remembered under this request ID; + /// - the worker's current request, if any, is another request that has + /// already answered, so no waiting request of this worker is one this + /// request would end by becoming current; + /// - the claim would accept the registration: the same live proof and + /// poll fence, on this queue; + /// - the namespace has no open query task and no query bookkeeping due + /// (`query_bookkeeping_idle`). + /// + /// A poll-row sweep due before or after the reads is left to the locked + /// path, which reserves and runs it. So is every fact this cannot prove + /// exactly; the locked path then keeps its own answers and refusals. + /// `snapshot_taken` runs once the snapshot has read its first row; only + /// tests use it, to order a racing commit against the snapshot. + pub(super) async fn peek_empty_poll( + &self, + body: &Value, + config: &QueryTaskConfig, + proof: &str, + fence: Option<&crate::runtime::worker_poll_fence::WorkerPollFence>, + keys: &PollKeys, + snapshot_taken: impl std::future::Future, + ) -> Result { + use crate::runtime::poll_sweeps::Kind; + let interval = self.poll_sweep_config.interval; + if self.poll_sweeps.is_due(Kind::Query, interval) { + return Ok(false); + } + // One fixed snapshot whatever the server's configured isolation. + let mut read = self.begin_repeatable_read().await?; + let proven = self.peek_empty_poll_in(&mut read, body, config, proof, fence, keys, snapshot_taken).await?; + read.commit().await?; + Ok(proven && !self.poll_sweeps.is_due(Kind::Query, interval)) + } + + #[allow(clippy::too_many_arguments)] + async fn peek_empty_poll_in( + &self, + tx: &mut Transaction<'_, DB>, + body: &Value, + config: &QueryTaskConfig, + proof: &str, + fence: Option<&crate::runtime::worker_poll_fence::WorkerPollFence>, + keys: &PollKeys, + snapshot_taken: impl std::future::Future, + ) -> Result { + let owner = text(body, "worker_id")?; + let queue = text(body, "task_queue")?; + let pending = Self::peek_poll_row(tx, &keys.pending).await?; + snapshot_taken.await; + if !matches!(pending, Peek::Absent) || !matches!(Self::peek_poll_row(tx, &keys.result).await?, Peek::Absent) { + return Ok(false); + } + match Self::peek_poll_row(tx, &keys.current).await? { + Peek::Absent => {} + Peek::Unproven => return Ok(false), + Peek::Live(current) => { + // Exactly the row `mark_current_if_fresh` and the immediate + // path write, naming another request. + let Some(other) = current + .as_object() + .filter(|fields| fields.len() == 1) + .and_then(|fields| fields.get("poll_request_id")) + .and_then(Value::as_str) + .filter(|other| *other != keys.request_id) + else { + return Ok(false); + }; + // A waiting request has a pending row. Between becoming + // current and starting to wait it has neither row, so only + // a remembered answer shows it has finished. + if !matches!(Self::peek_poll_row(tx, &PollKeys::key(&keys.scope, "pending", Some(other))).await?, Peek::Absent) + || !matches!(Self::peek_poll_row(tx, &PollKeys::key(&keys.scope, "result", Some(other))).await?, Peek::Live(_)) + { + return Ok(false); + } + } + } + let workers = Self::query("SELECT task_queue FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2 AND status='active'") + .bind(self.namespace.as_ref()) + .bind(owner) + .fetch_all(&mut **tx) + .await?; + if workers.len() != 1 || DB::string(&workers[0], "task_queue")? != queue { + return Ok(false); + } + if Self::query_worker_proof(&mut **tx, self.namespace.as_ref(), owner, config).await?.as_deref() != Some(proof) { + return Ok(false); + } + if let Some(fence) = fence + && !self.worker_poll_fence_holds_in(tx, fence, owner).await? + { + return Ok(false); + } + Self::query_bookkeeping_idle(tx, self.namespace.as_ref()).await + } + pub(super) async fn coordinated_query_poll( &self, body: &Value, @@ -376,6 +528,9 @@ where keys: &PollKeys, ) -> Result { if timeout == 0 { + if self.peek_empty_poll(body, config, proof, fence, keys, std::future::ready(())).await? { + return Ok(poll_result(None, "empty")); + } if let Some(response) = self.immediate_query_poll(body, config, proof, fence, keys).await? { return Ok(response); } diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 8c99338f5..382e5addd 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -437,6 +437,16 @@ where None, )) } + /// A read-only snapshot that is fixed even on a MySQL server configured + /// below REPEATABLE READ, for a lock-free answer that is only correct if + /// every read shares one committed state (Server #1057). + pub(super) async fn begin_repeatable_read(&self) -> Result> { + Ok(Transaction::new( + self.pool.begin_with(DB::REPEATABLE_READ_SNAPSHOT_BEGIN).await?, + self.history_budget, + None, + )) + } pub(crate) async fn start( &self, body: Value,