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/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/mcp/rack_app.rb,
lib/pgbus/mcp/redactor.rb,
lib/pgbus/rate_counter.rb,
lib/pgbus/web/streamer.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/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/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/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,
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,
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/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,
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/falcon_connection.rb,
app/controllers/pgbus/insights_controller.rb,
lib/generators/pgbus/add_outbox_generator.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

Defined Under Namespace

Modules: ActiveJob, Api, ApplicationHelper, AutovacuumTuning, CLI, Concurrency, ErrorReporter, EventBus, ExecutionPools, Generators, Instrumentation, Integrations, LogFormatter, MCP, Metrics, Outbox, PgmqSchema, Process, QueueFactory, QueueNameValidator, Recurring, RetryBackoff, Serializer, Streams, StreamsHelper, 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, 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"
VERSION = 
"0.11.0"

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.stopping ⇒ Object 



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

def stopping
  @stopping || false
end

Class Method Details

.client ⇒ Object 



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

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

.configuration ⇒ Object 



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

def configuration
  @configuration ||= Configuration.new
end

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

Yields:



152
153
154
155
156
157
158
159
 
# File 'lib/pgbus.rb', line 152

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

.loader ⇒ Object 



85
86
87
88
89
90
91
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
 
# File 'lib/pgbus.rb', line 85

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 



263
264
265
 
# File 'lib/pgbus.rb', line 263

def logger
  configuration.logger
end

.logger=(value) ⇒ Object 



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

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.



128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/pgbus.rb', line 128

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)


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

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)


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

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

.reset! ⇒ Object 



248
249
250
251
252
253
 
# File 'lib/pgbus.rb', line 248

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.



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

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.



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

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)


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

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)


213
214
215
 
# File 'lib/pgbus.rb', line 213

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



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

def stream_name_budget
  Streams::Key.queue_name_budget
end

.teardown_models_loader! ⇒ Object 



141
142
143
144
145
146
 
# File 'lib/pgbus.rb', line 141

def teardown_models_loader!
  return unless @models_loader

  @models_loader.unregister
  @models_loader = nil
end