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.

The ACK does not take a round trip of its own: it is held here and pipelined in front of the next fetch's LMOVE, so a worker draining a busy queue costs one round trip per job in total. See #flush_pending_acks for the paths that must send a held ACK before they stop fetching, and docs/idea/parity-divergences.md for the window that widens.

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

How long a fetcher may answer from its own copy of the paused SET before re-reading it. Deliberately equal to the default poll interval: a worker parked in BLMOVE already cannot observe a pause until that block returns, so caching the busy path for the same window leaves the fleet's worst-case pause latency where it was. A constant, never a config knob — see docs/plans/2026/08/06/101-faster-than-sidekiq/00-semantics-signoff.md.

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

Guards the generation bump only. Fetchers read the counter without it — a torn read is impossible for an Integer reference, and a fetcher that misses a bump by microseconds picks it up on its next pass.

Mutex.new

Constants included from Component

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

Class Attribute Summary collapse

Attributes included from Component

#config

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Component

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

Constructor Details

#initialize(capsule) ⇒ Reliable

Returns a new instance of Reliable.



144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/wurk/fetcher/reliable.rb', line 144

def initialize(capsule)
  super()
  @config = capsule
  @done = false
  @paused = nil
  @paused_generation = nil
  @paused_expires_at = 0.0
  @pending_acks = {}
  @pending_lock = ::Mutex.new
  @queue_keys = {}
  @queue_keys_pid = ::Process.pid
  @prefixed_queues = nil
  @prefixed_source = nil
end

Class Attribute Details

.paused_generation ⇒ Object (readonly)

Every fetcher in this process caches the paused SET against this counter, so bumping it expires all of them at once.



134
135
136
 
# File 'lib/wurk/fetcher/reliable.rb', line 134

def paused_generation
  @paused_generation
end

Class Method Details

.invalidate_paused_cache! ⇒ Object

Queue#pause!/#unpause! call this. Without it a host app that pauses a queue from inside a job would keep watching its own workers drain that queue for up to PAUSED_TTL — the one staleness the sign-off refuses.



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

def invalidate_paused_cache!
  PAUSED_GENERATION_LOCK.synchronize { @paused_generation += 1 }
end

.private_queue_name(public_queue, index = 0) ⇒ Object

Class-level: the name is a pure function of the public queue and this process's identity, and both the fetcher and the Reaper need it (the fetcher's units carry the string #queue_keys built for them, so nothing on the hot path calls this per job). 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.



119
120
121
122
 
# File 'lib/wurk/fetcher/reliable.rb', line 119

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.



238
239
240
241
242
243
244
245
246
247
 
# File 'lib/wurk/fetcher/reliable.rb', line 238

def bulk_requeue(in_progress)
  # First and unconditional — see #flush_pending_acks. Deliberately not
  # rescued: if the ACKs could not be sent we would be requeueing jobs
  # whose completion we failed to record. Leaving them in the private
  # list for the next boot's reaper is the safer of the two.
  flush_pending_acks
  return if in_progress.nil? || in_progress.empty?

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

#defer_ack(uow) ⇒ Object

Take custody of a finished job's LREM instead of sending it now. One slot per processor thread: a capsule shares a single fetcher across its processors, and each thread's fetch → execute → ACK cycle is strictly sequential, so a thread only ever writes its own slot. The lock is for the flush paths, which drain every slot from a different thread.

The slot holds a list rather than a single unit because a failed flush can hand an older ACK back to a thread that has already deferred a newer one (see #restore_pending_acks). Everywhere else it holds exactly one, and the array is reused empty rather than reallocated per job.



169
170
171
 
# File 'lib/wurk/fetcher/reliable.rb', line 169

def defer_ack(uow)
  @pending_lock.synchronize { (@pending_acks[::Thread.current] ||= []) << uow }
end

#flush_pending_acks ⇒ Object

Send every held ACK now, in one pipeline of its own.

Called from every path that stops fetching — nothing else would send them — and from #bulk_requeue, where it is a correctness requirement rather than an optimization: a finished job whose LREM is still pending is not in Manager#hard_shutdown's in-flight list, so the requeue Lua's LREM guard would still find its payload, RPUSH it onto the public queue, and run it a second time on every graceful shutdown.



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
 
# File 'lib/wurk/fetcher/reliable.rb', line 181

def flush_pending_acks
  pending = claim_pending_acks
  # A thread whose ACK already rode a fetch leaves its queue behind empty;
  # dropping the hash those queues lived in is also how the entry of a
  # processor thread that has since died is reclaimed.
  return if pending.each_value.all?(&:empty?)

  begin
    # Apply-safe for the same reason the piggybacked copy is: a replayed
    # LREM finds the payload already gone and removes nothing, and the
    # counter DEL is idempotent by definition. Claiming it buys the drain
    # path the full connection-blip backoff.
    config.redis(idempotent: true) do |conn|
      conn.pipelined { |pipe| pending.each_value { |uows| uows.each { |uow| uow.write_ack(pipe) } } }
    end
  rescue StandardError
    restore_pending_acks(pending)
    raise
  end
end

#queues_cmd ⇒ Object

Prefixed queue keys (queue:<name>) in fetch order. Strict mode preserves declaration order and, with nothing paused, hands back the prebuilt array as-is — the steady-state fetch allocates nothing here, matching Sidekiq's own strict path (fetch.rb:79-87). 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.



257
258
259
260
261
262
263
 
# File 'lib/wurk/fetcher/reliable.rb', line 257

def queues_cmd
  paused = paused_keys
  keys = config.mode == :strict ? prefixed_queues : prefixed_queues.shuffle.uniq
  return keys if paused.empty?

  keys.reject { |key| paused.include?(key) }
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.



207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/wurk/fetcher/reliable.rb', line 207

def retrieve_work
  if @done
    flush_pending_acks
    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 as fast as the CPU
  # allows. Mirrors Sidekiq's BasicFetch guard, upstream #4825.
  if queues.empty?
    flush_pending_acks
    sleep poll_interval
    return nil
  end

  walk(queues)
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.

Which is exactly why it flushes: after this, retrieve_work short-circuits for good, so an ACK held here would otherwise sit until shutdown.



273
274
275
276
277
278
279
280
281
 
# File 'lib/wurk/fetcher/reliable.rb', line 273

def terminate
  @done = true
  flush_pending_acks
rescue StandardError => e
  # Runs on the Manager's thread mid-shutdown, where a raise would skip
  # the rest of the quiet path. The ACKs are back in their slots, so the
  # next flush point (Processor's ensure) retries them.
  handle_exception(e, { context: 'Error flushing pending acks' })
end