Module: Pgbus

Defined in:
lib/pgbus.rb,
lib/pgbus/cli.rb,
lib/pgbus/mcp.rb,
lib/pgbus/batch.rb,
lib/pgbus/event.rb,
lib/pgbus/client.rb,
lib/pgbus/doctor.rb,
lib/pgbus/engine.rb,
lib/pgbus/outbox.rb,
lib/pgbus/cli/dlq.rb,
lib/pgbus/metrics.rb,
lib/pgbus/streams.rb,
lib/pgbus/support.rb,
lib/pgbus/testing.rb,
lib/pgbus/version.rb,
lib/pgbus/bus_record.rb,
lib/pgbus/mcp/runner.rb,
lib/pgbus/mcp/server.rb,
lib/pgbus/serializer.rb,
lib/pgbus/uniqueness.rb,
lib/pgbus/concurrency.rb,
lib/pgbus/dedup_cache.rb,
lib/pgbus/pgmq_schema.rb,
lib/pgbus/stat_buffer.rb,
lib/pgbus/streams/key.rb,
lib/pgbus/health_probe.rb,
lib/pgbus/mcp/rack_app.rb,
lib/pgbus/mcp/redactor.rb,
lib/pgbus/rate_counter.rb,
lib/pgbus/web/streamer.rb,
lib/pgbus/cli/dashboard.rb,
lib/pgbus/configuration.rb,
lib/pgbus/log_formatter.rb,
lib/pgbus/mcp/base_tool.rb,
lib/pgbus/outbox/poller.rb,
lib/pgbus/queue_factory.rb,
lib/pgbus/retry_backoff.rb,
lib/pgbus/error_reporter.rb,
lib/pgbus/process/worker.rb,
lib/pgbus/recurring/task.rb,
lib/pgbus/streams/cursor.rb,
lib/pgbus/web/health_app.rb,
lib/pgbus/web/stream_app.rb,
app/models/pgbus/job_stat.rb,
lib/pgbus/circuit_breaker.rb,
lib/pgbus/execution_pools.rb,
lib/pgbus/instrumentation.rb,
lib/pgbus/metrics/backend.rb,
lib/pgbus/streams/filters.rb,
lib/pgbus/web/data_source.rb,
app/models/pgbus/semaphore.rb,
lib/pgbus/process/consumer.rb,
lib/pgbus/streams/envelope.rb,
lib/pgbus/streams/presence.rb,
lib/pgbus/streams/renderer.rb,
lib/pgbus/testing/minitest.rb,
lib/pgbus/autovacuum_tuning.rb,
lib/pgbus/client/read_after.rb,
lib/pgbus/event_bus/handler.rb,
lib/pgbus/process/heartbeat.rb,
lib/pgbus/process/lifecycle.rb,
lib/pgbus/process/wake_pipe.rb,
lib/pgbus/streams/coalescer.rb,
lib/pgbus/table_maintenance.rb,
lib/pgbus/web/health_server.rb,
app/models/pgbus/batch_entry.rb,
app/models/pgbus/queue_state.rb,
app/models/pgbus/stream_stat.rb,
lib/pgbus/active_job/adapter.rb,
lib/pgbus/event_bus/registry.rb,
lib/pgbus/mcp/tools/dlq_tool.rb,
lib/pgbus/metrics/subscriber.rb,
lib/pgbus/process/dispatcher.rb,
lib/pgbus/process/notify_hub.rb,
lib/pgbus/process/queue_lock.rb,
lib/pgbus/process/supervisor.rb,
lib/pgbus/recurring/schedule.rb,
lib/pgbus/streams/streamable.rb,
lib/pgbus/testing/assertions.rb,
lib/pgbus/web/authentication.rb,
lib/pgbus/web/payload_filter.rb,
app/models/pgbus/outbox_entry.rb,
app/models/pgbus/stream_queue.rb,
lib/pgbus/active_job/executor.rb,
lib/pgbus/event_bus/publisher.rb,
lib/pgbus/mcp/health_analyzer.rb,
lib/pgbus/mcp/tools/jobs_tool.rb,
lib/pgbus/process/wake_signal.rb,
lib/pgbus/recurring/scheduler.rb,
lib/pgbus/streams/signed_name.rb,
app/models/pgbus/process_entry.rb,
lib/pgbus/client/notify_stream.rb,
lib/pgbus/database_tasks_guard.rb,
lib/pgbus/dedicated_connection.rb,
lib/pgbus/event_bus/subscriber.rb,
lib/pgbus/mcp/tools/locks_tool.rb,
lib/pgbus/mcp/tools/stats_tool.rb,
lib/pgbus/process/memory_usage.rb,
lib/pgbus/process/notify_probe.rb,
lib/pgbus/queue_name_validator.rb,
lib/pgbus/streams/pool_trigger.rb,
app/models/pgbus/recurring_task.rb,
app/models/pgbus/uniqueness_key.rb,
lib/pgbus/client/resizable_pool.rb,
lib/pgbus/concurrency/semaphore.rb,
lib/pgbus/failed_event_recorder.rb,
lib/pgbus/mcp/tools/health_tool.rb,
lib/pgbus/mcp/tools/queues_tool.rb,
lib/pgbus/recurring/command_job.rb,
lib/pgbus/streams/phlex_helpers.rb,
lib/pgbus/web/streamer/instance.rb,
lib/pgbus/web/streamer/listener.rb,
lib/pgbus/web/streamer/registry.rb,
app/helpers/pgbus/streams_helper.rb,
app/models/pgbus/processed_event.rb,
lib/pgbus/integrations/appsignal.rb,
lib/pgbus/process/signal_handler.rb,
lib/pgbus/web/metrics_serializer.rb,
lib/pgbus/web/streamer/heartbeat.rb,
lib/pgbus/web/streamer/io_writer.rb,
lib/pgbus/metrics/backends/statsd.rb,
lib/pgbus/process/notify_listener.rb,
lib/pgbus/recurring/config_loader.rb,
lib/pgbus/streams/pool_autoscaler.rb,
lib/pgbus/web/streamer/connection.rb,
lib/pgbus/web/streamer/hub_client.rb,
lib/pgbus/web/streamer/master_hub.rb,
app/models/pgbus/blocked_execution.rb,
lib/pgbus/client/connection_health.rb,
lib/pgbus/mcp/tools/processes_tool.rb,
lib/pgbus/mcp/tools/recurring_tool.rb,
app/models/pgbus/application_record.rb,
lib/generators/pgbus/migration_path.rb,
lib/pgbus/configuration/capsule_dsl.rb,
lib/pgbus/mcp/tools/dlq_detail_tool.rb,
lib/pgbus/mcp/tools/job_detail_tool.rb,
lib/pgbus/mcp/tools/throughput_tool.rb,
lib/pgbus/process/consumer_priority.rb,
lib/pgbus/process/primary_validator.rb,
lib/pgbus/web/streamer/hub_protocol.rb,
app/helpers/pgbus/application_helper.rb,
app/models/pgbus/recurring_execution.rb,
lib/pgbus/client/ensure_stream_queue.rb,
lib/pgbus/execution_pools/async_pool.rb,
lib/pgbus/process/readiness_snapshot.rb,
lib/pgbus/recurring/already_recorded.rb,
lib/pgbus/web/streamer/outbound_pump.rb,
app/controllers/pgbus/jobs_controller.rb,
lib/generators/pgbus/update_generator.rb,
lib/pgbus/execution_pools/thread_pool.rb,
lib/pgbus/mcp/tools/queue_detail_tool.rb,
lib/pgbus/metrics/backends/prometheus.rb,
lib/pgbus/metrics/prometheus_exporter.rb,
lib/pgbus/streams/turbo_broadcastable.rb,
lib/pgbus/web/streamer/stream_counter.rb,
app/controllers/pgbus/locks_controller.rb,
lib/generators/pgbus/install_generator.rb,
lib/pgbus/integrations/appsignal/probe.rb,
lib/pgbus/web/streamer/master_hub_boot.rb,
app/controllers/pgbus/events_controller.rb,
app/controllers/pgbus/locale_controller.rb,
app/controllers/pgbus/outbox_controller.rb,
app/controllers/pgbus/queues_controller.rb,
lib/pgbus/concurrency/blocked_execution.rb,
lib/pgbus/generators/migration_detector.rb,
lib/pgbus/streams/turbo_stream_override.rb,
app/controllers/pgbus/batches_controller.rb,
lib/pgbus/streams/broadcastable_override.rb,
lib/pgbus/web/streamer/failover_listener.rb,
lib/pgbus/web/streamer/falcon_connection.rb,
app/controllers/pgbus/insights_controller.rb,
lib/generators/pgbus/add_outbox_generator.rb,
lib/pgbus/process/wildcard_queue_resolver.rb,
app/controllers/pgbus/api/stats_controller.rb,
app/controllers/pgbus/dashboard_controller.rb,
app/controllers/pgbus/frontends_controller.rb,
app/controllers/pgbus/processes_controller.rb,
lib/generators/pgbus/add_presence_generator.rb,
lib/generators/pgbus/upgrade_pgmq_generator.rb,
lib/pgbus/integrations/appsignal/subscriber.rb,
app/controllers/pgbus/api/metrics_controller.rb,
app/controllers/pgbus/application_controller.rb,
app/controllers/pgbus/dead_letter_controller.rb,
lib/generators/pgbus/add_job_stats_generator.rb,
lib/generators/pgbus/add_recurring_generator.rb,
lib/pgbus/streams/watermark_cache_middleware.rb,
app/controllers/pgbus/api/insights_controller.rb,
lib/pgbus/generators/database_target_detector.rb,
lib/generators/pgbus/tune_autovacuum_generator.rb,
lib/generators/pgbus/tune_fillfactor_generator.rb,
lib/pgbus/web/streamer/stream_event_dispatcher.rb,
lib/generators/pgbus/add_queue_states_generator.rb,
lib/generators/pgbus/add_stream_stats_generator.rb,
app/controllers/pgbus/recurring_tasks_controller.rb,
lib/generators/pgbus/add_stream_queues_generator.rb,
lib/generators/pgbus/migrate_job_locks_generator.rb,
lib/generators/pgbus/add_uniqueness_keys_generator.rb,
lib/generators/pgbus/add_job_stats_latency_generator.rb,
lib/generators/pgbus/add_failed_events_index_generator.rb,
lib/generators/pgbus/add_job_stats_queue_index_generator.rb,
lib/generators/pgbus/add_processed_event_completion_generator.rb

