From 6594f94c9d6e740bf0ac5a599066763335efd2ba Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 2 Nov 2026 16:02:05 -0600 Subject: [PATCH 1/3] Native: answered queries no longer hold query queue capacity (#964) Native keeps every query task in dw_query_cache until its receipt expires (deadline plus at least 60 s), whatever its status. query_entries scanned the namespace with LIMIT 65 across all statuses, and admission refused at 64 entries. So 64 answered queries in a namespace refused every new query, and blocked the worker claim scan, until the oldest receipts expired. Python SDK 2.12.4 hit it at query 64 of a sequential loop; PHP 2.11.8 answered all 75. The scan now reads only open (pending or leased) entries on each backend, and a bounded sweep removes expired receipts, which the scan used to do. Answered, failed and timed-out receipts stay readable by key until they expire. Admission keeps its 16-open-task bound, as PHP's counts only pending task ids. answered_queries_release_queue_capacity runs 80 sequential queries on one run. It fails at query 64 on 923d45b3 and passes now. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 10 +++++++--- rust/src/runtime/queries.rs | 15 +++++++++++++++ rust/tests/execution/query_poll_requests.rs | 19 +++++++++++++++++++ 3 files changed, 41 insertions(+), 3 deletions(-) diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index b6c132b69..b42b9494e 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -72,7 +72,10 @@ pub(super) trait Backend: Database { // Local activities in the namespace, as PHP's operator metrics count them. const LOCAL_ACTIVITY_EXECUTIONS: &str = "SELECT status FROM activity_executions WHERE workflow_run_id IN (SELECT id FROM workflow_runs WHERE namespace=$1) AND activity_options->>'execution_mode'='local'"; const LOCAL_ACTIVITY_ATTEMPTS: &str = "SELECT p.status FROM activity_attempts p JOIN activity_executions a ON a.id=p.activity_execution_id WHERE p.workflow_run_id IN (SELECT id FROM workflow_runs WHERE namespace=$1) AND a.activity_options->>'execution_mode'='local'"; - const QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND value::jsonb->>'namespace'=$1 ORDER BY \"key\" LIMIT 65"; + // Only open query tasks: an answered task keeps its receipt until it + // expires, but must not hold queue capacity (Server #964). + const QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND value::jsonb->>'namespace'=$1 AND value::jsonb->>'status' IN ('pending','leased') ORDER BY \"key\" LIMIT 65"; + const QUERY_RECEIPTS_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 32"; const QUERY_CACHE_GET: &str = "SELECT value,expiration FROM dw_query_cache WHERE \"key\"=$1"; const QUERY_CACHE_INSERT: &str = @@ -223,7 +226,7 @@ SELECT id,ordinal FROM ordinals WHERE position IS NOT ordinal ORDER BY ordinal L "#; const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND json_extract(payload,'$.failure_id') COLLATE BINARY=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1"; 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 QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND json_extract(value,'$.namespace')=$1 AND json_extract(value,'$.status') IN ('pending','leased') 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#" @@ -536,7 +539,8 @@ impl Backend for MySql { Ok(row.try_get(field)?) } 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 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 ? AND JSON_UNQUOTE(JSON_EXTRACT(value,'$.status')) IN ('pending','leased') ORDER BY `key` LIMIT 65"; + const QUERY_RECEIPTS_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-query:%' AND expiration<=$1 ORDER BY `key` LIMIT 32"; 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)"; diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index 63a333793..79742cc6e 100644 --- a/rust/src/runtime/queries.rs +++ b/rust/src/runtime/queries.rs @@ -302,10 +302,25 @@ where for<'r> i64: Decode<'r, DB>, usize: ColumnIndex, { + /// The namespace's open query tasks (pending or leased), at most + /// `MAX_ENTRIES`. An answered, failed or timed-out task keeps its receipt + /// until it expires but holds no queue capacity, as in PHP, whose + /// admission counts only pending task ids (Server #964). Expired receipts + /// are removed a bounded batch at a time. pub(super) async fn query_entries( &self, tx: &mut Transaction<'_, DB>, ) -> Result> { + for expired in Self::query(DB::QUERY_RECEIPTS_EXPIRED) + .bind(now()) + .fetch_all(&mut **tx) + .await? + { + Self::query(DB::QUERY_CACHE_DELETE) + .bind(DB::string(&expired, "key")?) + .execute(&mut **tx) + .await?; + } let rows = Self::query(DB::QUERY_CACHE_SCAN) .bind(self.namespace.to_string()) .fetch_all(&mut **tx) diff --git a/rust/tests/execution/query_poll_requests.rs b/rust/tests/execution/query_poll_requests.rs index 1180755eb..b98bc96af 100644 --- a/rust/tests/execution/query_poll_requests.rs +++ b/rust/tests/execution/query_poll_requests.rs @@ -202,3 +202,22 @@ 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; +} From 5f78e8fec72019f736e90c2a4444965b7dfb6de1 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 2 Nov 2026 16:18:28 -0600 Subject: [PATCH 2/3] Native query cache: exact-key open sets and expiry pages, no table scans (#964) Security's hold on 6594f94c: filtering the namespace scan to open rows left it, and the expiry sweep, scanning every retained receipt in the shared cache table (JSON predicates and expiration have no index). A key range isn't safe either: on PostgreSQL the key column uses the database's collation (en_US.utf8 in the CI image), which doesn't order punctuation bytewise. Every request-path read is now by primary key: - Open set: one row per namespace (key from a namespace digest, value naming the namespace in full) lists its open task ids, validated as unique canonical ULIDs, at most 64. Admission adds the new id. Each read fetches the listed tasks by key, keeps only this namespace's pending or leased ones under their own id, and rewrites the set when members closed, expired or vanished. - Expiry: every task row is filed at admission under the minute it expires, in pages of at most 128 keys. A sweep behind a watermark row deletes expired keys by key, at most 30 minutes and 2 pages per call, and files a key again if its expiry moved. A row already expired when filed is deleted at once. All of it runs in the caller's transaction. - Cutover: 3.0.1 kept query rows with neither. A global marker walks the cache table one primary-key page (64 rows) per call, once for the whole server, listing open tasks in their namespaces' sets and filing every query row. No per-namespace fallback scan exists. Receipts keep their keys, so the waiting caller, idempotent retries and completions read them as before. thousands_of_retained_receipts_leave_ queries_unaffected runs 20 queries with 3000 answered receipts present. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 16 +- rust/src/runtime/queries.rs | 263 +++++++++++++++++--- rust/tests/execution/query_arguments.rs | 7 +- rust/tests/execution/query_poll_requests.rs | 31 +++ 4 files changed, 278 insertions(+), 39 deletions(-) diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index b42b9494e..bacef6360 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -72,10 +72,7 @@ pub(super) trait Backend: Database { // Local activities in the namespace, as PHP's operator metrics count them. const LOCAL_ACTIVITY_EXECUTIONS: &str = "SELECT status FROM activity_executions WHERE workflow_run_id IN (SELECT id FROM workflow_runs WHERE namespace=$1) AND activity_options->>'execution_mode'='local'"; const LOCAL_ACTIVITY_ATTEMPTS: &str = "SELECT p.status FROM activity_attempts p JOIN activity_executions a ON a.id=p.activity_execution_id WHERE p.workflow_run_id IN (SELECT id FROM workflow_runs WHERE namespace=$1) AND a.activity_options->>'execution_mode'='local'"; - // Only open query tasks: an answered task keeps its receipt until it - // expires, but must not hold queue capacity (Server #964). - const QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND value::jsonb->>'namespace'=$1 AND value::jsonb->>'status' IN ('pending','leased') ORDER BY \"key\" LIMIT 65"; - const QUERY_RECEIPTS_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 32"; + const QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND value::jsonb->>'namespace'=$1 ORDER BY \"key\" LIMIT 65"; const QUERY_CACHE_GET: &str = "SELECT value,expiration FROM dw_query_cache WHERE \"key\"=$1"; const QUERY_CACHE_INSERT: &str = @@ -83,6 +80,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"; @@ -226,7 +227,7 @@ SELECT id,ordinal FROM ordinals WHERE position IS NOT ordinal ORDER BY ordinal L "#; const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND json_extract(payload,'$.failure_id') COLLATE BINARY=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1"; 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 AND json_extract(value,'$.status') IN ('pending','leased') ORDER BY \"key\" LIMIT 65"; + 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#" @@ -539,8 +540,7 @@ impl Backend for MySql { Ok(row.try_get(field)?) } 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 ? AND JSON_UNQUOTE(JSON_EXTRACT(value,'$.status')) IN ('pending','leased') ORDER BY `key` LIMIT 65"; - const QUERY_RECEIPTS_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-query:%' AND expiration<=$1 ORDER BY `key` LIMIT 32"; + 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)"; @@ -566,6 +566,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"; diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index 79742cc6e..7c5871f8e 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,11 @@ fn poll_result(task: Option, status: &str) -> Value { fn now() -> i64 { chrono::Utc::now().timestamp() } +/// 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() { @@ -303,53 +320,237 @@ where usize: ColumnIndex, { /// The namespace's open query tasks (pending or leased), at most - /// `MAX_ENTRIES`. An answered, failed or timed-out task keeps its receipt - /// until it expires but holds no queue capacity, as in PHP, whose - /// admission counts only pending task ids (Server #964). Expired receipts - /// are removed a bounded batch at a time. + /// `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> { - for expired in Self::query(DB::QUERY_RECEIPTS_EXPIRED) - .bind(now()) - .fetch_all(&mut **tx) - .await? - { - Self::query(DB::QUERY_CACHE_DELETE) - .bind(DB::string(&expired, "key")?) - .execute(&mut **tx) - .await?; - } - 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![]); + }; + if set["namespace"] != namespace { + return Ok(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()); + } + } + Ok(ids) + } + + 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; + }; + Self::file_expiry(tx, &key, DB::number(row, "expiration")?).await?; + let Ok(task) = serde_json::from_str::(&DB::string(row, "value")?) else { + continue; + }; + if task["query_task_id"] == id + && canonical_id(id) + && matches!(task["status"].as_str(), Some("pending" | "leased")) + && let Some(namespace) = task["namespace"].as_str() + { + // A full set leaves the rest to time out by their deadlines. + let _ = Self::list_open(tx, namespace, id).await; + } + } + 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, @@ -530,6 +731,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. 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 b98bc96af..7c5b7d553 100644 --- a/rust/tests/execution/query_poll_requests.rs +++ b/rust/tests/execution/query_poll_requests.rs @@ -221,3 +221,34 @@ async fn answered_queries_release_queue_capacity() { 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; +} From 5544ba68bb75ae9ec26880126005d8160aefc3eb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 2 Nov 2026 16:27:27 -0600 Subject: [PATCH 3/3] Native query cache: keyed long-poll readiness; harden the cutover (#964) Security's re-review of 5f78e8fe: - query_poll_ready and poll_wake_snapshot still ran POLL_WAKE_QUERIES_SQL, the shared-table prefix plus JSON scan, on every long-poll probe. Both now use ready_query_keys: the namespace's open set read lock-free by key, then each listed task by key, with the old filters (queue, query deadline, pending or lease-expired). The statement is removed from every backend. - The cutover deletes an already-expired row and never lists it, closes a pending row past its query deadline before it could take a slot, and propagates every failure except a full set, so its page and marker roll back together. the_cutover_lists_only_answerable_rows: 128 stale 3.0.1 rows (expired, or pending past their deadline) leave three live queries answered. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 4 - rust/src/runtime/cross_kind_poll.rs | 15 +--- rust/src/runtime/queries.rs | 92 ++++++++++++++++----- rust/tests/execution/query_poll_requests.rs | 32 +++++++ 4 files changed, 106 insertions(+), 37 deletions(-) diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index bacef6360..177a112bc 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"; @@ -229,7 +228,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, @@ -541,7 +539,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 = @@ -603,7 +600,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 7c5871f8e..14bb94a37 100644 --- a/rust/src/runtime/queries.rs +++ b/rust/src/runtime/queries.rs @@ -80,6 +80,21 @@ 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) @@ -378,16 +393,45 @@ where let Some(set) = Self::cache_document(tx, &Self::open_set_key(namespace)).await? else { return Ok(vec![]); }; - if set["namespace"] != namespace { + 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 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()); + }; + 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(ids) + Ok(ready) } async fn write_open_set(tx: &mut Transaction<'_, DB>, namespace: &str, ids: &[String]) -> Result<()> { @@ -534,17 +578,30 @@ where let Some(id) = key.strip_prefix("dw-native-query:") else { continue; }; - Self::file_expiry(tx, &key, DB::number(row, "expiration")?).await?; - let Ok(task) = serde_json::from_str::(&DB::string(row, "value")?) else { + 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() { - // A full set leaves the rest to time out by their deadlines. - let _ = Self::list_open(tx, namespace, id).await; + 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; @@ -1032,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); } @@ -1051,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_poll_requests.rs b/rust/tests/execution/query_poll_requests.rs index 7c5b7d553..2340ebdc3 100644 --- a/rust/tests/execution/query_poll_requests.rs +++ b/rust/tests/execution/query_poll_requests.rs @@ -252,3 +252,35 @@ async fn thousands_of_retained_receipts_leave_queries_unaffected() { 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; +}