fix(manager-core): F-P-006 cap queue::depth scans at 10k rows
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>
This commit is contained in:
@@ -22,6 +22,13 @@ use uuid::Uuid;
|
|||||||
|
|
||||||
use crate::dead_letter_repo::{DeadLetterRepoError, NewDeadLetter};
|
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)]
|
#[derive(Debug, thiserror::Error)]
|
||||||
pub enum QueueRepoError {
|
pub enum QueueRepoError {
|
||||||
#[error("database error: {0}")]
|
#[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> {
|
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(
|
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(app_id.into_inner())
|
||||||
.bind(queue_name)
|
.bind(queue_name)
|
||||||
|
.bind(QUEUE_DEPTH_SCAN_CAP)
|
||||||
.fetch_one(&self.pool)
|
.fetch_one(&self.pool)
|
||||||
.await?;
|
.await?;
|
||||||
Ok(u64::try_from(n).unwrap_or(0))
|
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> {
|
async fn depth_pending(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError> {
|
||||||
let (n,): (i64,) = sqlx::query_as(
|
let (n,): (i64,) = sqlx::query_as(
|
||||||
"SELECT COUNT(*) FROM queue_messages \
|
"SELECT COUNT(*) FROM ( \
|
||||||
WHERE app_id = $1 AND queue_name = $2 \
|
SELECT 1 FROM queue_messages \
|
||||||
AND claim_token IS NULL \
|
WHERE app_id = $1 AND queue_name = $2 \
|
||||||
AND (deliver_after IS NULL OR deliver_after <= NOW())",
|
AND claim_token IS NULL \
|
||||||
|
AND (deliver_after IS NULL OR deliver_after <= NOW()) \
|
||||||
|
LIMIT $3 \
|
||||||
|
) sub",
|
||||||
)
|
)
|
||||||
.bind(app_id.into_inner())
|
.bind(app_id.into_inner())
|
||||||
.bind(queue_name)
|
.bind(queue_name)
|
||||||
|
.bind(QUEUE_DEPTH_SCAN_CAP)
|
||||||
.fetch_one(&self.pool)
|
.fetch_one(&self.pool)
|
||||||
.await?;
|
.await?;
|
||||||
Ok(u64::try_from(n).unwrap_or(0))
|
Ok(u64::try_from(n).unwrap_or(0))
|
||||||
|
|||||||
Reference in New Issue
Block a user