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: [],
  # Wurk extras, appended after the mirrored keys so the Sidekiq prefix
  # above stays byte-for-byte what a gem reading @options expects.
  #
  # Lifetime of a `status:<jid>` row (Wurk::Status), re-stamped on every
  # write, and the separate lifetime a `complete` row gets once the job
  # succeeded. Retention is off by default — nil means a finished job's
  # row expires on the same clock as a running one. Set it to seconds to
  # keep succeeded jobs around longer than they ran, or to 0 for Sidekiq's
  # own behavior, where a job that succeeds leaves nothing behind.
  status_ttl: Keys::STATUS_TTL,
  status_retention: nil,
  # Distributed tracing (Wurk::Telemetry). Seeded here, not written on
  # demand, so the read side answers on a config that `freeze!` has already
  # closed — and so a gem inspecting @options sees the same key set whether
  # or not the host traces. Flipping it is only half the gate; the
  # opentelemetry-api gem has to be present too (Wurk::Telemetry.enabled?).
  telemetry: false,
  # Bearer tokens for the machine-facing HTTP API (Wurk::API), as
  # `token => [scopes]`. Empty means the API does not exist: the engine
  # skips the mount and a directly mounted app answers 404, so a host that
  # never registers a token carries no new HTTP surface at all. Seeded here
  # rather than written on demand so the read side answers on a config that
  # `freeze!` has already closed.
  api_tokens: {},
  # Boundary caps for the HTTP API's produce plane. A body larger than
  # `api_max_body_bytes` is refused before it is parsed; a job whose args
  # serialize larger than `api_max_args_bytes` before it is pushed — the
  # second is the one that bounds what a bulk request lands in Redis, since
  # one request becomes many payloads. `api_idempotency_ttl` is how long an
  # `Idempotency-Key` is remembered: a replay window for a producer
  # retrying a lost response, not an audit log.
  api_max_body_bytes: 1_048_576,
  api_max_args_bytes: 65_536,
  api_idempotency_ttl: 3_600,
  # Allow-list of classes the HTTP API may enqueue. nil (the default) means
  # only an `admin` token may enqueue at all — see #api_enqueue_classes=.
  api_enqueue_classes: nil,
  # Three-state on purpose — see #api_read_only=. nil is "inherit", which
  # only mount mode 1 has anything to inherit from.
  api_read_only: nil,
  # Per-token request ceiling. nil is off: the API adds no Redis work to a
  # request until a host asks for a limit, and no number picked here would
  # be right for both a cron relay and a fan-out producer.
  api_rate_limit: nil,
  api_rate_limit_interval: :minute,
  # Cluster-wide ceiling per queue (Wurk::QueueSlot), as
  # `{ 'critical' => 20 }`. Empty is the whole point of seeding it here:
  # the fetch path reads this once at boot, so an app that caps nothing
  # resolves to "no capped queues" before the first fetch and never asks
  # again.
  global_concurrency: {}
}.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.



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/wurk/configuration.rb', line 153

def initialize(options = {})
  @options = deep_dup_defaults.merge(options)
  # Through the same door `global_concurrency=` uses, so a cap passed to
  # `Configuration.new` is validated rather than trusted, and the default
  # Hash arrives frozen: the fetch path resolves caps once at boot, so a
  # Hash still mutable here is one a caller can add a queue to and have
  # nothing read it. FrozenError is the honest answer to that.
  @options[:global_concurrency] = normalize_global_concurrency(@options[:global_concurrency])
  @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
  @telemetry_installed = false
  @frozen = false
end

Instance Attribute Details

#capsules ⇒ Object (readonly)

Returns the value of attribute capsules.



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

def capsules
  @capsules
end

#directory ⇒ Object (readonly)

Returns the value of attribute directory.



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

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.



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

def dogstatsd
  @dogstatsd
end

#logger ⇒ Object

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



693
694
695
 
# File 'lib/wurk/configuration.rb', line 693

