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 member — the SET carries no TTL, so
the member lingers after its heartbeat hash expires (60s) until
`ProcessSet#cleanup` prunes it. A `kill -9`ed sibling is reclaimed
the moment the supervisor reaps it, not 60s later. Pid reuse by an
unrelated process reads a dead owner as live; a swarm child does not
depend on this check, since its parent drains its lists by exact key
the moment it reaps it (#reclaim_exited), before the pid can be reused.
* any other incarnation — its pid means nothing in our namespace, so we
trust the heartbeat: the owner is alive iff its identity's `info`
hash still exists. Such reclaim waits out the 60s heartbeat TTL, the
next sweep and then the grace: 80-150s after the owner died.

A missing hash is not proof of death on its own. The verdict is a snapshot taken before a SCAN that can run for a long time on a big keyspace, a beat can be held back by CPU-bound jobs, and a Redis restarted from an older snapshot brings back the processes SET with every hash lapsed. So a heartbeat-judged orphan must stay orphaned for grace seconds on this process's own clock (OrphanGrace), and its identity must still have no info hash. Our own incarnation skips both: kill(0) is not a snapshot.

Every move goes through OrphanReclaim, which caps a job at Middleware::PoisonPill::RECOVERY_THRESHOLD recoveries within 72h: past the cap the job goes from the private list straight to the dead set instead of its queue, so a job that crashes its worker every time can't loop forever.

One sweep an interval: one SCAN of queue:*|* acting on every list, gated by a cluster lock so about one process sweeps per interval. Every list, and not only those this process serves: the lock is shared across the fleet, so a scoped sweep would wait for its turn to land on one of the processes serving the queue. The list's own queue is where its jobs go back to, whichever process moves them.

Every tick also runs SelfReconcile over this process's own lists, which is the one case liveness cannot answer: the owner is alive and has lost track of a job.

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

Constant Summary collapse

DEFAULT_INTERVAL =

Sweep cadence in seconds. 60s matches the heartbeat TTL — the floor below which cross-host orphans can't be detected anyway.

60
LOCK_TTL_RATIO =

The lock lapses before the winner's next tick, or a sole survivor would still find its own lock there and sweep every second interval.

0.8
DEFAULT_GRACE =

Seconds a heartbeat-judged private list stays orphaned before it is drained. A beat is due every 10s, and 25 CPU-bound jobs were measured holding one back 17.4s: at a 15s grace one trial in nine drained all 25 live jobs. The price is paid only by a host lost for good.

30
LOCK_KEY = 
'super_fetch:reaper'
SCAN_COUNT =

A sweep walks the whole keyspace, where almost nothing matches: COUNT is the work per SCAN call, so a bigger batch is the same server work in a tenth of the round trips — a million keys is 1,000 calls instead of 10,000. Still small enough that one call never holds Redis for more than a fraction of a millisecond.

1000
THREAD_NAME = 
'wurk-reaper'

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, interval: DEFAULT_INTERVAL, lock_key: LOCK_KEY, grace: DEFAULT_GRACE, reconcile_confirm: SelfReconcile::DEFAULT_CONFIRM) ⇒ Reaper

Returns a new instance of Reaper.



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/wurk/fetcher/reaper.rb', line 91

def initialize(config, interval: DEFAULT_INTERVAL, lock_key: LOCK_KEY, grace: DEFAULT_GRACE,
               reconcile_confirm: SelfReconcile::DEFAULT_CONFIRM)
  @config = config
  @interval = interval
  @grace = OrphanGrace.new(grace)
  @lock_key = lock_key
  @thread = nil
  @done = false
  @mutex = ::Mutex.new
  @sleeper = ::ConditionVariable.new
  @unparseable = UnparseableKeys.new(config)
  @next_sweep_at = 0.0
  @orphan_reclaim = OrphanReclaim.new(config)
  @self_reconcile = SelfReconcile.new(config, confirm: reconcile_confirm, stop: -> { done? })
end

Instance Attribute Details

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

Returns the value of attribute config.

#interval ⇒ Object (readonly)

Returns the value of attribute interval.



89
90
91
 
# File 'lib/wurk/fetcher/reaper.rb', line 89

def interval
  @interval
end

