From 97a6bd732fcb6329ce733459b70a86e036f6e03f Mon Sep 17 00:00:00 2001
From: MechaCat02 <fabian@diekaulbachs.de>
Date: Sun, 7 Jun 2026 20:16:33 +0200
Subject: [PATCH] fix(manager-core): F-P-006 cap queue::depth scans at 10k rows
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Scripts called queue::depth(name) / queue::depth_pending(name) per
invocation; each ran an unbounded COUNT(*) over the queue_messages
partial index. On a backed-up queue with millions of rows, every call
scanned the whole partition.

Wrap the predicate in a LIMIT-capped subquery so the worst-case scan
is bounded:

  SELECT COUNT(*) FROM (
      SELECT 1 FROM queue_messages WHERE ... LIMIT 10000
  ) sub

QUEUE_DEPTH_SCAN_CAP = 10_000. Callers that need an exact depth on
queues larger than 10k use the admin /apps/{id}/queues endpoint which
tolerates the slower COUNT for its much lower read frequency.

`list_for_app` (dashboard queue overview) is left at full COUNT —
separate finding F-P-006 second-bullet, deferred to v1.2 along with
per-queue running counters.

AUDIT.md anchor: F-P-006 (first half).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 crates/manager-core/src/queue_repo.rs | 34 +++++++++++++++++++++++----
 1 file changed, 29 insertions(+), 5 deletions(-)

diff --git a/crates/manager-core/src/queue_repo.rs b/crates/manager-core/src/queue_repo.rs
index d45d701..588bceb 100644
--- a/crates/manager-core/src/queue_repo.rs
+++ b/crates/manager-core/src/queue_repo.rs
@@ -22,6 +22,13 @@ use uuid::Uuid;
 
 use crate::dead_letter_repo::{DeadLetterRepoError, NewDeadLetter};
 
+/// F-P-006: maximum rows scanned by `depth()` / `depth_pending()`.
+/// Scripts calling `queue::depth(name)` per invocation paid an unbounded
+/// COUNT(*); capping the scan at 10k lets the SDK report "10000" as a
+/// floor for very-deep queues without ever touching more than that many
+/// index rows.
+const QUEUE_DEPTH_SCAN_CAP: i64 = 10_000;
+
 #[derive(Debug, thiserror::Error)]
 pub enum QueueRepoError {
     #[error("database error: {0}")]
@@ -262,11 +269,24 @@ impl QueueRepo for PostgresQueueRepo {
     }
 
     async fn depth(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError> {
+        // F-P-006: cap the scan at QUEUE_DEPTH_SCAN_CAP rows. Scripts
+        // calling queue::depth(name) per invocation were doing
+        // unbounded COUNT(*) over the queue_messages partial index;
+        // on a backed-up queue with millions of rows that meant a
+        // full-partition scan every call. The capped subquery means
+        // the worst-case cost is bounded; callers needing exact
+        // depth on huge queues should use the admin dashboard which
+        // tolerates the slower COUNT for its read frequency.
         let (n,): (i64,) = sqlx::query_as(
-            "SELECT COUNT(*) FROM queue_messages WHERE app_id = $1 AND queue_name = $2",
+            "SELECT COUNT(*) FROM ( \
+                 SELECT 1 FROM queue_messages \
+                 WHERE app_id = $1 AND queue_name = $2 \
+                 LIMIT $3 \
+             ) sub",
         )
         .bind(app_id.into_inner())
         .bind(queue_name)
+        .bind(QUEUE_DEPTH_SCAN_CAP)
         .fetch_one(&self.pool)
         .await?;
         Ok(u64::try_from(n).unwrap_or(0))
@@ -274,13 +294,17 @@ impl QueueRepo for PostgresQueueRepo {
 
     async fn depth_pending(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError> {
         let (n,): (i64,) = sqlx::query_as(
-            "SELECT COUNT(*) FROM queue_messages \
-             WHERE app_id = $1 AND queue_name = $2 \
-               AND claim_token IS NULL \
-               AND (deliver_after IS NULL OR deliver_after <= NOW())",
+            "SELECT COUNT(*) FROM ( \
+                 SELECT 1 FROM queue_messages \
+                 WHERE app_id = $1 AND queue_name = $2 \
+                   AND claim_token IS NULL \
+                   AND (deliver_after IS NULL OR deliver_after <= NOW()) \
+                 LIMIT $3 \
+             ) sub",
         )
         .bind(app_id.into_inner())
         .bind(queue_name)
+        .bind(QUEUE_DEPTH_SCAN_CAP)
         .fetch_one(&self.pool)
         .await?;
         Ok(u64::try_from(n).unwrap_or(0))