Class: Wurk::Cron::Poller

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

Overview

Once-per-minute tick driver. Only the cluster leader iterates the LoopSet and enqueues; non-leaders return early. Missed-tick warning when wall-clock has drifted more than MISSED_TICK_THRESHOLD seconds past the expected fire.

Constant Summary

Constants included from Wurk::Component

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

Instance Attribute Summary

Attributes included from Wurk::Component

#config

Instance Method Summary collapse

Methods included from Wurk::Component

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

Constructor Details

#initialize(config) ⇒ Poller

Returns a new instance of Poller.



522
523
524
525
526
527
528
529
530
 
# File 'lib/wurk/cron.rb', line 522

def initialize(config)
  @config = config
  @client = Client.new(config: config)
  @thread = nil
  # Operators never need to touch this; integration tests shrink it so a
  # due loop fires within the test window instead of waiting a full minute.
  @tick_interval = config[:cron_tick_interval] || DEFAULT_TICK_SECONDS
  @timer = TimerLoop.new(@tick_interval)
end

Instance Method Details

#enqueue_if_due(loop_obj) ⇒ Object

The marks advance before the push, via CAS: the leader gate is a cached read, so a second poller can reach the same due loop for a few seconds after a handover. Losing the CAS means another tick owns this slot — enqueue nothing.



571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
 
# File 'lib/wurk/cron.rb', line 571

def enqueue_if_due(loop_obj)
  return if loop_obj.paused?

  now = ::Time.now.to_i
  slot, mark = due_slot(loop_obj, now)
  return if slot.nil?

  future = loop_obj.next_fire_after(slot, now)
  return unless claim_fire?(loop_obj, mark, now, future)

  warn_missed_tick(loop_obj, slot, now)
  jid = enqueue_claimed!(loop_obj)
  record_history(loop_obj, jid, now)
  jid
end

#fire(loop_obj) ⇒ Object

Fire one loop right now, bypassing both the leader gate and the schedule due-check, recording history + advancing the fire marks exactly like a real tick. Powers Cron.fire! (deterministic specs / manual "run now"); the scheduled, leader-gated path stays #tick.



591
592
593
594
595
596
 
# File 'lib/wurk/cron.rb', line 591

def fire(loop_obj)
  now = ::Time.now.to_i
  jid = enqueue!(loop_obj)
  record_fire(loop_obj, jid, now, loop_obj.next_fire_at(now))
  jid
end

#start ⇒ Object

TimerLoop waits one interval before the first tick: don't fire a catch-up burst the instant we boot (the leader is barely settled), and let a short-lived process exit without ticking at all.



535
536
537
538
539
540
 
# File 'lib/wurk/cron.rb', line 535

def start
  return @thread if @thread

  @timer.reset
  @thread = safe_thread('cron-poller') { @timer.run { tick } }
end

#terminate ⇒ Object

Blocks until the thread is really gone: the launcher releases the cluster lock immediately after this returns, and a tick still in flight would enqueue loops the next leader is about to fire itself.

Cleared only on a confirmed join (Thread#join returns nil on timeout): a wedged thread must stay tracked so #start's guard returns it instead of calling @timer.reset, which would un-terminate the loop it is still inside and leave two tick threads double-enqueuing the same loops.



550
551
552
553
 
# File 'lib/wurk/cron.rb', line 550

def terminate
  @timer.terminate
  @thread = nil if @thread&.join(TimerLoop::JOIN_TIMEOUT)
end

#tick ⇒ Object

Leader-gated by the single cluster lock (Component#leader? reads dear-leader): non-leaders return early and never iterate the LoopSet. The Launcher owns the lock's renewal — the poller no longer runs (or expires) its own.



559
560
561
562
563
564
565
 
# File 'lib/wurk/cron.rb', line 559

def tick
  return unless leader?

  LoopSet.new(@config).each { |lp| enqueue_if_due(lp) }
rescue StandardError => e
  handle_exception(e, { context: 'cron-poller' })
end