diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 412103944..02735d708 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -51,7 +51,6 @@ pub(super) trait Backend: Database { const WORKER_REGISTRATION_SQL: &str = super::store::WORKER_REGISTRATION_SQL; const TASK_CANDIDATES_SQL: &str = super::store::TASK_CANDIDATES_SQL; const POLL_WAKE_TASKS_SQL: &str = super::cross_kind_poll::TASKS_SQL; - const POLL_WAKE_QUERIES_SQL: &str = "SELECT \"key\" AS id FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND expiration>$1 AND CAST(value AS jsonb)->>'namespace'=$2 AND CAST(value AS jsonb)->>'task_queue'=$3 AND CAST(value AS jsonb)->>'query_deadline_at'>$4 AND (CAST(value AS jsonb)->>'status'='pending' OR (CAST(value AS jsonb)->>'status'='leased' AND CAST(value AS jsonb)->>'lease_expires_at'<=$5)) ORDER BY \"key\" LIMIT 65"; const POLL_RECEIPT_CLEANUP_SQL: &str = super::store::POLL_RECEIPT_CLEANUP_SQL; const TIMER_TASK_SQL: &str; const TIMER_SCHEDULE_SQL: &str = "SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='TimerScheduled' AND payload->>'timer_id'=$2 ORDER BY sequence DESC LIMIT 1"; @@ -80,6 +79,10 @@ pub(super) trait Backend: Database { const QUERY_CACHE_UPDATE: &str = "UPDATE dw_query_cache SET value=$1,expiration=$2 WHERE \"key\"=$3"; const QUERY_CACHE_DELETE: &str = "DELETE FROM dw_query_cache WHERE \"key\"=$1"; + const QUERY_CACHE_DELETE_EXPIRED: &str = "DELETE FROM dw_query_cache WHERE \"key\"=$1 AND expiration<=$2"; + // One primary-key page of the whole cache table, for the one-time + // cutover of 3.0.1 query rows (Server #964). + const QUERY_CACHE_PAGE: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\">$1 ORDER BY \"key\" LIMIT 64"; // A bounded batch of expired per-request query poll rows. const QUERY_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 32"; const WORKFLOW_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 32"; @@ -235,7 +238,6 @@ SELECT id,ordinal FROM ordinals WHERE position IS NOT ordinal ORDER BY ordinal L const ACTIVITY_RESULT_EVENT_SQL: &str = "SELECT sequence,event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') AND (json_extract(payload,'$.activity_execution_id') COLLATE BINARY=$2 OR json_extract(payload,'$.activity.id') COLLATE BINARY=$3) ORDER BY sequence DESC LIMIT 1"; const QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND json_extract(value,'$.namespace')=$1 ORDER BY \"key\" LIMIT 65"; const SESSION_RUNNING_ATTEMPTS: &str = "SELECT COUNT(*) FROM activity_attempts p JOIN activity_executions e ON e.id=p.activity_execution_id JOIN workflow_tasks t ON t.id=p.workflow_task_id WHERE t.namespace=$1 AND p.status='running' AND p.lease_expires_at>$2 AND p.lease_owner=$3 AND json_extract(e.activity_options,'$.worker_session.session_id')=$4 AND p.workflow_task_id<>$5"; - const POLL_WAKE_QUERIES_SQL: &str = "SELECT \"key\" AS id FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND expiration>$1 AND json_extract(value,'$.namespace')=$2 AND json_extract(value,'$.task_queue')=$3 AND json_extract(value,'$.query_deadline_at')>$4 AND (json_extract(value,'$.status')='pending' OR (json_extract(value,'$.status')='leased' AND json_extract(value,'$.lease_expires_at')<=$5)) ORDER BY \"key\" LIMIT 65"; const POLL_WAKE_TASKS_SQL: &str = r#" WITH utc_tasks AS ( SELECT t.id,t.namespace,t.queue,t.workflow_run_id,t.status,t.available_at, @@ -551,7 +553,6 @@ impl Backend for MySql { } const ACTIVITY_RESULT_EVENT_SQL: &str = "SELECT sequence,event_type,payload FROM workflow_history_events WHERE CAST(workflow_run_id AS BINARY)=CAST(? AS BINARY) AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') AND (CAST(JSON_UNQUOTE(JSON_EXTRACT(payload,'$.activity_execution_id')) AS BINARY)=CAST(? AS BINARY) OR CAST(JSON_UNQUOTE(JSON_EXTRACT(payload,'$.activity.id')) AS BINARY)=CAST(? AS BINARY)) ORDER BY sequence DESC LIMIT 1"; const QUERY_CACHE_SCAN: &str = "SELECT `key`,value,expiration FROM dw_query_cache WHERE `key` LIKE 'dw-native-query:%' AND BINARY JSON_UNQUOTE(JSON_EXTRACT(value,'$.namespace'))=BINARY ? ORDER BY `key` LIMIT 65"; - const POLL_WAKE_QUERIES_SQL: &str = "SELECT `key` AS id FROM dw_query_cache WHERE `key` LIKE 'dw-native-query:%' AND expiration>? AND CAST(JSON_UNQUOTE(JSON_EXTRACT(value,'$.namespace')) AS BINARY)=CAST(? AS BINARY) AND CAST(JSON_UNQUOTE(JSON_EXTRACT(value,'$.task_queue')) AS BINARY)=CAST(? AS BINARY) AND CAST(JSON_UNQUOTE(JSON_EXTRACT(value,'$.query_deadline_at')) AS BINARY)>CAST(? AS BINARY) AND (JSON_UNQUOTE(JSON_EXTRACT(value,'$.status'))='pending' OR (JSON_UNQUOTE(JSON_EXTRACT(value,'$.status'))='leased' AND CAST(JSON_UNQUOTE(JSON_EXTRACT(value,'$.lease_expires_at')) AS BINARY)<=CAST(? AS BINARY))) ORDER BY `key` LIMIT 65"; const EXTERNAL_PAYLOAD_BYTES: &str = "SELECT CAST(COALESCE(SUM(size_bytes),0) AS SIGNED) FROM runtime_external_payloads WHERE namespace=$1"; const MEMO_UPSERT: &str = "INSERT INTO workflow_memos(workflow_run_id,workflow_instance_id,`key`,value,upserted_at_sequence,inherited_from_parent,created_at,updated_at,portable_value,portable_value_sequence) VALUES ($1,$2,$3,$4,$5,false,$6,$7,$8,$9) ON DUPLICATE KEY UPDATE value=VALUES(value),upserted_at_sequence=VALUES(upserted_at_sequence),inherited_from_parent=false,updated_at=VALUES(updated_at),portable_value=VALUES(portable_value),portable_value_sequence=VALUES(portable_value_sequence)"; const MEMO_DELETE: &str = @@ -576,6 +577,8 @@ impl Backend for MySql { const QUERY_CACHE_UPDATE: &str = "UPDATE dw_query_cache SET value=$1,expiration=$2 WHERE `key`=$3"; const QUERY_CACHE_DELETE: &str = "DELETE FROM dw_query_cache WHERE `key`=$1"; + const QUERY_CACHE_DELETE_EXPIRED: &str = "DELETE FROM dw_query_cache WHERE `key`=$1 AND expiration<=$2"; + const QUERY_CACHE_PAGE: &str = "SELECT `key`,value,expiration FROM dw_query_cache WHERE `key`>$1 ORDER BY `key` LIMIT 64"; const QUERY_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 32"; const WORKFLOW_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 32"; const QUERY_CACHE_LOCK: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3) ON DUPLICATE KEY UPDATE expiration=expiration"; @@ -611,7 +614,6 @@ impl Backend for MySql { || sql == Self::LOCAL_ACTIVITY_ATTEMPTS || sql == Self::ACTIVITY_RESULT_EVENT_SQL || sql == Self::OBSERVATION_FAILURE_EVENT - || sql == Self::POLL_WAKE_QUERIES_SQL || sql == Self::QUERY_CACHE_SCAN || sql == Self::SESSION_RUNNING_ATTEMPTS { diff --git a/rust/src/runtime/cross_kind_poll.rs b/rust/src/runtime/cross_kind_poll.rs index 3396c3e92..2db964378 100644 --- a/rust/src/runtime/cross_kind_poll.rs +++ b/rust/src/runtime/cross_kind_poll.rs @@ -71,19 +71,12 @@ where if kind != "query" { // Query timing uses the immutable UTC6 authority fields. This // readiness hint still grants no claim or execution authority. - let query_now = super::query_authority::wire(at); + // Read by key from the namespace's open set (Server #964). ready.extend( - Self::query(DB::POLL_WAKE_QUERIES_SQL) - .bind(at.timestamp()) - .bind(self.namespace.as_ref()) - .bind(text(body, "task_queue")?) - .bind(query_now.clone()) - .bind(query_now) - .fetch_all(&self.pool) + self.ready_query_keys(text(body, "task_queue")?) .await? - .iter() - .map(|row| DB::string(row, "id").map(|id| format!("query:{id}"))) - .collect::>>()?, + .into_iter() + .map(|key| format!("query:{key}")), ); } Ok(Some(PollWakeSnapshot { diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index 63a333793..14bb94a37 100644 --- a/rust/src/runtime/queries.rs +++ b/rust/src/runtime/queries.rs @@ -19,6 +19,18 @@ use tokio::sync::Semaphore; const PREFIX: &str = "dw-native-query:"; const MAX_PENDING: usize = 16; pub(super) const MAX_ENTRIES: usize = 64; +/// One row per namespace listing its open query task ids (Server #964). +const OPEN_PREFIX: &str = "dw-native-query-open:"; +/// Receipts filed by the minute their rows expire, and the sweep's watermark. +const EXPIRY_PREFIX: &str = "dw-native-query-expiry:"; +const EXPIRY_MARK: &str = "dw-native-query-expiry-mark"; +/// The one-time cutover of 3.0.1 query rows. +const CUTOVER_MARK: &str = "dw-native-query-cutover"; +const PAGE_KEYS: usize = 128; +const SWEEP_MINUTES: usize = 30; +const SWEEP_PAGES: usize = 2; +/// Bookkeeping rows never expire on their own. +const BOOKKEEPING_EXPIRATION: i64 = i64::MAX / 2; const SNAPSHOT_BYTES: usize = 1024 * 1024; pub(super) const ARGUMENT_BYTES: usize = 64 * 1024; pub(super) const RESULT_BYTES: usize = 256 * 1024; @@ -68,6 +80,26 @@ fn poll_result(task: Option, status: &str) -> Value { fn now() -> i64 { chrono::Utc::now().timestamp() } +/// An open set's ids: unique canonical ULIDs, at most `MAX_ENTRIES`, and +/// none from a row that names another namespace. +fn listed_ids(set: &Value, namespace: &str) -> Vec { + if set["namespace"] != namespace { + return vec![]; + } + let mut ids: Vec = vec![]; + for id in set["ids"].as_array().into_iter().flatten().filter_map(Value::as_str) { + if ids.len() < MAX_ENTRIES && canonical_id(id) && !ids.iter().any(|seen| seen == id) { + ids.push(id.to_owned()); + } + } + ids +} + +/// A query task id as native writes it: a ULID in its canonical text. +fn canonical_id(id: &str) -> bool { + id.parse::().is_ok_and(|parsed| parsed.to_string() == id) +} + fn key(id: &str) -> Result { // A task ID cannot address another application's cache key. if id.parse::().is_err() { @@ -302,39 +334,280 @@ where for<'r> i64: Decode<'r, DB>, usize: ColumnIndex, { + /// The namespace's open query tasks (pending or leased), at most + /// `MAX_ENTRIES`, read by key from the namespace's open set. An answered, + /// failed or timed-out task keeps its receipt, readable by key, until it + /// expires, but holds no queue capacity, as in PHP, whose admission counts + /// only pending task ids (Server #964). Nothing here scans the cache + /// table: every read is by primary key, with a fixed bound per call. pub(super) async fn query_entries( &self, tx: &mut Transaction<'_, DB>, ) -> Result> { - let rows = Self::query(DB::QUERY_CACHE_SCAN) - .bind(self.namespace.to_string()) - .fetch_all(&mut **tx) - .await?; + Self::cut_over_query_rows(tx).await?; + Self::sweep_query_receipts(tx).await?; + let namespace = self.namespace.as_ref(); + let listed = Self::open_set(tx, namespace).await?; let mut entries = vec![]; - for row in rows { - let key = DB::string(&row, "key")?; + let mut open = vec![]; + for id in &listed { + let key = key(id)?; + let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(&key).fetch_optional(&mut **tx).await? else { + continue; + }; if DB::number(&row, "expiration")? <= now() { - Self::query(DB::QUERY_CACHE_DELETE) - .bind(&key) - .execute(&mut **tx) - .await?; + Self::query(DB::QUERY_CACHE_DELETE).bind(&key).execute(&mut **tx).await?; continue; } let mut task: Value = serde_json::from_str(&DB::string(&row, "value")?)?; + // A member must be this namespace's task under its own id. + if task["query_task_id"] != id.as_str() || task["namespace"] != namespace { + continue; + } if authority::expire(&mut task, authority::now())? { Self::put_query(tx, &key, &task, authority::expiration(&task)?).await?; } - entries.push((key, task)); + if matches!(task["status"].as_str(), Some("pending" | "leased")) { + open.push(id.clone()); + entries.push((key, task)); + } } - if entries.len() > MAX_ENTRIES { - return Err(refuse( - StatusCode::SERVICE_UNAVAILABLE, - "query_task_queue_full", - )); + if open != listed { + Self::write_open_set(tx, namespace, &open).await?; } Ok(entries) } + /// The open-set row's key: one per namespace, from a digest so any + /// namespace name fits the cache key column. The row also names its + /// namespace in full. + fn open_set_key(namespace: &str) -> String { + use sha2::{Digest, Sha256}; + let digest: String = Sha256::digest(namespace.as_bytes()).iter().take(16).map(|byte| format!("{byte:02x}")).collect(); + format!("{OPEN_PREFIX}{digest}") + } + + /// 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> { + let Some(set) = Self::cache_document(tx, &Self::open_set_key(namespace)).await? else { + return Ok(vec![]); + }; + Ok(listed_ids(&set, namespace)) + } + + /// Open tasks of this namespace on `queue` a poll could claim now: read + /// lock-free from the namespace's open set, every row by key (Server + /// #964). Only a readiness hint; the claim decides under the lock. + pub(super) async fn ready_query_keys(&self, queue: &str) -> Result> { + let namespace = self.namespace.as_ref(); + let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(Self::open_set_key(namespace)).fetch_optional(&self.pool).await? else { + return Ok(vec![]); + }; + let set: Value = serde_json::from_str(&DB::string(&row, "value")?).unwrap_or(Value::Null); + let at = super::store::now(); + let wire_now = authority::wire(at); + let mut ready = vec![]; + for id in listed_ids(&set, namespace) { + let key = key(&id)?; + let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(&key).fetch_optional(&self.pool).await? else { + continue; + }; + if DB::number(&row, "expiration")? <= at.timestamp() { + continue; + } + let Ok(task) = serde_json::from_str::(&DB::string(&row, "value")?) else { + continue; + }; + let after = |field: &str| task[field].as_str().is_some_and(|time| time > wire_now.as_str()); + let claimable = match task["status"].as_str() { + Some("pending") => true, + Some("leased") => !after("lease_expires_at"), + _ => false, + }; + if task["query_task_id"] == id.as_str() && task["namespace"] == namespace && task["task_queue"] == queue + && after("query_deadline_at") && claimable + { + ready.push(key); + } + } + Ok(ready) + } + + async fn write_open_set(tx: &mut Transaction<'_, DB>, namespace: &str, ids: &[String]) -> Result<()> { + Self::put_cache_document(tx, &Self::open_set_key(namespace), &json!({"namespace": namespace, "ids": ids})).await + } + + /// Adds an open task to its namespace's set, at most `MAX_ENTRIES`. + async fn list_open(tx: &mut Transaction<'_, DB>, namespace: &str, id: &str) -> Result<()> { + let mut ids = Self::open_set(tx, namespace).await?; + if !ids.iter().any(|listed| listed == id) { + if ids.len() >= MAX_ENTRIES { + return Err(refuse(StatusCode::SERVICE_UNAVAILABLE, "query_task_queue_full")); + } + ids.push(id.to_owned()); + } + Self::write_open_set(tx, namespace, &ids).await + } + + /// A native bookkeeping row's document, by key. + async fn cache_document(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(serde_json::from_str(&DB::string(&row, "value")?).ok()) + } + + async fn put_cache_document(tx: &mut Transaction<'_, DB>, key: &str, value: &Value) -> Result<()> { + // Bookkeeping rows never expire on their own; the sweep and the open + // set maintain them. + let expiration = BOOKKEEPING_EXPIRATION; + let updated = Self::query(DB::QUERY_CACHE_UPDATE) + .bind(value.to_string()).bind(expiration).bind(key) + .execute(&mut **tx).await?; + if DB::affected(updated) == 0 { + Self::query(DB::QUERY_CACHE_INSERT) + .bind(key).bind(value.to_string()).bind(expiration) + .execute(&mut **tx).await?; + } + Ok(()) + } + + /// Files a query row under the minute it expires, for the sweep: a minute + /// keeps pages of at most `PAGE_KEYS` keys, named by number. + async fn file_expiry(tx: &mut Transaction<'_, DB>, key: &str, expiration: i64) -> Result<()> { + // A row already expired goes now: its minute may be behind the sweep. + if expiration <= now() { + Self::query(DB::QUERY_CACHE_DELETE_EXPIRED).bind(key).bind(now()).execute(&mut **tx).await?; + return Ok(()); + } + let minute = expiration.div_euclid(60); + let header_key = format!("{EXPIRY_PREFIX}{minute}"); + let header = Self::cache_document(tx, &header_key).await?.unwrap_or_else(|| json!({"first": 0, "pages": 0})); + let first = header["first"].as_i64().unwrap_or(0).max(0); + let mut pages = header["pages"].as_i64().unwrap_or(0).max(first); + let mut page = if pages > first { + Self::cache_document(tx, &format!("{header_key}:{}", pages - 1)).await? + } else { + None + }; + if page.as_ref().is_none_or(|page| page["keys"].as_array().is_none_or(|keys| keys.len() >= PAGE_KEYS)) { + pages += 1; + page = Some(json!({"keys": []})); + } + let mut page = page.unwrap_or_else(|| json!({"keys": []})); + if let Some(keys) = page["keys"].as_array_mut() { + keys.push(json!(key)); + } + Self::put_cache_document(tx, &format!("{header_key}:{}", pages - 1), &page).await?; + Self::put_cache_document(tx, &header_key, &json!({"first": first, "pages": pages})).await + } + + /// Deletes expired query rows, minute by minute behind a watermark: at + /// most `SWEEP_MINUTES` minutes, `SWEEP_PAGES` pages and their keys per + /// call, every one by key. A key that has not expired (its expiry moved) + /// is filed again under its own minute. Everything happens in the + /// caller's transaction, so a crash keeps or loses the whole step. + async fn sweep_query_receipts(tx: &mut Transaction<'_, DB>) -> Result<()> { + let current = now().div_euclid(60); + let Some(mark) = Self::cache_document(tx, EXPIRY_MARK).await?.and_then(|mark| mark["minute"].as_i64()) else { + // The first sweep starts here; earlier minutes hold nothing, as + // filing deletes a row that has already expired. + return Self::put_cache_document(tx, EXPIRY_MARK, &json!({"minute": current - 1})).await; + }; + let mut minute = mark; + let mut pages_left = SWEEP_PAGES; + for _ in 0..SWEEP_MINUTES { + if minute + 1 >= current || pages_left == 0 { + break; + } + let header_key = format!("{EXPIRY_PREFIX}{}", minute + 1); + if let Some(header) = Self::cache_document(tx, &header_key).await? { + let pages = header["pages"].as_i64().unwrap_or(0).max(0); + let mut first = header["first"].as_i64().unwrap_or(0).clamp(0, pages); + while first < pages && pages_left > 0 { + let page_key = format!("{header_key}:{first}"); + let keys = Self::cache_document(tx, &page_key).await? + .and_then(|page| page["keys"].as_array().cloned()) + .unwrap_or_default(); + for key in keys.iter().filter_map(Value::as_str).take(PAGE_KEYS) { + let deleted = Self::query(DB::QUERY_CACHE_DELETE_EXPIRED).bind(key).bind(now()).execute(&mut **tx).await?; + if DB::affected(deleted) == 0 + && let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&mut **tx).await? + { + let expiration = DB::number(&row, "expiration")?; + if expiration.div_euclid(60) > minute + 1 { + Self::file_expiry(tx, key, expiration).await?; + } + } + } + Self::query(DB::QUERY_CACHE_DELETE).bind(&page_key).execute(&mut **tx).await?; + first += 1; + pages_left -= 1; + } + if first < pages { + Self::put_cache_document(tx, &header_key, &json!({"first": first, "pages": pages})).await?; + break; + } + Self::query(DB::QUERY_CACHE_DELETE).bind(&header_key).execute(&mut **tx).await?; + } + minute += 1; + } + if minute != mark { + Self::put_cache_document(tx, EXPIRY_MARK, &json!({"minute": minute})).await?; + } + Ok(()) + } + + /// 3.0.1 kept query rows without open sets or expiry files. Until the + /// cutover marker says it is done, each call reads one primary-key page + /// of the cache table (at most 64 rows, whatever their keys), lists those + /// rows' open tasks in their namespaces' sets and files every one for + /// expiry. It runs once for the whole server, never per namespace. + async fn cut_over_query_rows(tx: &mut Transaction<'_, DB>) -> Result<()> { + let marker = Self::cache_document(tx, CUTOVER_MARK).await?.unwrap_or_else(|| json!({"after": "", "done": false})); + if marker["done"] == true { + return Ok(()); + } + let after = marker["after"].as_str().unwrap_or_default().to_owned(); + let rows = Self::query(DB::QUERY_CACHE_PAGE).bind(&after).fetch_all(&mut **tx).await?; + let mut last = after; + for row in &rows { + let key = DB::string(row, "key")?; + last.clone_from(&key); + let Some(id) = key.strip_prefix("dw-native-query:") else { + continue; + }; + let expiration = DB::number(row, "expiration")?; + // An expired row is deleted here and never listed. + Self::file_expiry(tx, &key, expiration).await?; + if expiration <= now() { + continue; + } + let Ok(mut task) = serde_json::from_str::(&DB::string(row, "value")?) else { + continue; + }; + // A task past its deadline closes before it could take a slot. + if authority::expire(&mut task, authority::now())? { + Self::put_query(tx, &key, &task, authority::expiration(&task)?).await?; + } + if task["query_task_id"] == id + && canonical_id(id) + && matches!(task["status"].as_str(), Some("pending" | "leased")) + && let Some(namespace) = task["namespace"].as_str() + { + match Self::list_open(tx, namespace, id).await { + // A full set leaves this task to its deadline; nothing + // else may pass silently, so the page rolls back. + Err(super::RuntimeError::Refused { reason: "query_task_queue_full", .. }) => {} + other => other?, + } + } + } + let done = rows.len() < 64; + Self::put_cache_document(tx, CUTOVER_MARK, &json!({"after": last, "done": done})).await + } + async fn put_query( tx: &mut Transaction<'_, DB>, key: &str, @@ -515,6 +788,8 @@ where .bind(authority::expiration(&task)?) .execute(&mut *tx) .await?; + Self::list_open(&mut tx, self.namespace.as_ref(), &task_id).await?; + Self::file_expiry(&mut tx, &key, authority::expiration(&task)?).await?; tx.commit().await?; self.wake.notify_waiters(); // Never hold the transition lock or a pooled connection across the wait. @@ -814,16 +1089,7 @@ where { return Ok(true); } - let at = super::store::now(); - let query_now = authority::wire(at); - let candidates = Self::query(DB::POLL_WAKE_QUERIES_SQL) - .bind(at.timestamp()) - .bind(self.namespace.as_ref()) - .bind(text(body, "task_queue")?) - .bind(query_now.clone()) - .bind(query_now) - .fetch_all(&self.pool) - .await?; + let candidates = self.ready_query_keys(text(body, "task_queue")?).await?; if candidates.is_empty() { return Ok(false); } @@ -833,7 +1099,7 @@ where let mut ready = false; for candidate in candidates { let Some(row) = Self::query(DB::QUERY_CACHE_GET) - .bind(DB::string(&candidate, "id")?) + .bind(&candidate) .fetch_optional(&mut *read) .await? else { diff --git a/rust/tests/execution/query_arguments.rs b/rust/tests/execution/query_arguments.rs index b2479c2cf..2c55f02b6 100644 --- a/rust/tests/execution/query_arguments.rs +++ b/rust/tests/execution/query_arguments.rs @@ -1,7 +1,10 @@ use super::*; +/// The cache's query task rows, without native's bookkeeping rows (open +/// sets and expiry pages, which share the table). async fn query_entries(database: &TestDatabase) -> i64 { - let sql = "SELECT COUNT(*) FROM dw_query_cache"; + let sql = "SELECT COUNT(*) FROM dw_query_cache WHERE substr(\"key\",1,16)='dw-native-query:'"; + let mysql = "SELECT COUNT(*) FROM dw_query_cache WHERE LEFT(`key`,16)='dw-native-query:'"; match database { TestDatabase::Sqlite(dir) => { let pool = sqlx::SqlitePool::connect(&format!( @@ -28,7 +31,7 @@ async fn query_entries(database: &TestDatabase) -> i64 { .connect_with(options.as_ref().clone()) .await .unwrap(); - let count = sqlx::query_scalar(sql).fetch_one(&pool).await.unwrap(); + let count = sqlx::query_scalar(mysql).fetch_one(&pool).await.unwrap(); pool.close().await; count } diff --git a/rust/tests/execution/query_poll_requests.rs b/rust/tests/execution/query_poll_requests.rs index 1180755eb..2340ebdc3 100644 --- a/rust/tests/execution/query_poll_requests.rs +++ b/rust/tests/execution/query_poll_requests.rs @@ -202,3 +202,85 @@ async fn duplicate_polls_with_one_request_id_share_one_claim() { runtime.close().await; database.remove().await; } + +// An answered query keeps its receipt but holds no queue capacity (Server +// #964): more sequential queries than the cache's open-task bound (64) all +// succeed against one run, as against published PHP. +#[tokio::test] +async fn answered_queries_release_queue_capacity() { + let (database, runtime, app, path) = queryable("sequential-queries").await; + for index in 0..80 { + let caller = admit_query(&app, &path); + let (claimed, _) = poll_waiting(&app, &format!("sequential-{index}"), 5).await; + let task = claimed["task"].clone(); + assert!(task.is_object(), "query {index} was not claimed: {claimed}"); + finish_query(&app, &task).await; + let (status, answer) = caller.await.unwrap(); + assert_eq!(status, StatusCode::OK, "query {index}: {answer}"); + } + runtime.close().await; + database.remove().await; +} + +// Retained receipts hold no capacity and cost no scan (Server #964): with +// thousands of answered tasks already in the cache, as 3.0.1 left them, +// later queries are admitted, claimed and answered as usual, while the +// one-time cutover files the old rows a bounded page at a time. +#[tokio::test] +async fn thousands_of_retained_receipts_leave_queries_unaffected() { + let (database, runtime, app, path) = queryable("many-receipts").await; + let expires = chrono::Utc::now().timestamp() + 3600; + let rows: Vec = (0..3000) + .map(|_| { + let id = ulid::Ulid::new().to_string(); + format!("('dw-native-query:{id}','{{\"query_task_id\":\"{id}\",\"namespace\":\"default\",\"status\":\"completed\"}}',{expires})") + }) + .collect(); + for chunk in rows.chunks(500) { + // Column order is the cache table's: key, value, expiration. + database.execute(sqlx::AssertSqlSafe(format!("INSERT INTO dw_query_cache VALUES {}", chunk.join(",")))).await; + } + for index in 0..20 { + let caller = admit_query(&app, &path); + let (claimed, _) = poll_waiting(&app, &format!("receipts-{index}"), 5).await; + let task = claimed["task"].clone(); + assert!(task.is_object(), "query {index} was not claimed: {claimed}"); + finish_query(&app, &task).await; + let (status, answer) = caller.await.unwrap(); + assert_eq!(status, StatusCode::OK, "query {index}: {answer}"); + } + runtime.close().await; + database.remove().await; +} + +// The cutover never lists a 3.0.1 row that can no longer be answered (Server +// #964): an expired row is deleted, and a pending row past its query +// deadline closes first. Otherwise 128 such rows would fill the namespace's +// open set and refuse the live queries below as query_task_queue_full. +#[tokio::test] +async fn the_cutover_lists_only_answerable_rows() { + let (database, runtime, app, path) = queryable("stale-cutover").await; + let now = chrono::Utc::now(); + let wire = |time: chrono::DateTime| time.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string(); + let row = |expiration: i64| { + let id = ulid::Ulid::new().to_string(); + let task = format!( + "{{\"query_task_id\":\"{id}\",\"namespace\":\"default\",\"task_queue\":\"test\",\"status\":\"pending\",\"query_admitted_at\":\"{}\",\"query_deadline_at\":\"{}\",\"receipt_expires_at\":\"{}\"}}", + wire(now - chrono::Duration::seconds(600)), wire(now - chrono::Duration::seconds(300)), wire(now + chrono::Duration::seconds(3600))); + format!("('dw-native-query:{id}','{task}',{expiration})") + }; + let mut rows: Vec = (0..64).map(|_| row(now.timestamp() - 30)).collect(); + rows.extend((0..64).map(|_| row(now.timestamp() + 3600))); + database.execute(sqlx::AssertSqlSafe(format!("INSERT INTO dw_query_cache VALUES {}", rows.join(",")))).await; + for index in 0..3 { + let caller = admit_query(&app, &path); + let (claimed, _) = poll_waiting(&app, &format!("stale-{index}"), 5).await; + let task = claimed["task"].clone(); + assert!(task.is_object(), "query {index} was not claimed: {claimed}"); + finish_query(&app, &task).await; + let (status, answer) = caller.await.unwrap(); + assert_eq!(status, StatusCode::OK, "query {index}: {answer}"); + } + runtime.close().await; + database.remove().await; +}