Class: Wurk::CLI

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

Overview

Standalone CLI. Loads from exe/wurk — never loads wurk/rails so the binary works without the Rails engine (the host app might not be Rails). Singleton because there is exactly one process-wide CLI; tests construct fresh .new instances to keep state isolated.

Spec: docs/target/sidekiq-free.md §21 (Sidekiq::CLI).

Constant Summary collapse

MIN_REDIS_VERSION =

Minimum Redis version Wurk supports — same as Sidekiq 8.x. The job JSON format and Lua scripts rely on commands introduced in Redis 7.

'7.0.0'
COMMANDS =

Subcommands. Bare wurk runs the worker — the historical shape, and the only one the swarm binaries accept — so a subcommand is never anything but the first argument, and every argument after it is that command's.

%w[api].freeze
DEFAULT_API_BIND =

wurk api defaults. 0.0.0.0 because the point of the machine API is for another service to reach it; it is bearer-gated, and answers 404 to everything until a token exists. --bind 127.0.0.1 narrows it.

'0.0.0.0'
DEFAULT_API_PORT = 
7433
NO_API_TOKEN_MESSAGE =

Serving the API with nothing registered would bind a port and 404 every request — the surface is off until a token exists, so say so at boot rather than leaving an operator to diagnose it over HTTP.

<<~MSG
  No API token is registered, so every request would answer 404.
  Register one where the app loaded by -r configures wurk:

      Wurk.configuration.api_token(ENV.fetch('WURK_API_TOKEN'), scopes: %i[enqueue read])
MSG
BACKTRACE_DUMPER =

Thread-backtrace dumper used by both TTIN and INFO. Same body — INFO is the modern name, TTIN is kept for parity with older Sidekiq users.

lambda do |cli|
  Thread.list.each do |thread|
    cli.logger.warn "Thread TID-#{(thread.object_id ^ ::Process.pid).to_s(36)} #{thread.name}"
    if thread.backtrace
      cli.logger.warn thread.backtrace.join("\n")
    else
      cli.logger.warn '<no backtrace available>'
    end
  end
end
SIGNAL_HANDLERS = 
{
  'INT' => ->(_cli) { raise Interrupt },
  'TERM' => ->(_cli) { raise Interrupt },
  'TSTP' => lambda do |cli|
    cli.logger.info 'Received TSTP, no longer accepting new work'
    cli.launcher.quiet
  end,
  'TTIN' => BACKTRACE_DUMPER,
  'INFO' => BACKTRACE_DUMPER,
  'USR2' => lambda do |cli|
    cli.logger.info 'Received USR2, reopening logs'
    # reopen_logs is private — an explicit receiver needs __send__.
    cli.__send__(:reopen_logs)
  end
}.freeze
OPTION_FLAGS =

Table-driven so adding a flag doesn't grow define_value_flags's ABC size and the surface matches the Sidekiq docs row-for-row. The 5th column is the assignment transform: :to_i parses as Integer, :append pushes onto a list (only -q uses that), nil = assign as-is.

[
  ['-c', '--concurrency INT',      :concurrency, 'processor threads to use', :to_i],
  ['-e', '--environment ENV',      :environment, 'Application environment'],
  ['-g', '--tag TAG',              :tag,         'Process tag for procline'],
  ['-q', '--queue QUEUE[,WEIGHT]', :queues,      'Queues to process with optional weights', :append],
  ['-r', '--require [PATH|DIR]',   :require,     'Location of Rails app or .rb file to require'],
  ['-t', '--timeout NUM',          :timeout,     'Shutdown timeout', :to_i],
  ['-v', '--verbose',              :verbose,     'Print more verbose output'],
  ['-C', '--config PATH',          :config_file, 'path to YAML config file']
].freeze
API_OPTION_FLAGS =

Same table shape, offered only when api is the subcommand: none of it means anything to a worker process, and wurk --help stays the worker's.

[
  ['-b', '--bind ADDRESS', :api_bind,   "Address to bind (default #{DEFAULT_API_BIND})"],
  ['-p', '--port PORT',    :api_port,   "Port to listen on (default #{DEFAULT_API_PORT})", :to_i],
  ['-s', '--server NAME',  :api_server, 'Rack handler to serve with (default: the first installed)']
].freeze

Constants included from Component

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

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Component

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

Constructor Details

#initialize ⇒ CLI

Returns a new instance of CLI.



117
118
119
120
121
122
123
 
# File 'lib/wurk/cli.rb', line 117

def initialize
  @config = nil
  @launcher = nil
  @environment = nil
  @parser = nil
  @command = nil
end

Instance Attribute Details

#command ⇒ Object (readonly)

The subcommand parse pulled off argv, or nil for the worker runner. The binary picks the mode from it — exe/wurk runs the API for api.



106
107
108
 
# File 'lib/wurk/cli.rb', line 106

def command
  @command
end

#config ⇒ Object

Returns the value of attribute config.



102
103
104
 
# File 'lib/wurk/cli.rb', line 102

def config
  @config
end

#environment ⇒ Object

Returns the value of attribute environment.



102
103
104
 
# File 'lib/wurk/cli.rb', line 102

def environment
  @environment
end

#launcher ⇒ Object

Returns the value of attribute launcher.



102
103
104
 
# File 'lib/wurk/cli.rb', line 102

def launcher
  @launcher
end

Class Method Details

.instance ⇒ Object 



108
109
110
 
# File 'lib/wurk/cli.rb', line 108

def self.instance
  @instance ||= new
end

