Module: Wurk::Middleware::PoisonPill

Defined in:
lib/wurk/fetcher/poison_pill.rb

Overview

Pro parity (§3.2): poison-pill detection for reliable-fetch orphans. When a job is recovered out of a private list, a per-jid counter at super_fetch:recovered:<jid> goes up by one with a 72h TTL. Recoveries up to RECOVERY_THRESHOLD (3) requeue; the one that takes the counter past it (the 4th orphaning) is the poison pill: the payload is moved to the dead set, jobs.poison is emitted to statsd, and recovery callbacks fire so operators can be paged.

Counter key TTL is wire-compat with Sidekiq Pro — third-party tooling that watches super_fetch:recovered:* expects 72h.

Not a chain middleware, so the file lives with the fetcher; the constant keeps its Middleware path because Pro initializers name it. No server-middleware registration: the producer is Fetcher::OrphanReclaim, whose script bumps the counter and takes the requeue-or-kill decision in the same step as the move, then hands the outcome to PoisonPill.recovered! for the notifications. The consumer is the ACK — Fetcher::Reliable::UnitOfWork#acknowledge pipelines PoisonPill.clear_in next to its LREM, so an attempt that finished starts the next one at zero without costing a round trip of its own.

Defined Under Namespace

Classes: Pill, Poisoned, Registration

Constant Summary collapse

RECOVERY_THRESHOLD = 
3
RECOVERY_TTL = 
72 * 60 * 60
KEY_PREFIX = 
'super_fetch:recovered:'

Class Method Summary collapse

Class Method Details

.callbacks ⇒ Object 



161
162
163
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 161

def callbacks
  @callbacks ||= []
end

.clear!(jid) ⇒ Object

Resets the counter for a jid. The hot path uses clear_in instead — this is the standalone form for the API/dashboard and for callers with no pipeline of their own.



97
98
99
100
101
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 97

def clear!(jid)
  return if jid.nil? || jid.to_s.empty?

  Wurk.redis { |conn| conn.call('DEL', counter_key(jid)) }
end

.clear_in(pipe, jid) ⇒ Object

Queue the counter reset onto a pipeline the caller already has open.

The ACK is the reset point. An attempt that got as far as acking — returned, or raised and booked its retry — is proof the job did not take its worker down, and that is the only thing this counter measures. Without the reset, three reclaims caused by three unrelated crashes inside 72h dead-set a job that has been completing all along (jids do come back: a UI/API retry re-pushes the same one, as does any client that supplies its own). Riding a round trip the ACK already makes is what keeps that free for the jobs — nearly all of them — that were never reclaimed at all. Guards the blank jid for the same reason clear! and recovery_count do, rather than relying on the one caller to do it: counter_key(nil) is the bare KEY_PREFIX, so an unguarded DEL here would delete a key that is shared rather than per-job. The caller's own check stays -- it also skips queueing the command at all -- but the invariant belongs on the method that builds the key.



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

def clear_in(pipe, jid)
  return if jid.nil? || jid.to_s.empty?

  pipe.call_v(['DEL', counter_key(jid)])
end

.counter_key(jid) ⇒ Object 



126
127
128
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 126

def counter_key(jid)
  "#{KEY_PREFIX}#{jid}"
end

.emit(metric, klass, queue) ⇒ Object 



180
181
182
183
184
185
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 180

def emit(metric, klass, queue)
  tags = []
  tags << "class:#{klass}" if klass
  tags << "queue:#{queue}" if queue
  Wurk::Metrics::Statsd.increment(metric, tags: tags.empty? ? nil : tags)
end

.fire_callbacks(pill) ⇒ Object 



208
209
210
211
212
213
214
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 208

def fire_callbacks(pill)
  callbacks.each do |cb|
    cb.call(pill)
  rescue StandardError => e
    Wurk.configuration.handle_exception(e, context: 'Wurk::Middleware::PoisonPill')
  end
end

.fire_super_fetch(config, payload, pill) ⇒ Object

Invoke the Pro recovery callback registered via config.super_fetch! { }. No-op unless one is registered. jobstr is the raw job JSON so a Pro |jobstr, pill| block sees exactly what Sidekiq Pro hands it.



219
220
221
222
223
224
225
226
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 219

def fire_super_fetch(config, payload, pill)
  cb = config.super_fetch_callback
  return unless cb

  cb.call(payload, pill)
rescue StandardError => e
  config.handle_exception(e, context: 'Wurk::Middleware::PoisonPill')
end

.mark_poison(payload, job, queue:, count:) ⇒ Object

