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/fair_share.rb,
lib/pgbus/mcp/runner.rb,
lib/pgbus/mcp/server.rb,
lib/pgbus/serializer.rb,
lib/pgbus/uniqueness.rb,
lib/pgbus/batch/sweep.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,
lib/pgbus/web/job_context.rb,
app/models/pgbus/semaphore.rb,
lib/pgbus/client/fair_read.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/current_attributes.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/batch_id.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/batch_execution.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/active_job/current_attributes.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_batch_executions_generator.rb,
lib/generators/pgbus/add_job_stats_latency_generator.rb,
lib/generators/pgbus/add_batch_callback_jobs_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, CurrentAttributes, DatabaseTasksGuard, DedicatedConnection, ErrorReporter, EventBus, ExecutionPools, FairShare, 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, BatchExecution, BatchesController, BlockedExecution, BusRecord, CircuitBreaker, Client, ConcurrencyLimitExceeded, Configuration, ConfigurationError, ConnectionCircuitOpenError, CurrentAttributesError, 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.15.2"

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.stopping ⇒ Object 



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

def stopping
  @stopping || false
end

Class Method Details

.client ⇒ Object 



194
195
196
 
# File 'lib/pgbus.rb', line 194

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

.configuration ⇒ Object 



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

def configuration
  @configuration ||= Configuration.new
end

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

Yields:



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

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) 


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

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 



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
131
132
 
# File 'lib/pgbus.rb', line 94

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 



296
297
298
 
# File 'lib/pgbus.rb', line 296

def logger
  configuration.logger
end

.logger=(value) ⇒ Object 



300
301
302
 
# File 'lib/pgbus.rb', line 300

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.



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

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)


268
269
270
 
# File 'lib/pgbus.rb', line 268

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)


277
278
279
 
# File 'lib/pgbus.rb', line 277

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

.reset! ⇒ Object 



281
282
283
284
285
286
 
# File 'lib/pgbus.rb', line 281

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.



292
293
294
 
# File 'lib/pgbus.rb', line 292

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.



217
218
219
220
221
222
223
 
# File 'lib/pgbus.rb', line 217

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)


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

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)


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

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



259
260
261
 
# File 'lib/pgbus.rb', line 259

def stream_name_budget
  Streams::Key.queue_name_budget
end

.teardown_models_loader! ⇒ Object 



150
151
152
153
154
155
 
# File 'lib/pgbus.rb', line 150

def teardown_models_loader!
  return unless @models_loader

  @models_loader.unregister
  @models_loader = nil
end