def logger
  @logger ||= default_logger
end

#redis_config ⇒ Object (readonly)

Returns the value of attribute redis_config.



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

def redis_config
  @redis_config
end

#super_fetch_callback ⇒ Object (readonly)

Returns the value of attribute super_fetch_callback.



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

def super_fetch_callback
  @super_fetch_callback
end

#thread_priority ⇒ Object

Returns the value of attribute thread_priority.



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

def thread_priority
  @thread_priority
end

Instance Method Details

#[](key) ⇒ Object

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



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

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

#[]=(key, val) ⇒ Object 



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

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

#api_enabled? ⇒ Boolean

Returns whether the HTTP API exists at all. The engine's mount is conditional on this.

Returns:

  • (Boolean) —

    whether the HTTP API exists at all. The engine's mount is conditional on this.



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

def api_enabled? = !@options[:api_tokens].empty?

#api_enqueue_classes ⇒ Array<String>, ...

Returns classes the HTTP API may enqueue.

Returns:

  • (Array<String>, :any, nil) —

    classes the HTTP API may enqueue.



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

def api_enqueue_classes = @options[:api_enqueue_classes]

#api_enqueue_classes=(classes) ⇒ Object

The allow-list JobUtil::TRANSIENT_ATTRIBUTES deliberately is not — that is a strip-list, right for a producer already running your code:

config.api_enqueue_classes = %w[Billing::Charge ReportJob]

Unset, only an admin token may enqueue at all. class is a code selector — a producer token that can name any constant in the host is choosing which of its jobs to run — so an enqueue-scoped credential has to be told exactly what it is for. Once a list exists it binds every token, admin included: a host that writes one means it.

:any turns the check off, which is what an enqueue-scoped relay that legitimately pushes arbitrary classes needs — the alternative is handing it admin and the destructive routes that come with it.



551
552
553
554
555
 
# File 'lib/wurk/configuration.rb', line 551

def api_enqueue_classes=(classes)
  guard_frozen!
  require_relative 'api/validation'
  @options[:api_enqueue_classes] = Wurk::API::Validation.enqueueable_classes!(classes)
end

#api_idempotency_ttl ⇒ Integer

Returns seconds an Idempotency-Key stays replayable.

Returns:

  • (Integer) —

    seconds an Idempotency-Key stays replayable.



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

def api_idempotency_ttl = @options[:api_idempotency_ttl]

#api_idempotency_ttl=(seconds) ⇒ Object 



530
531
532
 
# File 'lib/wurk/configuration.rb', line 530

def api_idempotency_ttl=(seconds)
  @options[:api_idempotency_ttl] = api_limit!(:api_idempotency_ttl, seconds)
end

#api_max_args_bytes ⇒ Integer

Returns largest args any single job may serialize to.

Returns:

  • (Integer) —

    largest args any single job may serialize to.



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

def api_max_args_bytes = @options[:api_max_args_bytes]

#api_max_args_bytes=(bytes) ⇒ Object 



523
524
525
 
# File 'lib/wurk/configuration.rb', line 523

def api_max_args_bytes=(bytes)
  @options[:api_max_args_bytes] = api_limit!(:api_max_args_bytes, bytes)
end

#api_max_body_bytes ⇒ Integer

Returns largest request body the produce plane will read.

Returns:

  • (Integer) —

    largest request body the produce plane will read.



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

def api_max_body_bytes = @options[:api_max_body_bytes]

#api_max_body_bytes=(bytes) ⇒ Object 



516
517
518
 
# File 'lib/wurk/configuration.rb', line 516

def api_max_body_bytes=(bytes)
  @options[:api_max_body_bytes] = api_limit!(:api_max_body_bytes, bytes)
end

#api_rate_limit ⇒ Integer?

Returns requests one token may make per interval.

Returns:

  • (Integer, nil) —

    requests one token may make per interval.



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

def api_rate_limit = @options[:api_rate_limit]

#api_rate_limit=(count) ⇒ Object

