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'

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::LEADER_CACHE_TTL_MS, Component::PROCESS_NONCE

Instance Attribute Summary

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, hostname, #hostname, identity, #identity, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, tid, #tid, #watchdog

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 Method Details

#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

#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

#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

#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