Class: Wurk::Fetcher::Reliable

Inherits:
Fetcher
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/fetcher/reliable.rb

Overview

Default fetcher. Each public queue is paired with a per-process private list (queue:<name>|<host>|<pid>|<nonce>|<idx>); a job is moved atomically from the public tail to the private head via LMOVE, and stays there until the Processor explicitly ACKs (LREM). SIGKILL between fetch and ack leaves the job in the private list, where the next boot of this process reclaims it via bulk_requeue.

Priority handling: iterate queues_cmd in order with non-blocking LMOVE, then fall back to a blocking BLMOVE on the first queue so an empty poll doesn't spin Redis. BLMOVE has no multi-key form, so blocking on a single queue is the best Redis gives us. The block timeout defaults to TIMEOUT (2s) and is overridable per the Pro super_fetch §3.3 config.fetch_poll_interval knob.

Spec: docs/target/sidekiq-pro.md §3 (super_fetch, §3.3 poll interval), docs/target/sidekiq-free.md §15 (TIMEOUT=2).

Defined Under Namespace

Classes: UnitOfWork

Constant Summary collapse

TIMEOUT =

Default BLMOVE block timeout; overridable via config.fetch_poll_interval.

2
QUIET_PAUSE =

Backoff for the quieted short-circuit. Manager#quiet terminates the shared fetcher before it terminates the processors, and Processor#run loops on its own flag — so in that window every processor would spin on an instant nil. Kept below Manager::PAUSE_TIME, which #stop sleeps immediately after #quiet, so this pause adds no drain latency.

0.05

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::LEADER_CACHE_TTL_MS, Component::PROCESS_NONCE

Instance Attribute Summary

Attributes included from Component

#config

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #identity, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, #tid, #watchdog

Constructor Details

#initialize(capsule) ⇒ Reliable

Returns a new instance of Reliable.



94
95
96
97
98
 
# File 'lib/wurk/fetcher/reliable.rb', line 94

def initialize(capsule)
  super()
  @config = capsule
  @done = false
end

Class Method Details

.private_queue_name(public_queue, index = 0) ⇒ Object

Class-level so UnitOfWork can compute the private list without carrying a back-reference to its parent fetcher. Index defaults to 0 — we run one fetcher per capsule today. Multi-processor topology (one private list per processor slot) is a future Manager concern.

The nonce marks the incarnation. host+pid alone is ambiguous once PID namespaces are in play: a restarted container reuses both, so the reaper's kill(0) liveness check would read a dead owner's list as live (jobs stranded) or a live owner's as dead (job run twice). Keys written before the nonce existed stay reclaimable — Reaper#parse_owner accepts both shapes.



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

def self.private_queue_name(public_queue, index = 0)
  host = ENV['DYNO'] || Socket.gethostname
  "#{public_queue}|#{host}|#{::Process.pid}|#{Component::PROCESS_NONCE}|#{index}"
end

Instance Method Details

#bulk_requeue(in_progress) ⇒ Object

Called on shutdown for jobs the Processor couldn't finish in time. Atomically moves each still-private UoW back to its public queue via the RELIABLE_REQUEUE Lua (LREM-guarded RPUSH): the job leaves the per-process private list and reappears on the public queue in one hop, so it's visible immediately after a deploy instead of waiting for the next boot's reaper. The guard makes the move idempotent against the cross-thread job-read race in Manager#hard_shutdown — a Processor that ACKed in that window is a no-op (LREM misses, RPUSH skipped), so a finished job is never resurrected. Sidekiq Pro super_fetch §3 retains in-flight in the private list until the next boot; we prefer the immediate move so a rolling deploy recovers work without a restart.



139
140
141
142
143
 
# File 'lib/wurk/fetcher/reliable.rb', line 139

def bulk_requeue(in_progress)
  return if in_progress.nil? || in_progress.empty?

  config.redis { |conn| requeue_pipelined(conn, in_progress) }
end

#queues_cmd ⇒ Object

Prefixed queue keys (queue:<name>) in fetch order. Strict mode preserves declaration order. Random/weighted shuffle each call — @queues is pre-expanded by weight in Capsule#queues=, so uniform shuffle yields weighted fairness; .uniq trims duplicates. Paused queues are filtered after shuffle so the membership test runs on the smallest possible set.



151
152
153
154
155
156
 
# File 'lib/wurk/fetcher/reliable.rb', line 151

def queues_cmd
  names = config.mode == :strict ? config.queues : config.queues.shuffle.uniq
  paused = paused_names
  names = names.reject { |q| paused.include?(q) } unless paused.empty?
  names.map { |q| "#{Keys::QUEUE_PREFIX}#{q}" }
end

#retrieve_work ⇒ Object

Every pass that yields no job has to cost wall-clock time: Processor#run drives process_one in a bare until @done loop with no pause of its own, so any nil returned instantly turns N processor threads into a hot loop. The blocking BLMOVE pays that cost on the normal empty-queue path; the two short-circuits below have to pay it themselves.



105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/wurk/fetcher/reliable.rb', line 105

def retrieve_work
  if @done
    sleep QUIET_PAUSE
    return nil
  end

  queues = queues_cmd
  # Nothing fetchable — every queue paused, or none configured. Back off a
  # full poll interval rather than re-running queues_cmd (an SMEMBERS per
  # pass, on the main pool) as fast as the CPU allows. Mirrors Sidekiq's
  # BasicFetch guard, upstream #4825.
  if queues.empty?
    sleep poll_interval
    return nil
  end

  queues.each do |public_q|
    uow = lmove(public_q)
    return uow if uow
  end
  blmove(queues.first)
end

#terminate ⇒ Object

Quiet hook (Manager#quiet). Flips the drain flag so retrieve_work short-circuits: once quieted, no processor can pull a fresh UoW, even one sitting in the between-jobs window (Processor#run only re-checks its own @done between iterations). Quiet is one-way — matches Sidekiq TSTP (spec §21.3), there is no un-terminate.



163
164
165
 
# File 'lib/wurk/fetcher/reliable.rb', line 163

def terminate
  @done = true
end