From 338a585e563c5c57212721787cbed78d93001c47 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 8 Nov 2026 02:31:50 +0000 Subject: [PATCH 1/3] Answer a proven-empty non-waiting query poll without the transition lock Server #1057. A new non-waiting query poll is answered "empty" from one read-only snapshot, writing nothing, only while every fact of that answer holds at the snapshot's one instant: - nothing is pending or remembered under the request ID; - the worker's current request, if any, is another request that has already answered, so no waiting request would be ended by this one; - the claim would accept the registration (proof, fence, queue); - the namespace's query bookkeeping is idle: the 3.0.1 cutover is done, the receipt sweep has no minute to retire, and the 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 one cannot see the other. A due poll-row sweep, checked before and after the reads without being reserved, leaves the poll to the locked path. So does any row the reader cannot recognize exactly (malformed, expired, foreign or ambiguous); the locked path keeps its own answers, refusals and storage faults. A retry of an unremembered empty answer polls again, as a later poll would. Controls: zero transition locks and no cache writes when proven; forced lock for every unproven fact; a newer request still ends an older waiting one; the registration-queue refusal is unchanged; and a barrier-ordered race in both commit orders shows a query committed after the snapshot is never lost and one committed before it is never answered empty. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/poll_sweeps.rs | 35 ++- rust/src/runtime/queries.rs | 40 +++ rust/src/runtime/query_empty_poll_tests.rs | 343 +++++++++++++++++++++ rust/src/runtime/query_poll_requests.rs | 178 ++++++++++- 4 files changed, 582 insertions(+), 14 deletions(-) create mode 100644 rust/src/runtime/query_empty_poll_tests.rs 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/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..7fb4c404d --- /dev/null +++ b/rust/src/runtime/query_empty_poll_tests.rs @@ -0,0 +1,343 @@ +//! 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. +use super::{CUTOVER_MARK, EXPIRY_MARK, read_arguments}; +use crate::codec::{Value as Payload, ValueCodec}; +use crate::runtime::{QueryTaskConfig, RuntimeError, store::{Store, now}, 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())); + } + match rows.last() { + Some((key, ..)) if page.len() == 64 => after.clone_from(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:?}"), + } + } + + /// 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()); + // The poll-row sweep stays reserved for the test once warmed. + store.poll_sweep_config.interval = Duration::from_secs(3600); + store.empty_fixture().await; + store.qualify_lock_free_empty_answer().await; + store.qualify_bookkeeping_forces_lock().await; + store.qualify_request_rows_force_lock().await; + store.qualify_waiting_request_is_superseded().await; + store.qualify_registration_mismatch_keeps_its_refusal().await; + store.qualify_query_arrival_race().await; + // A due poll-row sweep is left to the locked path, which takes it. + store.poll_sweeps = Arc::default(); + store.assert_forced_lock("sweep-due", "poll-row sweep due").await; + store.assert_lock_free_empty("after-sweep", "sweep reserved again").await; + store.close().await; +} diff --git a/rust/src/runtime/query_poll_requests.rs b/rust/src/runtime/query_poll_requests.rs index bc82c785b..7eb7ca48d 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,125 @@ 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); + } + let mut read = self.begin_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 +527,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); } From 86ae87f1e2622c4009d01e71c5ab6a37cfc8e5ba Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 8 Nov 2026 02:55:34 +0000 Subject: [PATCH 2/3] Use the query cache clock in the empty-poll controls The controls wrote the receipt sweep's minute mark from the store's DateTime clock, which has no div_euclid; the bookkeeping rows use the query cache's epoch-second clock. Also collect the cache rows without returning the vector while a borrow of it is matched. Test-only. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/query_empty_poll_tests.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/rust/src/runtime/query_empty_poll_tests.rs b/rust/src/runtime/query_empty_poll_tests.rs index 7fb4c404d..e5116c8ee 100644 --- a/rust/src/runtime/query_empty_poll_tests.rs +++ b/rust/src/runtime/query_empty_poll_tests.rs @@ -1,9 +1,10 @@ //! 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. -use super::{CUTOVER_MARK, EXPIRY_MARK, read_arguments}; +// `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, now}, worker_poll_fence::PollAdmission}; +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}; @@ -81,8 +82,9 @@ store_methods! { for row in &page { rows.push((DB::string(row, "key").unwrap(), DB::string(row, "value").unwrap(), DB::number(row, "expiration").unwrap())); } - match rows.last() { - Some((key, ..)) if page.len() == 64 => after.clone_from(key), + let last = rows.last().map(|(key, ..)| key.clone()); + match last { + Some(key) if page.len() == 64 => after = key, _ => return rows, } } From 77c2f3b4c3abd70afdde24a2b8eb457123ec7420 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 8 Nov 2026 03:47:54 +0000 Subject: [PATCH 3/3] Fix one snapshot for the lock-free empty query poll on MySQL Server #1057 review: the empty answer is right only if every read shares one committed snapshot. MySQL's ordinary snapshot begin keeps the session's isolation, and WITH CONSISTENT SNAPSHOT is ignored below REPEATABLE READ, so a server configured for READ COMMITTED would not supply one. The peek now begins with REPEATABLE_READ_SNAPSHOT_BEGIN: on MySQL and MariaDB a SET TRANSACTION (next transaction only, never the session or server) followed by the consistent-snapshot start; SQLite and PostgreSQL keep their ordinary snapshot, which already is one. Every other read keeps begin_read unchanged. Controls: - MySQL/MariaDB, from a READ COMMITTED session: the ordinary begin sees a later commit (the missing premise); the new begin hides it and stays read-only; the session is still READ COMMITTED afterwards. - The whole empty-poll suite (zero-lock proof, forced-lock controls, superseded waiting request, kept refusal, two-order barrier race) now runs on SQLite, MySQL/MariaDB and PostgreSQL through one entry point. - A unit check pins each backend's begin statement. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 24 +++++++++ rust/src/runtime/mysql.rs | 58 ++++++++++++++++++++++ rust/src/runtime/postgres.rs | 10 ++++ rust/src/runtime/query_empty_poll_tests.rs | 32 +++++++----- rust/src/runtime/query_poll_requests.rs | 3 +- rust/src/runtime/store.rs | 10 ++++ 6 files changed, 123 insertions(+), 14 deletions(-) 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/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/query_empty_poll_tests.rs b/rust/src/runtime/query_empty_poll_tests.rs index e5116c8ee..b730f1184 100644 --- a/rust/src/runtime/query_empty_poll_tests.rs +++ b/rust/src/runtime/query_empty_poll_tests.rs @@ -269,6 +269,24 @@ store_methods! { } } + /// 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) { @@ -328,18 +346,6 @@ store_methods! { 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()); - // The poll-row sweep stays reserved for the test once warmed. - store.poll_sweep_config.interval = Duration::from_secs(3600); - store.empty_fixture().await; - store.qualify_lock_free_empty_answer().await; - store.qualify_bookkeeping_forces_lock().await; - store.qualify_request_rows_force_lock().await; - store.qualify_waiting_request_is_superseded().await; - store.qualify_registration_mismatch_keeps_its_refusal().await; - store.qualify_query_arrival_race().await; - // A due poll-row sweep is left to the locked path, which takes it. - store.poll_sweeps = Arc::default(); - store.assert_forced_lock("sweep-due", "poll-row sweep due").await; - store.assert_lock_free_empty("after-sweep", "sweep reserved again").await; + 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 7eb7ca48d..e79d393b6 100644 --- a/rust/src/runtime/query_poll_requests.rs +++ b/rust/src/runtime/query_poll_requests.rs @@ -448,7 +448,8 @@ where if self.poll_sweeps.is_due(Kind::Query, interval) { return Ok(false); } - let mut read = self.begin_read().await?; + // 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)) 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,