Class: Pgbus::Client

Inherits:
Object
  • Object
show all
Includes:
EnsureStreamQueue, NotifyStream, ReadAfter
Defined in:
lib/pgbus/client.rb,
lib/pgbus/client/read_after.rb,
lib/pgbus/client/notify_stream.rb,
lib/pgbus/client/ensure_stream_queue.rb

Defined Under Namespace

Modules: EnsureStreamQueue, NotifyStream, ReadAfter

Constant Summary collapse

NOTIFY_THROTTLE_MS =

Throttle window for PGMQ’s enable_notify_insert trigger. Postgres NOTIFYs are coalesced into one wake-up per window, so a value of 250ms means: at most 4 broadcasts/sec per queue, regardless of insert rate. The trigger is a Postgres-level concern; exposing it as a setting never came up in practice and changing it on the fly would require re-running the trigger DDL on every queue.

250

Constants included from ReadAfter

ReadAfter::DEFAULT_LIMIT

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from NotifyStream

#notify_stream

Methods included from EnsureStreamQueue

#ensure_stream_queue

Methods included from ReadAfter

#read_after, #stream_current_msg_id, #stream_oldest_msg_id

Constructor Details

#initialize(config = Pgbus.configuration) ⇒ Client

Returns a new instance of Client.



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/pgbus/client.rb', line 27

def initialize(config = Pgbus.configuration)
  # Define the PGMQ module before requiring the gem so that Zeitwerk's
  # eager_load (called inside pgmq.rb) can resolve the constant.
  # Without this, Ruby 4.0 + Zeitwerk 2.7.5 raises NameError because
  # eager_load runs const_get(:Client) on PGMQ before the module is defined.
  PGMQ_REQUIRE_MUTEX.synchronize do
    Object.const_set(:PGMQ, Module.new) unless defined?(::PGMQ)
    require "pgmq"
  end
  @config = config
  conn_opts = config.connection_options
  @shared_connection = conn_opts.is_a?(Proc)

  if @shared_connection
    # When using the Rails lambda path (-> { AR::Base.connection.raw_connection }),
    # the Proc returns the same underlying PG::Connection that ActiveRecord uses.
    # PG::Connection (libpq) is not thread-safe — concurrent access causes
    # segfaults and result corruption. Force pool_size=1 and serialize all
    # operations through a mutex.
    @pgmq = PGMQ::Client.new(conn_opts, pool_size: 1, pool_timeout: config.pool_timeout)
    @pgmq_mutex = Mutex.new
  else
    # With a String URL or Hash params, pgmq-ruby creates its own dedicated
    # PG::Connection per pool slot — no shared state with ActiveRecord.
    # Use the resolved pool size (auto-tuned from worker thread counts
    # unless explicitly set) and let pgmq-ruby's connection_pool handle
    # concurrency internally (no mutex needed).
    @pgmq = PGMQ::Client.new(conn_opts, pool_size: config.resolved_pool_size, pool_timeout: config.pool_timeout)
    @pgmq_mutex = nil
  end

  @queues_created = Concurrent::Map.new
  @stream_indexes_created = Concurrent::Map.new
  @queue_strategy = QueueFactory.for(config)
  @schema_ensured = false
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



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

def config
  @config
end

#pgmq ⇒ Object (readonly)

Returns the value of attribute pgmq.



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

def pgmq
  @pgmq
end

Instance Method Details

#archive_batch(queue_name, msg_ids, prefixed: true) ⇒ Object

Batch archive — moves multiple messages to the archive table in one call.



208
209
210
211
212
213
 
# File 'lib/pgbus/client.rb', line 208

def archive_batch(queue_name, msg_ids, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.archive_batch(name, msg_ids) }
  end
end

#archive_message(queue_name, msg_id, prefixed: true) ⇒ Object

Archive a message. Pass prefixed: false when queue_name is already the full PGMQ queue name.



200
201
202
203
204
205
 
# File 'lib/pgbus/client.rb', line 200

def archive_message(queue_name, msg_id, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.archive(name, msg_id) }
  end
end

#bind_topic(pattern, queue_name) ⇒ Object

Topic routing



355
356
357
358
359
360
361
 
# File 'lib/pgbus/client.rb', line 355