The script already put payload in the dead set; DeadSet#kill writes the same member again, so the trim and the death handlers stay DeadSet's.

Death handlers fire (notify_failure defaults to true). A poison kill is a death like any other exhaustion: suppressing it strands every batch that owns the job — Batch::DeathHandler is a death handler, so a silent kill leaves the batch's pending count stuck forever and neither :death nor :complete ever runs. Pro's spec is silent here (docs/target/sidekiq-pro.md §12); recorded in docs/idea/parity-divergences.md.



196
197
198
199
200
201
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 196

def mark_poison(payload, job, queue:, count:)
  emit('jobs.poison', job['class'], queue)
  Wurk::Metrics::ProcessCounters.increment(:poison_kills)
  Wurk::DeadSet.new.kill(payload, ex: poisoned_error(job['class'], count))
  fire_callbacks(jid: job['jid'], klass: job['class'], count: count, queue: queue)
end

.on_poison(&block) ⇒ Object

Register a callback fired when a poison pill is detected. Callbacks receive a single Hash {jid:, klass:, count:, queue:} — matches Sidekiq Pro's documented shape so consumers can drop in unchanged.

Dedups by callable identity — an initializer that re-runs (Rails code reloading, a re-required file) and hands back the same stored Proc each time won't accumulate duplicate closures (and the bindings they retain). A fresh block literal per call still registers separately, same as before. Returns a Registration the caller can #remove! to deregister explicitly.

Raises:

  • (ArgumentError) 


140
141
142
143
144
145
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 140

def on_poison(&block)
  raise ArgumentError, 'block required' unless block

  callbacks << block unless callbacks.any? { |cb| cb.equal?(block) }
  Registration.new(callbacks, block)
end

.parse(payload) ⇒ Object

The job Hash, or nil for anything that is not a JSON object.



173
174
175
176
177
178
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 173

def parse(payload)
  job = Wurk.load_json(payload)
  job if job.is_a?(::Hash)
rescue ::JSON::ParserError
  nil
end

.poisoned_error(klass, count) ⇒ Object 



203
204
205
206
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 203

def poisoned_error(klass, count)
  message = "#{klass || 'job'} was orphaned #{count} times without completing"
  Poisoned.new(message).tap { |e| e.set_backtrace(caller) }
end

.recovered!(payload, job, count:, queue: nil, config: Wurk.configuration) ⇒ Symbol

Called per recovered orphan job, once Fetcher::OrphanReclaim has moved it: count is the recovery counter its script wrote (0 for a payload with no jid). Returns :poison when the count is past the threshold — the script already put the job in the dead set, never on its queue — and :recovered otherwise. Emits jobs.recovered.fetch for every parseable job, jobs.poison only on the kill path.

Fires the Pro super_fetch! recovery callback (config.super_fetch_callback) exactly once per call: (jobstr, nil) on plain recovery, (jobstr, pill) on the kill path. The poison-only on_poison Hash callbacks fire independently inside #mark_poison.

Parameters:

  • payload (String) —

    the job JSON as it sat in the private list.

  • job (Hash, nil) —

    the payload parsed by parse.

  • count (Integer) —

    the recovery count after this recovery.

  • queue (String, nil) (defaults to: nil) —

    the public queue name (without queue:).

  • config (Configuration) (defaults to: Wurk.configuration) —

    config that owns the super_fetch! callback.

Returns:

  • (Symbol) —

    :recovered | :poison



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 69

def recovered!(payload, job, count:, queue: nil, config: Wurk.configuration)
  unless job
    fire_super_fetch(config, payload, nil)
    return :recovered
  end

  emit('jobs.recovered.fetch', job['class'], queue)
  if count > RECOVERY_THRESHOLD
    mark_poison(payload, job, queue: queue, count: count)
    fire_super_fetch(config, payload, Pill.new(jid: job['jid'], klass: job['class'], count: count, queue: queue))
    :poison
  else
    fire_super_fetch(config, payload, nil)
    :recovered
  end
end

.recovery_count(jid) ⇒ Object

Reads the current recovery counter without bumping it. Used by tests and dashboards; returns 0 for jobs that have never been recovered.



88
89
90
91
92
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 88

def recovery_count(jid)
  return 0 if jid.nil? || jid.to_s.empty?

  Wurk.redis { |conn| conn.call('GET', counter_key(jid)) }.to_i
end

.reset! ⇒ Object

Test-only reset.



166
167
168
 
# File 'lib/wurk/fetcher/poison_pill.rb', line 166

def reset!
  @callbacks = []
end