Requests per api_rate_limit_interval, per token, across the fleet — enforced by a Wurk::Limiter sliding window rather than a second limiter, so an API ceiling shows up on the Limiters page beside every other one and resets on the same Redis clock. nil turns it off.



597
598
599
600
 
# File 'lib/wurk/configuration.rb', line 597

def api_rate_limit=(count)
  guard_frozen!
  @options[:api_rate_limit] = count.nil? ? nil : api_limit!(:api_rate_limit, count)
end

#api_rate_limit_interval ⇒ Symbol, Integer

Returns the window the limit is counted over.

Returns:

  • (Symbol, Integer) —

    the window the limit is counted over.



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

def api_rate_limit_interval = @options[:api_rate_limit_interval]

#api_rate_limit_interval=(interval) ⇒ Object

:second :minute :hour :day or a raw Integer of seconds — whatever Limiter.window accepts, validated here so a typo raises in the initializer that wrote it rather than on the first throttled request.



608
609
610
611
612
 
# File 'lib/wurk/configuration.rb', line 608

def api_rate_limit_interval=(interval)
  guard_frozen!
  Wurk::Limiter.interval_seconds(interval, allow_integer: true)
  @options[:api_rate_limit_interval] = interval
end

#api_read_only ⇒ Boolean?

Returns true/false when the host said so, nil to inherit.

Returns:

  • (Boolean, nil) —

    true/false when the host said so, nil to inherit.



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

def api_read_only = @options[:api_read_only]

#api_read_only=(value) ⇒ Object

Freezes the API's writes — enqueue included, not only the destructive admin routes. Same rule as the dashboard's Wurk::Web.config.read_only and deliberately the same sentence: a read-only deployment answers reads. Carving enqueue out would mean a viewer-only deploy could still be made to run arbitrary work, and would leave an operator holding two different definitions of the same word.

Three states, because the two planes are not always one deployment:

nil (default)  inherit. Mounted inside the engine, the API is part of
             the dashboard's deployment and `WURK_WEB_READ_ONLY=1`
             freezes both. Mounted on its own path or run standalone
             there is nothing to inherit from and this means off.
true           frozen wherever it is mounted. `WURK_API_READ_ONLY=1`
             sets it, which is the only door mode 3 has.
false          live even inside a read-only dashboard — the host that
             wants a viewer-only UI and a working producer on the
             same mount, said out loud.


578
579
580
581
 
# File 'lib/wurk/configuration.rb', line 578

def api_read_only=(value)
  guard_frozen!
  @options[:api_read_only] = value.nil? ? nil : truthy_setting?(value)
end

#api_read_only? ⇒ Boolean

Returns the resolved verdict for a mount with nothing to inherit; App layers mode 1's inherited flag over it.

Returns:

  • (Boolean) —

    the resolved verdict for a mount with nothing to inherit; App layers mode 1's inherited flag over it.



585
586
587
588
 
# File 'lib/wurk/configuration.rb', line 585

def api_read_only?
  value = @options[:api_read_only]
  value.nil? ? ENV['WURK_API_READ_ONLY'] == '1' : value
end

#api_token(token, scopes:) ⇒ Object

Registers a bearer token for the HTTP API and the scopes it grants — any of :enqueue, :read, :admin (which grants the other two):

# config/initializers/wurk.rb
Wurk.configuration.api_token ENV.fetch('WURK_API_TOKEN'), scopes: %i[enqueue read]

Registered unconditionally, not inside configure_server/configure_client — the process that serves the API has to be the one holding the credential, and neither block runs in every such process. A Puma-cluster web process never enters server mode (Wurk::RailsBoot refuses to fork a swarm from one), so a token registered in configure_server would be absent from the exact process serving the engine-nested mount.

Registering the first token is what brings the API into existence; with none the app is never mounted. Re-registering a token replaces its scopes. Loading wurk/api/auth is deferred to this call so a host that serves no HTTP API never pays for openssl/digest in every swarm child.



