Module: Wurk::BootChecks

Defined in:
lib/wurk/boot_checks.rb

Overview

What a standalone runner says once at boot, after the app loaded. Log only, and never allowed to stop a boot: none of this changes behaviour by backend, version or peers.

Constant Summary collapse

TESTED_FLOORS =

Oldest version each non-Redis backend has been run against.

{ 'dragonfly' => '1.34.0', 'valkey' => '8.0.0' }.freeze
MAX_LISTED = 
20
SUBCLASS = 
::Module.instance_method(:<)
NAME = 
::Module.instance_method(:name)
UNIQUE_OPTIONS = 
%w[unique_for unique_until].freeze
STRICT_ENV = 
'WURK_STRICT'

Class Method Summary collapse

Class Method Details

.active_job_queue_name(queue) ⇒ Object 



168
169
170
171
 
# File 'lib/wurk/boot_checks.rb', line 168

def active_job_queue_name(queue)
  prefix = ::ActiveJob::Base.queue_name_prefix
  prefix ? "#{prefix}#{::ActiveJob::Base.queue_name_delimiter}#{queue}" : queue.to_s
end

.active_record_pool ⇒ Object

The pool size the app configured, read without opening a connection; nil without Active Record or a configured database. Rails 8.1 renamed pool.



284
285
286
287
288
289
290
291
 
# File 'lib/wurk/boot_checks.rb', line 284

def active_record_pool
  return unless defined?(::ActiveRecord::Base)

  db = ::ActiveRecord::Base.connection_db_config
  db.respond_to?(:max_connections) ? db.max_connections : db.pool
rescue ::ActiveRecord::ActiveRecordError
  nil
end

.backend(info) ⇒ Object 



37
38
39
40
41
42
 
# File 'lib/wurk/boot_checks.rb', line 37

def backend(info)
  return ['dragonfly', info['dragonfly_version'].to_s.delete_prefix('df-v')] if info['dragonfly_version']
  return ['valkey', info['valkey_version'].to_s] if info['valkey_version']

  ['redis', info['redis_version'].to_s]
end

.flag_real_sidekiq_gem(logger, specs: ::Gem.loaded_specs, strict: ENV[STRICT_ENV] == '1') ⇒ Object

The real gem next to Wurk loads both: Sidekiq::* is then partly theirs, partly ours. Gems that depend on sidekiq pull it in unless the shim gem, which depends on Wurk instead, satisfies the dependency. Runs as Wurk loads.

Raises:



198
199
200
201
202
203
204
 
# File 'lib/wurk/boot_checks.rb', line 198

def flag_real_sidekiq_gem(logger, specs: ::Gem.loaded_specs, strict: ENV[STRICT_ENV] == '1')
  message = real_sidekiq_message(specs, logger)
  return unless message
  raise Wurk::Error, message if strict

  logger.error { "#{message} Set #{STRICT_ENV}=1 to refuse to boot instead." }
end

.framework_queue_set(framework, mod) ⇒ Object 



156
157
158
159
160
 
# File 'lib/wurk/boot_checks.rb', line 156

def framework_queue_set(framework, mod)
  return {} unless mod.respond_to?(:queues) && mod.queues.respond_to?(:to_h)

  mod.queues.to_h.transform_keys { |kind| "#{framework} #{kind}" }
end

.framework_queues ⇒ Object

{ queue => what Rails sends there }, for apps whose ActiveJob adapter is Wurk.



146
147
148
149
150
151
152
153
154
 
# File 'lib/wurk/boot_checks.rb', line 146

def framework_queues
  return {} unless wurk_active_job?

  named = {}
  named['mail delivery'] = ::ActionMailer::Base.deliver_later_queue_name if defined?(::ActionMailer::Base)
  named.merge!(framework_queue_set('Active Storage', defined?(::ActiveStorage) && ::ActiveStorage))
  named.merge!(framework_queue_set('Action Mailbox', defined?(::ActionMailbox) && ::ActionMailbox))
  named.compact.to_h { |what, queue| [active_job_queue_name(queue), what] }
end

.job_classes_setting(options) ⇒ Object

Workers, and ActiveJob classes that take sidekiq_options (Wurk::Job::Options). A walk over every class meets classes that redefine these with other arguments (Faker::Travel::Airport.name takes required keywords), and one such class must not cost the app this check.



