Class: Wurk::Launcher

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/launcher.rb

Overview

Top-level supervisor inside each worker process. Owns the Manager pool (one per Capsule), the scheduler poller, and the heartbeat thread. The heartbeat WIRE lives in Wurk::Heartbeat — Launcher owns lifecycle, signal dispatch, and stats rollup; Heartbeat owns the Redis writes.

Lifecycle:

* `run(async_beat:)` — freeze config, start heartbeat, poller, managers.
* `quiet`            — stop fetching across all managers + poller.
* `stop`             — graceful drain inside `config[:timeout]`.
* `heartbeat`        — one-shot beat (also driven by the heartbeat thread).

flush_stats rolls per-process Processor counters (PROCESSED / FAILURE / EXPIRED) into the global + per-day Redis strings every beat. Per-day keys carry STATS_TTL so old days expire automatically.

Spec: docs/target/sidekiq-free.md §12 (Sidekiq::Launcher).

Constant Summary collapse

STATS_TTL =

5 years, in seconds. Per-day stat:processed:YYYY-MM-DD / stat:failed:YYYY-MM-DD / stat:expired:YYYY-MM-DD strings carry this TTL so they roll off without manual cleanup.

5 * 365 * 24 * 60 * 60
BEAT_PAUSE =

Re-exported for test/third-party callers that read it off Launcher (Sidekiq's drop-in surface). The single source of truth is Heartbeat.

Heartbeat::BEAT_PAUSE
PROCTITLES =

The process title, upstream's format and program name word for word: sidekiq 8.1.5 myapp [3 of 10 busy], then stopping once quieted. Probes and runbooks that ps | grep '^sidekiq' or read the busy count keep working across the gem swap. Mutable like upstream's, so code that appends its own segment still can. docs/target/ is silent; the oracle is Sidekiq 8.1's Launcher (docs/deployment.md, "Process titles").

[ # rubocop:disable Style/MutableConstant
  proc { 'sidekiq' },
  proc { ::Sidekiq::VERSION },
  proc { |_me, data| data['tag'] },
  proc { |me, _data| "[#{Processor::WORK_STATE.size} of #{me.config.total_concurrency} busy]" },
  proc { |me, _data| 'stopping' if me.stopping? }
]
BOOT_RECLAIM_JOIN_TIMEOUT =

Bound on how long #stop waits for the boot-time reclaim sweep before moving on — it can still be scanning a large keyspace when a fast shutdown lands right after boot; teardown must not hang on it.

5
TEARDOWN_GRACE =

How far past timeout the Redis-touching teardown may run, and the least it always gets. Below Swarm::SHUTDOWN_GRACE so the parent's SIGKILL stays the backstop rather than the normal end.

4
MIN_TEARDOWN_BUDGET = 
1
KILL_JOIN = 
0.25

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, embedded: false) ⇒ Launcher

Returns a new instance of Launcher.



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/wurk/launcher.rb', line 67

def initialize(config, embedded: false)
  @config = config
  @embedded = embedded
  # @done is "quieted": stop fetching, stay alive, report quiet=true. It
  # deliberately does NOT stop the heartbeat — a quieted process that stopped
  # beating would never publish quiet=true and would expire out of the live
  # set (#236). Only #stop ends the beat, by terminating @beat_timer.
  @done = false
  @shutdown_gate = ShutdownGate.new
  @loop_threads = ::ThreadGroup.new
  @beat_timer = TimerLoop.new(BEAT_PAUSE)
  @managers = build_managers
  build_loops
  @leader = build_leader
  @reaper = build_reaper
  @health_server = build_health_server
  reset_thread_state
end

Instance Attribute Details

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

Returns the value of attribute config.

#heartbeat_thread ⇒ Object (readonly)

Used by tests to inspect the heartbeat thread; not part of the Sidekiq public surface.



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

def heartbeat_thread
  @heartbeat_thread
end

#history ⇒ Object

Returns the value of attribute history.



64
65
66
 
# File 'lib/wurk/launcher.rb', line 64

def history
  @history
end

#managers ⇒ Object

Returns the value of attribute managers.



64
65
66
 
# File 'lib/wurk/launcher.rb', line 64

def managers
  @managers