498
499
500
501
502
503
504
 
# File 'lib/wurk/configuration.rb', line 498

def api_token(token, scopes:)
  guard_frozen!
  require_relative 'api/auth'
  value, granted = Wurk::API::Auth.credential!(token, scopes)
  @options[:api_tokens][value] = granted
  nil
end

#api_tokens ⇒ Hash{String => Array<Symbol>}

Returns registered tokens and their scopes.

Returns:

  • (Hash{String => Array<Symbol>}) —

    registered tokens and their scopes



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

def api_tokens = @options[:api_tokens]

#average_scheduled_poll_interval=(interval) ⇒ Object 



346
347
348
 
# File 'lib/wurk/configuration.rb', line 346

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

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

Yields:

  • (cap) 


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

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) 


230
231
232
233
 
# File 'lib/wurk/configuration.rb', line 230

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.



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

def concurrency = default_capsule.concurrency

#concurrency=(val) ⇒ Object

Parameters:

  • val (Integer) —

    threads per worker process



203
204
205
 
# File 'lib/wurk/configuration.rb', line 203

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

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

Yields:

  • (_self)

Yield Parameters:



717
718
719
 
# File 'lib/wurk/configuration.rb', line 717

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

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

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

Yields:

  • (_self)

Yield Parameters:



713
714
715
 
# File 'lib/wurk/configuration.rb', line 713

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

#death_handlers ⇒ Object 



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

def death_handlers
  @options[:death_handlers]
end

#default_capsule ⇒ Object 



217
218
219
 
# File 'lib/wurk/configuration.rb', line 217

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

#dig(*keys) ⇒ Object 



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

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

#error_handlers ⇒ Object

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



325
326
327
 
# File 'lib/wurk/configuration.rb', line 325

def error_handlers
  @options[:error_handlers]
end

#fetch ⇒ Object 



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

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

#fetch_poll_interval ⇒ Object 



359
360
361
 
# File 'lib/wurk/configuration.rb', line 359

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



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

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

#freeze! ⇒ Object

Guarded on the capsule table, not @frozen: a swarm child reaches this with prepare_for_fork! already run in its parent, so @frozen is true while the capsules it has just configured are still open.



785
786
787
788
789
790
791
792
 
# File 'lib/wurk/configuration.rb', line 785

def freeze!
  return self if @capsules.frozen?

  prepare_for_fork!
  @capsules.each_value(&:freeze)
  @capsules.freeze
  self
end

#frozen? ⇒ Boolean

Returns:

  • (Boolean) 


794
795
796
 
# File 'lib/wurk/configuration.rb', line 794

def frozen?
  @frozen
end

#global_concurrency ⇒ Hash{String => Integer}

Returns queue name → cluster-wide ceiling. Always frozen — the default included, normalized in #initialize; see #global_concurrency=.

Returns:

  • (Hash{String => Integer}) —

    queue name → cluster-wide ceiling. Always frozen — the default included, normalized in #initialize; see #global_concurrency=.



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

def global_concurrency = @options[:global_concurrency]

#global_concurrency=(caps) ⇒ Object

Caps how many jobs from a queue may run at once across the whole cluster, whatever the worker count:

Wurk.configure_server { |config| config.global_concurrency = { critical: 20 } }

Not the same thing as concurrency, which is threads in one process, nor as a Limiter, which is keyed per lock and gates inside the job (by which point the job is already claimed off the queue). This is a ceiling on the queue itself, enforced where the work is picked up.

Names are normalized to Strings and the result is frozen, because the fetch path resolves it once at boot: a cap added by mutating this Hash afterwards would be read by nothing, so it raises rather than doing nothing. Assign a new Hash to change one.



635
636
637
638
 
# File 'lib/wurk/configuration.rb', line 635

def global_concurrency=(caps)
  guard_frozen!
  @options[:global_concurrency] = normalize_global_concurrency(caps)
end

