Class: Wurk::Fetcher::SelfReconcile

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

Overview

A live worker's check of its own private lists.

The Reaper leaves a live owner's list alone, and a live owner only ever settles the jobs its threads know about. So a payload no thread knows about would wait there until the process exits. Two ways in:

* a claim applied but its reply was lost (Redis restart, a stall past the
read timeout, a dropped socket): the client re-sends and claims another
job, so one job per thread per incident;
* a job that failed while Redis was down, where the retry booking and the
requeue both raised.

A payload counts as stranded only if it is in the list and held by nothing: no thread's claim (Fetcher::Reliable::CLAIM_KEY, which covers the whole of a running job however long it runs) and no pending ACK. The list is read first and the holders second, so a claim landing between the two reads as held. What the reads cannot see is a unit in transit: claimed but not yet stamped, or an ACK on the wire. Those last microseconds to seconds, which is why one sighting proves nothing: it has to be seen stranded again at least confirm seconds later before it moves.

Counted rather than compared as sets, because a client re-send can leave two byte-identical payloads, and a running job must not vouch for its twin.

Defined Under Namespace

Classes: Suspect

Constant Summary collapse

DEFAULT_CONFIRM =

Above anything a unit can spend in transit: the longest a held ACK sits on the wire is the pool's blip backoff plus its socket timeouts, ~18s.

30

Instance Method Summary collapse

Constructor Details

#initialize(config, confirm: DEFAULT_CONFIRM, stop: -> { false }) ⇒ SelfReconcile

A payload put back goes through the same poison accounting as a reclaim (OrphanReclaim), so a job that kills its processor thread every time still hits the poison cap.



39
40
41
42
43
44
45
 
# File 'lib/wurk/fetcher/self_reconcile.rb', line 39

def initialize(config, confirm: DEFAULT_CONFIRM, stop: -> { false })
  @config = config
  @confirm = confirm
  @stop = stop
  @orphan_reclaim = OrphanReclaim.new(config)
  @suspects = {}
end

Instance Method Details

#sweep ⇒ Object

One pass. Returns how many payloads went back to their public queues.



48
49
50
51
52
53
54
55
56
57
 
# File 'lib/wurk/fetcher/self_reconcile.rb', line 48

def sweep
  lists = own_lists
  return 0 if lists.empty?

  crowded = crowded_lists(lists)
  return forget_all if crowded.empty?

  contents = read(crowded.keys)
  settle(crowded, contents, held_by_list)
end