end

#metrics_flusher ⇒ Object

Returns the value of attribute metrics_flusher.



64
65
66
 
# File 'lib/wurk/launcher.rb', line 64

def metrics_flusher
  @metrics_flusher
end

#metrics_rollup ⇒ Object

Returns the value of attribute metrics_rollup.



64
65
66
 
# File 'lib/wurk/launcher.rb', line 64

def metrics_rollup
  @metrics_rollup
end

#periodic_poller ⇒ Object

Returns the value of attribute periodic_poller.



64
65
66
 
# File 'lib/wurk/launcher.rb', line 64

def periodic_poller
  @periodic_poller
end

#poller ⇒ Object

Returns the value of attribute poller.



64
65
66
 
# File 'lib/wurk/launcher.rb', line 64

def poller
  @poller
end

#queue_consumers ⇒ Object

Returns the value of attribute queue_consumers.



64
65
66
 
# File 'lib/wurk/launcher.rb', line 64

def queue_consumers
  @queue_consumers
end

#queue_rollup ⇒ Object

Returns the value of attribute queue_rollup.



64
65
66
 
# File 'lib/wurk/launcher.rb', line 64

def queue_rollup
  @queue_rollup
end

Instance Method Details

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

#dump_threads ⇒ Object

Logs every thread's backtrace — Sidekiq's TTIN, which the dashboard's "dump threads" queues. Handled in-process rather than re-delivered: a swarm child or embedded host may not trap TTIN, and its default disposition stops the whole process.



221
222
223
224
225
226
 
# File 'lib/wurk/launcher.rb', line 221

def dump_threads
  Thread.list.each do |thread|
    logger.warn "Thread TID-#{(thread.object_id ^ ::Process.pid).to_s(36)} #{thread.name}"
    logger.warn(thread.backtrace ? thread.backtrace.join("\n") : '<no backtrace available>')
  end
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_stats ⇒ Object

Rolls in-process Processor counters into Redis. Pipelined so a single round trip covers all writes. Skips when all counters are zero to avoid touching keys we have nothing to add to.



196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/wurk/launcher.rb', line 196

def flush_stats
  processed = Processor::PROCESSED.reset
  failed = Processor::FAILURE.reset
  expired = Processor::EXPIRED.reset
  return if processed.zero? && failed.zero? && expired.zero?

  write_stats(processed, failed, expired)
rescue StandardError => e
  # The counters were reset above, so a raise here drops this window's stats
  # for good: #write_stats cannot claim apply-safety and the pool no longer
  # replays it through a blip. Accepted deliberately — adding them back
  # would double-count the case where the INCRBYs did land and only the
  # reply was lost, and the per-job at-least-once semantics don't apply to
  # *counters*. The next beat resets again.
  handle_exception(e, { context: 'flush_stats' })
end

#handle_exception(ex, ctx = {}) ⇒ Object Originally defined in module Component

#heartbeat ⇒ Object

One-shot beat. Public for embedded mode (and for tests) — the heartbeat thread calls this on BEAT_PAUSE cadence.



188
189
190
191
 
# File 'lib/wurk/launcher.rb', line 188

def heartbeat
  flush_stats
  ❤ # rubocop:disable Naming/AsciiIdentifiers
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

#quiet ⇒ Object

Idempotent. Flips stopping? true, halts fetching across every Manager + the poller, then fires the :quiet event in reverse registration order so teardown hooks run LIFO.



132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/wurk/launcher.rb', line 132

def quiet
  return if @done

  @done = true
  @managers.each(&:quiet)
  @poller&.terminate
  # The cron poller is intentionally NOT terminated here: a TSTP-quieted
  # leader still enqueues periodic jobs — it only stops fetching for itself.
  # Loops stop only on full shutdown (#stop). Spec: sidekiq-ent.md §2.6.
  fire_event(:quiet, reverse: true)
end

#real_ms ⇒ Object Originally defined in module Component

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

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

#run(async_beat: true) ⇒ Object

Boot order matters:

