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.



79
80
81
 
# File 'app/models/pgbus/stream_queue.rb', line 79

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.



111
112
113
114
115
116
117
118
119
120
 
# File 'app/models/pgbus/stream_queue.rb', line 111

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.



94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'app/models/pgbus/stream_queue.rb', line 94

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.



87
88
89
 
# File 'app/models/pgbus/stream_queue.rb', line 87

def known_names
  all_names | fingerprint_matched_names
end

.record!(queue_name) ⇒ Object

Inserts 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 insert failed (so callers like backfill! can report accurately).

Deliberately raw SQL rather than upsert(unique_by:) (issue #401): Rails resolves unique_by: through the pool's schema cache, which caches a negative data_source_exists? probe permanently — one wrong first probe (while the table genuinely exists and the live table_exists? guard above passes) poisons every subsequent record! in the process with "No unique index found". The unique index is owned by this gem's own migration, so there is nothing to resolve.



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'app/models/pgbus/stream_queue.rb', line 48

def record!(queue_name)
  return false unless table_exists?

  conn = connection
  conn.execute(
    "INSERT INTO #{conn.quote_table_name(table_name)} (queue_name) " \
    "VALUES (#{conn.quote(queue_name)}) ON CONFLICT (queue_name) DO NOTHING"
  )
  # 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 insert above has committed.
  @all_names&.add(queue_name)
  true
rescue ActiveRecord::ActiveRecordError => e
  Pgbus.logger.debug { "[Pgbus] Failed to record stream queue #{queue_name}: #{e.message}" }
  false
rescue StandardError => e
  log_record_failure(queue_name, e)
  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.



130
131
132
 
# File 'app/models/pgbus/stream_queue.rb', line 130

def reset_cache!
  @all_names = nil
end

.stream?(queue_name) ⇒ Boolean

Returns:

  • (Boolean) 


122
123
124
 
# File 'app/models/pgbus/stream_queue.rb', line 122

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) 


136
137
138
139
140
141
142
143
 
# File 'app/models/pgbus/stream_queue.rb', line 136

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