Class: Pgbus::StreamQueue

Inherits:
BusRecord
  • Object
show all
Defined in:
app/models/pgbus/stream_queue.rb

Overview

Registry of physical PGMQ queue names that back streams (as opposed to job queues). Stream and job queues share the same #{queue_prefix}_ namespace (see Configuration#queue_name), so there is no reliable way to tell them apart by name alone. ensure_stream_queue records each stream queue here on first broadcast/subscribe; consumers that iterate pgmq.meta (the dispatcher's stream-archive prune and orphan sweep, compact_archives, and the worker's wildcard resolution) use this registry to classify a queue as a stream.

Degrades safely: on an install that has not run the migration, the table is absent — record! no-ops and all_names returns an empty set, so the pre-registry behavior (streams treated as job queues by maintenance) is preserved rather than raising.

Pre-registry dormant streams (issue #366): queues created before this table existed never re-broadcast, so they never call record!. Those queues still carry the archive a_<name>_msg_id_idx index that only ensure_stream_queue creates — fingerprint_matched_names discovers them, known_names unions them with the registry for classification, and backfill! (via rake pgbus:streams:backfill_registry) writes the missing rows so the registry is complete.

Class Method Summary collapse

Class Method Details

.all_names ⇒ Object

Set of all registered physical stream queue names. Memoized so a dispatcher maintenance pass or a wildcard re-resolve does one query, not one per queue. Callers that need freshness (a long-lived process picking up newly-created streams) call reset_cache! at the top of their loop.



59
60
61
 
# File 'app/models/pgbus/stream_queue.rb', line 59

def all_names
  @all_names ||= load_names
end

.backfill! ⇒ Object

Registers fingerprint-matched queues missing from the registry. Idempotent. Returns the number of names successfully persisted (failed upserts are not counted — see record!). No-ops (returns 0) when the registry table is absent.



91
92
93
94
95
96
97
98
99
100
 
# File 'app/models/pgbus/stream_queue.rb', line 91

def backfill!
  return 0 unless table_exists?

  missing = fingerprint_matched_names - all_names
  return 0 if missing.empty?

  registered = missing.count { |name| record!(name) }
  reset_cache! if registered.positive?
  registered
end

.fingerprint_matched_names ⇒ Object

Physical queue names that carry the stream-only archive msg_id index (a_<queue>_msg_id_idx). Job queues never get this index. Safe to call when pgmq is absent — returns an empty set rather than raising.



74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'app/models/pgbus/stream_queue.rb', line 74

def fingerprint_matched_names
  Set.new(connection.select_values(<<~SQL.squish))
    SELECT m.queue_name
    FROM pgmq.meta m
    INNER JOIN pg_indexes i
      ON i.schemaname = 'pgmq'
     AND i.indexname = 'a_' || m.queue_name || '_msg_id_idx'
  SQL
rescue StandardError => e
  Pgbus.logger.debug { "[Pgbus] Failed to discover stream queue fingerprints: #{e.message}" }
  Set.new
end

.known_names ⇒ Object

Registry ∪ fingerprint-matched names. Use this for classification (health verdict exclusion, orphan sweep, wildcard worker exclusion) so dormant pre-registry streams are not treated as job queues. Does not write — call backfill! to persist the fingerprint matches.



67
68
69
 
# File 'app/models/pgbus/stream_queue.rb', line 67

def known_names
  all_names | fingerprint_matched_names
end

.record!(queue_name) ⇒ Object

Upserts the physical queue name. Idempotent and cheap to call on every broadcast; the caller (ensure_stream_queue) also memoizes per-process, so the DB write happens once per stream per process. Errors are swallowed — a registry hiccup must never abort a broadcast. Returns true on a successful write, false when the table is absent or the upsert failed (so callers like backfill! can report accurately).



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'app/models/pgbus/stream_queue.rb', line 35

def record!(queue_name)
  return false unless table_exists?

  upsert({ queue_name: queue_name }, unique_by: :queue_name)
  # Keep the in-process cache consistent with the write so a subsequent
  # stream? check reflects this registration without a re-query. Only
  # update an ALREADY-LOADED cache — if @all_names is still nil (this
  # process hasn't queried the registry yet), seeding it here would
  # fabricate a one-entry set and silently hide every other
  # already-registered stream until the next reset_cache!. Leaving it
  # nil lets the next all_names call do a real load, which already
  # includes this row since the upsert above has committed.
  @all_names&.add(queue_name)
  true
rescue StandardError => e
  Pgbus.logger.debug { "[Pgbus] Failed to record stream queue #{queue_name}: #{e.message}" }
  false
end

.reset_cache! ⇒ Object

Drops the memoized set so the next all_names/stream? re-queries. Called at the start of each dispatcher maintenance pass and each wildcard resolution so freshly-created streams are picked up without a process restart.



110
111
112
 
# File 'app/models/pgbus/stream_queue.rb', line 110

def reset_cache!
  @all_names = nil
end

.stream?(queue_name) ⇒ Boolean

Returns:

  • (Boolean) 


102
103
104
 
# File 'app/models/pgbus/stream_queue.rb', line 102

def stream?(queue_name)
  all_names.include?(queue_name)
end

.table_exists? ⇒ Boolean

Memoized like StreamStat: a successful probe sticks; a transient error returns false without caching, so the next call retries.

Returns:

  • (Boolean) 


116
117
118
119
120
121
122
123
 
# File 'app/models/pgbus/stream_queue.rb', line 116

def table_exists?
  return @table_exists if defined?(@table_exists) && @table_exists

  @table_exists = connection.table_exists?(table_name)
rescue StandardError => e
  Pgbus.logger.debug { "[Pgbus] Failed to check stream queue table: #{e.message}" }
  false
end