#global_concurrency? ⇒ Boolean

Returns whether any queue is capped. The one question the fetcher asks at boot to decide whether it has a gate to run at all.

Returns:

  • (Boolean) —

    whether any queue is capped. The one question the fetcher asks at boot to decide whether it has a gate to run at all.



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

def global_concurrency? = !@options[:global_concurrency].empty?

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



699
700
701
702
703
704
705
706
707
708
709
 
# File 'lib/wurk/configuration.rb', line 699

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) 


667
668
669
670
671
672
673
674
675
676
677
678
 
# File 'lib/wurk/configuration.rb', line 667

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..65_535).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 



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

def history_collector = @options[:history_collector]

#history_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


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

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

#history_interval ⇒ Object 



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

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

#inspect ⇒ Object 



798
799
800
 
# File 'lib/wurk/configuration.rb', line 798

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

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

Returns:

  • (Boolean) 


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

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

#lookup(name, default_class = nil) ⇒ Object

Memoizes on a miss, and the first miss can land long after boot (an extension resolved on its first tick), which is why freeze! leaves @directory writable: frozen, that lookup raised FrozenError instead of building the default. register — the host-facing half — still refuses writes past the freeze, so the closed surface is unchanged.



319
320
321
 
# File 'lib/wurk/configuration.rb', line 319

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.



753
754
755
756
 
# File 'lib/wurk/configuration.rb', line 753

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



742
743
744
 
# File 'lib/wurk/configuration.rb', line 742

def memory_limit_mb
  @memory_limit_mb || env_memory_limit_mb
end

#memory_limit_mb=(value) ⇒ Object 



746
747
748
749
 
# File 'lib/wurk/configuration.rb', line 746

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

#merge!(other) ⇒ Object 



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

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

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



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

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

#on(event, &block) ⇒ Object

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

Raises:

  • (ArgumentError) 


682
683
684
685
686
687
688
689
 
# File 'lib/wurk/configuration.rb', line 682

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) 


336
337
338
339
340
 
# File 'lib/wurk/configuration.rb', line 336

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:



411
412
413
414
415
416
 
# File 'lib/wurk/configuration.rb', line 411

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

#prepare_for_fork! ⇒ Object

The pre-fork half of freeze!, and the only half a forking parent can run: capsules stay writable until each child has applied its slot (ChildBoot#apply_slot_to_config) and opened its own Redis pools. What is left is slot-independent — the options Hash and every capsule's middleware chains — so the swarm parent settles it once and every child inherits the result copy-on-write instead of allocating and dirtying its own copy. Freezing the options here also makes a post-fork option write raise in the child that wrote it, rather than silently diverging from its siblings.

@directory is deliberately left out — see #lookup.



768
769
770
771
772
773
774
775
776
777
778
779
780
 
# File 'lib/wurk/configuration.rb', line 768

def prepare_for_fork!
  # The capsule every swarm child configures (ChildBoot reaches for it by
  # name), and the one a client-only config builds on its first enqueue.
  # Materialized here so its chains are shared rather than rebuilt N times
  # — and so `freeze!` can't close `@capsules` around a name that is only
  # ever resolved later, which turned that first resolution into a
  # FrozenError on the frozen Hash.
  default_capsule
  @capsules.each_value(&:prepare_shared!)
  @options.freeze
  @frozen = true
  self
end

#queues ⇒ Object 



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

def queues = default_capsule.queues

#queues=(val) ⇒ Object 



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

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

#redis(idempotent: false) ⇒ Object 



270
271
272
 
# File 'lib/wurk/configuration.rb', line 270

def redis(idempotent: false, &)
  PoolCheckout.trusted(redis_pool, idempotent, &)
end

#redis=(hash) ⇒ Object

Validated here, in the process running the initializer, rather than later in whichever process first builds a pool. The swarm's children are the ones that construct pools, so a bad key used to kill every child on boot while the parent stayed up and healthy — Running pod, passing probe, zero jobs processed (#283).



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

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

#redis_error_handlers ⇒ Object 



342
343
344
 
# File 'lib/wurk/configuration.rb', line 342

def redis_error_handlers
  @options[:redis_error_handlers]
end

#redis_pool ⇒ Object 



253
254
255
 
# File 'lib/wurk/configuration.rb', line 253

def redis_pool
  default_capsule.redis_pool
end

#register(name, instance) ⇒ Object

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



309
310
311
312
 
# File 'lib/wurk/configuration.rb', line 309

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.



388
389
390
391
 
# File 'lib/wurk/configuration.rb', line 388

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.



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

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) 


