Module: Wurk::Middleware::PoisonPill

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

Overview

Pro parity (§3.2): poison-pill detection for reliable-fetch orphans. When a job is recovered out of a dead process's private list, we INCR a per-jid counter at super_fetch:recovered:<jid> with a 72h TTL. Once the counter crosses RECOVERY_THRESHOLD (3) the next recovery is treated as a 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.

No server-middleware registration: the producer is Reaper#drain, which drives the lifecycle directly via track!(payload, queue:) on each orphan it moves back to the public queue. 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

Constant Summary collapse

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

Class Method Summary collapse

Class Method Details

.bump_counter(jid) ⇒ Object

No apply-safety claim: INCR is additive, so a block replayed after a lost reply bumps twice and can carry a healthy job past RECOVERY_THRESHOLD into the dead set. Raising instead only leaves the count short — Reaper#drain rescues, and the job is already back on its public queue, so it just misses one poison check.



164
165
166
167
168
169
170
171
 
# File 'lib/wurk/middleware/poison_pill.rb', line 164

def bump_counter(jid)
  key = counter_key(jid)
  Wurk.redis do |conn|
    count = conn.call('INCR', key).to_i
    conn.call('EXPIRE', key, RECOVERY_TTL)
    count
  end
end

.callbacks ⇒ Object 



139
140
141
 
# File 'lib/wurk/middleware/poison_pill.rb', line 139

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.



96
97
98
99
100
 
# File 'lib/wurk/middleware/poison_pill.rb', line 96

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.



119
120
121
122
123
 
# File 'lib/wurk/middleware/poison_pill.rb', line 119

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

  pipe.call('DEL', counter_key(jid))
end

.counter_key(jid) ⇒ Object 



125
126
127
 
# File 'lib/wurk/middleware/poison_pill.rb', line 125

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

.emit(metric, klass, queue) ⇒ Object 



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

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 



198
199
200
201
202
203
204
 
# File 'lib/wurk/middleware/poison_pill.rb', line 198

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.



209
210
211
212
213
214
215
216
217
 
# File 'lib/wurk/middleware/poison_pill.rb', line 209

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

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

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

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.



186
187
188
189
190
191
 
# File 'lib/wurk/middleware/poison_pill.rb', line 186

def mark_poison(payload, job, queue:, count:)
  emit('jobs.poison', job['class'], queue)
  json = payload.is_a?(String) ? payload : Wurk.dump_json(job)
  Wurk::DeadSet.new.kill(json, 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.

Raises:

  • (ArgumentError) 


132
133
134
135
136
137
 
# File 'lib/wurk/middleware/poison_pill.rb', line 132

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

  callbacks << block
  block
end

.parse(payload) ⇒ Object

---- internals --------------------------------------------------



150
151
152
153
154
155
156
157
 
# File 'lib/wurk/middleware/poison_pill.rb', line 150

def parse(payload)
  return payload if payload.is_a?(::Hash)
  return nil unless payload.is_a?(::String)

  Wurk.load_json(payload)
rescue ::JSON::ParserError
  nil
end

.poisoned_error(klass, count) ⇒ Object 



193
194
195
196
 
# File 'lib/wurk/middleware/poison_pill.rb', line 193

def poisoned_error(klass, count)
  message = "#{klass || 'job'} was recovered #{count} times without completing"
  Poisoned.new(message).tap { |e| e.set_backtrace(caller) }
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.



87
88
89
90
91
 
# File 'lib/wurk/middleware/poison_pill.rb', line 87

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.



144
145
146
 
# File 'lib/wurk/middleware/poison_pill.rb', line 144

def reset!
  @callbacks = []
end

.track!(payload, queue: nil, config: Wurk.configuration) ⇒ Symbol

Called per recovered orphan job. Returns :poison when the threshold was crossed and the job was killed; :recovered when the job is being re-pushed (caller's responsibility — we don't touch the queue here). Emits jobs.recovered.fetch on every call, 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, Hash) —

    the job JSON or pre-parsed hash.

  • 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; defaults to the global so non-reaper callers (tests) need not pass it.

Returns:

  • (Symbol) —

    :recovered | :poison



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/wurk/middleware/poison_pill.rb', line 63

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

  jid = job['jid']
  klass = job['class']
  emit('jobs.recovered.fetch', klass, queue)

  count = bump_counter(jid) if jid && !jid.empty?
  if count && count >= RECOVERY_THRESHOLD
    mark_poison(payload, job, queue: queue, count: count)
    fire_super_fetch(config, payload, Pill.new(jid: jid, klass: klass, count: count, queue: queue))
    :poison
  else
    fire_super_fetch(config, payload, nil)
    :recovered
  end
end