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



191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
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
 
# File 'lib/pgbus/configuration.rb', line 191

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

  @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_host = nil
  @worker_notify_port = nil
  @worker_notify_database_url = nil

  @pgmq_schema_mode = :auto

  @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
  # Retained for backward compatibility only — NO LONGER USED for queue
  # naming or stream detection. Stream queues are named like job queues
  # (`#{queue_prefix}_<name>`, see #queue_name) and are identified via the
  # `pgbus_stream_queues` registry (Pgbus::StreamQueue), not by this prefix.
  # Setting it has no effect. See issue #308.
  @streams_queue_prefix = "pgbus_stream"
  # 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
  # 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

Event bus



73
74
75
 
# File 'lib/pgbus/configuration.rb', line 73

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.



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

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.



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

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.



66
67
68
 
# File 'lib/pgbus/configuration.rb', line 66

def archive_retention
  @archive_retention
end

#base_controller_class ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

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.



44
45
46
 
# File 'lib/pgbus/configuration.rb', line 44

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.



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

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.



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

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



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

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



39
40
41
 
# File 'lib/pgbus/configuration.rb', line 39

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



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

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



82
83
84
 
# File 'lib/pgbus/configuration.rb', line 82

def error_reporters
  @error_reporters
end

#event_consumers ⇒ Object

Event consumers



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

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



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

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



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

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



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

def health_port
  @health_port
end

#idempotency_ttl ⇒ Object

rubocop:disable Style/AccessorGrouping



74
75
76
 
# File 'lib/pgbus/configuration.rb', line 74

def idempotency_ttl
  @idempotency_ttl
end

#insights_default_minutes ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

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



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

def listen_notify
  @listen_notify
end

#log_format ⇒ Object

rubocop:disable Style/AccessorGrouping



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

def log_format
  @log_format
end

#logger ⇒ Object

Logging



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

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



47
48
49
 
# File 'lib/pgbus/configuration.rb', line 47

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


182
183
184
 
# File 'lib/pgbus/configuration.rb', line 182

def metrics_backend
  @metrics_backend
end

#metrics_enabled ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

def metrics_enabled
  @metrics_enabled
end

#outbox_batch_size ⇒ Object

Transactional outbox



69
70
71
 
# File 'lib/pgbus/configuration.rb', line 69

def outbox_batch_size
  @outbox_batch_size
end

#outbox_enabled ⇒ Object

Transactional outbox



69
70
71
 
# File 'lib/pgbus/configuration.rb', line 69

def outbox_enabled
  @outbox_enabled
end

#outbox_poll_interval ⇒ Object

Transactional outbox



69
70
71
 
# File 'lib/pgbus/configuration.rb', line 69

def outbox_poll_interval
  @outbox_poll_interval
end

#outbox_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



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

def outbox_retention
  @outbox_retention
end

#pgmq_schema_mode ⇒ Object

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



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

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



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

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



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

def recurring_enabled
  @recurring_enabled
end

#recurring_execution_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



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

def recurring_execution_retention
  @recurring_execution_retention
end

#recurring_schedule_interval ⇒ Object

Recurring jobs



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

def recurring_schedule_interval
  @recurring_schedule_interval
end

#recurring_tasks ⇒ Object

Recurring jobs



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

def recurring_tasks
  @recurring_tasks
end

#recurring_tasks_file ⇒ Object

Recurring jobs



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

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.



979
980
981
982
983
984
985
986
987
988
989
990
991
992
 
# File 'lib/pgbus/configuration.rb', line 979

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.



111
112
113
 
# File 'lib/pgbus/configuration.rb', line 111

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.



51
52
53
 
# File 'lib/pgbus/configuration.rb', line 51

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.



51
52
53
 
# File 'lib/pgbus/configuration.rb', line 51

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.



51
52
53
 
# File 'lib/pgbus/configuration.rb', line 51

def retry_backoff_max
  @retry_backoff_max
end

#return_to_app_url ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

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

#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



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

def stats_enabled
  @stats_enabled
end

#stats_flush_interval ⇒ Object

Job stats



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

def stats_flush_interval
  @stats_flush_interval
end

#stats_flush_size ⇒ Object

Job stats



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

def stats_flush_size
  @stats_flush_size
end

#stats_retention ⇒ Object

rubocop:disable Style/AccessorGrouping



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

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


182
183
184
 
# File 'lib/pgbus/configuration.rb', line 182

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


182
183
184
 
# File 'lib/pgbus/configuration.rb', line 182

def statsd_port
  @statsd_port
end

#streams_application_name ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_application_name
  @streams_application_name
end

#streams_broadcast_queue ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_broadcast_queue
  @streams_broadcast_queue
end

#streams_database_url ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_database_url
  @streams_database_url
end

#streams_default_broadcast_mode ⇒ Object

rubocop:disable Style/AccessorGrouping



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

