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?). Pro reopens Sidekiq::Queue with pause!/unpause!/paused? and prepends pause support onto Sidekiq::BasicFetch, so pausing is already enforced with no Roundhouse fetcher installed — and Pro's key ("paused") is not ours ("roundhouse:paused"), so writing our own set there would do nothing.

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: Pro prepends pause support onto Sidekiq::BasicFetch (and SuperFetch honors it too), so any Pro worker enforces pauses whether or not our fetcher is installed.

Returns:

  • (Boolean) 


88
89
90
91
92
 
# File 'lib/roundhouse_ui/pause.rb', line 88

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.



61
62
63
 
# File 'lib/roundhouse_ui/pause.rb', line 61

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.



78
79
80
 
# File 'lib/roundhouse_ui/pause.rb', line 78

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) 


27
28
29
 
# File 'lib/roundhouse_ui/pause.rb', line 27

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

.pause!(queue) ⇒ Object

Under Pro, go through Sidekiq::Queue#pause! rather than writing PRO_KEY ourselves: Pro's fetchers read that set once at startup and afterwards only update on the "pro:config" pubsub message that pause! publishes. A bare SADD would leave running workers pulling the queue until they restarted.



35
36
37
38
39
 
# File 'lib/roundhouse_ui/pause.rb', line 35

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) 


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

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

.paused_queues ⇒ Object 



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

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

.paused_set ⇒ Object 



55
56
57
 
# File 'lib/roundhouse_ui/pause.rb', line 55

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.



68
69
70
71
72
73
 
# File 'lib/roundhouse_ui/pause.rb', line 68

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 



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

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

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