186
187
188
189
190
191
192
193
 
# File 'lib/wurk/boot_checks.rb', line 186

def job_classes_setting(options)
  ::ObjectSpace.each_object(::Class).filter_map do |klass|
    next unless SUBCLASS.bind_call(klass, Wurk::Worker) || SUBCLASS.bind_call(klass, Wurk::Job::Options)

    name = NAME.bind_call(klass)
    name if name && klass.get_sidekiq_options.values_at(*options).any?
  end.sort
end

.listing(names) ⇒ Object 



293
294
295
296
 
# File 'lib/wurk/boot_checks.rb', line 293

def listing(names)
  extra = names.size - MAX_LISTED
  names.first(MAX_LISTED).join(', ') + (extra.positive? ? " and #{extra} more" : '')
end

.locked_job_classes ⇒ Object 



180
 
# File 'lib/wurk/boot_checks.rb', line 180

def locked_job_classes = job_classes_setting(%w[lock])

.log_backend(logger, info) ⇒ Object 



44
45
46
47
48
49
50
51
 
# File 'lib/wurk/boot_checks.rb', line 44

def log_backend(logger, info)
  name, version = backend(info)
  logger.info { "Wurk #{Wurk::VERSION} on #{name} #{version}" }
  floor = TESTED_FLOORS[name]
  return unless floor && ::Gem::Version.correct?(version) && ::Gem::Version.new(version) < ::Gem::Version.new(floor)

  logger.warn { "#{name} #{version} is older than the oldest version Wurk is tested against (#{floor})." }
end

.log_client_only(logger, reason) ⇒ Object

DEBUG, not INFO: Wurk logs to stdout, and every rails runner, rake task and wurk preflight --app boots the app; their stdout is someone's output.



260
261
262
 
# File 'lib/wurk/boot_checks.rb', line 260

def log_client_only(logger, reason)
  logger.debug { "wurk: client only (#{reason}); this process enqueues and runs no jobs" }
end

.log_process_role(config, role, ar_pool: active_record_pool) ⇒ Object

The role line of a process that runs every capsule's threads itself.



236
237
238
239
240
241
242
 
# File 'lib/wurk/boot_checks.rb', line 236

def log_process_role(config, role, ar_pool: active_record_pool)
  capsules = config.capsules.values
  slots = capsules.map { |cap| [capsules.size > 1 ? "capsule #{cap.name}" : '1 process', cap.concurrency] }
  log_role(config, role, slots, per_process: slots.sum(&:last), ar_pool: ar_pool)
rescue StandardError => e
  role_skipped(config.logger, e)
end

.log_role(config, role, slots, per_process:, ar_pool:) ⇒ Object

One line per process: what it is, and how big. per_process is the most job threads one process runs, which is what its Active Record pool covers.



250
251
252
253
254
255
256
 
# File 'lib/wurk/boot_checks.rb', line 250

