Class: Pgbus::Configuration

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/configuration.rb,
lib/pgbus/configuration/capsule_dsl.rb

Defined Under Namespace

Classes: CapsuleDSL

Constant Summary collapse

VALID_ROLES = 
%i[workers dispatcher scheduler consumers outbox].freeze
SHUTDOWN_TIMEOUT_MARGIN = 
5
VALID_LOG_FORMATS = 
%i[text json].freeze
VALID_BROADCAST_MODES = 
%i[ephemeral durable].freeze
VALID_GROUP_MODES = 
[nil, :fifo, :round_robin].freeze
VALID_CONNECTION_GUC_MODES = 
%i[options session].freeze
VALID_PGMQ_SCHEMA_MODES = 
%i[auto extension embedded].freeze
VALID_DOCTOR_ON_BOOT_MODES = 
[nil, :report, :strict].freeze
VALID_STREAMS_LISTEN_SCOPES = 
%i[master process].freeze
VALID_WORKER_NOTIFY_SCOPES = 
%i[supervisor fork].freeze
POOL_SIZE_WARN_THRESHOLD =

Returns the connection pool size to use for the PGMQ client.

If pool_size was explicitly set, returns that value unchanged. Otherwise auto-derives from the threads needed by the roles this process actually runs (respects Configuration#roles from --workers-only / --scheduler-only / --dispatcher-only):

workers role     → sum(workers.threads)
consumers role   → sum(event_consumers.threads)
dispatcher role  → +1
scheduler role   → +1

A --scheduler-only deployment that has 50 worker threads configured only needs 1 connection (for the scheduler), not 52.

Auto-tune protects users from the common pitfall of running 15 worker threads with a hand-set pool_size of 5 (resulting in ConnectionPool timeouts under load). Setting pool_size explicitly is still supported for advanced cases where you need a tighter or looser pool than the default formula provides.

NOTE: this covers the JOB pool only. On the dedicated-connection path (database_url / connection_params) every Client also opens a separate streams pool of streams_pool_size (default 5) for durable-stream publish + replay (issue #315). Both pools are lazy, so an idle process holds few real connections, but when budgeting Postgres/PgBouncer max_connections, size for +resolved_pool_size + streams_pool_size+ per process — every forked worker/dispatcher/scheduler/consumer builds its own streams pool even if it rarely touches streams.

50
ASYNC_POOL_CONNECTIONS =

Connections needed per async worker: one for the reactor's serial execution, one for polling, one for headroom. Fibers share connections because only one runs at a time per reactor thread.

3

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Configuration

Returns a new instance of Configuration.



227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
 
# File 'lib/pgbus/configuration.rb', line 227

def initialize
  @database_url = nil
  @connection_params = nil
  @pool_size = nil
  @pool_timeout = 5

  @default_queue = "default"
  @queue_prefix = "pgbus"

  @workers = [{ queues: %w[default], threads: 5 }]
  @roles = nil
  @polling_interval = 0.1
  @visibility_timeout = 30

  @prefetch_limit = nil
  @execution_mode = :threads

  @max_jobs_per_worker = nil
  @max_memory_mb = nil
  @max_worker_lifetime = nil

  @stall_threshold = 90
  @read_timeout = 30
  @drain_timeout = 30
  @shutdown_timeout = nil

  @dispatch_interval = 1.0

  @circuit_breaker_enabled = true

  @max_retries = 5
  @retry_backoff = 5          # seconds — first VT-retry delay
  @retry_backoff_max = 300    # 5 minutes cap
  @retry_backoff_jitter = 0.15

  @priority_levels = nil
  @default_priority = 1
  @group_mode = nil

  @archive_retention = 7 * 24 * 3600 # 7 days

  @outbox_enabled = false
  @outbox_poll_interval = 1.0
  @outbox_batch_size = 100
  @outbox_retention = 24 * 3600 # 1 day

  @idempotency_ttl = 7 * 24 * 3600 # 7 days
  @allowed_global_id_models = nil # nil = allow all (for backwards compat)

  @logger = (defined?(Rails) && Rails.respond_to?(:logger) && Rails.logger) || Logger.new($stdout)
  @log_format = :text
  @error_reporters = []

  @listen_notify = true

  @worker_notify_wakeup = nil
  @worker_notify_scope = :supervisor
  @streams_listen_scope = :master
  @worker_notify_host = nil
  @worker_notify_port = nil
  @worker_notify_database_url = nil

  @pgmq_schema_mode = :auto
  @doctor_on_boot = nil

  @event_consumers = nil

  @recurring_tasks = nil
  @recurring_schedule_interval = 1.0
  @recurring_tasks_file = nil
  @recurring_tasks_files = nil
  @recurring_enabled = true
  @recurring_execution_retention = 7 * 24 * 3600 # 7 days

  @zombie_detection = true

  @stats_enabled = true
  @stats_retention = 30 * 24 * 3600 # 30 days
  # StatBuffer flush thresholds. Buffered stats are lost on SIGKILL, so a
  # smaller size/interval shrinks the residual loss window at the cost of
  # more frequent bulk inserts. Defaults mirror StatBuffer's own constants.
  @stats_flush_size = StatBuffer::DEFAULT_FLUSH_SIZE
  @stats_flush_interval = StatBuffer::DEFAULT_FLUSH_INTERVAL

  @connects_to = nil
  @require_primary = false
  @connection_guc_mode = :options

  @web_auth = nil
  @web_refresh_interval = 5000
  @web_per_page = 25
  @web_live_updates = true
  @web_data_source = nil
  @insights_default_minutes = 60 # 1 hour
  @base_controller_class = "::ActionController::Base"
  @return_to_app_url = nil
  @metrics_enabled = true
  @web_filter_parameters = nil # nil = auto-detect from Rails, then fall back to defaults
  @web_filter_sensitive = true
  @deprecation_warned = Set.new

  # HTTP health endpoints — nil port disables the standalone server.
  @health_port = nil
  @health_bind = "127.0.0.1"

  @streams_enabled = true
  @streams_path = nil
  # streams_queue_prefix was removed in 1.0 (issue #335): it had been an
  # inert no-op since #308 (stream queues are named like job queues,
  # `#{queue_prefix}_<name>`, and identified via the pgbus_stream_queues
  # registry). Setting it now raises NoMethodError — remove it from your
  # initializer.
  # Streamer-only connection overrides. The Streamer's Listener owns a
  # dedicated long-lived `wait_for_notify` PG connection that can't go
  # through a PgBouncer in transaction mode (LISTEN/NOTIFY don't survive
  # transaction-pool COMMIT boundaries — see PlanetScale's docs). Setting
  # any of these overrides only the Streamer's connection options; the
  # worker, dispatcher, and client publish paths keep using the regular
  # `database_url` / `connection_params` (typically pooled).
  #
  #   streams_host          — override host only
  #   streams_port          — override port only (most common case:
  #                           pooler is on 6432, direct is 5432)
  #   streams_database_url  — full URL override; takes precedence over
  #                           the host/port surgicals when set
  @streams_host = nil
  @streams_port = nil
  @streams_database_url = nil
  # Dedicated DB connection pool for the streamer's publish + replay hot
  # paths (Client#send_stream_message and read_after/stream_current_msg_id).
  # Isolated from the job pool (`pool_size`) so a saturated worker pool
  # can't delay broadcast INSERTs and the dispatcher's per-wake read_after
  # reuses a persistent connection instead of a fresh PG.connect per call
  # (issue #315). A small fixed default — streams fan out from one
  # dispatcher thread per Puma worker, so a handful of connections is
  # plenty; raise it only for very high broadcast concurrency. Ignored on
  # the shared-ActiveRecord (Proc) connection path, where libpq isn't
  # thread-safe and streams keep using the single serialized connection.
  @streams_pool_size = 5
  @streams_pool_timeout = 5
  # Streams-POOL-only connection overrides (issue #358). The pool above
  # defaults to wherever the streamer connects (streams_host/port/
  # database_url), which is right for a separate streams database — but
  # wrong for the pooler-bypass pattern, where streams_port pins the
  # LISTEN connection to the direct Postgres port purely because LISTEN
  # dies in transaction pooling. The pool's traffic (broadcast INSERTs,
  # replay reads) is plain pooler-safe SQL, and on a direct port with a
  # low max_connections ceiling every process's streams_pool_size
  # connections eat scarce direct slots. Set any of these to route the
  # POOL independently of the LISTEN connection:
  #
  #   c.streams_port = 5432        # LISTEN bypasses the pooler
  #   c.streams_pool_port = 6432   # the pool stays on the pooler
  @streams_pool_host = nil
  @streams_pool_port = nil
  @streams_pool_database_url = nil
  # Self-tuning streams-pool autoscaling (issue #323). Opt-in, default off.
  # When true, a per-web-process control loop grows the dedicated streams
  # pool into a FAIR SHARE of live Postgres connection headroom under
  # saturation and shrinks it back to streams_pool_size (the baseline/floor)
  # when idle — and emergency-shrinks immediately if the DB runs critically
  # low on connections. No connection-count config: every threshold derives
  # from live max_connections. streams_pool_max is an OPTIONAL hard ceiling
  # (nil = the dynamic fair share is the effective cap).
  @streams_pool_autoscale = false
  @streams_pool_max = nil
  # Slow maintenance cadence (seconds) — the autoscaler runs as a periodic
  # check on the streamer's idle LISTEN connection (pghero-style), not a busy
  # loop. 5 minutes: the interval itself debounces, so there is no per-sample
  # hysteresis; a sustained burst converges over a few checks.
  @streams_pool_autoscale_interval = 300.0
  # Distinctive application_name prefix stamped on streams-pool connections
  # so the autoscaler can count peer processes (DISTINCT application_name
  # LIKE '<prefix>_%') from pg_stat_activity. Per-process suffix is appended
  # at build time. Namespace it per deployment if several pgbus fleets share
  # one database. NOTE: a transaction-mode PgBouncer strips application_name;
  # peer inference then degrades to "assume alone" (the safety margin +
  # emergency-shrink cushion this) — connect the streamer DIRECTLY for
  # accurate autoscaling.
  @streams_application_name = "pgbus_streams"
  @streams_signed_name_secret = nil
  @streams_default_retention = 5 * 60 # 5 minutes
  @streams_retention = {}
  @streams_heartbeat_interval = 15
  @streams_max_connections = 2_000
  @streams_idle_timeout = 3_600 # 1 hour
  # 250ms — this value plays two roles: (1) the TCP keepalive
  # interval for the streamer's PG LISTEN connection, and (2) the
  # upper bound on how long Dispatcher#handle_connect waits for
  # the Listener to acknowledge a synchronous ensure_listening
  # call. 5s was unbounded enough to drop messages on a
  # realistic subscribe burst; 250ms keeps the connect-path race
  # window tight while still leaving headroom over a typical
  # PG keepalive interval.
  @streams_listen_health_check_ms = 250
  @streams_write_deadline_ms = 5_000
  # Deadline (ms) for a socket write made INSIDE the dispatcher's
  # serial fanout loop (handle_durable_wake / handle_ephemeral_wake).
  # Kept far below streams_write_deadline_ms (5s) because these writes
  # run one after another on the single dispatcher thread: K
  # slow-but-not-yet-dead clients stack the stall, so the whole-thread
  # worst case is ~K * this_value before each is marked dead
  # (issue #315 item 3). A slow-but-alive client that can't absorb a
  # fanout frame in this window is marked dead and reconnects — its
  # EventSource Last-Event-ID then replays the missed frames from the
  # durable archive (or triggers a fresh re-render for ephemeral
  # streams, which have no archive). Connect-replay writes keep the
  # full streams_write_deadline_ms. Set this equal to
  # streams_write_deadline_ms to restore the pre-#315 timing.
  @streams_fanout_write_deadline_ms = 250
  # Optional cap on dispatch-queue depth checked by the Listener before
  # it pushes a durable WakeMessage. 0 (default) = unbounded — today's
  # behavior. A positive value makes the Listener DROP a durable wake
  # when the queue is at/over the cap; this is safe because the next
  # durable wake for that stream re-reads from the min cursor. It
  # bounds DISTINCT-STREAM backlog only — a single hot stream's wakes
  # coalesce (drain_wakes_for) so its depth stays low regardless.
  # Ephemeral wakes (which carry the only copy of their HTML) and
  # Connect/Disconnect messages are NEVER dropped (issue #315 item 3).
  @streams_dispatch_queue_limit = 0
  # Number of writer threads that flush durable-stream fanout socket writes
  # OFF the single dispatcher thread (issue #321). 0 (default) = inline —
  # byte-for-byte the pre-#321 behavior, no pump, no extra allocation. A
  # positive value spawns N writer threads; each durable broadcast fanout
  # write is handed to a writer (partitioned by connection.id.hash % N, so
  # a connection's frames stay ordered and its per-io mutex is respected),
  # freeing the dispatcher to service the next wake/connect. Fast clients
  # then stop waiting behind slow ones. EPHEMERAL fanout and connect-replay
  # writes always stay inline regardless of this value: ephemerals have no
  # archive to replay, so they must not risk an async drop (see #321 B1).
  @streams_writer_threads = 0
  # Per-connection cap on the writer's outbound buffer when
  # streams_writer_threads > 0. 0 (default) = unbounded. A positive value
  # drops the OLDEST buffered durable frame for a connection whose writer
  # can't keep up — safe because durable frames live in the a_<stream>
  # archive and are re-read on reconnect via Last-Event-ID; it's a memory
  # guard against a pathologically slow-but-alive client, not a delivery
  # guarantee. Ignored when streams_writer_threads == 0 (issue #321).
  @streams_writer_buffer_limit = 0
  # EXPERIMENTAL — exempt from the 1.0 stability promise.
  @streams_falcon_streaming_body = false
  # Opt-in: when true, the Dispatcher writes one row to
  # pgbus_stream_stats per broadcast/connect/disconnect. Default
  # off because stream event volume can be much higher than job
  # volume and the Insights surface is only useful if operators
  # actually look at it. Separate from #stats_enabled (which
  # gates pgbus_job_stats recording) on purpose — operators
  # usually want job stats on and stream stats off, or vice versa.
  @streams_stats_enabled = false
  @streams_test_mode = false
  @streams_default_broadcast_mode = :ephemeral
  @streams_orphan_sweep_interval = 3600    # 1 hour
  @streams_orphan_threshold = 86_400       # 24 hours
  @streams_durable_patterns = []
  # Dedicated queue for turbo-rails' async broadcast jobs
  # (Turbo::Streams::ActionBroadcastJob and siblings). nil (default) leaves
  # them on the default queue, where a `broadcasts_to`/`broadcasts_refreshes`
  # render+broadcast can wait behind long-running jobs before the browser
  # sees the update. Set to a queue name (e.g. "realtime") and back it with
  # a dedicated worker capsule to isolate broadcast latency from job
  # throughput. Applied at engine boot when turbo-rails is loaded. See #311.
  @streams_broadcast_queue = nil
  # EXPERIMENTAL (both presence settings) — exempt from the 1.0 stability
  # promise. Streams matching these patterns get connection-driven presence:
  # auto-join on SSE connect, auto-leave on disconnect, touch on the
  # keepalive heartbeat (issue #169). Empty by default (opt-in).
  @streams_presence_patterns = []
  # Extracts a presence member { id:, metadata: } from a connection's
  # authorize-hook context. Defaults to nil, which uses the built-in
  # extractor (see #presence_member_for): a Hash with :member_id/:id,
  # or an object responding to #id.
  @streams_presence_member = nil

  # AppSignal: auto-on when the appsignal gem is loaded; probe runs in
  # the same process, so the operator can disable it independently.
  @appsignal_enabled = true
  @appsignal_probe_enabled = true

  # Generic metrics adapter: off by default (no subscription installed).
  @metrics_backend = nil
  @statsd_host = "127.0.0.1"
  @statsd_port = 8125

  @eager_validation = true
end

Instance Attribute Details

#allowed_global_id_models ⇒ Object

GlobalID allowlist for job arguments (_aj_globalid) and EventBus payloads (_global_id). nil = allow-all; [] = deny-all; Array of Class/Module = only those models may resolve. Enforced in Serializer (issue #368 for the job path).



86
87
88
 
# File 'lib/pgbus/configuration.rb', line 86

def allowed_global_id_models
  @allowed_global_id_models
end

#appsignal_enabled ⇒ Object

AppSignal integration (auto-loaded when ::Appsignal is defined and this is true). Set to false to opt out without uninstalling the appsignal gem.



210
211
212
 
# File 'lib/pgbus/configuration.rb', line 210

def appsignal_enabled
  @appsignal_enabled
end

#appsignal_probe_enabled ⇒ Object

AppSignal integration (auto-loaded when ::Appsignal is defined and this is true). Set to false to opt out without uninstalling the appsignal gem.



210
211
212
 
# File 'lib/pgbus/configuration.rb', line 210

def appsignal_probe_enabled
  @appsignal_probe_enabled
end

#archive_retention ⇒ Object

Archive compaction. Only the user-facing retention window is configurable; the loop interval and batch size are tuned via constants on Pgbus::Process::Dispatcher.



76
77
78
 
# File 'lib/pgbus/configuration.rb', line 76

def archive_retention
  @archive_retention
end

#base_controller_class ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def base_controller_class
  @base_controller_class
end

#circuit_breaker_enabled ⇒ Object

Circuit breaker. Only enabled is user-facing — the trip threshold and backoff curve are tuned via constants on Pgbus::CircuitBreaker because they are implementation details that have never been worth exposing.



54
55
56
 
# File 'lib/pgbus/configuration.rb', line 54

def circuit_breaker_enabled
  @circuit_breaker_enabled
end

#connection_guc_mode ⇒ Object

How pgbus forwards database.yml GUCs (variables: such as client_min_messages, and the read-timeout statement_timeout) onto its dedicated pgmq connections:

:options (default) — bake them into the libpq `options` STARTUP param
                   (-c key=value). Works everywhere EXCEPT a
                   transaction-mode PgBouncer, which rejects the
                   `options` startup param with a FATAL.
:session           — apply them via post-connect `SET` statements on a
                   fresh connection instead, so a transaction-mode
                   pooler accepts them. Dedicated-connection path only.

See issue #332.



151
152
153
 
# File 'lib/pgbus/configuration.rb', line 151

def connection_guc_mode
  @connection_guc_mode
end

#connection_params ⇒ Object

Connection settings



8
9
10
 
# File 'lib/pgbus/configuration.rb', line 8

def connection_params
  @connection_params
end

#connects_to ⇒ Object

Multi-database support (optional separate database for pgbus tables) Set to { database: { writing: :pgbus, reading: :pgbus } } to use a separate database. Requires a matching entry in config/database.yml under the "pgbus" key.



130
131
132
 
# File 'lib/pgbus/configuration.rb', line 130

def connects_to
  @connects_to
end

#database_url ⇒ Object

Connection settings



8
9
10
 
# File 'lib/pgbus/configuration.rb', line 8

def database_url
  @database_url
end

#default_priority ⇒ Object

Priority queues



64
65
66
 
# File 'lib/pgbus/configuration.rb', line 64

def default_priority
  @default_priority
end

#default_queue ⇒ Object

Queue settings



11
12
13
 
# File 'lib/pgbus/configuration.rb', line 11

def default_queue
  @default_queue
end

#dispatch_interval ⇒ Object

Dispatcher settings



49
50
51
 
# File 'lib/pgbus/configuration.rb', line 49

def dispatch_interval
  @dispatch_interval
end

#doctor_on_boot ⇒ Object

Run doctor preflight checks inside the supervisor at boot (issue #347), so an entrypoint need not boot Rails twice (once for pgbus doctor, once for pgbus start). nil/false (default) — off, byte-identical to today. :report — run the preflight and log the report, always continue booting. :strict — additionally REFUSE to boot (raise before forking any worker) when a genuinely-fatal, non-transient check fails. The strict set is deliberately narrow (Configuration, PGMQ schema) so a transient boot-time DB blip — which the lenient queue bootstrap is designed to ride out via child crash-and-backoff — never aborts a whole fleet's cold boot in lockstep. Set via the --doctor / --doctor-strict start flags too.



117
118
119
 
# File 'lib/pgbus/configuration.rb', line 117

def doctor_on_boot
  @doctor_on_boot
end

#drain_timeout ⇒ Object

Liveness probe: supervisor kills a worker whose claim loop has not advanced for longer than stall_threshold seconds (default 90). read_timeout caps how long a single PGMQ read can block (default 30s), so a dead socket raises instead of parking the loop forever. drain_timeout bounds the graceful-shutdown drain phase (default 30s): a job still running after this is abandoned to shutdown's own termination wait, so recycling/deploy never wedges on a permanently-stuck job.



36
37
38
 
# File 'lib/pgbus/configuration.rb', line 36

def drain_timeout
  @drain_timeout
end

#eager_validation ⇒ Object

When true (default), Pgbus.configure runs validate! after the block yields, so an invalid value fails loud at boot instead of dormant until a worker path consumes it. Set false to opt out (exotic setups that intentionally hold a transiently-invalid config); explicit validate! still works.



225
226
227
 
# File 'lib/pgbus/configuration.rb', line 225

def eager_validation
  @eager_validation
end

#error_reporters ⇒ Object

Error reporting — array of callable objects invoked on caught exceptions. Each receives (exception, context_hash) or (exception, context_hash, config).



96
97
98
 
# File 'lib/pgbus/configuration.rb', line 96

def error_reporters
  @error_reporters
end

#event_consumers ⇒ Object

Event consumers



120
121
122
 
# File 'lib/pgbus/configuration.rb', line 120

def event_consumers
  @event_consumers
end

#execution_mode ⇒ Object

Worker settings



14
15
16
 
# File 'lib/pgbus/configuration.rb', line 14

def execution_mode
  @execution_mode
end

#group_mode ⇒ Object

Grouped reads (PGMQ v1.11.0+ FIFO grouping). EXPERIMENTAL — exempt from the 1.0 stability promise; the shape may change in a minor release. nil = disabled (default read_batch behavior). :fifo = use read_grouped (drains oldest group first, throughput-optimized). :round_robin = use read_grouped_rr (fair round-robin across groups).



71
72
73
 
# File 'lib/pgbus/configuration.rb', line 71

def group_mode
  @group_mode
end

#health_bind ⇒ Object

HTTP health endpoints (liveness/readiness for orchestrators like Kubernetes). health_port nil (default) leaves the standalone server in the supervisor disabled — mount Pgbus::Web::HealthApp in the host Rails app instead. When set, Supervisor#run serves /livez and /readyz over a plain TCPServer bound to health_bind (default localhost).



172
173
174
 
# File 'lib/pgbus/configuration.rb', line 172

def health_bind
  @health_bind
end

#health_port ⇒ Object

HTTP health endpoints (liveness/readiness for orchestrators like Kubernetes). health_port nil (default) leaves the standalone server in the supervisor disabled — mount Pgbus::Web::HealthApp in the host Rails app instead. When set, Supervisor#run serves /livez and /readyz over a plain TCPServer bound to health_bind (default localhost).



172
173
174
 
# File 'lib/pgbus/configuration.rb', line 172

def health_port
  @health_port
end

#idempotency_ttl ⇒ Object

Event bus



88
89
90
 
# File 'lib/pgbus/configuration.rb', line 88

def idempotency_ttl
  @idempotency_ttl
end

#insights_default_minutes ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def insights_default_minutes
  @insights_default_minutes
end

#listen_notify ⇒ Object

LISTEN/NOTIFY. Only the on/off switch is user-facing — the throttle interval is a Postgres-side tuning knob that lives as a constant on Pgbus::Client (NOTIFY_THROTTLE_MS).



101
102
103
 
# File 'lib/pgbus/configuration.rb', line 101

def listen_notify
  @listen_notify
end

#log_format ⇒ Object

rubocop:disable Style/AccessorGrouping



92
93
94
 
# File 'lib/pgbus/configuration.rb', line 92

def log_format
  @log_format
end

#logger ⇒ Object

Logging



91
92
93
 
# File 'lib/pgbus/configuration.rb', line 91

def logger
  @logger
end

#max_jobs_per_worker ⇒ Object

Worker recycling



27
28
29
 
# File 'lib/pgbus/configuration.rb', line 27

def max_jobs_per_worker
  @max_jobs_per_worker
end

#max_memory_mb ⇒ Object

Worker recycling



27
28
29
 
# File 'lib/pgbus/configuration.rb', line 27

def max_memory_mb
  @max_memory_mb
end

#max_retries ⇒ Object

Dead letter queue



57
58
59
 
# File 'lib/pgbus/configuration.rb', line 57

def max_retries
  @max_retries
end

#max_worker_lifetime ⇒ Object

Worker recycling



27
28
29
 
# File 'lib/pgbus/configuration.rb', line 27

def max_worker_lifetime
  @max_worker_lifetime
end

#metrics_backend ⇒ Object

Generic metrics adapter (Pgbus::Metrics). Consumes the same pgbus.* instrumentation events as AppSignal and forwards them to a backend:

nil (default) — install nothing, zero overhead
:prometheus   — in-process registry, scraped via PrometheusExporter
:statsd       — UDP datagrams to statsd_host:statsd_port
a Pgbus::Metrics::Backend instance — a custom backend


218
219
220
 
# File 'lib/pgbus/configuration.rb', line 218

def metrics_backend
  @metrics_backend
end

#metrics_enabled ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def metrics_enabled
  @metrics_enabled
end

#outbox_batch_size ⇒ Object

Transactional outbox



79
80
81
 
# File 'lib/pgbus/configuration.rb', line 79

def outbox_batch_size
  @outbox_batch_size
end

#outbox_enabled ⇒ Object

Transactional outbox



79
80
81
 
# File 'lib/pgbus/configuration.rb', line 79

def outbox_enabled
  @outbox_enabled
end

#outbox_poll_interval ⇒ Object

Transactional outbox



79
80
81
 
# File 'lib/pgbus/configuration.rb', line 79

def outbox_poll_interval
  @outbox_poll_interval
end

#outbox_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



80
81
82
 
# File 'lib/pgbus/configuration.rb', line 80

def outbox_retention
  @outbox_retention
end

#pgmq_schema_mode ⇒ Object

PGMQ schema installation mode (:auto, :extension, :embedded)



104
105
106
 
# File 'lib/pgbus/configuration.rb', line 104

def pgmq_schema_mode
  @pgmq_schema_mode
end

#polling_interval ⇒ Object

Worker settings



14
15
16
 
# File 'lib/pgbus/configuration.rb', line 14

def polling_interval
  @polling_interval
end

#pool_size ⇒ Object

Connection settings



8
9
10
 
# File 'lib/pgbus/configuration.rb', line 8

def pool_size
  @pool_size
end

#pool_timeout ⇒ Object

Connection settings



8
9
10
 
# File 'lib/pgbus/configuration.rb', line 8

def pool_timeout
  @pool_timeout
end

#prefetch_limit ⇒ Object

Worker settings



14
15
16
 
# File 'lib/pgbus/configuration.rb', line 14

def prefetch_limit
  @prefetch_limit
end

#priority_levels ⇒ Object

Priority queues



64
65
66
 
# File 'lib/pgbus/configuration.rb', line 64

def priority_levels
  @priority_levels
end

#queue_prefix ⇒ Object

Queue settings



11
12
13
 
# File 'lib/pgbus/configuration.rb', line 11

def queue_prefix
  @queue_prefix
end

#read_timeout ⇒ Object

Liveness probe: supervisor kills a worker whose claim loop has not advanced for longer than stall_threshold seconds (default 90). read_timeout caps how long a single PGMQ read can block (default 30s), so a dead socket raises instead of parking the loop forever. drain_timeout bounds the graceful-shutdown drain phase (default 30s): a job still running after this is abandoned to shutdown's own termination wait, so recycling/deploy never wedges on a permanently-stuck job.



36
37
38
 
# File 'lib/pgbus/configuration.rb', line 36

def read_timeout
  @read_timeout
end

#recurring_enabled ⇒ Object

Recurring jobs



123
124
125
 
# File 'lib/pgbus/configuration.rb', line 123

def recurring_enabled
  @recurring_enabled
end

#recurring_execution_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



125
126
127
 
# File 'lib/pgbus/configuration.rb', line 125

def recurring_execution_retention
  @recurring_execution_retention
end

#recurring_schedule_interval ⇒ Object

Recurring jobs



123
124
125
 
# File 'lib/pgbus/configuration.rb', line 123

def recurring_schedule_interval
  @recurring_schedule_interval
end

#recurring_tasks ⇒ Object

Recurring jobs



123
124
125
 
# File 'lib/pgbus/configuration.rb', line 123

def recurring_tasks
  @recurring_tasks
end

#recurring_tasks_file ⇒ Object

Recurring jobs



123
124
125
 
# File 'lib/pgbus/configuration.rb', line 123

def recurring_tasks_file
  @recurring_tasks_file
end

#recurring_tasks_files ⇒ Object

recurring_tasks_file (singular) is deprecated in favor of the plural recurring_tasks_files. The engine still writes the singular internally to record which default file was loaded, so it stays a plain accessor; the deprecation warning fires only when a user has set BOTH — the case where the singular is silently ignored — pointing them at the plural.



1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
 
# File 'lib/pgbus/configuration.rb', line 1194

def recurring_tasks_files
  if @recurring_tasks_files
    if @recurring_tasks_file
      warn_deprecated_config(
        :recurring_tasks_file,
        "recurring_tasks_files is set, so recurring_tasks_file is ignored — " \
        "consolidate onto recurring_tasks_files (plural)"
      )
    end
    return @recurring_tasks_files
  end

  @recurring_tasks_file ? [@recurring_tasks_file] : nil
end

#require_primary ⇒ Object

Reject a job-pool connection that landed on a read-only replica (pg_is_in_recovery() => t). Default false — a correctly-configured single-primary deployment sees no change. Set true when a read/write splitting pooler (pgdog/pgcat) could route pgmq's VOLATILE read/archive to a replica, which otherwise makes workers silently read nothing. See issue #332.



138
139
140
 
# File 'lib/pgbus/configuration.rb', line 138

def require_primary
  @require_primary
end

#retry_backoff ⇒ Object

Retry backoff for the VT-based retry path (unhandled exceptions). Jobs can override these per-class via Pgbus::RetryBackoff::JobMixin.



61
62
63
 
# File 'lib/pgbus/configuration.rb', line 61

def retry_backoff
  @retry_backoff
end

#retry_backoff_jitter ⇒ Object

Retry backoff for the VT-based retry path (unhandled exceptions). Jobs can override these per-class via Pgbus::RetryBackoff::JobMixin.



61
62
63
 
# File 'lib/pgbus/configuration.rb', line 61

def retry_backoff_jitter
  @retry_backoff_jitter
end

#retry_backoff_max ⇒ Object

Retry backoff for the VT-based retry path (unhandled exceptions). Jobs can override these per-class via Pgbus::RetryBackoff::JobMixin.



61
62
63
 
# File 'lib/pgbus/configuration.rb', line 61

def retry_backoff_max
  @retry_backoff_max
end

#return_to_app_url ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def return_to_app_url
  @return_to_app_url
end

#roles ⇒ Object

Supervisor role selection. nil = boot all roles (default behavior). Array of role symbols = boot only the listed roles. Set via the CLI flags --workers-only / --scheduler-only / --dispatcher-only, or directly in an initializer for advanced cases.



22
23
24
 
# File 'lib/pgbus/configuration.rb', line 22

def roles
  @roles
end

#shutdown_timeout ⇒ Object

Resolved supervisor SIGKILL deadline: the explicit value when set, otherwise drain_timeout + SHUTDOWN_TIMEOUT_MARGIN (see attr_writer docs).



1279
1280
1281
 
# File 'lib/pgbus/configuration.rb', line 1279

def shutdown_timeout
  @shutdown_timeout || (drain_timeout + SHUTDOWN_TIMEOUT_MARGIN)
end

#stall_threshold ⇒ Object

Liveness probe: supervisor kills a worker whose claim loop has not advanced for longer than stall_threshold seconds (default 90). read_timeout caps how long a single PGMQ read can block (default 30s), so a dead socket raises instead of parking the loop forever. drain_timeout bounds the graceful-shutdown drain phase (default 30s): a job still running after this is abandoned to shutdown's own termination wait, so recycling/deploy never wedges on a permanently-stuck job.



36
37
38
 
# File 'lib/pgbus/configuration.rb', line 36

def stall_threshold
  @stall_threshold
end

#stats_enabled ⇒ Object

Job stats



158
159
160
 
# File 'lib/pgbus/configuration.rb', line 158

def stats_enabled
  @stats_enabled
end

#stats_flush_interval ⇒ Object

Job stats



158
159
160
 
# File 'lib/pgbus/configuration.rb', line 158

def stats_flush_interval
  @stats_flush_interval
end

#stats_flush_size ⇒ Object

Job stats



158
159
160
 
# File 'lib/pgbus/configuration.rb', line 158

def stats_flush_size
  @stats_flush_size
end

#stats_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



159
160
161
 
# File 'lib/pgbus/configuration.rb', line 159

def stats_retention
  @stats_retention
end

#statsd_host ⇒ Object

Generic metrics adapter (Pgbus::Metrics). Consumes the same pgbus.* instrumentation events as AppSignal and forwards them to a backend:

nil (default) — install nothing, zero overhead
:prometheus   — in-process registry, scraped via PrometheusExporter
:statsd       — UDP datagrams to statsd_host:statsd_port
a Pgbus::Metrics::Backend instance — a custom backend


218
219
220
 
# File 'lib/pgbus/configuration.rb', line 218

def statsd_host
  @statsd_host
end

#statsd_port ⇒ Object

Generic metrics adapter (Pgbus::Metrics). Consumes the same pgbus.* instrumentation events as AppSignal and forwards them to a backend:

nil (default) — install nothing, zero overhead
:prometheus   — in-process registry, scraped via PrometheusExporter
:statsd       — UDP datagrams to statsd_host:statsd_port
a Pgbus::Metrics::Backend instance — a custom backend


218
219
220
 
# File 'lib/pgbus/configuration.rb', line 218

def statsd_port
  @statsd_port
end

#streams_application_name ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_application_name
  @streams_application_name
end

#streams_broadcast_queue ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_broadcast_queue
  @streams_broadcast_queue
end

#streams_database_url ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_database_url
  @streams_database_url
end

#streams_default_broadcast_mode ⇒ Object

rubocop:disable Style/AccessorGrouping



190
191
192
 
# File 'lib/pgbus/configuration.rb', line 190

def streams_default_broadcast_mode
  @streams_default_broadcast_mode
end

#streams_default_retention ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_default_retention
  @streams_default_retention
end

#streams_dispatch_queue_limit ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_dispatch_queue_limit
  @streams_dispatch_queue_limit
end

#streams_durable_patterns ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_durable_patterns
  @streams_durable_patterns
end

#streams_enabled ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_enabled
  @streams_enabled
end

#streams_falcon_streaming_body ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_falcon_streaming_body
  @streams_falcon_streaming_body
end

#streams_fanout_write_deadline_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_fanout_write_deadline_ms
  @streams_fanout_write_deadline_ms
end

#streams_heartbeat_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_heartbeat_interval
  @streams_heartbeat_interval
end

#streams_host ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_host
  @streams_host
end

#streams_idle_timeout ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_idle_timeout
  @streams_idle_timeout
end

#streams_listen_health_check_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_listen_health_check_ms
  @streams_listen_health_check_ms
end

#streams_listen_scope ⇒ Object

Where the streams LISTEN connection lives (issue #382):

:master (default) — ONE shared listener in the preforking web master
(MasterHub); workers connect lazily over a Unix socket and fall back
to a per-worker listener whenever the hub is absent or dies.
:process — one listener per web process: the pre-0.13 behavior, and
the automatic behavior on single-mode / non-preforking servers.


653
654
655
 
# File 'lib/pgbus/configuration.rb', line 653

def streams_listen_scope
  @streams_listen_scope
end

#streams_max_connections ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_max_connections
  @streams_max_connections
end

#streams_orphan_sweep_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_orphan_sweep_interval
  @streams_orphan_sweep_interval
end

#streams_orphan_threshold ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_orphan_threshold
  @streams_orphan_threshold
end

#streams_path ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_path
  @streams_path
end

#streams_pool_autoscale ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_pool_autoscale
  @streams_pool_autoscale
end

#streams_pool_autoscale_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_pool_autoscale_interval
  @streams_pool_autoscale_interval
end

#streams_pool_database_url ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_pool_database_url
  @streams_pool_database_url
end

#streams_pool_host ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_pool_host
  @streams_pool_host
end

#streams_pool_max ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_pool_max
  @streams_pool_max
end

#streams_pool_port ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_pool_port
  @streams_pool_port
end

#streams_pool_size ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_pool_size
  @streams_pool_size
end

#streams_pool_timeout ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_pool_timeout
  @streams_pool_timeout
end

#streams_port ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_port
  @streams_port
end

#streams_presence_member ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_presence_member
  @streams_presence_member
end

#streams_presence_patterns ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_presence_patterns
  @streams_presence_patterns
end

#streams_retention ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_retention
  @streams_retention
end

#streams_signed_name_secret ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_signed_name_secret
  @streams_signed_name_secret
end

#streams_stats_enabled ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_stats_enabled
  @streams_stats_enabled
end

#streams_test_mode ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_test_mode
  @streams_test_mode
end

#streams_write_deadline_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_write_deadline_ms
  @streams_write_deadline_ms
end

#streams_writer_buffer_limit ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_writer_buffer_limit
  @streams_writer_buffer_limit
end

#streams_writer_threads ⇒ Object

Streams (turbo-rails replacement, SSE-based)



175
176
177
 
# File 'lib/pgbus/configuration.rb', line 175

def streams_writer_threads
  @streams_writer_threads
end

#visibility_timeout ⇒ Object

rubocop:disable Style/AccessorGrouping



15
16
17
 
# File 'lib/pgbus/configuration.rb', line 15

def visibility_timeout
  @visibility_timeout
end

#web_auth ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def web_auth
  @web_auth
end

#web_data_source ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def web_data_source
  @web_data_source
end

#web_filter_parameters ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def web_filter_parameters
  @web_filter_parameters
end

#web_filter_sensitive ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def web_filter_sensitive
  @web_filter_sensitive
end

#web_live_updates ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def web_live_updates
  @web_live_updates
end

#web_per_page ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def web_per_page
  @web_per_page
end

#web_refresh_interval ⇒ Object

Web dashboard



162
163
164
 
# File 'lib/pgbus/configuration.rb', line 162

def web_refresh_interval
  @web_refresh_interval
end

#worker_notify_database_url ⇒ Object

NOTIFY-gated worker wakeups. When true, workers are woken by a NotifyListener PG connection that LISTENs on their queues' INSERT channels instead of blind-polling. Defaults to the value of listen_notify. The worker_notify_* overrides mirror streams_* so the LISTEN connection can bypass PgBouncer.



197
198
199
 
# File 'lib/pgbus/configuration.rb', line 197

def worker_notify_database_url
  @worker_notify_database_url
end

#worker_notify_host ⇒ Object

NOTIFY-gated worker wakeups. When true, workers are woken by a NotifyListener PG connection that LISTENs on their queues' INSERT channels instead of blind-polling. Defaults to the value of listen_notify. The worker_notify_* overrides mirror streams_* so the LISTEN connection can bypass PgBouncer.



197
198
199
 
# File 'lib/pgbus/configuration.rb', line 197

def worker_notify_host
  @worker_notify_host
end

#worker_notify_port ⇒ Object

NOTIFY-gated worker wakeups. When true, workers are woken by a NotifyListener PG connection that LISTENs on their queues' INSERT channels instead of blind-polling. Defaults to the value of listen_notify. The worker_notify_* overrides mirror streams_* so the LISTEN connection can bypass PgBouncer.



197
198
199
 
# File 'lib/pgbus/configuration.rb', line 197

def worker_notify_port
  @worker_notify_port
end

#worker_notify_scope ⇒ Object

Where the LISTEN connection lives (issue #381):

:supervisor (default) — ONE shared NotifyListener in the supervisor
process for the whole host; forks are woken over per-fork pipes.
Direct-connection footprint: 1, regardless of capsule/consumer count.
:fork — one NotifyListener (and one dedicated PG connection) per
worker/consumer fork: the pre-0.13 behavior, kept as an escape hatch.


206
207
208
 
# File 'lib/pgbus/configuration.rb', line 206

def worker_notify_scope
  @worker_notify_scope
end

#worker_notify_wakeup ⇒ Object

NOTIFY-gated worker wakeups. When true, workers are woken by a NotifyListener PG connection that LISTENs on their queues' INSERT channels instead of blind-polling. Defaults to the value of listen_notify. The worker_notify_* overrides mirror streams_* so the LISTEN connection can bypass PgBouncer.



197
198
199
 
# File 'lib/pgbus/configuration.rb', line 197

def worker_notify_wakeup
  @worker_notify_wakeup
end

#workers ⇒ Object

rubocop:disable Style/AccessorGrouping



15
16
17
 
# File 'lib/pgbus/configuration.rb', line 15

def workers
  @workers
end

#zombie_detection ⇒ Object

Zombie message detection — logs a warning when a message is redelivered (read_ct > 1) without any prior failure recorded in pgbus_failed_events.



155
156
157
 
# File 'lib/pgbus/configuration.rb', line 155

def zombie_detection
  @zombie_detection
end

Instance Method Details

#capsule(name, queues:, threads:) ⇒ Object

Define a named capsule and append it to the workers list.

c.capsule :critical, queues: %w[critical], threads: 5
c.capsule :gated, queues: %w[gated], threads: 1, single_active_consumer: true

Names must be unique. Queues must not overlap with capsules already defined (would cause double-processing). Composes with the string DSL — c.workers "..." followed by c.capsule :name, ... appends the named capsule to the list parsed from the string.

Raises:



1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
 
# File 'lib/pgbus/configuration.rb', line 1093

def capsule(name, queues:, threads:, **)
  raise Pgbus::ConfigurationError, "capsule queues must be a non-empty Array" unless queues.is_a?(Array) && queues.any?
  raise Pgbus::ConfigurationError, "capsule threads must be a positive Integer" unless threads.is_a?(Integer) && threads.positive?

  normalized_name = name.to_s
  @workers ||= []

  if @workers.any? { |c| capsule_name(c) == normalized_name }
    raise Pgbus::ConfigurationError, "capsule #{name.inspect} is already defined"
  end

  validate_no_queue_overlap!(queues)

  @workers << { name: normalized_name, queues: queues, threads: threads, ** }
end

#capsule_named(name) ⇒ Object

Look up a capsule by its name. Accepts symbol or string. Returns the matching Hash, or nil. Used by the CLI's --capsule selector.



1111
1112
1113
1114
1115
1116
 
# File 'lib/pgbus/configuration.rb', line 1111

def capsule_named(name)
  return nil unless @workers

  key = name.to_s
  @workers.find { |c| capsule_name(c) == key }
end

#connection_options ⇒ Object 



1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
 
# File 'lib/pgbus/configuration.rb', line 1296

def connection_options
  if database_url
    database_url
  elsif connection_params
    # An explicit connection_params Hash may carry a database.yml-style
    # `:variables` block; forward it the same way the AR-extracted path does
    # so client_min_messages etc. actually reach pgmq's connections and a
    # non-libpq :variables key never reaches PG.connect (issue #332). A
    # non-Hash connection_params (e.g. a Proc returning a raw connection)
    # passes through untouched.
    if connection_params.is_a?(Hash)
      forward_connection_variables(connection_params.except(:variables), connection_params[:variables])
    else
      connection_params
    end
  elsif defined?(ActiveRecord::Base)
    # Extract connection config from ActiveRecord so pgmq-ruby creates its
    # own dedicated PG connections. Sharing AR's raw_connection via a Proc
    # is NOT thread-safe: the ConnectionPool caches the PG::Connection from
    # whichever thread first called the Proc, then hands that same object to
    # other threads — causing nil results, segfaults, and data corruption
    # when AR and PGMQ issue concurrent queries on the same connection.
    extract_ar_connection_hash
  else
    raise ConfigurationError, "No database connection configured. " \
                              "Set Pgbus.configuration.database_url, connection_params, or use with Rails."
  end
end

#dashboard_filter_parameters ⇒ Object

dashboard_filter_* unify onto the incumbent web_ prefix.



1223
1224
1225
 
# File 'lib/pgbus/configuration.rb', line 1223

def dashboard_filter_parameters
  web_filter_parameters
end

#dashboard_filter_parameters=(value) ⇒ Object 



1227
1228
1229
1230
 
# File 'lib/pgbus/configuration.rb', line 1227

def dashboard_filter_parameters=(value)
  warn_deprecated_config(:dashboard_filter_parameters, "use web_filter_parameters instead")
  self.web_filter_parameters = value
end

#dashboard_filter_sensitive ⇒ Object 



1232
1233
1234
 
# File 'lib/pgbus/configuration.rb', line 1232

def dashboard_filter_sensitive
  web_filter_sensitive
end

#dashboard_filter_sensitive=(value) ⇒ Object 



1236
1237
1238
1239
 
# File 'lib/pgbus/configuration.rb', line 1236

def dashboard_filter_sensitive=(value)
  warn_deprecated_config(:dashboard_filter_sensitive, "use web_filter_sensitive instead")
  self.web_filter_sensitive = value
end

#dead_letter_queue_name(name) ⇒ Object 



518
519
520
 
# File 'lib/pgbus/configuration.rb', line 518

def dead_letter_queue_name(name)
  "#{queue_name(name)}#{Pgbus::DEAD_LETTER_SUFFIX}"
end

#execution_mode_for(worker_config) ⇒ Object

Returns the execution mode for a specific worker config hash, falling back to the global execution_mode setting.



534
535
536
537
 
# File 'lib/pgbus/configuration.rb', line 534

def execution_mode_for(worker_config)
  mode = worker_config[:execution_mode] || execution_mode
  ExecutionPools.normalize_mode(mode)
end

#presence_member_for(context) ⇒ Object

Derives a presence member { id:, metadata: } from a connection's authorize-hook context, or nil when no member can be derived (an anonymous connection — presence is simply skipped). Uses streams_presence_member when configured; otherwise the built-in extractor handles the common shapes:

- Hash with :member_id (or :id) and optional :metadata
- an object responding to #id (e.g. a User model)

The id is always coerced to a String and metadata defaults to {}.



999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
 
# File 'lib/pgbus/configuration.rb', line 999

def presence_member_for(context)
  return nil if context.nil?

  raw = streams_presence_member ? streams_presence_member.call(context) : default_presence_member(context)
  return nil unless raw.is_a?(Hash)

  id = raw[:id]
  return nil if id.nil? || id.to_s.empty?

  { id: id.to_s, metadata: raw[:metadata] || {} }
end

#priority_queue_name(name, priority) ⇒ Object 



522
523
524
 
# File 'lib/pgbus/configuration.rb', line 522

def priority_queue_name(name, priority)
  "#{queue_name(name)}_p#{priority}"
end

#priority_queue_names(name) ⇒ Object 



526
527
528
529
530
 
# File 'lib/pgbus/configuration.rb', line 526

def priority_queue_names(name)
  return [queue_name(name)] unless priority_levels && priority_levels > 1

  (0...priority_levels).map { |p| priority_queue_name(name, p) }
end

#queue_name(name) ⇒ Object 



513
514
515
516
 
# File 'lib/pgbus/configuration.rb', line 513

def queue_name(name)
  full = "#{queue_prefix}_#{name}"
  QueueNameValidator.normalize(full)
end

#resolved_pool_size ⇒ Object 



1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
 
# File 'lib/pgbus/configuration.rb', line 1283

def resolved_pool_size
  return pool_size if pool_size

  total = 0
  total += sum_thread_counts(workers, default_threads: 5, group: "worker") if role_enabled?(:workers)
  total += sum_thread_counts(event_consumers, default_threads: 3, group: "event_consumer") if role_enabled?(:consumers)
  total += 1 if role_enabled?(:dispatcher)
  total += 1 if role_enabled?(:scheduler)

  warn_if_oversized(total)
  total
end

#role_enabled?(role) ⇒ Boolean

Returns true if the given role should be booted by the supervisor. When roles is nil (the default), every role is enabled — this matches the legacy single-process behavior. When roles is set (e.g. via the CLI's --workers-only / --scheduler-only / --dispatcher-only flags), only the listed roles boot.

Accepts symbol or string for case-insensitive comparison.

Returns:

  • (Boolean) 


1125
1126
1127
1128
1129
 
# File 'lib/pgbus/configuration.rb', line 1125

def role_enabled?(role)
  return true if @roles.nil?

  @roles.include?(role.to_s.downcase.to_sym)
end

#skip_recurring ⇒ Object

skip_recurring was the only negative-polarity toggle among the *_enabled switches. It now aliases recurring_enabled with inverted polarity.



1213
1214
1215
 
# File 'lib/pgbus/configuration.rb', line 1213

def skip_recurring # rubocop:disable Naming/PredicateMethod
  !recurring_enabled
end

#skip_recurring=(value) ⇒ Object 



1217
1218
1219
1220
 
# File 'lib/pgbus/configuration.rb', line 1217

def skip_recurring=(value)
  warn_deprecated_config(:skip_recurring, "use recurring_enabled (positive polarity) instead")
  self.recurring_enabled = !value
end

#stream_durable?(name) ⇒ Boolean

Returns true if the given stream name should be durable based on streams_durable_patterns (exact string or Regexp match) or the streams_default_broadcast_mode fallback.

Returns:

  • (Boolean) 


975
976
977
978
979
980
 
# File 'lib/pgbus/configuration.rb', line 975

def stream_durable?(name)
  patterns = streams_durable_patterns || []
  return true if patterns.any? { |p| p.is_a?(Regexp) ? p.match?(name) : p == name }

  streams_default_broadcast_mode == :durable
end

#stream_presence?(name) ⇒ Boolean

Returns true if the given stream name should have connection-driven presence based on streams_presence_patterns (exact string or Regexp match). Presence is opt-in, so the default (no patterns) is false. See issue #169.

Returns:

  • (Boolean) 


986
987
988
989
 
# File 'lib/pgbus/configuration.rb', line 986

def stream_presence?(name)
  patterns = streams_presence_patterns || []
  patterns.any? { |p| p.is_a?(Regexp) ? p.match?(name) : p == name }
end

#streams_connection_options ⇒ Object

Connection options the Streamer's dedicated LISTEN/NOTIFY PG connection should use. Defaults to connection_options (same as workers and the publish path). If any of streams_database_url, streams_host, or streams_port is set, the Streamer's connection is reconfigured — everything else keeps using the base options.

The typical use is "workers go through PgBouncer, streamer goes direct":

c.connects_to = { database: { writing: :pgbus } }   # pooler
c.streams_port = 5432                               # direct

Precedence: streams_database_url > streams_host/port override > base.



1337
1338
1339
 
# File 'lib/pgbus/configuration.rb', line 1337

def streams_connection_options
  override_connection_options(url: streams_database_url, host: streams_host, port: streams_port)
end

#streams_pool_connection_options ⇒ Object

Connection options for the dedicated streams PGMQ pool — durable-broadcast publish INSERTs and the dispatcher's replay reads (issue #315). Defaults to streams_connection_options, so a genuinely separate streams database carries the pool with it. When any of streams_pool_database_url / streams_pool_host / streams_pool_port is set, the pool is routed independently — applied to the BASE options, exactly like the other two override groups.

Why this exists (issue #358): only the streamer's LISTEN connection needs to bypass a transaction-mode pooler (LISTEN dies at COMMIT boundaries); the pool's INSERT/SELECT traffic is pooler-safe. Without this split, a streams_port direct-port pin drags streams_pool_size connections per process onto the direct port's low max_connections ceiling:

c.streams_port = 5432        # LISTEN bypasses the pooler
c.streams_pool_port = 6432   # the pool stays on the pooler

Precedence: streams_pool_database_url > streams_pool_host/port over the base > streams_connection_options.



1360
1361
1362
1363
1364
1365
1366
 
# File 'lib/pgbus/configuration.rb', line 1360

def streams_pool_connection_options
  return streams_connection_options unless streams_pool_database_url || streams_pool_host || streams_pool_port

  override_connection_options(
    url: streams_pool_database_url, host: streams_pool_host, port: streams_pool_port
  )
end

#validate! ⇒ Object

Raises:



697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
 
# File 'lib/pgbus/configuration.rb', line 697

def validate!
  if pool_size && !(pool_size.is_a?(Numeric) && pool_size.positive?)
    raise Pgbus::ConfigurationError, "pool_size must be a positive number or nil (auto-tune)"
  end

  raise Pgbus::ConfigurationError, "pool_timeout must be > 0" unless pool_timeout.is_a?(Numeric) && pool_timeout.positive?
  raise Pgbus::ConfigurationError, "polling_interval must be > 0" unless polling_interval.is_a?(Numeric) && polling_interval.positive?
  unless visibility_timeout.is_a?(Numeric) && visibility_timeout.positive?
    raise Pgbus::ConfigurationError,
          "visibility_timeout must be > 0"
  end
  raise Pgbus::ConfigurationError, "max_retries must be >= 0" unless max_retries.is_a?(Integer) && max_retries >= 0
  raise Pgbus::ConfigurationError, "retry_backoff must be > 0" unless retry_backoff.is_a?(Numeric) && retry_backoff.positive?
  unless retry_backoff_max.is_a?(Numeric) && retry_backoff_max.positive?
    raise Pgbus::ConfigurationError,
          "retry_backoff_max must be > 0"
  end
  unless retry_backoff_jitter.is_a?(Numeric) && retry_backoff_jitter >= 0 && retry_backoff_jitter <= 1
    raise Pgbus::ConfigurationError, "retry_backoff_jitter must be between 0 and 1"
  end

  unless stall_threshold.nil? || (stall_threshold.is_a?(Numeric) && stall_threshold.positive?)
    raise Pgbus::ConfigurationError, "stall_threshold must be a positive number or nil to disable"
  end
  unless read_timeout.nil? || (read_timeout.is_a?(Numeric) && read_timeout.positive?)
    raise Pgbus::ConfigurationError, "read_timeout must be a positive number or nil to disable"
  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
  unless stats_flush_interval.is_a?(Numeric) && stats_flush_interval.positive?
    raise Pgbus::ConfigurationError, "stats_flush_interval must be a positive number"
  end

  unless health_port.nil? || (health_port.is_a?(Integer) && health_port.between?(1, 65_535))
    raise Pgbus::ConfigurationError, "health_port must be an integer between 1 and 65535 or nil to disable"
  end

  raise Pgbus::ConfigurationError, "health_bind must be a non-empty String" unless health_bind.is_a?(String) && !health_bind.empty?

  raise Pgbus::ConfigurationError, "statsd_host must be a non-empty String" unless statsd_host.is_a?(String) && !statsd_host.empty?

  unless statsd_port.is_a?(Integer) && statsd_port.between?(1, 65_535)
    raise Pgbus::ConfigurationError, "statsd_port must be an integer between 1 and 65535"
  end

  # Validate global execution_mode
  validate_execution_mode!(execution_mode)

  Array(workers).each do |w|
    threads = w[:threads] || 5
    raise Pgbus::ConfigurationError, "worker threads must be > 0" unless threads.is_a?(Integer) && threads.positive?

    # Validate per-worker execution_mode override if present
    mode = w[:execution_mode]
    validate_execution_mode!(mode) if mode
  end

  if prefetch_limit && !(prefetch_limit.is_a?(Integer) && prefetch_limit.positive?)
    raise Pgbus::ConfigurationError,
          "prefetch_limit must be > 0"
  end

  if priority_levels && !(priority_levels.is_a?(Integer) && priority_levels >= 1 && priority_levels <= 10)
    raise Pgbus::ConfigurationError, "priority_levels must be an integer between 1 and 10"
  end

  unless insights_default_minutes.is_a?(Integer) && insights_default_minutes.positive?
    raise Pgbus::ConfigurationError, "insights_default_minutes must be a positive integer"
  end

  raise Pgbus::ConfigurationError, "require_primary must be true or false" unless [true, false].include?(require_primary)

  validate_job_path_gaps!
  validate_streams!
  validate_metrics_backend!

  self
end

#validate_job_path_gaps! ⇒ Object

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 dashboard open. Mirrors the idioms already used above (issue #335).

Raises:



809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
 
# File 'lib/pgbus/configuration.rb', line 809

def validate_job_path_gaps!
  # Worker recycling trio: nil disables the limit; a set value must be a
  # positive Numeric (mirror stall_threshold).
  { max_jobs_per_worker: max_jobs_per_worker, max_memory_mb: max_memory_mb,
    max_worker_lifetime: max_worker_lifetime }.each do |name, value|
    next if value.nil? || (value.is_a?(Numeric) && value.positive?)

    raise Pgbus::ConfigurationError, "#{name} must be a positive number or nil to disable"
  end

  # Interval knobs: positive Numeric, never nil (mirror polling_interval).
  { dispatch_interval: dispatch_interval, outbox_poll_interval: outbox_poll_interval,
    recurring_schedule_interval: recurring_schedule_interval }.each do |name, value|
    raise Pgbus::ConfigurationError, "#{name} must be > 0" unless value.is_a?(Numeric) && value.positive?
  end

  unless outbox_batch_size.is_a?(Integer) && outbox_batch_size.positive?
    raise Pgbus::ConfigurationError, "outbox_batch_size must be a positive integer"
  end

  # 0 is a valid priority level (queue_factory clamps to 0..priority_levels-1).
  unless default_priority.is_a?(Integer) && default_priority >= 0
    raise Pgbus::ConfigurationError, "default_priority must be a non-negative integer"
  end

  # Queue-name components: a nil/empty/non-String prefix silently corrupts
  # every derived queue name (mirror health_bind).
  { queue_prefix: queue_prefix, default_queue: default_queue }.each do |name, value|
    raise Pgbus::ConfigurationError, "#{name} must be a non-empty String" unless value.is_a?(String) && !value.empty?
  end

  # web_auth gates the dashboard; a non-callable value silently leaves it
  # open (mirror streams_presence_member).
  unless web_auth.nil? || web_auth.respond_to?(:call)
    raise Pgbus::ConfigurationError, "web_auth must respond to #call (a Proc/lambda) or be nil"
  end

  raise Pgbus::ConfigurationError, "error_reporters must be an Array" unless error_reporters.is_a?(Array)

  # connects_to is splatted into ActiveRecord's connects_to at engine boot;
  # a non-Hash raises a raw TypeError there — surface a clean message here.
  return if connects_to.nil? || connects_to.is_a?(Hash)

  raise Pgbus::ConfigurationError, "connects_to must be a Hash or nil"
end

#validate_metrics_backend! ⇒ Object

Raises:



961
962
963
964
965
966
967
968
969
970
 
# File 'lib/pgbus/configuration.rb', line 961

def validate_metrics_backend!
  value = metrics_backend
  return if value.nil?
  return if %i[prometheus statsd].include?(value)
  return if defined?(Pgbus::Metrics::Backend) && value.is_a?(Pgbus::Metrics::Backend)

  raise Pgbus::ConfigurationError,
        "metrics_backend must be nil, :prometheus, :statsd, or a " \
        "Pgbus::Metrics::Backend instance (got #{value.inspect})"
end

#validate_shutdown_timeout! ⇒ Object

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.



785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
 
# File 'lib/pgbus/configuration.rb', line 785

def validate_shutdown_timeout!
  explicit = @shutdown_timeout
  # 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 finite 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

#validate_streams! ⇒ Object

Raises:



855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
 
# File 'lib/pgbus/configuration.rb', line 855

def validate_streams!
  unless streams_default_retention.is_a?(Numeric) && streams_default_retention >= 0
    raise Pgbus::ConfigurationError, "streams_default_retention must be a non-negative number"
  end

  unless streams_max_connections.is_a?(Integer) && streams_max_connections.positive?
    raise Pgbus::ConfigurationError, "streams_max_connections must be a positive integer"
  end

  unless streams_heartbeat_interval.is_a?(Numeric) && streams_heartbeat_interval.positive?
    raise Pgbus::ConfigurationError, "streams_heartbeat_interval must be a positive number"
  end

  unless streams_idle_timeout.is_a?(Numeric) && streams_idle_timeout.positive?
    raise Pgbus::ConfigurationError, "streams_idle_timeout must be a positive number"
  end

  unless streams_listen_health_check_ms.is_a?(Integer) && streams_listen_health_check_ms.positive?
    raise Pgbus::ConfigurationError, "streams_listen_health_check_ms must be a positive integer"
  end

  unless streams_write_deadline_ms.is_a?(Integer) && streams_write_deadline_ms.positive?
    raise Pgbus::ConfigurationError, "streams_write_deadline_ms must be a positive integer"
  end

  unless streams_fanout_write_deadline_ms.is_a?(Integer) && streams_fanout_write_deadline_ms.positive?
    raise Pgbus::ConfigurationError, "streams_fanout_write_deadline_ms must be a positive integer"
  end

  # >= 0, NOT .positive? — 0 is the valid "unbounded" sentinel.
  unless streams_dispatch_queue_limit.is_a?(Integer) && streams_dispatch_queue_limit >= 0
    raise Pgbus::ConfigurationError, "streams_dispatch_queue_limit must be a non-negative integer (0 = unbounded)"
  end

  # >= 0, NOT .positive? — 0 is the valid "inline" sentinel.
  unless streams_writer_threads.is_a?(Integer) && streams_writer_threads >= 0
    raise Pgbus::ConfigurationError, "streams_writer_threads must be a non-negative integer (0 = inline)"
  end

  # >= 0, NOT .positive? — 0 is the valid "unbounded" sentinel.
  unless streams_writer_buffer_limit.is_a?(Integer) && streams_writer_buffer_limit >= 0
    raise Pgbus::ConfigurationError, "streams_writer_buffer_limit must be a non-negative integer (0 = unbounded)"
  end

  unless streams_pool_size.is_a?(Integer) && streams_pool_size.positive?
    raise Pgbus::ConfigurationError, "streams_pool_size must be a positive integer"
  end

  unless streams_pool_timeout.is_a?(Numeric) && streams_pool_timeout.positive?
    raise Pgbus::ConfigurationError, "streams_pool_timeout must be a positive number"
  end

  validate_streams_autoscale!

  raise Pgbus::ConfigurationError, "streams_retention must be a Hash" unless streams_retention.is_a?(Hash)

  unless streams_broadcast_queue.nil? ||
         (streams_broadcast_queue.is_a?(String) && !streams_broadcast_queue.empty?)
    raise Pgbus::ConfigurationError,
          "streams_broadcast_queue must be a non-empty String (a queue name) or nil to disable"
  end

  if streams_orphan_sweep_interval && !(streams_orphan_sweep_interval.is_a?(Numeric) && streams_orphan_sweep_interval.positive?)
    raise Pgbus::ConfigurationError, "streams_orphan_sweep_interval must be a positive number or nil to disable"
  end

  unless streams_durable_patterns.is_a?(Array)
    raise Pgbus::ConfigurationError,
          "streams_durable_patterns must be an Array of strings/regex"
  end

  unless streams_presence_patterns.is_a?(Array)
    raise Pgbus::ConfigurationError,
          "streams_presence_patterns must be an Array of strings/regex"
  end

  if !streams_presence_member.nil? && !streams_presence_member.respond_to?(:call)
    raise Pgbus::ConfigurationError, "streams_presence_member must respond to #call (a Proc/lambda) or be nil"
  end

  return if streams_orphan_threshold.nil?
  return if streams_orphan_threshold.is_a?(Numeric) && streams_orphan_threshold.positive?

  raise Pgbus::ConfigurationError, "streams_orphan_threshold must be a positive number or nil to disable"
end

#validate_streams_autoscale! ⇒ Object

Raises:



941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
 
# File 'lib/pgbus/configuration.rb', line 941

def validate_streams_autoscale!
  raise Pgbus::ConfigurationError, "streams_pool_autoscale must be true or false" unless [true, false].include?(streams_pool_autoscale)

  # nil = no hard cap (dynamic fair-share is the ceiling). A positive value
  # below the baseline is an inverted ceiling — fail loud at boot.
  unless streams_pool_max.nil? ||
         (streams_pool_max.is_a?(Integer) && streams_pool_max >= streams_pool_size)
    raise Pgbus::ConfigurationError,
          "streams_pool_max must be an Integer >= streams_pool_size " \
          "(#{streams_pool_size}) or nil to leave growth uncapped"
  end

  unless streams_pool_autoscale_interval.is_a?(Numeric) && streams_pool_autoscale_interval.positive?
    raise Pgbus::ConfigurationError, "streams_pool_autoscale_interval must be a positive number"
  end

  raise Pgbus::ConfigurationError, "streams_application_name must be a non-empty String" \
    unless streams_application_name.is_a?(String) && !streams_application_name.empty?
end

#worker_notify_connection_options ⇒ Object

Connection options for the Worker's dedicated NotifyListener connection. Mirrors streams_connection_options: defaults to the base connection_options, overridable via worker_notify_database_url / worker_notify_host / worker_notify_port so the LISTEN connection can bypass PgBouncer.



1372
1373
1374
 
# File 'lib/pgbus/configuration.rb', line 1372

def worker_notify_connection_options
  override_connection_options(url: worker_notify_database_url, host: worker_notify_host, port: worker_notify_port)
end

#worker_notify_wakeup? ⇒ Boolean

Resolved notify wakeup flag: defaults to listen_notify when nil.

Returns:

  • (Boolean) 


1377
1378
1379
1380
1381
1382
1383
 
# File 'lib/pgbus/configuration.rb', line 1377

def worker_notify_wakeup?
  if @worker_notify_wakeup.nil?
    listen_notify
  else
    @worker_notify_wakeup
  end
end