From e009a7fcef52f941fa7c0e40632e6c27b833a66e Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 09:01:25 +1300 Subject: [PATCH 1/6] fix(worker): keep planner statistics fresh on tables autovacuum never analyses (0.6) Autovacuum does not analyse partitioned tables, so the thirteen ring parents have no inherited statistics, and the two head tables (queue_claim_heads, queue_enqueue_heads) never reach the autoanalyze threshold, so the planner prices lane joins against them at its defaults. The maintenance leader now runs ANALYZE on them when it is elected and every five minutes (ClientBuilder::statistics_refresh_interval, Python statistics_refresh_interval_ms; zero disables). One relation per transaction, with a bounded lock_timeout and statement_timeout, partitioned parents with SKIP_LOCKED, in a task beside the leader loop so a slow ANALYZE never delays rotate, prune or promote ticks. The 0.6 series cannot add a migration version, so the SECURITY DEFINER function awa.analyze_queue_storage_relation() ships as the idempotent schema patch analyze_queue_storage_relation (applied by `awa migrate` on a v040 schema, exported as R__analyze_queue_storage_relation). A missing function or an owner that may not analyse the tables is logged once and recorded in awa.maintenance.analyze.attempts without affecting job processing. --- CHANGELOG.md | 37 ++ awa-metrics/src/lib.rs | 49 +- .../analyze_queue_storage_relation.sql | 116 ++++ awa-model/src/migrations.rs | 6 + awa-python/python/awa/_awa.pyi | 1 + awa-python/python/awa/client.py | 7 + awa-python/src/client.rs | 6 +- awa-python/tests/test_start_config.py | 14 + awa-worker/src/client.rs | 21 + awa-worker/src/lib.rs | 1 + awa-worker/src/maintenance.rs | 34 ++ awa-worker/src/statistics.rs | 347 +++++++++++ awa/tests/statistics_refresh_test.rs | 563 ++++++++++++++++++ docs/architecture.md | 3 + docs/configuration.md | 11 + docs/grafana/README.md | 2 + docs/security.md | 2 + 17 files changed, 1218 insertions(+), 2 deletions(-) create mode 100644 awa-model/migrations/patches/analyze_queue_storage_relation.sql create mode 100644 awa-worker/src/statistics.rs create mode 100644 awa/tests/statistics_refresh_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 879a17deb..069c34283 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,43 @@ Notable changes between releases. Detailed migration notes for storage transitio ## [Unreleased] +### Added + +- **The maintenance leader keeps planner statistics fresh on the queue-storage + tables autovacuum never analyses.** Autovacuum does not analyse partitioned + tables, so the thirteen ring parents (`ready_entries`, `done_entries`, + `leases`, ...) had no inherited statistics, and the two small head tables + (`queue_claim_heads`, `queue_enqueue_heads`) never reach the autoanalyze + threshold, so the planner priced lane joins against them at its defaults + while they drove joins over hundreds of thousands of retained rows. The + leader now runs `ANALYZE` on them when it is elected and then every five + minutes. `ClientBuilder::statistics_refresh_interval` (Python + `statistics_refresh_interval_ms`) changes the cadence; zero disables it. It + analyses one relation per transaction with a 2 s `lock_timeout`, skips + children that prune holds, and never makes prune wait: a prune attempt that + lands while ANALYZE samples the oldest slot's tables returns `blocked` and + retries on its normal backoff. Results are in + `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration` + (scope `heads` or `ring_parent`; outcome `ok`, `skipped_privilege`, + `skipped_lock`, `unavailable` or `failed`). +- **Schema patch `analyze_queue_storage_relation` lets a runtime role that does + not own the tables refresh statistics.** `ANALYZE` skips a table its caller + may not analyse, and the documented runtime grants do not include + `MAINTAIN`, so the refresh runs through the `SECURITY DEFINER` function + `awa.analyze_queue_storage_relation(schema, relation)`, owned by the role + that applied the patch. It analyses only a head table or partitioned table + of a queue-storage schema owned by the owner of the `awa` schema, and + `EXECUTE` is granted to `PUBLIC` like the other runtime helpers. The 0.6 + series cannot add a migration version, so it ships like + `wait_free_dirty_marks`: **run `awa migrate` as the schema owner after + upgrading** (it applies the patch on a v040 schema; `awa migrate --sql` + and `--extract-to` emit it as the repeatable script + `R__analyze_queue_storage_relation`). No grant changes are needed. The same + function is awa 0.7 migration v053, which replaces it cleanly. A binary + running before the patch, or against a function whose owner may not analyse + the tables, logs one warning naming what is missing and skips the refresh + without affecting job processing. Earlier binaries never call the function. + ### Fixed - **`admin::job_kind_overviews` on queue storage no longer depends on the diff --git a/awa-metrics/src/lib.rs b/awa-metrics/src/lib.rs index 3f81abce3..34e374998 100644 --- a/awa-metrics/src/lib.rs +++ b/awa-metrics/src/lib.rs @@ -70,6 +70,8 @@ pub mod names { pub const MAINTENANCE_ROTATE_SKIPPED_ROWS: &str = "awa.maintenance.rotate.skipped_rows"; pub const MAINTENANCE_PRUNE_ATTEMPTS: &str = "awa.maintenance.prune.attempts"; pub const MAINTENANCE_PRUNE_SKIPPED_ROWS: &str = "awa.maintenance.prune.skipped_rows"; + pub const MAINTENANCE_ANALYZE_ATTEMPTS: &str = "awa.maintenance.analyze.attempts"; + pub const MAINTENANCE_ANALYZE_DURATION: &str = "awa.maintenance.analyze.duration"; pub const STORAGE_TRANSITION_READY: &str = "awa.storage.transition_ready"; pub const STORAGE_CANONICAL_LIVE_BACKLOG: &str = "awa.storage.canonical_live_backlog"; pub const STORAGE_LIVE_RUNTIME_CAPABILITY: &str = "awa.storage.live_runtime_capability"; @@ -244,6 +246,13 @@ pub struct AwaMetrics { /// rather than one per tick. Fleets alert on this rather than /// scraping the matching `tracing::warn!` line. Issue #242. pub maintenance_branch_overrun_total: Counter, + /// Planner-statistics `ANALYZE` calls by scope (`heads`, `ring_parent`) + /// and outcome (`ok`, `skipped_privilege`, `skipped_lock`, + /// `unavailable`, `failed`). One increment per relation per refresh pass. + pub maintenance_analyze_attempts: Counter, + /// Wall-clock duration of one planner-statistics `ANALYZE` call, by scope + /// and outcome. + pub maintenance_analyze_duration_seconds: Histogram, /// Current ring `current_slot` per ring, sampled from each rotate call. /// The slot number itself isn't meaningful but the rate of advance is — /// dashboards plot `rate(slot_changes)` to see whether rotation is @@ -537,6 +546,21 @@ impl AwaMetrics { ) .with_unit("{episode}") .build(), + maintenance_analyze_attempts: meter + .u64_counter(names::MAINTENANCE_ANALYZE_ATTEMPTS) + .with_description( + "Planner-statistics ANALYZE calls by scope (heads/ring_parent) and outcome", + ) + .with_unit("{attempt}") + .build(), + maintenance_analyze_duration_seconds: meter + .f64_histogram(names::MAINTENANCE_ANALYZE_DURATION) + .with_description( + "Duration of one planner-statistics ANALYZE call by scope and outcome", + ) + .with_unit("s") + .with_boundaries(WAIT_DURATION_BUCKETS_SECONDS.to_vec()) + .build(), ring_current_slot: meter .i64_gauge(names::RING_CURRENT_SLOT) .with_description("Current slot index per ring (queue/lease/claim)") @@ -1092,6 +1116,24 @@ impl AwaMetrics { self.maintenance_branch_overrun_total.add(1, &attrs); } + /// Record one planner-statistics `ANALYZE` call. `scope` is `"heads"` or + /// `"ring_parent"`; `outcome` is `"ok"`, `"skipped_privilege"`, + /// `"skipped_lock"`, `"unavailable"` or `"failed"`. + pub fn record_statistics_analyze( + &self, + scope: &'static str, + outcome: &'static str, + duration: Duration, + ) { + let attrs = [ + opentelemetry::KeyValue::new("awa.analyze.scope", scope), + opentelemetry::KeyValue::new("awa.analyze.outcome", outcome), + ]; + self.maintenance_analyze_attempts.add(1, &attrs); + self.maintenance_analyze_duration_seconds + .record(duration.as_secs_f64(), &attrs); + } + /// Record a ring prune outcome. pub fn record_prune_outcome(&self, ring: &'static str, outcome: &awa_model::PruneOutcome) { let (label, reason, count) = match outcome { @@ -1176,7 +1218,7 @@ mod tests { .build(); let metrics = AwaMetrics::new(&provider.meter("awa-metrics-test")); - let seconds: [(&str, &Histogram, &[f64]); 8] = [ + let seconds: [(&str, &Histogram, &[f64]); 9] = [ ( names::ENQUEUE_DURATION, &metrics.enqueue_duration_seconds, @@ -1192,6 +1234,11 @@ mod tests { &metrics.maintenance_branch_duration_seconds, &WAIT_DURATION_BUCKETS_SECONDS, ), + ( + names::MAINTENANCE_ANALYZE_DURATION, + &metrics.maintenance_analyze_duration_seconds, + &WAIT_DURATION_BUCKETS_SECONDS, + ), ( names::DISPATCH_WAKE_TO_CLAIM_DURATION, &metrics.dispatch_wake_to_claim_seconds, diff --git a/awa-model/migrations/patches/analyze_queue_storage_relation.sql b/awa-model/migrations/patches/analyze_queue_storage_relation.sql new file mode 100644 index 000000000..67a377369 --- /dev/null +++ b/awa-model/migrations/patches/analyze_queue_storage_relation.sql @@ -0,0 +1,116 @@ +-- Schema patch analyze_queue_storage_relation: planner statistics helper for +-- the queue-storage tables autovacuum never analyses (#630). +-- Counterpart of awa 0.7 migration v053, minus the schema_version row; v053 +-- replaces this body with one that checks the schema through +-- awa.queue_storage_schemas(), which 0.6 does not have. +-- +-- 0.6 delivery: the 0.6 series cannot add a migration version (0.7 owns v041 +-- and above), so this ships as an idempotent schema patch. `awa migrate` +-- applies it when the schema is exactly at v040 and the probe in +-- migrations.rs reports it missing; `awa migrate --sql` prints it as a +-- repeatable R__ script for external runners. +-- +-- Autovacuum does not analyse partitioned tables, so the ring parents have no +-- inherited statistics, and it never reaches the autoanalyze threshold on the +-- two small head tables (queue_claim_heads, queue_enqueue_heads). The +-- maintenance leader refreshes both on an interval. Runtimes usually connect +-- as a role that does not own the tables, and ANALYZE skips with a warning a +-- table its caller may not analyse, so the refresh runs through this +-- SECURITY DEFINER function owned by the role that applies the patch, which +-- owns the tables. +-- +-- The function analyses one relation per call; the caller runs each call as +-- its own transaction, so the locks ANALYZE takes on a partitioned table's +-- children are held for that table rather than for the whole refresh. +-- SKIP_LOCKED skips a child that ring prune holds ACCESS EXCLUSIVE instead of +-- waiting for it and aborting the statistics gathered so far. +-- +-- Definer boundary: only a head table or a partitioned table of a queue-storage +-- schema (the AWA-specific claim_ready_runtime sentinel) that is owned by the +-- owner of the awa schema is analysed, from quoted identifiers only. ANALYZE +-- writes statistics and nothing else. +-- +-- Returns 'analyzed'; 'skipped_privilege' when the function owner may not +-- analyse the relation; or 'missing' when the relation does not exist. +-- Takes no lock on any queue-storage table and completes in milliseconds. +-- Releases before the patch never call it; a binary that runs before the patch +-- logs once that the function is missing and skips the refresh. + +CREATE OR REPLACE FUNCTION awa.analyze_queue_storage_relation( + p_schema TEXT, + p_relation TEXT +) +RETURNS TEXT +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = pg_catalog, pg_temp +AS $$ +DECLARE + v_relid OID; + v_relkind "char"; + v_relowner OID; + v_may_analyze BOOLEAN; +BEGIN + IF to_regprocedure(format( + '%I.claim_ready_runtime(text,bigint,double precision,double precision)', + p_schema + )) IS NULL THEN + RAISE EXCEPTION '% is not a queue-storage schema', p_schema + USING ERRCODE = '22023'; + END IF; + + v_relid := to_regclass(format('%I.%I', p_schema, p_relation)); + IF v_relid IS NULL THEN + RETURN 'missing'; + END IF; + + SELECT class.relkind, class.relowner INTO v_relkind, v_relowner + FROM pg_class AS class WHERE class.oid = v_relid; + IF NOT ( + v_relkind = 'p' + OR p_relation IN ('queue_claim_heads', 'queue_enqueue_heads') + ) THEN + RAISE EXCEPTION '%.% is not a partitioned or head table of a queue-storage schema', + p_schema, p_relation + USING ERRCODE = '22023'; + END IF; + + IF v_relowner IS DISTINCT FROM ( + SELECT namespace.nspowner FROM pg_namespace AS namespace WHERE namespace.nspname = 'awa' + ) THEN + RAISE EXCEPTION '%.% is not owned by the awa schema owner', p_schema, p_relation + USING ERRCODE = '42501'; + END IF; + + -- ANALYZE skips, with a warning rather than an error, a table the caller + -- may not analyse: the owner, a MAINTAIN holder (PostgreSQL 17+), a + -- member of pg_maintain, or the database owner may analyse it. + SELECT + pg_has_role(current_user, v_relowner, 'USAGE') + OR pg_has_role( + current_user, + (SELECT database.datdba FROM pg_database AS database + WHERE database.datname = current_database()), + 'USAGE' + ) + OR CASE + WHEN current_setting('server_version_num')::int >= 170000 + THEN has_table_privilege(v_relid, 'MAINTAIN') + ELSE FALSE + END + INTO v_may_analyze; + + IF NOT v_may_analyze THEN + RETURN 'skipped_privilege'; + END IF; + + IF v_relkind = 'p' THEN + EXECUTE format('ANALYZE (SKIP_LOCKED) %I.%I', p_schema, p_relation); + ELSE + EXECUTE format('ANALYZE %I.%I', p_schema, p_relation); + END IF; + RETURN 'analyzed'; +END +$$; + +GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(TEXT, TEXT) TO PUBLIC; diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs index 193c465fc..204329e16 100644 --- a/awa-model/src/migrations.rs +++ b/awa-model/src/migrations.rs @@ -347,6 +347,12 @@ pub const SCHEMA_PATCHES: &[SchemaPatch] = &[ ")" ), }, + SchemaPatch { + name: "analyze_queue_storage_relation", + description: "Planner statistics helper for tables autovacuum never analyses (awa 0.7 migration v053)", + sql: include_str!("../migrations/patches/analyze_queue_storage_relation.sql"), + applied_probe: "SELECT to_regprocedure('awa.analyze_queue_storage_relation(text,text)') IS NOT NULL", + }, ]; /// Schema patches a database still needs to reach this binary's schema shape. diff --git a/awa-python/python/awa/_awa.pyi b/awa-python/python/awa/_awa.pyi index fcd040853..b589b8de6 100644 --- a/awa-python/python/awa/_awa.pyi +++ b/awa-python/python/awa/_awa.pyi @@ -575,6 +575,7 @@ class Client: deadline_rescue_interval_ms: int | None = None, callback_rescue_interval_ms: int | None = None, terminal_count_rollup_interval_ms: int | None = None, + statistics_refresh_interval_ms: int | None = None, queue_storage_schema: str | None = None, queue_storage_queue_slot_count: int = 16, queue_storage_lease_slot_count: int = 8, diff --git a/awa-python/python/awa/client.py b/awa-python/python/awa/client.py index a237cc838..2ee6b5175 100644 --- a/awa-python/python/awa/client.py +++ b/awa-python/python/awa/client.py @@ -859,6 +859,7 @@ async def start( deadline_rescue_interval_ms: int | None = None, callback_rescue_interval_ms: int | None = None, terminal_count_rollup_interval_ms: int | None = None, + statistics_refresh_interval_ms: int | None = None, queue_storage_schema: str | None = None, queue_storage_queue_slot_count: int = DEFAULT_QUEUE_STORAGE_QUEUE_SLOT_COUNT, queue_storage_lease_slot_count: int = DEFAULT_QUEUE_STORAGE_LEASE_SLOT_COUNT, @@ -900,6 +901,11 @@ async def start( counters. Exact reads include pending deltas, so the default favors fewer hot counter updates over immediate folding. + ``statistics_refresh_interval_ms`` controls how often the maintenance + leader runs ``ANALYZE`` on the queue-storage head tables and + partitioned ring parents, which autovacuum never analyses (default 5 + minutes; ``0`` disables the refresh). + Queue dicts also accept ``claimers`` and ``claim_batch_size`` for hot queue-storage queues. ``claimers`` adds dispatcher/claimer loops that share the queue's worker permits; benchmark ``2`` or ``4`` before @@ -921,6 +927,7 @@ async def start( deadline_rescue_interval_ms=deadline_rescue_interval_ms, callback_rescue_interval_ms=callback_rescue_interval_ms, terminal_count_rollup_interval_ms=terminal_count_rollup_interval_ms, + statistics_refresh_interval_ms=statistics_refresh_interval_ms, queue_storage_schema=queue_storage_schema, queue_storage_queue_slot_count=queue_storage_queue_slot_count, queue_storage_lease_slot_count=queue_storage_lease_slot_count, diff --git a/awa-python/src/client.rs b/awa-python/src/client.rs index 7de483c14..8d91ae673 100644 --- a/awa-python/src/client.rs +++ b/awa-python/src/client.rs @@ -1666,7 +1666,7 @@ impl PyClient { }) } - #[pyo3(signature = (queues=None, *, poll_interval_ms=200, global_max_workers=None, completed_retention_hours=None, failed_retention_hours=None, descriptor_retention_days=None, cleanup_batch_size=None, leader_election_interval_ms=None, heartbeat_interval_ms=None, promote_interval_ms=None, heartbeat_rescue_interval_ms=None, heartbeat_staleness_ms=None, deadline_rescue_interval_ms=None, callback_rescue_interval_ms=None, terminal_count_rollup_interval_ms=None, queue_storage_schema=None, queue_storage_queue_slot_count=16, queue_storage_lease_slot_count=8, queue_storage_claim_slot_count=8, queue_storage_queue_stripe_count=1, queue_storage_queue_rotate_interval_ms=1000, queue_storage_lease_rotate_interval_ms=1000, queue_storage_claim_rotate_interval_ms=None, storage_transition_role=None))] + #[pyo3(signature = (queues=None, *, poll_interval_ms=200, global_max_workers=None, completed_retention_hours=None, failed_retention_hours=None, descriptor_retention_days=None, cleanup_batch_size=None, leader_election_interval_ms=None, heartbeat_interval_ms=None, promote_interval_ms=None, heartbeat_rescue_interval_ms=None, heartbeat_staleness_ms=None, deadline_rescue_interval_ms=None, callback_rescue_interval_ms=None, terminal_count_rollup_interval_ms=None, statistics_refresh_interval_ms=None, queue_storage_schema=None, queue_storage_queue_slot_count=16, queue_storage_lease_slot_count=8, queue_storage_claim_slot_count=8, queue_storage_queue_stripe_count=1, queue_storage_queue_rotate_interval_ms=1000, queue_storage_lease_rotate_interval_ms=1000, queue_storage_claim_rotate_interval_ms=None, storage_transition_role=None))] #[allow(clippy::too_many_arguments)] fn start<'py>( &self, @@ -1686,6 +1686,7 @@ impl PyClient { deadline_rescue_interval_ms: Option, callback_rescue_interval_ms: Option, terminal_count_rollup_interval_ms: Option, + statistics_refresh_interval_ms: Option, queue_storage_schema: Option, queue_storage_queue_slot_count: u32, queue_storage_lease_slot_count: u32, @@ -1811,6 +1812,9 @@ impl PyClient { } builder = builder.terminal_count_rollup_interval(Duration::from_millis(ms)); } + if let Some(ms) = statistics_refresh_interval_ms { + builder = builder.statistics_refresh_interval(Duration::from_millis(ms)); + } if queue_storage_queue_slot_count == 0 { return Err(pyo3::exceptions::PyValueError::new_err( "queue_storage_queue_slot_count must be > 0", diff --git a/awa-python/tests/test_start_config.py b/awa-python/tests/test_start_config.py index 8a68c0123..b0f8823e9 100644 --- a/awa-python/tests/test_start_config.py +++ b/awa-python/tests/test_start_config.py @@ -316,12 +316,26 @@ async def handle(job): deadline_rescue_interval_ms=1_000, callback_rescue_interval_ms=1_000, terminal_count_rollup_interval_ms=1_000, + statistics_refresh_interval_ms=1_000, leader_election_interval_ms=200, promote_interval_ms=500, ) await client.shutdown() +@pytest.mark.asyncio +async def test_statistics_refresh_can_be_disabled(client): + """statistics_refresh_interval_ms=0 disables the refresh; start() accepts it.""" + queue = "cfg_statistics_refresh_off" + + @client.task(ConfigTestJob, queue=queue) + async def handle(job): + return None + + await client.start([(queue, 1)], statistics_refresh_interval_ms=0) + await client.shutdown() + + @pytest.mark.asyncio async def test_per_queue_retention_in_dict_config(client): """Dict form with per-queue retention config starts successfully.""" diff --git a/awa-worker/src/client.rs b/awa-worker/src/client.rs index e5a498dbb..e0d86abd9 100644 --- a/awa-worker/src/client.rs +++ b/awa-worker/src/client.rs @@ -148,6 +148,7 @@ pub struct ClientBuilder { leader_check_interval: Option, priority_aging_interval: Option, terminal_count_rollup_interval: Option, + statistics_refresh_interval: Option, completed_retention: Option, failed_retention: Option, descriptor_retention: Option, @@ -206,6 +207,7 @@ impl ClientBuilder { leader_check_interval: None, priority_aging_interval: None, terminal_count_rollup_interval: None, + statistics_refresh_interval: None, completed_retention: None, failed_retention: None, descriptor_retention: None, @@ -848,6 +850,20 @@ impl ClientBuilder { self } + /// Set how often the maintenance leader refreshes planner statistics for + /// the queue-storage tables autovacuum never analyses (default: 5 minutes). + /// + /// Autovacuum does not analyse partitioned tables, and never reaches its + /// threshold on the two small head tables, so the leader runs `ANALYZE` + /// on both when it is elected and then every interval, through + /// `awa.analyze_queue_storage_relation()` (a schema patch) so that a runtime + /// role that does not own the tables still works. `Duration::ZERO` + /// disables the refresh. Queue storage only. + pub fn statistics_refresh_interval(mut self, interval: Duration) -> Self { + self.statistics_refresh_interval = Some(interval); + self + } + /// Set how often queue depth/lag metrics are published (default: 30s). pub fn queue_stats_interval(mut self, interval: Duration) -> Self { self.queue_stats_interval = Some(interval); @@ -1097,6 +1113,7 @@ impl ClientBuilder { leader_check_interval: self.leader_check_interval, priority_aging_interval: self.priority_aging_interval, terminal_count_rollup_interval: self.terminal_count_rollup_interval, + statistics_refresh_interval: self.statistics_refresh_interval, completed_retention: self.completed_retention, failed_retention: self.failed_retention, descriptor_retention: self.descriptor_retention, @@ -1198,6 +1215,7 @@ pub struct Client { leader_check_interval: Option, priority_aging_interval: Option, terminal_count_rollup_interval: Option, + statistics_refresh_interval: Option, completed_retention: Option, failed_retention: Option, descriptor_retention: Option, @@ -1650,6 +1668,9 @@ impl Client { if let Some(interval) = self.terminal_count_rollup_interval { maintenance = maintenance.terminal_count_rollup_interval(interval); } + if let Some(interval) = self.statistics_refresh_interval { + maintenance = maintenance.statistics_refresh_interval(interval); + } if let Some(retention) = self.completed_retention { maintenance = maintenance.completed_retention(retention); } diff --git a/awa-worker/src/lib.rs b/awa-worker/src/lib.rs index 29a4bfdd2..6112ebc6c 100644 --- a/awa-worker/src/lib.rs +++ b/awa-worker/src/lib.rs @@ -13,6 +13,7 @@ pub mod maintenance; pub mod metrics; mod notify_hub; mod runtime; +mod statistics; mod storage; // Re-exports diff --git a/awa-worker/src/maintenance.rs b/awa-worker/src/maintenance.rs index b7c134177..4934e7e23 100644 --- a/awa-worker/src/maintenance.rs +++ b/awa-worker/src/maintenance.rs @@ -1,5 +1,6 @@ use crate::executor::DlqPolicy; use crate::runtime::InFlightMap; +use crate::statistics::{RefreshGuard, StatisticsRefresher}; use crate::storage::{QueueStorageRuntime, RuntimeStorage}; use awa_model::cron::{ atomic_enqueue, list_cron_jobs, upsert_cron_job, CronJobRow, CronMissedFirePolicy, @@ -707,12 +708,17 @@ pub struct MaintenanceService { priority_aging_interval: Duration, batch_operations_interval: Duration, terminal_count_rollup_interval: Duration, + /// How often the leader refreshes planner statistics for the queue-storage + /// tables autovacuum never analyses. Zero disables the refresh. + statistics_refresh_interval: Duration, /// How long a descriptor catalog row can sit without being refreshed /// before the maintenance leader deletes it. Zero disables cleanup. /// Default: 30 days. descriptor_retention: Duration, } +/// Default cadence of the planner-statistics refresh on the maintenance leader. +pub(crate) const DEFAULT_STATISTICS_REFRESH_INTERVAL: Duration = Duration::from_secs(300); const PROMOTE_BATCH_SIZE: i64 = 4_096; const PROMOTE_MAX_BATCHES_PER_TICK: usize = 32; const CRON_CATCH_UP_LIMIT: usize = 1_000; @@ -773,6 +779,7 @@ impl MaintenanceService { priority_aging_interval: Duration::from_secs(60), batch_operations_interval: Duration::from_secs(1), terminal_count_rollup_interval: Duration::from_secs(30), + statistics_refresh_interval: DEFAULT_STATISTICS_REFRESH_INTERVAL, descriptor_retention: Duration::from_secs(30 * 86400), // 30d } } @@ -799,6 +806,14 @@ impl MaintenanceService { self } + /// Set how often the leader runs `ANALYZE` on the queue-storage tables + /// autovacuum never analyses (default: 5 minutes). `Duration::ZERO` + /// disables the refresh. + pub fn statistics_refresh_interval(mut self, interval: Duration) -> Self { + self.statistics_refresh_interval = interval; + self + } + /// How long a descriptor catalog row can go without being re-synced /// before the maintenance leader deletes it (default: 30 days). Set /// to `Duration::ZERO` to disable — useful if you maintain the catalog @@ -1052,6 +1067,7 @@ impl MaintenanceService { cron_eval_cancel.clone(), self.cron_eval_interval, )); + let _statistics_refresh = self.spawn_statistics_refresh(); loop { tokio::select! { @@ -1245,6 +1261,24 @@ impl MaintenanceService { } } + /// Start the planner-statistics refresh for this leadership term. The + /// returned guard stops it when dropped. + fn spawn_statistics_refresh(&self) -> Option { + if self.statistics_refresh_interval.is_zero() { + return None; + } + let runtime = self.storage.queue_storage()?; + Some(RefreshGuard::spawn( + StatisticsRefresher::new( + self.pool.clone(), + runtime.store.schema().to_string(), + self.metrics.clone(), + ), + self.statistics_refresh_interval, + &self.cancel, + )) + } + fn stop_cron_evaluator(cancel: &CancellationToken, task: &JoinHandle<()>) { cancel.cancel(); task.abort(); diff --git a/awa-worker/src/statistics.rs b/awa-worker/src/statistics.rs new file mode 100644 index 000000000..04792cf9c --- /dev/null +++ b/awa-worker/src/statistics.rs @@ -0,0 +1,347 @@ +//! Planner-statistics refresh for the queue-storage tables autovacuum never +//! analyses: the two head tables, which stay below the autoanalyze threshold, +//! and the partitioned ring parents, which autovacuum skips entirely. +//! +//! The maintenance leader runs one pass on election and one every interval. +//! Each relation is analysed through `awa.analyze_queue_storage_relation` +//! (the `analyze_queue_storage_relation` schema patch, `SECURITY DEFINER`) in its own short transaction with a +//! bounded `lock_timeout` and `statement_timeout`, so the refresh never holds +//! locks across relations and cannot wait on ring prune for long. + +use crate::metrics::AwaMetrics; +use sqlx::PgPool; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, warn}; + +/// How long one relation's ANALYZE may wait for a lock before the pass moves +/// on. Ring prune holds `ACCESS EXCLUSIVE` on a slot's children only +/// briefly, so a wait this long means something else holds the table. +const LOCK_TIMEOUT: Duration = Duration::from_secs(2); + +/// Upper bound for one relation. A ring parent analyses in under a few +/// seconds even on a loaded host; this only stops a stuck statement. +const STATEMENT_TIMEOUT: Duration = Duration::from_secs(120); + +const HEAD_TABLES: [&str; 2] = ["queue_claim_heads", "queue_enqueue_heads"]; + +const SQLSTATE_INSUFFICIENT_PRIVILEGE: &str = "42501"; +const SQLSTATE_UNDEFINED_FUNCTION: &str = "42883"; +const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03"; + +/// Which kind of relation an analyse call covered; bounds metric cardinality. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RelationScope { + Heads, + RingParent, +} + +impl RelationScope { + pub(crate) fn as_str(self) -> &'static str { + match self { + RelationScope::Heads => "heads", + RelationScope::RingParent => "ring_parent", + } + } +} + +/// Result of analysing one relation. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum AnalyzeOutcome { + Ok, + /// The function owner (or the runtime role, if it cannot execute the + /// function) may not analyse the relation. + SkippedPrivilege, + /// `lock_timeout` expired waiting on a conflicting lock. + SkippedLock, + /// `awa.analyze_queue_storage_relation` is not installed (the schema + /// patch has not been applied). + Unavailable, + Failed, +} + +impl AnalyzeOutcome { + pub(crate) fn as_str(self) -> &'static str { + match self { + AnalyzeOutcome::Ok => "ok", + AnalyzeOutcome::SkippedPrivilege => "skipped_privilege", + AnalyzeOutcome::SkippedLock => "skipped_lock", + AnalyzeOutcome::Unavailable => "unavailable", + AnalyzeOutcome::Failed => "failed", + } + } +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub(crate) struct PassSummary { + pub(crate) ok: usize, + pub(crate) skipped_privilege: usize, + pub(crate) skipped_lock: usize, + pub(crate) unavailable: usize, + pub(crate) failed: usize, +} + +impl PassSummary { + fn record(&mut self, outcome: AnalyzeOutcome) { + match outcome { + AnalyzeOutcome::Ok => self.ok += 1, + AnalyzeOutcome::SkippedPrivilege => self.skipped_privilege += 1, + AnalyzeOutcome::SkippedLock => self.skipped_lock += 1, + AnalyzeOutcome::Unavailable => self.unavailable += 1, + AnalyzeOutcome::Failed => self.failed += 1, + } + } +} + +/// Refreshes planner statistics for one queue-storage schema. +pub(crate) struct StatisticsRefresher { + pool: PgPool, + schema: String, + metrics: AwaMetrics, + warned_privilege: AtomicBool, + warned_unavailable: AtomicBool, +} + +impl StatisticsRefresher { + pub(crate) fn new(pool: PgPool, schema: String, metrics: AwaMetrics) -> Self { + Self { + pool, + schema, + metrics, + warned_privilege: AtomicBool::new(false), + warned_unavailable: AtomicBool::new(false), + } + } + + /// Run a pass now and then every `interval` until `cancel` fires. + pub(crate) async fn run(&self, interval: Duration, cancel: CancellationToken) { + let mut timer = tokio::time::interval(interval); + timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tokio::select! { + _ = cancel.cancelled() => return, + _ = timer.tick() => { + tokio::select! { + _ = cancel.cancelled() => return, + _ = self.run_pass() => {} + } + } + } + } + } + + /// Analyse the head tables, then every partitioned table of the schema. + pub(crate) async fn run_pass(&self) -> PassSummary { + let started = Instant::now(); + let mut summary = PassSummary::default(); + + match awa_model::storage::queue_storage_schema_ready(&self.pool, &self.schema).await { + Ok(true) => {} + Ok(false) => { + debug!(schema = %self.schema, "Queue-storage schema is not prepared yet; skipping statistics refresh"); + return summary; + } + Err(err) => { + warn!(error = %err, schema = %self.schema, "Could not check the queue-storage schema for statistics refresh"); + self.record( + RelationScope::RingParent, + AnalyzeOutcome::Failed, + started.elapsed(), + ); + summary.record(AnalyzeOutcome::Failed); + return summary; + } + } + + let ring_parents = match self.ring_parents().await { + Ok(parents) => parents, + Err(err) => { + warn!(error = %err, schema = %self.schema, "Could not list ring tables for statistics refresh"); + self.record( + RelationScope::RingParent, + AnalyzeOutcome::Failed, + started.elapsed(), + ); + summary.record(AnalyzeOutcome::Failed); + Vec::new() + } + }; + + let targets = HEAD_TABLES + .iter() + .map(|name| (RelationScope::Heads, (*name).to_string())) + .chain( + ring_parents + .into_iter() + .map(|name| (RelationScope::RingParent, name)), + ); + + for (scope, relation) in targets { + let relation_started = Instant::now(); + let outcome = self.analyze_relation(&relation).await; + self.record(scope, outcome, relation_started.elapsed()); + summary.record(outcome); + } + + debug!( + schema = %self.schema, + elapsed_ms = started.elapsed().as_millis() as u64, + ok = summary.ok, + skipped_privilege = summary.skipped_privilege, + skipped_lock = summary.skipped_lock, + unavailable = summary.unavailable, + failed = summary.failed, + "Planner statistics refresh finished" + ); + summary + } + + fn record(&self, scope: RelationScope, outcome: AnalyzeOutcome, elapsed: Duration) { + self.metrics + .record_statistics_analyze(scope.as_str(), outcome.as_str(), elapsed); + } + + async fn ring_parents(&self) -> Result, sqlx::Error> { + sqlx::query_scalar( + r#" + SELECT class.relname::text + FROM pg_class AS class + JOIN pg_namespace AS namespace ON namespace.oid = class.relnamespace + WHERE namespace.nspname = $1 AND class.relkind = 'p' + ORDER BY class.relname + "#, + ) + .bind(&self.schema) + .fetch_all(&self.pool) + .await + } + + async fn analyze_relation(&self, relation: &str) -> AnalyzeOutcome { + match self.call_analyze_function(relation).await { + Ok(status) => match status.as_str() { + "analyzed" => AnalyzeOutcome::Ok, + "skipped_privilege" => { + self.warn_privilege_once( + relation, + "the owner of awa.analyze_queue_storage_relation() may not ANALYZE this table; \ + run migrations as the role that owns the queue-storage tables, or grant the \ + function owner MAINTAIN on them (PostgreSQL 17+)", + ); + AnalyzeOutcome::SkippedPrivilege + } + // The relation is not part of this schema's shape; nothing to analyse. + "missing" => AnalyzeOutcome::Ok, + other => { + warn!( + status = other, + relation, "Unexpected statistics refresh status" + ); + AnalyzeOutcome::Failed + } + }, + Err(err) => self.classify_error(relation, &err), + } + } + + async fn call_analyze_function(&self, relation: &str) -> Result { + let mut tx = self.pool.begin().await?; + sqlx::query( + "SELECT set_config('lock_timeout', $1, true), set_config('statement_timeout', $2, true)", + ) + .bind(format!("{}ms", LOCK_TIMEOUT.as_millis())) + .bind(format!("{}ms", STATEMENT_TIMEOUT.as_millis())) + .execute(&mut *tx) + .await?; + let status: String = + sqlx::query_scalar("SELECT awa.analyze_queue_storage_relation($1, $2)") + .bind(&self.schema) + .bind(relation) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok(status) + } + + fn classify_error(&self, relation: &str, err: &sqlx::Error) -> AnalyzeOutcome { + let code = match err { + sqlx::Error::Database(db_err) => db_err.code().map(|code| code.into_owned()), + _ => None, + }; + match code.as_deref() { + Some(SQLSTATE_LOCK_NOT_AVAILABLE) => { + debug!( + relation, + "Statistics refresh timed out waiting for a lock; retrying next interval" + ); + AnalyzeOutcome::SkippedLock + } + Some(SQLSTATE_UNDEFINED_FUNCTION) => { + if !self.warned_unavailable.swap(true, Ordering::Relaxed) { + warn!( + schema = %self.schema, + "Planner statistics refresh is disabled: awa.analyze_queue_storage_relation() \ + is missing. Run `awa migrate` to apply the analyze_queue_storage_relation schema patch" + ); + } + AnalyzeOutcome::Unavailable + } + Some(SQLSTATE_INSUFFICIENT_PRIVILEGE) => { + self.warn_privilege_once( + relation, + "the runtime role may not execute awa.analyze_queue_storage_relation(); \ + GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) to it", + ); + AnalyzeOutcome::SkippedPrivilege + } + _ => { + warn!(relation, error = %err, "Planner statistics refresh failed for relation"); + AnalyzeOutcome::Failed + } + } + } + + fn warn_privilege_once(&self, relation: &str, missing: &str) { + if self.warned_privilege.swap(true, Ordering::Relaxed) { + return; + } + warn!( + schema = %self.schema, + relation, + "Planner statistics refresh skipped: {missing}. Set the statistics refresh \ + interval to zero to disable the refresh" + ); + } +} + +/// Stops the refresh task when the maintenance leader loses leadership or +/// shuts down. +pub(crate) struct RefreshGuard { + cancel: CancellationToken, +} + +impl RefreshGuard { + pub(crate) fn spawn( + refresher: StatisticsRefresher, + interval: Duration, + parent: &CancellationToken, + ) -> Self { + let cancel = parent.child_token(); + let task_cancel = cancel.clone(); + tokio::spawn(async move { + debug!( + schema = %refresher.schema, + interval_s = interval.as_secs(), + "Planner statistics refresh started" + ); + refresher.run(interval, task_cancel).await; + }); + Self { cancel } + } +} + +impl Drop for RefreshGuard { + fn drop(&mut self) { + self.cancel.cancel(); + } +} diff --git a/awa/tests/statistics_refresh_test.rs b/awa/tests/statistics_refresh_test.rs new file mode 100644 index 000000000..0c1946b12 --- /dev/null +++ b/awa/tests/statistics_refresh_test.rs @@ -0,0 +1,563 @@ +//! The maintenance leader refreshes planner statistics for the queue-storage +//! tables autovacuum never analyses (the head tables and the partitioned ring +//! parents) on election and every interval, without delaying ring prune. + +use awa::model::{migrations, PruneOutcome, QueueStorage, QueueStorageConfig}; +use awa::{Client, JobContext, JobError, JobResult, QueueConfig, Worker}; +use opentelemetry::KeyValue; +use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; +use opentelemetry_sdk::metrics::{InMemoryMetricExporter, SdkMeterProvider}; +use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; +use sqlx::PgPool; +use std::str::FromStr; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::{Duration, Instant}; + +const ROLE_PASSWORD: &str = "statistics_refresh_test"; +const RING_PARENTS: [&str; 13] = [ + "done_entries", + "lease_claim_batches", + "lease_claim_closure_batches", + "lease_claim_closures", + "lease_claims", + "leases", + "queue_terminal_count_deltas", + "ready_claim_attempt_batches", + "ready_entries", + "ready_segments", + "ready_tombstones", + "receipt_completion_batches", + "receipt_completion_tombstones", +]; + +static DATABASE_COUNTER: AtomicU32 = AtomicU32::new(0); + +async fn run_sql(pool: &PgPool, sql: String) -> Result<(), sqlx::Error> { + sqlx::raw_sql(&sql).execute(pool).await.map(|_| ()) +} + +fn base_database_url() -> String { + std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string()) +} + +struct StatisticsProbeWorker; + +#[async_trait::async_trait] +impl Worker for StatisticsProbeWorker { + fn kind(&self) -> &'static str { + "statistics_probe_job" + } + + async fn perform(&self, _ctx: &JobContext) -> Result { + Ok(JobResult::Completed) + } +} + +/// A throwaway database with the awa schema migrated, optionally owned by a +/// non-superuser role the way a production schema owner is. +struct TestDatabase { + admin: PgPool, + name: String, + roles: Vec, + options: PgConnectOptions, + /// Connects as the schema owner (or the superuser when no owner role). + owner_pool: PgPool, +} + +impl TestDatabase { + async fn create(owner_role: Option<&str>, extra_roles: &[&str]) -> Self { + let admin = PgPoolOptions::new() + .max_connections(2) + .connect(&base_database_url()) + .await + .expect("connect admin pool"); + let name = format!( + "awa_stats_{}_{}", + std::process::id(), + DATABASE_COUNTER.fetch_add(1, Ordering::SeqCst) + ); + let mut roles = Vec::new(); + for role in owner_role.iter().chain(extra_roles.iter()) { + run_sql( + &admin, + format!( + "DROP ROLE IF EXISTS {role}; CREATE ROLE {role} LOGIN PASSWORD '{ROLE_PASSWORD}'" + ), + ) + .await + .expect("create role"); + roles.push((*role).to_string()); + } + let owner_clause = owner_role + .map(|role| format!(" OWNER {role}")) + .unwrap_or_default(); + run_sql(&admin, format!("CREATE DATABASE {name}{owner_clause}")) + .await + .expect("create database"); + let options = PgConnectOptions::from_str(&base_database_url()) + .expect("parse DATABASE_URL") + .database(&name); + let owner_options = match owner_role { + Some(role) => options.clone().username(role).password(ROLE_PASSWORD), + None => options.clone(), + }; + let owner_pool = PgPoolOptions::new() + .max_connections(8) + .connect_with(owner_options) + .await + .expect("connect owner pool"); + migrations::run(&owner_pool).await.expect("run migrations"); + Self { + admin, + name, + roles, + options, + owner_pool, + } + } + + /// A pool for a role other than the schema owner, with the runtime grants + /// from the database-roles guide. + async fn runtime_pool(&self, role: &str) -> PgPool { + run_sql( + &self.owner_pool, + format!( + r#" + GRANT USAGE ON SCHEMA awa TO {role}; + GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON ALL TABLES IN SCHEMA awa TO {role}; + GRANT USAGE, SELECT, UPDATE ON ALL SEQUENCES IN SCHEMA awa TO {role}; + GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa TO {role}; + REVOKE EXECUTE ON FUNCTION + awa.install_queue_storage_substrate(TEXT, INT, INT, INT, BOOLEAN) FROM {role}; + "# + ), + ) + .await + .expect("grant runtime privileges"); + PgPoolOptions::new() + .max_connections(8) + .connect_with(self.options.clone().username(role).password(ROLE_PASSWORD)) + .await + .expect("connect runtime pool") + } + + async fn drop(self) { + self.owner_pool.close().await; + let _ = run_sql( + &self.admin, + format!("DROP DATABASE IF EXISTS {} WITH (FORCE)", self.name), + ) + .await; + for role in &self.roles { + let _ = run_sql(&self.admin, format!("DROP ROLE IF EXISTS {role}")).await; + } + self.admin.close().await; + } +} + +async fn wait_for(what: &str, timeout: Duration, mut check: F) +where + F: FnMut() -> Fut, + Fut: std::future::Future, +{ + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if check().await { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!("timed out waiting for {what}"); +} + +async fn stats_rows(pool: &PgPool, schema: &str, table: &str, inherited_only: bool) -> i64 { + sqlx::query_scalar( + "SELECT count(*) FROM pg_stats + WHERE schemaname = $1 AND tablename = $2 AND (inherited OR NOT $3)", + ) + .bind(schema) + .bind(table) + .bind(inherited_only) + .fetch_one(pool) + .await + .expect("read pg_stats") +} + +async fn parent_reltuples(pool: &PgPool, schema: &str, table: &str) -> f32 { + sqlx::query_scalar("SELECT reltuples FROM pg_class WHERE oid = to_regclass($1)") + .bind(format!("{schema}.{table}")) + .fetch_one(pool) + .await + .expect("read reltuples") +} + +async fn analyze_count(pool: &PgPool, schema: &str, table: &str) -> i64 { + sqlx::query_scalar( + "SELECT COALESCE(analyze_count, 0)::bigint FROM pg_stat_user_tables + WHERE schemaname = $1 AND relname = $2", + ) + .bind(schema) + .bind(table) + .fetch_one(pool) + .await + .expect("read analyze_count") +} + +fn probe_client( + pool: PgPool, + statistics_interval: Duration, + storage: Option, +) -> Client { + let mut builder = Client::builder(pool) + .queue( + "statistics_probe", + QueueConfig { + max_workers: 1, + poll_interval: Duration::from_millis(100), + ..QueueConfig::default() + }, + ) + .register_worker(StatisticsProbeWorker) + .leader_election_interval(Duration::from_millis(100)) + .leader_check_interval(Duration::from_millis(100)) + .statistics_refresh_interval(statistics_interval); + if let Some(config) = storage { + builder = builder + .queue_storage(config, Duration::from_secs(600), Duration::from_secs(600)) + .claim_rotate_interval(Duration::from_secs(600)); + } + builder.build().expect("build client") +} + +async fn seed_ready_rows(store: &QueueStorage, pool: &PgPool, rows_per_queue: i64) { + for queue in ["stats_a", "stats_b", "stats_c"] { + store + .enqueue_batch(pool, queue, 2, rows_per_queue) + .await + .expect("enqueue rows"); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn leader_refreshes_statistics_on_election_and_every_interval() { + let database = TestDatabase::create(None, &[]).await; + let pool = &database.owner_pool; + let store = QueueStorage::new(QueueStorageConfig::default()).expect("queue storage config"); + seed_ready_rows(&store, pool, 200).await; + + assert_eq!(stats_rows(pool, "awa", "queue_claim_heads", false).await, 0); + assert_eq!( + stats_rows(pool, "awa", "queue_enqueue_heads", false).await, + 0 + ); + assert_eq!(stats_rows(pool, "awa", "ready_entries", true).await, 0); + assert!(parent_reltuples(pool, "awa", "ready_entries").await < 0.0); + + let client = probe_client(pool.clone(), Duration::from_secs(1), None); + client.start().await.expect("start client"); + + wait_for( + "first statistics pass on election", + Duration::from_secs(30), + || async { + stats_rows(pool, "awa", "queue_claim_heads", false).await > 0 + && stats_rows(pool, "awa", "queue_enqueue_heads", false).await > 0 + && stats_rows(pool, "awa", "ready_entries", true).await > 0 + }, + ) + .await; + let partitioned: Vec = sqlx::query_scalar( + "SELECT relname::text FROM pg_class + WHERE relnamespace = 'awa'::regnamespace AND relkind = 'p' ORDER BY relname", + ) + .fetch_all(pool) + .await + .expect("list partitioned tables"); + assert_eq!( + partitioned, RING_PARENTS, + "the refresh covers every ring parent" + ); + wait_for( + "every ring parent analysed", + Duration::from_secs(30), + || async { + for parent in RING_PARENTS { + if analyze_count(pool, "awa", parent).await < 1 { + return false; + } + } + true + }, + ) + .await; + let first_tuples = parent_reltuples(pool, "awa", "ready_entries").await; + assert!( + (550.0..=650.0).contains(&first_tuples), + "parent reltuples after the first pass: {first_tuples}" + ); + + seed_ready_rows(&store, pool, 2_000).await; + wait_for( + "a later pass to see the new rows", + Duration::from_secs(30), + || async { parent_reltuples(pool, "awa", "ready_entries").await > 5_000.0 }, + ) + .await; + assert!(analyze_count(pool, "awa", "ready_entries").await >= 2); + assert!(analyze_count(pool, "awa", "queue_claim_heads").await >= 2); + + client.shutdown(Duration::from_secs(5)).await; + database.drop().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn zero_interval_disables_the_refresh() { + let database = TestDatabase::create(None, &[]).await; + let pool = &database.owner_pool; + let store = QueueStorage::new(QueueStorageConfig::default()).expect("queue storage config"); + seed_ready_rows(&store, pool, 100).await; + + let client = probe_client(pool.clone(), Duration::ZERO, None); + client.start().await.expect("start client"); + wait_for("leadership", Duration::from_secs(10), || async { + client.health_check().await.maintenance_alive + }) + .await; + tokio::time::sleep(Duration::from_secs(3)).await; + + assert_eq!(stats_rows(pool, "awa", "queue_claim_heads", false).await, 0); + assert_eq!(stats_rows(pool, "awa", "ready_entries", true).await, 0); + + client.shutdown(Duration::from_secs(5)).await; + database.drop().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn non_owner_runtime_role_refreshes_statistics() { + let suffix = uuid::Uuid::new_v4().simple().to_string(); + let owner = format!("stats_owner_{suffix}"); + let runtime = format!("stats_runtime_{suffix}"); + let database = TestDatabase::create(Some(&owner), &[&runtime]).await; + let store = QueueStorage::new(QueueStorageConfig::default()).expect("queue storage config"); + seed_ready_rows(&store, &database.owner_pool, 200).await; + let runtime_pool = database.runtime_pool(&runtime).await; + + let is_superuser_or_owner: bool = sqlx::query_scalar( + "SELECT rolsuper OR pg_has_role(current_user, 'pg_maintain', 'USAGE') + OR has_table_privilege('awa.queue_claim_heads', 'MAINTAIN') + OR (SELECT relowner = (SELECT oid FROM pg_roles WHERE rolname = current_user) + FROM pg_class WHERE oid = 'awa.queue_claim_heads'::regclass) + FROM pg_roles WHERE rolname = current_user", + ) + .fetch_one(&runtime_pool) + .await + .expect("check runtime role privileges"); + assert!( + !is_superuser_or_owner, + "the runtime role must not be able to analyse the tables itself" + ); + + let client = probe_client(runtime_pool.clone(), Duration::from_secs(1), None); + client + .start() + .await + .expect("start client as the runtime role"); + wait_for( + "statistics written through a non-owner role", + Duration::from_secs(30), + || async { + stats_rows(&database.owner_pool, "awa", "queue_claim_heads", false).await > 0 + && stats_rows(&database.owner_pool, "awa", "ready_entries", true).await > 0 + }, + ) + .await; + + client.shutdown(Duration::from_secs(5)).await; + runtime_pool.close().await; + database.drop().await; +} + +fn sum_by_outcome( + resource_metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics], + outcome: &str, +) -> u64 { + let mut total = 0; + for resource in resource_metrics { + for scope in resource.scope_metrics() { + for metric in scope.metrics() { + if metric.name() != "awa.maintenance.analyze.attempts" { + continue; + } + if let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() { + total += sum + .data_points() + .filter(|point| { + point.attributes().any(|KeyValue { key, value, .. }| { + key.as_str() == "awa.analyze.outcome" && value.as_str() == outcome + }) + }) + .map(|point| point.value()) + .sum::(); + } + } + } + } + total +} + +/// A refresh that cannot run is recorded and leaves the runtime healthy: the +/// helper's owner may not analyse the tables, or the schema predates the +/// helper. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unrunnable_refresh_is_recorded_and_does_not_fail_the_runtime() { + let exporter = InMemoryMetricExporter::default(); + let provider = SdkMeterProvider::builder() + .with_periodic_exporter(exporter.clone()) + .build(); + opentelemetry::global::set_meter_provider(provider.clone()); + + let suffix = uuid::Uuid::new_v4().simple().to_string(); + let owner = format!("stats_owner_{suffix}"); + let helper = format!("stats_helper_{suffix}"); + let database = TestDatabase::create(Some(&owner), &[&helper]).await; + let store = QueueStorage::new(QueueStorageConfig::default()).expect("queue storage config"); + seed_ready_rows(&store, &database.owner_pool, 50).await; + + // The function now runs as a role that owns nothing and holds no MAINTAIN. + run_sql(&database.admin, format!("GRANT {helper} TO {owner}")) + .await + .expect("let the schema owner hand over the function"); + run_sql( + &database.owner_pool, + format!( + "GRANT USAGE, CREATE ON SCHEMA awa TO {helper}; + ALTER FUNCTION awa.analyze_queue_storage_relation(text, text) OWNER TO {helper}" + ), + ) + .await + .expect("hand the helper to an unprivileged role"); + + let client = probe_client( + database.owner_pool.clone(), + Duration::from_millis(200), + None, + ); + client.start().await.expect("start client"); + wait_for("a skipped pass", Duration::from_secs(30), || async { + provider.force_flush().expect("flush metrics"); + let metrics = exporter.get_finished_metrics().expect("read metrics"); + sum_by_outcome(&metrics, "skipped_privilege") > 0 + }) + .await; + assert_eq!( + stats_rows(&database.owner_pool, "awa", "queue_claim_heads", false).await, + 0, + "a skipped pass writes no statistics" + ); + assert!(client.health_check().await.healthy); + client.shutdown(Duration::from_secs(5)).await; + + // A schema that predates the helper skips the same way. + sqlx::raw_sql("DROP FUNCTION awa.analyze_queue_storage_relation(text, text)") + .execute(&database.owner_pool) + .await + .expect("drop the helper"); + let client = probe_client( + database.owner_pool.clone(), + Duration::from_millis(200), + None, + ); + client + .start() + .await + .expect("start client without the helper"); + wait_for("an unavailable pass", Duration::from_secs(30), || async { + provider.force_flush().expect("flush metrics"); + let metrics = exporter.get_finished_metrics().expect("read metrics"); + sum_by_outcome(&metrics, "unavailable") > 0 + }) + .await; + assert!(client.health_check().await.healthy); + client.shutdown(Duration::from_secs(5)).await; + + provider.force_flush().expect("flush metrics"); + let metrics = exporter.get_finished_metrics().expect("read metrics"); + assert_eq!(sum_by_outcome(&metrics, "failed"), 0); + database.drop().await; +} + +/// Back-to-back statistics passes (a far higher duty cycle than the default +/// five-minute interval) run while rotation and prune cycle the ring: prune +/// keeps reclaiming slots and only defers an attempt that lands while ANALYZE +/// is sampling the slot's tables. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn statistics_refresh_does_not_stop_prune_reclaiming_slots() { + let database = TestDatabase::create(None, &[]).await; + let pool = &database.owner_pool; + let schema = "awa_stats_ring"; + let config = QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + claim_slot_count: 2, + lease_claim_receipts: false, + ..Default::default() + }; + let store = QueueStorage::new(config.clone()).expect("queue storage config"); + store.prepare_schema(pool).await.expect("prepare schema"); + + let queue = "statistics_ring"; + let analyze_before = analyze_count(pool, schema, "ready_entries").await; + let client = probe_client(pool.clone(), Duration::from_millis(1), Some(config)); + client.start().await.expect("start client"); + + let (mut pruned, mut deferred) = (0_u32, 0_u32); + let started = Instant::now(); + for _ in 0..60 { + store + .enqueue_batch(pool, queue, 1, 600) + .await + .expect("enqueue"); + loop { + let claims = store.claim_batch(pool, queue, 600).await.expect("claim"); + if claims.is_empty() { + break; + } + store.complete_batch(pool, &claims).await.expect("complete"); + } + store.rotate(pool).await.expect("rotate"); + match store + .prune_oldest(pool, Duration::ZERO) + .await + .expect("prune never errors under ANALYZE") + { + PruneOutcome::Pruned { .. } => pruned += 1, + PruneOutcome::Blocked { .. } => deferred += 1, + PruneOutcome::Noop | PruneOutcome::SkippedActive { .. } => {} + } + if started.elapsed() > Duration::from_secs(60) { + break; + } + } + let passes = analyze_count(pool, schema, "ready_entries").await - analyze_before; + eprintln!("prune under ANALYZE: pruned={pruned} deferred={deferred} analyze_passes={passes}"); + + client.shutdown(Duration::from_secs(5)).await; + assert!( + passes >= 3, + "the refresh ran {passes} times during the loop" + ); + assert!(pruned >= 20, "prune reclaimed {pruned} slots"); + + // With the refresh stopped, the oldest sealed slot reclaims immediately. + store.rotate(pool).await.expect("rotate"); + assert!(!matches!( + store + .prune_oldest(pool, Duration::ZERO) + .await + .expect("prune"), + PruneOutcome::Blocked { .. } + )); + database.drop().await; +} diff --git a/docs/architecture.md b/docs/architecture.md index 06b967bc3..b0c620ffe 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -212,6 +212,9 @@ Queue storage has three independent rings, each advanced by the elected maintena The maintenance tick for each ring is deliberately small: attempt one rotate, then attempt one prune. If a partition is busy, blocked by a lock, current, or still live, the tick records a skipped/blocked outcome and tries again on a future interval. +The leader also refreshes planner statistics on the two head tables and the ring parents when it is elected and every five minutes ([Planner statistics](configuration.md#planner-statistics)): autovacuum never analyses partitioned tables or reaches its threshold on the head tables. It runs beside the ring ticks rather than in them, so a slow `ANALYZE` never delays a rotate or prune tick. + + The common safety pattern is: 1. Lock the ring-state row with `FOR UPDATE`. diff --git a/docs/configuration.md b/docs/configuration.md index 868dfb3cf..7e39bf2e8 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -497,6 +497,7 @@ DLQ rows are different: `dlq_entries` is a separate hold table and the maintenan | `cleanup_batch_size` | `1000` | Max rows deleted per pass. Raise for very high throughput; lower if you want gentler IO. | | `dlq_cleanup_batch_size` | `1000` | DLQ-specific batch size. | | `ClientBuilder::terminal_count_rollup_interval` (Rust) / `terminal_count_rollup_interval_ms` (Python kwarg) | `30s` | Queue storage only: how often maintenance folds pending `done_entries` terminal-count deltas for sealed queue slots into compact live counters. Exact reads include pending deltas and retained compact receipt batches, so this affects compaction/WAL pressure rather than correctness. | +| `ClientBuilder::statistics_refresh_interval` (Rust) / `statistics_refresh_interval_ms` (Python kwarg) | `5m` | Queue storage only: how often the maintenance leader runs `ANALYZE` on the two head tables and the partitioned ring parents, which autovacuum never analyses; also run once when a runtime becomes leader. `0` disables it. See [Planner statistics](#planner-statistics). | ## Queue storage tuning @@ -520,6 +521,16 @@ The maintenance leader folds pending `done_entries` deltas from sealed queue slo Raising the interval reduces folded-counter churn but leaves more pending delta rows for exact reads to sum. Lowering it folds counters sooner, which can help if exact-count reads are frequent and the `done_entries` delta ledger is growing faster than maintenance drains it. Under long reader transactions, lowering the interval does not force rollup; the useful fix is to release the pinned snapshot or let queue prune reclaim the whole sealed segment. Compact receipt successes are already retained as compact batch rows, so this interval does not change their hot-path write count. +### Planner statistics + +Autovacuum does not analyse partitioned tables, so the ring parents (`ready_entries`, `done_entries`, `leases`, and the other ten partitioned tables of the schema) have no inherited statistics. It also never reaches its analyze threshold on `queue_claim_heads` and `queue_enqueue_heads`, which hold a few rows per lane, so the planner prices joins against them at its defaults. Admin and metrics queries that join retained ready rows to the heads tables then pick plans for a few hundred rows while reading hundreds of thousands. + +The maintenance leader runs `ANALYZE` on both when it is elected and every `statistics_refresh_interval` (default `5m`). One pass takes well under a second on a small ring and a few seconds on a ring with millions of retained rows. It has no measurable effect on claim or completion throughput. + +- **Locks.** Each relation is analysed in its own transaction with a 2 s `lock_timeout` and a 120 s `statement_timeout`. Children that ring prune holds are skipped (`SKIP_LOCKED`). Prune never waits for the refresh: its own 10 ms lock timeout applies, so an attempt that lands while `ANALYZE` samples the oldest slot's tables returns `blocked` and retries on its usual backoff. `ANALYZE` of a partitioned table samples every child, so that window cannot be avoided. +- **Privileges.** The refresh calls `awa.analyze_queue_storage_relation(schema, relation)` (a schema patch applied by `awa migrate`), a `SECURITY DEFINER` function owned by the role that ran it, so the runtime role needs no `MAINTAIN` grant and does not have to own the tables. If the function is missing (run `awa migrate`) or its owner may not analyse the tables, the leader logs one warning that names what is missing and skips the pass; job processing is unaffected. +- **Metrics.** `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration`, by `awa.analyze.scope` (`heads`, `ring_parent`) and `awa.analyze.outcome` (`ok`, `skipped_privilege`, `skipped_lock`, `unavailable`, `failed`). A sustained `skipped_lock` rate means something else holds the tables; any `skipped_privilege` or `unavailable` means the refresh is not running. + ### Rust ```rust diff --git a/docs/grafana/README.md b/docs/grafana/README.md index 823102462..dde57d5cd 100644 --- a/docs/grafana/README.md +++ b/docs/grafana/README.md @@ -157,6 +157,8 @@ All metrics use the `awa` OTel meter name and are exported via OTLP to your conf | `awa.maintenance.promote_batch_size` | Histogram | state | Jobs promoted | | `awa.maintenance.promote_duration` | Histogram (s) | state | Promotion time | | `awa.maintenance.rescues` | Counter | rescue_kind | Jobs rescued | +| `awa.maintenance.analyze.attempts` | Counter | scope, outcome | Planner-statistics `ANALYZE` calls on the head tables and ring parents (`ok`, `skipped_privilege`, `skipped_lock`, `unavailable`, `failed`) | +| `awa.maintenance.analyze.duration` | Histogram (s) | scope, outcome | Duration of one planner-statistics `ANALYZE` call | ### Heartbeat diff --git a/docs/security.md b/docs/security.md index a146635bb..7bc07cb29 100644 --- a/docs/security.md +++ b/docs/security.md @@ -224,6 +224,8 @@ The runtime also directly upserts into the descriptor catalogs on startup and on These writes are not trigger-driven; they come from `ClientBuilder::build()` / `AsyncClient.start()` and from the snapshot reporter. The broad grant already covers them. +The maintenance leader refreshes planner statistics for the queue-storage head tables and ring parents (see [Planner statistics](configuration.md#planner-statistics)). `ANALYZE` skips a table its caller may not analyse, so a runtime that does not own the tables calls `awa.analyze_queue_storage_relation(schema, relation)`, a `SECURITY DEFINER` function installed by the `analyze_queue_storage_relation` schema patch under the migrator role and executable by `PUBLIC`. The runtime needs no `MAINTAIN` grant. The function only analyses a head table or partitioned table of a queue-storage schema that is owned by the owner of the `awa` schema, which is why migrations must run with the owner role as the effective role (step 2 above). A function owned by a role that may not analyse the tables makes every pass report `skipped_privilege` and logs one warning. + Other PostgreSQL features used at runtime: | Feature | Purpose | Privilege needed | From cb52274b258935609fa10f61e92f641ee6ae4117 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 10:10:20 +1300 Subject: [PATCH 2/6] test(worker): wait for the last ANALYZE before asserting prune is unblocked (0.6) --- awa/tests/statistics_refresh_test.rs | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/awa/tests/statistics_refresh_test.rs b/awa/tests/statistics_refresh_test.rs index 0c1946b12..5679c30f5 100644 --- a/awa/tests/statistics_refresh_test.rs +++ b/awa/tests/statistics_refresh_test.rs @@ -550,7 +550,24 @@ async fn statistics_refresh_does_not_stop_prune_reclaiming_slots() { ); assert!(pruned >= 20, "prune reclaimed {pruned} slots"); - // With the refresh stopped, the oldest sealed slot reclaims immediately. + // Shutdown can leave one ANALYZE statement finishing server-side; once it + // is gone the oldest sealed slot reclaims without being blocked. + wait_for( + "the last ANALYZE to finish", + Duration::from_secs(30), + || async { + let running: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_stat_activity + WHERE state = 'active' AND query LIKE '%analyze_queue_storage_relation%' + AND pid <> pg_backend_pid()", + ) + .fetch_one(pool) + .await + .expect("read pg_stat_activity"); + running == 0 + }, + ) + .await; store.rotate(pool).await.expect("rotate"); assert!(!matches!( store From c22d35b70f6419420b9bafe32b71acef0f194e20 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 11:53:01 +1300 Subject: [PATCH 3/6] fix(worker): restrict statistics refresh to ring parents, cover standby leaders, report lock skips (0.6) - Analyse only the known ring parents, not every partitioned table in the schema, so a substrate sharing a schema with application tables never analyses them. - A leader that resolved canonical at start but holds the queue-storage runtime as standby refreshes the queue-storage schema once routing flips to it, checked on every pass. - Take the parent's SHARE UPDATE EXCLUSIVE lock before ANALYZE (SKIP_LOCKED), so a conflicting lock on the parent fails the call with lock_not_available (recorded as skipped_lock) instead of skipping every partition and reporting analyzed. Correct the documentation of what SKIP_LOCKED covers. - Reference the pull request, not an issue number, in the patch header. - Tests: refresh on election before the first interval, both rejection guards of the helper from the runtime role, and a locked parent. --- CHANGELOG.md | 5 +- .../analyze_queue_storage_relation.sql | 19 +++- awa-worker/src/maintenance.rs | 6 +- awa-worker/src/statistics.rs | 52 ++++++++++- awa/tests/statistics_refresh_test.rs | 88 +++++++++++++++++++ docs/configuration.md | 2 +- 6 files changed, 163 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 069c34283..d3d265447 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,8 +16,9 @@ Notable changes between releases. Detailed migration notes for storage transitio leader now runs `ANALYZE` on them when it is elected and then every five minutes. `ClientBuilder::statistics_refresh_interval` (Python `statistics_refresh_interval_ms`) changes the cadence; zero disables it. It - analyses one relation per transaction with a 2 s `lock_timeout`, skips - children that prune holds, and never makes prune wait: a prune attempt that + analyses one relation per transaction with a 2 s `lock_timeout` (a call that + times out waiting for a lock is retried next interval and recorded as + `skipped_lock`) and never makes prune wait: a prune attempt that lands while ANALYZE samples the oldest slot's tables returns `blocked` and retries on its normal backoff. Results are in `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration` diff --git a/awa-model/migrations/patches/analyze_queue_storage_relation.sql b/awa-model/migrations/patches/analyze_queue_storage_relation.sql index 67a377369..e2b3d50af 100644 --- a/awa-model/migrations/patches/analyze_queue_storage_relation.sql +++ b/awa-model/migrations/patches/analyze_queue_storage_relation.sql @@ -1,5 +1,5 @@ -- Schema patch analyze_queue_storage_relation: planner statistics helper for --- the queue-storage tables autovacuum never analyses (#630). +-- the queue-storage tables autovacuum never analyses (#631). -- Counterpart of awa 0.7 migration v053, minus the schema_version row; v053 -- replaces this body with one that checks the schema through -- awa.queue_storage_schemas(), which 0.6 does not have. @@ -19,11 +19,19 @@ -- SECURITY DEFINER function owned by the role that applies the patch, which -- owns the tables. -- --- The function analyses one relation per call; the caller runs each call as +-- The function analyses one relation per call and the caller runs each call as -- its own transaction, so the locks ANALYZE takes on a partitioned table's -- children are held for that table rather than for the whole refresh. --- SKIP_LOCKED skips a child that ring prune holds ACCESS EXCLUSIVE instead of --- waiting for it and aborting the statistics gathered so far. +-- +-- Locking on a partitioned table, per the ANALYZE documentation: SKIP_LOCKED +-- applies when ANALYZE starts work on a relation, so a child that ring prune +-- holds ACCESS EXCLUSIVE is skipped during the per-child analysis, but +-- sampling the children for the parent's statistics can still wait for it. +-- That wait is bounded by the caller's lock_timeout and fails the call with +-- SQLSTATE 55P03, which the caller records as skipped_lock. SKIP_LOCKED also +-- skips every partition silently when the parent itself is locked, so the +-- function takes the parent's SHARE UPDATE EXCLUSIVE lock first and lets that +-- wait fail instead. -- -- Definer boundary: only a head table or a partitioned table of a queue-storage -- schema (the AWA-specific claim_ready_runtime sentinel) that is owned by the @@ -105,6 +113,9 @@ BEGIN END IF; IF v_relkind = 'p' THEN + EXECUTE format( + 'LOCK TABLE ONLY %I.%I IN SHARE UPDATE EXCLUSIVE MODE', p_schema, p_relation + ); EXECUTE format('ANALYZE (SKIP_LOCKED) %I.%I', p_schema, p_relation); ELSE EXECUTE format('ANALYZE %I.%I', p_schema, p_relation); diff --git a/awa-worker/src/maintenance.rs b/awa-worker/src/maintenance.rs index 4934e7e23..7371c59c1 100644 --- a/awa-worker/src/maintenance.rs +++ b/awa-worker/src/maintenance.rs @@ -1267,12 +1267,16 @@ impl MaintenanceService { if self.statistics_refresh_interval.is_zero() { return None; } - let runtime = self.storage.queue_storage()?; + let (runtime, standby) = match self.storage.queue_storage() { + Some(runtime) => (runtime, false), + None => (self.standby_queue_storage.as_ref()?, true), + }; Some(RefreshGuard::spawn( StatisticsRefresher::new( self.pool.clone(), runtime.store.schema().to_string(), self.metrics.clone(), + standby, ), self.statistics_refresh_interval, &self.cancel, diff --git a/awa-worker/src/statistics.rs b/awa-worker/src/statistics.rs index 04792cf9c..cfc97ef57 100644 --- a/awa-worker/src/statistics.rs +++ b/awa-worker/src/statistics.rs @@ -26,6 +26,25 @@ const STATEMENT_TIMEOUT: Duration = Duration::from_secs(120); const HEAD_TABLES: [&str; 2] = ["queue_claim_heads", "queue_enqueue_heads"]; +/// The partitioned ring parents a queue-storage substrate installs. A schema +/// can hold other partitioned tables (a substrate in an application schema), +/// which this refresh must never analyse. +const RING_PARENTS: [&str; 13] = [ + "done_entries", + "lease_claim_batches", + "lease_claim_closure_batches", + "lease_claim_closures", + "lease_claims", + "leases", + "queue_terminal_count_deltas", + "ready_claim_attempt_batches", + "ready_entries", + "ready_segments", + "ready_tombstones", + "receipt_completion_batches", + "receipt_completion_tombstones", +]; + const SQLSTATE_INSUFFICIENT_PRIVILEGE: &str = "42501"; const SQLSTATE_UNDEFINED_FUNCTION: &str = "42883"; const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03"; @@ -99,16 +118,26 @@ pub(crate) struct StatisticsRefresher { pool: PgPool, schema: String, metrics: AwaMetrics, + /// Set for a leader whose own storage is canonical and whose queue-storage + /// runtime is only a standby: a pass runs only once routing has flipped to + /// `schema`. + require_active_schema: bool, warned_privilege: AtomicBool, warned_unavailable: AtomicBool, } impl StatisticsRefresher { - pub(crate) fn new(pool: PgPool, schema: String, metrics: AwaMetrics) -> Self { + pub(crate) fn new( + pool: PgPool, + schema: String, + metrics: AwaMetrics, + require_active_schema: bool, + ) -> Self { Self { pool, schema, metrics, + require_active_schema, warned_privilege: AtomicBool::new(false), warned_unavailable: AtomicBool::new(false), } @@ -136,6 +165,25 @@ impl StatisticsRefresher { let started = Instant::now(); let mut summary = PassSummary::default(); + if self.require_active_schema { + match sqlx::query_scalar::<_, Option>( + "SELECT awa.active_queue_storage_schema()", + ) + .fetch_one(&self.pool) + .await + { + Ok(active) if active.as_deref() == Some(self.schema.as_str()) => {} + Ok(_) => { + debug!(schema = %self.schema, "Routing has not flipped to this schema yet; skipping statistics refresh"); + return summary; + } + Err(err) => { + warn!(error = %err, "Could not resolve the active queue-storage schema for statistics refresh"); + return summary; + } + } + } + match awa_model::storage::queue_storage_schema_ready(&self.pool, &self.schema).await { Ok(true) => {} Ok(false) => { @@ -209,10 +257,12 @@ impl StatisticsRefresher { FROM pg_class AS class JOIN pg_namespace AS namespace ON namespace.oid = class.relnamespace WHERE namespace.nspname = $1 AND class.relkind = 'p' + AND class.relname = ANY($2) ORDER BY class.relname "#, ) .bind(&self.schema) + .bind(&RING_PARENTS[..]) .fetch_all(&self.pool) .await } diff --git a/awa/tests/statistics_refresh_test.rs b/awa/tests/statistics_refresh_test.rs index 5679c30f5..7322639a0 100644 --- a/awa/tests/statistics_refresh_test.rs +++ b/awa/tests/statistics_refresh_test.rs @@ -311,6 +311,26 @@ async fn leader_refreshes_statistics_on_election_and_every_interval() { database.drop().await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn leader_refreshes_statistics_on_election_before_the_first_interval() { + let database = TestDatabase::create(None, &[]).await; + let pool = &database.owner_pool; + let store = QueueStorage::new(QueueStorageConfig::default()).expect("queue storage config"); + seed_ready_rows(&store, pool, 200).await; + + let client = probe_client(pool.clone(), Duration::from_secs(300), None); + client.start().await.expect("start client"); + wait_for( + "a statistics pass on election", + Duration::from_secs(30), + || async { stats_rows(pool, "awa", "ready_entries", true).await > 0 }, + ) + .await; + + client.shutdown(Duration::from_secs(5)).await; + database.drop().await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn zero_interval_disables_the_refresh() { let database = TestDatabase::create(None, &[]).await; @@ -358,6 +378,30 @@ async fn non_owner_runtime_role_refreshes_statistics() { "the runtime role must not be able to analyse the tables itself" ); + for (schema, relation, message) in [ + ( + "public", + "queue_claim_heads", + "is not a queue-storage schema", + ), + ("awa", "jobs", "is not a partitioned or head table"), + ] { + let error = + sqlx::query_scalar::<_, String>("SELECT awa.analyze_queue_storage_relation($1, $2)") + .bind(schema) + .bind(relation) + .fetch_one(&runtime_pool) + .await + .expect_err("the helper rejects what it may not analyse"); + let database_error = error.as_database_error().expect("database error"); + assert_eq!(database_error.code().as_deref(), Some("22023")); + assert!( + database_error.message().contains(message), + "{schema}.{relation}: {}", + database_error.message() + ); + } + let client = probe_client(runtime_pool.clone(), Duration::from_secs(1), None); client .start() @@ -378,6 +422,50 @@ async fn non_owner_runtime_role_refreshes_statistics() { database.drop().await; } +/// A conflicting lock on a ring parent fails the call with `lock_not_available` +/// rather than letting `ANALYZE (SKIP_LOCKED)` skip every partition and report +/// success. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn locked_ring_parent_fails_the_call_instead_of_reporting_analyzed() { + let database = TestDatabase::create(None, &[]).await; + let pool = &database.owner_pool; + + let mut blocker = pool.begin().await.expect("begin blocker"); + sqlx::query("LOCK TABLE ONLY awa.ready_entries IN SHARE UPDATE EXCLUSIVE MODE") + .execute(blocker.as_mut()) + .await + .expect("lock the parent"); + + let mut call = pool.begin().await.expect("begin call"); + sqlx::query("SELECT set_config('lock_timeout', '200ms', true)") + .execute(call.as_mut()) + .await + .expect("set lock_timeout"); + let error = sqlx::query_scalar::<_, String>( + "SELECT awa.analyze_queue_storage_relation('awa', 'ready_entries')", + ) + .fetch_one(call.as_mut()) + .await + .expect_err("the locked parent must not report analyzed"); + assert_eq!( + error + .as_database_error() + .and_then(|error| error.code()) + .as_deref(), + Some("55P03") + ); + call.rollback().await.expect("rollback call"); + + blocker.rollback().await.expect("release the parent"); + let status: String = + sqlx::query_scalar("SELECT awa.analyze_queue_storage_relation('awa', 'ready_entries')") + .fetch_one(pool) + .await + .expect("analyse once the lock is gone"); + assert_eq!(status, "analyzed"); + database.drop().await; +} + fn sum_by_outcome( resource_metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics], outcome: &str, diff --git a/docs/configuration.md b/docs/configuration.md index 7e39bf2e8..0fd661248 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -527,7 +527,7 @@ Autovacuum does not analyse partitioned tables, so the ring parents (`ready_entr The maintenance leader runs `ANALYZE` on both when it is elected and every `statistics_refresh_interval` (default `5m`). One pass takes well under a second on a small ring and a few seconds on a ring with millions of retained rows. It has no measurable effect on claim or completion throughput. -- **Locks.** Each relation is analysed in its own transaction with a 2 s `lock_timeout` and a 120 s `statement_timeout`. Children that ring prune holds are skipped (`SKIP_LOCKED`). Prune never waits for the refresh: its own 10 ms lock timeout applies, so an attempt that lands while `ANALYZE` samples the oldest slot's tables returns `blocked` and retries on its usual backoff. `ANALYZE` of a partitioned table samples every child, so that window cannot be avoided. +- **Locks.** Each relation is analysed in its own transaction with a 2 s `lock_timeout` and a 120 s `statement_timeout`. `SKIP_LOCKED` skips a child that ring prune holds during the per-child analysis, but sampling for the parent's statistics waits for it up to the `lock_timeout`; a call that times out is recorded as `skipped_lock` and retried on the next interval. Prune never waits for the refresh: its own 10 ms lock timeout applies, so an attempt that lands while `ANALYZE` samples the oldest slot's tables returns `blocked` and retries on its usual backoff. `ANALYZE` of a partitioned table samples every child, so that window cannot be avoided. - **Privileges.** The refresh calls `awa.analyze_queue_storage_relation(schema, relation)` (a schema patch applied by `awa migrate`), a `SECURITY DEFINER` function owned by the role that ran it, so the runtime role needs no `MAINTAIN` grant and does not have to own the tables. If the function is missing (run `awa migrate`) or its owner may not analyse the tables, the leader logs one warning that names what is missing and skips the pass; job processing is unaffected. - **Metrics.** `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration`, by `awa.analyze.scope` (`heads`, `ring_parent`) and `awa.analyze.outcome` (`ok`, `skipped_privilege`, `skipped_lock`, `unavailable`, `failed`). A sustained `skipped_lock` rate means something else holds the tables; any `skipped_privilege` or `unavailable` means the refresh is not running. From 8857f8983e0787948127d3ec18e6ccc1ab410ce2 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 17:51:33 +1300 Subject: [PATCH 4/6] fix(worker): restrict who may call the statistics helper and what it may analyse (0.6) - Revoke EXECUTE on awa.analyze_queue_storage_relation from PUBLIC. The SHARE UPDATE EXCLUSIVE lock it takes lasts until the caller's transaction ends, so any role could hold it open and block ring prune. The patch grants EXECUTE to the roles that already hold INSERT, UPDATE and DELETE on the awa head tables, and the function requires the calling session user to hold those privileges on the target schema. - Give the function its own lock_timeout. - Accept only the two head tables and the thirteen ring parents inside the function, not any partitioned table of the schema. - Probe for the function once per pass, so a binary ahead of the patch makes one catalog query instead of fifteen failing calls and server ERROR lines. Pass-level outcomes use a new "pass" metric scope. - Correct the patch header, which claimed the function takes no lock. - Tests: a role without EXECUTE and a role without head-table privileges are denied, the patch grants EXECUTE to existing grantees, non-ring tables are rejected, and a missing helper costs one probe and no calls. --- CHANGELOG.md | 13 +- awa-metrics/src/lib.rs | 6 +- .../analyze_queue_storage_relation.sql | 113 ++++++++++--- awa-worker/src/statistics.rs | 129 +++++++++++++-- awa/tests/statistics_refresh_test.rs | 149 +++++++++++++++++- docs/configuration.md | 4 +- docs/grafana/README.md | 2 +- docs/security.md | 2 +- 8 files changed, 371 insertions(+), 47 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d3d265447..ffdfb2bd3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,7 @@ Notable changes between releases. Detailed migration notes for storage transitio lands while ANALYZE samples the oldest slot's tables returns `blocked` and retries on its normal backoff. Results are in `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration` - (scope `heads` or `ring_parent`; outcome `ok`, `skipped_privilege`, + (scope `heads`, `ring_parent` or `pass`; outcome `ok`, `skipped_privilege`, `skipped_lock`, `unavailable` or `failed`). - **Schema patch `analyze_queue_storage_relation` lets a runtime role that does not own the tables refresh statistics.** `ANALYZE` skips a table its caller @@ -30,13 +30,18 @@ Notable changes between releases. Detailed migration notes for storage transitio `MAINTAIN`, so the refresh runs through the `SECURITY DEFINER` function `awa.analyze_queue_storage_relation(schema, relation)`, owned by the role that applied the patch. It analyses only a head table or partitioned table - of a queue-storage schema owned by the owner of the `awa` schema, and - `EXECUTE` is granted to `PUBLIC` like the other runtime helpers. The 0.6 + of a queue-storage schema owned by the owner of the `awa` schema. It is not + executable by `PUBLIC`, because the lock it takes lasts until the caller's + transaction ends and would block ring prune: the patch grants `EXECUTE` to + the roles that already hold INSERT, UPDATE and DELETE on the head tables + (the runtime grantees), and the function requires the calling session user to + hold them. A role created later needs the documented + `GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa`. The 0.6 series cannot add a migration version, so it ships like `wait_free_dirty_marks`: **run `awa migrate` as the schema owner after upgrading** (it applies the patch on a v040 schema; `awa migrate --sql` and `--extract-to` emit it as the repeatable script - `R__analyze_queue_storage_relation`). No grant changes are needed. The same + `R__analyze_queue_storage_relation`). The same function is awa 0.7 migration v053, which replaces it cleanly. A binary running before the patch, or against a function whose owner may not analyse the tables, logs one warning naming what is missing and skips the refresh diff --git a/awa-metrics/src/lib.rs b/awa-metrics/src/lib.rs index 34e374998..ba85f7c13 100644 --- a/awa-metrics/src/lib.rs +++ b/awa-metrics/src/lib.rs @@ -246,7 +246,7 @@ pub struct AwaMetrics { /// rather than one per tick. Fleets alert on this rather than /// scraping the matching `tracing::warn!` line. Issue #242. pub maintenance_branch_overrun_total: Counter, - /// Planner-statistics `ANALYZE` calls by scope (`heads`, `ring_parent`) + /// Planner-statistics `ANALYZE` calls by scope (`heads`, `ring_parent`, `pass`) /// and outcome (`ok`, `skipped_privilege`, `skipped_lock`, /// `unavailable`, `failed`). One increment per relation per refresh pass. pub maintenance_analyze_attempts: Counter, @@ -1116,8 +1116,8 @@ impl AwaMetrics { self.maintenance_branch_overrun_total.add(1, &attrs); } - /// Record one planner-statistics `ANALYZE` call. `scope` is `"heads"` or - /// `"ring_parent"`; `outcome` is `"ok"`, `"skipped_privilege"`, + /// Record one planner-statistics `ANALYZE` call. `scope` is `"heads"`, + /// `"ring_parent"` or `"pass"` (a pass-level probe or failure); `outcome` is `"ok"`, `"skipped_privilege"`, /// `"skipped_lock"`, `"unavailable"` or `"failed"`. pub fn record_statistics_analyze( &self, diff --git a/awa-model/migrations/patches/analyze_queue_storage_relation.sql b/awa-model/migrations/patches/analyze_queue_storage_relation.sql index e2b3d50af..e109bcdc9 100644 --- a/awa-model/migrations/patches/analyze_queue_storage_relation.sql +++ b/awa-model/migrations/patches/analyze_queue_storage_relation.sql @@ -10,14 +10,18 @@ -- migrations.rs reports it missing; `awa migrate --sql` prints it as a -- repeatable R__ script for external runners. -- --- Autovacuum does not analyse partitioned tables, so the ring parents have no --- inherited statistics, and it never reaches the autoanalyze threshold on the --- two small head tables (queue_claim_heads, queue_enqueue_heads). The --- maintenance leader refreshes both on an interval. Runtimes usually connect --- as a role that does not own the tables, and ANALYZE skips with a warning a --- table its caller may not analyse, so the refresh runs through this --- SECURITY DEFINER function owned by the role that applies the patch, which --- owns the tables. +-- Autovacuum does not analyse partitioned tables, so the ring parents +-- (ready_entries, done_entries, leases, ...) have no inherited statistics. It +-- also never reaches the autoanalyze threshold on the two small head tables +-- (queue_claim_heads, queue_enqueue_heads), so the planner prices lane joins +-- against them at its defaults (320 rows, n_distinct 200) while they drive +-- joins over hundreds of thousands of retained rows. +-- +-- The maintenance leader refreshes both on an interval. Runtimes usually +-- connect as a role that does not own the tables, and ANALYZE skips with a +-- warning a table its caller may not analyse, so the refresh runs through +-- this SECURITY DEFINER function owned by the migration role, which owns the +-- tables. -- -- The function analyses one relation per call and the caller runs each call as -- its own transaction, so the locks ANALYZE takes on a partitioned table's @@ -27,22 +31,35 @@ -- applies when ANALYZE starts work on a relation, so a child that ring prune -- holds ACCESS EXCLUSIVE is skipped during the per-child analysis, but -- sampling the children for the parent's statistics can still wait for it. --- That wait is bounded by the caller's lock_timeout and fails the call with +-- That wait is bounded by the function's lock_timeout and fails the call with -- SQLSTATE 55P03, which the caller records as skipped_lock. SKIP_LOCKED also -- skips every partition silently when the parent itself is locked, so the -- function takes the parent's SHARE UPDATE EXCLUSIVE lock first and lets that -- wait fail instead. -- --- Definer boundary: only a head table or a partitioned table of a queue-storage --- schema (the AWA-specific claim_ready_runtime sentinel) that is owned by the --- owner of the awa schema is analysed, from quoted identifiers only. ANALYZE --- writes statistics and nothing else. +-- Definer boundary: only a head table or ring parent of a queue-storage schema +-- (the AWA-specific claim_ready_runtime sentinel) that is owned by the owner of +-- the awa schema is analysed, from quoted identifiers only. ANALYZE writes +-- statistics and nothing else. +-- Only the fixed list of ring parents and the two head tables is accepted; +-- a migration that adds a partitioned table to the substrate must extend it. +-- +-- Who may call it: the lock above lasts until the caller's transaction ends +-- and blocks ring prune of that table, so EXECUTE is not granted to PUBLIC. It +-- goes to the roles that already hold INSERT, UPDATE and DELETE on the head +-- tables of the awa schema (the runtime grantees), and the function itself +-- requires the calling session user to hold those privileges on the target +-- schema's queue_claim_heads. A role created later needs the documented +-- `GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa` (or default privileges). -- -- Returns 'analyzed'; 'skipped_privilege' when the function owner may not -- analyse the relation; or 'missing' when the relation does not exist. --- Takes no lock on any queue-storage table and completes in milliseconds. +-- -- Releases before the patch never call it; a binary that runs before the patch --- logs once that the function is missing and skips the refresh. +-- probes for the function, logs once that it is missing and skips the refresh. +-- Creating or replacing it takes no table lock and completes in milliseconds; +-- calling it takes SHARE UPDATE EXCLUSIVE on the relation until the caller's +-- transaction ends. CREATE OR REPLACE FUNCTION awa.analyze_queue_storage_relation( p_schema TEXT, @@ -52,8 +69,10 @@ RETURNS TEXT LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, pg_temp +SET lock_timeout = '2s' AS $$ DECLARE + v_heads REGCLASS; v_relid OID; v_relkind "char"; v_relowner OID; @@ -67,6 +86,38 @@ BEGIN USING ERRCODE = '22023'; END IF; + IF p_relation <> ALL (ARRAY[ + 'queue_claim_heads', + 'queue_enqueue_heads', + 'done_entries', + 'lease_claim_batches', + 'lease_claim_closure_batches', + 'lease_claim_closures', + 'lease_claims', + 'leases', + 'queue_terminal_count_deltas', + 'ready_claim_attempt_batches', + 'ready_entries', + 'ready_segments', + 'ready_tombstones', + 'receipt_completion_batches', + 'receipt_completion_tombstones' + ]) THEN + RAISE EXCEPTION '%.% is not a ring parent or head table of a queue-storage schema', + p_schema, p_relation + USING ERRCODE = '22023'; + END IF; + + v_heads := to_regclass(format('%I.queue_claim_heads', p_schema)); + IF v_heads IS NULL + OR NOT has_table_privilege(session_user, v_heads, 'INSERT') + OR NOT has_table_privilege(session_user, v_heads, 'UPDATE') + OR NOT has_table_privilege(session_user, v_heads, 'DELETE') THEN + RAISE EXCEPTION 'permission denied: % may not maintain queue storage in schema %', + session_user, p_schema + USING ERRCODE = '42501'; + END IF; + v_relid := to_regclass(format('%I.%I', p_schema, p_relation)); IF v_relid IS NULL THEN RETURN 'missing'; @@ -74,14 +125,6 @@ BEGIN SELECT class.relkind, class.relowner INTO v_relkind, v_relowner FROM pg_class AS class WHERE class.oid = v_relid; - IF NOT ( - v_relkind = 'p' - OR p_relation IN ('queue_claim_heads', 'queue_enqueue_heads') - ) THEN - RAISE EXCEPTION '%.% is not a partitioned or head table of a queue-storage schema', - p_schema, p_relation - USING ERRCODE = '22023'; - END IF; IF v_relowner IS DISTINCT FROM ( SELECT namespace.nspowner FROM pg_namespace AS namespace WHERE namespace.nspname = 'awa' @@ -124,4 +167,26 @@ BEGIN END $$; -GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(TEXT, TEXT) TO PUBLIC; +REVOKE EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(TEXT, TEXT) FROM PUBLIC; + +DO $grant$ +DECLARE + v_grantee TEXT; +BEGIN + FOR v_grantee IN + SELECT pg_get_userbyid(privilege.grantee) + FROM pg_class AS heads + CROSS JOIN LATERAL aclexplode(heads.relacl) AS privilege + WHERE heads.oid = to_regclass('awa.queue_claim_heads') + AND privilege.grantee <> 0 + AND privilege.privilege_type IN ('INSERT', 'UPDATE', 'DELETE') + GROUP BY privilege.grantee + HAVING count(DISTINCT privilege.privilege_type) = 3 + LOOP + EXECUTE format( + 'GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) TO %I', + v_grantee + ); + END LOOP; +END +$grant$; diff --git a/awa-worker/src/statistics.rs b/awa-worker/src/statistics.rs index cfc97ef57..b808f9cce 100644 --- a/awa-worker/src/statistics.rs +++ b/awa-worker/src/statistics.rs @@ -54,6 +54,8 @@ const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03"; pub(crate) enum RelationScope { Heads, RingParent, + /// The pass as a whole: a probe or listing failed, or the helper is absent. + Pass, } impl RelationScope { @@ -61,6 +63,7 @@ impl RelationScope { match self { RelationScope::Heads => "heads", RelationScope::RingParent => "ring_parent", + RelationScope::Pass => "pass", } } } @@ -193,7 +196,31 @@ impl StatisticsRefresher { Err(err) => { warn!(error = %err, schema = %self.schema, "Could not check the queue-storage schema for statistics refresh"); self.record( - RelationScope::RingParent, + RelationScope::Pass, + AnalyzeOutcome::Failed, + started.elapsed(), + ); + summary.record(AnalyzeOutcome::Failed); + return summary; + } + } + + match self.helper_installed().await { + Ok(true) => {} + Ok(false) => { + self.warn_unavailable_once(); + self.record( + RelationScope::Pass, + AnalyzeOutcome::Unavailable, + started.elapsed(), + ); + summary.record(AnalyzeOutcome::Unavailable); + return summary; + } + Err(err) => { + warn!(error = %err, schema = %self.schema, "Could not probe for the statistics helper"); + self.record( + RelationScope::Pass, AnalyzeOutcome::Failed, started.elapsed(), ); @@ -207,7 +234,7 @@ impl StatisticsRefresher { Err(err) => { warn!(error = %err, schema = %self.schema, "Could not list ring tables for statistics refresh"); self.record( - RelationScope::RingParent, + RelationScope::Pass, AnalyzeOutcome::Failed, started.elapsed(), ); @@ -250,6 +277,18 @@ impl StatisticsRefresher { .record_statistics_analyze(scope.as_str(), outcome.as_str(), elapsed); } + /// One catalog lookup per pass. A missing helper (a binary running ahead + /// of the schema) then costs this query instead of a failing call, and a + /// server-side ERROR line, per relation; the pass interval is the + /// re-probe interval. + async fn helper_installed(&self) -> Result { + sqlx::query_scalar( + "SELECT to_regprocedure('awa.analyze_queue_storage_relation(text,text)') IS NOT NULL", + ) + .fetch_one(&self.pool) + .await + } + async fn ring_parents(&self) -> Result, sqlx::Error> { sqlx::query_scalar( r#" @@ -327,19 +366,14 @@ impl StatisticsRefresher { AnalyzeOutcome::SkippedLock } Some(SQLSTATE_UNDEFINED_FUNCTION) => { - if !self.warned_unavailable.swap(true, Ordering::Relaxed) { - warn!( - schema = %self.schema, - "Planner statistics refresh is disabled: awa.analyze_queue_storage_relation() \ - is missing. Run `awa migrate` to apply the analyze_queue_storage_relation schema patch" - ); - } + self.warn_unavailable_once(); AnalyzeOutcome::Unavailable } Some(SQLSTATE_INSUFFICIENT_PRIVILEGE) => { self.warn_privilege_once( relation, - "the runtime role may not execute awa.analyze_queue_storage_relation(); \ + "the runtime role may not execute awa.analyze_queue_storage_relation(), or lacks \ + INSERT, UPDATE and DELETE on the queue head tables; \ GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) to it", ); AnalyzeOutcome::SkippedPrivilege @@ -351,6 +385,17 @@ impl StatisticsRefresher { } } + fn warn_unavailable_once(&self) { + if self.warned_unavailable.swap(true, Ordering::Relaxed) { + return; + } + warn!( + schema = %self.schema, + "Planner statistics refresh is disabled: awa.analyze_queue_storage_relation() \ + is missing. Run `awa migrate` to apply the analyze_queue_storage_relation schema patch" + ); + } + fn warn_privilege_once(&self, relation: &str, missing: &str) { if self.warned_privilege.swap(true, Ordering::Relaxed) { return; @@ -395,3 +440,67 @@ impl Drop for RefreshGuard { self.cancel.cancel(); } } + +#[cfg(test)] +mod tests { + use super::*; + use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; + use std::str::FromStr; + + /// A binary ahead of the schema must probe once and attempt no calls: a + /// failing call per relation would count as `unavailable` fifteen times + /// and write fifteen server-side ERROR lines per pass. + #[tokio::test] + async fn missing_helper_costs_one_probe_and_no_calls() { + let url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string()); + let options = PgConnectOptions::from_str(&url).expect("parse DATABASE_URL"); + let admin = PgPoolOptions::new() + .max_connections(1) + .connect_with(options.clone()) + .await + .expect("connect admin"); + let database = format!("awa_stats_unit_{}", std::process::id()); + sqlx::raw_sql(&format!("DROP DATABASE IF EXISTS {database} WITH (FORCE)")) + .execute(&admin) + .await + .expect("drop leftover database"); + sqlx::raw_sql(&format!("CREATE DATABASE {database}")) + .execute(&admin) + .await + .expect("create database"); + let pool = PgPoolOptions::new() + .max_connections(2) + .connect_with(options.database(&database)) + .await + .expect("connect test database"); + awa_model::migrations::run(&pool).await.expect("migrate"); + + let refresher = StatisticsRefresher::new( + pool.clone(), + "awa".to_string(), + AwaMetrics::from_global(), + false, + ); + let present = refresher.run_pass().await; + assert_eq!(present.ok, 15, "{present:?}"); + + sqlx::raw_sql("DROP FUNCTION awa.analyze_queue_storage_relation(text, text)") + .execute(&pool) + .await + .expect("drop the helper"); + let absent = refresher.run_pass().await; + assert_eq!( + absent, + PassSummary { + unavailable: 1, + ..PassSummary::default() + } + ); + pool.close().await; + sqlx::raw_sql(&format!("DROP DATABASE IF EXISTS {database} WITH (FORCE)")) + .execute(&admin) + .await + .expect("drop database"); + } +} diff --git a/awa/tests/statistics_refresh_test.rs b/awa/tests/statistics_refresh_test.rs index 7322639a0..69bd3aebb 100644 --- a/awa/tests/statistics_refresh_test.rs +++ b/awa/tests/statistics_refresh_test.rs @@ -384,7 +384,7 @@ async fn non_owner_runtime_role_refreshes_statistics() { "queue_claim_heads", "is not a queue-storage schema", ), - ("awa", "jobs", "is not a partitioned or head table"), + ("awa", "jobs", "is not a ring parent or head table"), ] { let error = sqlx::query_scalar::<_, String>("SELECT awa.analyze_queue_storage_relation($1, $2)") @@ -422,6 +422,150 @@ async fn non_owner_runtime_role_refreshes_statistics() { database.drop().await; } +async fn call_helper( + pool: &PgPool, + schema: &str, + relation: &str, +) -> Result { + sqlx::query_scalar("SELECT awa.analyze_queue_storage_relation($1, $2)") + .bind(schema) + .bind(relation) + .fetch_one(pool) + .await +} + +fn denied(error: &sqlx::Error, message: &str) { + let database_error = error.as_database_error().expect("database error"); + assert_eq!(database_error.code().as_deref(), Some("42501")); + assert!( + database_error.message().contains(message), + "unexpected message: {}", + database_error.message() + ); +} + +/// EXECUTE is not public: holding the lock the helper takes until the end of +/// a transaction would block ring prune of that table. A role needs EXECUTE and +/// write privileges on the head tables, and the migration grants EXECUTE to +/// the roles that already have them. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn execute_is_limited_to_roles_that_may_maintain_queue_storage() { + let suffix = uuid::Uuid::new_v4().simple().to_string(); + let owner = format!("stats_owner_{suffix}"); + let runtime = format!("stats_runtime_{suffix}"); + let bystander = format!("stats_bystander_{suffix}"); + let reader = format!("stats_reader_{suffix}"); + let database = TestDatabase::create(Some(&owner), &[&runtime, &bystander, &reader]).await; + let owner_pool = &database.owner_pool; + run_sql( + owner_pool, + format!( + "GRANT USAGE ON SCHEMA awa TO {bystander}, {reader}, {runtime}; + GRANT SELECT ON ALL TABLES IN SCHEMA awa TO {reader}; + GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) TO {reader}; + GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON ALL TABLES IN SCHEMA awa TO {runtime}" + ), + ) + .await + .expect("grant privileges"); + let connect = |role: String| { + let options = database.options.clone(); + async move { + PgPoolOptions::new() + .max_connections(2) + .connect_with(options.username(&role).password(ROLE_PASSWORD)) + .await + .expect("connect role pool") + } + }; + let bystander_pool = connect(bystander.clone()).await; + let reader_pool = connect(reader.clone()).await; + let runtime_pool = connect(runtime.clone()).await; + + denied( + &call_helper(&bystander_pool, "awa", "queue_claim_heads") + .await + .expect_err("a role without EXECUTE is denied"), + "permission denied for function", + ); + denied( + &call_helper(&reader_pool, "awa", "queue_claim_heads") + .await + .expect_err("EXECUTE without write privileges on the head tables is denied"), + "may not maintain queue storage", + ); + + // The runtime role holds write privileges but was granted no function + // EXECUTE: re-running the migration grants it, as an upgrade does. + denied( + &call_helper(&runtime_pool, "awa", "queue_claim_heads") + .await + .expect_err("write privileges alone do not grant EXECUTE"), + "permission denied for function", + ); + for (name, _, sql) in migrations::schema_patch_sql(migrations::SCHEMA_PATCHES) { + if name == "analyze_queue_storage_relation" { + run_sql(owner_pool, sql).await.expect("re-apply the patch"); + } + } + assert_eq!( + call_helper(&runtime_pool, "awa", "queue_claim_heads") + .await + .expect("the migration granted EXECUTE to the runtime grantee"), + "analyzed" + ); + + for pool in [bystander_pool, reader_pool, runtime_pool] { + pool.close().await; + } + database.drop().await; +} + +/// The helper analyses the ring parents and head tables only, not any other +/// partitioned table of the schema. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn helper_rejects_a_partitioned_table_that_is_not_a_ring_parent() { + let suffix = uuid::Uuid::new_v4().simple().to_string(); + let owner = format!("stats_owner_{suffix}"); + let runtime = format!("stats_runtime_{suffix}"); + let database = TestDatabase::create(Some(&owner), &[&runtime]).await; + run_sql( + &database.owner_pool, + "CREATE TABLE awa.application_events (id int) PARTITION BY LIST (id)".to_string(), + ) + .await + .expect("create a partitioned application table"); + let runtime_pool = database.runtime_pool(&runtime).await; + + for relation in ["application_events", "ready_entries_0", "jobs_hot"] { + let error = call_helper(&runtime_pool, "awa", relation) + .await + .expect_err("only ring parents and head tables are accepted"); + let database_error = error.as_database_error().expect("database error"); + assert_eq!( + database_error.code().as_deref(), + Some("22023"), + "{relation}" + ); + assert!( + database_error + .message() + .contains("is not a ring parent or head table"), + "{relation}: {}", + database_error.message() + ); + } + assert_eq!( + call_helper(&runtime_pool, "awa", "ready_entries") + .await + .expect("a ring parent is accepted"), + "analyzed" + ); + + runtime_pool.close().await; + database.drop().await; +} + /// A conflicting lock on a ring parent fails the call with `lock_not_available` /// rather than letting `ANALYZE (SKIP_LOCKED)` skip every partition and report /// success. @@ -520,7 +664,8 @@ async fn unrunnable_refresh_is_recorded_and_does_not_fail_the_runtime() { &database.owner_pool, format!( "GRANT USAGE, CREATE ON SCHEMA awa TO {helper}; - ALTER FUNCTION awa.analyze_queue_storage_relation(text, text) OWNER TO {helper}" + ALTER FUNCTION awa.analyze_queue_storage_relation(text, text) OWNER TO {helper}; + GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) TO {owner}" ), ) .await diff --git a/docs/configuration.md b/docs/configuration.md index 0fd661248..83715cb76 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -528,8 +528,8 @@ Autovacuum does not analyse partitioned tables, so the ring parents (`ready_entr The maintenance leader runs `ANALYZE` on both when it is elected and every `statistics_refresh_interval` (default `5m`). One pass takes well under a second on a small ring and a few seconds on a ring with millions of retained rows. It has no measurable effect on claim or completion throughput. - **Locks.** Each relation is analysed in its own transaction with a 2 s `lock_timeout` and a 120 s `statement_timeout`. `SKIP_LOCKED` skips a child that ring prune holds during the per-child analysis, but sampling for the parent's statistics waits for it up to the `lock_timeout`; a call that times out is recorded as `skipped_lock` and retried on the next interval. Prune never waits for the refresh: its own 10 ms lock timeout applies, so an attempt that lands while `ANALYZE` samples the oldest slot's tables returns `blocked` and retries on its usual backoff. `ANALYZE` of a partitioned table samples every child, so that window cannot be avoided. -- **Privileges.** The refresh calls `awa.analyze_queue_storage_relation(schema, relation)` (a schema patch applied by `awa migrate`), a `SECURITY DEFINER` function owned by the role that ran it, so the runtime role needs no `MAINTAIN` grant and does not have to own the tables. If the function is missing (run `awa migrate`) or its owner may not analyse the tables, the leader logs one warning that names what is missing and skips the pass; job processing is unaffected. -- **Metrics.** `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration`, by `awa.analyze.scope` (`heads`, `ring_parent`) and `awa.analyze.outcome` (`ok`, `skipped_privilege`, `skipped_lock`, `unavailable`, `failed`). A sustained `skipped_lock` rate means something else holds the tables; any `skipped_privilege` or `unavailable` means the refresh is not running. +- **Privileges.** The refresh calls `awa.analyze_queue_storage_relation(schema, relation)` (a schema patch applied by `awa migrate`), a `SECURITY DEFINER` function owned by the role that ran it, so the runtime role needs no `MAINTAIN` grant and does not have to own the tables. It is executable only by roles that hold INSERT, UPDATE and DELETE on the queue head tables (the patch grants it to the existing ones). Each pass first probes for the function; if it is missing (run `awa migrate`) or its owner may not analyse the tables, the leader logs one warning that names what is missing and skips the pass without making failing calls; job processing is unaffected. +- **Metrics.** `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration`, by `awa.analyze.scope` (`heads`, `ring_parent`, `pass`) and `awa.analyze.outcome` (`ok`, `skipped_privilege`, `skipped_lock`, `unavailable`, `failed`). A sustained `skipped_lock` rate means something else holds the tables; any `skipped_privilege` or `unavailable` means the refresh is not running. ### Rust diff --git a/docs/grafana/README.md b/docs/grafana/README.md index dde57d5cd..e836066be 100644 --- a/docs/grafana/README.md +++ b/docs/grafana/README.md @@ -157,7 +157,7 @@ All metrics use the `awa` OTel meter name and are exported via OTLP to your conf | `awa.maintenance.promote_batch_size` | Histogram | state | Jobs promoted | | `awa.maintenance.promote_duration` | Histogram (s) | state | Promotion time | | `awa.maintenance.rescues` | Counter | rescue_kind | Jobs rescued | -| `awa.maintenance.analyze.attempts` | Counter | scope, outcome | Planner-statistics `ANALYZE` calls on the head tables and ring parents (`ok`, `skipped_privilege`, `skipped_lock`, `unavailable`, `failed`) | +| `awa.maintenance.analyze.attempts` | Counter | scope (`heads`, `ring_parent`, `pass`), outcome | Planner-statistics `ANALYZE` calls on the head tables and ring parents (`ok`, `skipped_privilege`, `skipped_lock`, `unavailable`, `failed`) | | `awa.maintenance.analyze.duration` | Histogram (s) | scope, outcome | Duration of one planner-statistics `ANALYZE` call | ### Heartbeat diff --git a/docs/security.md b/docs/security.md index 7bc07cb29..49f836711 100644 --- a/docs/security.md +++ b/docs/security.md @@ -224,7 +224,7 @@ The runtime also directly upserts into the descriptor catalogs on startup and on These writes are not trigger-driven; they come from `ClientBuilder::build()` / `AsyncClient.start()` and from the snapshot reporter. The broad grant already covers them. -The maintenance leader refreshes planner statistics for the queue-storage head tables and ring parents (see [Planner statistics](configuration.md#planner-statistics)). `ANALYZE` skips a table its caller may not analyse, so a runtime that does not own the tables calls `awa.analyze_queue_storage_relation(schema, relation)`, a `SECURITY DEFINER` function installed by the `analyze_queue_storage_relation` schema patch under the migrator role and executable by `PUBLIC`. The runtime needs no `MAINTAIN` grant. The function only analyses a head table or partitioned table of a queue-storage schema that is owned by the owner of the `awa` schema, which is why migrations must run with the owner role as the effective role (step 2 above). A function owned by a role that may not analyse the tables makes every pass report `skipped_privilege` and logs one warning. +The maintenance leader refreshes planner statistics for the queue-storage head tables and ring parents (see [Planner statistics](configuration.md#planner-statistics)). `ANALYZE` skips a table its caller may not analyse, so a runtime that does not own the tables calls `awa.analyze_queue_storage_relation(schema, relation)`, a `SECURITY DEFINER` function installed by the `analyze_queue_storage_relation` schema patch under the migrator role and executable only by roles that hold `INSERT`, `UPDATE` and `DELETE` on the head tables (the patch grants it to the existing ones; a role created later gets it from `GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa`), because the lock it takes lasts until the caller's transaction ends and would block ring prune. The runtime needs no `MAINTAIN` grant. The function only analyses the two head tables and the thirteen ring parents of a queue-storage schema that is owned by the owner of the `awa` schema, which is why migrations must run with the owner role as the effective role (step 2 above). A function owned by a role that may not analyse the tables makes every pass report `skipped_privilege` and logs one warning. Other PostgreSQL features used at runtime: From 9ed7c2cee4a0455f87f8e636425f147664e7b059 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 21:20:54 +1300 Subject: [PATCH 5/6] Repair stale statistics helpers and skip missing runtime privileges --- CHANGELOG.md | 5 +- .../analyze_queue_storage_relation.sql | 3 +- awa-model/src/migrations.rs | 46 +++++- awa-worker/src/statistics.rs | 156 ++++++++++++++++-- awa/tests/statistics_refresh_test.rs | 96 +++++++++++ docs/configuration.md | 2 +- docs/security.md | 2 +- 7 files changed, 290 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ffdfb2bd3..7058b951e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,10 +42,13 @@ Notable changes between releases. Detailed migration notes for storage transitio upgrading** (it applies the patch on a v040 schema; `awa migrate --sql` and `--extract-to` emit it as the repeatable script `R__analyze_queue_storage_relation`). The same - function is awa 0.7 migration v053, which replaces it cleanly. A binary + function is awa 0.7 migration v054, which replaces it cleanly. A binary running before the patch, or against a function whose owner may not analyse the tables, logs one warning naming what is missing and skips the refresh without affecting job processing. Earlier binaries never call the function. + The patch also repairs an older helper definition, PUBLIC EXECUTE or missing + grants for existing runtime roles; a missing runtime privilege skips a pass + before calling the helper. ### Fixed diff --git a/awa-model/migrations/patches/analyze_queue_storage_relation.sql b/awa-model/migrations/patches/analyze_queue_storage_relation.sql index e109bcdc9..288d00d62 100644 --- a/awa-model/migrations/patches/analyze_queue_storage_relation.sql +++ b/awa-model/migrations/patches/analyze_queue_storage_relation.sql @@ -1,6 +1,6 @@ -- Schema patch analyze_queue_storage_relation: planner statistics helper for -- the queue-storage tables autovacuum never analyses (#631). --- Counterpart of awa 0.7 migration v053, minus the schema_version row; v053 +-- Counterpart of awa 0.7 migration v054, minus the schema_version row; v054 -- replaces this body with one that checks the schema through -- awa.queue_storage_schemas(), which 0.6 does not have. -- @@ -78,6 +78,7 @@ DECLARE v_relowner OID; v_may_analyze BOOLEAN; BEGIN + -- awa_statistics_helper_v1: bump when the definition or grants change. IF to_regprocedure(format( '%I.claim_ready_runtime(text,bigint,double precision,double precision)', p_schema diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs index 204329e16..7cec64c16 100644 --- a/awa-model/src/migrations.rs +++ b/awa-model/src/migrations.rs @@ -295,6 +295,40 @@ macro_rules! prepared_lane_sequences_pending_sql { }; } +/// An existing function can still predate the hardened definition or its +/// runtime grants. Keep the patch pending until all three are current. +macro_rules! analyze_queue_storage_relation_pending_sql { + () => { + "SELECT 'awa.analyze_queue_storage_relation definition'::text AS target \ + WHERE NOT EXISTS ( \ + SELECT 1 FROM pg_proc AS helper \ + WHERE helper.oid = to_regprocedure('awa.analyze_queue_storage_relation(text,text)') \ + AND helper.prosecdef \ + AND helper.prosrc LIKE '%awa_statistics_helper_v1%' \ + AND helper.proconfig @> ARRAY['search_path=pg_catalog, pg_temp', 'lock_timeout=2s']::text[]) \ + UNION ALL \ + SELECT 'PUBLIC EXECUTE on awa.analyze_queue_storage_relation' \ + FROM pg_proc AS helper \ + WHERE helper.oid = to_regprocedure('awa.analyze_queue_storage_relation(text,text)') \ + AND EXISTS (SELECT 1 FROM aclexplode(COALESCE(helper.proacl, acldefault('f', helper.proowner))) AS privilege \ + WHERE privilege.grantee = 0 AND privilege.privilege_type = 'EXECUTE') \ + UNION ALL \ + SELECT format('EXECUTE on awa.analyze_queue_storage_relation for %s', pg_get_userbyid(runtime.grantee)) \ + FROM ( \ + SELECT privilege.grantee \ + FROM pg_class AS heads \ + CROSS JOIN LATERAL aclexplode(COALESCE(heads.relacl, acldefault('r', heads.relowner))) AS privilege \ + WHERE heads.oid = to_regclass('awa.queue_claim_heads') \ + AND privilege.grantee <> 0 \ + AND privilege.privilege_type IN ('INSERT', 'UPDATE', 'DELETE') \ + GROUP BY privilege.grantee \ + HAVING count(DISTINCT privilege.privilege_type) = 3 \ + ) AS runtime \ + WHERE NOT COALESCE(has_function_privilege( \ + runtime.grantee, to_regprocedure('awa.analyze_queue_storage_relation(text,text)'), 'EXECUTE'), false)" + }; +} + /// SQL returning one `text` row per object or schema that still needs the /// named patch, for patches that touch more than one place. fn patch_targets_sql(name: &str) -> Option<&'static str> { @@ -303,6 +337,10 @@ fn patch_targets_sql(name: &str) -> Option<&'static str> { prepared_lane_sequences_pending_sql!(), " ORDER BY 1" )), + "analyze_queue_storage_relation" => Some(concat!( + analyze_queue_storage_relation_pending_sql!(), + " ORDER BY 1" + )), _ => None, } } @@ -349,9 +387,13 @@ pub const SCHEMA_PATCHES: &[SchemaPatch] = &[ }, SchemaPatch { name: "analyze_queue_storage_relation", - description: "Planner statistics helper for tables autovacuum never analyses (awa 0.7 migration v053)", + description: "Planner statistics helper for tables autovacuum never analyses (awa 0.7 migration v054)", sql: include_str!("../migrations/patches/analyze_queue_storage_relation.sql"), - applied_probe: "SELECT to_regprocedure('awa.analyze_queue_storage_relation(text,text)') IS NOT NULL", + applied_probe: concat!( + "SELECT NOT EXISTS (", + analyze_queue_storage_relation_pending_sql!(), + ")" + ), }, ]; diff --git a/awa-worker/src/statistics.rs b/awa-worker/src/statistics.rs index b808f9cce..4de61e80a 100644 --- a/awa-worker/src/statistics.rs +++ b/awa-worker/src/statistics.rs @@ -116,6 +116,13 @@ impl PassSummary { } } +#[derive(Debug, PartialEq, Eq)] +enum HelperAccess { + Available, + Missing, + Denied, +} + /// Refreshes planner statistics for one queue-storage schema. pub(crate) struct StatisticsRefresher { pool: PgPool, @@ -205,9 +212,9 @@ impl StatisticsRefresher { } } - match self.helper_installed().await { - Ok(true) => {} - Ok(false) => { + match self.helper_access().await { + Ok(HelperAccess::Available) => {} + Ok(HelperAccess::Missing) => { self.warn_unavailable_once(); self.record( RelationScope::Pass, @@ -217,8 +224,23 @@ impl StatisticsRefresher { summary.record(AnalyzeOutcome::Unavailable); return summary; } + Ok(HelperAccess::Denied) => { + self.warn_privilege_once( + "queue_claim_heads", + "the runtime role needs EXECUTE on awa.analyze_queue_storage_relation(text, text), \ + and the login role needs INSERT, UPDATE and DELETE on this schema's queue_claim_heads; \ + restore those grants as the schema owner", + ); + self.record( + RelationScope::Pass, + AnalyzeOutcome::SkippedPrivilege, + started.elapsed(), + ); + summary.record(AnalyzeOutcome::SkippedPrivilege); + return summary; + } Err(err) => { - warn!(error = %err, schema = %self.schema, "Could not probe for the statistics helper"); + warn!(error = %err, schema = %self.schema, "Could not probe access to the statistics helper"); self.record( RelationScope::Pass, AnalyzeOutcome::Failed, @@ -277,16 +299,33 @@ impl StatisticsRefresher { .record_statistics_analyze(scope.as_str(), outcome.as_str(), elapsed); } - /// One catalog lookup per pass. A missing helper (a binary running ahead - /// of the schema) then costs this query instead of a failing call, and a - /// server-side ERROR line, per relation; the pass interval is the - /// re-probe interval. - async fn helper_installed(&self) -> Result { - sqlx::query_scalar( - "SELECT to_regprocedure('awa.analyze_queue_storage_relation(text,text)') IS NOT NULL", + /// Check the helper and its caller privileges once per pass, so a missing + /// migration or grant does not produce a failing call for every relation. + async fn helper_access(&self) -> Result { + let (installed, allowed): (bool, bool) = sqlx::query_as( + r#" + SELECT helper IS NOT NULL, + COALESCE( + has_function_privilege(helper, 'EXECUTE') + AND has_table_privilege(session_user, heads, 'INSERT') + AND has_table_privilege(session_user, heads, 'UPDATE') + AND has_table_privilege(session_user, heads, 'DELETE'), + false + ) + FROM ( + SELECT to_regprocedure('awa.analyze_queue_storage_relation(text,text)') AS helper, + to_regclass(format('%I.queue_claim_heads', $1::text)) AS heads + ) AS objects + "#, ) + .bind(&self.schema) .fetch_one(&self.pool) - .await + .await?; + Ok(match (installed, allowed) { + (false, _) => HelperAccess::Missing, + (true, false) => HelperAccess::Denied, + (true, true) => HelperAccess::Available, + }) } async fn ring_parents(&self) -> Result, sqlx::Error> { @@ -314,8 +353,8 @@ impl StatisticsRefresher { self.warn_privilege_once( relation, "the owner of awa.analyze_queue_storage_relation() may not ANALYZE this table; \ - run migrations as the role that owns the queue-storage tables, or grant the \ - function owner MAINTAIN on them (PostgreSQL 17+)", + change the function's owner to the table owner with ALTER FUNCTION, \ + or grant the function owner MAINTAIN on them (PostgreSQL 17+)", ); AnalyzeOutcome::SkippedPrivilege } @@ -503,4 +542,93 @@ mod tests { .await .expect("drop database"); } + + /// Missing EXECUTE or login-role DML is a single skipped pass, rather + /// than fifteen server-side permission errors. + #[tokio::test] + async fn missing_helper_privileges_skip_the_pass_without_calling_it() { + let url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string()); + let options = PgConnectOptions::from_str(&url).expect("parse DATABASE_URL"); + let admin = PgPoolOptions::new() + .max_connections(1) + .connect_with(options.clone()) + .await + .expect("connect admin"); + let database = format!("awa_stats_access_unit_{}", std::process::id()); + let role = format!("awa_stats_access_{}", std::process::id()); + sqlx::raw_sql(&format!("DROP DATABASE IF EXISTS {database} WITH (FORCE)")) + .execute(&admin) + .await + .expect("drop leftover database"); + sqlx::raw_sql(&format!("DROP ROLE IF EXISTS {role}; CREATE ROLE {role} LOGIN PASSWORD 'statistics_access_test'")) + .execute(&admin) + .await + .expect("create test role"); + sqlx::raw_sql(&format!("CREATE DATABASE {database}")) + .execute(&admin) + .await + .expect("create database"); + let owner = PgPoolOptions::new() + .max_connections(2) + .connect_with(options.clone().database(&database)) + .await + .expect("connect owner"); + awa_model::migrations::run(&owner).await.expect("migrate"); + sqlx::raw_sql(&format!( + "GRANT USAGE ON SCHEMA awa TO {role}; \ + GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA awa TO {role}; \ + REVOKE EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) FROM {role}" + )) + .execute(&owner) + .await + .expect("grant head-table privileges only"); + let runtime = PgPoolOptions::new() + .max_connections(2) + .connect_with( + options + .database(&database) + .username(&role) + .password("statistics_access_test"), + ) + .await + .expect("connect runtime"); + let refresher = StatisticsRefresher::new( + runtime.clone(), + "awa".to_string(), + AwaMetrics::from_global(), + false, + ); + let denied = PassSummary { + skipped_privilege: 1, + ..PassSummary::default() + }; + assert_eq!(refresher.run_pass().await, denied); + sqlx::raw_sql(&format!( + "GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) TO {role}" + )) + .execute(&owner) + .await + .expect("grant helper EXECUTE"); + assert_eq!(refresher.run_pass().await.ok, 15); + sqlx::raw_sql(&format!( + "REVOKE INSERT ON awa.queue_claim_heads FROM {role}" + )) + .execute(&owner) + .await + .expect("revoke login-role head privilege"); + assert_eq!(refresher.run_pass().await, denied); + + runtime.close().await; + owner.close().await; + sqlx::raw_sql(&format!("DROP DATABASE {database} WITH (FORCE)")) + .execute(&admin) + .await + .expect("drop database"); + sqlx::raw_sql(&format!("DROP ROLE {role}")) + .execute(&admin) + .await + .expect("drop test role"); + admin.close().await; + } } diff --git a/awa/tests/statistics_refresh_test.rs b/awa/tests/statistics_refresh_test.rs index 69bd3aebb..41d9d8185 100644 --- a/awa/tests/statistics_refresh_test.rs +++ b/awa/tests/statistics_refresh_test.rs @@ -811,3 +811,99 @@ async fn statistics_refresh_does_not_stop_prune_reclaiming_slots() { )); database.drop().await; } + +/// The versionless patch must repair an existing older helper, its public +/// grant, and missing runtime grants without changing the schema version. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn migration_repairs_a_stale_statistics_helper_and_missing_runtime_grants() { + let suffix = uuid::Uuid::new_v4().simple().to_string(); + let owner = format!("stats_owner_{suffix}"); + let runtime = format!("stats_runtime_{suffix}"); + let database = TestDatabase::create(Some(&owner), &[&runtime]).await; + let runtime_pool = database.runtime_pool(&runtime).await; + let owner_pool = &database.owner_pool; + run_sql( + owner_pool, + format!( + "CREATE OR REPLACE FUNCTION awa.analyze_queue_storage_relation(p_schema text, p_relation text) \ + RETURNS text LANGUAGE plpgsql SECURITY DEFINER AS $$ BEGIN RETURN 'stale'; END $$; \ + GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) TO PUBLIC; \ + REVOKE EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) FROM {runtime}" + ), + ) + .await + .expect("install an older helper with PUBLIC EXECUTE"); + assert!(migrations::pending_schema_patches(owner_pool) + .await + .expect("pending patches") + .iter() + .any(|patch| patch.name == "analyze_queue_storage_relation")); + migrations::run(owner_pool) + .await + .expect("repair the helper"); + assert_eq!( + call_helper(&runtime_pool, "awa", "queue_claim_heads") + .await + .expect("runtime helper"), + "analyzed" + ); + let public_execute: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_proc AS helper, \ + LATERAL aclexplode(COALESCE(helper.proacl, acldefault('f', helper.proowner))) AS privilege \ + WHERE helper.oid = 'awa.analyze_queue_storage_relation(text,text)'::regprocedure \ + AND privilege.grantee = 0 AND privilege.privilege_type = 'EXECUTE')" + ).fetch_one(owner_pool).await.expect("read PUBLIC grant"); + assert!(!public_execute, "the patch revokes PUBLIC EXECUTE"); + assert!(!migrations::pending_schema_patches(owner_pool) + .await + .expect("pending after repair") + .iter() + .any(|patch| patch.name == "analyze_queue_storage_relation")); + + run_sql( + owner_pool, + "GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) TO PUBLIC" + .to_string(), + ) + .await + .expect("restore the obsolete PUBLIC grant"); + assert!(migrations::pending_schema_patches(owner_pool) + .await + .expect("pending PUBLIC grant") + .iter() + .any(|patch| patch.name == "analyze_queue_storage_relation")); + migrations::run(owner_pool) + .await + .expect("remove obsolete PUBLIC grant"); + + run_sql( + owner_pool, + format!( + "REVOKE EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) FROM {runtime}" + ), + ) + .await + .expect("remove the runtime grant"); + assert!(migrations::pending_schema_patches(owner_pool) + .await + .expect("pending missing grant") + .iter() + .any(|patch| patch.name == "analyze_queue_storage_relation")); + migrations::run(owner_pool) + .await + .expect("repair missing runtime grant"); + assert_eq!( + call_helper(&runtime_pool, "awa", "queue_claim_heads") + .await + .expect("restored runtime helper"), + "analyzed" + ); + assert_eq!( + migrations::current_version(owner_pool) + .await + .expect("schema version"), + migrations::CURRENT_VERSION + ); + runtime_pool.close().await; + database.drop().await; +} diff --git a/docs/configuration.md b/docs/configuration.md index 83715cb76..7e586fc84 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -528,7 +528,7 @@ Autovacuum does not analyse partitioned tables, so the ring parents (`ready_entr The maintenance leader runs `ANALYZE` on both when it is elected and every `statistics_refresh_interval` (default `5m`). One pass takes well under a second on a small ring and a few seconds on a ring with millions of retained rows. It has no measurable effect on claim or completion throughput. - **Locks.** Each relation is analysed in its own transaction with a 2 s `lock_timeout` and a 120 s `statement_timeout`. `SKIP_LOCKED` skips a child that ring prune holds during the per-child analysis, but sampling for the parent's statistics waits for it up to the `lock_timeout`; a call that times out is recorded as `skipped_lock` and retried on the next interval. Prune never waits for the refresh: its own 10 ms lock timeout applies, so an attempt that lands while `ANALYZE` samples the oldest slot's tables returns `blocked` and retries on its usual backoff. `ANALYZE` of a partitioned table samples every child, so that window cannot be avoided. -- **Privileges.** The refresh calls `awa.analyze_queue_storage_relation(schema, relation)` (a schema patch applied by `awa migrate`), a `SECURITY DEFINER` function owned by the role that ran it, so the runtime role needs no `MAINTAIN` grant and does not have to own the tables. It is executable only by roles that hold INSERT, UPDATE and DELETE on the queue head tables (the patch grants it to the existing ones). Each pass first probes for the function; if it is missing (run `awa migrate`) or its owner may not analyse the tables, the leader logs one warning that names what is missing and skips the pass without making failing calls; job processing is unaffected. +- **Privileges.** The refresh calls `awa.analyze_queue_storage_relation(schema, relation)`, a `SECURITY DEFINER` function installed by schema patch `analyze_queue_storage_relation`. The runtime needs EXECUTE on the helper, and the login role needs INSERT, UPDATE and DELETE on the target schema's `queue_claim_heads`; the migration grants EXECUTE to existing runtime grantees. A missing helper or caller privilege skips the pass before making calls and logs one warning. Run `awa migrate` for a missing helper; restore the grants as the schema owner for `skipped_privilege`. If the helper owner may not analyse the tables, transfer the function to the table owner or grant its owner `MAINTAIN` on PostgreSQL 17+. See the [database roles guide](security.md) for the ownership command. - **Metrics.** `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration`, by `awa.analyze.scope` (`heads`, `ring_parent`, `pass`) and `awa.analyze.outcome` (`ok`, `skipped_privilege`, `skipped_lock`, `unavailable`, `failed`). A sustained `skipped_lock` rate means something else holds the tables; any `skipped_privilege` or `unavailable` means the refresh is not running. ### Rust diff --git a/docs/security.md b/docs/security.md index 49f836711..2823c579e 100644 --- a/docs/security.md +++ b/docs/security.md @@ -224,7 +224,7 @@ The runtime also directly upserts into the descriptor catalogs on startup and on These writes are not trigger-driven; they come from `ClientBuilder::build()` / `AsyncClient.start()` and from the snapshot reporter. The broad grant already covers them. -The maintenance leader refreshes planner statistics for the queue-storage head tables and ring parents (see [Planner statistics](configuration.md#planner-statistics)). `ANALYZE` skips a table its caller may not analyse, so a runtime that does not own the tables calls `awa.analyze_queue_storage_relation(schema, relation)`, a `SECURITY DEFINER` function installed by the `analyze_queue_storage_relation` schema patch under the migrator role and executable only by roles that hold `INSERT`, `UPDATE` and `DELETE` on the head tables (the patch grants it to the existing ones; a role created later gets it from `GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa`), because the lock it takes lasts until the caller's transaction ends and would block ring prune. The runtime needs no `MAINTAIN` grant. The function only analyses the two head tables and the thirteen ring parents of a queue-storage schema that is owned by the owner of the `awa` schema, which is why migrations must run with the owner role as the effective role (step 2 above). A function owned by a role that may not analyse the tables makes every pass report `skipped_privilege` and logs one warning. +The maintenance leader refreshes planner statistics for the queue-storage head tables and ring parents (see [Planner statistics](configuration.md#planner-statistics)). `ANALYZE` skips a table its caller may not analyse, so a runtime that does not own the tables calls `awa.analyze_queue_storage_relation(schema, relation)`, a `SECURITY DEFINER` function installed by the `analyze_queue_storage_relation` schema patch under the migrator role and executable only by roles that hold `INSERT`, `UPDATE` and `DELETE` on the head tables (the patch grants it to the existing ones; a role created later gets it from `GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa`), because the lock it takes lasts until the caller's transaction ends and would block ring prune. The runtime needs no `MAINTAIN` grant. The function only analyses the two head tables and the thirteen ring parents of a queue-storage schema that is owned by the owner of the `awa` schema, which is why migrations must run with the owner role as the effective role (step 2 above). A function owned by a role that may not analyse the tables makes every pass report `skipped_privilege` and logs one warning. Transfer a wrongly owned helper with `ALTER FUNCTION awa.analyze_queue_storage_relation(text, text) OWNER TO awa_owner` as a role permitted to change both owners, or grant its owner `MAINTAIN` on the target tables on PostgreSQL 17+. Replacing the function body leaves its owner unchanged. Other PostgreSQL features used at runtime: From 0e080f95d117a9eb67404df8968008faf44b714f Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 10:01:51 +1300 Subject: [PATCH 6/6] Guard pre-17 statistics checks and distinguish privilege warnings --- CHANGELOG.md | 2 ++ awa-worker/src/statistics.rs | 19 ++++++++++++------- awa/tests/statistics_refresh_test.rs | 25 +++++++++++++++++++------ 3 files changed, 33 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d9c219a62..511d334e2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -74,6 +74,8 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- Statistics-refresh permission warnings distinguish caller grants from + helper/table ownership and ANALYZE privileges. - **Ring prune and terminal-count rollup report `Blocked` when any of their statements times out on a lock.** Prune sets a short `lock_timeout` (10 ms by default) before it takes `ACCESS EXCLUSIVE` on a slot's partitions, but diff --git a/awa-worker/src/statistics.rs b/awa-worker/src/statistics.rs index 4de61e80a..83f598911 100644 --- a/awa-worker/src/statistics.rs +++ b/awa-worker/src/statistics.rs @@ -132,7 +132,8 @@ pub(crate) struct StatisticsRefresher { /// runtime is only a standby: a pass runs only once routing has flipped to /// `schema`. require_active_schema: bool, - warned_privilege: AtomicBool, + warned_caller_privilege: AtomicBool, + warned_analyze_privilege: AtomicBool, warned_unavailable: AtomicBool, } @@ -148,7 +149,8 @@ impl StatisticsRefresher { schema, metrics, require_active_schema, - warned_privilege: AtomicBool::new(false), + warned_caller_privilege: AtomicBool::new(false), + warned_analyze_privilege: AtomicBool::new(false), warned_unavailable: AtomicBool::new(false), } } @@ -226,6 +228,7 @@ impl StatisticsRefresher { } Ok(HelperAccess::Denied) => { self.warn_privilege_once( + &self.warned_caller_privilege, "queue_claim_heads", "the runtime role needs EXECUTE on awa.analyze_queue_storage_relation(text, text), \ and the login role needs INSERT, UPDATE and DELETE on this schema's queue_claim_heads; \ @@ -351,6 +354,7 @@ impl StatisticsRefresher { "analyzed" => AnalyzeOutcome::Ok, "skipped_privilege" => { self.warn_privilege_once( + &self.warned_analyze_privilege, relation, "the owner of awa.analyze_queue_storage_relation() may not ANALYZE this table; \ change the function's owner to the table owner with ALTER FUNCTION, \ @@ -410,10 +414,11 @@ impl StatisticsRefresher { } Some(SQLSTATE_INSUFFICIENT_PRIVILEGE) => { self.warn_privilege_once( + &self.warned_analyze_privilege, relation, - "the runtime role may not execute awa.analyze_queue_storage_relation(), or lacks \ - INSERT, UPDATE and DELETE on the queue head tables; \ - GRANT EXECUTE ON FUNCTION awa.analyze_queue_storage_relation(text, text) to it", + "the statistics helper was denied access; check the helper's owner, table ownership \ + and ANALYZE privileges; also recheck runtime helper EXECUTE and login-role \ + head-table INSERT, UPDATE and DELETE grants if they changed after this pass's privilege probe", ); AnalyzeOutcome::SkippedPrivilege } @@ -435,8 +440,8 @@ impl StatisticsRefresher { ); } - fn warn_privilege_once(&self, relation: &str, missing: &str) { - if self.warned_privilege.swap(true, Ordering::Relaxed) { + fn warn_privilege_once(&self, warned: &AtomicBool, relation: &str, missing: &str) { + if warned.swap(true, Ordering::Relaxed) { return; } warn!( diff --git a/awa/tests/statistics_refresh_test.rs b/awa/tests/statistics_refresh_test.rs index 41d9d8185..fffb2c4f7 100644 --- a/awa/tests/statistics_refresh_test.rs +++ b/awa/tests/statistics_refresh_test.rs @@ -363,16 +363,29 @@ async fn non_owner_runtime_role_refreshes_statistics() { seed_ready_rows(&store, &database.owner_pool, 200).await; let runtime_pool = database.runtime_pool(&runtime).await; - let is_superuser_or_owner: bool = sqlx::query_scalar( + let server_version_num: i32 = + sqlx::query_scalar("SELECT current_setting('server_version_num')::int") + .fetch_one(&runtime_pool) + .await + .expect("read PostgreSQL version"); + // pg_maintain and the MAINTAIN privilege were introduced in PostgreSQL 17. + // Older supported servers must never evaluate either privilege check. + let privilege_query = if server_version_num >= 170000 { "SELECT rolsuper OR pg_has_role(current_user, 'pg_maintain', 'USAGE') OR has_table_privilege('awa.queue_claim_heads', 'MAINTAIN') OR (SELECT relowner = (SELECT oid FROM pg_roles WHERE rolname = current_user) FROM pg_class WHERE oid = 'awa.queue_claim_heads'::regclass) - FROM pg_roles WHERE rolname = current_user", - ) - .fetch_one(&runtime_pool) - .await - .expect("check runtime role privileges"); + FROM pg_roles WHERE rolname = current_user" + } else { + "SELECT rolsuper + OR (SELECT relowner = (SELECT oid FROM pg_roles WHERE rolname = current_user) + FROM pg_class WHERE oid = 'awa.queue_claim_heads'::regclass) + FROM pg_roles WHERE rolname = current_user" + }; + let is_superuser_or_owner: bool = sqlx::query_scalar(privilege_query) + .fetch_one(&runtime_pool) + .await + .expect("check runtime role privileges"); assert!( !is_superuser_or_owner, "the runtime role must not be able to analyse the tables itself"