.reset_instance! ⇒ Object

Test seam: parallel suites can't share the singleton.



113
114
115
 
# File 'lib/wurk/cli.rb', line 113

def self.reset_instance!
  @instance = nil
end

Instance Method Details

#handle_signal(sig) ⇒ Object 



231
232
233
234
235
236
237
 
# File 'lib/wurk/cli.rb', line 231

def handle_signal(sig)
  logger.debug { "Got #{sig} signal" }
  handler = SIGNAL_HANDLERS[sig]
  return logger.warn("No #{sig} signal handler registered, ignoring") unless handler

  handler.call(self)
end

#parse(args = ARGV.dup) ⇒ Object

parse is split from run so tests can drive option parsing without touching Redis or booting the host app.



127
128
129
130
131
132
133
134
 
# File 'lib/wurk/cli.rb', line 127

def parse(args = ARGV.dup)
  @config ||= Wurk.default_configuration
  @command = extract_command!(args)
  setup_options(args)
  initialize_logger
  validate!
  self
end

#run(boot_app: true, warmup: true) ⇒ Object

boot_app: / warmup: are test seams. Production always passes true.



137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/wurk/cli.rb', line 137

def run(boot_app: true, warmup: true)
  # Mark server mode BEFORE the app loads so `configure_server` blocks in
  # the required initializer actually fire (they gate on `config.server?`).
  # Matches Sidekiq, which sets `Sidekiq.server = true` before requiring
  # the app in its CLI. Skipping this silently drops server middleware,
  # error handlers, and lifecycle hooks registered via configure_server.
  enter_server_mode
  boot_application if boot_app
  self_read, self_write = ::IO.pipe
  begin
    trap_signals(self_write)
    validate_redis!
    validate_pool_sizes!
    @config[:identity] = identity
    # Force lazy server-middleware chain so worker threads don't race
    # against each other constructing it. Spec: Sidekiq::CLI line 104.
    @config.server_middleware
    warm_up_process if warmup
    fire_event(:startup, reverse: false, reraise: true)
    launch(self_read)
  ensure
    # Every exit path — clean return, a raise out of validate/startup, and
    # the SystemExit that `launch` unwinds on Interrupt. Embedded and test
    # callers survive `run`, so a leaked pair is a real FD leak.
    self_read.close
    self_write.close
  end
end

#run_api(boot_app: true, handler: nil) ⇒ Object

wurk api — serve the machine-facing HTTP API and nothing else: no fetcher, no heartbeat, no job ever runs here. This is mount mode 3, the standalone one: the same Rack app the engine nests and a host can mount itself, served without Rails.

Server mode is entered for the reason #run does it — a host that registers its token inside a configure_server block would otherwise boot an API with no credential and 404 every request. Nothing starts as a side effect: no lifecycle event fires and no Launcher is ever built.

handler: is a test seam; production resolves one from the installed gems.

Raises:

  • (ArgumentError) 


217
218
219
220
221
222
223
224
225
226
227
228
229
 
# File 'lib/wurk/cli.rb', line 217

def run_api(boot_app: true, handler: nil)
  enter_server_mode
  boot_application if boot_app
  validate_redis!
  # Load the app here, not on the first request: a broken load should fail
  # the boot an operator is watching, not the first client to call.
  require_relative 'api/app'
  raise ArgumentError, NO_API_TOKEN_MESSAGE unless @config.api_enabled?

  handler ||= api_handler(@config[:api_server])
  logger.info { "Wurk API listening on http://#{api_bind}:#{api_port}#{Wurk::API::VERSION_PREFIX}" }
  handler.run(Wurk::API, Host: api_bind, Port: api_port)
end

#run_swarm(boot_app: true, warmup: true) ⇒ Object

Standalone multi-process boot — the sidekiqswarm entry point (Ent §7). The parent loads the app once, forks N worker children per the configured topology, then supervises them (respawn on crash, rolling restart on SIGUSR1, memory-based recycling). The parent itself never fetches.

This is the only way to get fork-based parallelism without Rails — the railtie auto-boot path is Rails-only. Honors the swarm preload knobs (WURK_PRELOAD/SIDEKIQ_PRELOAD Bundler groups, WURK_PRELOAD_APP/ SIDEKIQ_PRELOAD_APP whole-app eager-load) and boots Process.warmup before the fork so children share warmed pages (copy-on-write). Spec §7.

Raises:

  • (ArgumentError) 


176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/wurk/cli.rb', line 176

def run_swarm(boot_app: true, warmup: true)
  raise ArgumentError, "the swarm runner takes no subcommand; run `wurk #{@command}` instead" if @command

  # Server mode before the app loads — see #run. The flag rides through the
  # fork into every child (the config object is copied), so configure_server
  # blocks registered in the parent take effect in the workers.
  enter_server_mode
  if boot_app
    preload_bundler_groups
    boot_application
    eager_load_application
  end
  validate_redis!
  validate_pool_sizes!
  @config[:identity] = identity
  warm_up_process if warmup
  @swarm = Wurk::Swarm.new(topology: @config.topology, config: @config,
                           shutdown_timeout: @config[:timeout] || Swarm::DEFAULT_SHUTDOWN_TIMEOUT)
  begin
    @swarm.boot(install_signals: true)
    @swarm.supervise
  ensure
    # A fork failure part-way through `boot`, or anything raising out of
    # `supervise`, otherwise leaves live children behind with no supervisor.
    # A no-op once the loop already drained (no children, pipe closed).
    @swarm.shutdown
  end
end