Class: Wurk::Configuration

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/configuration.rb

Overview

Owns runtime knobs (concurrency, queues, timeouts, lifecycle events, error/death handlers) and the registry of Capsules. Single source of truth for everything the swarm / managers / processors need to boot.

Spec: docs/target/sidekiq-free.md §4 (Sidekiq::Config).

Constant Summary collapse

DEFAULTS =

Mirrors Sidekiq::Config::DEFAULTS. Order and keys are part of the drop-in contract — third-party gems read @options via [] / fetch / dig.

{
  labels: Set.new,
  require: '.',
  environment: nil,
  concurrency: 5,
  timeout: 25,
  poll_interval_average: nil,
  average_scheduled_poll_interval: 5,
  on_complex_arguments: :raise,
  max_iteration_runtime: nil,
  error_handlers: [],
  death_handlers: [],
  lifecycle_events: {
    startup: [],
    fork: [],
    quiet: [],
    shutdown: [],
    exit: [],
    heartbeat: [],
    beat: [],
    leader: []
  },
  dead_max_jobs: 10_000,
  dead_timeout_in_seconds: 180 * 24 * 60 * 60,
  reloader: proc { |&b| b.call },
  backtrace_cleaner: ->(bt) { bt },
  logged_job_attributes: %w[bid tags],
  redis_idle_timeout: nil,
  redis_error_handlers: []
}.freeze
LIFECYCLE_EVENTS =

:fork fires only inside swarm children, after fork + internal AR/Redis reconnect — apps reopen sockets / non-fork-safe libs there (Ent §7.4).

%i[startup fork quiet shutdown exit heartbeat beat leader].freeze
DEFAULT_THREAD_PRIORITY = 
-1
REDIS_ERROR_CLASSES =

