Module: Pgbus::Process::ConsumerPriority

Defined in:
lib/pgbus/process/consumer_priority.rb

Overview

Implements consumer priority by checking whether higher-priority workers are active for the same queues. When a higher-priority worker is healthy and not at its prefetch limit, lower-priority workers yield by using a longer polling interval.

Inspired by LavinMQ's consumer priority where higher-priority consumers are served first and lower-priority consumers wait until all higher-priority consumers are at their prefetch limit.

Constant Summary collapse

CACHE_TTL =

Time-to-live (seconds) for cached max_active_priority lookups. The underlying data only changes on heartbeat cadence, so a short TTL avoids ~10 queries/second per prioritized worker at the default 0.1s polling interval.

5

Class Method Summary collapse

Class Method Details

.effective_polling_interval(base_interval:, my_priority:, max_priority:) ⇒ Object

Calculate the effective polling interval for this worker. Higher-priority workers use the base interval. Lower-priority workers multiply by a backoff factor.



90
91
92
93
94
95
 
# File 'lib/pgbus/process/consumer_priority.rb', line 90

def self.effective_polling_interval(base_interval:, my_priority:, max_priority:)
  return base_interval if my_priority >= max_priority

  # Lower-priority workers back off: 3x the base interval
  base_interval * 3
end

.max_active_priority(queues, my_pid) ⇒ Object

Returns the highest consumer_priority among healthy workers that share at least one queue with the given queue list, excluding the current worker (by PID).

Results are cached per (sorted queues, pid) for CACHE_TTL seconds. On a miss the query runs outside the lock (a duplicate query on a race is harmless), then the value is stored under the mutex.



42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/pgbus/process/consumer_priority.rb', line 42

def self.max_active_priority(queues, my_pid)
  key = [queues.sort, my_pid]
  now = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)

  cached = @cache_mutex.synchronize { @cache[key] }
  return cached[:value] if cached && (now - cached[:at]) < CACHE_TTL

  value = compute_max_active_priority(queues, my_pid)
  @cache_mutex.synchronize { @cache[key] = { value: value, at: now } }
  value
end

.reset_cache! ⇒ Object

Clears all cached max_active_priority entries. Intended for spec isolation; also safe to call at runtime.



56
57
58
 
# File 'lib/pgbus/process/consumer_priority.rb', line 56

def self.reset_cache!
  @cache_mutex.synchronize { @cache.clear }
end

.should_yield?(queues:, my_priority:, my_pid:) ⇒ Boolean

Check if this worker should yield to a higher-priority worker. Returns true if a higher-priority healthy worker exists for any of the given queues.

Returns:

  • (Boolean) 


26
27
28
29
30
31
32
33
 
# File 'lib/pgbus/process/consumer_priority.rb', line 26

def self.should_yield?(queues:, my_priority:, my_pid:)
  return false if my_priority >= max_active_priority(queues, my_pid)

  true
rescue StandardError => e
  Pgbus.logger.debug { "[Pgbus] Consumer priority check failed: #{e.message}" }
  false
end