Class: Wurk::QueueConsumers::Watch

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/queue_consumers.rb

Overview

Warns about non-empty queues no live process subscribes to: such jobs wait forever with no error, retry or report. Leader only, one tick a minute.

A queue is reported only after grace seconds unconsumed on every tick. A rolling restart or a deploy that adds a queue passes through that state briefly, and one leader's in-memory clock is enough to ride it out. A leader change restarts the clock, which only delays a warning.

Constant Summary collapse

TICK = 
60
GRACE = 
300
THROTTLE = 
3600

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, tick: TICK, grace: GRACE, throttle: THROTTLE, clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) }) ⇒ Watch

Returns a new instance of Watch.



55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/wurk/queue_consumers.rb', line 55

def initialize(config, tick: TICK, grace: GRACE, throttle: THROTTLE,
               clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) })
  @config = config
  @grace = grace
  @throttle = throttle
  @clock = clock
  @timer = TimerLoop.new(tick)
  @unconsumed_since = {}
  @warned_at = {}
  @thread = nil
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



53
54
55
 
# File 'lib/wurk/queue_consumers.rb', line 53

def config
  @config
end

#config ⇒ Object (readonly) Originally defined in module Component

Returns the value of attribute config.

Instance Method Details

#check ⇒ Object

One pass, bypassing the leader gate. Returns the queues warned about.



88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/wurk/queue_consumers.rb', line 88

def check
  now = @clock.call
  unconsumed = read_unconsumed
  forget_settled(unconsumed, now)
  unconsumed.filter_map do |queue, size|
    @unconsumed_since[queue] ||= now
    next unless warn_due?(queue, now)

    @warned_at[queue] = now
    logger.warn { warning(queue, size) }
    queue
  end
end

#default_tag(dir = Dir.pwd) ⇒ Object Originally defined in module Component

#fire_event(event, oneshot: true, reverse: false, reraise: false) ⇒ Object Originally defined in module Component

Invokes lifecycle hooks for event. Hooks run in registration order (or LIFO when reverse: true, used for teardown). A raise in one hook is reported via handle_exception and does NOT stop the next hook unless reraise: true (used in tests / fail-fast boot). oneshot: true clears the bucket after dispatch so the event can't fire twice.

#handle_exception(ex, ctx = {}) ⇒ Object Originally defined in module Component

#hostname ⇒ Object Originally defined in module Component

#identity ⇒ Object Originally defined in module Component

#leader? ⇒ Boolean Originally defined in module Component

True iff this process currently holds the cluster dear-leader lock. Cached per Component instance for LEADER_CACHE_TTL_MS (~5s): cron and the metrics rollups call this every tick, and an uncached GET would double their Redis traffic at short intervals for no benefit — the lock's own renewal cadence (60s+, spec §6.1) easily tolerates a few-second-stale read. Returns false unconditionally when WURK_LEADER=false (or SIDEKIQ_LEADER=false) is set on the process (opt-out hot-standby). Any Redis error is swallowed → false, so a transient partition can't propagate as an exception into user code.

Spec: docs/target/sidekiq-ent.md §6.1.

Returns:

  • (Boolean)

#logger ⇒ Object Originally defined in module Component

--- delegated to config -------------------------------------------

#mono_ms ⇒ Object Originally defined in module Component

#process_nonce ⇒ Object Originally defined in module Component

#real_ms ⇒ Object Originally defined in module Component

--- clocks ---------------------------------------------------------

#redis(idempotent: false) ⇒ Object Originally defined in module Component

#safe_thread(name, priority: nil, &block) ⇒ Object Originally defined in module Component

Spawns a named thread that runs block under watchdog(name). The parent must retain the returned Thread; otherwise GC may not, but report_on_exception is disabled so we don't double-log on death.

Priority resolution matches Sidekiq (component.rb:44-48): explicit argument, then config.thread_priority, then -1. Ruby's default of 0 buys a 100ms timeslice; each negative step halves it, so -1 keeps a CPU-heavy capsule from starving its siblings for a whole tick.

#start ⇒ Object 



67
68
69
70
71
72
 
# File 'lib/wurk/queue_consumers.rb', line 67

def start
  return @thread if @thread

  @timer.reset
  @thread = safe_thread('queue-consumers') { @timer.run { tick } }
end

#terminate ⇒ Object 



74
75
76
77
 
# File 'lib/wurk/queue_consumers.rb', line 74

def terminate
  @timer.terminate
  @thread = nil if @thread&.join(TimerLoop::JOIN_TIMEOUT)
end

#tick ⇒ Object 



79
80
81
82
83
84
85
 
# File 'lib/wurk/queue_consumers.rb', line 79

def tick
  return unless leader?

  check
rescue StandardError => e
  handle_exception(e, { context: 'queue-consumers' })
end

#tid ⇒ Object Originally defined in module Component

#watchdog(last_words) ⇒ Object Originally defined in module Component

Wraps a block at a thread boundary: any unhandled exception is reported via handle_exception (so it lands in error_handlers / the log) and then re-raised. last_words is the component label included in the context.