Defined Under Namespace

Modules: ActiveJob, Api, ApplicationHelper, AutovacuumTuning, CLI, Concurrency, DatabaseTasksGuard, DedicatedConnection, ErrorReporter, EventBus, ExecutionPools, Generators, Instrumentation, Integrations, LogFormatter, MCP, Metrics, Outbox, PgmqSchema, Process, QueueFactory, QueueNameValidator, Recurring, RetryBackoff, Serializer, Streams, StreamsHelper, Support, TableMaintenance, Testing, Uniqueness, Web Classes: ApplicationController, ApplicationRecord, Batch, BatchEntry, BatchesController, BlockedExecution, BusRecord, CircuitBreaker, Client, ConcurrencyLimitExceeded, Configuration, ConfigurationError, ConnectionCircuitOpenError, DashboardController, DeadLetterController, DeadLetterError, DedupCache, Doctor, Engine, EnqueueError, Error, Event, EventsController, ExecutionPoolError, FailedEventRecorder, FrontendsController, HealthProbe, InsightsController, JobNotUnique, JobStat, JobsController, LocaleController, LocksController, OutboxController, OutboxEntry, ProcessEntry, ProcessedEvent, ProcessesController, QueueNotFoundError, QueueState, QueuesController, RateCounter, ReadTimeoutError, RecurringExecution, RecurringTask, RecurringTasksController, SchemaNotReady, Semaphore, SerializationError, StatBuffer, StreamQueue, StreamStat, UniquenessKey

