Class: Wurk::Cron::Loop

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

Overview

One registered loop. Carries identity, schedule, options, and the cached parser. Immutable after register! — re-registering the same (schedule, klass, options) triple no-ops.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(schedule:, klass:, options: {}, tz: nil, lid: nil, paused: nil) ⇒ Loop

paused: is the loop HASH's paused field when read back from Redis — the runtime owns it (Web UI pause/unpause, ent §2.4). A code-declared paused: option only seeds that field at first registration, so it must not keep the loop paused once the field says otherwise.

Raises:

  • (ArgumentError) 


301
302
303
304
305
306
307
308
309
310
311
312
 
# File 'lib/wurk/cron.rb', line 301

def initialize(schedule:, klass:, options: {}, tz: nil, lid: nil, paused: nil)
  raise ArgumentError, 'klass must be a String' unless klass.is_a?(String) && !klass.empty?

  @schedule = schedule
  @klass = klass
  @options = stringify_options(options)
  @tz = tz
  @parser = Parser.cached(schedule)
  @lid = lid || Cron.lid(schedule, klass, @options)
  @paused = paused.nil? ? [true, '1', 1].include?(@options['paused']) : paused
  @fire_marks = nil
end

Instance Attribute Details

#fire_marks ⇒ Object

The raw lf / nf strings of the hash this loop was read from, so a tick decides on the read it already made; nil for a loop built in code.



316
317
318
 
# File 'lib/wurk/cron.rb', line 316

def fire_marks
  @fire_marks
end

#klass ⇒ Object (readonly)

Returns the value of attribute klass.



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

def klass
  @klass
end

#lid ⇒ Object (readonly)

Returns the value of attribute lid.



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

def lid
  @lid
end

#options ⇒ Object (readonly)

Returns the value of attribute options.



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

def options
  @options
end

#schedule ⇒ Object (readonly)

Returns the value of attribute schedule.



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

def schedule
  @schedule
end

#tz ⇒ Object (readonly)

Returns the value of attribute tz.



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

def tz
  @tz
end

Class Method Details

.from_redis(lid, hash) ⇒ Object 



444
445
446
447
448
449
450
 
# File 'lib/wurk/cron.rb', line 444

def self.from_redis(lid, hash)
  h = hash.is_a?(Array) ? hash.each_slice(2).to_h : hash
  opts = h['options'] ? JSON.parse(h['options']) : {}
  tz = h['tz'].to_s.empty? ? nil : h['tz']
  new(lid: lid, schedule: h['schedule'], klass: h['klass'], options: opts, tz: tz,
      paused: h['paused'] == '1')
end

Instance Method Details

#args ⇒ Object

args: is the static argument list splatted into perform. A bare value (a Hash most often) is one argument — Array({ 'a' => 1 }) would turn it into [['a', 1]].



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

def args
  value = @options['args']
  return [] if value.nil?

  value.is_a?(::Array) ? value : [value]
end

#history ⇒ Object 



370
371
372
373
374
375
 
# File 'lib/wurk/cron.rb', line 370

def history
  Wurk.redis do |c|
    entries = c.call('LRANGE', "#{HISTORY_PREFIX}#{@lid}", 0, -1)
    entries.map { |e| JSON.parse(e) }
  end
end

#hourly? ⇒ Boolean

True when the schedule fires every hour (hour field is * / all 24). Only such a schedule legitimately runs in both repeated fall-back hours; a fixed-hour slot (even a multi-hour one like 0 1,2 * * *) must dedup.

Returns:

  • (Boolean) 


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

def hourly?
  parser.fields[1].size == 24
end

#job_item ⇒ Object

The payload one tick pushes. Ent §2.2: a tick is klass.perform_async(*args) with the loop's retry / queue laid over the worker's own options, so the resolved class goes to the client — JobUtil#normalize_item merges its get_sidekiq_options (unique_for, encrypt, expires_in, …) only for a Class. Only the overrides the loop actually declares are set.



355
356
357
358
359
360
361
 
# File 'lib/wurk/cron.rb', line 355

def job_item
  worker = worker_class
  item = { 'class' => worker.respond_to?(:get_sidekiq_options) ? worker : @klass, 'args' => args }
  item['queue'] = @options['queue'] if @options.key?('queue')
  item['retry'] = @options['retry'] if @options.key?('retry')
  item
end

#last_fired_at ⇒ Object

