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.

Instance Attribute Summary collapse

Instance Method Summary collapse

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 Attribute Details

#config ⇒ Object (readonly) Originally defined in module Wurk::Component

Returns the value of attribute config.

Instance Method Details

#default_tag(dir = Dir.pwd) ⇒ Object Originally defined in module Wurk::Component

#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

#fire_event(event, oneshot: true, reverse: false, reraise: false) ⇒ Object Originally defined in module Wurk::Component

Invokes lifecycle hooks for event. Hooks run in registration order (or LIFO when reverse: true, used for teardown). A raise in one hook is reported via handle_exception and does NOT stop the next hook unless reraise: true (used in tests / fail-fast boot). oneshot: true clears the bucket after dispatch so the event can't fire twice.

#handle_exception(ex, ctx = {}) ⇒ Object Originally defined in module Wurk::Component

#hostname ⇒ Object Originally defined in module Wurk::Component

#identity ⇒ Object Originally defined in module Wurk::Component

#leader? ⇒ Boolean Originally defined in module Wurk::Component

True iff this process currently holds the cluster dear-leader lock. Cached per Component instance for LEADER_CACHE_TTL_MS (~5s): cron and the metrics rollups call this every tick, and an uncached GET would double their Redis traffic at short intervals for no benefit — the lock's own renewal cadence (60s+, spec §6.1) easily tolerates a few-second-stale read. Returns false unconditionally when WURK_LEADER=false (or SIDEKIQ_LEADER=false) is set on the process (opt-out hot-standby). Any Redis error is swallowed → false, so a transient partition can't propagate as an exception into user code.

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

Returns:

  • (Boolean)

#logger ⇒ Object Originally defined in module Wurk::Component

--- delegated to config -------------------------------------------

#mono_ms ⇒ Object Originally defined in module Wurk::Component

#process_nonce ⇒ Object Originally defined in module Wurk::Component

#real_ms ⇒ Object Originally defined in module Wurk::Component

--- clocks ---------------------------------------------------------

#redis(idempotent: false) ⇒ Object Originally defined in module Wurk::Component

#safe_thread(name, priority: nil, &block) ⇒ Object Originally defined in module Wurk::Component

Spawns a named thread that runs block under watchdog(name). The parent must retain the returned Thread; otherwise GC may not, but report_on_exception is disabled so we don't double-log on death.

Priority resolution matches Sidekiq (component.rb:44-48): explicit argument, then config.thread_priority, then -1. Ruby's default of 0 buys a 100ms timeslice; each negative step halves it, so -1 keeps a CPU-heavy capsule from starving its siblings for a whole tick.

#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

#tid ⇒ Object Originally defined in module Wurk::Component

#watchdog(last_words) ⇒ Object Originally defined in module Wurk::Component

Wraps a block at a thread boundary: any unhandled exception is reported via handle_exception (so it lands in error_handlers / the log) and then re-raised. last_words is the component label included in the context.