1. freeze! the config so mutations after fork are visible mistakes.
 A swarm child finds the slot-independent half already frozen by its
 parent (Configuration#prepare_for_fork!); what closes here is the
 capsules, which stayed open for this child's slot and pools.
2. write the first heartbeat SYNCHRONOUSLY, then spawn the beat thread.
 Ahead of the managers on purpose: the moment a manager fetches, this
 process holds jobs in its private list, and a booting sibling's
 reclaim sweep decides whether those are orphans by whether the owner
 is in `processes`. Fetching before the first beat lands let a live
 worker's in-flight job be reclaimed and run twice. One pipelined
 round trip on the time-to-first-job path is the price.
3. start the managers. Everything below is periodic background work
 whose first tick is seconds away, so it has nothing to gain from
 going ahead of them.
4. start the reaper and the boot-time reclaim sweep, ahead of the
 periodic loops: a process restarted outside a swarm (exe/wurk under
 systemd, a pod) has its previous incarnation's jobs recovered from
 here. The sweep takes the cluster reaper lock, so a fleet booting
 together makes one keyspace pass; a swarm child's SIGKILLed sibling
 is recovered by the parent, not by this sweep.
5. write the `config.periodic` registrations (Ent registers loops on
 startup), then start the periodic loops. None of them ticks at
 zero — TimerLoop#run waits an interval before its first yield,
 Scheduled::Poller waits INITIAL_WAIT, and Leader waits
 DEFAULT_INITIAL_WAIT — so starting them last costs a few thread
 spawns and nothing else. Both leader-
 gated pollers are also safe to start before leadership settles: a
 non-leader tick just returns early.
6. start the health probe server LAST so the listener doesn't
 accept k8s probes until the rest of the launcher is up.

A no-op once a shutdown has been claimed: a TERM that lands between the child installing its traps and calling this used to have its drain finish first, after which run booted anyway — a leader campaign and a process entry nobody would ever stop. A shutdown requested from another thread mid-boot waits for the boot to finish (ShutdownGate#boot), so the teardown never races half-started components.



124
125
126
127
 
# File 'lib/wurk/launcher.rb', line 124

def run(async_beat: true)
  @shutdown_gate.boot { boot(async_beat) }
  nil
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.

#shutting_down? ⇒ Boolean

Past quiet: #stop has been called and this process is on its way out.

Returns:

  • (Boolean) 


182
183
184
 
# File 'lib/wurk/launcher.rb', line 182

def shutting_down?
  @shutting_down == true
end

#stop ⇒ Object

Graceful shutdown, single-shot: several requests can be in flight at once (a dashboard-queued TERM, the embedded host's own Embedded#stop, a Manager that can no longer hold its concurrency), so the first caller drains and the rest wait it out — see ShutdownGate. Deadline is monotonic so wall-clock skew can't extend it. Managers stop in parallel threads so a slow capsule doesn't block its siblings — and each drain is guarded, because a capsule that blows up mid-drain (Redis down during bulk_requeue) used to surface out of join and skip the whole teardown tail, leaving the leader lock held, the process listed as live, and its port open.

The joins get one shared budget of deadline + TimerLoop::JOIN_TIMEOUT, not an open-ended wait. Manager#stop is meant to run a little past the deadline — on expiry it bulk_requeues the in-flight UnitsOfWork, kills the threads and gives them a ~3s window to run their ensure blocks — but past that it is wedged (a fetcher.terminate / bulk_requeue parked on a Redis that stopped answering), and every further second spent waiting comes out of the teardown tail below, which the swarm parent only allows shutdown_timeout + Swarm::SHUTDOWN_GRACE for before it SIGKILLs us mid-drain.



163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/wurk/launcher.rb', line 163

def stop
  @shutdown_gate.run do
    deadline = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + (@config[:timeout] || 25)
    @teardown_deadline = deadline + TEARDOWN_GRACE
    @shutting_down = true
    quiet
    stoppers = @managers.map { |m| Thread.new { teardown_step('manager') { m.stop(deadline) } } }
    fire_event(:shutdown, reverse: true)
    @shutdown_gate.join_within(stoppers, deadline + TimerLoop::JOIN_TIMEOUT)
  ensure
    release_components
  end
end

#stopping? ⇒ Boolean

Returns:

  • (Boolean) 


177
178
179
 
# File 'lib/wurk/launcher.rb', line 177

def stopping?
  @done
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.