430
431
432
433
434
435
436
437
438
 
# File 'lib/wurk/configuration.rb', line 430

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) 


721
722
723
 
# File 'lib/wurk/configuration.rb', line 721

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

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

Yields:

  • (@server_chain) 


235
236
237
238
 
# File 'lib/wurk/configuration.rb', line 235

def server_middleware
  yield @server_chain if block_given?
  @server_chain
end

#super_fetch!(&block) ⇒ Object

Sidekiq Pro's opt-in toggle for reliable fetch. Already the default in Wurk — the fetcher is always the reliable BLMOVE fetcher with orphan reclamation — so the toggle is a no-op beyond capturing the recovery callback. It exists so a Pro initializer drops in unchanged instead of raising NoMethodError.

NOTE: reliable_scheduler! below is NOT a no-op. The default scheduled_enq pops then pushes and has a job-loss window (Wurk::Scheduled::Enq); only the toggle swaps in the atomic promoter.

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.



379
380
381
382
 
# File 'lib/wurk/configuration.rb', line 379

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

#telemetry=(enabled) ⇒ Object

Opt into Wurk::Telemetry — a producer span carrying W3C trace context on the job hash at enqueue, a linked consumer span on execute:

Wurk.configure_server { |config| config.telemetry = true }

Loading wurk/telemetry is deferred to this call, which is the only reason require "wurk" can leave opentelemetry-api untouched in an app that doesn't trace.

Warns when the host opted in but the gem is missing: tracing then stays off, and a silent no-op is indistinguishable from a broken exporter.



461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
 
# File 'lib/wurk/configuration.rb', line 461

def telemetry=(enabled)
  guard_frozen!
  @options[:telemetry] = enabled ? true : false
  # Turning tracing back off still has to unregister, but must not *load*
  # the module to do it — that require is the whole cost this flag gates,
  # and a config that never opted in has nothing registered to undo. Tracked
  # per instance rather than off `defined?(Wurk::Telemetry)`: whether some
  # *other* config loaded the module says nothing about this one's chain.
  return if !@options[:telemetry] && !@telemetry_installed

  require_relative 'telemetry'
  @telemetry_installed = true
  Wurk::Telemetry.install!(self)
  return if !@options[:telemetry] || Wurk::Telemetry.available?

  logger.warn { 'config.telemetry = true but opentelemetry-api is not installed; tracing stays off' }
end

#telemetry? ⇒ Boolean

Returns whether the host asked for tracing. Half the gate; Telemetry.enabled? is the whole one.

Returns:

  • (Boolean) —

    whether the host asked for tracing. Half the gate; Telemetry.enabled? is the whole one.



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

def telemetry? = @options[:telemetry] == true

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



729
730
731
 
# File 'lib/wurk/configuration.rb', line 729

def topology
  @topology ||= default_topology
end

#topology=(value) ⇒ Object 



733
734
735
736
 
# File 'lib/wurk/configuration.rb', line 733

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

#total_concurrency ⇒ Object 



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

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.



652
653
654
655
 
# File 'lib/wurk/configuration.rb', line 652

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



286
287
288
 
# File 'lib/wurk/configuration.rb', line 286

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

#web_pool_size=(size) ⇒ Object 



290
291
292
293
 
# File 'lib/wurk/configuration.rb', line 290

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.



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

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