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

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.



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

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.



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

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



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

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.



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

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



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

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



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

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



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

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