Redis client / pool errors that the pool wrapper already retried before re-raising. Logged one level up (WARN, not INFO) so a transient blip surfaces in ops dashboards without drowning steady-state noise (#101). RedisClient + ConnectionPool are always loaded before this file (capsule → redis_pool requires both), so referencing them here is safe.

[RedisClient::Error, ConnectionPool::TimeoutError].freeze
ERROR_HANDLER =

Default error handler. Wraps the report in the thread-local Wurk::Context so logger formatters/JSON layouts can pick up jid/bid/tags. full_message (with backtrace) in dev/debug, detailed_message in prod — mirrors the Sidekiq behavior so log scrapers built for one work for both.

Spec: docs/target/sidekiq-free.md §4.3.

lambda do |ex, ctx, cfg = Wurk.configuration|
  safe_ctx = ctx || {}
  Wurk::Context.with(safe_ctx) do
    dev = $DEBUG || ENV['WURK_DEBUG'] || cfg.logger.debug?
    msg = dev ? ex.full_message : ex.detailed_message
    level = REDIS_ERROR_CLASSES.any? { |k| ex.is_a?(k) } ? :warn : :info
    cfg.logger.public_send(level) { msg }
  end
end
WEB_POOL_DEFAULT_SIZE =

Default connection count for the dedicated web pool (#web_redis_pool).

5
WEB_POOL_TIMEOUT =

Deliberately short checkout wait for the web pool: a saturated dashboard should fail fast rather than tie up a web-server thread queuing for a slot.

1.0
HISTORY_DEFAULT_INTERVAL =

--- Historical metrics snapshotter (Ent §5) -------------------------

30

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Configuration

Returns a new instance of Configuration.



90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/wurk/configuration.rb', line 90

def initialize(options = {})
  @options = deep_dup_defaults.merge(options)
  @options[:error_handlers] << ERROR_HANDLER if @options[:error_handlers].empty?
  @capsules = {}
  @directory = {}
  @client_chain = Middleware::Chain.new
  @server_chain = Middleware::Chain.new
  @redis_config = { url: ENV.fetch('REDIS_URL', 'redis://localhost:6379/0') }
  @web_redis_pool = nil
  @logger = nil
  @thread_priority = DEFAULT_THREAD_PRIORITY
  @frozen = false
end

Instance Attribute Details

#capsules ⇒ Object (readonly)

Returns the value of attribute capsules.



78
79
80
 
# File 'lib/wurk/configuration.rb', line 78

def capsules
  @capsules
end

#directory ⇒ Object (readonly)

Returns the value of attribute directory.



78
79
80
 
# File 'lib/wurk/configuration.rb', line 78

def directory
  @directory
end

#dogstatsd ⇒ Object

Pro parity: callable that builds the statsd / dogstatsd client. Invoked once per process AFTER fork; see Wurk::Metrics::Statsd.client. Assignable as a Proc, lambda, or any object responding to #call:

config.dogstatsd = -> { Datadog::Statsd.new('host', 8125) }

Spec: docs/target/sidekiq-pro.md §9.1.



88
89
90
 
# File 'lib/wurk/configuration.rb', line 88

def dogstatsd
  @dogstatsd
end

#logger ⇒ Object

--- Logger -----------------------------------------------------------



408
409
410
 
# File 'lib/wurk/configuration.rb', line 408

def logger
  @logger ||= default_logger
end

#redis_config ⇒ Object (readonly)

Returns the value of attribute redis_config.



78
79
80
 
# File 'lib/wurk/configuration.rb', line 78

def redis_config
  @redis_config
end

#super_fetch_callback ⇒ Object (readonly)

Returns the value of attribute super_fetch_callback.



78
79
80
 
# File 'lib/wurk/configuration.rb', line 78

def super_fetch_callback
  @super_fetch_callback
end

#thread_priority ⇒ Object

Returns the value of attribute thread_priority.



79
80
81
 
# File 'lib/wurk/configuration.rb', line 79

def thread_priority
  @thread_priority
end

Instance Method Details

#[](key) ⇒ Object

--- Hash-like options access -----------------------------------------



106
 
# File 'lib/wurk/configuration.rb', line 106

def [](key) = @options.[](key)

#[]=(key, val) ⇒ Object 



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

def []=(key, val)
  guard_frozen!
  @options[key] = val
end

#average_scheduled_poll_interval=(interval) ⇒ Object 



265
266
267
 
# File 'lib/wurk/configuration.rb', line 265

def average_scheduled_poll_interval=(interval)
  @options[:average_scheduled_poll_interval] = interval
end

#capsule(name) {|cap| ... } ⇒ Object

Yields:

  • (cap) 


151
152
153
154
155
156
 
# File 'lib/wurk/configuration.rb', line 151

def capsule(name)
  name = name.to_s
  cap = @capsules[name] ||= Capsule.new(name, self)
  yield cap if block_given?
  cap
end

#client_middleware {|@client_chain| ... } ⇒ Object

--- Middleware -------------------------------------------------------

Yields:

  • (@client_chain) 


160
161
162
163
 
# File 'lib/wurk/configuration.rb', line 160

def client_middleware
  yield @client_chain if block_given?
  @client_chain
end

#concurrency ⇒ Integer

Returns threads per worker process for the default capsule (default 5). The process count is separate — set via WURK_COUNT (defaults to the CPU count). With a single capsule, total in-flight jobs = WURK_COUNT × concurrency; with multiple capsules see #total_concurrency for the cluster aggregate.

Returns:

  • (Integer) —

    threads per worker process for the default capsule (default 5). The process count is separate — set via WURK_COUNT (defaults to the CPU count). With a single capsule, total in-flight jobs = WURK_COUNT × concurrency; with multiple capsules see #total_concurrency for the cluster aggregate.



130
 
# File 'lib/wurk/configuration.rb', line 130

def concurrency = default_capsule.concurrency

#concurrency=(val) ⇒ Object

Parameters:

  • val (Integer) —

    threads per worker process



133
134
135
 
# File 'lib/wurk/configuration.rb', line 133

def concurrency=(val)
  default_capsule.concurrency = val
end

#configure_client {|_self| ... } ⇒ Object

Yields:

  • (_self)

Yield Parameters:



432
433
434
 
# File 'lib/wurk/configuration.rb', line 432

def configure_client(&block)
  yield self if block && !server?
end

#configure_server {|_self| ... } ⇒ Object

--- Configure blocks (Sidekiq.configure_server / _client) -----------

Yields:

  • (_self)

Yield Parameters:



428
429
430
 
# File 'lib/wurk/configuration.rb', line 428

def configure_server(&block)
  yield self if block && server?
end

#death_handlers ⇒ Object 



248
249
250
 
# File 'lib/wurk/configuration.rb', line 248

def death_handlers
  @options[:death_handlers]
end

#default_capsule ⇒ Object 



147
148
149
 
# File 'lib/wurk/configuration.rb', line 147

def default_capsule(&)
  capsule('default', &)
end

#dig(*keys) ⇒ Object 



121
 
# File 'lib/wurk/configuration.rb', line 121

def dig(*keys) = @options.dig(*keys)

#error_handlers ⇒ Object

--- Handlers ---------------------------------------------------------



244
245
246
 
# File 'lib/wurk/configuration.rb', line 244

def error_handlers
  @options[:error_handlers]
end

#fetch ⇒ Object 



113
 
# File 'lib/wurk/configuration.rb', line 113

def fetch(*, &) = @options.fetch(*, &)

#fetch_poll_interval ⇒ Object 



278
279
280
 
# File 'lib/wurk/configuration.rb', line 278

def fetch_poll_interval
  @options[:fetch_poll_interval]
end

#fetch_poll_interval=(seconds) ⇒ Object

Reliable-fetch empty-poll backoff: the BLMOVE block timeout (seconds) used when every served queue is empty. Pro super_fetch §3.3's fetch_poll_interval knob. Unset (nil) → the fetcher's default (Wurk::Fetcher::Reliable::TIMEOUT, 2s). Also readable as config[:fetch_poll_interval].



274
275
276
 
# File 'lib/wurk/configuration.rb', line 274

def fetch_poll_interval=(seconds)
  @options[:fetch_poll_interval] = seconds
end

#freeze! ⇒ Object 



473
474
475
476
477
478
479
480
481
482
 
# File 'lib/wurk/configuration.rb', line 473

def freeze!
  return self if @frozen

  @capsules.each_value(&:freeze)
  @capsules.freeze
  @options.freeze
  @directory.freeze
  @frozen = true
  self
end

#frozen? ⇒ Boolean

Returns:

  • (Boolean) 


484
485
486
 
# File 'lib/wurk/configuration.rb', line 484

def frozen?
  @frozen
end

#handle_exception(ex, ctx = {}) ⇒ Object 



414
415
416
417
418
419
420
421
422
423
424
 
# File 'lib/wurk/configuration.rb', line 414

def handle_exception(ex, ctx = {})
  if error_handlers.empty?
    logger.error("#{ctx} #{ex.class}: #{ex.message}")
  else
    error_handlers.each do |handler|
      handler.call(ex, ctx, self)
    rescue StandardError => e
      logger.error("error_handler raised: #{e.class}: #{e.message}")
    end
  end
end

#health_check(port:, bind: '0.0.0.0', ready_window: 30) ⇒ Object

Opt-in thin HTTP listener inside the worker process for k8s probes. When called, the Launcher will start a TCP server on port bound to bind exposing GET /live (200 while not stopping) and GET /ready (200 only when Redis is reachable AND heartbeat fired within ready_window seconds).

Off by default — call this in a configure_server block to enable. Spec: docs/target/sidekiq-ent.md §7.1.2.

Raises:

  • (ArgumentError) 


382
383
384
385
386
387
388
389
390
391
392
393
 
# File 'lib/wurk/configuration.rb', line 382

def health_check(port:, bind: '0.0.0.0', ready_window: 30)
  guard_frozen!
  p = Integer(port)
  rw = Integer(ready_window)
  raise ArgumentError, 'port must be between 0 and 65535' unless (0..65535).cover?(p)
  raise ArgumentError, 'ready_window must be > 0' unless rw.positive?

  b = bind.to_s
  raise ArgumentError, 'bind must be a non-empty string' if b.empty?

  @options[:health_check_options] = { port: p, bind: b, ready_window: rw }
end

#history_collector ⇒ Object 



357
 
# File 'lib/wurk/configuration.rb', line 357

def history_collector = @options[:history_collector]

#history_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


355
 
# File 'lib/wurk/configuration.rb', line 355

def history_enabled? = @options.key?(:history_interval)

#history_interval ⇒ Object 



356
 
# File 'lib/wurk/configuration.rb', line 356

def history_interval = @options.fetch(:history_interval, HISTORY_DEFAULT_INTERVAL)

#inspect ⇒ Object 



488
489
490
 
# File 'lib/wurk/configuration.rb', line 488

def inspect
  "#<#{self.class} capsules=#{@capsules.keys} concurrency=#{total_concurrency}>"
end

#key?(key) ⇒ Boolean Also known as: has_key?

Returns:

  • (Boolean) 


114
 
# File 'lib/wurk/configuration.rb', line 114

def key?(key) = @options.key?(key)

#lookup(name, default_class = nil) ⇒ Object 



238
239
240
 
# File 'lib/wurk/configuration.rb', line 238

def lookup(name, default_class = nil)
  @directory[name] ||= default_class&.new
end

#memory_limit_kb ⇒ Object

Threshold in KB, the unit the swarm compares against /proc//statm (pages × 4KB). nil when recycling is disabled.



468
469
470
471
 
# File 'lib/wurk/configuration.rb', line 468

def memory_limit_kb
  mb = memory_limit_mb
  mb&.positive? ? mb * 1024 : nil
end

#memory_limit_mb ⇒ Object

Memory-based child recycling (Sidekiq Ent §7.5): the swarm parent TERMs (and respawns) any child whose RSS exceeds this many MB. Set in code or via SIDEKIQ_MAXMEM_MB (WURK_MAXMEM_MB is the native alias); an explicit value wins over the env. nil/0 disables recycling (the default).



457
458
459
 
# File 'lib/wurk/configuration.rb', line 457

def memory_limit_mb
  @memory_limit_mb || env_memory_limit_mb
end

#memory_limit_mb=(value) ⇒ Object 



461
462
463
464
 
# File 'lib/wurk/configuration.rb', line 461

def memory_limit_mb=(value)
  guard_frozen!
  @memory_limit_mb = value.nil? ? nil : Integer(value)
end

#merge!(other) ⇒ Object 



116
117
118
119
 
# File 'lib/wurk/configuration.rb', line 116

def merge!(other)
  guard_frozen!
  @options.merge!(other)
end

#new_redis_pool(size, name = 'custom') ⇒ Object 



190
191
192
 
# File 'lib/wurk/configuration.rb', line 190

def new_redis_pool(size, name = 'custom')
  build_redis_pool(size: size, name: name)
end

#on(event, &block) ⇒ Object

--- Lifecycle hooks --------------------------------------------------

Raises:

  • (ArgumentError) 


397
398
399
400
401
402
403
404
 
# File 'lib/wurk/configuration.rb', line 397

def on(event, &block)
  raise ArgumentError, "block required for on(#{event.inspect})" unless block
  unless LIFECYCLE_EVENTS.include?(event)
    raise ArgumentError, "invalid event #{event.inspect}, must be one of #{LIFECYCLE_EVENTS.inspect}"
  end

  @options[:lifecycle_events][event] << block
end

#on_redis_error(&block) ⇒ Object

Telemetry hook fired by RedisPool on every transient-error retry and final give-up. The block receives one Hash: { error:, attempt:, retried:, pool: }. Opt-in — pools stay silent until a handler is registered.

Raises:

  • (ArgumentError) 


255
256
257
258
259
 
# File 'lib/wurk/configuration.rb', line 255

def on_redis_error(&block)
  raise ArgumentError, 'block required for on_redis_error' unless block

  @options[:redis_error_handlers] << block
end

#periodic {|mgr| ... } ⇒ Wurk::Cron::Manager

Yields a Wurk::Cron::Manager so the host app can register periodic jobs at boot. Manager state is shared per-process so multiple config.periodic blocks accumulate (matches Sidekiq Ent §2.1). This is the native replacement for the sidekiq-cron gem.

Spec: docs/target/sidekiq-ent.md §2.

Examples:

Register cron jobs at boot

Wurk.configure_server do |config|
  config.periodic do |mgr|
    mgr.register("*/5 * * * *", ReportJob)
    mgr.register("0 0 * * *", NightlyJob, tz: "UTC")
  end
end

Yield Parameters:

Returns:



326
327
328
329
330
331
 
# File 'lib/wurk/configuration.rb', line 326

def periodic
  require_relative 'cron'
  @periodic_manager ||= Wurk::Cron::Manager.new(self)
  yield @periodic_manager if block_given?
  @periodic_manager
end

#queues ⇒ Object 



137
 
# File 'lib/wurk/configuration.rb', line 137

def queues = default_capsule.queues

#queues=(val) ⇒ Object 



139
140
141
 
# File 'lib/wurk/configuration.rb', line 139

def queues=(val)
  default_capsule.queues = val
end

#redis ⇒ Object 



194
195
196
 
# File 'lib/wurk/configuration.rb', line 194

def redis(&)
  redis_pool.with(&)
end

#redis=(hash) ⇒ Object

--- Redis ------------------------------------------------------------



172
173
174
175
 
# File 'lib/wurk/configuration.rb', line 172

def redis=(hash)
  guard_frozen!
  @redis_config = @redis_config.merge(hash.transform_keys(&:to_sym))
end

#redis_error_handlers ⇒ Object 



261
262
263
 
# File 'lib/wurk/configuration.rb', line 261

def redis_error_handlers
  @options[:redis_error_handlers]
end

#redis_pool ⇒ Object 



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

def redis_pool
  default_capsule.redis_pool
end

#register(name, instance) ⇒ Object

--- Service locator (extension registry) ----------------------------



233
234
235
236
 
# File 'lib/wurk/configuration.rb', line 233

def register(name, instance)
  guard_frozen!
  @directory[name] = instance
end

#reliable_scheduler! ⇒ Object

Pro reliable scheduler (§4): promote due jobs from retry/schedule onto their target queue in a single atomic Lua (ZRANGEBYSCORE+ZREM+LPUSH), closing the pop→push job-loss window of the default poller. Swaps the pluggable scheduled_enq for the atomic promoter; idempotent.



303
304
305
306
 
# File 'lib/wurk/configuration.rb', line 303

def reliable_scheduler!(*)
  self[:scheduled_enq] = Wurk::Scheduled::ReliableEnq
  nil
end

#reset_redis_pools! ⇒ Object

Disconnect and drop every capsule's cached pools (main + fetch) plus the web pool. Used by Wurk::Swarm so the parent never leaks sockets into forks and each child can build fresh ones.



184
185
186
187
188
 
# File 'lib/wurk/configuration.rb', line 184

def reset_redis_pools!
  @capsules.each_value(&:reset_redis_pools!)
  @web_redis_pool&.disconnect!
  @web_redis_pool = nil
end

#retain_history(seconds = HISTORY_DEFAULT_INTERVAL, &block) ⇒ Object

Enables the Ent Historical Metrics snapshotter: every seconds the cluster leader emits a statsd-shaped snapshot to the configured dogstatsd client. With no block the default §5.2 gauge set is published; a block receives the dogstatsd client s and collects custom metrics instead. The Launcher starts the snapshotter only when this has been called.

Spec: docs/target/sidekiq-ent.md §5.1.

Raises:

  • (ArgumentError) 


345
346
347
348
349
350
351
352
353
 
# File 'lib/wurk/configuration.rb', line 345

def retain_history(seconds = HISTORY_DEFAULT_INTERVAL, &block)
  guard_frozen!
  interval = Float(seconds)
  raise ArgumentError, 'retain_history interval must be > 0' unless interval.positive?

  @options[:history_interval] = interval
  @options[:history_collector] = block
  nil
end

#server? ⇒ Boolean

Returns:

  • (Boolean) 


436
437
438
 
# File 'lib/wurk/configuration.rb', line 436

def server?
  @options[:server] == true
end

#server_middleware {|@server_chain| ... } ⇒ Object

Yields:

  • (@server_chain) 


165
166
167
168
 
# File 'lib/wurk/configuration.rb', line 165

def server_middleware
  yield @server_chain if block_given?
  @server_chain
end

#super_fetch!(&block) ⇒ Object

Sidekiq Pro's opt-in toggles for reliable fetch and the reliable scheduler. Both are already the default in Wurk — the fetcher is always the reliable BLMOVE fetcher with orphan reclamation, and the scheduler is always atomic Lua — so the toggle itself is a no-op. They exist only so a Pro initializer drops in unchanged instead of raising NoMethodError.

The optional block is Pro's recovery callback: |jobstr, pill|, fired once per orphan recovery (pill nil) and once on a poison kill (pill responds to .jid/.klass/.count/.queue). The reaper drives it via Wurk::Middleware::PoisonPill.track!. Spec: docs/target/sidekiq-pro.md §3.1.



294
295
296
297
 
# File 'lib/wurk/configuration.rb', line 294

def super_fetch!(*, &block)
  @super_fetch_callback = block if block
  nil
end

#topology ⇒ Object

Worker topology for the swarm. When the host hasn't declared one (the railtie path), default to a single flat fork running the default capsule's queues + concurrency. Assign a custom Wurk::Topology (via topology=) for specialized slots.



444
445
446
 
# File 'lib/wurk/configuration.rb', line 444

def topology
  @topology ||= default_topology
end

#topology=(value) ⇒ Object 



448
449
450
451
 
# File 'lib/wurk/configuration.rb', line 448

def topology=(value)
  guard_frozen!
  @topology = value
end

#total_concurrency ⇒ Object 



143
144
145
 
# File 'lib/wurk/configuration.rb', line 143

def total_concurrency
  @capsules.each_value.sum(&:concurrency)
end

#web ⇒ Object

Web UI configuration: the authorization hook and read-only mode. Returns the process-wide Wurk::Web.config singleton so config.web.read_only = true and the engine middleware share one source of truth. Lazy-requires the web layer to keep standalone boot lean.

Spec: docs/target/sidekiq-ent.md §9.2.



367
368
369
370
 
# File 'lib/wurk/configuration.rb', line 367

def web
  require_relative 'web/config'
  Wurk::Web.config
end

#web_pool_size ⇒ Object

Connections in the dedicated web pool. config.web_pool_size = N overrides the default; independent of config.redis[:size], which sizes the worker capsules — the two pools are deliberately disjoint (#101).



210
211
212
 
# File 'lib/wurk/configuration.rb', line 210

def web_pool_size
  @options[:web_pool_size] || WEB_POOL_DEFAULT_SIZE
end

#web_pool_size=(size) ⇒ Object 



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

def web_pool_size=(size)
  guard_frozen!
  @options[:web_pool_size] = Integer(size)
end

#web_redis_pool ⇒ Object

Dedicated Redis pool for the dashboard / JSON API / SSE, disjoint from every worker capsule's pool. Dashboard load — an API burst, a long-lived SSE stream — can no longer drain the connections a co-located (embedded) worker needs to fetch and heartbeat, and vice versa: the #101 0/N pool-exhaustion incident. Lazy, so a headless worker that never serves the dashboard builds nothing; web entry points route Wurk.redis here through Wurk::Web::PoolScope. The Configuration instance is never frozen (only its @options/@capsules are), so this ||= is safe to fire post-boot.



227
228
229
 
# File 'lib/wurk/configuration.rb', line 227

def web_redis_pool
  @web_redis_pool ||= build_redis_pool(size: web_pool_size, name: 'web', pool_timeout: WEB_POOL_TIMEOUT)
end