Module: Pgbus::Client::FairRead

Included in:
Pgbus::Client
Defined in:
lib/pgbus/client/fair_read.rb

Overview

Fair share reads (issue #426): a weighted, work-conserving interleave across fair-share keys within one queue. Replaces read_batch on the worker when config.fair_share is set.

Scheduling rule per read of qty: for every key that has visible messages, rank that key's visible messages 1..qty oldest-visible first (vt, msg_id); a message's virtual time is rank / weight; take the qty messages with the lowest (virtual_time, msg_id). Weight 3 vs 1 yields a 3:1 split under contention; a lone key fills the whole batch. This is batch-level weighted fair queuing — proportional within each batch, memoryless across batches (no deficit carry-over).

Cost model: key enumeration is a loose index scan over (key, vt, msg_id) — one probe per key with visible work; keys whose messages are all invisible (in flight / delayed / in retry backoff) are skipped at the index level because vt is in the index. Per-key candidates are a bounded LIMIT qty index range. Roughly O(K · log n + K · qty), K = keys with visible work, independent of backlog depth. Within a key the order is (vt, msg_id) — FIFO for immediate enqueues; retried/delayed messages sort by when they became visible — a deliberate deviation from pgmq.read's pure msg_id order so the per-key lookup never sorts a tenant's whole visible backlog.

Candidates are selected before locking (same shape as pgmq.read_grouped_rr), so under concurrent readers a batch can come back short; the worker loop re-reads immediately while it has capacity.

Constant Summary collapse

FAIR_INDEX_KEY_EXPR =

The read expressions are written against the alias m; the index expression is the same shape minus the alias so the planner matches it.

"COALESCE(message->>'#{FairShare::METADATA_KEY}', '')".freeze
FAIR_KEY_EXPR = 
"COALESCE(m.message->>'#{FairShare::METADATA_KEY}', '')".freeze
FAIR_WEIGHT_EXPR = 
"COALESCE((m.message->>'#{FairShare::WEIGHT_KEY}')::numeric, 1)".freeze
FAIR_INDEX_SUFFIX = 
"_fair_idx"

Instance Method Summary collapse

Instance Method Details

#ensure_fair_index(queue_name) ⇒ Object

Idempotent, memoized per process. Uses CREATE INDEX CONCURRENTLY so enabling fair share on an existing, populated queue never blocks enqueues. A queue table that does not exist yet is left alone (its creation path builds the index non-concurrently); any other failure is logged with the remediation and NOT memoized so the next ensure retries.



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/pgbus/client/fair_read.rb', line 49

def ensure_fair_index(queue_name)
  full_name = fair_queue_name(queue_name)
  return if fair_indexes_ensured[full_name]

  with_stale_connection_retry do
    synchronized { exec_ddl(fair_index_sql(full_name, concurrently: true)) }
  end
  fair_indexes_ensured[full_name] = true
rescue StandardError => e
  if duplicate_relation_error?(e)
    fair_indexes_ensured[full_name] = true
  elsif undefined_table_error?(e)
    Pgbus.logger.debug { "[Pgbus] Fair index deferred — queue table #{full_name} not created yet" }
  else
    Pgbus.logger.error do
      "[Pgbus] Could not create fair index #{fair_index_name(full_name)} on pgmq.q_#{full_name}: " \
        "#{e.class}: #{e.message}. A failed CONCURRENTLY build leaves an INVALID index behind — " \
        "run `DROP INDEX IF EXISTS pgmq.#{fair_index_name(full_name)}` and restart the worker to retry."
    end
  end
end

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



39
40
41
42
 
# File 'lib/pgbus/client/fair_read.rb', line 39

def read_batch_fair(queue_name, qty:, vt: nil)
  full_name = fair_queue_name(queue_name)
  guarded_read { fair_read_step(full_name, qty, vt || config.visibility_timeout) }
end