Instance Method Details

#boot_reclaim! ⇒ Object

The sweep a booting process runs, under the same lock as the periodic one: a rolling restart or a deploy of N processes is one keyspace pass, not N. A replacement's sweep is no longer what recovers the swarm child it replaces; the parent did that by exact key when it reaped the child (#reclaim_exited). What this pass still finds is other incarnations' lists, which wait on the heartbeat TTL and the grace either way.



179
180
181
 
# File 'lib/wurk/fetcher/reaper.rb', line 179

def boot_reclaim!
  reap
end

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

#drain_settled ⇒ Object

Lists whose grace has run out, whichever sweep started it. Runs on every tick with no lock: the sweep lock rotates between processes, and waiting to win it again is what made reclaim slower as a fleet grew. Two processes draining one list is safe, each move takes one job once.



203
204
205
206
207
208
209
 
# File 'lib/wurk/fetcher/reaper.rb', line 203

def drain_settled
  @grace.take_settled.sum do |key, (public_q, host, pid, nonce)|
    next 0 if done?

    owner_gone?(host, pid, nonce) ? drain(key, public_q) : 0
  end
end

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

#reap ⇒ Object

One loop tick: a sweep when this process wins the per-interval lock. Returns the jobs reclaimed.



156
157
158
 
# File 'lib/wurk/fetcher/reaper.rb', line 156

def reap
  acquire_lock? ? reclaim! : 0
end

#reclaim! ⇒ Object

One unguarded sweep over every private list. 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.

Honours #stop at each boundary: a stop bounding its join leaves the sweep a straggler by design, and the straggler finishes the list it is already draining (its LMOVEs stay applied) but starts nothing new — otherwise it keeps requeueing orphans and holding the sweep locks past the stop, into whatever runs next.



169
170
171
 
# File 'lib/wurk/fetcher/reaper.rb', line 169

def reclaim!
  sweep.tap { wake_for_settling }
end

#reclaim_exited(pid, queue_names) ⇒ Object

Drains the private lists of a process known to be dead: a swarm child its parent has just reaped. The parent knows the pid, the queues the child served and, since the nonce is inherited across fork, the rest of each list's name, so neither a SCAN nor a liveness check is needed. One pipelined LLEN finds the lists holding anything (none, after a drained exit) and raises rather than paying the pool's backoff once per list when Redis is away. Returns the jobs moved.



190
191
192
193
194
195
196
197
 
# File 'lib/wurk/fetcher/reaper.rb', line 190

def reclaim_exited(pid, queue_names)
  lists = queue_names.to_h do |name|
    public_q = Keys.queue(name)
    [Reliable.private_queue_name(public_q, pid: pid), public_q]
  end
  lengths = redis(idempotent: true) { |c| c.pipelined { |p| lists.each_key { |l| p.call('LLEN', l) } } }
  lists.zip(lengths).sum { |(list, public_q), len| len.positive? ? drain(list, public_q) : 0 }
end

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

#running? ⇒ Boolean

Returns:

  • (Boolean) 


150
151
152
 
# File 'lib/wurk/fetcher/reaper.rb', line 150

def running?
  !@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.

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

A LIVE loop is the one and only sweep loop and is handed back as-is. A dead one is not: a stop whose bounded join timed out leaves its straggler referenced, and once that straggler has wound down the reference is a thread that will never sweep again — handing it back would leave the reaper dead for good, silently, since start looked like it had worked. A fresh loop takes its place; while the straggler is still alive it keeps the sweep to itself, so a restart still cannot run two loops at once.



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

def start
  @mutex.synchronize do
    return @thread if @thread&.alive?

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

#stop ⇒ Object

Bounded at TimerLoop::JOIN_TIMEOUT like every other periodic component. A 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. The sweep honours @done at its queue/list/page boundaries, so a straggler runs at most one more Redis round trip past the stop, not the rest of the keyspace. Its own tick_once rescues and reports.



142
143
144
145
146
147
148
 
# File 'lib/wurk/fetcher/reaper.rb', line 142

def stop
  @mutex.synchronize do
    @done = true
    @sleeper.signal
  end
  @thread = nil if @thread&.join(TimerLoop::JOIN_TIMEOUT)
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.