Epoch of the most recent fire, or nil if this loop has never run. The poller LPUSHes [fired_at, jid] tuples so index 0 is newest; we read only that one entry instead of the whole history list.



380
381
382
383
384
385
386
387
388
389
390
391
 
# File 'lib/wurk/cron.rb', line 380

def last_fired_at
  raw = Wurk.redis { |c| c.call('LINDEX', "#{HISTORY_PREFIX}#{@lid}", 0) }
  return nil if raw.nil?

  entry = JSON.parse(raw)
  return nil unless entry.is_a?(Array)

  fired_at = entry[0]
  fired_at.is_a?(Numeric) ? fired_at.to_i : nil
rescue JSON::ParserError
  nil
end

#local_components(epoch) ⇒ Object 



425
426
427
 
# File 'lib/wurk/cron.rb', line 425

def local_components(epoch)
  parser.local_components(epoch, @tz)
end

#next_fire_after(fired_slot, advance_from = fired_slot) ⇒ Object

Next scheduled fire after firing at fired_slot. advance_from (the poller passes now) is the search origin so missed ticks are not backfilled. The DST guard: on a fall-back the clock rolls back, so the same wall-clock minute recurs at a later UTC instant. That equality alone isn't enough to skip — it would also drop a legitimate hourly run whose repeated hour is real. So we only suppress the duplicate for a FIXED daily slot; an hourly schedule (wildcard hour) keeps both fold hours. A sub-hourly schedule (e.g. */30) lands on a different minute, so the equality never even triggers. Spec: no DST double-fire without skipping legitimate hourly runs (sidekiq-ent.md §2.6).



417
418
419
420
421
422
423
 
# File 'lib/wurk/cron.rb', line 417

def next_fire_after(fired_slot, advance_from = fired_slot)
  nxt = next_fire_at(advance_from)
  return nxt if nxt.nil? || local_components(nxt) != local_components(fired_slot)
  return nxt if hourly? # repeated fall-back hour is a genuine second run

  next_fire_at(nxt) # fixed daily slot: drop the fold duplicate
end

#next_fire_at(from_epoch = ::Time.now.to_i) ⇒ Object 



403
404
405
 
# File 'lib/wurk/cron.rb', line 403

def next_fire_at(from_epoch = ::Time.now.to_i)
  parser.next_fire_at(from_epoch, @tz)
end

#parser ⇒ Object 



318
319
320
 
# File 'lib/wurk/cron.rb', line 318

def parser
  @parser ||= Parser.cached(@schedule)
end

#paused? ⇒ Boolean

Returns:

  • (Boolean) 


322
323
324
 
# File 'lib/wurk/cron.rb', line 322

def paused?
  @paused
end

#queue ⇒ Object

The queue a tick lands in: the loop's own queue:, else the worker's sidekiq_options queue, as perform_async would pick (ent §2.2).



328
329
330
 
# File 'lib/wurk/cron.rb', line 328

def queue
  @options['queue'] || worker_option('queue') || 'default'
end

#retry_value ⇒ Object

The loop's retry:, else the worker's, else Sidekiq's default true.



343
344
345
346
347
348
 
# File 'lib/wurk/cron.rb', line 343

def retry_value
  return @options['retry'] if @options.key?('retry')

  worker_retry = worker_option('retry')
  worker_retry.nil? || worker_retry
end

#to_redis_hash ⇒ Object 



393
394
395
396
397
398
399
400
401
 
# File 'lib/wurk/cron.rb', line 393

def to_redis_hash
  {
    'schedule' => @schedule,
    'klass' => @klass,
    'options' => Wurk.dump_json(@options),
    'tz' => tz_name.to_s,
    'paused' => paused? ? '1' : '0'
  }
end

#tz_name ⇒ Object 



436
437
438
439
440
441
442
 
# File 'lib/wurk/cron.rb', line 436

def tz_name
  return nil if @tz.nil?
  return @tz.name if @tz.respond_to?(:name)
  return @tz.identifier if @tz.respond_to?(:identifier)

  @tz.to_s
end

#worker_class ⇒ Object

The worker constant, or nil when it is not loadable in this process.



364
365
366
367
368
 
# File 'lib/wurk/cron.rb', line 364

def worker_class
  @klass.split('::').inject(::Object) { |mod, name| mod.const_get(name) }
rescue ::NameError
  nil
end