Class: Wurk::Watchdog

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

Overview

One timer thread per Capsule — so one per Manager — that cuts jobs which outlive a bound. #watch arms a bound around a block; once it passes, this thread raises the caller's exception into the thread that armed it. Soft by construction: an exception the job can rescue and the retry layer can book, never Thread#kill.

Not stdlib Timeout, which since Ruby 3.1 also runs one shared monitor thread — that half of the folklore is out of date. What it still doesn't do is contain its own raise: Timeout::Request#interrupt calls Thread#raise with no interrupt mask, so a bound that wins the race against the block returning lands at whatever checkpoint the thread reaches next — inside a Processor, the ACK or the next job. Its own docs tell every caller to hand-roll the handle_interrupt sandwich that avoids this; #watch writes it once, for every bounded job. It is also ~1.5x cheaper per call (08-watchdog-measurement.md in the slice plan) and accepts a bound that has already passed, which Timeout.timeout rejects outright — an absolute deadline read off an old payload is exactly that.

Nothing bounded, nothing running: the thread is spawned by the first #watch, so a process whose jobs declare no bound never has one. The Capsule holds this like it holds the fetcher; the Manager drives its lifecycle (Manager#stop, not #quiet — a quieted process still has in-flight jobs, and a bound shorter than the drain budget must still fire).

Constant Summary collapse

SCAN_INTERVAL =

Scan cadence, and therefore the overshoot: a bound fires somewhere in [deadline, deadline + SCAN_INTERVAL). Job bounds are wall-clock seconds, so 500ms of slack is invisible; waking at the nearest deadline instead would buy precision nobody asked for and cost a signal on every arm.

0.5
THREAD_NAME = 
'watchdog'

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(capsule, interval: SCAN_INTERVAL) ⇒ Watchdog

Returns a new instance of Watchdog.



44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/wurk/watchdog.rb', line 44

def initialize(capsule, interval: SCAN_INTERVAL)
  @config = @capsule = capsule
  @timer = TimerLoop.new(interval)
  @lock = ::Mutex.new
  # Keyed by a counter, not by Thread or by the Entry itself: one job can
  # hold two bounds at once (a per-attempt `timeout` inside an absolute
  # `deadline`), and Struct identity is value equality, so twin entries
  # would collide.
  @armed = {}
  @seq = 0
  @thread = nil
end

Instance Attribute Details

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

Returns the value of attribute config.

Instance Method Details

#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

Capsule doesn't define handle_exception (it's a Configuration method); override Component's delegation so error handlers fire.



113
114
115
 
# File 'lib/wurk/watchdog.rb', line 113

def handle_exception(ex, ctx = {})
  @capsule.config.handle_exception(ex, ctx)
end

#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

#running? ⇒ Boolean

The two assertable halves of "zero cost when unconfigured, nothing leaked when used": no bound was ever armed → no thread; every armed bound is retracted on every exit path → nothing accumulates.

Returns:

  • (Boolean) 


102
103
104
105
 
# File 'lib/wurk/watchdog.rb', line 102

def running?
  thread = @lock.synchronize { @thread }
  !thread.nil? && thread.alive?
end

#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.

#size ⇒ Object 



107
108
109
 
# File 'lib/wurk/watchdog.rb', line 107

def size
  @lock.synchronize { @armed.size }
end

#terminate ⇒ Object

Idempotent, and a no-op when nothing was ever armed. Bounded join like every other periodic component: a scan blocked on a raise must not hold the process's shutdown open. The thread reference is deliberately kept on a join timeout — a wedged scan stays tracked, so a later #watch returns it rather than spawning a second one alongside it.



94
95
96
97
 
# File 'lib/wurk/watchdog.rb', line 94

def terminate
  @timer.terminate
  @lock.synchronize { @thread }&.join(TimerLoop::JOIN_TIMEOUT)
end

#tid ⇒ Object Originally defined in module Component

#watch(seconds, exception, message = nil) ⇒ Object

Runs the block with seconds of wall-clock on it. Still on this thread's stack when that runs out → exception (a class, with an optional message) is raised into it. A bound that already passed is not special-cased; it fires on the next scan.

The interrupt pair is the containment guarantee, and the whole reason this exists instead of Timeout.timeout. :never on the outer scope means a raise that wins the race against #disarm is delivered when that scope pops — inside this method — instead of at whatever checkpoint the thread reaches next, which by then can be the ACK, or the next job. :immediate on the inner scope is what lets a wedged job be interrupted at all; without it the outer mask would hold the raise until the block it is meant to cut short returned on its own.

The mask only contains a raise that is issued while this frame is on the stack; #tick is what guarantees that, by taking the same @lock #disarm takes and holding it across the raise.



74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/wurk/watchdog.rb', line 74

def watch(seconds, exception, message = nil)
  bound_id = arm(seconds, exception, message)
  Thread.handle_interrupt(exception => :never) do
    Thread.handle_interrupt(exception => :immediate) { yield } # rubocop:disable Style/ExplicitBlockArgument
  ensure
    # Masked against everything, not just `exception`: a job holding two
    # bounds runs this inside the other one's `:immediate` scope, and a raise
    # landing mid-retraction would strand this entry — still armed, no longer
    # retractable, free to fire into whatever the thread picks up next.
    # `:never` defers rather than discards, so the other bound is still
    # delivered, one frame later.
    Thread.handle_interrupt(::Object => :never) { disarm(bound_id) }
  end
end

#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.