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>|<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

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.



63
64
65
66
67
 
# File 'lib/wurk/fetcher/reliable.rb', line 63

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.



58
59
60
61
 
# File 'lib/wurk/fetcher/reliable.rb', line 58

def self.private_queue_name(public_queue, index = 0)
  host = ENV['DYNO'] || Socket.gethostname
  "#{public_queue}|#{host}|#{::Process.pid}|#{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.



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

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.



105
106
107
108
109
110
 
# File 'lib/wurk/fetcher/reliable.rb', line 105

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 



69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/wurk/fetcher/reliable.rb', line 69

def retrieve_work
  return nil if @done

  queues = queues_cmd
  return nil if queues.empty?

  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.



117
118
119
 
# File 'lib/wurk/fetcher/reliable.rb', line 117

def terminate
  @done = true
end