def log_role(config, role, slots, per_process:, ar_pool:)
  sizes = slots.map do |label, threads|
    "#{label} × #{threads} threads (redis pool #{redis_pool_size(config, threads)}, fetch #{threads})"
  end
  config.logger.info { "wurk: #{role}; #{sizes.join(' + ')}#{"; AR pool #{ar_pool}" if ar_pool}" }
  warn_on_small_ar_pool(config.logger, ar_pool, per_process)
end

.log_swarm_role(config, role, ar_pool: active_record_pool) ⇒ Object

The role line of a process that forks the topology's children and runs no jobs itself. Its children each run the threads of one slot.



226
227
228
229
230
231
232
233
 
# File 'lib/wurk/boot_checks.rb', line 226

def log_swarm_role(config, role, ar_pool: active_record_pool)
  slots = config.topology.slots.map do |slot|
    ["#{slot.count} #{slot.count == 1 ? 'process' : 'processes'}", slot.concurrency]
  end
  log_role(config, role, slots, per_process: slots.map(&:last).max, ar_pool: ar_pool)
rescue StandardError => e
  role_skipped(config.logger, e)
end

.own_queues(config) ⇒ Object 



173
174
175
176
177
178
 
# File 'lib/wurk/boot_checks.rb', line 173

def own_queues(config)
  queues = config.capsules.each_value.flat_map { |cap| cap.weights.keys }
  return queues unless config.topology_assigned?

  queues | config.topology.slots.flat_map { |slot| Array(slot.queues).map { |entry| entry.to_s.split(',').first } }
end

.process_infos(config) ⇒ Object 



65
66
67
68
69
70
 
# File 'lib/wurk/boot_checks.rb', line 65

def process_infos(config)
  config.redis(idempotent: true) do |conn|
    ids = conn.call('SMEMBERS', Keys::PROCESSES)
    ids.empty? ? [] : ids.zip(conn.pipelined { |pipe| ids.each { |id| pipe.call('HGET', id, 'info') } })
  end
end

.real_sidekiq_message(specs, logger) ⇒ Object 



206
207
208
209
210
211
212
213
214
215
216
217
 
# File 'lib/wurk/boot_checks.rb', line 206

def real_sidekiq_message(specs, logger)
  real = specs['sidekiq']
  return if real.nil? || real.dependencies.any? { |dep| dep.name == 'wurk' }

  "The bundle holds the real sidekiq #{real.version} gem, so it loads next to Wurk and the process runs " \
    "half on each. Pulled in by: #{sidekiq_dependents(specs)}. Replace it with the shim in the Gemfile: " \
    "gem 'sidekiq', github: 'developerz-ai/wurk', tag: 'v#{Wurk::VERSION}', " \
    "glob: 'ecosystem/sidekiq-shim/*.gemspec'."
rescue StandardError => e
  logger.warn { "wurk: could not check the bundle for the real sidekiq gem: #{e.class}: #{e.message}" }
  nil
end

.redis_pool_size(config, threads) ⇒ Object

Capsule#main_pool_size, without building a pool: a swarm parent describes slots whose pools only its children build.



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

def redis_pool_size(config, threads)
  config.redis_config[:size] || [threads + Capsule::POOL_HEADROOM, Capsule::MIN_POOL_SIZE].max
end

.role_skipped(logger, error) ⇒ Object 



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

def role_skipped(logger, error)
  logger.warn { "wurk role line skipped: #{error.class}: #{error.message}" }
end

.run(config, info) ⇒ Object

info is nil when there is no Redis to ask (wurk boot-check without --redis): only the checks that read the loaded app run then.



24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/wurk/boot_checks.rb', line 24

def run(config, info)
  if info
    log_backend(config.logger, info)
    warn_on_foreign_processes(config)
    warn_on_unconsumed_framework_queues(config)
  end
  warn_on_ignored_locks(config.logger)
  warn_on_ignored_unique(config.logger)
  warn_on_client_config_only(config)
rescue StandardError => e
  config.logger.warn { "wurk boot checks skipped: #{e.class}: #{e.message}" }
end

.sidekiq_dependents(specs) ⇒ Object 



219
220
221
222
 
# File 'lib/wurk/boot_checks.rb', line 219

def sidekiq_dependents(specs)
  names = specs.each_value.select { |spec| spec.dependencies.any? { |dep| dep.name == 'sidekiq' } }.map(&:name)
  names.empty? ? 'the Gemfile itself' : listing(names.sort)
end

.warn_on_client_config_only(config) ⇒ Object

Stock Sidekiq skips a configure_client block in a server without a word, and so does Configuration: an app normally configures both roles. A server where only client blocks were written ran none of them, so the Redis settings, middleware and handlers they hold are all absent here.



106
107
108
109
110
111
112
113
114
 
# File 'lib/wurk/boot_checks.rb', line 106

def warn_on_client_config_only(config)
  return unless config.client_config_only?

  config.logger.warn do
    'This is a server process, so its configure_client blocks were skipped, and no configure_server block ran: ' \
      'whatever those blocks set (config.redis, middleware, handlers) does not apply to these workers. Set it ' \
      'in a configure_server block too, or outside both.'
  end
end

.warn_on_foreign_processes(config) ⇒ Object

Both runtimes share processes and the same queues, so a mix is safe only if it is deliberate. Wurk's heartbeat always carries cores; stock does not.



55
56
57
58
59
60
61
62
63
 
# File 'lib/wurk/boot_checks.rb', line 55

def warn_on_foreign_processes(config)
  foreign = process_infos(config).reject { |_, info| info.nil? || ::JSON.parse(info).key?('cores') }.map(&:first)
  return if foreign.empty?

  config.logger.warn do
    "Live processes that are not Wurk share this Redis: #{listing(foreign)}. " \
      'They fetch from the same queues. If this is not a deliberate rolling cutover, stop them.'
  end
end

.warn_on_ignored_locks(logger) ⇒ Object

Without the gem's middleware the option does nothing: the "unique" job simply enqueues twice. Only classes the app has loaded by now are seen.



74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/wurk/boot_checks.rb', line 74

def warn_on_ignored_locks(logger)
  return if defined?(::SidekiqUniqueJobs)

  classes = locked_job_classes
  return if classes.empty?

  logger.warn do
    'These jobs set `lock:` but sidekiq-unique-jobs is not loaded, so the option does nothing and ' \
      "duplicates will run: #{listing(classes)}. Keep the gem, or move them to `unique_for:`."
  end
end

.warn_on_ignored_topology(config, runner) ⇒ Object

The topology only shapes what a swarm forks; wurk and an embedded worker run one process with the capsules' threads, so a declared one silently does nothing there.



119
120
121
122
123
124
125
126
 
# File 'lib/wurk/boot_checks.rb', line 119

def warn_on_ignored_topology(config, runner)
  return unless config.topology_assigned?

  config.logger.warn do
    "config.topology is set, but #{runner} runs a single process and ignores it: its slots' queues and " \
      'concurrency are not used here. Run wurkswarm to fork it, or set the queues and concurrency on the capsules.'
  end
end

.warn_on_ignored_unique(logger) ⇒ Object

Only the middleware unique! installs reads these options, so without it every push enqueues. Turning it on for the app is a human call, not a boot check's: this only says so.



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

def warn_on_ignored_unique(logger)
  return if Wurk::Unique.enabled?

  classes = job_classes_setting(UNIQUE_OPTIONS)
  return if classes.empty?

  logger.warn do
    'These jobs set `unique_for:` or `unique_until:` but unique jobs are not enabled, so the options do ' \
      "nothing and duplicates enqueue: #{listing(classes)}. Enable them once at boot, in an initializer: " \
      '`Sidekiq::Enterprise.unique!`.'
  end
end

.warn_on_small_ar_pool(logger, ar_pool, threads) ⇒ Object

Every job thread that touches the database holds a connection; with fewer in the pool the rest wait, then fail with ActiveRecord::ConnectionTimeoutError.



272
273
274
275
276
277
278
279
280
 
# File 'lib/wurk/boot_checks.rb', line 272

def warn_on_small_ar_pool(logger, ar_pool, threads)
  return unless ar_pool && ar_pool < threads

  logger.warn do
    "The Active Record pool holds #{ar_pool} connections but this process runs up to #{threads} job threads, " \
      'so jobs that use the database wait for a connection and then fail on a timeout. Set `pool:` in ' \
      "config/database.yml to at least #{threads} (plus the web server's threads when jobs run in the web process)."
  end
end

.warn_on_unconsumed_framework_queues(config) ⇒ Object

Rails enqueues its own jobs (mail delivery, Active Storage analysis and purge, Action Mailbox routing) to queues an app never names itself, so they are the ones that go unlistened. Read from Redis, so only with it.



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

def warn_on_unconsumed_framework_queues(config)
  wanted = framework_queues
  return if wanted.empty?

  listened = own_queues(config) | config.redis(idempotent: true) { |conn| QueueConsumers.counts(conn).keys }
  missing = wanted.except(*listened)
  return if missing.empty?

  config.logger.warn do
    "Rails enqueues to #{listing(missing.map { |queue, what| "#{queue} (#{what})" })}, which neither this " \
      'process nor any live one listens to, so those jobs wait forever. Add the queues to a worker.'
  end
end

.wurk_active_job? ⇒ Boolean

Returns:

  • (Boolean) 


162
163
164
165
166
 
# File 'lib/wurk/boot_checks.rb', line 162

def wurk_active_job?
  return false unless defined?(::ActiveJob::Base)

  ::ActiveJob::Base.queue_adapter.class.name.to_s.match?(/Wurk|Sidekiq/)
end