Class: Wurk::Fetcher::Reliable

Inherits:
Fetcher
  • Object
show all
Includes:
Component, AtMaxmemory, Capped, ManyQueues
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
PAUSED_CLOCK =

The clock that cache is timed on, read once per fetch. A TTL in seconds does not need CLOCK_MONOTONIC's nanoseconds, and that clock is a real syscall wherever the kernel's clocksource has no vDSO read (hpet, some hypervisors), while the coarse one is a vDSO read everywhere Linux has it. It lags by up to its resolution, which the cache deadline gives back (PAUSED_CLOCK_SLACK), so the cache still never outlives PAUSED_TTL.

defined?(::Process::CLOCK_MONOTONIC_COARSE) ? ::Process::CLOCK_MONOTONIC_COARSE : ::Process::CLOCK_MONOTONIC
PAUSED_CLOCK_SLACK = 
::Process.clock_getres(PAUSED_CLOCK)
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
CLAIM_KEY =

Thread variable holding the unit this thread claimed and has not settled; what Fetcher::SelfReconcile reads to tell a running job from a stranded one. A thread variable, not Thread#[]: those are per fiber, and a job running inside a Fiber would read as holding nothing.

:wurk_fetcher_claim

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.



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/wurk/fetcher/reliable.rb', line 124

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
  @claim_keys = nil
  @strict_unpaused = nil
  @rest_empty_until = 0.0
  @oom_reported_at = 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.



114
115
116
 
# File 'lib/wurk/fetcher/reliable.rb', line 114

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.



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

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

.private_queue_name(public_queue, index = 0, pid: ::Process.pid) ⇒ 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 — PrivateListKey accepts both shapes. pid: lets a swarm parent name a reaped child's lists (Reaper#reclaim_exited).



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

def self.private_queue_name(public_queue, index = 0, pid: ::Process.pid)
  "#{public_queue}|#{Component.hostname}|#{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 (guarded RPUSH, then LREM): 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.



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

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) }
rescue RedisClient::OutOfMemoryError => e
  requeue_past_maxmemory(in_progress, e)
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.



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

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 guard would still find its payload, RPUSH it onto the public queue, and run it a second time on every graceful shutdown.



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

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

#held_acks ⇒ Object

The units whose ACK is waiting here for a fetch to ride. A snapshot, for Fetcher::SelfReconcile.



215
216
217
 
# File 'lib/wurk/fetcher/reliable.rb', line 215

def held_acks
  @pending_lock.synchronize { @pending_acks.values.flatten }
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); with a queue paused it hands back one filtered array per paused-cache refresh (#strict_unpaused). Random/weighted shuffle each call — @queues is pre-expanded by weight in Capsule#queues=, so uniform shuffle yields weighted fairness — then trim duplicates and paused queues in place on that fresh array. The result is frozen only when it is reused across passes; ManyQueues#claim_keys keys its cache on that.



253
254
255
256
257
258
259
260
261
262
 
# File 'lib/wurk/fetcher/reliable.rb', line 253

def queues_cmd
  paused = paused_keys
  keys, distinct = prefixed_queues
  return paused.empty? ? keys : strict_unpaused(keys, paused) if config.mode == :strict

  keys = keys.shuffle
  keys.uniq! unless distinct
  keys.reject! { |key| paused.include?(key) } unless paused.empty?
  keys
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.



191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/wurk/fetcher/reliable.rb', line 191

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)
rescue RedisClient::OutOfMemoryError => e
  claim_at_maxmemory(queues, e)
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.



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

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.