def streams_default_broadcast_mode
  @streams_default_broadcast_mode
end

#streams_default_retention ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_default_retention
  @streams_default_retention
end

#streams_dispatch_queue_limit ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_dispatch_queue_limit
  @streams_dispatch_queue_limit
end

#streams_durable_patterns ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_durable_patterns
  @streams_durable_patterns
end

#streams_enabled ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_enabled
  @streams_enabled
end

#streams_falcon_streaming_body ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_falcon_streaming_body
  @streams_falcon_streaming_body
end

#streams_fanout_write_deadline_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_fanout_write_deadline_ms
  @streams_fanout_write_deadline_ms
end

#streams_heartbeat_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_heartbeat_interval
  @streams_heartbeat_interval
end

#streams_host ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_host
  @streams_host
end

#streams_idle_timeout ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_idle_timeout
  @streams_idle_timeout
end

#streams_listen_health_check_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_listen_health_check_ms
  @streams_listen_health_check_ms
end

#streams_max_connections ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_max_connections
  @streams_max_connections
end

#streams_orphan_sweep_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_orphan_sweep_interval
  @streams_orphan_sweep_interval
end

#streams_orphan_threshold ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_orphan_threshold
  @streams_orphan_threshold
end

#streams_path ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_path
  @streams_path
end

#streams_pool_autoscale ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_pool_autoscale
  @streams_pool_autoscale
end

#streams_pool_autoscale_interval ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_pool_autoscale_interval
  @streams_pool_autoscale_interval
end

#streams_pool_max ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_pool_max
  @streams_pool_max
end

#streams_pool_size ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_pool_size
  @streams_pool_size
end

#streams_pool_timeout ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_pool_timeout
  @streams_pool_timeout
end

#streams_port ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_port
  @streams_port
end

#streams_presence_member ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_presence_member
  @streams_presence_member
end

#streams_presence_patterns ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_presence_patterns
  @streams_presence_patterns
end

#streams_queue_prefix ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_queue_prefix
  @streams_queue_prefix
end

#streams_retention ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_retention
  @streams_retention
end

#streams_signed_name_secret ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_signed_name_secret
  @streams_signed_name_secret
end

#streams_stats_enabled ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_stats_enabled
  @streams_stats_enabled
end

#streams_test_mode ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_test_mode
  @streams_test_mode
end

#streams_write_deadline_ms ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_write_deadline_ms
  @streams_write_deadline_ms
end

#streams_writer_buffer_limit ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

def streams_writer_buffer_limit
  @streams_writer_buffer_limit
end

#streams_writer_threads ⇒ Object

Streams (turbo-rails replacement, SSE-based)



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

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



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

def web_auth
  @web_auth
end

#web_data_source ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

def web_data_source
  @web_data_source
end

#web_filter_parameters ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

def web_filter_parameters
  @web_filter_parameters
end

#web_filter_sensitive ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

def web_filter_sensitive
  @web_filter_sensitive
end

#web_live_updates ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

def web_live_updates
  @web_live_updates
end

#web_per_page ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

def web_per_page
  @web_per_page
end

#web_refresh_interval ⇒ Object

Web dashboard



135
136
137
 
# File 'lib/pgbus/configuration.rb', line 135

def web_refresh_interval
  @web_refresh_interval
end

#worker_notify_database_url ⇒ Object

NOTIFY-gated worker wakeups. When true, each Worker fork owns a dedicated NotifyListener PG connection that LISTENs on its queues' INSERT channels and wakes the loop on a real insert. Defaults to the value of listen_notify. The worker_notify_* overrides mirror streams_* so the LISTEN connection can bypass PgBouncer.



169
170
171
 
# File 'lib/pgbus/configuration.rb', line 169

def worker_notify_database_url
  @worker_notify_database_url
end

#worker_notify_host ⇒ Object

NOTIFY-gated worker wakeups. When true, each Worker fork owns a dedicated NotifyListener PG connection that LISTENs on its queues' INSERT channels and wakes the loop on a real insert. Defaults to the value of listen_notify. The worker_notify_* overrides mirror streams_* so the LISTEN connection can bypass PgBouncer.



169
170
171
 
# File 'lib/pgbus/configuration.rb', line 169

def worker_notify_host
  @worker_notify_host
end

#worker_notify_port ⇒ Object

NOTIFY-gated worker wakeups. When true, each Worker fork owns a dedicated NotifyListener PG connection that LISTENs on its queues' INSERT channels and wakes the loop on a real insert. Defaults to the value of listen_notify. The worker_notify_* overrides mirror streams_* so the LISTEN connection can bypass PgBouncer.



169
170
171
 
# File 'lib/pgbus/configuration.rb', line 169

def worker_notify_port
  @worker_notify_port
end

#worker_notify_wakeup ⇒ Object

