From 308862bd15329a004987c9ca69ecc3045a79c2d3 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Tue, 4 Aug 2026 10:47:29 +0200 Subject: [PATCH 1/3] =?UTF-8?q?fix(metrics):=20vt-aware=20oldest=5Fclaimab?= =?UTF-8?q?le=5Fage=5Fsec=20=E2=80=94=20one=20delayed=20job=20no=20longer?= =?UTF-8?q?=20reads=20as=20a=20degraded=20queue?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary pgmq's oldest_msg_age_sec ignores vt, so a scheduled or backoff-parked message made age-based latency alerts fire for hours on healthy queues. Every metrics surface now exposes oldest_claimable_age_sec (now() - min(vt) over rows with vt <= now()): DataSource (dashboard/API/MCP), Prometheus gauge, AppSignal gauge, Client#oldest_claimable_ages, CLI CLAIMABLE column. AppSignal queue_latency now derives from the claimable age (always emitted, 0 = no claimable backlog). Dashboard queue tables split depth into claimable vs parked and show the claimable age. ## Test Coverage - data_source specs: SQL selects min(vt) scoped to claimable rows; parked-only queue maps to nil - metrics_serializer spec: new gauge emitted, nil-skipped - probe spec: queue_latency from claimable age, 0 when parked-only; raw gauge unchanged - client spec: prefixed single-queue read + all-queues hash via raw connection - cli spec: CLAIMABLE column, dash for parked-only - integration spec (real PG): delayed message -> nil claimable while raw age counts it; in-flight excluded ## Verification - [x] bundle exec rubocop passes - [x] bundle exec rspec passes (only the 2 known pre-existing i18n baseline failures) - [x] bun run lint:herb passes - [x] docs rake lint passes Refs #389 --- CHANGELOG.md | 4 ++ README.md | 1 + .../pgbus/dashboard/_queues_table.html.erb | 8 +-- app/views/pgbus/queues/_queues_list.html.erb | 8 +-- app/views/pgbus/queues/show.html.erb | 2 + config/locales/da.yml | 8 ++- config/locales/de.yml | 8 ++- config/locales/en.yml | 8 ++- config/locales/es.yml | 8 ++- config/locales/fi.yml | 8 ++- config/locales/fr.yml | 8 ++- config/locales/it.yml | 8 ++- config/locales/ja.yml | 8 ++- config/locales/nb.yml | 8 ++- config/locales/nl.yml | 8 ++- config/locales/pt.yml | 8 ++- config/locales/sv.yml | 8 ++- docs/app/views/docs/pages/observability.rb | 8 ++- lib/pgbus/cli.rb | 10 ++-- lib/pgbus/client.rb | 34 +++++++++++++ lib/pgbus/integrations/appsignal/probe.rb | 12 +++-- lib/pgbus/mcp/tools/queues_tool.rb | 4 ++ lib/pgbus/web/data_source.rb | 7 ++- lib/pgbus/web/metrics_serializer.rb | 9 ++++ spec/dummy/lib/stub_data_source.rb | 12 +++-- .../integration/claimable_age_metrics_spec.rb | 42 ++++++++++++++++ spec/pgbus/cli_spec.rb | 23 +++++++-- spec/pgbus/client_spec.rb | 41 +++++++++++++++ .../integrations/appsignal/probe_spec.rb | 43 +++++++++++++--- .../web/data_source_batched_metrics_spec.rb | 50 +++++++++++++++++++ spec/pgbus/web/data_source_spec.rb | 22 ++++++++ spec/pgbus/web/metrics_serializer_spec.rb | 17 +++++-- spec/support/pgbus/stub_data_source.rb | 6 ++- 33 files changed, 400 insertions(+), 59 deletions(-) create mode 100644 spec/integration/claimable_age_metrics_spec.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index a5f203c6..8315f54e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ ## [Unreleased] +### Fixed + +- **Queue-age metrics no longer count vt-parked (scheduled/retrying) messages — one delayed job stops reading as a degraded queue (issue #389).** ⚠️ **Behavior change on the AppSignal `pgbus_queue_latency` gauge.** pgmq's `oldest_msg_age_sec` is computed from `enqueued_at` and ignores `vt`, but a job enqueued with `wait:` or parked on a long retry backoff lives in the queue table with a future `vt` — that *is* the delayed-delivery mechanism. So a single parked message made the age metric grow at wall-clock rate for hours on an otherwise drained queue, and any latency alert thresholding on it fired continuously ("oldest message is 17045s old" on a healthy queue with depth 1, `read_ct` 0). Every metrics surface now also exposes **`oldest_claimable_age_sec`** — `now() - min(vt)` over rows with `vt <= now()`, i.e. the age of the oldest message actually *eligible for pickup*: an immediately-enqueued message contributes from enqueue time (matching the old number on a plain backlog), a scheduled/backoff-parked message contributes nothing until due, an in-flight message (vt pushed forward) is excluded, and nil means "no claimable backlog" even when the table is non-empty. Surfaces: `Web::DataSource` (dashboard, JSON API, MCP `pgbus_queues` tool), a new Prometheus gauge `pgbus_queue_oldest_claimable_age_seconds`, a new AppSignal gauge `pgbus_queue_oldest_claimable_age_seconds`, `Pgbus::Client#oldest_claimable_ages` (raw-SQL reader, since pgmq's `metrics_result` type is frozen upstream), and a CLAIMABLE column in `pgbus queues`. The AppSignal **`pgbus_queue_latency` gauge now derives from the claimable age** and always emits — `(claimable_age || 0) * 1000`, 0 = no claimable backlog — so existing latency alerts stop false-firing with no dashboard changes; the raw `pgbus_queue_oldest_message_age_seconds` gauge keeps its enqueue-time semantics everywhere. The dashboard queue tables additionally split depth into **Parked** (`depth − visible`) and show the claimable age in place of the raw age, so a queue holding only backoff retries reads visibly healthy. Refs #389. + ### Added - **Health-checked rolling restarts for the job container (issue #386).** ⚠️ **Behavior change on the standalone `/readyz`.** The supervisor's `health_port` server previously answered `/readyz` with the cluster-wide HealthAnalyzer verdict — so during a rolling deploy a freshly-booted container could pass an orchestrator's health gate on the strength of the *old* container's still-heartbeating workers, and the old container (with all its capacity) was stopped before the new one had forked a single child. The standalone `/readyz` is now **container-local**: 200 only when *this* supervisor verified its connection, bootstrapped queues, forked every configured child, and all of them are currently alive — with 503 bodies `BOOTING` (pre-boot), `DEGRADED` (a child died and is waiting out crash-restart backoff — precisely the state a deploy gate must fail on, keeping the old container running), and `DRAINING` (stop signal received). No database access on the probe path; the supervisor publishes an immutable snapshot per monitor pass and the accept thread reads it. The Rails-mounted `Pgbus::Web::HealthApp` keeps the cluster-wide verdict unchanged. Alongside it: **`pgbus-health`**, a shipped executable probe for docker `HEALTHCHECK` blocks (plain Ruby + stdlib sockets, loads neither Bundler nor the gem — cheap at 1–5s intervals, works in curl-less images; exit 0/1/2 = healthy/unhealthy/usage), and a README "Rolling restarts (Kamal, docker)" guide covering the healthcheck block, stop-timeout alignment, overlap-window duplicate-supervisor safety, and the `read_ct`-vs-deploy-kill DLQ caveat. Refs #386. diff --git a/README.md b/README.md index cfeb82ab..c0b50e0d 100644 --- a/README.md +++ b/README.md @@ -991,6 +991,7 @@ When `config.metrics_enabled = true` (default), the dashboard exposes Prometheus | Metric | Description | |--------|-------------| +| `pgbus_queue_oldest_claimable_age_seconds` | Age of the oldest message eligible for pickup (visibility timeout elapsed) — safe to alert on: scheduled/backoff-parked messages don't count until due | | `pgbus_table_dead_tuples` | Dead tuple count per PGMQ table | | `pgbus_table_live_tuples` | Live tuple count per PGMQ table | | `pgbus_table_bloat_ratio` | Dead / (dead + live) per table | diff --git a/app/views/pgbus/dashboard/_queues_table.html.erb b/app/views/pgbus/dashboard/_queues_table.html.erb index 29a41bdf..702009e0 100644 --- a/app/views/pgbus/dashboard/_queues_table.html.erb +++ b/app/views/pgbus/dashboard/_queues_table.html.erb @@ -12,7 +12,8 @@ <%= t("pgbus.dashboard.queues_table.headers.queue") %> <%= t("pgbus.dashboard.queues_table.headers.depth") %> <%= t("pgbus.dashboard.queues_table.headers.visible") %> - <%= t("pgbus.dashboard.queues_table.headers.oldest") %> + <%= t("pgbus.dashboard.queues_table.headers.parked") %> + <%= t("pgbus.dashboard.queues_table.headers.oldest_claimable") %> <%= t("pgbus.dashboard.queues_table.headers.total") %> @@ -25,12 +26,13 @@ <%= pgbus_number(q[:queue_length]) %> <%= pgbus_number(q[:queue_visible_length]) %> - <%= q[:oldest_msg_age_sec] || "—" %> + <%= pgbus_number(q[:queue_length] - q[:queue_visible_length]) %> + <%= q[:oldest_claimable_age_sec] || "—" %> <%= pgbus_number(q[:total_messages]) %> <% end %> <% if @queues.empty? %> - <%= t("pgbus.dashboard.queues_table.empty") %> + <%= t("pgbus.dashboard.queues_table.empty") %> <% end %> diff --git a/app/views/pgbus/queues/_queues_list.html.erb b/app/views/pgbus/queues/_queues_list.html.erb index ddad678f..c5972f96 100644 --- a/app/views/pgbus/queues/_queues_list.html.erb +++ b/app/views/pgbus/queues/_queues_list.html.erb @@ -6,7 +6,8 @@ <%= t("pgbus.queues.queues_list.headers.queue") %> <%= t("pgbus.queues.queues_list.headers.depth") %> <%= t("pgbus.queues.queues_list.headers.visible") %> - <%= t("pgbus.queues.queues_list.headers.oldest") %> + <%= t("pgbus.queues.queues_list.headers.parked") %> + <%= t("pgbus.queues.queues_list.headers.oldest_claimable") %> <%= t("pgbus.queues.queues_list.headers.newest") %> <%= t("pgbus.queues.queues_list.headers.total_ever") %> <%= t("pgbus.queues.queues_list.headers.actions") %> @@ -24,7 +25,8 @@ <%= pgbus_number(q[:queue_length]) %> <%= pgbus_number(q[:queue_visible_length]) %> - <%= q[:oldest_msg_age_sec] || "—" %> + <%= pgbus_number(q[:queue_length] - q[:queue_visible_length]) %> + <%= q[:oldest_claimable_age_sec] || "—" %> <%= q[:newest_msg_age_sec] || "—" %> <%= pgbus_number(q[:total_messages]) %> @@ -51,7 +53,7 @@ <% end %> <% if @queues.empty? %> - <%= t("pgbus.queues.queues_list.empty") %> + <%= t("pgbus.queues.queues_list.empty") %> <% end %> diff --git a/app/views/pgbus/queues/show.html.erb b/app/views/pgbus/queues/show.html.erb index 1225593b..ebbb1975 100644 --- a/app/views/pgbus/queues/show.html.erb +++ b/app/views/pgbus/queues/show.html.erb @@ -13,6 +13,8 @@

<%= t("pgbus.queues.show.depth") %> <%= @queue[:queue_length] %> | <%= t("pgbus.queues.show.visible") %> <%= @queue[:queue_visible_length] %> | + <%= t("pgbus.queues.show.parked") %> <%= @queue[:queue_length] - @queue[:queue_visible_length] %> | + <%= t("pgbus.queues.show.oldest_claimable") %> <%= @queue[:oldest_claimable_age_sec] || "—" %> | <%= t("pgbus.queues.show.total") %> <%= pgbus_number(@queue[:total_messages]) %>

<% end %> diff --git a/config/locales/da.yml b/config/locales/da.yml index 4c77443b..58eb3eb2 100644 --- a/config/locales/da.yml +++ b/config/locales/da.yml @@ -65,7 +65,8 @@ da: empty: Ingen køer fundet headers: depth: Dybde - oldest: Ældste (s) + oldest_claimable: Ældste tilgængelig (s) + parked: Parkeret queue: Kø total: Total visible: Synlig @@ -475,7 +476,8 @@ da: actions: Handlinger depth: Dybde newest: Nyeste (s) - oldest: Ældste (s) + oldest_claimable: Ældste tilgængelig (s) + parked: Parkeret queue: Kø total_ever: Total nogensinde visible: Synlig @@ -516,6 +518,8 @@ da: scheduled: 'Planlagt:' timezone: 'Tidszone:' visible_at: 'Synlig fra:' + oldest_claimable: 'Ældste tilgængelig:' + parked: 'Parkeret:' pause: Pause pause_confirm: Pause behandling? purge_confirm: Rens alle beskeder? diff --git a/config/locales/de.yml b/config/locales/de.yml index 13fd0da0..5cfbcc9b 100644 --- a/config/locales/de.yml +++ b/config/locales/de.yml @@ -65,7 +65,8 @@ de: empty: Keine Warteschlangen gefunden headers: depth: Tiefe - oldest: Älteste (s) + oldest_claimable: Älteste verfügbar (s) + parked: Geparkt queue: Warteschlange total: Gesamt visible: Sichtbar @@ -475,7 +476,8 @@ de: actions: Aktionen depth: Tiefe newest: Neueste (s) - oldest: Älteste (s) + oldest_claimable: Älteste verfügbar (s) + parked: Geparkt queue: Warteschlange total_ever: Insgesamt jemals visible: Sichtbar @@ -516,6 +518,8 @@ de: scheduled: 'Geplant:' timezone: 'Zeitzone:' visible_at: 'Sichtbar ab:' + oldest_claimable: 'Älteste verfügbar:' + parked: 'Geparkt:' pause: Pause pause_confirm: Verarbeitung pausieren? purge_confirm: Alle Nachrichten löschen? diff --git a/config/locales/en.yml b/config/locales/en.yml index d510e2cc..79dbc532 100644 --- a/config/locales/en.yml +++ b/config/locales/en.yml @@ -65,7 +65,8 @@ en: empty: No queues found headers: depth: Depth - oldest: Oldest (s) + oldest_claimable: Oldest claimable (s) + parked: Parked queue: Queue total: Total visible: Visible @@ -475,7 +476,8 @@ en: actions: Actions depth: Depth newest: Newest (s) - oldest: Oldest (s) + oldest_claimable: Oldest claimable (s) + parked: Parked queue: Queue total_ever: Total Ever visible: Visible @@ -516,6 +518,8 @@ en: scheduled: 'Scheduled at:' timezone: 'Timezone:' visible_at: 'Visible at:' + oldest_claimable: 'Oldest claimable:' + parked: 'Parked:' pause: Pause pause_confirm: Pause processing? purge_confirm: Purge all messages? diff --git a/config/locales/es.yml b/config/locales/es.yml index 4ee6de0d..4c3190c5 100644 --- a/config/locales/es.yml +++ b/config/locales/es.yml @@ -65,7 +65,8 @@ es: empty: No se encontraron colas headers: depth: Profundidad - oldest: El más antiguo (s) + oldest_claimable: Más antiguo disponible (s) + parked: Aparcados queue: Cola total: Total visible: Visible @@ -475,7 +476,8 @@ es: actions: Acciones depth: Profundidad newest: Más nuevo (s) - oldest: Más antiguo (s) + oldest_claimable: Más antiguo disponible (s) + parked: Aparcados queue: Cola total_ever: Total acumulado visible: Visible @@ -516,6 +518,8 @@ es: scheduled: 'Programado:' timezone: 'Zona horaria:' visible_at: 'Visible en:' + oldest_claimable: 'Más antiguo disponible:' + parked: 'Aparcados:' pause: Pausar pause_confirm: "¿Pausar el procesamiento?" purge_confirm: "¿Purgar todos los mensajes?" diff --git a/config/locales/fi.yml b/config/locales/fi.yml index 1cc22ccc..c3d4d4a2 100644 --- a/config/locales/fi.yml +++ b/config/locales/fi.yml @@ -65,7 +65,8 @@ fi: empty: Jonot eivät löytyneet headers: depth: Syvyys - oldest: Vanhin (s) + oldest_claimable: Vanhin saatavilla (s) + parked: Pysäköidyt queue: Jono total: Yhteensä visible: Näkyvissä @@ -475,7 +476,8 @@ fi: actions: Toiminnot depth: Syvyys newest: Uusimmat (s) - oldest: Vanhimmat (s) + oldest_claimable: Vanhin saatavilla (s) + parked: Pysäköidyt queue: Jono total_ever: Yhteensä koskaan visible: Näkyvissä @@ -516,6 +518,8 @@ fi: scheduled: 'Aikataulutettu:' timezone: 'Aikavyöhyke:' visible_at: 'Näkyvissä:' + oldest_claimable: 'Vanhin saatavilla:' + parked: 'Pysäköidyt:' pause: Tauko pause_confirm: Keskeytetäänkö käsittely? purge_confirm: Tyhjennetäänkö kaikki viestit? diff --git a/config/locales/fr.yml b/config/locales/fr.yml index d5aa9ec6..48864770 100644 --- a/config/locales/fr.yml +++ b/config/locales/fr.yml @@ -65,7 +65,8 @@ fr: empty: Aucune file d'attente trouvée headers: depth: Profondeur - oldest: Le plus ancien (s) + oldest_claimable: Plus ancien disponible (s) + parked: Différés queue: File d'attente total: Total visible: Visible @@ -475,7 +476,8 @@ fr: actions: Actions depth: Profondeur newest: Le plus récent (s) - oldest: Le plus ancien (s) + oldest_claimable: Plus ancien disponible (s) + parked: Différés queue: File d'attente total_ever: Total jamais visible: Visible @@ -516,6 +518,8 @@ fr: scheduled: 'Planifié :' timezone: 'Fuseau horaire :' visible_at: 'Visible à :' + oldest_claimable: 'Plus ancien disponible :' + parked: 'Différés :' pause: Pause pause_confirm: Mettre en pause le traitement ? purge_confirm: Purger tous les messages ? diff --git a/config/locales/it.yml b/config/locales/it.yml index 46717511..ec351ecf 100644 --- a/config/locales/it.yml +++ b/config/locales/it.yml @@ -65,7 +65,8 @@ it: empty: Nessuna coda trovata headers: depth: Profondità - oldest: Più vecchio (s) + oldest_claimable: Più vecchio disponibile (s) + parked: Posticipati queue: Coda total: Totale visible: Visibile @@ -475,7 +476,8 @@ it: actions: Azioni depth: Profondità newest: Più recente (s) - oldest: Più vecchio (s) + oldest_claimable: Più vecchio disponibile (s) + parked: Posticipati queue: Coda total_ever: Totale mai visible: Visibile @@ -516,6 +518,8 @@ it: scheduled: 'Programmato:' timezone: 'Fuso orario:' visible_at: 'Visibile alle:' + oldest_claimable: 'Più vecchio disponibile:' + parked: 'Posticipati:' pause: Pausa pause_confirm: Mettere in pausa l'elaborazione? purge_confirm: Eliminare tutti i messaggi? diff --git a/config/locales/ja.yml b/config/locales/ja.yml index 6b08c43c..aa962368 100644 --- a/config/locales/ja.yml +++ b/config/locales/ja.yml @@ -65,7 +65,8 @@ ja: empty: キューが見つかりません headers: depth: 深さ - oldest: 最古 (秒) + oldest_claimable: 取得可能な最古 (秒) + parked: 待機中 queue: キュー total: 合計 visible: 表示中 @@ -475,7 +476,8 @@ ja: actions: アクション depth: 深さ newest: 最新 (秒) - oldest: 最古 (秒) + oldest_claimable: 取得可能な最古 (秒) + parked: 待機中 queue: キュー total_ever: 合計数 visible: 表示中 @@ -516,6 +518,8 @@ ja: scheduled: スケジュール済み： timezone: タイムゾーン： visible_at: 表示可能日時： + oldest_claimable: '取得可能な最古:' + parked: '待機中:' pause: 一時停止 pause_confirm: 処理を一時停止しますか？ purge_confirm: すべてのメッセージを削除しますか？ diff --git a/config/locales/nb.yml b/config/locales/nb.yml index 8ec75aed..9cf90204 100644 --- a/config/locales/nb.yml +++ b/config/locales/nb.yml @@ -65,7 +65,8 @@ nb: empty: Ingen køer funnet headers: depth: Dybde - oldest: Eldste (s) + oldest_claimable: Eldste tilgjengelig (s) + parked: Parkert queue: Kø total: Totalt visible: Synlig @@ -475,7 +476,8 @@ nb: actions: Handlinger depth: Dybde newest: Nyeste (s) - oldest: Eldste (s) + oldest_claimable: Eldste tilgjengelig (s) + parked: Parkert queue: Kø total_ever: Totalt noensinne visible: Synlig @@ -516,6 +518,8 @@ nb: scheduled: 'Planlagt:' timezone: 'Tidssone:' visible_at: 'Synlig fra:' + oldest_claimable: 'Eldste tilgjengelig:' + parked: 'Parkert:' pause: Pause pause_confirm: Pause behandling? purge_confirm: Rens alle meldinger? diff --git a/config/locales/nl.yml b/config/locales/nl.yml index b45df4bb..5e49eaa9 100644 --- a/config/locales/nl.yml +++ b/config/locales/nl.yml @@ -65,7 +65,8 @@ nl: empty: Geen wachtrijen gevonden headers: depth: Diepte - oldest: Oudste (s) + oldest_claimable: Oudste beschikbaar (s) + parked: Geparkeerd queue: Wachtrij total: Totaal visible: Zichtbaar @@ -475,7 +476,8 @@ nl: actions: Acties depth: Diepte newest: Nieuwste (s) - oldest: Oudste (s) + oldest_claimable: Oudste beschikbaar (s) + parked: Geparkeerd queue: Wachtrij total_ever: Totaal ooit visible: Zichtbaar @@ -516,6 +518,8 @@ nl: scheduled: 'Gepland:' timezone: 'Tijdzone:' visible_at: 'Zichtbaar op:' + oldest_claimable: 'Oudste beschikbaar:' + parked: 'Geparkeerd:' pause: Pauzeren pause_confirm: Verwerking pauzeren? purge_confirm: Alle berichten verwijderen? diff --git a/config/locales/pt.yml b/config/locales/pt.yml index 8f824c6e..1a88854a 100644 --- a/config/locales/pt.yml +++ b/config/locales/pt.yml @@ -65,7 +65,8 @@ pt: empty: Nenhuma fila encontrada headers: depth: Profundidade - oldest: Mais antigo (s) + oldest_claimable: Mais antigo disponível (s) + parked: Adiados queue: Fila total: Total visible: Visível @@ -475,7 +476,8 @@ pt: actions: Ações depth: Profundidade newest: Mais novo (s) - oldest: Mais antigo (s) + oldest_claimable: Mais antigo disponível (s) + parked: Adiados queue: Fila total_ever: Total de todos os tempos visible: Visível @@ -516,6 +518,8 @@ pt: scheduled: 'Agendado:' timezone: 'Fuso horário:' visible_at: 'Visível em:' + oldest_claimable: 'Mais antigo disponível:' + parked: 'Adiados:' pause: Pausar pause_confirm: Pausar processamento? purge_confirm: Limpar todas as mensagens? diff --git a/config/locales/sv.yml b/config/locales/sv.yml index b0c617a5..9b7b374b 100644 --- a/config/locales/sv.yml +++ b/config/locales/sv.yml @@ -65,7 +65,8 @@ sv: empty: Inga köer hittades headers: depth: Djup - oldest: Äldst (s) + oldest_claimable: Äldsta tillgänglig (s) + parked: Parkerade queue: Kö total: Totalt visible: Synliga @@ -475,7 +476,8 @@ sv: actions: Åtgärder depth: Djup newest: Nyaste (s) - oldest: Äldsta (s) + oldest_claimable: Äldsta tillgänglig (s) + parked: Parkerade queue: Kö total_ever: Totalt någonsin visible: Synliga @@ -516,6 +518,8 @@ sv: scheduled: 'Schemalagt:' timezone: 'Tidszon:' visible_at: 'Synlig vid:' + oldest_claimable: 'Äldsta tillgänglig:' + parked: 'Parkerade:' pause: Pausa pause_confirm: Pausa bearbetning? purge_confirm: Rensa alla meddelanden? diff --git a/docs/app/views/docs/pages/observability.rb b/docs/app/views/docs/pages/observability.rb index efe9f384..e3ac0799 100644 --- a/docs/app/views/docs/pages/observability.rb +++ b/docs/app/views/docs/pages/observability.rb @@ -144,7 +144,13 @@ def appsignal Load the `appsignal` gem and pgbus auto-installs a subscriber and a minutely probe — background-job transactions for every job and handler, `pgbus_` counters and distributions, and gauges for queue depth, oldest-message age, - DLQ depth, dead tuples, and MVCC horizon. Four importable dashboards ship + DLQ depth, dead tuples, and MVCC horizon. `pgbus_queue_latency` is computed + from the oldest *claimable* message (visibility timeout elapsed), so a + queue holding only scheduled or backoff-parked jobs reads 0 — alert on it + without false positives from one delayed job; the raw + `pgbus_queue_oldest_message_age_seconds` gauge keeps enqueue-time + semantics, and `pgbus_queue_oldest_claimable_age_seconds` reports the + claimable age itself. Four importable dashboards ship with the gem — `pgbus dashboard --list` enumerates them and `pgbus dashboard ` prints import-ready JSON for AppSignal's "Import dashboard" dialog. Opt out with `config.appsignal_enabled = false`. diff --git a/lib/pgbus/cli.rb b/lib/pgbus/cli.rb index 0ab6843b..13d0c6e7 100644 --- a/lib/pgbus/cli.rb +++ b/lib/pgbus/cli.rb @@ -177,14 +177,16 @@ def run_doctor def list_queues Pgbus.client.list_queues metrics = Pgbus.client.metrics + claimable_ages = Pgbus.client.oldest_claimable_ages - puts "QUEUE DEPTH VISIBLE OLDEST (s) TOTAL " - puts "-" * 95 + puts "QUEUE DEPTH VISIBLE OLDEST (s) CLAIMABLE (s) TOTAL " + puts "-" * 111 Array(metrics).each do |m| - puts format("%-40s %-10s %-10s %-15s %-15s", + puts format("%-40s %-10s %-10s %-15s %-15s %-15s", m.queue_name, m.queue_length, m.queue_visible_length, - m.oldest_msg_age_sec || "-", m.total_messages) + m.oldest_msg_age_sec || "-", claimable_ages[m.queue_name] || "-", + m.total_messages) end end diff --git a/lib/pgbus/client.rb b/lib/pgbus/client.rb index 7b10900a..aa584e17 100644 --- a/lib/pgbus/client.rb +++ b/lib/pgbus/client.rb @@ -543,6 +543,28 @@ def metrics(queue_name = nil) end end + # Age (seconds) of the oldest message actually eligible for pickup, i.e. + # whose visibility timeout has elapsed. Unlike pgmq's oldest_msg_age_sec + # (computed from enqueued_at), a scheduled or backoff-parked message — + # future vt — contributes nothing until it comes due, so a queue holding + # only parked messages reads nil ("no claimable backlog") instead of an + # age growing at wall-clock rate (issue #389). pgmq's metrics_result type + # is frozen upstream, so this lives here rather than in the SQL function. + # + # With a queue name: the age for that (prefixed) queue, or nil. + # Without: a hash of every physical queue in pgmq.meta to its age. + def oldest_claimable_ages(queue_name = nil) + with_raw_connection do |conn| + if queue_name + claimable_age_for(conn, config.queue_name(queue_name)) + else + names = conn.exec("SELECT queue_name FROM pgmq.meta ORDER BY queue_name") + .map { |row| row["queue_name"] } + names.to_h { |name| [name, claimable_age_for(conn, name)] } + end + end + end + # Snapshot of the PGMQ connection pool: {size:, available:, pool_timeout:}. # # Reads pgmq-ruby's own pool counters (@pgmq.stats -> {size:, available:}) @@ -948,6 +970,18 @@ def install_pgmq_schema(conn) end end + # queue_name is a physical (already prefixed) queue name; sanitized to a + # bare identifier before interpolation, same as the dashboard's DataSource. + def claimable_age_for(conn, queue_name) + qtable = "q_#{QueueNameValidator.sanitize!(queue_name)}" + row = conn.exec(<<~SQL).first + SELECT EXTRACT(epoch FROM (NOW() - min(vt)))::int AS age_sec + FROM pgmq.#{qtable} + WHERE vt <= NOW() + SQL + row && row["age_sec"]&.to_i + end + def with_raw_connection opts = config.connection_options owned = false diff --git a/lib/pgbus/integrations/appsignal/probe.rb b/lib/pgbus/integrations/appsignal/probe.rb index 9b155af8..ce5ea321 100644 --- a/lib/pgbus/integrations/appsignal/probe.rb +++ b/lib/pgbus/integrations/appsignal/probe.rb @@ -85,10 +85,14 @@ def track_queues gauge "queue_visible_depth", q[:queue_visible_length], tags gauge "queue_paused", q[:paused] ? 1 : 0, tags age = q[:oldest_msg_age_sec] - if age - gauge "queue_oldest_message_age_seconds", age, tags - gauge "queue_latency", age * 1_000, tags - end + gauge "queue_oldest_message_age_seconds", age, tags if age + claimable_age = q[:oldest_claimable_age_sec] + gauge "queue_oldest_claimable_age_seconds", claimable_age, tags if claimable_age + # Latency = time the oldest *claimable* message has waited for + # pickup; a queue holding only vt-parked (scheduled/backoff) + # messages is healthy, so 0 — not the raw enqueued_at age, which + # grows at wall-clock rate on a parked message (issue #389). + gauge "queue_latency", (claimable_age || 0) * 1_000, tags end rescue StandardError => e log_failure("queue metrics", e) diff --git a/lib/pgbus/mcp/tools/queues_tool.rb b/lib/pgbus/mcp/tools/queues_tool.rb index 261c2d7d..db8f652b 100644 --- a/lib/pgbus/mcp/tools/queues_tool.rb +++ b/lib/pgbus/mcp/tools/queues_tool.rb @@ -14,6 +14,10 @@ class QueuesTool < BaseTool (messages whose visibility timeout has expired and are ready to be claimed), oldest/newest message age in seconds, lifetime total, and paused state. Use this to answer "are any queues backed up?". + oldest_claimable_age_sec is the age of the oldest message actually + eligible for pickup — nil means no claimable backlog, so a queue whose + oldest_msg_age_sec keeps growing while oldest_claimable_age_sec is nil + holds only scheduled/backoff-parked messages and is healthy. DESC input_schema(properties: {}, required: []) diff --git a/lib/pgbus/web/data_source.rb b/lib/pgbus/web/data_source.rb index c9271834..b9f596c8 100644 --- a/lib/pgbus/web/data_source.rb +++ b/lib/pgbus/web/data_source.rb @@ -1177,7 +1177,8 @@ def batched_queue_metrics(queue_names) (SELECT EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int FROM pgmq.#{qtable}) AS oldest_msg_age_sec, (SELECT CASE WHEN is_called THEN last_value ELSE 0 END FROM pgmq.#{seq_name}) AS total_messages, (SELECT max(read_ct) FROM pgmq.#{qtable}) AS max_read_ct, - (SELECT count(*) FROM pgmq.#{qtable} WHERE vt <= NOW() AND read_ct = 0) AS visible_unread_length + (SELECT count(*) FROM pgmq.#{qtable} WHERE vt <= NOW() AND read_ct = 0) AS visible_unread_length, + (SELECT EXTRACT(epoch FROM (NOW() - min(vt)))::int FROM pgmq.#{qtable} WHERE vt <= NOW()) AS oldest_claimable_age_sec SQL rescue StandardError => e Pgbus.logger.debug { "[Pgbus::Web] Skipping queue metrics for #{name}: #{e.message}" } @@ -1194,6 +1195,7 @@ def batched_queue_metrics(queue_names) queue_length: row["queue_length"].to_i, queue_visible_length: row["queue_visible_length"].to_i, oldest_msg_age_sec: row["oldest_msg_age_sec"]&.to_i, + oldest_claimable_age_sec: row["oldest_claimable_age_sec"]&.to_i, newest_msg_age_sec: row["newest_msg_age_sec"]&.to_i, total_messages: row["total_messages"].to_i, max_read_ct: row["max_read_ct"]&.to_i, @@ -1216,6 +1218,7 @@ def queue_metrics_via_sql(queue_name) count(CASE WHEN vt <= NOW() THEN 1 END) AS queue_visible_length, EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int AS newest_msg_age_sec, EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int AS oldest_msg_age_sec, + EXTRACT(epoch FROM (NOW() - min(vt) FILTER (WHERE vt <= NOW())))::int AS oldest_claimable_age_sec, max(read_ct) AS max_read_ct, count(CASE WHEN vt <= NOW() AND read_ct = 0 THEN 1 END) AS visible_unread_length FROM pgmq.#{qtable} @@ -1229,6 +1232,7 @@ def queue_metrics_via_sql(queue_name) q_summary.queue_visible_length, q_summary.newest_msg_age_sec, q_summary.oldest_msg_age_sec, + q_summary.oldest_claimable_age_sec, q_summary.max_read_ct, q_summary.visible_unread_length, all_metrics.total_messages @@ -1242,6 +1246,7 @@ def queue_metrics_via_sql(queue_name) queue_length: row["queue_length"].to_i, queue_visible_length: row["queue_visible_length"].to_i, oldest_msg_age_sec: row["oldest_msg_age_sec"]&.to_i, + oldest_claimable_age_sec: row["oldest_claimable_age_sec"]&.to_i, newest_msg_age_sec: row["newest_msg_age_sec"]&.to_i, total_messages: row["total_messages"].to_i, max_read_ct: row["max_read_ct"]&.to_i, diff --git a/lib/pgbus/web/metrics_serializer.rb b/lib/pgbus/web/metrics_serializer.rb index adbd85c5..aa6347dc 100644 --- a/lib/pgbus/web/metrics_serializer.rb +++ b/lib/pgbus/web/metrics_serializer.rb @@ -54,6 +54,15 @@ def append_queue_metrics(lines) end end + gauge(lines, "pgbus_queue_oldest_claimable_age_seconds", + "Age of the oldest message eligible for pickup (visibility timeout elapsed)") do + queues.filter_map do |q| + next unless q[:oldest_claimable_age_sec] + + [q[:oldest_claimable_age_sec], { queue: q[:name] }] + end + end + gauge(lines, "pgbus_queue_paused", "Whether the queue is paused (1) or active (0)") do queues.map { |q| [q[:paused] ? 1 : 0, { queue: q[:name] }] } end diff --git a/spec/dummy/lib/stub_data_source.rb b/spec/dummy/lib/stub_data_source.rb index c28ce911..ef45d461 100644 --- a/spec/dummy/lib/stub_data_source.rb +++ b/spec/dummy/lib/stub_data_source.rb @@ -19,13 +19,17 @@ def summary_stats def queues_with_metrics [ { name: "pgbus_default", queue_length: 85, queue_visible_length: 62, - oldest_msg_age_sec: 300, newest_msg_age_sec: 2, total_messages: 12_450 }, + oldest_msg_age_sec: 300, oldest_claimable_age_sec: 240, + newest_msg_age_sec: 2, total_messages: 12_450 }, { name: "pgbus_mailers", queue_length: 22, queue_visible_length: 18, - oldest_msg_age_sec: 45, newest_msg_age_sec: 1, total_messages: 8_320 }, + oldest_msg_age_sec: 45, oldest_claimable_age_sec: 30, + newest_msg_age_sec: 1, total_messages: 8_320 }, { name: "pgbus_events", queue_length: 15, queue_visible_length: 13, - oldest_msg_age_sec: 120, newest_msg_age_sec: 5, total_messages: 45_000 }, + oldest_msg_age_sec: 120, oldest_claimable_age_sec: 100, + newest_msg_age_sec: 5, total_messages: 45_000 }, { name: "pgbus_default_dlq", queue_length: 3, queue_visible_length: 3, - oldest_msg_age_sec: 7200, newest_msg_age_sec: 3600, total_messages: 47 } + oldest_msg_age_sec: 7200, oldest_claimable_age_sec: 7200, + newest_msg_age_sec: 3600, total_messages: 47 } ] end diff --git a/spec/integration/claimable_age_metrics_spec.rb b/spec/integration/claimable_age_metrics_spec.rb new file mode 100644 index 00000000..434c46b5 --- /dev/null +++ b/spec/integration/claimable_age_metrics_spec.rb @@ -0,0 +1,42 @@ +# frozen_string_literal: true + +require_relative "../integration_helper" + +RSpec.describe "Claimable age metrics (integration)", :integration do + let(:client) { Pgbus.client } + + before do + client.ensure_queue("claimable_test") + client.purge_queue("claimable_test") + end + + describe "#oldest_claimable_ages" do + it "reports nil for a queue holding only a delayed message, while the raw age counts it" do + client.send_message("claimable_test", { "delayed" => true }, delay: 3600) + + expect(client.oldest_claimable_ages("claimable_test")).to be_nil + + metrics = client.metrics("claimable_test") + expect(metrics.queue_length.to_i).to eq(1) + expect(metrics.oldest_msg_age_sec.to_i).to be >= 0 + end + + it "reports the age once a message is eligible for pickup" do + client.send_message("claimable_test", { "now" => true }) + + age = client.oldest_claimable_ages("claimable_test") + expect(age).to be_an(Integer) + expect(age).to be >= 0 + + all = client.oldest_claimable_ages + expect(all.fetch("pgbus_int_claimable_test")).to be_within(1).of(age) + end + + it "excludes an in-flight message whose visibility timeout was pushed forward" do + client.send_message("claimable_test", { "in_flight" => true }) + client.read_batch("claimable_test", qty: 1, vt: 60) + + expect(client.oldest_claimable_ages("claimable_test")).to be_nil + end + end +end diff --git a/spec/pgbus/cli_spec.rb b/spec/pgbus/cli_spec.rb index 57f1e0fc..0199683d 100644 --- a/spec/pgbus/cli_spec.rb +++ b/spec/pgbus/cli_spec.rb @@ -310,20 +310,37 @@ end describe ".list_queues" do - it "prints formatted table with queue metrics" do + it "prints formatted table with queue metrics including the claimable age" do metric = double("metric", queue_name: "pgbus_test_default", queue_length: 10, queue_visible_length: 8, oldest_msg_age_sec: 42, total_messages: 100) - allow(mock_client).to receive(:metrics).and_return([metric]) + allow(mock_client).to receive_messages(metrics: [metric], + oldest_claimable_ages: { "pgbus_test_default" => 12 }) output = capture_stdout { described_class.list_queues } expect(output).to include("QUEUE") + expect(output).to include("CLAIMABLE") expect(output).to include("pgbus_test_default") - expect(output).to include("10") + expect(output).to include("12") + end + + it "prints a dash for a queue holding only vt-parked messages" do + metric = double("metric", + queue_name: "pgbus_test_webhooks", + queue_length: 1, + queue_visible_length: 0, + oldest_msg_age_sec: 17_045, + total_messages: 500) + allow(mock_client).to receive_messages(metrics: [metric], + oldest_claimable_ages: { "pgbus_test_webhooks" => nil }) + + output = capture_stdout { described_class.list_queues } + + expect(output).to match(/pgbus_test_webhooks.*17045.*-/) end end diff --git a/spec/pgbus/client_spec.rb b/spec/pgbus/client_spec.rb index 6677fda5..1e397799 100644 --- a/spec/pgbus/client_spec.rb +++ b/spec/pgbus/client_spec.rb @@ -468,6 +468,47 @@ def initialize(*args, **kwargs); end end end + describe "#oldest_claimable_ages" do + let(:raw_conn) { double("PG::Connection") } + + before { allow(client).to receive(:with_raw_connection).and_yield(raw_conn) } + + context "with a queue_name" do + it "returns the vt-aware age of the oldest claimable message in the prefixed queue" do + allow(raw_conn).to receive(:exec) + .with(/min\(vt\).*FROM pgmq\.q_pgbus_test_default.*WHERE vt <= NOW\(\)/m) + .and_return([{ "age_sec" => "42" }]) + + expect(client.oldest_claimable_ages("default")).to eq(42) + end + + it "returns nil when only vt-parked (scheduled/retrying) messages remain" do + allow(raw_conn).to receive(:exec).and_return([{ "age_sec" => nil }]) + + expect(client.oldest_claimable_ages("default")).to be_nil + end + end + + context "without a queue_name" do + it "maps every queue in pgmq.meta to its claimable age" do + allow(raw_conn).to receive(:exec) + .with(/FROM pgmq\.meta/) + .and_return([{ "queue_name" => "pgbus_test_default" }, { "queue_name" => "pgbus_test_mailers" }]) + allow(raw_conn).to receive(:exec) + .with(/FROM pgmq\.q_pgbus_test_default/m) + .and_return([{ "age_sec" => "10" }]) + allow(raw_conn).to receive(:exec) + .with(/FROM pgmq\.q_pgbus_test_mailers/m) + .and_return([{ "age_sec" => nil }]) + + expect(client.oldest_claimable_ages).to eq( + "pgbus_test_default" => 10, + "pgbus_test_mailers" => nil + ) + end + end + end + describe "#pool_stats" do it "returns pgmq pool stats merged with the configured pool_timeout" do allow(mock_pgmq).to receive(:stats).and_return({ size: 5, available: 3 }) diff --git a/spec/pgbus/integrations/appsignal/probe_spec.rb b/spec/pgbus/integrations/appsignal/probe_spec.rb index d47676b5..9fc05c98 100644 --- a/spec/pgbus/integrations/appsignal/probe_spec.rb +++ b/spec/pgbus/integrations/appsignal/probe_spec.rb @@ -20,8 +20,14 @@ def self.set_gauge(name, value, tags = {}) Class.new do def queues_with_metrics [ - { name: "pgbus_default", queue_length: 42, queue_visible_length: 30, oldest_msg_age_sec: 5.0, paused: false }, - { name: "pgbus_critical", queue_length: 0, queue_visible_length: 0, oldest_msg_age_sec: nil, paused: true } + { name: "pgbus_default", queue_length: 42, queue_visible_length: 30, oldest_msg_age_sec: 5.0, + oldest_claimable_age_sec: 3.0, paused: false }, + { name: "pgbus_critical", queue_length: 0, queue_visible_length: 0, oldest_msg_age_sec: nil, + oldest_claimable_age_sec: nil, paused: true }, + # The issue #389 incident shape: one backoff-parked message (future vt) — + # raw age grows at wall-clock rate while nothing is claimable. + { name: "pgbus_webhooks", queue_length: 1, queue_visible_length: 0, oldest_msg_age_sec: 17_045.0, + oldest_claimable_age_sec: nil, paused: false } ] end @@ -95,22 +101,45 @@ def pool_stats expect(depth_gauge[2]).to eq(queue: "pgbus_default") end - it "records queue latency gauge per queue without a hostname tag" do + it "records queue latency from the claimable age, not the raw message age" do runner = Pgbus::Integrations::Appsignal::Probe::Runner.new(data_source: fake_data_source) runner.call latency = appsignal_class.gauges.find { |g| g[0] == "pgbus_queue_latency" && g[2][:queue] == "pgbus_default" } expect(latency).not_to be_nil - expect(latency[1]).to eq(5000.0) + expect(latency[1]).to eq(3000.0) expect(latency[2]).to eq(queue: "pgbus_default") end - it "skips queue latency when oldest_msg_age_sec is nil" do + it "reports zero queue latency when nothing is claimable, even with a vt-parked message" do runner = Pgbus::Integrations::Appsignal::Probe::Runner.new(data_source: fake_data_source) runner.call - critical_latency = appsignal_class.gauges.find { |g| g[0] == "pgbus_queue_latency" && g[2][:queue] == "pgbus_critical" } - expect(critical_latency).to be_nil + parked_latency = appsignal_class.gauges.find { |g| g[0] == "pgbus_queue_latency" && g[2][:queue] == "pgbus_webhooks" } + expect(parked_latency).not_to be_nil + expect(parked_latency[1]).to eq(0) + + empty_latency = appsignal_class.gauges.find { |g| g[0] == "pgbus_queue_latency" && g[2][:queue] == "pgbus_critical" } + expect(empty_latency).not_to be_nil + expect(empty_latency[1]).to eq(0) + end + + it "records the claimable age gauge only when a claimable backlog exists" do + runner = Pgbus::Integrations::Appsignal::Probe::Runner.new(data_source: fake_data_source) + runner.call + + claimable = appsignal_class.gauges.select { |g| g[0] == "pgbus_queue_oldest_claimable_age_seconds" } + expect(claimable.map { |g| g[2][:queue] }).to eq(["pgbus_default"]) + expect(claimable.first[1]).to eq(3.0) + end + + it "keeps the raw oldest message age gauge for vt-parked messages" do + runner = Pgbus::Integrations::Appsignal::Probe::Runner.new(data_source: fake_data_source) + runner.call + + raw = appsignal_class.gauges.find { |g| g[0] == "pgbus_queue_oldest_message_age_seconds" && g[2][:queue] == "pgbus_webhooks" } + expect(raw).not_to be_nil + expect(raw[1]).to eq(17_045.0) end it "records active_processes scoped to the current host with hostname tag" do diff --git a/spec/pgbus/web/data_source_batched_metrics_spec.rb b/spec/pgbus/web/data_source_batched_metrics_spec.rb index 357e1757..70b90f35 100644 --- a/spec/pgbus/web/data_source_batched_metrics_spec.rb +++ b/spec/pgbus/web/data_source_batched_metrics_spec.rb @@ -90,6 +90,56 @@ expect(result.first[:visible_unread_length]).to eq(4) end + it "selects a vt-aware oldest_claimable_age_sec scoped to claimable rows" do + allow(conn).to receive(:select_values) + .with(a_string_matching(/pgmq\.meta/)) + .and_return(%w[pgbus_default]) + + allow(conn).to receive(:quote) { |v| "'#{v}'" } + + captured_sql = nil + allow(conn).to receive(:select_all) do |sql, _label| + captured_sql = sql + double(to_a: [{ "queue_name" => "pgbus_default", "queue_length" => "5", + "queue_visible_length" => "3", "newest_msg_age_sec" => "10", + "oldest_msg_age_sec" => "100", "total_messages" => "500", + "max_read_ct" => "2", "visible_unread_length" => "0", + "oldest_claimable_age_sec" => "42" }]) + end + + result = data_source.queues_with_metrics + + # Age of the oldest message eligible for pickup: min(vt) over claimable + # rows, so a scheduled/backoff-parked message (future vt) contributes + # nothing until it comes due (issue #389). + expect(captured_sql).to include("min(vt)") + expect(captured_sql).to include("oldest_claimable_age_sec") + expect(result.first[:oldest_claimable_age_sec]).to eq(42) + end + + it "reports nil claimable age for a queue holding only vt-parked messages" do + allow(conn).to receive(:select_values) + .with(a_string_matching(/pgmq\.meta/)) + .and_return(%w[pgbus_default]) + + allow(conn).to receive(:quote) { |v| "'#{v}'" } + + # The issue #389 incident: one retry-parked message (future vt) — the raw + # age grows at wall-clock rate while nothing is eligible for pickup. + allow(conn).to receive(:select_all) + .with(anything, "Pgbus Batched Queue Metrics") + .and_return(double(to_a: [{ "queue_name" => "pgbus_default", "queue_length" => "1", + "queue_visible_length" => "0", "newest_msg_age_sec" => "17045", + "oldest_msg_age_sec" => "17045", "total_messages" => "500", + "max_read_ct" => "0", "visible_unread_length" => "0", + "oldest_claimable_age_sec" => nil }])) + + result = data_source.queues_with_metrics + + expect(result.first[:oldest_msg_age_sec]).to eq(17_045) + expect(result.first[:oldest_claimable_age_sec]).to be_nil + end + it "maps a NULL max_read_ct (empty queue) to nil, not 0" do allow(conn).to receive(:select_values) .with(a_string_matching(/pgmq\.meta/)) diff --git a/spec/pgbus/web/data_source_spec.rb b/spec/pgbus/web/data_source_spec.rb index 2101606f..af7bef95 100644 --- a/spec/pgbus/web/data_source_spec.rb +++ b/spec/pgbus/web/data_source_spec.rb @@ -158,6 +158,28 @@ expect(data_source.queue_detail("missing")).to be_nil end + + it "exposes a vt-aware oldest_claimable_age_sec, nil when only parked messages remain" do + captured_sql = nil + allow(mock_connection).to receive(:select_one) do |sql, _label| + captured_sql = sql + { + "queue_length" => 1, + "queue_visible_length" => 0, + "oldest_msg_age_sec" => 17_045, + "newest_msg_age_sec" => 17_045, + "oldest_claimable_age_sec" => nil, + "total_messages" => 500 + } + end + + result = data_source.queue_detail("pgbus_critical") + + expect(captured_sql).to include("oldest_claimable_age_sec") + expect(captured_sql).to include("min(vt)") + expect(result[:oldest_msg_age_sec]).to eq(17_045) + expect(result[:oldest_claimable_age_sec]).to be_nil + end end describe "#summary_stats" do diff --git a/spec/pgbus/web/metrics_serializer_spec.rb b/spec/pgbus/web/metrics_serializer_spec.rb index 8fc9df29..91542407 100644 --- a/spec/pgbus/web/metrics_serializer_spec.rb +++ b/spec/pgbus/web/metrics_serializer_spec.rb @@ -10,11 +10,14 @@ let(:queue_metrics) do [ { name: "pgbus_default", queue_length: 42, queue_visible_length: 40, - total_messages: 1000, oldest_msg_age_sec: 120, newest_msg_age_sec: 1, paused: false }, + total_messages: 1000, oldest_msg_age_sec: 120, oldest_claimable_age_sec: 90, + newest_msg_age_sec: 1, paused: false }, { name: "pgbus_default_dlq", queue_length: 3, queue_visible_length: 3, - total_messages: 50, oldest_msg_age_sec: 3600, newest_msg_age_sec: 60, paused: false }, + total_messages: 50, oldest_msg_age_sec: 3600, oldest_claimable_age_sec: 3600, + newest_msg_age_sec: 60, paused: false }, { name: "pgbus_critical", queue_length: 0, queue_visible_length: 0, - total_messages: 500, oldest_msg_age_sec: nil, newest_msg_age_sec: nil, paused: true } + total_messages: 500, oldest_msg_age_sec: nil, oldest_claimable_age_sec: nil, + newest_msg_age_sec: nil, paused: true } ] end @@ -110,6 +113,14 @@ expect(output).not_to include('pgbus_queue_oldest_message_age_seconds{queue="pgbus_critical"}') end + it "includes vt-aware oldest claimable age" do + expect(output).to include('pgbus_queue_oldest_claimable_age_seconds{queue="pgbus_default"} 90') + end + + it "omits oldest claimable age when no claimable backlog exists" do + expect(output).not_to include('pgbus_queue_oldest_claimable_age_seconds{queue="pgbus_critical"}') + end + it "includes queue paused gauge (1 for paused, 0 for active)" do expect(output).to include('pgbus_queue_paused{queue="pgbus_default"} 0') expect(output).to include('pgbus_queue_paused{queue="pgbus_critical"} 1') diff --git a/spec/support/pgbus/stub_data_source.rb b/spec/support/pgbus/stub_data_source.rb index f33cc9bd..0cc87d0b 100644 --- a/spec/support/pgbus/stub_data_source.rb +++ b/spec/support/pgbus/stub_data_source.rb @@ -203,9 +203,11 @@ def default_health_stats def default_queues [ { name: "pgbus_default", queue_length: 10, queue_visible_length: 8, - oldest_msg_age_sec: 120, newest_msg_age_sec: 5, total_messages: 500 }, + oldest_msg_age_sec: 120, oldest_claimable_age_sec: 90, + newest_msg_age_sec: 5, total_messages: 500 }, { name: "pgbus_default_dlq", queue_length: 2, queue_visible_length: 2, - oldest_msg_age_sec: 3600, newest_msg_age_sec: 1800, total_messages: 5 } + oldest_msg_age_sec: 3600, oldest_claimable_age_sec: 3600, + newest_msg_age_sec: 1800, total_messages: 5 } ] end From 2a7d0591d076bf47ab068a1fba1f03092bdb388f Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Tue, 4 Aug 2026 11:46:21 +0200 Subject: [PATCH 2/3] fix(review): synchronize claimable-age raw query, centralize parked_length, doc clarifications MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit review fixes on PR #390: - Client#oldest_claimable_ages wraps the raw query in `synchronized` — on the shared-Proc path with_raw_connection yields the same AR connection every @pgmq operation uses, so an unserialized query could interleave mid-protocol - parked depth derived once in DataSource (parked_length) instead of three views - MCP queues_tool description: nil claimable age = nothing currently eligible (scheduled, backoff-parked, OR in-flight), not proof of health - README: document that the claimable gauge omits its series when nil - integration spec: derive queue key via config.queue_name; tolerate wall-clock growth between calls on a contended server - batched-metrics spec: fixture comment matches the read_ct=0 wait:-re-enqueue incident shape --- README.md | 2 +- .../pgbus/dashboard/_queues_table.html.erb | 2 +- app/views/pgbus/queues/_queues_list.html.erb | 2 +- app/views/pgbus/queues/show.html.erb | 2 +- lib/pgbus/client.rb | 20 ++++++++++++------- lib/pgbus/mcp/tools/queues_tool.rb | 8 +++++--- lib/pgbus/web/data_source.rb | 2 ++ spec/dummy/lib/stub_data_source.rb | 8 ++++---- .../integration/claimable_age_metrics_spec.rb | 4 +++- .../web/data_source_batched_metrics_spec.rb | 6 ++++-- spec/support/pgbus/stub_data_source.rb | 4 ++-- 11 files changed, 37 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index c0b50e0d..709ccf75 100644 --- a/README.md +++ b/README.md @@ -991,7 +991,7 @@ When `config.metrics_enabled = true` (default), the dashboard exposes Prometheus | Metric | Description | |--------|-------------| -| `pgbus_queue_oldest_claimable_age_seconds` | Age of the oldest message eligible for pickup (visibility timeout elapsed) — safe to alert on: scheduled/backoff-parked messages don't count until due | +| `pgbus_queue_oldest_claimable_age_seconds` | Age of the oldest message eligible for pickup (visibility timeout elapsed) — safe to alert on: scheduled/backoff-parked messages don't count until due; the series is omitted entirely when no claimable backlog exists (while the raw `pgbus_queue_oldest_message_age_seconds` gauge may still report a parked message's age) | | `pgbus_table_dead_tuples` | Dead tuple count per PGMQ table | | `pgbus_table_live_tuples` | Live tuple count per PGMQ table | | `pgbus_table_bloat_ratio` | Dead / (dead + live) per table | diff --git a/app/views/pgbus/dashboard/_queues_table.html.erb b/app/views/pgbus/dashboard/_queues_table.html.erb index 702009e0..624d0116 100644 --- a/app/views/pgbus/dashboard/_queues_table.html.erb +++ b/app/views/pgbus/dashboard/_queues_table.html.erb @@ -26,7 +26,7 @@ <%= pgbus_number(q[:queue_length]) %> <%= pgbus_number(q[:queue_visible_length]) %> - <%= pgbus_number(q[:queue_length] - q[:queue_visible_length]) %> + <%= pgbus_number(q[:parked_length]) %> <%= q[:oldest_claimable_age_sec] || "—" %> <%= pgbus_number(q[:total_messages]) %> diff --git a/app/views/pgbus/queues/_queues_list.html.erb b/app/views/pgbus/queues/_queues_list.html.erb index c5972f96..1e0835c0 100644 --- a/app/views/pgbus/queues/_queues_list.html.erb +++ b/app/views/pgbus/queues/_queues_list.html.erb @@ -25,7 +25,7 @@ <%= pgbus_number(q[:queue_length]) %> <%= pgbus_number(q[:queue_visible_length]) %> - <%= pgbus_number(q[:queue_length] - q[:queue_visible_length]) %> + <%= pgbus_number(q[:parked_length]) %> <%= q[:oldest_claimable_age_sec] || "—" %> <%= q[:newest_msg_age_sec] || "—" %> <%= pgbus_number(q[:total_messages]) %> diff --git a/app/views/pgbus/queues/show.html.erb b/app/views/pgbus/queues/show.html.erb index ebbb1975..216ee902 100644 --- a/app/views/pgbus/queues/show.html.erb +++ b/app/views/pgbus/queues/show.html.erb @@ -13,7 +13,7 @@

<%= t("pgbus.queues.show.depth") %> <%= @queue[:queue_length] %> | <%= t("pgbus.queues.show.visible") %> <%= @queue[:queue_visible_length] %> | - <%= t("pgbus.queues.show.parked") %> <%= @queue[:queue_length] - @queue[:queue_visible_length] %> | + <%= t("pgbus.queues.show.parked") %> <%= @queue[:parked_length] %> | <%= t("pgbus.queues.show.oldest_claimable") %> <%= @queue[:oldest_claimable_age_sec] || "—" %> | <%= t("pgbus.queues.show.total") %> <%= pgbus_number(@queue[:total_messages]) %>

diff --git a/lib/pgbus/client.rb b/lib/pgbus/client.rb index aa584e17..991ecb32 100644 --- a/lib/pgbus/client.rb +++ b/lib/pgbus/client.rb @@ -553,14 +553,20 @@ def metrics(queue_name = nil) # # With a queue name: the age for that (prefixed) queue, or nil. # Without: a hash of every physical queue in pgmq.meta to its age. + # + # synchronized: on the shared-Proc path with_raw_connection yields the same + # AR raw connection every @pgmq operation runs on — an unserialized query + # here could interleave with a concurrent PGMQ call mid-protocol. def oldest_claimable_ages(queue_name = nil) - with_raw_connection do |conn| - if queue_name - claimable_age_for(conn, config.queue_name(queue_name)) - else - names = conn.exec("SELECT queue_name FROM pgmq.meta ORDER BY queue_name") - .map { |row| row["queue_name"] } - names.to_h { |name| [name, claimable_age_for(conn, name)] } + synchronized do + with_raw_connection do |conn| + if queue_name + claimable_age_for(conn, config.queue_name(queue_name)) + else + names = conn.exec("SELECT queue_name FROM pgmq.meta ORDER BY queue_name") + .map { |row| row["queue_name"] } + names.to_h { |name| [name, claimable_age_for(conn, name)] } + end end end end diff --git a/lib/pgbus/mcp/tools/queues_tool.rb b/lib/pgbus/mcp/tools/queues_tool.rb index db8f652b..8ba31fae 100644 --- a/lib/pgbus/mcp/tools/queues_tool.rb +++ b/lib/pgbus/mcp/tools/queues_tool.rb @@ -15,9 +15,11 @@ class QueuesTool < BaseTool claimed), oldest/newest message age in seconds, lifetime total, and paused state. Use this to answer "are any queues backed up?". oldest_claimable_age_sec is the age of the oldest message actually - eligible for pickup — nil means no claimable backlog, so a queue whose - oldest_msg_age_sec keeps growing while oldest_claimable_age_sec is nil - holds only scheduled/backoff-parked messages and is healthy. + eligible for pickup — nil means no message is currently claimable: + every remaining message is scheduled, backoff-parked, or in flight + with a future visibility timeout. A queue whose oldest_msg_age_sec + keeps growing while oldest_claimable_age_sec stays nil has no + starving backlog — nothing is waiting for a worker. DESC input_schema(properties: {}, required: []) diff --git a/lib/pgbus/web/data_source.rb b/lib/pgbus/web/data_source.rb index b9f596c8..d97ca4ed 100644 --- a/lib/pgbus/web/data_source.rb +++ b/lib/pgbus/web/data_source.rb @@ -1194,6 +1194,7 @@ def batched_queue_metrics(queue_names) name: row["queue_name"], queue_length: row["queue_length"].to_i, queue_visible_length: row["queue_visible_length"].to_i, + parked_length: row["queue_length"].to_i - row["queue_visible_length"].to_i, oldest_msg_age_sec: row["oldest_msg_age_sec"]&.to_i, oldest_claimable_age_sec: row["oldest_claimable_age_sec"]&.to_i, newest_msg_age_sec: row["newest_msg_age_sec"]&.to_i, @@ -1245,6 +1246,7 @@ def queue_metrics_via_sql(queue_name) name: queue_name, queue_length: row["queue_length"].to_i, queue_visible_length: row["queue_visible_length"].to_i, + parked_length: row["queue_length"].to_i - row["queue_visible_length"].to_i, oldest_msg_age_sec: row["oldest_msg_age_sec"]&.to_i, oldest_claimable_age_sec: row["oldest_claimable_age_sec"]&.to_i, newest_msg_age_sec: row["newest_msg_age_sec"]&.to_i, diff --git a/spec/dummy/lib/stub_data_source.rb b/spec/dummy/lib/stub_data_source.rb index ef45d461..87cac869 100644 --- a/spec/dummy/lib/stub_data_source.rb +++ b/spec/dummy/lib/stub_data_source.rb @@ -18,16 +18,16 @@ def summary_stats def queues_with_metrics [ - { name: "pgbus_default", queue_length: 85, queue_visible_length: 62, + { name: "pgbus_default", queue_length: 85, queue_visible_length: 62, parked_length: 23, oldest_msg_age_sec: 300, oldest_claimable_age_sec: 240, newest_msg_age_sec: 2, total_messages: 12_450 }, - { name: "pgbus_mailers", queue_length: 22, queue_visible_length: 18, + { name: "pgbus_mailers", queue_length: 22, queue_visible_length: 18, parked_length: 4, oldest_msg_age_sec: 45, oldest_claimable_age_sec: 30, newest_msg_age_sec: 1, total_messages: 8_320 }, - { name: "pgbus_events", queue_length: 15, queue_visible_length: 13, + { name: "pgbus_events", queue_length: 15, queue_visible_length: 13, parked_length: 2, oldest_msg_age_sec: 120, oldest_claimable_age_sec: 100, newest_msg_age_sec: 5, total_messages: 45_000 }, - { name: "pgbus_default_dlq", queue_length: 3, queue_visible_length: 3, + { name: "pgbus_default_dlq", queue_length: 3, queue_visible_length: 3, parked_length: 0, oldest_msg_age_sec: 7200, oldest_claimable_age_sec: 7200, newest_msg_age_sec: 3600, total_messages: 47 } ] diff --git a/spec/integration/claimable_age_metrics_spec.rb b/spec/integration/claimable_age_metrics_spec.rb index 434c46b5..b4aec46d 100644 --- a/spec/integration/claimable_age_metrics_spec.rb +++ b/spec/integration/claimable_age_metrics_spec.rb @@ -28,8 +28,10 @@ expect(age).to be_an(Integer) expect(age).to be >= 0 + # The age is wall-clock-relative and only grows between the two calls; + # an exact match would flake when connecting to a contended server. all = client.oldest_claimable_ages - expect(all.fetch("pgbus_int_claimable_test")).to be_within(1).of(age) + expect(all.fetch(client.config.queue_name("claimable_test"))).to be >= age end it "excludes an in-flight message whose visibility timeout was pushed forward" do diff --git a/spec/pgbus/web/data_source_batched_metrics_spec.rb b/spec/pgbus/web/data_source_batched_metrics_spec.rb index 70b90f35..40e8e598 100644 --- a/spec/pgbus/web/data_source_batched_metrics_spec.rb +++ b/spec/pgbus/web/data_source_batched_metrics_spec.rb @@ -115,6 +115,7 @@ expect(captured_sql).to include("min(vt)") expect(captured_sql).to include("oldest_claimable_age_sec") expect(result.first[:oldest_claimable_age_sec]).to eq(42) + expect(result.first[:parked_length]).to eq(2) end it "reports nil claimable age for a queue holding only vt-parked messages" do @@ -124,8 +125,9 @@ allow(conn).to receive(:quote) { |v| "'#{v}'" } - # The issue #389 incident: one retry-parked message (future vt) — the raw - # age grows at wall-clock rate while nothing is eligible for pickup. + # The issue #389 incident: one vt-parked message — an ActiveJob retry + # re-enqueued with wait:, so read_ct is 0 and vt is hours in the future. + # The raw age grows at wall-clock rate while nothing is eligible for pickup. allow(conn).to receive(:select_all) .with(anything, "Pgbus Batched Queue Metrics") .and_return(double(to_a: [{ "queue_name" => "pgbus_default", "queue_length" => "1", diff --git a/spec/support/pgbus/stub_data_source.rb b/spec/support/pgbus/stub_data_source.rb index 0cc87d0b..4fc1cce3 100644 --- a/spec/support/pgbus/stub_data_source.rb +++ b/spec/support/pgbus/stub_data_source.rb @@ -202,10 +202,10 @@ def default_health_stats def default_queues [ - { name: "pgbus_default", queue_length: 10, queue_visible_length: 8, + { name: "pgbus_default", queue_length: 10, queue_visible_length: 8, parked_length: 2, oldest_msg_age_sec: 120, oldest_claimable_age_sec: 90, newest_msg_age_sec: 5, total_messages: 500 }, - { name: "pgbus_default_dlq", queue_length: 2, queue_visible_length: 2, + { name: "pgbus_default_dlq", queue_length: 2, queue_visible_length: 2, parked_length: 0, oldest_msg_age_sec: 3600, oldest_claimable_age_sec: 3600, newest_msg_age_sec: 1800, total_messages: 5 } ] From e067fcf646e16c54efa5181db37cddebeda7316f Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Tue, 4 Aug 2026 12:10:13 +0200 Subject: [PATCH 3/3] fix(review): route claimable-age query through pooled @pgmq connection, show raw age on queue detail MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Client#oldest_claimable_ages now uses @pgmq.with_connection (health-checked, bounded by the statement/socket timeouts from Client#initialize) instead of a fresh unbounded PG.connect per call — same migration notify_trigger_current? made; locally this cut the integration spec from ~150s to ~9s on a contended server - queues/show summary line now shows the raw oldest age alongside the claimable age (new pgbus.queues.show.oldest key in all 12 locales) --- app/views/pgbus/queues/show.html.erb | 1 + config/locales/da.yml | 1 + config/locales/de.yml | 1 + config/locales/en.yml | 1 + config/locales/es.yml | 1 + config/locales/fi.yml | 1 + config/locales/fr.yml | 1 + config/locales/it.yml | 1 + config/locales/ja.yml | 1 + config/locales/nb.yml | 1 + config/locales/nl.yml | 1 + config/locales/pt.yml | 1 + config/locales/sv.yml | 1 + lib/pgbus/client.rb | 12 ++++++++---- spec/pgbus/client_spec.rb | 2 +- 15 files changed, 22 insertions(+), 5 deletions(-) diff --git a/app/views/pgbus/queues/show.html.erb b/app/views/pgbus/queues/show.html.erb index 216ee902..8065cc46 100644 --- a/app/views/pgbus/queues/show.html.erb +++ b/app/views/pgbus/queues/show.html.erb @@ -14,6 +14,7 @@ <%= t("pgbus.queues.show.depth") %> <%= @queue[:queue_length] %> | <%= t("pgbus.queues.show.visible") %> <%= @queue[:queue_visible_length] %> | <%= t("pgbus.queues.show.parked") %> <%= @queue[:parked_length] %> | + <%= t("pgbus.queues.show.oldest") %> <%= @queue[:oldest_msg_age_sec] || "—" %> | <%= t("pgbus.queues.show.oldest_claimable") %> <%= @queue[:oldest_claimable_age_sec] || "—" %> | <%= t("pgbus.queues.show.total") %> <%= pgbus_number(@queue[:total_messages]) %>

diff --git a/config/locales/da.yml b/config/locales/da.yml index 58eb3eb2..0c9b344e 100644 --- a/config/locales/da.yml +++ b/config/locales/da.yml @@ -518,6 +518,7 @@ da: scheduled: 'Planlagt:' timezone: 'Tidszone:' visible_at: 'Synlig fra:' + oldest: 'Ældste:' oldest_claimable: 'Ældste tilgængelig:' parked: 'Parkeret:' pause: Pause diff --git a/config/locales/de.yml b/config/locales/de.yml index 5cfbcc9b..e02aad16 100644 --- a/config/locales/de.yml +++ b/config/locales/de.yml @@ -518,6 +518,7 @@ de: scheduled: 'Geplant:' timezone: 'Zeitzone:' visible_at: 'Sichtbar ab:' + oldest: 'Älteste:' oldest_claimable: 'Älteste verfügbar:' parked: 'Geparkt:' pause: Pause diff --git a/config/locales/en.yml b/config/locales/en.yml index 79dbc532..df456900 100644 --- a/config/locales/en.yml +++ b/config/locales/en.yml @@ -518,6 +518,7 @@ en: scheduled: 'Scheduled at:' timezone: 'Timezone:' visible_at: 'Visible at:' + oldest: 'Oldest:' oldest_claimable: 'Oldest claimable:' parked: 'Parked:' pause: Pause diff --git a/config/locales/es.yml b/config/locales/es.yml index 4c3190c5..dc26fb54 100644 --- a/config/locales/es.yml +++ b/config/locales/es.yml @@ -518,6 +518,7 @@ es: scheduled: 'Programado:' timezone: 'Zona horaria:' visible_at: 'Visible en:' + oldest: 'Más antiguo:' oldest_claimable: 'Más antiguo disponible:' parked: 'Aparcados:' pause: Pausar diff --git a/config/locales/fi.yml b/config/locales/fi.yml index c3d4d4a2..57b0ea4b 100644 --- a/config/locales/fi.yml +++ b/config/locales/fi.yml @@ -518,6 +518,7 @@ fi: scheduled: 'Aikataulutettu:' timezone: 'Aikavyöhyke:' visible_at: 'Näkyvissä:' + oldest: 'Vanhin:' oldest_claimable: 'Vanhin saatavilla:' parked: 'Pysäköidyt:' pause: Tauko diff --git a/config/locales/fr.yml b/config/locales/fr.yml index 48864770..a9c61663 100644 --- a/config/locales/fr.yml +++ b/config/locales/fr.yml @@ -518,6 +518,7 @@ fr: scheduled: 'Planifié :' timezone: 'Fuseau horaire :' visible_at: 'Visible à :' + oldest: 'Plus ancien :' oldest_claimable: 'Plus ancien disponible :' parked: 'Différés :' pause: Pause diff --git a/config/locales/it.yml b/config/locales/it.yml index ec351ecf..4d8adc53 100644 --- a/config/locales/it.yml +++ b/config/locales/it.yml @@ -518,6 +518,7 @@ it: scheduled: 'Programmato:' timezone: 'Fuso orario:' visible_at: 'Visibile alle:' + oldest: 'Più vecchio:' oldest_claimable: 'Più vecchio disponibile:' parked: 'Posticipati:' pause: Pausa diff --git a/config/locales/ja.yml b/config/locales/ja.yml index aa962368..7082ab75 100644 --- a/config/locales/ja.yml +++ b/config/locales/ja.yml @@ -518,6 +518,7 @@ ja: scheduled: スケジュール済み： timezone: タイムゾーン： visible_at: 表示可能日時： + oldest: '最古:' oldest_claimable: '取得可能な最古:' parked: '待機中:' pause: 一時停止 diff --git a/config/locales/nb.yml b/config/locales/nb.yml index 9cf90204..51957ca7 100644 --- a/config/locales/nb.yml +++ b/config/locales/nb.yml @@ -518,6 +518,7 @@ nb: scheduled: 'Planlagt:' timezone: 'Tidssone:' visible_at: 'Synlig fra:' + oldest: 'Eldste:' oldest_claimable: 'Eldste tilgjengelig:' parked: 'Parkert:' pause: Pause diff --git a/config/locales/nl.yml b/config/locales/nl.yml index 5e49eaa9..8b3aab68 100644 --- a/config/locales/nl.yml +++ b/config/locales/nl.yml @@ -518,6 +518,7 @@ nl: scheduled: 'Gepland:' timezone: 'Tijdzone:' visible_at: 'Zichtbaar op:' + oldest: 'Oudste:' oldest_claimable: 'Oudste beschikbaar:' parked: 'Geparkeerd:' pause: Pauzeren diff --git a/config/locales/pt.yml b/config/locales/pt.yml index 1a88854a..70aa6ac2 100644 --- a/config/locales/pt.yml +++ b/config/locales/pt.yml @@ -518,6 +518,7 @@ pt: scheduled: 'Agendado:' timezone: 'Fuso horário:' visible_at: 'Visível em:' + oldest: 'Mais antigo:' oldest_claimable: 'Mais antigo disponível:' parked: 'Adiados:' pause: Pausar diff --git a/config/locales/sv.yml b/config/locales/sv.yml index 9b7b374b..ee67eda8 100644 --- a/config/locales/sv.yml +++ b/config/locales/sv.yml @@ -518,6 +518,7 @@ sv: scheduled: 'Schemalagt:' timezone: 'Tidszon:' visible_at: 'Synlig vid:' + oldest: 'Äldsta:' oldest_claimable: 'Äldsta tillgänglig:' parked: 'Parkerade:' pause: Pausa diff --git a/lib/pgbus/client.rb b/lib/pgbus/client.rb index 991ecb32..f43a59e9 100644 --- a/lib/pgbus/client.rb +++ b/lib/pgbus/client.rb @@ -554,12 +554,16 @@ def metrics(queue_name = nil) # With a queue name: the age for that (prefixed) queue, or nil. # Without: a hash of every physical queue in pgmq.meta to its age. # - # synchronized: on the shared-Proc path with_raw_connection yields the same - # AR raw connection every @pgmq operation runs on — an unserialized query - # here could interleave with a concurrent PGMQ call mid-protocol. + # Routes through the pooled @pgmq.with_connection (health-checked, bounded + # by the statement/socket timeouts applied at Client#initialize) rather + # than a fresh unbounded PG.connect per call — same rationale as + # notify_trigger_current?. synchronized: on the shared-Proc path @pgmq + # rides the AR raw connection, so the query must serialize against + # concurrent PGMQ operations. One checkout spans all per-queue queries; + # nothing nests inside it, so the shared pool_size=1 path is safe. def oldest_claimable_ages(queue_name = nil) synchronized do - with_raw_connection do |conn| + @pgmq.with_connection do |conn| if queue_name claimable_age_for(conn, config.queue_name(queue_name)) else diff --git a/spec/pgbus/client_spec.rb b/spec/pgbus/client_spec.rb index 1e397799..49486bb7 100644 --- a/spec/pgbus/client_spec.rb +++ b/spec/pgbus/client_spec.rb @@ -471,7 +471,7 @@ def initialize(*args, **kwargs); end describe "#oldest_claimable_ages" do let(:raw_conn) { double("PG::Connection") } - before { allow(client).to receive(:with_raw_connection).and_yield(raw_conn) } + before { allow(mock_pgmq).to receive(:with_connection).and_yield(raw_conn) } context "with a queue_name" do it "returns the vt-aware age of the oldest claimable message in the prefixed queue" do