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

Class Attribute Summary collapse

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

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

Instance Attribute Details

#config ⇒ Object (readonly) Originally defined in module Component

Returns the value of attribute config.

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

#default_tag(dir = Dir.pwd) ⇒ Object Originally defined in module Component

#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

#fire_event(event, oneshot: true, reverse: false, reraise: false) ⇒ Object Originally defined in module Component

Invokes lifecycle hooks for event. Hooks run in registration order (or LIFO when reverse: true, used for teardown). A raise in one hook is reported via handle_exception and does NOT stop the next hook unless reraise: true (used in tests / fail-fast boot). oneshot: true clears the bucket after dispatch so the event can't fire twice.

#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

#hostname ⇒ Object Originally defined in module Component

#identity ⇒ Object Originally defined in module Component

#leader? ⇒ Boolean Originally defined in module Component

True iff this process currently holds the cluster dear-leader lock. Cached per Component instance for LEADER_CACHE_TTL_MS (~5s): cron and the metrics rollups call this every tick, and an uncached GET would double their Redis traffic at short intervals for no benefit — the lock's own renewal cadence (60s+, spec §6.1) easily tolerates a few-second-stale read. Returns false unconditionally when WURK_LEADER=false (or SIDEKIQ_LEADER=false) is set on the process (opt-out hot-standby). Any Redis error is swallowed → false, so a transient partition can't propagate as an exception into user code.

Spec: docs/target/sidekiq-ent.md §6.1.

Returns:

  • (Boolean)

#logger ⇒ Object Originally defined in module Component

--- delegated to config -------------------------------------------

#mono_ms ⇒ Object Originally defined in module Component

#process_nonce ⇒ Object Originally defined in module Component

#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

#real_ms ⇒ Object Originally defined in module Component

--- clocks ---------------------------------------------------------

#redis(idempotent: false) ⇒ Object Originally defined in module Component

#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

#safe_thread(name, priority: nil, &block) ⇒ Object Originally defined in module Component

Spawns a named thread that runs block under watchdog(name). The parent must retain the returned Thread; otherwise GC may not, but report_on_exception is disabled so we don't double-log on death.

Priority resolution matches Sidekiq (component.rb:44-48): explicit argument, then config.thread_priority, then -1. Ruby's default of 0 buys a 100ms timeslice; each negative step halves it, so -1 keeps a CPU-heavy capsule from starving its siblings for a whole tick.

#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

#tid ⇒ Object Originally defined in module Component

#watchdog(last_words) ⇒ Object Originally defined in module Component

Wraps a block at a thread boundary: any unhandled exception is reported via handle_exception (so it lands in error_handlers / the log) and then re-raised. last_words is the component label included in the context.