Constant Summary collapse

DEAD_LETTER_SUFFIX =

Suffix appended to a queue name to derive its dead-letter companion (e.g. "pgbus_default" -> "pgbus_default_dlq"). Hard-coded here because changing it on a running deployment would orphan every existing DLQ message; nothing in the codebase or in user reports has ever needed this to be configurable.

"_dlq"
DATABASE_TASK_PREFIXES =

Rake task-name prefixes during which pgbus must NOT open a database connection: schema management (an idle session on a dedicated pgbus database blocks that same process's DROP DATABASE during db:test:purge — issue #409) and asset precompile (where a database may legitimately not exist). Consulted by Pgbus.database_task?.

%w[db: db_test: assets: webpacker: yarn:].freeze
VERSION = 
"0.14.1"

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.stopping ⇒ Object 



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

def stopping
  @stopping || false
end

Class Method Details

.client ⇒ Object 



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

def client
  @client ||= Client.new(configuration)
end

.configuration ⇒ Object 



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

def configuration
  @configuration ||= Configuration.new
end

.configure {|configuration| ... } ⇒ Object

Yields:



183
184
185
186
187
188
189
190
 
# File 'lib/pgbus.rb', line 183

def configure
  yield configuration
  # Fail loud at boot on an invalid value rather than leaving it dormant
  # until a worker path consumes it. Checking the flag after the yield
  # lets a block opt out for itself via `c.eager_validation = false`.
  configuration.validate! if configuration.eager_validation
  configuration
end

.database_task? ⇒ Boolean

True when the current process is running a rake task (db:, assets:, … — see DATABASE_TASK_PREFIXES) during which pgbus must not touch the database: the database may not exist yet, and an idle connection on a dedicated pgbus database blocks that same process's DROP DATABASE during db:test:purge (issue #409). Use it to guard boot-time database touches in an initializer:

Rails.application.config.after_initialize do
Pgbus::StreamQueue.table_exists? unless Pgbus.database_task?
end

False outside a Rake run, and false (never raise into boot) when detection itself fails.

Returns:

  • (Boolean) 


172
173
174
175
176
177
178
179
180
181
 
# File 'lib/pgbus.rb', line 172

def database_task?
  return false unless defined?(::Rake) && ::Rake.respond_to?(:application)

  ::Rake.application.top_level_tasks.any? do |task|
    DATABASE_TASK_PREFIXES.any? { |prefix| task.to_s.start_with?(prefix) }
  end
rescue StandardError => e
  logger.debug { "[Pgbus] database_task? detection failed, assuming non-schema: #{e.class}: #{e.message}" }
  false
end

.loader ⇒ Object 



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/pgbus.rb', line 92

def loader
  @loader ||= begin
    loader = Zeitwerk::Loader.for_gem
    loader.inflector.inflect(
      "pgbus" => "Pgbus",
      "cli" => "CLI",
      "dlq" => "DLQ",
      "dsl" => "DSL",
      "capsule_dsl" => "CapsuleDSL",
      "mcp" => "MCP"
    )
    loader.ignore("#{__dir__}/generators")
    loader.ignore("#{__dir__}/active_job")
    loader.ignore("#{__dir__}/pgbus/testing")
    # The MCP diagnostic server is optional — its tool classes subclass
    # MCP::Tool from the (optional) `mcp` gem. Keeping it out of Zeitwerk
    # means we never reference the gem's constants at autoload time;
    # `Pgbus::MCP.load!` (called by the `pgbus mcp` CLI command) requires
    # the gem and loads the subsystem explicitly.
    loader.ignore("#{__dir__}/pgbus/mcp")
    loader.ignore("#{__dir__}/pgbus/mcp.rb")
    # The Phlex stream helper references Phlex::Rails::HelperMacros from the
    # optional `phlex-rails` gem. Keep it out of Zeitwerk so eager_load never
    # touches those constants; a Phlex app requires it explicitly via
    # `require "pgbus/streams/phlex_helpers"`.
    loader.ignore("#{__dir__}/pgbus/streams/phlex_helpers.rb")
    # Vendor integrations are loaded conditionally (when the vendor gem
    # is present) by lib/pgbus/engine.rb. Keeping them out of Zeitwerk
    # means we don't reference vendor constants at autoload time.
    loader.ignore("#{__dir__}/pgbus/integrations")
    # lib/puma/plugin/pgbus_streams.rb is a Puma plugin — it's required
    # explicitly by the user from config/puma.rb via `plugin :pgbus_streams`.
    # Without this ignore, Zeitwerk scans lib/puma/ under the pgbus loader
    # root and tries to autoload Puma::Plugin, which collides with the real
    # Puma::Plugin class defined by the puma gem itself.
    loader.ignore("#{__dir__}/puma")
    loader
  end
end

.logger ⇒ Object 



294
295
296
 
# File 'lib/pgbus.rb', line 294

def logger
  configuration.logger
end

.logger=(value) ⇒ Object 



298
299
300
 
# File 'lib/pgbus.rb', line 298

def logger=(value)
  configuration.logger = value
end

.models_loader ⇒ Object

Separate loader for app/models used only in non-Rails contexts (specs, standalone scripts). When the Engine boots, Rails' autoloader takes over app/models and this loader is torn down to avoid conflicts.



135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/pgbus.rb', line 135

def models_loader
  models_dir = File.expand_path("../app/models", __dir__)
  return nil unless File.directory?(models_dir)

  @models_loader ||= begin
    loader = Zeitwerk::Loader.new
    loader.tag = "pgbus-models"
    loader.push_dir(models_dir)
    loader.setup
    loader
  end
end

.publish(routing_key, payload, headers: nil, delay: 0) ⇒ Object

Publish an event to the bus — the top-level shortcut for Pgbus::EventBus::Publisher.publish, symmetric with Pgbus.stream.

Pgbus.publish("orders.created", { order_id: 42 })
Pgbus.publish("orders.created", order, headers: { "x-trace" => id }, delay: 30)


266
267
268
 
# File 'lib/pgbus.rb', line 266

def publish(routing_key, payload, headers: nil, delay: 0)
  EventBus::Publisher.publish(routing_key, payload, headers: headers, delay: delay)
end

.publish_later(routing_key, payload, delay:, headers: nil) ⇒ Object

Publish an event with a delay. Mirrors Pgbus::EventBus::Publisher.publish_later; delay: is required (a publish_later with no delay is just publish).

Pgbus.publish_later("orders.reminder", { order_id: 42 }, delay: 1.hour)


275
276
277
 
# File 'lib/pgbus.rb', line 275

def publish_later(routing_key, payload, delay:, headers: nil)
  EventBus::Publisher.publish_later(routing_key, payload, delay: delay, headers: headers)
end

.reset! ⇒ Object 



279
280
281
282
283
284
 
# File 'lib/pgbus.rb', line 279

def reset!
  @client&.close
  @client = nil
  @configuration = nil
  @stream_cache = nil
end

.reset_client! ⇒ Object

Discard the inherited PGMQ client after fork. Do NOT call close — the parent's @pgmq_mutex is in undefined state post-fork and attempting to acquire it can deadlock. The next call to Pgbus.client will lazily create a fresh one.



290
291
292
 
# File 'lib/pgbus.rb', line 290

def reset_client!
  @client = nil
end

.stream(streamables, durable: nil) ⇒ Object

Entry point for the streams subsystem — Pgbus.stream(name).broadcast(html) or Pgbus.stream(@order).current_msg_id. Defined on Pgbus itself rather than inside lib/pgbus/streams.rb because that file is only Zeitwerk-loaded when Pgbus::Streams::Stream is first referenced — the chicken-and-egg problem means Pgbus.stream(...) would be undefined on the first call. Referencing Streams::Stream inside the method body forces Zeitwerk to load lib/pgbus/streams.rb lazily on first use, which is fine.

Caches Stream instances by logical name so high-frequency callers (e.g. Turbo::StreamsChannel.broadcast_stream_to inside an after_update_commit callback firing 1000x/sec) don't allocate a new Stream + Mutex per broadcast. The cache is process-local; reset! clears it. The cache key is the resolved name string, not the raw streamables, so Pgbus.stream(@order) and Pgbus.stream(@order) in the same process return the same instance. durable: defaults to nil so the configuration resolves the mode: streams_durable_patterns first (exact string or regex match), then streams_default_broadcast_mode. Pass durable: true/false explicitly to bypass the resolver for this Stream instance.



215
216
217
218
219
220
221
 
# File 'lib/pgbus.rb', line 215

def stream(streamables, durable: nil)
  name = Streams::Stream.name_from(streamables)
  resolved = durable.nil? ? configuration.stream_durable?(name) : durable
  cache_key = "#{name}:#{resolved ? "d" : "e"}"
  @stream_cache ||= Concurrent::Map.new
  @stream_cache.compute_if_absent(cache_key) { Streams::Stream.new(streamables, durable: resolved) }
end

.stream_key(*parts) ⇒ Object

Compose a short, pgbus-safe stream identifier from any mix of records, strings, symbols, and arrays. Delegates to Pgbus::Streams::Key.stream_key; raises ArgumentError if the resulting key would overflow the pgbus queue-name budget. See lib/pgbus/streams/key.rb for the digest policy and rationale.

Pgbus.stream_key(chat, :messages)
# => "ai_chat_3a4f9c21b7d20e18:messages"

Pgbus.stream(Pgbus.stream_key(chat, :messages)).broadcast(html)


233
234
235
 
# File 'lib/pgbus.rb', line 233

def stream_key(*parts, **)
  Streams::Key.stream_key(*parts, **)
end

.stream_key!(key) ⇒ Object

Accepts an already-built stream key verbatim (no re-keying), only enforcing the queue-name budget. Use when you hold a key string and want to pass it to both turbo_stream_from and a broadcaster without the colon-separator guard raising on the second call.

key = Pgbus.stream_key(chat, :messages)
Pgbus.stream(Pgbus.stream_key!(key)).broadcast(html)


244
245
246
 
# File 'lib/pgbus.rb', line 244

def stream_key!(key)
  Streams::Key.stream_key!(key)
end

.stream_name_budget ⇒ Object

The maximum length (in characters) a stream name may be before it overflows the pgbus queue-name budget. A stream name becomes the physical queue #{queue_prefix}_<name>, and PGMQ caps queue names, so the usable budget is `QueueNameValidator::MAX_QUEUE_NAME_LENGTH - queue_prefix.length

  • 1. Depends on config.queue_prefix`. Use it to size or truncate stream identifiers up front instead of hand-computing the budget at every call site.

    name = candidate.length > Pgbus.stream_name_budget ? Pgbus.stream_key(record) : candidate



257
258
259
 
# File 'lib/pgbus.rb', line 257

def stream_name_budget
  Streams::Key.queue_name_budget
end

.teardown_models_loader! ⇒ Object 



148
149
150
151
152
153
 
# File 'lib/pgbus.rb', line 148

def teardown_models_loader!
  return unless @models_loader

  @models_loader.unregister
  @models_loader = nil
end