From e10b3a96741afe786d1c770c0a2ea0b40fc4446d Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Mon, 3 Aug 2026 07:41:55 +0200 Subject: [PATCH 1/3] =?UTF-8?q?feat(deploy):=20health-checked=20rolling=20?= =?UTF-8?q?restarts=20=E2=80=94=20container-local=20/readyz,=20aligned=20s?= =?UTF-8?q?hutdown=20budget,=20pgbus-health=20probe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The supervisor's standalone /readyz previously answered with the cluster-wide HealthAnalyzer verdict, so during a rolling deploy a freshly-booted container passed the orchestrator's health gate on the strength of the OLD container's still-heartbeating workers. It now answers container-local: 200 only once THIS supervisor verified its connection, bootstrapped queues, and every child it forked is alive — 503 BOOTING / DEGRADED (child in crash backoff) / DRAINING otherwise. The supervisor publishes an immutable ReadinessSnapshot per monitor pass (refreshed after reap-and-restart, so clean recycles never flap); the Rails-mounted HealthApp keeps the cluster verdict. Shutdown budgets become alignable: new config.shutdown_timeout (default drain_timeout + 5) replaces the supervisor's hardcoded 30s SIGKILL deadline, Consumer's drain wait follows drain_timeout instead of a hardcoded 30s, and Worker's post-drain residual wait drops to 5s. New pgbus-health executable: stdlib-only probe (never loads Bundler or the gem) for docker HEALTHCHECK blocks — exit 0/1/2. ## Test Coverage - configuration_spec: shutdown_timeout derivation, validation, warning - readiness_snapshot_spec: ready?/status truth table - supervisor_spec: boot/degrade/drain transitions, /readyz wiring, configurable SIGKILL deadline - worker_spec / consumer_spec: pool-wait bounds - health_app_spec: local-readiness verdicts + error path - health_probe_spec: real-socket probe against HealthServer, exit codes, no-gem-load guard ## Verification - [x] bundle exec rake rubocop — 546 files, no offenses - [x] touched specs green (747 examples) - [x] full suite vs main, same seed: identical 2 pre-existing i18n failures, zero regressions Refs #386 --- CHANGELOG.md | 6 + README.md | 66 +++++++++- exe/pgbus-health | 9 ++ lib/pgbus/configuration.rb | 38 ++++++ lib/pgbus/health_probe.rb | 123 ++++++++++++++++++ lib/pgbus/process/consumer.rb | 5 +- lib/pgbus/process/readiness_snapshot.rb | 30 +++++ lib/pgbus/process/supervisor.rb | 50 ++++++- lib/pgbus/process/worker.rb | 18 ++- lib/pgbus/web/health_app.rb | 24 +++- spec/pgbus/configuration_spec.rb | 50 +++++++ spec/pgbus/health_probe_spec.rb | 102 +++++++++++++++ spec/pgbus/process/consumer_spec.rb | 10 ++ spec/pgbus/process/readiness_snapshot_spec.rb | 63 +++++++++ spec/pgbus/process/supervisor_spec.rb | 105 ++++++++++++++- spec/pgbus/process/worker_spec.rb | 10 ++ spec/pgbus/web/health_app_spec.rb | 87 +++++++++++++ 17 files changed, 783 insertions(+), 13 deletions(-) create mode 100755 exe/pgbus-health create mode 100644 lib/pgbus/health_probe.rb create mode 100644 lib/pgbus/process/readiness_snapshot.rb create mode 100644 spec/pgbus/health_probe_spec.rb create mode 100644 spec/pgbus/process/readiness_snapshot_spec.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index 02c932c9..a5f203c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ### 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. + +### Changed + +- **Shutdown budgets are now alignable end-to-end (issue #386).** New `config.shutdown_timeout` bounds how long the supervisor waits for children after forwarding TERM before escalating to SIGKILL — previously a hardcoded 30s, which silently SIGKILLed workers mid-drain the moment `drain_timeout` was raised past it. Default derives `drain_timeout + 5` so the deadline tracks the drain window automatically; an explicit value below `drain_timeout` logs a boot warning. `Consumer#shutdown`'s pool wait (its only drain bound) now follows `config.drain_timeout` instead of a hardcoded 30s, and `Worker#shutdown`'s post-drain residual wait drops from a second full 30s window to 5s — the drain loop already waited `drain_timeout`, and a job still running has proven it won't finish. Rule of thumb: orchestrator stop grace period > `shutdown_timeout` > `drain_timeout`. Refs #386. + - **Streams: one LISTEN connection per web host — `streams_listen_scope` (issue #382).** ⚠️ **Default behavior change.** Previously every Puma worker lazily opened its own dedicated streams LISTEN connection on first SSE use, so a web host pinned one direct connection per worker. Under the new default (`streams_listen_scope = :master`) the `pgbus_streams` Puma plugin runs a **MasterHub** in the preforking master: ONE `Web::Streamer::Listener` on the refcounted union of every worker's stream channels, fanning wakes — **including ephemeral payloads** — out to workers over a Unix domain socket with length-prefixed frames (`Streamer::HubProtocol`). Workers connect lazily (nothing is inherited across fork) and the synchronous `ensure_listening` ack contract is preserved cross-process: a sub is registered before LISTEN executes and acked only after, so the no-lost-broadcast guarantee holds. Backpressure follows the streams rules: durable wakes are droppable at a per-worker cap (they self-heal via `read_after`), **ephemeral wakes are never dropped** — a worker that stops draining is evicted, which triggers its own fallback. **Fallback is per-worker listeners, not loss**: whenever the hub is absent or dies (no `preload_app!`, single-mode Puma, crash, eviction) each worker's `FailoverListener` swaps in a real per-worker `Listener` and re-LISTENs its recorded subscriptions — connection footprint balloons back to pre-#382 levels (census-visible) but no broadcast semantics change; the worker stays local until it recycles. Measured (local PG, n=50): the master→worker hop is noise-level free — single-broadcast SSE roundtrip p50 16.00ms via the hub vs 16.93ms per-worker. **`:master` effectively requires `preload_app!`** (the hub waits for the app's pgbus initializer; without it the deadline expires quietly and workers stay per-worker). **Rollback:** `config.streams_listen_scope = :process`. Refs #382, builds on the #381 patterns. - **Host-level shared LISTEN: `worker_notify_scope` — the supervisor now owns ONE direct LISTEN connection for the whole host (issue #381).** ⚠️ **Default behavior change.** Previously every worker fork and every consumer fork opened its own dedicated LISTEN connection (`NotifyListener`), so a host's direct-connection footprint scaled with fork count — on transaction-pool PgBouncer platforms those connections come out of the scarcest slice of `max_connections`, and a 5-capsule + 2-consumer host pinned 7. Under the new default (`config.worker_notify_scope = :supervisor`) the supervisor runs a single `NotifyHub`: one `NotifyListener` on the union of every capsule's and consumer's queue channels (wildcards via the shared resolver, consumer sets via the registry), fanning wakes out to forks over per-fork pipes (`W` wake / `H` healthy / `P` degraded bytes; a fork whose pipe reports degraded or reaches EOF falls back to fast polling exactly like a failed local listener). Footprint drops to **1 direct LISTEN connection per job host**, verified by integration test: routing is per-fork (an insert wakes only the forks reading that queue, wildcard capsules unconditionally), and `pg_terminate_backend` on the shared connection is survived — reconnect, re-LISTEN, wakes flow again. **Rollback:** `config.worker_notify_scope = :fork` restores the previous per-fork listeners byte-for-byte. Dedicated LISTEN connections are now census-tagged `application_name=pgbus-listen` so `pg_stat_activity` can count them. Refs #381. diff --git a/README.md b/README.md index d7581656..a4b28363 100644 --- a/README.md +++ b/README.md @@ -1117,14 +1117,18 @@ For the **HTTP** transport, point the client at the mounted URL with a streamabl ### Health endpoints (liveness / readiness) -For orchestrators like Kubernetes, Pgbus exposes two HTTP probes: `/livez` (is the serving process up?) and `/readyz` (are queues draining, or is a worker silently wedged?). `/readyz` runs the same `OK` / `DEGRADED` / `STALLED` verdict as the MCP `pgbus_health` tool — `STALLED` (visible backlog while workers heart-beat but don't claim) fails readiness. +For orchestrators like Kubernetes, Pgbus exposes two HTTP probes: `/livez` (is the serving process up?) and `/readyz`. Readiness means different things in the two places the probes are served: + +- **Mounted in Rails** (`Pgbus::Web::HealthApp`): `/readyz` runs the cluster-wide `OK` / `DEGRADED` / `STALLED` verdict, same as the MCP `pgbus_health` tool — `STALLED` (visible backlog while workers heart-beat but don't claim) fails readiness. +- **Standalone from the supervisor** (`health_port`): `/readyz` is **container-local** — did *this* supervisor finish booting, and are all the children *it* forked alive? That is the signal a rolling deploy's health gate needs; the cluster verdict would let a brand-new container pass on the strength of the *old* container's workers. | Path | Method | 200 | 503 | Touches DB | |---|---|---|---|---| | `/livez` | GET | always (`ok`) | never | no | -| `/readyz` | GET | verdict `OK` or `DEGRADED` | verdict `STALLED`, or DB unreachable (`{"status":"ERROR"}`) | yes | +| `/readyz` (mounted) | GET | verdict `OK` or `DEGRADED` | verdict `STALLED`, or DB unreachable (`{"status":"ERROR"}`) | yes | +| `/readyz` (supervisor) | GET | `OK` — booted, all children live | `BOOTING`, `DEGRADED` (child down), `DRAINING` (stopping) | no | -Unknown paths return `404`; non-`GET` methods return `405`. The `/readyz` body is the verdict JSON, so a probe failure is self-describing in the pod's event log. +Unknown paths return `404`; non-`GET` methods return `405`. The `/readyz` body is JSON, so a probe failure is self-describing in the pod's event log. #### Mount in your Rails app @@ -1165,6 +1169,61 @@ readinessProbe: httpGet: { path: /readyz, port: 9394 } ``` +The supervisor's `/readyz` answers from its own state, never the database: + +```json +{ "status": "OK", "expected": 3, "live": 3 } +``` + +- `BOOTING` (503) until the connection is verified, queues are bootstrapped, and every configured child has been forked. `expected` is stamped at that instant. +- `OK` (200) while all expected children are in the fork table. A clean worker recycle never dips the count — the snapshot refreshes after reap-and-restart each monitor pass. +- `DEGRADED` (503) when a child died and is waiting out crash-restart backoff. During a rolling deploy this is the desired failure mode: a crash-looping replacement never goes ready, so the old container keeps running. +- `DRAINING` (503) the moment a stop signal arrives. + +#### `pgbus-health`: container HEALTHCHECK probe + +`pgbus-health` ships with the gem: a dependency-free probe (plain Ruby + stdlib sockets — no Bundler, no Rails, nothing else loaded) that GETs `127.0.0.1:/readyz` and exits `0` on 200, `1` on anything else, `2` on usage errors. Cheap enough for a 1–5s `HEALTHCHECK` interval, and it works in images without curl: + +```bash +pgbus-health --port 9394 # or PGBUS_HEALTH_PORT=9394 pgbus-health +pgbus-health --port 9394 --path /livez --timeout 2 +``` + +### Rolling restarts (Kamal, docker) + +Kamal distributions with per-role health checks (for example the [`dash` branch](https://github.com/mhenrixon/kamal)) can rolling-restart a non-proxied job role: start the new container, poll its docker `HEALTHCHECK` until healthy, and only then `docker stop` the old one. Wire the pgbus container into that gate: + +```yaml +# config/deploy.yml +servers: + job: + hosts: [...] + cmd: bin/pgbus start + healthcheck: + cmd: bin/pgbus-health --port 9394 + interval: 5s + start_period: 30s # cover Rails boot + queue bootstrap + stop_timeout: 45 # must exceed pgbus shutdown_timeout (see below) +env: + clear: + PGBUS_HEALTH_PORT: 9394 +``` + +(`bundle binstubs pgbus` generates `bin/pgbus-health`; adjust the path if your image invokes gem executables differently.) + +**The shutdown timeline.** On `docker stop`, SIGTERM reaches the supervisor and readiness flips to `DRAINING`; children stop claiming work and drain in-flight jobs for up to `drain_timeout` (default 30s); the supervisor waits `shutdown_timeout` (default `drain_timeout + 5`) before SIGKILLing stragglers. Align the three knobs outside-in: + +``` +orchestrator stop_timeout > pgbus shutdown_timeout > pgbus drain_timeout + 45s 35s (derived) 30s +``` + +If the orchestrator's stop grace period is *shorter* than `shutdown_timeout`, docker SIGKILLs the whole tree mid-drain and the graceful path never gets to finish. Raising `drain_timeout` raises the derived `shutdown_timeout` automatically; raise `stop_timeout` to match. + +**The overlap window is safe by construction.** Between "new container healthy" and "old container stopped", two supervisors run against the same database. Nothing double-fires: queue claims use `FOR UPDATE SKIP LOCKED`, `single_active_consumer` queues arbitrate via session-level advisory locks (released the instant a killed process's connection dies), two live recurring schedulers dedup on the `(task_key, run_at)` unique record, and dispatcher maintenance is idempotent. "One scheduler per deployment" is a steady-state rule; a deploy window may briefly violate it without consequence. + +**What a hard kill still costs.** Jobs killed past the drain window are redelivered after their visibility timeout (at-least-once holds) — but PGMQ's `read_ct` increments exactly like a logical failure, so a long-running job that straddles *repeated* deploy kills can be pushed to the DLQ without its code ever raising. `zombie_detection` logs exactly this pattern (`read_ct > 1` with no recorded failure). Keep jobs shorter than `drain_timeout`, or raise it (and `stop_timeout`) for queues that can't be. For `idempotent!` event handlers there is a separate crash-window caveat tracked in [#385](https://github.com/mhenrixon/pgbus/issues/385). + ### Boot diagnostics banner `Supervisor#run` logs a one-block banner right after the heartbeat starts and before queues bootstrap, so a misconfigured deployment states its actual settings instead of forcing an operator to attach a console. Every line is `"[Pgbus] boot:"`-prefixed and renders cleanly under both the `:text` and `:json` log formatters: @@ -2111,6 +2170,7 @@ Curated headline options for the README. The full operator reference (with types | `zombie_detection` | `true` | Detect and reclaim work from crashed workers | | `read_timeout` | `30` | Seconds before a single PGMQ read is bounded (libpq `statement_timeout` + `tcp_user_timeout` on a dedicated connection; nil disables) | | `drain_timeout` | `30` | Seconds to wait for in-flight jobs during graceful shutdown before abandoning them | +| `shutdown_timeout` | `drain_timeout + 5` | Seconds the supervisor waits for children after TERM before SIGKILL; an orchestrator's stop grace period must exceed it | | `stall_threshold` | `300` | Seconds without progress before a worker is considered stalled | | `priority_levels` | `nil` | Number of priority sub-queues (nil = disabled, 2-10) | | `default_priority` | `1` | Default priority for jobs without explicit priority | diff --git a/exe/pgbus-health b/exe/pgbus-health new file mode 100755 index 00000000..d6722970 --- /dev/null +++ b/exe/pgbus-health @@ -0,0 +1,9 @@ +#!/usr/bin/env ruby +# frozen_string_literal: true + +# Container HEALTHCHECK probe (issue #386). Deliberately loads ONLY the +# probe file — never the pgbus gem, Bundler, or Rails — because a docker +# HEALTHCHECK runs this every few seconds. +require_relative "../lib/pgbus/health_probe" + +exit Pgbus::HealthProbe.run(ARGV) diff --git a/lib/pgbus/configuration.rb b/lib/pgbus/configuration.rb index 69f5ce45..d39edfc8 100644 --- a/lib/pgbus/configuration.rb +++ b/lib/pgbus/configuration.rb @@ -35,6 +35,16 @@ class Configuration # wait, so recycling/deploy never wedges on a permanently-stuck job. attr_accessor :stall_threshold, :read_timeout, :drain_timeout + # shutdown_timeout bounds how long the supervisor waits for its children + # after forwarding TERM before escalating to SIGKILL. nil (default) derives + # drain_timeout + SHUTDOWN_TIMEOUT_MARGIN, so raising drain_timeout keeps + # the supervisor's deadline above the workers' drain window. An orchestrator + # stop grace period (Kamal stop_timeout, Kubernetes terminationGracePeriod) + # should exceed this value, or docker SIGKILLs the whole tree first. + attr_writer :shutdown_timeout + + SHUTDOWN_TIMEOUT_MARGIN = 5 + # Dispatcher settings attr_accessor :dispatch_interval @@ -238,6 +248,7 @@ def initialize @stall_threshold = 90 @read_timeout = 30 @drain_timeout = 30 + @shutdown_timeout = nil @dispatch_interval = 1.0 @@ -712,6 +723,8 @@ def validate! end raise Pgbus::ConfigurationError, "drain_timeout must be > 0" unless drain_timeout.is_a?(Numeric) && drain_timeout.positive? + validate_shutdown_timeout! + unless stats_flush_size.is_a?(Integer) && stats_flush_size.positive? raise Pgbus::ConfigurationError, "stats_flush_size must be a positive integer" end @@ -765,6 +778,25 @@ def validate! self end + # An explicit shutdown_timeout must be a positive number; nil keeps the + # derived drain_timeout + margin default. A value below drain_timeout is + # legal but self-defeating (the supervisor SIGKILLs workers mid-drain), so + # it warns instead of raising. + def validate_shutdown_timeout! + explicit = @shutdown_timeout + unless explicit.nil? || (explicit.is_a?(Numeric) && explicit.positive?) + raise Pgbus::ConfigurationError, + "shutdown_timeout must be a positive number or nil (defaults to drain_timeout + #{SHUTDOWN_TIMEOUT_MARGIN})" + end + + return unless explicit && explicit < drain_timeout + + Pgbus.logger.warn do + "[Pgbus] shutdown_timeout (#{explicit}s) is below drain_timeout (#{drain_timeout}s) — " \ + "the supervisor will SIGKILL workers before their drain window ends" + end + end + # Pre-1.0 surface-freeze: reject malformed values for core job-path keys at # boot rather than failing deep in a worker/dispatcher/poller/scheduler # thread, per-enqueue, or by silently corrupting queue names / leaving the @@ -1237,6 +1269,12 @@ def dashboard_filter_sensitive=(value) # because only one runs at a time per reactor thread. ASYNC_POOL_CONNECTIONS = 3 + # Resolved supervisor SIGKILL deadline: the explicit value when set, + # otherwise drain_timeout + SHUTDOWN_TIMEOUT_MARGIN (see attr_writer docs). + def shutdown_timeout + @shutdown_timeout || (drain_timeout + SHUTDOWN_TIMEOUT_MARGIN) + end + def resolved_pool_size return pool_size if pool_size diff --git a/lib/pgbus/health_probe.rb b/lib/pgbus/health_probe.rb new file mode 100644 index 00000000..0d86dd37 --- /dev/null +++ b/lib/pgbus/health_probe.rb @@ -0,0 +1,123 @@ +# frozen_string_literal: true + +require "socket" + +module Pgbus + # Dependency-free readiness probe for container HEALTHCHECKs (issue #386). + # + # exe/pgbus-health loads this file via require_relative and nothing else: + # a docker HEALTHCHECK runs the probe every few seconds, so it must never + # drag in Bundler, Zeitwerk, Rails, or the rest of the gem. Only Ruby's + # bundled socket library is allowed here. + # + # healthcheck: + # cmd: bin/pgbus-health # port from PGBUS_HEALTH_PORT + # cmd: bin/pgbus-health --port 9394 --path /livez + # + # Exit codes: 0 healthy (HTTP 2xx), 1 unhealthy (non-2xx, refused, timeout), + # 2 usage error (no/invalid port). + class HealthProbe + EXIT_OK = 0 + EXIT_UNHEALTHY = 1 + EXIT_USAGE = 2 + + DEFAULT_PATH = "/readyz" + DEFAULT_TIMEOUT = 2.0 + HOST = "127.0.0.1" + + USAGE = "usage: pgbus-health [--port PORT] [--path PATH] [--timeout SECONDS]\n " \ + "port falls back to the PGBUS_HEALTH_PORT environment variable\n" + + def self.run(argv, env: ENV, out: $stdout, err: $stderr) + new(argv, env: env, out: out, err: err).run + end + + def initialize(argv, env: ENV, out: $stdout, err: $stderr) + @out = out + @err = err + @path = DEFAULT_PATH + @timeout = DEFAULT_TIMEOUT + @port = env["PGBUS_HEALTH_PORT"] + @usage_error = false + parse(argv) + end + + def run + return usage_failure if @usage_error + + port = Integer(@port, exception: false) + return usage_failure unless port + + probe(port) + end + + private + + # Hand-rolled flag parsing: three flags do not justify optparse in a + # script whose reason to exist is loading nothing. + def parse(argv) + args = argv.dup + until args.empty? + flag = args.shift + value = args.shift + case flag + when "--port" then @port = value + when "--path" then @path = value + when "--timeout" then @timeout = value.to_f + else + return @usage_error = true + end + return @usage_error = true if value.nil? + end + end + + def usage_failure + @err.write(USAGE) + EXIT_USAGE + end + + def probe(port) + status = http_status(port) + healthy = status&.between?(200, 299) + @out.write("pgbus-health: #{@path} -> #{status || "no response"}\n") + healthy ? EXIT_OK : EXIT_UNHEALTHY + rescue SystemCallError, IOError => e + @err.write("pgbus-health: #{@path} -> #{e.class}: #{e.message}\n") + EXIT_UNHEALTHY + end + + # Minimal HTTP/1.0 exchange: send the request, read just the status line. + # The deadline covers connect and read together. + def http_status(port) + deadline = monotonic_now + @timeout + Socket.tcp(HOST, port, connect_timeout: @timeout) do |sock| + sock.write("GET #{@path} HTTP/1.0\r\nHost: #{HOST}\r\nConnection: close\r\n\r\n") + line = read_status_line(sock, deadline) + code = line&.split(" ", 3)&.fetch(1, nil) + Integer(code, exception: false) + end + end + + def read_status_line(sock, deadline) + buffer = +"" + until buffer.include?("\n") + remaining = deadline - monotonic_now + return nil if remaining <= 0 || !sock.wait_readable(remaining) + + chunk = sock.read_nonblock(1024, exception: false) + return nil if chunk.nil? # EOF before a full status line + next if chunk == :wait_readable # spurious wakeup — re-wait on the deadline + + buffer << chunk + end + buffer[/\A[^\r\n]*/] + end + + # ::Process, not Process — inside the Pgbus namespace the bare constant + # resolves to Pgbus::Process (the process model), which is also why this + # file must never be renamed into that namespace. + def monotonic_now + ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + end + end +end diff --git a/lib/pgbus/process/consumer.rb b/lib/pgbus/process/consumer.rb index aa24ba72..bad6429b 100644 --- a/lib/pgbus/process/consumer.rb +++ b/lib/pgbus/process/consumer.rb @@ -480,7 +480,10 @@ def emit_pool_stats def shutdown stop_wake_source @pool.shutdown - @pool.wait_for_termination(30) + # The consumer has no quiesce-gated drain loop like Worker's, so this + # wait IS its drain window — bound it by the same knob workers use + # instead of a hardcoded 30s (issue #386). + @pool.wait_for_termination(config.drain_timeout) @stat_buffer&.stop @heartbeat&.stop restore_signals diff --git a/lib/pgbus/process/readiness_snapshot.rb b/lib/pgbus/process/readiness_snapshot.rb new file mode 100644 index 00000000..ace6cf43 --- /dev/null +++ b/lib/pgbus/process/readiness_snapshot.rb @@ -0,0 +1,30 @@ +# frozen_string_literal: true + +module Pgbus + module Process + # Immutable container-local readiness state, published by the supervisor + # (one atomic swap per monitor pass) and read by the standalone health + # server's accept thread — the immutability is what makes the cross-thread + # handoff safe without a lock (issue #386). + # + # `expected` is the child count forked by boot_processes; `live` is the + # current fork-table size. A child sitting in crash-restart backoff keeps + # `live < expected`, which is exactly the signal a rolling deploy's health + # gate needs to fail on: the replacement container never goes ready, and + # the orchestrator keeps the old container running. + ReadinessSnapshot = Data.define(:booted, :shutting_down, :expected, :live) do + def ready? + booted && !shutting_down && live >= expected + end + + # DRAINING wins over BOOTING: a supervisor told to stop mid-boot is + # leaving, not arriving, and must never look like it will become ready. + def status + return "DRAINING" if shutting_down + return "BOOTING" unless booted + + ready? ? "OK" : "DEGRADED" + end + end + end +end diff --git a/lib/pgbus/process/supervisor.rb b/lib/pgbus/process/supervisor.rb index c45cc059..a17e72cc 100644 --- a/lib/pgbus/process/supervisor.rb +++ b/lib/pgbus/process/supervisor.rb @@ -44,6 +44,15 @@ def initialize(config: Pgbus.configuration, forks: {}, shutting_down: false, @pending_restarts = pending_restarts @crash_counts = Hash.new(0) @notify_hub = notify_hub + @readiness = Concurrent::AtomicReference.new( + ReadinessSnapshot.new(booted: false, shutting_down: shutting_down, expected: 0, live: forks.size) + ) + end + + # The current container-local readiness state. Safe to call from any + # thread (the health server's accept thread reads it per probe). + def readiness_snapshot + @readiness.get end def shutting_down? @@ -97,6 +106,7 @@ def run start_notify_hub boot_processes + mark_booted monitor_loop ensure shutdown @@ -105,17 +115,40 @@ def run def graceful_shutdown Pgbus.logger.info { "[Pgbus] Supervisor: graceful shutdown requested" } @shutting_down = true + refresh_readiness signal_children("TERM") end def immediate_shutdown Pgbus.logger.warn { "[Pgbus] Supervisor: immediate shutdown requested" } @shutting_down = true + refresh_readiness signal_children("QUIT") end private + # Boot is complete: connection verified, queues bootstrapped, every + # configured child forked. The fork-table size at this instant becomes + # the readiness baseline — roles that legitimately declined to boot + # (scheduler with no recurring tasks) are simply absent from it. + def mark_booted + @booted = true + @expected_children = @forks.size + refresh_readiness + end + + # Publish a fresh snapshot; the swapped-in Data is immutable, so the + # health server's accept thread always reads a consistent state. + def refresh_readiness + @readiness.set( + ReadinessSnapshot.new( + booted: !!@booted, shutting_down: @shutting_down, + expected: @expected_children || 0, live: @forks.size + ) + ) + end + # Log a single boot diagnostics banner: the settings that actually # determine whether this deployment works. One consecutive block of # "[Pgbus] boot:"-prefixed info lines so it reads cleanly under both the @@ -537,6 +570,9 @@ def monitor_loop # refresh, and fork status broadcast (issue #381). @notify_hub&.tick end + # After reap + restarts so a clean recycle (reaped and re-forked in + # the same pass) never dips the published live count (issue #386). + refresh_readiness interruptible_sleep(FORK_WAIT) end end @@ -821,7 +857,12 @@ def start_heartbeat def start_health_server return unless config.health_port - @health_server = Pgbus::Web::HealthServer.new(port: config.health_port, bind: config.health_bind) + # The standalone server answers /readyz from THIS supervisor's + # container-local snapshot — a rolling deploy's health gate must + # measure the new container, not the fleet-wide verdict a sibling + # container's workers can satisfy (issue #386). + app = Pgbus::Web::HealthApp.new(local_readiness: -> { readiness_snapshot }) + @health_server = Pgbus::Web::HealthServer.new(port: config.health_port, bind: config.health_bind, app: app) @health_server.start end @@ -876,8 +917,11 @@ def start_notify_hub end def shutdown - # Wait for all children with timeout - deadline = Time.now + 30 + # Wait for children to drain and exit, bounded by config.shutdown_timeout + # (default drain_timeout + 5) so raising the drain window can never + # mean SIGKILLing workers mid-drain. An orchestrator's stop grace + # period should exceed this value (issue #386). + deadline = Time.now + config.shutdown_timeout until @forks.empty? || Time.now > deadline reap_children diff --git a/lib/pgbus/process/worker.rb b/lib/pgbus/process/worker.rb index 47119ac1..d0f20da6 100644 --- a/lib/pgbus/process/worker.rb +++ b/lib/pgbus/process/worker.rb @@ -155,6 +155,12 @@ def last_loop_tick NOTIFY_RETRY_BASE_SECONDS = 5 NOTIFY_RETRY_MAX_SECONDS = 300 + # Residual pool wait in #shutdown, AFTER the drain loop already spent up + # to config.drain_timeout on in-flight jobs. Short by design: a job still + # running has proven it won't finish, and this wait competes with the + # supervisor's shutdown_timeout deadline (issue #386). + POOL_TERMINATION_WAIT = 5 + def run setup_signals start_heartbeat @@ -175,9 +181,9 @@ def run break if @lifecycle.stopped? # quiesced? (all slots free), not idle? (any slot free) — exiting - # with work still in flight abandons those jobs to the 30s - # wait_for_termination timeout in shutdown. Bounded by - # config.drain_timeout so a stuck job can't wedge the loop forever. + # with work still in flight abandons those jobs to shutdown's short + # POOL_TERMINATION_WAIT residual. Bounded by config.drain_timeout so + # a stuck job can't wedge the loop forever. break if @lifecycle.draining? && (@pool.quiesced? || drain_deadline_exceeded?) claim_and_execute if @lifecycle.can_process? @@ -792,7 +798,11 @@ def shutdown Pgbus.logger.info { "[Pgbus] Worker draining thread pool..." } stop_wake_source @pool.shutdown - @pool.wait_for_termination(30) + # Residual wait only: the drain loop already waited up to + # config.drain_timeout for in-flight jobs. A job still running here has + # proven it won't finish; waiting another full window would push the + # worker past the supervisor's shutdown_timeout deadline (issue #386). + @pool.wait_for_termination(POOL_TERMINATION_WAIT) @stat_buffer&.stop @queue_lock&.unlock_all @heartbeat&.stop diff --git a/lib/pgbus/web/health_app.rb b/lib/pgbus/web/health_app.rb index a23fb5e6..a1474625 100644 --- a/lib/pgbus/web/health_app.rb +++ b/lib/pgbus/web/health_app.rb @@ -49,8 +49,15 @@ class HealthApp # @param data_source [Pgbus::Web::DataSource, nil] read layer for /readyz. # nil (the default) builds a fresh DataSource per readiness check, which # avoids serving stale metrics from a long-lived app's memoized instance. - def initialize(data_source: nil) + # @param local_readiness [#call, nil] when set, /readyz answers from this + # callable's {Process::ReadinessSnapshot} instead of the cluster-wide + # analyzer — the supervisor's standalone HealthServer passes its own + # snapshot so a rolling deploy's health gate measures THIS container, + # not the fleet (issue #386). The Rails-mounted app leaves it nil and + # keeps the cluster verdict. + def initialize(data_source: nil, local_readiness: nil) @data_source = data_source + @local_readiness = local_readiness end def call(env) @@ -69,6 +76,8 @@ def livez end def readyz + return local_readyz if @local_readiness + # HealthAnalyzer lives in the MCP namespace, which is excluded from # Zeitwerk (its *tools* subclass the optional `mcp` gem). The analyzer # itself has no gem dependency, so require just that one file — the @@ -83,6 +92,19 @@ def readyz [503, JSON_HEADERS.dup, [{ status: "ERROR", error: e.message }.to_json]] end + # Container-local readiness: no database, no analyzer — just the + # supervisor's published snapshot. The error path mirrors the cluster + # readyz: 503 ERROR, logged, never swallowed. + def local_readyz + snapshot = @local_readiness.call + status = snapshot.ready? ? 200 : 503 + body = { status: snapshot.status, expected: snapshot.expected, live: snapshot.live } + [status, JSON_HEADERS.dup, [body.to_json]] + rescue StandardError => e + Pgbus.logger.error { "[Pgbus::Web::HealthApp] local readiness check failed: #{e.class}: #{e.message}" } + [503, JSON_HEADERS.dup, [{ status: "ERROR", error: e.message }.to_json]] + end + # Reuse an injected DataSource (tests, an app that wants one shared # instance); otherwise build a fresh one each check so per-instance # memoization can never serve stale queue/process metrics. diff --git a/spec/pgbus/configuration_spec.rb b/spec/pgbus/configuration_spec.rb index 0869343b..54dcf455 100644 --- a/spec/pgbus/configuration_spec.rb +++ b/spec/pgbus/configuration_spec.rb @@ -169,6 +169,22 @@ expect(config.drain_timeout).to eq(30) end + it "defaults shutdown_timeout to drain_timeout + 5" do + expect(config.shutdown_timeout).to eq(35) + end + + it "keeps shutdown_timeout above a raised drain_timeout" do + config.drain_timeout = 60 + + expect(config.shutdown_timeout).to eq(65) + end + + it "respects an explicit shutdown_timeout over the derived default" do + config.shutdown_timeout = 120 + + expect(config.shutdown_timeout).to eq(120) + end + it "enables eager_validation by default" do expect(config.eager_validation).to be(true) end @@ -1055,6 +1071,40 @@ expect { config.validate! }.not_to raise_error end + it "rejects non-numeric shutdown_timeout" do + config.shutdown_timeout = "45" + expect { config.validate! }.to raise_error(Pgbus::ConfigurationError, /shutdown_timeout/) + end + + it "rejects zero shutdown_timeout" do + config.shutdown_timeout = 0 + expect { config.validate! }.to raise_error(Pgbus::ConfigurationError, /shutdown_timeout/) + end + + it "accepts nil shutdown_timeout (derived default)" do + config.shutdown_timeout = nil + expect { config.validate! }.not_to raise_error + end + + it "warns when an explicit shutdown_timeout is below drain_timeout" do + allow(Pgbus.logger).to receive(:warn) + config.drain_timeout = 60 + config.shutdown_timeout = 45 + + config.validate! + + expect(Pgbus.logger).to have_received(:warn) + end + + it "does not warn when shutdown_timeout covers drain_timeout" do + allow(Pgbus.logger).to receive(:warn) + config.shutdown_timeout = 45 + + config.validate! + + expect(Pgbus.logger).not_to have_received(:warn) + end + it "rejects zero stats_flush_size" do config.stats_flush_size = 0 expect { config.validate! }.to raise_error(Pgbus::ConfigurationError, /stats_flush_size/) diff --git a/spec/pgbus/health_probe_spec.rb b/spec/pgbus/health_probe_spec.rb new file mode 100644 index 00000000..30700824 --- /dev/null +++ b/spec/pgbus/health_probe_spec.rb @@ -0,0 +1,102 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Pgbus::HealthProbe do + let(:out) { StringIO.new } + let(:err) { StringIO.new } + + def run_probe(argv, env: {}) + described_class.run(argv, env: env, out: out, err: err) + end + + # A real HealthServer bound to an OS-assigned port, answering with a fixed + # Rack response — the probe speaks actual HTTP to an actual socket. + def with_server(status) + app = ->(_env) { [status, { "Content-Type" => "application/json" }, ["{}"]] } + server = Pgbus::Web::HealthServer.new(port: 0, app: app) + server.start + yield server.port + ensure + server&.stop + end + + it "exits 0 when the endpoint answers 200" do + with_server(200) do |port| + expect(run_probe(["--port", port.to_s])).to eq(described_class::EXIT_OK) + end + end + + it "exits 1 when the endpoint answers 503" do + with_server(503) do |port| + expect(run_probe(["--port", port.to_s])).to eq(described_class::EXIT_UNHEALTHY) + end + end + + it "exits 1 when nothing listens on the port" do + # Bind then release a port so we hold a port number that refuses connections. + server = TCPServer.new("127.0.0.1", 0) + port = server.addr[1] + server.close + + expect(run_probe(["--port", port.to_s])).to eq(described_class::EXIT_UNHEALTHY) + end + + it "reads the port from PGBUS_HEALTH_PORT when no flag is given" do + with_server(200) do |port| + expect(run_probe([], env: { "PGBUS_HEALTH_PORT" => port.to_s })).to eq(described_class::EXIT_OK) + end + end + + it "probes the given --path" do + captured_path = nil + app = lambda do |env| + captured_path = env["PATH_INFO"] + [200, { "Content-Type" => "text/plain" }, ["ok"]] + end + server = Pgbus::Web::HealthServer.new(port: 0, app: app) + server.start + + run_probe(["--port", server.port.to_s, "--path", "/livez"]) + + expect(captured_path).to eq("/livez") + ensure + server&.stop + end + + it "defaults the path to /readyz" do + captured_path = nil + app = lambda do |env| + captured_path = env["PATH_INFO"] + [200, { "Content-Type" => "text/plain" }, ["ok"]] + end + server = Pgbus::Web::HealthServer.new(port: 0, app: app) + server.start + + run_probe(["--port", server.port.to_s]) + + expect(captured_path).to eq("/readyz") + ensure + server&.stop + end + + it "exits 2 with usage on stderr when no port is available anywhere" do + expect(run_probe([])).to eq(described_class::EXIT_USAGE) + expect(err.string).to include("--port") + end + + it "exits 2 on a non-numeric port" do + expect(run_probe(["--port", "banana"])).to eq(described_class::EXIT_USAGE) + end + + # The whole point of the probe: a docker HEALTHCHECK runs it every few + # seconds, so it must never drag in Bundler, Zeitwerk, or the pgbus gem. + it "loads standalone without pulling in the gem" do + script = 'require_relative "lib/pgbus/health_probe"; ' \ + "exit(defined?(Pgbus::Client) || defined?(Zeitwerk) || defined?(Pgbus::Web) ? 1 : 0)" + result = system(RbConfig.ruby, "--disable-gems", "-e", script, + chdir: File.expand_path("../..", __dir__)) + + expect(result).to be true + end +end diff --git a/spec/pgbus/process/consumer_spec.rb b/spec/pgbus/process/consumer_spec.rb index cd15886d..47a487fe 100644 --- a/spec/pgbus/process/consumer_spec.rb +++ b/spec/pgbus/process/consumer_spec.rb @@ -626,6 +626,16 @@ def wait_for(timeout: 2) consumer.send(:shutdown) expect(fake_listener).to have_received(:stop) end + + it "bounds the drain wait by config.drain_timeout, not a hardcoded 30s (issue #386)" do + config = Pgbus::Configuration.new + config.drain_timeout = 42 + consumer = described_class.new(topics: ["orders.#"], config: config) + + consumer.send(:shutdown) + + expect(mock_pool).to have_received(:wait_for_termination).with(42) + end end # Operational parity with Worker (issue #274). A consumer fork must be diff --git a/spec/pgbus/process/readiness_snapshot_spec.rb b/spec/pgbus/process/readiness_snapshot_spec.rb new file mode 100644 index 00000000..e16e1cea --- /dev/null +++ b/spec/pgbus/process/readiness_snapshot_spec.rb @@ -0,0 +1,63 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Pgbus::Process::ReadinessSnapshot do + describe "#ready?" do + it "is not ready before boot completes" do + snapshot = described_class.new(booted: false, shutting_down: false, expected: 0, live: 0) + + expect(snapshot.ready?).to be false + end + + it "is ready once booted with all expected children live" do + snapshot = described_class.new(booted: true, shutting_down: false, expected: 3, live: 3) + + expect(snapshot.ready?).to be true + end + + it "is not ready when a child is missing" do + snapshot = described_class.new(booted: true, shutting_down: false, expected: 3, live: 2) + + expect(snapshot.ready?).to be false + end + + it "is not ready while shutting down, even with all children live" do + snapshot = described_class.new(booted: true, shutting_down: true, expected: 3, live: 3) + + expect(snapshot.ready?).to be false + end + + it "is ready for a zero-child deployment once booted" do + snapshot = described_class.new(booted: true, shutting_down: false, expected: 0, live: 0) + + expect(snapshot.ready?).to be true + end + end + + describe "#status" do + it "reports BOOTING before boot completes" do + snapshot = described_class.new(booted: false, shutting_down: false, expected: 0, live: 0) + + expect(snapshot.status).to eq("BOOTING") + end + + it "reports OK when ready" do + snapshot = described_class.new(booted: true, shutting_down: false, expected: 2, live: 2) + + expect(snapshot.status).to eq("OK") + end + + it "reports DEGRADED when a child is missing" do + snapshot = described_class.new(booted: true, shutting_down: false, expected: 2, live: 1) + + expect(snapshot.status).to eq("DEGRADED") + end + + it "reports DRAINING while shutting down, taking precedence over BOOTING" do + snapshot = described_class.new(booted: false, shutting_down: true, expected: 0, live: 0) + + expect(snapshot.status).to eq("DRAINING") + end + end +end diff --git a/spec/pgbus/process/supervisor_spec.rb b/spec/pgbus/process/supervisor_spec.rb index ddf96345..a94bbb2d 100644 --- a/spec/pgbus/process/supervisor_spec.rb +++ b/spec/pgbus/process/supervisor_spec.rb @@ -873,10 +873,26 @@ def banner supervisor.run - expect(Pgbus::Web::HealthServer).to have_received(:new).with(port: 9394, bind: "0.0.0.0") + expect(Pgbus::Web::HealthServer).to have_received(:new) + .with(port: 9394, bind: "0.0.0.0", app: an_instance_of(Pgbus::Web::HealthApp)) expect(health_server).to have_received(:start) end + it "wires /readyz to the supervisor's container-local readiness (issue #386)" do + config.health_port = 9394 + captured_app = nil + allow(Pgbus::Web::HealthServer).to receive(:new) do |**kwargs| + captured_app = kwargs[:app] + health_server + end + + supervisor.run + status, _headers, body = captured_app.call("REQUEST_METHOD" => "GET", "PATH_INFO" => "/readyz") + + expect(status).to eq(200) + expect(JSON.parse(body.join)).to include("status" => "OK", "expected" => 0, "live" => 0) + end + it "starts the health server after the heartbeat" do config.health_port = 9394 call_order = [] @@ -897,6 +913,93 @@ def banner end end + describe "readiness snapshot (issue #386)" do + it "starts BOOTING and not ready" do + snapshot = described_class.new.readiness_snapshot + + expect(snapshot.ready?).to be false + expect(snapshot.status).to eq("BOOTING") + end + + it "becomes ready once boot completes with all forked children counted" do + supervisor = described_class.new(forks: { 101 => { type: :worker }, 102 => { type: :dispatcher } }) + + supervisor.send(:mark_booted) + snapshot = supervisor.readiness_snapshot + + expect(snapshot.ready?).to be true + expect(snapshot.expected).to eq(2) + expect(snapshot.live).to eq(2) + end + + it "degrades when a child dies and has not been restarted" do + supervisor = described_class.new(forks: { 101 => { type: :worker }, 102 => { type: :worker } }) + supervisor.send(:mark_booted) + + supervisor.forks.delete(102) + supervisor.send(:refresh_readiness) + snapshot = supervisor.readiness_snapshot + + expect(snapshot.ready?).to be false + expect(snapshot.status).to eq("DEGRADED") + expect(snapshot.live).to eq(1) + end + + it "flips to DRAINING on graceful_shutdown" do + supervisor = described_class.new(forks: { 101 => { type: :worker } }) + supervisor.send(:mark_booted) + allow(Process).to receive(:kill) + + supervisor.graceful_shutdown + + expect(supervisor.readiness_snapshot.status).to eq("DRAINING") + expect(supervisor.readiness_snapshot.ready?).to be false + end + + it "flips to DRAINING on immediate_shutdown" do + supervisor = described_class.new(forks: { 101 => { type: :worker } }) + supervisor.send(:mark_booted) + allow(Process).to receive(:kill) + + supervisor.immediate_shutdown + + expect(supervisor.readiness_snapshot.status).to eq("DRAINING") + end + + it "is marked booted by #run after boot_processes, before monitor_loop runs" do + supervisor = described_class.new + mock_client = build_mock_client + allow(Pgbus).to receive(:client).and_return(mock_client) + allow(mock_client).to receive_messages( + verify_connection!: true, ensure_all_queues: nil, pgmq_schema_version: "1.5.0" + ) + allow(supervisor).to receive_messages(setup_signals: nil, log_boot_banner: nil, boot_processes: nil) + seen = nil + allow(supervisor).to receive(:monitor_loop) { seen = supervisor.readiness_snapshot } + + supervisor.run + + expect(seen.booted).to be true + end + end + + describe "shutdown deadline (issue #386)" do + it "waits config.shutdown_timeout for children before escalating to SIGKILL" do + config = Pgbus::Configuration.new + config.shutdown_timeout = 42 + supervisor = described_class.new(config: config, forks: { 5001 => { type: :worker } }) + allow(supervisor).to receive(:interruptible_sleep) + allow(Process).to receive(:waitpid2).and_return(nil) + allow(Process).to receive(:kill) + t0 = Time.now + allow(Time).to receive(:now).and_return(t0, t0 + 41, t0 + 43) + + supervisor.send(:shutdown) + + expect(Process).to have_received(:kill).with("KILL", 5001) + end + end + describe "recurring_tasks_configured? (private)" do let(:supervisor) { described_class.new } diff --git a/spec/pgbus/process/worker_spec.rb b/spec/pgbus/process/worker_spec.rb index 5015a3fa..09c82f1b 100644 --- a/spec/pgbus/process/worker_spec.rb +++ b/spec/pgbus/process/worker_spec.rb @@ -1548,6 +1548,16 @@ def wait_for(timeout: 2) worker.notify_listener = nil expect { worker.send(:shutdown) }.not_to raise_error end + + it "bounds the residual pool wait with POOL_TERMINATION_WAIT, not a second drain window (issue #386)" do + # The drain loop already waited up to config.drain_timeout for in-flight + # jobs; a job still running here has proven it won't finish, and a long + # residual wait only pushes the worker past the supervisor's + # shutdown_timeout deadline. + worker.send(:shutdown) + + expect(pool).to have_received(:wait_for_termination).with(described_class::POOL_TERMINATION_WAIT) + end end end diff --git a/spec/pgbus/web/health_app_spec.rb b/spec/pgbus/web/health_app_spec.rb index 348fe0b0..2d9075db 100644 --- a/spec/pgbus/web/health_app_spec.rb +++ b/spec/pgbus/web/health_app_spec.rb @@ -133,6 +133,93 @@ def body_of(response) end end + describe "container-local readiness (issue #386)" do + subject(:app) { described_class.new(local_readiness: -> { snapshot }) } + + let(:snapshot) do + Pgbus::Process::ReadinessSnapshot.new(booted: true, shutting_down: false, expected: 3, live: 3) + end + + it "returns 200 with the local status when the snapshot is ready" do + status, headers, = get("/readyz") + + expect(status).to eq(200) + expect(headers["Content-Type"]).to eq("application/json") + end + + it "includes status, expected, and live in the body" do + body = JSON.parse(body_of(get("/readyz"))) + + expect(body).to eq("status" => "OK", "expected" => 3, "live" => 3) + end + + it "never builds the cluster analyzer in local mode" do + get("/readyz") + + expect(Pgbus::MCP::HealthAnalyzer).not_to have_received(:new) + end + + it "leaves /livez unconditional" do + status, = get("/livez") + + expect(status).to eq(200) + end + + context "when the container is still booting" do + let(:snapshot) do + Pgbus::Process::ReadinessSnapshot.new(booted: false, shutting_down: false, expected: 0, live: 0) + end + + it "returns 503 BOOTING" do + status, = get("/readyz") + + expect(status).to eq(503) + expect(JSON.parse(body_of(get("/readyz")))["status"]).to eq("BOOTING") + end + end + + context "when the container is draining" do + let(:snapshot) do + Pgbus::Process::ReadinessSnapshot.new(booted: true, shutting_down: true, expected: 3, live: 3) + end + + it "returns 503 DRAINING" do + status, = get("/readyz") + + expect(status).to eq(503) + expect(JSON.parse(body_of(get("/readyz")))["status"]).to eq("DRAINING") + end + end + + context "when a child is missing" do + let(:snapshot) do + Pgbus::Process::ReadinessSnapshot.new(booted: true, shutting_down: false, expected: 3, live: 2) + end + + it "returns 503 DEGRADED with the counts" do + status, = get("/readyz") + body = JSON.parse(body_of(get("/readyz"))) + + expect(status).to eq(503) + expect(body).to eq("status" => "DEGRADED", "expected" => 3, "live" => 2) + end + end + + context "when the readiness callable raises" do + subject(:app) { described_class.new(local_readiness: -> { raise StandardError, "boom" }) } + + it "returns 503 with an ERROR body and logs" do + allow(Pgbus.logger).to receive(:error) + + status, = get("/readyz") + + expect(status).to eq(503) + expect(JSON.parse(body_of(get("/readyz")))["status"]).to eq("ERROR") + expect(Pgbus.logger).to have_received(:error).at_least(:once) + end + end + end + describe "default DataSource" do it "builds a fresh Pgbus::Web::DataSource when none is injected" do allow(Pgbus::Web::DataSource).to receive(:new).and_return(data_source) From c0de3081d66f98919b08f5823b91667ede387d03 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Mon, 3 Aug 2026 07:42:02 +0200 Subject: [PATCH 2/3] docs(site): Rolling restarts page + container-local /readyz semantics New Operations page covering the health-gated rolling restart flow: the container-local readiness gate, pgbus-health HEALTHCHECK wiring, the stop_timeout > shutdown_timeout > drain_timeout alignment rule, overlap-window duplicate-supervisor safety, and the read_ct-vs-deploy- kill DLQ caveat. Observability page's probe table now distinguishes mounted (cluster verdict) from standalone (container-local) /readyz, and the config reference documents shutdown_timeout (drift spec green). Refs #386 --- docs/app/models/config_reference.rb | 5 +- docs/app/models/doc.rb | 6 + docs/app/views/docs/pages/observability.rb | 18 +- docs/app/views/docs/pages/rolling_restarts.rb | 190 ++++++++++++++++++ 4 files changed, 212 insertions(+), 7 deletions(-) create mode 100644 docs/app/views/docs/pages/rolling_restarts.rb diff --git a/docs/app/models/config_reference.rb b/docs/app/models/config_reference.rb index 8cdf47ed..0769aeb0 100644 --- a/docs/app/models/config_reference.rb +++ b/docs/app/models/config_reference.rb @@ -106,7 +106,10 @@ module ConfigReference { name: "health_bind", type: "String", default: '"127.0.0.1"', desc: "Bind address for the health server." }, { name: "stall_threshold", type: "Numeric", default: "300", desc: "Seconds without progress before a worker is stalled." }, { name: "read_timeout", type: "Numeric", default: "30", desc: "Read timeout for worker fetches." }, - { name: "drain_timeout", type: "Numeric", default: "30", desc: "Seconds to wait for in-flight jobs to finish during graceful shutdown before abandoning them." } + { name: "drain_timeout", type: "Numeric", default: "30", desc: "Seconds to wait for in-flight jobs to finish during graceful shutdown before abandoning them." }, + { name: "shutdown_timeout", type: "Numeric, nil", default: "drain_timeout + 5", + desc: "Seconds the supervisor waits for children after TERM before SIGKILL. nil derives drain_timeout + 5; " \ + "an orchestrator's stop grace period (Kamal stop_timeout, K8s terminationGracePeriodSeconds) must exceed it." } ], "Streams (SSE)" => [ { name: "streams_enabled", type: "Boolean", default: "true", desc: "Enable the SSE streams transport." }, diff --git a/docs/app/models/doc.rb b/docs/app/models/doc.rb index 48bc696e..12ee7ac6 100644 --- a/docs/app/models/doc.rb +++ b/docs/app/models/doc.rb @@ -21,6 +21,7 @@ class Doc page "Installation", group: "Getting started" page "Quick start", group: "Getting started" page "Configuration", group: "Getting started" + page "Rolling restarts", group: "Operations" # Guide page "Architecture", group: "Guide" @@ -33,6 +34,7 @@ class Doc page "Recurring tasks", group: "Guide", slug: "recurring-tasks", view: "RecurringTasks" page "Transactional outbox", group: "Guide", slug: "outbox", view: "Outbox" page "Real-time streams", group: "Guide", slug: "streams", view: "Streams" + page "Rolling restarts", group: "Operations" # Operations page "Running workers", group: "Operations", slug: "running-workers", view: "RunningWorkers" @@ -40,17 +42,21 @@ class Doc page "Observability", group: "Operations" page "Performance & tuning", group: "Operations", slug: "performance-tuning", view: "PerformanceTuning" page "Separate database", group: "Operations", slug: "separate-database", view: "SeparateDatabase" + page "Rolling restarts", group: "Operations" # Testing page "Testing", group: "Testing" + page "Rolling restarts", group: "Operations" # Migrate page "Upgrading pgbus", group: "Migrate", slug: "upgrading-pgbus", view: "UpgradingPgbus" page "From Sidekiq", group: "Migrate", slug: "from-sidekiq", view: "FromSidekiq" page "From SolidQueue", group: "Migrate", slug: "from-solid-queue", view: "FromSolidQueue" page "From GoodJob", group: "Migrate", slug: "from-good-job", view: "FromGoodJob" + page "Rolling restarts", group: "Operations" # Reference page "Configuration reference", group: "Reference", slug: "configuration-reference", view: "ConfigurationReference" page "CLI & generators", group: "Reference", slug: "cli-generators", view: "CliGenerators" + page "Rolling restarts", group: "Operations" end diff --git a/docs/app/views/docs/pages/observability.rb b/docs/app/views/docs/pages/observability.rb index 6c38de04..efe9f384 100644 --- a/docs/app/views/docs/pages/observability.rb +++ b/docs/app/views/docs/pages/observability.rb @@ -199,17 +199,23 @@ def health DocsUI::Section("Health endpoints", description: "Liveness and readiness for Kubernetes.") do md <<~'MD' Pgbus exposes two HTTP probes: `/livez` (is the serving process up?) and - `/readyz` (are queues draining, or is a worker silently wedged?). `/readyz` - runs the same `OK` / `DEGRADED` / `STALLED` verdict as the MCP - `pgbus_health` tool — `DEGRADED` deliberately stays ready; only the - silent-wedge `STALLED` signal fails readiness. + `/readyz`. Readiness means different things in the two places it is served. + Mounted in Rails, `/readyz` runs the same cluster-wide `OK` / `DEGRADED` / + `STALLED` verdict as the MCP `pgbus_health` tool — `DEGRADED` deliberately + stays ready; only the silent-wedge `STALLED` signal fails readiness. Served + standalone from the supervisor (`health_port`), `/readyz` is + **container-local**: did *this* supervisor finish booting, and are all the + children it forked alive? That is the signal a rolling deploy's health gate + needs — see [Rolling restarts](/docs/rolling-restarts). MD DocsUI::Table( [ "Path", "Method", "200", "503", "Touches DB" ], [ [ [ :code, "/livez" ], "GET", [ :md, "always (`ok`)" ], "never", "no" ], - [ [ :code, "/readyz" ], "GET", [ :md, "verdict `OK` or `DEGRADED`" ], - [ :md, "verdict `STALLED`, or DB unreachable (`{\"status\":\"ERROR\"}`)" ], "yes" ] + [ [ :md, "`/readyz` (mounted)" ], "GET", [ :md, "verdict `OK` or `DEGRADED`" ], + [ :md, "verdict `STALLED`, or DB unreachable (`{\"status\":\"ERROR\"}`)" ], "yes" ], + [ [ :md, "`/readyz` (supervisor)" ], "GET", [ :md, "`OK` — booted, all children live" ], + [ :md, "`BOOTING`, `DEGRADED` (child down), `DRAINING` (stopping)" ], "no" ] ] ) md <<~'MD' diff --git a/docs/app/views/docs/pages/rolling_restarts.rb b/docs/app/views/docs/pages/rolling_restarts.rb new file mode 100644 index 00000000..a0588133 --- /dev/null +++ b/docs/app/views/docs/pages/rolling_restarts.rb @@ -0,0 +1,190 @@ +# frozen_string_literal: true + +# Zeitwerk resolves this compact reference through the directory-implied +# namespaces (app/views/docs/pages/ → Views::Docs::Pages), so there's no need +# for the 4-level nested-module ceremony. +class Views::Docs::Pages::RollingRestarts < DocsUI::Page + title "Rolling restarts" + eyebrow "Operations" + + def lead = "Zero-lost-capacity deploys for the job container: health-gate the new supervisor, drain the old one." + + def content + how_it_works + readiness_gate + probe + shutdown_budget + overlap_window + hard_kill_costs + end + + private + + def how_it_works + DocsUI::Section("How a health-gated rolling restart works", + description: "Start new, prove healthy, then stop old.") do + md <<~'MD' + Orchestrators with per-role health checks — Kamal distributions such as the + [`dash` branch](https://github.com/mhenrixon/kamal), or any docker-level + `HEALTHCHECK`-driven deploy — replace a job container in three steps: start + the new container, poll its health check until it reports healthy, and only + then `docker stop` the old one. If the new container never goes healthy, the + old one keeps running. + + Pgbus participates on both ends: the supervisor's standalone `/readyz` is the + health gate for the *new* container, and the graceful-drain pipeline + (`drain_timeout` / `shutdown_timeout`) bounds the stop of the *old* one. + MD + end + end + + def readiness_gate + DocsUI::Section("The readiness gate is container-local", + description: "The standalone /readyz answers for THIS supervisor, not the fleet.") do + md <<~'MD' + When `health_port` is set, the supervisor's `/readyz` answers from its own + state — never the database. That distinction matters precisely during a + rolling deploy: a cluster-wide verdict would let a freshly-booted container + pass the gate on the strength of the *old* container's still-heartbeating + workers, and the orchestrator would stop the old container before the new + one had forked a single child. (The Rails-mounted `Pgbus::Web::HealthApp` + keeps the cluster-wide verdict — the two probes answer different questions.) + + The body is a snapshot of the supervisor's fork table: + MD + DocsUI::Code(<<~'JSON', filename: "GET /readyz", lexer: :json) + { "status": "OK", "expected": 3, "live": 3 } + JSON + DocsUI::Table( + [ "Status", "HTTP", "Meaning" ], + [ + [ [ :code, "BOOTING" ], "503", + "Connection not yet verified, queues not bootstrapped, or children not yet forked." ], + [ [ :code, "OK" ], "200", "Every child forked at boot is currently alive." ], + [ [ :code, "DEGRADED" ], "503", + "A child died and is waiting out crash-restart backoff." ], + [ [ :code, "DRAINING" ], "503", "A stop signal arrived; the container is leaving." ] + ] + ) + DocsUI::Callout(:tip) do + plain "A crash-looping replacement container never reaches " + code { "OK" } + plain " — the deploy gate fails and the orchestrator keeps the old container " + plain "running. That is the failure mode you want." + end + md <<~'MD' + A clean worker recycle (`max_jobs_per_worker`, `max_memory_mb`, + `max_worker_lifetime`) never flaps readiness: the snapshot refreshes after + the reap-and-restart step of each monitor pass, so a recycled worker is + already replaced by the time the next probe reads it. + MD + end + end + + def probe + DocsUI::Section("pgbus-health: the HEALTHCHECK probe", + description: "A dependency-free probe cheap enough for 1–5s intervals.") do + md <<~'MD' + The gem ships a `pgbus-health` executable: plain Ruby and stdlib sockets, + loading neither Bundler, nor Rails, nor the rest of the gem — so a docker + `HEALTHCHECK` can run it every few seconds, and it works in images without + curl. It GETs `127.0.0.1:/readyz` and exits `0` on HTTP 200, `1` on + anything else (non-200, refused, timeout), `2` on usage errors. + MD + DocsUI::Code(<<~'SH', filename: "shell", lexer: :shell) + pgbus-health --port 9394 # or PGBUS_HEALTH_PORT=9394 pgbus-health + pgbus-health --port 9394 --path /livez --timeout 2 + SH + md <<~'MD' + Wire it into a Kamal role (`bundle binstubs pgbus` generates + `bin/pgbus-health`): + MD + DocsUI::Code(<<~'YAML', filename: "config/deploy.yml", lexer: :yaml) + servers: + job: + hosts: [...] + cmd: bin/pgbus start + healthcheck: + cmd: bin/pgbus-health --port 9394 + interval: 5s + start_period: 30s # cover Rails boot + queue bootstrap + stop_timeout: 45 # must exceed pgbus shutdown_timeout + env: + clear: + PGBUS_HEALTH_PORT: 9394 + YAML + end + end + + def shutdown_budget + DocsUI::Section("Aligning the shutdown budget", + description: "stop_timeout > shutdown_timeout > drain_timeout.") do + md <<~'MD' + On `docker stop`, SIGTERM reaches the supervisor and readiness flips to + `DRAINING`. Children stop claiming new work and finish in-flight jobs for up + to `drain_timeout` (default 30s). The supervisor then waits + `shutdown_timeout` — default `drain_timeout + 5` — before SIGKILLing + stragglers. The orchestrator's stop grace period sits outside both: + MD + DocsUI::Code(<<~'TEXT', filename: "budget alignment", lexer: :text) + orchestrator stop_timeout > pgbus shutdown_timeout > pgbus drain_timeout + 45s 35s (derived) 30s + TEXT + DocsUI::Callout(:warning) do + plain "If the orchestrator's stop grace period is shorter than " + code { "shutdown_timeout" } + plain ", docker SIGKILLs the whole tree mid-drain and the graceful path " + plain "never finishes. Raising " + code { "drain_timeout" } + plain " raises the derived " + code { "shutdown_timeout" } + plain " automatically — raise the orchestrator's stop timeout to match." + end + md <<~'MD' + An explicit `shutdown_timeout` below `drain_timeout` logs a boot warning: + it guarantees mid-drain kills. + MD + end + end + + def overlap_window + DocsUI::Section("The overlap window", + description: "Two supervisors briefly share the database — by design, safely.") do + md <<~'MD' + Between "new container healthy" and "old container stopped", two supervisors + run against the same database. Nothing double-fires: + + - Queue claims use `FOR UPDATE SKIP LOCKED` — a message goes to exactly one + worker regardless of how many are reading. + - `single_active_consumer` queues arbitrate through session-level advisory + locks, released by Postgres the instant a killed process's connection dies. + - Two live recurring schedulers dedup on the `(task_key, run_at)` unique + record — the loser of the insert race skips the occurrence. + - Dispatcher maintenance is idempotent; two dispatchers just do some + redundant work. + + "One scheduler per deployment" is a steady-state rule; a deploy window may + briefly violate it without consequence. + MD + end + end + + def hard_kill_costs + DocsUI::Section("What a hard kill still costs", + description: "At-least-once holds, but read_ct counts deploy kills as failures.") do + md <<~'MD' + Jobs killed past the drain window are redelivered after their visibility + timeout — at-least-once holds. But PGMQ's `read_ct` increments exactly like + a logical failure, so a long-running job that straddles *repeated* deploy + kills can be pushed to the dead-letter queue without its code ever raising. + `zombie_detection` logs exactly this pattern (`read_ct > 1` with no recorded + failure for the message). + + Keep jobs shorter than `drain_timeout`, or raise it (together with the + orchestrator's stop timeout) for queues that can't be. For `idempotent!` + event handlers there is a separate crash-window caveat tracked in + [pgbus#385](https://github.com/mhenrixon/pgbus/issues/385). + MD + end + end +end From 28950b63890436334325d3dce548daacd22d389d Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Mon, 3 Aug 2026 09:30:54 +0200 Subject: [PATCH 3/3] =?UTF-8?q?fix(deploy):=20harden=20review=20findings?= =?UTF-8?q?=20=E2=80=94=20intended-fork=20baseline,=20finite=20shutdown=5F?= =?UTF-8?q?timeout,=20probe=20input=20validation,=20docs=20registry=20dedu?= =?UTF-8?q?pe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit review on #388: - Readiness baseline now counts INTENDED boot-time forks (max of attempt count and fork table), so a failed fork at boot leaves the gate DEGRADED instead of silently lowering expected — a rolling deploy can no longer promote a container missing workers. Restart re-forks after boot never inflate the baseline. - validate! rejects non-finite / non-real shutdown_timeout (INFINITY previously blew up Supervisor#shutdown before child cleanup). - pgbus-health exits 2 on out-of-range ports (was a SocketError backtrace) and on non-numeric/non-positive --timeout (was a silent 0.0 deadline that failed every probe); SocketError added to the probe rescue. - docs page generator had injected the Rolling restarts registry line after every group — deduped to one entry under Operations. - README: language identifier on the budget-alignment fence (MD040). Refs #386 --- README.md | 2 +- docs/app/models/doc.rb | 7 +------ lib/pgbus/configuration.rb | 9 +++++++-- lib/pgbus/health_probe.rb | 17 +++++++++++++---- lib/pgbus/process/supervisor.rb | 25 +++++++++++++++++++++---- spec/pgbus/configuration_spec.rb | 15 +++++++++++++++ spec/pgbus/health_probe_spec.rb | 13 +++++++++++++ spec/pgbus/process/supervisor_spec.rb | 26 ++++++++++++++++++++++++++ 8 files changed, 97 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index a4b28363..cfeb82ab 100644 --- a/README.md +++ b/README.md @@ -1213,7 +1213,7 @@ env: **The shutdown timeline.** On `docker stop`, SIGTERM reaches the supervisor and readiness flips to `DRAINING`; children stop claiming work and drain in-flight jobs for up to `drain_timeout` (default 30s); the supervisor waits `shutdown_timeout` (default `drain_timeout + 5`) before SIGKILLing stragglers. Align the three knobs outside-in: -``` +```text orchestrator stop_timeout > pgbus shutdown_timeout > pgbus drain_timeout 45s 35s (derived) 30s ``` diff --git a/docs/app/models/doc.rb b/docs/app/models/doc.rb index 12ee7ac6..32de6f55 100644 --- a/docs/app/models/doc.rb +++ b/docs/app/models/doc.rb @@ -21,7 +21,6 @@ class Doc page "Installation", group: "Getting started" page "Quick start", group: "Getting started" page "Configuration", group: "Getting started" - page "Rolling restarts", group: "Operations" # Guide page "Architecture", group: "Guide" @@ -34,7 +33,6 @@ class Doc page "Recurring tasks", group: "Guide", slug: "recurring-tasks", view: "RecurringTasks" page "Transactional outbox", group: "Guide", slug: "outbox", view: "Outbox" page "Real-time streams", group: "Guide", slug: "streams", view: "Streams" - page "Rolling restarts", group: "Operations" # Operations page "Running workers", group: "Operations", slug: "running-workers", view: "RunningWorkers" @@ -42,21 +40,18 @@ class Doc page "Observability", group: "Operations" page "Performance & tuning", group: "Operations", slug: "performance-tuning", view: "PerformanceTuning" page "Separate database", group: "Operations", slug: "separate-database", view: "SeparateDatabase" - page "Rolling restarts", group: "Operations" + page "Rolling restarts", group: "Operations" # Testing page "Testing", group: "Testing" - page "Rolling restarts", group: "Operations" # Migrate page "Upgrading pgbus", group: "Migrate", slug: "upgrading-pgbus", view: "UpgradingPgbus" page "From Sidekiq", group: "Migrate", slug: "from-sidekiq", view: "FromSidekiq" page "From SolidQueue", group: "Migrate", slug: "from-solid-queue", view: "FromSolidQueue" page "From GoodJob", group: "Migrate", slug: "from-good-job", view: "FromGoodJob" - page "Rolling restarts", group: "Operations" # Reference page "Configuration reference", group: "Reference", slug: "configuration-reference", view: "ConfigurationReference" page "CLI & generators", group: "Reference", slug: "cli-generators", view: "CliGenerators" - page "Rolling restarts", group: "Operations" end diff --git a/lib/pgbus/configuration.rb b/lib/pgbus/configuration.rb index d39edfc8..396e8000 100644 --- a/lib/pgbus/configuration.rb +++ b/lib/pgbus/configuration.rb @@ -784,9 +784,14 @@ def validate! # it warns instead of raising. def validate_shutdown_timeout! explicit = @shutdown_timeout - unless explicit.nil? || (explicit.is_a?(Numeric) && explicit.positive?) + # Finite real only: Float::INFINITY would blow up Supervisor#shutdown's + # `Time.now + shutdown_timeout` before any child cleanup ran, and a + # Complex would crash `positive?` — reject both here, at boot. + valid = explicit.is_a?(Numeric) && explicit.real? && explicit.finite? && explicit.positive? + unless explicit.nil? || valid raise Pgbus::ConfigurationError, - "shutdown_timeout must be a positive number or nil (defaults to drain_timeout + #{SHUTDOWN_TIMEOUT_MARGIN})" + "shutdown_timeout must be a positive finite number or nil " \ + "(defaults to drain_timeout + #{SHUTDOWN_TIMEOUT_MARGIN})" end return unless explicit && explicit < drain_timeout diff --git a/lib/pgbus/health_probe.rb b/lib/pgbus/health_probe.rb index 0d86dd37..645b3013 100644 --- a/lib/pgbus/health_probe.rb +++ b/lib/pgbus/health_probe.rb @@ -46,7 +46,9 @@ def run return usage_failure if @usage_error port = Integer(@port, exception: false) - return usage_failure unless port + # Out-of-range ports would reach Socket.tcp and raise SocketError — a + # backtrace where a HEALTHCHECK needs a deterministic exit code. + return usage_failure unless port&.between?(1, 65_535) probe(port) end @@ -60,14 +62,21 @@ def parse(argv) until args.empty? flag = args.shift value = args.shift + return @usage_error = true if value.nil? + case flag when "--port" then @port = value when "--path" then @path = value - when "--timeout" then @timeout = value.to_f + when "--timeout" + # A typo'd timeout must be a usage error, not `to_f`'s silent 0.0 — + # a zero deadline reports the container unhealthy on every probe. + timeout = Float(value, exception: false) + return @usage_error = true unless timeout&.positive? + + @timeout = timeout else return @usage_error = true end - return @usage_error = true if value.nil? end end @@ -81,7 +90,7 @@ def probe(port) healthy = status&.between?(200, 299) @out.write("pgbus-health: #{@path} -> #{status || "no response"}\n") healthy ? EXIT_OK : EXIT_UNHEALTHY - rescue SystemCallError, IOError => e + rescue SystemCallError, IOError, SocketError => e @err.write("pgbus-health: #{@path} -> #{e.class}: #{e.message}\n") EXIT_UNHEALTHY end diff --git a/lib/pgbus/process/supervisor.rb b/lib/pgbus/process/supervisor.rb index a17e72cc..a85922c4 100644 --- a/lib/pgbus/process/supervisor.rb +++ b/lib/pgbus/process/supervisor.rb @@ -44,6 +44,7 @@ def initialize(config: Pgbus.configuration, forks: {}, shutting_down: false, @pending_restarts = pending_restarts @crash_counts = Hash.new(0) @notify_hub = notify_hub + @intended_children = 0 @readiness = Concurrent::AtomicReference.new( ReadinessSnapshot.new(booted: false, shutting_down: shutting_down, expected: 0, live: forks.size) ) @@ -129,15 +130,26 @@ def immediate_shutdown private # Boot is complete: connection verified, queues bootstrapped, every - # configured child forked. The fork-table size at this instant becomes - # the readiness baseline — roles that legitimately declined to boot - # (scheduler with no recurring tasks) are simply absent from it. + # configured child fork ATTEMPTED. The baseline is the larger of the + # intended-attempt count and the fork-table size: a boot-time fork + # failure (EAGAIN/ENOMEM, logged-and-swallowed in fork_*) leaves + # intended > live, so the readiness gate reports DEGRADED instead of + # blessing a container that is missing workers. Roles that legitimately + # declined to boot (scheduler with no recurring tasks) never reach a + # fork_* method and are counted by neither side. def mark_booted @booted = true - @expected_children = @forks.size + @expected_children = [@intended_children, @forks.size].max refresh_readiness end + # Count a child the configuration intends this boot to run. Called at + # the top of every fork_* method — before the fork can fail — and only + # pre-boot, so restart_child's re-forks never inflate the baseline. + def note_intended_child + @intended_children += 1 unless @booted + end + # Publish a fresh snapshot; the swapped-in Data is immutable, so the # health server's accept thread always reads a consistent state. def refresh_readiness @@ -286,6 +298,7 @@ def boot_processes end def fork_worker(worker_config, slot: nil) + note_intended_child queues = worker_config[:queues] || [config.default_queue] threads = worker_config[:threads] || 5 single_active = worker_config[:single_active_consumer] || false @@ -370,6 +383,7 @@ def register_fork_with_hub(pid, wake_writer, queues) end def fork_dispatcher + note_intended_child pid = fork do restore_signals setup_child_process @@ -397,6 +411,7 @@ def boot_scheduler end def fork_scheduler + note_intended_child pid = fork do restore_signals setup_child_process @@ -460,6 +475,7 @@ def boot_consumers end def fork_consumer(consumer_config, slot: nil) + note_intended_child # Array() so a consumer entry without :topics can't NoMethodError the # supervisor on the topics.join log lines below. topics = Array(consumer_config[:topics]) @@ -537,6 +553,7 @@ def boot_outbox_poller end def fork_outbox_poller + note_intended_child pid = fork do restore_signals setup_child_process diff --git a/spec/pgbus/configuration_spec.rb b/spec/pgbus/configuration_spec.rb index 54dcf455..e02e32b9 100644 --- a/spec/pgbus/configuration_spec.rb +++ b/spec/pgbus/configuration_spec.rb @@ -1086,6 +1086,21 @@ expect { config.validate! }.not_to raise_error end + it "rejects an infinite shutdown_timeout" do + config.shutdown_timeout = Float::INFINITY + expect { config.validate! }.to raise_error(Pgbus::ConfigurationError, /shutdown_timeout/) + end + + it "rejects a NaN shutdown_timeout" do + config.shutdown_timeout = Float::NAN + expect { config.validate! }.to raise_error(Pgbus::ConfigurationError, /shutdown_timeout/) + end + + it "rejects a non-real shutdown_timeout without crashing" do + config.shutdown_timeout = Complex(45, 1) + expect { config.validate! }.to raise_error(Pgbus::ConfigurationError, /shutdown_timeout/) + end + it "warns when an explicit shutdown_timeout is below drain_timeout" do allow(Pgbus.logger).to receive(:warn) config.drain_timeout = 60 diff --git a/spec/pgbus/health_probe_spec.rb b/spec/pgbus/health_probe_spec.rb index 30700824..e7b66ee2 100644 --- a/spec/pgbus/health_probe_spec.rb +++ b/spec/pgbus/health_probe_spec.rb @@ -89,6 +89,19 @@ def with_server(status) expect(run_probe(["--port", "banana"])).to eq(described_class::EXIT_USAGE) end + it "exits 2 on an out-of-range port instead of crashing in Socket.tcp" do + expect(run_probe(["--port", "0"])).to eq(described_class::EXIT_USAGE) + expect(run_probe(["--port", "70000"])).to eq(described_class::EXIT_USAGE) + end + + it "exits 2 on a non-numeric --timeout instead of silently probing with 0" do + expect(run_probe(["--port", "9394", "--timeout", "abc"])).to eq(described_class::EXIT_USAGE) + end + + it "exits 2 on a non-positive --timeout" do + expect(run_probe(["--port", "9394", "--timeout", "0"])).to eq(described_class::EXIT_USAGE) + end + # The whole point of the probe: a docker HEALTHCHECK runs it every few # seconds, so it must never drag in Bundler, Zeitwerk, or the pgbus gem. it "loads standalone without pulling in the gem" do diff --git a/spec/pgbus/process/supervisor_spec.rb b/spec/pgbus/process/supervisor_spec.rb index a94bbb2d..11f16d3f 100644 --- a/spec/pgbus/process/supervisor_spec.rb +++ b/spec/pgbus/process/supervisor_spec.rb @@ -966,6 +966,32 @@ def banner expect(supervisor.readiness_snapshot.status).to eq("DRAINING") end + it "fails the readiness gate when a boot-time fork failed (intended > live)" do + supervisor = described_class.new + allow(supervisor).to receive(:fork).and_return(nil) + + supervisor.send(:fork_worker, { queues: ["default"] }, slot: 0) + supervisor.send(:mark_booted) + snapshot = supervisor.readiness_snapshot + + expect(snapshot.expected).to eq(1) + expect(snapshot.live).to eq(0) + expect(snapshot.status).to eq("DEGRADED") + end + + it "does not inflate the baseline when a child is re-forked after boot" do + supervisor = described_class.new + allow(supervisor).to receive(:fork).and_return(7001) + + supervisor.send(:fork_worker, { queues: ["default"] }, slot: 0) + supervisor.send(:mark_booted) + allow(supervisor).to receive(:fork).and_return(7002) + supervisor.send(:fork_worker, { queues: ["default"] }, slot: 0) + supervisor.send(:refresh_readiness) + + expect(supervisor.readiness_snapshot.expected).to eq(1) + end + it "is marked booted by #run after boot_processes, before monitor_loop runs" do supervisor = described_class.new mock_client = build_mock_client