Class: Wurk::Fetcher::Reliable

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

Overview

Fetcher::Reliable reopened, not a second definition: the fetch loop owns which job to claim, and this owns what a claimed job still has to write on its way out — the ACK, the poison-pill counter and the global-concurrency slot. Required from the bottom of reliable.rb, once that class exists, the way lua.rb requires its loader.

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

Constants included from Capped

Capped::AT_CAPACITY, Capped::CAPPED_BACKOFF, Capped::CLAIMED

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, #hostname, identity, #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.



102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/wurk/fetcher/reliable.rb', line 102

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.



92
93
94
 
# File 'lib/wurk/fetcher/reliable.rb', line 92

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.



97
98
99
 
# File 'lib/wurk/fetcher/reliable.rb', line 97

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.



77
78
79
80
 
# File 'lib/wurk/fetcher/reliable.rb', line 77

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.



196
197
198
199
200
201
202
203
204
205
 
# File 'lib/wurk/fetcher/reliable.rb', line 196

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.



127
128
129
 
# File 'lib/wurk/fetcher/reliable.rb', line 127

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.



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

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.



215
216
217
218
219
220
221
 
# File 'lib/wurk/fetcher/reliable.rb', line 215

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.



165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/wurk/fetcher/reliable.rb', line 165

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.



231
232
233
234
235
236
237
238
239
 
# File 'lib/wurk/fetcher/reliable.rb', line 231

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