NOTIFY-gated worker wakeups. When true, each Worker fork owns a dedicated NotifyListener PG connection that LISTENs on its queues' INSERT channels and wakes the loop on a real insert. Defaults to the value of listen_notify. The worker_notify_* overrides mirror streams_* so the LISTEN connection can bypass PgBouncer.



169
170
171
 
# File 'lib/pgbus/configuration.rb', line 169

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.



128
129
130
 
# File 'lib/pgbus/configuration.rb', line 128

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:



878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
 
# File 'lib/pgbus/configuration.rb', line 878

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.



896
897
898
899
900
901
 
# File 'lib/pgbus/configuration.rb', line 896

def capsule_named(name)
  return nil unless @workers

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

#connection_options ⇒ Object 



1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
 
# File 'lib/pgbus/configuration.rb', line 1075

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.



1008
1009
1010
 
# File 'lib/pgbus/configuration.rb', line 1008

def dashboard_filter_parameters
  web_filter_parameters
end

#dashboard_filter_parameters=(value) ⇒ Object 



1012
1013
1014
1015
 
# File 'lib/pgbus/configuration.rb', line 1012

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 



1017
1018
1019
 
# File 'lib/pgbus/configuration.rb', line 1017

def dashboard_filter_sensitive
  web_filter_sensitive
end

#dashboard_filter_sensitive=(value) ⇒ Object 



1021
1022
1023
1024
 
# File 'lib/pgbus/configuration.rb', line 1021

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 



463
464
465
 
# File 'lib/pgbus/configuration.rb', line 463

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.



479
480
481
482
 
# File 'lib/pgbus/configuration.rb', line 479

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



784
785
786
787
788
789
790
791
792
793
794
 
# File 'lib/pgbus/configuration.rb', line 784

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 



467
468
469
 
# File 'lib/pgbus/configuration.rb', line 467

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

#priority_queue_names(name) ⇒ Object 



471
472
473
474
475
 
# File 'lib/pgbus/configuration.rb', line 471

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 



458
459
460
461
 
# File 'lib/pgbus/configuration.rb', line 458

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

#resolved_pool_size ⇒ Object 



1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
 
# File 'lib/pgbus/configuration.rb', line 1062

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) 


910
911
912
913
914
 
# File 'lib/pgbus/configuration.rb', line 910

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.



998
999
1000
 
# File 'lib/pgbus/configuration.rb', line 998

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

#skip_recurring=(value) ⇒ Object 



1002
1003
1004
1005
 
# File 'lib/pgbus/configuration.rb', line 1002

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) 


760
761
762
763
764
765
 
# File 'lib/pgbus/configuration.rb', line 760

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) 


771
772
773
774
 
# File 'lib/pgbus/configuration.rb', line 771

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.



1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
 
# File 'lib/pgbus/configuration.rb', line 1116

def streams_connection_options
  return streams_database_url if streams_database_url

  base = connection_options
  return base unless streams_host || streams_port

  case base
  when Hash
    result = base.dup
    result[:host] = streams_host if streams_host
    result[:port] = streams_port if streams_port
    result
  when String
    # libpq's conninfo parser takes later key=value pairs as overrides
    # for earlier ones, so we just append. Handles both URI form
    # (postgres://...) and key=value form.
    parts = [base]
    parts << "host=#{streams_host}" if streams_host
    parts << "port=#{streams_port}" if streams_port
    parts.join(" ")
  else
    base
  end
end

#validate! ⇒ Object

Raises:



559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
 
# File 'lib/pgbus/configuration.rb', line 559

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?

  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_streams!
  validate_metrics_backend!

  self
end

#validate_metrics_backend! ⇒ Object

Raises:



746
747
748
749
750
751
752
753
754
755
 
# File 'lib/pgbus/configuration.rb', line 746

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_streams! ⇒ Object

Raises:



640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
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
 
# File 'lib/pgbus/configuration.rb', line 640

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:



726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
 
# File 'lib/pgbus/configuration.rb', line 726

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.



1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
 
# File 'lib/pgbus/configuration.rb', line 1145

def worker_notify_connection_options
  return worker_notify_database_url if worker_notify_database_url

  base = connection_options
  return base unless worker_notify_host || worker_notify_port

  case base
  when Hash
    result = base.dup
    result[:host] = worker_notify_host if worker_notify_host
    result[:port] = worker_notify_port if worker_notify_port
    result
  when String
    parts = [base]
    parts << "host=#{worker_notify_host}" if worker_notify_host
    parts << "port=#{worker_notify_port}" if worker_notify_port
    parts.join(" ")
  else
    base
  end
end

#worker_notify_wakeup? ⇒ Boolean

Resolved notify wakeup flag: defaults to listen_notify when nil.

Returns:

  • (Boolean) 


1168
1169
1170
1171
1172
1173
1174
 
# File 'lib/pgbus/configuration.rb', line 1168

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