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

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::LEADER_CACHE_TTL_MS, Component::PROCESS_NONCE

Instance Attribute Summary collapse

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #identity, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, tid, #tid, #watchdog

Constructor Details

#initialize(config, embedded: false) ⇒ Launcher

Returns a new instance of Launcher.



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/wurk/launcher.rb', line 56

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
  @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

#cron_poller ⇒ Object

Returns the value of attribute cron_poller.



54
55
56
 
# File 'lib/wurk/launcher.rb', line 54

def cron_poller
  @cron_poller
end

#heartbeat_thread ⇒ Object (readonly)

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



200
201
202
 
# File 'lib/wurk/launcher.rb', line 200

def heartbeat_thread
  @heartbeat_thread
end

#history ⇒ Object

Returns the value of attribute history.



54
55
56
 
# File 'lib/wurk/launcher.rb', line 54

def history
  @history
end

#managers ⇒ Object

Returns the value of attribute managers.



54
55
56
 
# File 'lib/wurk/launcher.rb', line 54

def managers
  @managers
end

#metrics_flusher ⇒ Object

Returns the value of attribute metrics_flusher.



54
55
56
 
# File 'lib/wurk/launcher.rb', line 54

def metrics_flusher
  @metrics_flusher
end

#metrics_rollup ⇒ Object

Returns the value of attribute metrics_rollup.



54
55
56
 
# File 'lib/wurk/launcher.rb', line 54

def metrics_rollup
  @metrics_rollup
end

#poller ⇒ Object

Returns the value of attribute poller.



54
55
56
 
# File 'lib/wurk/launcher.rb', line 54

def poller
  @poller
end

#queue_rollup ⇒ Object

Returns the value of attribute queue_rollup.



54
55
56
 
# File 'lib/wurk/launcher.rb', line 54

def queue_rollup
  @queue_rollup
end

Instance Method Details

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



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/wurk/launcher.rb', line 181

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

#heartbeat ⇒ Object

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



173
174
175
176
 
# File 'lib/wurk/launcher.rb', line 173

def heartbeat
  flush_stats
  beat
end

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



124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/wurk/launcher.rb', line 124

def quiet
  return if @done

  @done = true
  @managers.each(&:quiet)
  @poller&.terminate
  # The cron poller is intentionally NOT terminated here: a USR1-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

#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. start the managers FIRST. They are the only thing here on the
 time-to-first-job path (the number bench/vs_sidekiq.rb measures);
 everything below is periodic background work whose first tick is
 seconds away, so it has nothing to gain from going ahead of them.
3. spawn the heartbeat thread, so the dashboard sees the process
 right after it can pick up jobs.
4. start the reaper and the boot-time reclaim sweep — kill-9 recovery,
 not decoration: a SIGKILLed sibling's in-flight jobs wait on this,
 so it stays ahead of the periodic loops.
5. 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.


96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/wurk/launcher.rb', line 96

def run(async_beat: true)
  @started_at = Time.now.to_f
  # Default each capsule's fetcher + materialize its lazy pools/middleware
  # before the config freezes. Every entry point (swarm child, standalone
  # CLI, embedded) runs through here, so none boots with a nil fetcher.
  @config.capsules.each_value(&:prepare!)
  @config.freeze!
  @managers.each(&:start)
  @heartbeat_thread = safe_thread('heartbeat', &method(:start_heartbeat)) if async_beat
  @reaper.start
  # Run on a background thread so /ready probe isn't delayed by a large
  # orphan sweep (reaper.reclaim! is atomic, but can scan many entries).
  @boot_reclaim_thread = safe_thread('boot-reclaim', &method(:boot_reclaim))
  start_periodic_components
  @health_server&.start
rescue StandardError
  # Boot is not atomic: whatever raised (a health-check port already bound,
  # ThreadError at the OS thread limit) leaves the steps before it holding
  # threads, sockets and a leader campaign — and the caller is about to drop
  # its only reference to us, so nothing else can ever release them. Guarded,
  # because the caller must see the boot failure, not a rollback failure.
  teardown_step('boot-rollback') { stop }
  raise
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.



155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/wurk/launcher.rb', line 155

def stop
  @shutdown_gate.run do
    deadline = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + (@config[:timeout] || 25)
    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) 


167
168
169
 
# File 'lib/wurk/launcher.rb', line 167

def stopping?
  @done
end