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.



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
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
 
# File 'lib/pgbus/configuration.rb', line 256

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
  @fair_share = nil
  @event_fair_share = nil
  @current_attributes = nil

  @archive_retention = 7 * 24 * 3600 # 7 days
  @batch_retention = 7 * 24 * 3600 # 7 days
  @batch_sweep_interval = 300 # 5 minutes
  @batch_stall_threshold = 300 # 5 minutes, solid_queue's stalled_for default

  @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).



115
116
117
 
# File 'lib/pgbus/configuration.rb', line 115

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.



239
240
241
 
# File 'lib/pgbus/configuration.rb', line 239

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.



239
240
241
 
# File 'lib/pgbus/configuration.rb', line 239

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.



99
100
101
 
# File 'lib/pgbus/configuration.rb', line 99

def archive_retention
  @archive_retention
end

#base_controller_class ⇒ Object

Web dashboard



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

def base_controller_class
  @base_controller_class
end

#batch_retention ⇒ Object

Batch execution-row sweep + finished-batch cleanup. Retention nil disables cleanup (same sentinel as archive_retention). Sweep interval is required. Stall threshold: how long a pending batch may go without a new execution row (or an orphan row without a msg_id) before the sweep repairs it.



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

def batch_retention
  @batch_retention
end

#batch_stall_threshold ⇒ Object

Batch execution-row sweep + finished-batch cleanup. Retention nil disables cleanup (same sentinel as archive_retention). Sweep interval is required. Stall threshold: how long a pending batch may go without a new execution row (or an orphan row without a msg_id) before the sweep repairs it.



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

def batch_stall_threshold
  @batch_stall_threshold
end

#batch_sweep_interval ⇒ Object

Batch execution-row sweep + finished-batch cleanup. Retention nil disables cleanup (same sentinel as archive_retention). Sweep interval is required. Stall threshold: how long a pending batch may go without a new execution row (or an orphan row without a msg_id) before the sweep repairs it.



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

def batch_sweep_interval
  @batch_sweep_interval
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.



180
181
182
 
# File 'lib/pgbus/configuration.rb', line 180

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.



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

def connects_to
  @connects_to
end

#current_attributes ⇒ Object

Persist ActiveSupport::CurrentAttributes across enqueue → perform (issue #430). nil = off (default). :auto = every CurrentAttributes subclass; an Array of classes/names; or a Hash of class/name => { only: [...] } / { except: [...] }. Normalized to :auto or an Array of { name:, only:, except: } specs (classes are resolved lazily by name).



94
95
96
 
# File 'lib/pgbus/configuration.rb', line 94

def current_attributes
  @current_attributes
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.



146
147
148
 
# File 'lib/pgbus/configuration.rb', line 146

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.



254
255
256
 
# File 'lib/pgbus/configuration.rb', line 254

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).



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

def error_reporters
  @error_reporters
end

#event_consumers ⇒ Object

Event consumers



149
150
151
 
# File 'lib/pgbus/configuration.rb', line 149

def event_consumers
  @event_consumers
end

#event_fair_share ⇒ Object

Fair share for event-bus consumers (issue #427). nil = disabled. Any #call-able receiving the Pgbus::Event at publish time (routing_key, payload as passed to publish, headers) and returning nil, a key, or [key, weight] exactly like fair_share. The key rides in the event envelope and consumers read subscriber queues with Client#read_batch_fair. Independent of fair_share (jobs).



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

def event_fair_share
  @event_fair_share
end

#execution_mode ⇒ Object

Worker settings



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

def execution_mode
  @execution_mode
end

#fair_share ⇒ Object

Fair share scheduling across tenants (issue #426). nil = disabled. Any #call-able receiving the ActiveJob instance at enqueue time and returning nil (unkeyed), a key (String/Symbol/Integer), or [key, weight] (weight: positive Numeric, default 1). Keyed jobs are read with Client#read_batch_fair — a weighted, work-conserving interleave across keys within each queue. Mutually exclusive with group_mode.



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

def fair_share
  @fair_share
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).



201
202
203
 
# File 'lib/pgbus/configuration.rb', line 201

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).



201
202
203
 
# File 'lib/pgbus/configuration.rb', line 201

def health_port
  @health_port
end

#idempotency_ttl ⇒ Object

Event bus



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

def idempotency_ttl
  @idempotency_ttl
end

#insights_default_minutes ⇒ Object

Web dashboard



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

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).



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

def listen_notify
  @listen_notify
end

#log_format ⇒ Object

rubocop:disable Style/AccessorGrouping



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

def log_format
  @log_format
end

#logger ⇒ Object

Logging



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

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


247
248
249
 
# File 'lib/pgbus/configuration.rb', line 247

def metrics_backend
  @metrics_backend
end

#metrics_enabled ⇒ Object

Web dashboard



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

def metrics_enabled
  @metrics_enabled
end

#outbox_batch_size ⇒ Object

Transactional outbox



108
109
110
 
# File 'lib/pgbus/configuration.rb', line 108

def outbox_batch_size
  @outbox_batch_size
end

