Module: RoundhouseUi::Pause

Defined in:
lib/roundhouse_ui/pause.rb

Overview

Roundhouse's own queue-pause registry — pure OSS, no Sidekiq Pro.

Paused queue names live in a Redis set. RoundhouseUi::Fetch consults this set and skips paused queues when pulling work, so a paused queue stops being consumed without stopping the worker process.

When Sidekiq Pro is loaded we defer to its registry instead (see .native?). Under Pro we defer to its own pause API (see .native?), so pausing is enforced with no Roundhouse fetcher installed. Pro's key is not ours.

Constant Summary collapse

KEY = 
"roundhouse:paused"
FETCH_FLAG =

liveness beacon set by the fetcher

"roundhouse:fetch_alive"
PRO_KEY =

Sidekiq Pro's own registry

"paused"

Class Method Summary collapse

Class Method Details

.fetch_installed? ⇒ Boolean

True when a RoundhouseUi::Fetch has reported in recently — i.e. pausing will take effect. When false, the UI warns instead of pretending.

Under Pro no beacon is needed: any Pro worker enforces pauses whether or not our fetcher is installed.

Returns:

  • (Boolean) 


83
84
85
86
87
 
# File 'lib/roundhouse_ui/pause.rb', line 83

def fetch_installed?
  return true if native?

  Sidekiq.redis { |conn| conn.call("EXISTS", FETCH_FLAG) } == 1
end

.key ⇒ Object

Which registry reads come from. Reads are plain set lookups in both cases (no pubsub involved), so they can share one implementation.



57
58
59
 
# File 'lib/roundhouse_ui/pause.rb', line 57

def key
  native? ? PRO_KEY : KEY
end

.mark_fetch_alive!(ttl = 30) ⇒ Object

The fetcher calls this periodically so the web UI can tell whether pausing is actually enforced (the worker and web run in separate processes). The short TTL means the flag disappears soon after all Roundhouse fetchers stop.



74
75
76
 
# File 'lib/roundhouse_ui/pause.rb', line 74

def mark_fetch_alive!(ttl = 30)
  Sidekiq.redis { |conn| conn.call("SET", FETCH_FLAG, "1", "EX", ttl) }
end

.native? ⇒ Boolean

True when Sidekiq Pro's queue-pause API is available. Feature-detected on the method rather than defined?(Sidekiq::Pro) so it tracks the actual capability across Pro versions. Cheap (no Redis), so it isn't memoized — loading Pro mid-process would otherwise be missed.

Returns:

  • (Boolean) 


25
26
27
 
# File 'lib/roundhouse_ui/pause.rb', line 25

def native?
  defined?(::Sidekiq::Queue) && ::Sidekiq::Queue.method_defined?(:pause!)
end

.pause!(queue) ⇒ Object

Under Pro, always go through Sidekiq::Queue#pause! — writing PRO_KEY directly does not reach already-running workers.



31
32
33
34
35
 
# File 'lib/roundhouse_ui/pause.rb', line 31

def pause!(queue)
  return ::Sidekiq::Queue.new(queue.to_s).pause! if native?

  Sidekiq.redis { |conn| conn.call("SADD", KEY, queue.to_s) }
end

.paused?(queue) ⇒ Boolean

Returns:

  • (Boolean) 


43
44
45
 
# File 'lib/roundhouse_ui/pause.rb', line 43

def paused?(queue)
  Sidekiq.redis { |conn| conn.call("SISMEMBER", key, queue.to_s) } == 1
end

.paused_queues ⇒ Object 



47
48
49
 
# File 'lib/roundhouse_ui/pause.rb', line 47

def paused_queues
  Sidekiq.redis { |conn| conn.call("SMEMBERS", key) }.sort
end

.paused_set ⇒ Object 



51
52
53
 
# File 'lib/roundhouse_ui/pause.rb', line 51

def paused_set
  Set.new(Sidekiq.redis { |conn| conn.call("SMEMBERS", key) })
end

.reject_paused(queue_keys) ⇒ Object

Given the redis queue keys BasicFetch would poll (e.g. "queue:default"), drop any whose queue is paused. Pure given the paused set, so it's unit testable without a running Sidekiq.



64
65
66
67
68
69
 
# File 'lib/roundhouse_ui/pause.rb', line 64

def reject_paused(queue_keys)
  paused = paused_set
  return queue_keys if paused.empty?

  queue_keys.reject { |key| paused.include?(key.to_s.delete_prefix("queue:")) }
end

.unpause!(queue) ⇒ Object 



37
38
39
40
41
 
# File 'lib/roundhouse_ui/pause.rb', line 37

def unpause!(queue)
  return ::Sidekiq::Queue.new(queue.to_s).unpause! if native?

  Sidekiq.redis { |conn| conn.call("SREM", KEY, queue.to_s) }
end