Class: Wurk::Fetcher::OrphanReclaim

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/fetcher/orphan_reclaim.rb

Overview

Puts one payload from a private list back where it belongs, for both ways an orphan is found: the Reaper draining a dead owner's list, and SelfReconcile returning a payload its own live process lost track of.

The poison-pill verdict is taken in the same script that moves the job (lua/reclaim_orphan.lua), so a job past the cap goes straight from the private list to the dead set and is never visible on its public queue. The notifications (death handlers, super_fetch!, on_poison) follow once the script has returned, for a job no worker can be running.

Constant Summary collapse

THRESHOLD = 
Middleware::PoisonPill::RECOVERY_THRESHOLD
TTL = 
Middleware::PoisonPill::RECOVERY_TTL

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ OrphanReclaim

Returns a new instance of OrphanReclaim.



24
25
26
 
# File 'lib/wurk/fetcher/orphan_reclaim.rb', line 24

def initialize(config)
  @config = config
end

Instance Method Details

#call(private_list, public_q, payload) ⇒ Object

Returns true when this call took the payload off the private list, false when it had already left (another reaper, an ACK).

Sent once, never re-sent on a fresh socket: a replay of a script that already ran finds the payload gone and answers "someone else moved it", which skipped the death handlers of a kill (#908), and a list holding two identical copies would move the second. A reply lost after the script ran is settled by reading back what it wrote instead (#settle_lost_reply).



36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/wurk/fetcher/orphan_reclaim.rb', line 36

def call(private_list, public_q, payload)
  job = Middleware::PoisonPill.parse(payload)
  keys = move_keys(private_list, public_q, job)
  count = moved_count(keys, payload, now)
  return false unless count

  Metrics::ProcessCounters.increment(:reaper_reclaimed)
  unless count == :uncounted
    Middleware::PoisonPill.recovered!(payload, job, count: count, queue: queue_name(public_q),
                                                    config: @config)
  end
  true
end