#outbox_enabled ⇒ Object

Transactional outbox



108
109
110
 
# File 'lib/pgbus/configuration.rb', line 108

def outbox_enabled
  @outbox_enabled
end

#outbox_poll_interval ⇒ Object

Transactional outbox



108
109
110
 
# File 'lib/pgbus/configuration.rb', line 108

def outbox_poll_interval
  @outbox_poll_interval
end

#outbox_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



109
110
111
 
# File 'lib/pgbus/configuration.rb', line 109

def outbox_retention
  @outbox_retention
end

#pgmq_schema_mode ⇒ Object

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



133
134
135
 
# File 'lib/pgbus/configuration.rb', line 133

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



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

def recurring_enabled
  @recurring_enabled
end

#recurring_execution_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



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

def recurring_execution_retention
  @recurring_execution_retention
end

#recurring_schedule_interval ⇒ Object

Recurring jobs



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

def recurring_schedule_interval
  @recurring_schedule_interval
end

#recurring_tasks ⇒ Object

Recurring jobs



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

def recurring_tasks
  @recurring_tasks
end

#recurring_tasks_file ⇒ Object

Recurring jobs



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

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.



1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
 
# File 'lib/pgbus/configuration.rb', line 1274

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.



167
168
169
 
# File 'lib/pgbus/configuration.rb', line 167

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



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

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).



1359
1360
1361
 
# File 'lib/pgbus/configuration.rb', line 1359

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



187
188
189
 
# File 'lib/pgbus/configuration.rb', line 187

def stats_enabled
  @stats_enabled
end

#stats_flush_interval ⇒ Object

Job stats



187
188
189
 
# File 'lib/pgbus/configuration.rb', line 187

def stats_flush_interval
  @stats_flush_interval
end

#stats_flush_size ⇒ Object

Job stats



187
188
189
 
# File 'lib/pgbus/configuration.rb', line 187

def stats_flush_size
  @stats_flush_size
end

#stats_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



188
189
190
 
# File 'lib/pgbus/configuration.rb', line 188

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


247
248
249
 
# File 'lib/pgbus/configuration.rb', line 247

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


247
248
249
 
# File 'lib/pgbus/configuration.rb', line 247

def statsd_port
  @statsd_port
end

#streams_application_name ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_application_name
  @streams_application_name
end

#streams_broadcast_queue ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_broadcast_queue
  @streams_broadcast_queue
end

#streams_database_url ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_database_url
  @streams_database_url
end

#streams_default_broadcast_mode ⇒ Object

rubocop:disable Style/AccessorGrouping



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

def streams_default_broadcast_mode
  @streams_default_broadcast_mode
end

#streams_default_retention ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_default_retention
  @streams_default_retention
end

#streams_dispatch_queue_limit ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_dispatch_queue_limit
  @streams_dispatch_queue_limit
end

#streams_durable_patterns ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_durable_patterns
  @streams_durable_patterns
end

#streams_enabled ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_enabled
  @streams_enabled
end

#streams_falcon_streaming_body ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_falcon_streaming_body
  @streams_falcon_streaming_body
end

#streams_fanout_write_deadline_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_fanout_write_deadline_ms
  @streams_fanout_write_deadline_ms
end

#streams_heartbeat_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_heartbeat_interval
  @streams_heartbeat_interval
end

#streams_host ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_host
  @streams_host
end

#streams_idle_timeout ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_idle_timeout
  @streams_idle_timeout
end

#streams_listen_health_check_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

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.


710
711
712
 
# File 'lib/pgbus/configuration.rb', line 710

def streams_listen_scope
  @streams_listen_scope
end

#streams_max_connections ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_max_connections
  @streams_max_connections
end

#streams_orphan_sweep_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_orphan_sweep_interval
  @streams_orphan_sweep_interval
end

#streams_orphan_threshold ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_orphan_threshold
  @streams_orphan_threshold
end

#streams_path ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_path
  @streams_path
end

#streams_pool_autoscale ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_pool_autoscale
  @streams_pool_autoscale
end

#streams_pool_autoscale_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_pool_autoscale_interval
  @streams_pool_autoscale_interval
end

#streams_pool_database_url ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_pool_database_url
  @streams_pool_database_url
end

#streams_pool_host ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_pool_host
  @streams_pool_host
end

#streams_pool_max ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_pool_max
  @streams_pool_max
end

#streams_pool_port ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_pool_port
  @streams_pool_port
end

#streams_pool_size ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_pool_size
  @streams_pool_size
end

#streams_pool_timeout ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_pool_timeout
  @streams_pool_timeout
end

#streams_port ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_port
  @streams_port
end

#streams_presence_member ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_presence_member
  @streams_presence_member
end

#streams_presence_patterns ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_presence_patterns
  @streams_presence_patterns
end

#streams_retention ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_retention
  @streams_retention
end

#streams_signed_name_secret ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_signed_name_secret
  @streams_signed_name_secret
end

#streams_stats_enabled ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_stats_enabled
  @streams_stats_enabled
end

#streams_test_mode ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_test_mode
  @streams_test_mode
end