def bind_topic(pattern, queue_name)
  full_name = config.queue_name(queue_name)
  with_stale_connection_retry do
    ensure_queue(queue_name)
    synchronized { @pgmq.bind_topic(pattern, full_name) }
  end
end

#close ⇒ Object 



376
377
378
 
# File 'lib/pgbus/client.rb', line 376

def close
  synchronized { @pgmq.close }
end

#delete_batch(queue_name, msg_ids, prefixed: true) ⇒ Object

Batch delete — permanently removes multiple messages in one call.



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

def delete_batch(queue_name, msg_ids, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.delete_batch(name, msg_ids) }
  end
end

#delete_message(queue_name, msg_id, prefixed: true) ⇒ Object

Delete a message. Pass prefixed: false when queue_name is already the full PGMQ queue name (e.g. from priority sub-queues or dashboard).



191
192
193
194
195
196
 
# File 'lib/pgbus/client.rb', line 191

def delete_message(queue_name, msg_id, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.delete(name, msg_id) }
  end
end

#drop_queue(queue_name, prefixed: true) ⇒ Object 



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

def drop_queue(queue_name, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  result = with_stale_connection_retry do
    synchronized { @pgmq.drop_queue(name) }
  end
  @queues_created.delete(name)
  result
end

#ensure_all_queues ⇒ Object 



69
70
71
72
73
 
# File 'lib/pgbus/client.rb', line 69

def ensure_all_queues
  queue_names = collect_configured_queues
  Pgbus.logger.info { "[Pgbus] Bootstrapping #{queue_names.size} queue(s): #{queue_names.join(", ")}" }
  queue_names.each { |name| ensure_queue(name) }
end

#ensure_dead_letter_queue(name) ⇒ Object 



75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/pgbus/client.rb', line 75

def ensure_dead_letter_queue(name)
  dlq_name = config.dead_letter_queue_name(name)
  return if @queues_created[dlq_name]

  @queues_created.compute_if_absent(dlq_name) do
    synchronized do
      @pgmq.create(dlq_name)
      tune_autovacuum(dlq_name)
    end
    true
  end
end

#ensure_queue(name) ⇒ Object 



64
65
66
67
 
# File 'lib/pgbus/client.rb', line 64

def ensure_queue(name)
  ensure_pgmq_schema
  @queue_strategy.physical_queue_names(name).each { |pq| ensure_single_queue(pq) }
end

#list_queues ⇒ Object 



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

def list_queues
  with_stale_connection_retry do
    synchronized { @pgmq.list_queues }
  end
end

#message_exists?(queue_name, msg_id: nil, uniqueness_key: nil) ⇒ Boolean

Check whether a message exists in the given queue.

Pass either msg_id for a fast primary-key lookup, or uniqueness_key to scan the queue for any message whose payload carries that key in the pgbus_uniqueness_key JSONB field. The latter is used by the dispatcher reaper to determine if a uniqueness lock with msg_id=0 (placeholder) still has a corresponding queue message.

queue_name may be either a logical name (e.g. “default”) or an already prefixed physical name (e.g. “pgbus_default”). The client normalizes both.

Returns:

true  — the message definitely exists in the queue
false — the message definitely does not exist
nil   — could not determine (e.g. queue table missing or unknown error).
        Callers MUST treat nil as "exists" for safety.

Returns:

  • (Boolean) 


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

def message_exists?(queue_name, msg_id: nil, uniqueness_key: nil)
  has_msg_id = !msg_id.nil?
  has_uniqueness_key = !uniqueness_key.nil?
  raise ArgumentError, "pass exactly one of msg_id or uniqueness_key" unless has_msg_id ^ has_uniqueness_key

  full_name = resolve_full_queue_name(queue_name)
  sanitized = QueueNameValidator.sanitize!(full_name)

  synchronized do
    with_raw_connection do |conn|
      if has_msg_id
        msg_id_present?(conn, sanitized, msg_id.to_i)
      else
        uniqueness_key_present?(conn, sanitized, uniqueness_key)
      end
    end
  end
rescue ActiveRecord::StatementInvalid => e
  raise unless undefined_table_error?(e)

  nil
rescue StandardError => e
  raise unless defined?(PG::UndefinedTable) && e.is_a?(PG::UndefinedTable)

  nil
end

#metrics(queue_name = nil) ⇒ Object 



256
257
258
259
260
261
262
263
264
265
266
 
# File 'lib/pgbus/client.rb', line 256

def metrics(queue_name = nil)
  with_stale_connection_retry do
    synchronized do
      if queue_name
        @pgmq.metrics(config.queue_name(queue_name))
      else
        @pgmq.metrics_all
      end
    end
  end
end

#move_to_dead_letter(queue_name, message) ⇒ Object 



241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/pgbus/client.rb', line 241

def move_to_dead_letter(queue_name, message)
  dlq_name = config.dead_letter_queue_name(queue_name)
  full_queue = config.queue_name(queue_name)

  with_stale_connection_retry do
    ensure_dead_letter_queue(queue_name)
    synchronized do
      @pgmq.transaction do |txn|
        txn.produce(dlq_name, message.message, headers: message.headers)
        txn.delete(full_queue, message.msg_id.to_i)
      end
    end
  end
end

#publish_to_topic(routing_key, payload, headers: nil, delay: 0) ⇒ Object 



363
364
365
366
367
368
369
370
371
372
373
374
 
# File 'lib/pgbus/client.rb', line 363

def publish_to_topic(routing_key, payload, headers: nil, delay: 0)
  with_stale_connection_retry do
    synchronized do
      @pgmq.produce_topic(
        routing_key,
        serialize(payload),
        headers: headers && serialize(headers),
        delay: delay
      )
    end
  end
end

#purge_archive(queue_name, older_than:, batch_size: 1000) ⇒ Object 



333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
 
# File 'lib/pgbus/client.rb', line 333

def purge_archive(queue_name, older_than:, batch_size: 1000)
  full_name = config.queue_name(queue_name)
  sanitized = QueueNameValidator.sanitize!(full_name)
  total = 0

  sql = "DELETE FROM pgmq.a_#{sanitized} " \
        "WHERE ctid = ANY(ARRAY(SELECT ctid FROM pgmq.a_#{sanitized} WHERE enqueued_at < $1 LIMIT $2))"

  loop do
    deleted = synchronized do
      with_raw_connection do |conn|
        conn.exec_params(sql, [older_than, batch_size]).cmd_tuples
      end
    end
    total += deleted
    break if deleted < batch_size
  end

  total
end

#purge_queue(queue_name, prefixed: true) ⇒ Object 



274
275
276
277
278
279
 
# File 'lib/pgbus/client.rb', line 274

def purge_queue(queue_name, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.purge_queue(name) }
  end
end

#read_batch(queue_name, qty:, vt: nil) ⇒ Object 



118
119
120
121
122
123
124
125
 
# File 'lib/pgbus/client.rb', line 118

def read_batch(queue_name, qty:, vt: nil)
  full_name = config.queue_name(queue_name)
  Instrumentation.instrument("pgbus.client.read_batch", queue: full_name, qty: qty) do
    with_stale_connection_retry do
      synchronized { @pgmq.read_batch(full_name, vt: vt || config.visibility_timeout, qty: qty) }
    end
  end
end

#read_batch_prioritized(queue_name, qty:, vt: nil) ⇒ Object

Read from priority sub-queues, highest priority (p0) first. Returns [priority_queue_name, messages] pairs.



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/pgbus/client.rb', line 129

def read_batch_prioritized(queue_name, qty:, vt: nil)
  unless @queue_strategy.priority?
    return (read_batch(queue_name, qty: qty, vt: vt) || []).map do |m|
      [config.queue_name(queue_name), m]
    end
  end

  remaining = qty
  results = []

  config.priority_queue_names(queue_name).each do |pq_name|
    break if remaining <= 0

    msgs = Instrumentation.instrument("pgbus.client.read_batch", queue: pq_name, qty: remaining) do
      with_stale_connection_retry do
        synchronized { @pgmq.read_batch(pq_name, vt: vt || config.visibility_timeout, qty: remaining) }
      end
    end || []

    msgs.each { |m| results << [pq_name, m] }
    remaining -= msgs.size
  end

  results
end

#read_message(queue_name, vt: nil) ⇒ Object 



109
110
111
112
113
114
115
116
 
# File 'lib/pgbus/client.rb', line 109

def read_message(queue_name, vt: nil)
  full_name = config.queue_name(queue_name)
  Instrumentation.instrument("pgbus.client.read_message", queue: full_name) do
    with_stale_connection_retry do
      synchronized { @pgmq.read(full_name, vt: vt || config.visibility_timeout) }
    end
  end
end

#read_multi(queue_names, qty:, vt: nil, limit: nil) ⇒ Object

Read from multiple queues in a single SQL query (UNION ALL). Each returned message includes a queue_name field identifying its source. queue_names should be logical names (prefix is added automatically).

‘qty` is the per-queue cap (pgmq-ruby semantics), so without `limit:` the caller receives up to `queue_count * qty` messages. Pass `limit:` to cap the total across all queues — required when feeding a fixed-size pool, otherwise the pool can overflow on multi-queue reads (issue #123).



178
179
180
181
182
183
184
185
186
187
 
# File 'lib/pgbus/client.rb', line 178

def read_multi(queue_names, qty:, vt: nil, limit: nil)
  full_names = queue_names.map { |q| config.queue_name(q) }
  Instrumentation.instrument("pgbus.client.read_multi", queues: full_names, qty: qty, limit: limit) do
    with_stale_connection_retry do
      synchronized do
        @pgmq.read_multi(full_names, vt: vt || config.visibility_timeout, qty: qty, limit: limit)
      end
    end
  end
end

#read_with_poll(queue_name, qty:, vt: nil, max_poll_seconds: 5, poll_interval_ms: 100) ⇒ Object 



155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/pgbus/client.rb', line 155

def read_with_poll(queue_name, qty:, vt: nil, max_poll_seconds: 5, poll_interval_ms: 100)
  full_name = config.queue_name(queue_name)
  with_stale_connection_retry do
    synchronized do
      @pgmq.read_with_poll(
        full_name,
        vt: vt || config.visibility_timeout,
        qty: qty,
        max_poll_seconds: max_poll_seconds,
        poll_interval_ms: poll_interval_ms
      )
    end
  end
end

#send_batch(queue_name, payloads, headers: nil, delay: 0) ⇒ Object 



98
99
100
101
102
103
104
105
106
107
 
# File 'lib/pgbus/client.rb', line 98

def send_batch(queue_name, payloads, headers: nil, delay: 0)
  full_name = config.queue_name(queue_name)
  serialized, serialized_headers = serialize_batch(payloads, headers)
  Instrumentation.instrument("pgbus.client.send_batch", queue: full_name, size: payloads.size) do
    with_stale_connection_retry do
      ensure_queue(queue_name)
      synchronized { @pgmq.produce_batch(full_name, serialized, headers: serialized_headers, delay: delay) }
    end
  end
end

#send_message(queue_name, payload, headers: nil, delay: 0, priority: nil) ⇒ Object 



88
89
90
91
92
93
94
95
96
 
# File 'lib/pgbus/client.rb', line 88

def send_message(queue_name, payload, headers: nil, delay: 0, priority: nil)
  target = @queue_strategy.target_queue(queue_name, priority)
  Instrumentation.instrument("pgbus.client.send_message", queue: target) do
    with_stale_connection_retry do
      ensure_queue(queue_name)
      synchronized { @pgmq.produce(target, serialize(payload), headers: headers && serialize(headers), delay: delay) }
    end
  end
end

#set_visibility_timeout(queue_name, msg_id, vt:, prefixed: true) ⇒ Object

Set visibility timeout. Pass prefixed: false when queue_name is already the full PGMQ queue name.



225
226
227
228
229
230
 
# File 'lib/pgbus/client.rb', line 225

def set_visibility_timeout(queue_name, msg_id, vt:, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.set_vt(name, msg_id, vt: vt) }
  end
end

#transaction(&block) ⇒ Object

Open a PGMQ transaction. The caller block may run twice if the first attempt hits a pre-flight stale-connection error — safe because no SQL was sent on the first attempt (the connection was dead before the BEGIN).



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

def transaction(&block)
  with_stale_connection_retry do
    synchronized { @pgmq.transaction(&block) }
  end
end