Class: Wurk::Fetcher::Reaper

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

Overview

Orphan reclamation for the reliable fetcher (Pro super_fetch §3.2).

The Reliable fetcher moves each job from a public queue into a per-process private list (queue:<public>|<host>|<pid>|<nonce>|<idx>) and leaves it there until the Processor ACKs. A SIGKILLed or crashed worker therefore strands its in-flight jobs in private lists that nobody will ever ACK. The Reaper is the recovery half: it periodically scans for private lists whose owning process is gone and atomically moves their jobs back to the public queue so a live worker re-runs them.

Liveness is decided per owner:

* our own host and nonce — the pid was minted in our PID namespace, so
the OS is authoritative: `Process.kill(0, pid)`. This is instant and
ignores a stale `processes` SET entry whose 60s TTL hasn't lapsed
yet, so a `kill -9`ed sibling is reclaimed the moment the supervisor
reaps it rather than 60s later. (Pid reuse by an unrelated process in
the same tree is the one blind spot — the supervisor respawns with a
fresh pid, so it does not arise in practice.)
* any other incarnation — its pid means nothing in our namespace, so we
trust the heartbeat: the owner is alive iff its identity is a live
`processes` member (one whose `info` hash still exists). Such reclaim
therefore waits out the 60s heartbeat TTL, exactly as the spec says.

Re-pushed jobs run through Wurk::Middleware::PoisonPill, which caps a job at RECOVERY_THRESHOLD recoveries within 72h: past the cap the job is killed into the dead set instead of re-queued, so a job that crashes its worker every time can't loop forever.

The reaper runs two passes, exactly as super_fetch's sweeper does:

* a *scoped* sweep every interval ("1/min within process group"): SCANs
only the public queues this process serves, gated by a cluster `SET NX
EX` lock so one process sweeps per interval. The cheap common path.
* a *full* sweep at most once an hour ("full SCAN 1/hr"): SCANs the whole
`queue:*|*` keyspace, gated by its own hourly lock, so private lists
whose public queue no live process serves — a renamed/decommissioned
queue, or a dead host's queue no survivor consumes — are recovered too,
not stranded forever.

Spec: docs/target/sidekiq-pro.md §3.2.

Constant Summary collapse

DEFAULT_INTERVAL =

Sweep cadence in seconds; also the cluster-lock TTL so exactly one process sweeps per interval. 60s matches the heartbeat TTL — the floor below which cross-host orphans can't be detected anyway.

60
FULL_INTERVAL =

Full-keyspace sweep cadence + its lock TTL: at most once per hour across the fleet, since a global SCAN is far costlier than the scoped pass.

3600
LOCK_KEY = 
'super_fetch:reaper'
FULL_LOCK_KEY = 
'super_fetch:reaper:full'
SCAN_COUNT = 
100
THREAD_NAME = 
'wurk-reaper'

Constants included from Component

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

Instance Attribute Summary collapse

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

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

Constructor Details

#initialize(config, interval: DEFAULT_INTERVAL, lock_key: LOCK_KEY, full_interval: FULL_INTERVAL, full_lock_key: FULL_LOCK_KEY) ⇒ Reaper

Returns a new instance of Reaper.



69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/wurk/fetcher/reaper.rb', line 69

def initialize(config, interval: DEFAULT_INTERVAL, lock_key: LOCK_KEY,
               full_interval: FULL_INTERVAL, full_lock_key: FULL_LOCK_KEY)
  @config = config
  @interval = interval
  @lock_key = lock_key
  @full_interval = full_interval
  @full_lock_key = full_lock_key
  @thread = nil
  @done = false
  @mutex = ::Mutex.new
  @sleeper = ::ConditionVariable.new
end

Instance Attribute Details

#interval ⇒ Object (readonly)

Returns the value of attribute interval.



67
68
69
 
# File 'lib/wurk/fetcher/reaper.rb', line 67

def interval
  @interval
end

Instance Method Details

#reap ⇒ Object

One loop tick: the scoped sweep when this process wins the per-interval lock, plus the full-keyspace sweep when it also wins the hourly lock. Returns the total jobs reclaimed across both.



120
121
122
123
124
 
# File 'lib/wurk/fetcher/reaper.rb', line 120

def reap
  reclaimed = acquire_lock? ? reclaim! : 0
  reclaimed += reclaim_full! if acquire_full_lock?
  reclaimed
end

#reclaim! ⇒ Object

One unguarded sweep over every served queue. Returns the number of jobs reclaimed (re-queued or killed). Public so boot paths and tests can drive a deterministic pass without the cluster lock.



129
130
131
132
 
# File 'lib/wurk/fetcher/reaper.rb', line 129

def reclaim!
  owners = live_owners
  served_queues.sum { |public_q| reclaim_queue(public_q, owners) }
end

#reclaim_full! ⇒ Object

One unguarded full-keyspace sweep: every queue:*|* private list, even ones whose public queue this process doesn't serve. Returns the number of jobs reclaimed. Public so boot paths and tests can drive it without the hourly lock.



138
139
140
141
142
143
144
145
146
147
 
# File 'lib/wurk/fetcher/reaper.rb', line 138

def reclaim_full!
  owners = live_owners
  reclaimed = 0
  each_full_private_list do |key, public_q, host, pid, nonce|
    next if owner_alive?(host, pid, nonce, owners)

    reclaimed += drain(key, public_q)
  end
  reclaimed
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


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

def running?
  !@thread.nil? && @thread.alive?
end

#start ⇒ Object

Spawns the sweep loop. Idempotent. The loop waits one interval before its first sweep so booting processes don't dogpile Redis and so an un-stopped launcher in a unit test never touches the keyspace.



85
86
87
88
89
90
91
92
93
 
# File 'lib/wurk/fetcher/reaper.rb', line 85

def start
  @mutex.synchronize do
    return @thread if @thread

    @done = false
    @thread = spawn_loop_thread
  end
  @thread
end

#stop ⇒ Object

Bounded at TimerLoop::JOIN_TIMEOUT like every other periodic component. The full-keyspace sweep is exactly the tick that outlasts a stop: it SCANs the whole queue:*|* keyspace and drains what it finds, so on a large or slow Redis it can still be running when shutdown lands. Waiting it out held the entire process's teardown open past the swarm parent's SHUTDOWN_GRACE — which SIGKILLs the child mid-drain, the one outcome this component exists to recover from. A straggler is left running instead — and still referenced, like every other periodic component, so a restart after a timed-out stop can't spawn a second sweep loop alongside it. Its own tick_once rescues and reports.



105
106
107
108
109
110
111
 
# File 'lib/wurk/fetcher/reaper.rb', line 105

def stop
  @mutex.synchronize do
    @done = true
    @sleeper.signal
  end
  @thread = nil if @thread&.join(TimerLoop::JOIN_TIMEOUT)
end