#streams_write_deadline_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_write_deadline_ms
  @streams_write_deadline_ms
end

#streams_writer_buffer_limit ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

def streams_writer_buffer_limit
  @streams_writer_buffer_limit
end

#streams_writer_threads ⇒ Object

Streams (turbo-rails replacement, SSE-based)



204
205
206
 
# File 'lib/pgbus/configuration.rb', line 204

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



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

def web_auth
  @web_auth
end

#web_data_source ⇒ Object

Web dashboard



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

def web_data_source
  @web_data_source
end

#web_filter_parameters ⇒ Object

Web dashboard



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

def web_filter_parameters
  @web_filter_parameters
end

#web_filter_sensitive ⇒ Object

Web dashboard



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

def web_filter_sensitive
  @web_filter_sensitive
end

#web_live_updates ⇒ Object

Web dashboard



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

def web_live_updates
  @web_live_updates
end

#web_per_page ⇒ Object

Web dashboard



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

def web_per_page
  @web_per_page
end

#web_refresh_interval ⇒ Object

Web dashboard



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

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.



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

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.



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

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.



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

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.


235
236
237
 
# File 'lib/pgbus/configuration.rb', line 235

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.



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

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.



184
185
186
 
# File 'lib/pgbus/configuration.rb', line 184

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:



1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
 
# File 'lib/pgbus/configuration.rb', line 1161

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.



1179
1180
1181
1182
1183
1184
 
# File 'lib/pgbus/configuration.rb', line 1179

def capsule_named(name)
  return nil unless @workers

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

#connection_options ⇒ Object 



1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
 
# File 'lib/pgbus/configuration.rb', line 1376

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.



1303
1304
1305
 
# File 'lib/pgbus/configuration.rb', line 1303

def dashboard_filter_parameters
  web_filter_parameters
end

#dashboard_filter_parameters=(value) ⇒ Object 



1307
1308
1309
1310
 
# File 'lib/pgbus/configuration.rb', line 1307

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 



1312
1313
1314
 
# File 'lib/pgbus/configuration.rb', line 1312

def dashboard_filter_sensitive
  web_filter_sensitive
end

#dashboard_filter_sensitive=(value) ⇒ Object 



1316
1317
1318
1319
 
# File 'lib/pgbus/configuration.rb', line 1316

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 



553
554
555
 
# File 'lib/pgbus/configuration.rb', line 553

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.



569
570
571
572
 
# File 'lib/pgbus/configuration.rb', line 569

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 {}.



1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
 
# File 'lib/pgbus/configuration.rb', line 1067

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 



557
558
559
 
# File 'lib/pgbus/configuration.rb', line 557

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

#priority_queue_names(name) ⇒ Object 



561
562
563
564
565
 
# File 'lib/pgbus/configuration.rb', line 561

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 



548
549
550
551
 
# File 'lib/pgbus/configuration.rb', line 548

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

#resolved_pool_size ⇒ Object 



1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
 
# File 'lib/pgbus/configuration.rb', line 1363

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) 


1193
1194
1195
1196
1197
 
# File 'lib/pgbus/configuration.rb', line 1193

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.



1293
1294
1295
 
# File 'lib/pgbus/configuration.rb', line 1293

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

#skip_recurring=(value) ⇒ Object 



1297
1298
1299
1300
 
# File 'lib/pgbus/configuration.rb', line 1297

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) 


1043
1044
1045
1046
1047
1048
 
# File 'lib/pgbus/configuration.rb', line 1043

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) 


1054
1055
1056
1057
 
# File 'lib/pgbus/configuration.rb', line 1054

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.



1417
1418
1419
 
# File 'lib/pgbus/configuration.rb', line 1417

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.



1440
1441
1442
1443
1444
1445
1446
 
# File 'lib/pgbus/configuration.rb', line 1440

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:



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
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
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
 
# File 'lib/pgbus/configuration.rb', line 754

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!
  validate_fair_share!

  self
end

#validate_fair_share! ⇒ Object

Raises:



915
916
917
918
919
920
921
 
# File 'lib/pgbus/configuration.rb', line 915

def validate_fair_share!
  return unless fair_share && group_mode

  raise Pgbus::ConfigurationError,
        "fair_share and group_mode are mutually exclusive — fair_share interleaves across keys " \
        "within a queue, group_mode serializes PGMQ FIFO groups; pick one"
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:



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
 
# File 'lib/pgbus/configuration.rb', line 867

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,
    batch_sweep_interval: batch_sweep_interval,
    batch_stall_threshold: batch_stall_threshold }.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:



1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
 
# File 'lib/pgbus/configuration.rb', line 1029

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.



843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
 
# File 'lib/pgbus/configuration.rb', line 843

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:



923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
 
# File 'lib/pgbus/configuration.rb', line 923

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:



1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
 
# File 'lib/pgbus/configuration.rb', line 1009

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.



1452
1453
1454
 
# File 'lib/pgbus/configuration.rb', line 1452

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) 


1457
1458
1459
1460
1461
1462
1463
 
# File 'lib/pgbus/configuration.rb', line 1457

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