Class: Wurk::Cron::Poller

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

Overview

Once-per-minute tick driver, plus one tick the moment this process wins the lock: at a tick's phase, a leader that lives under a minute would never tick. Only the cluster leader iterates the LoopSet and enqueues; non-leaders return early. A slot that wall-clock has passed by more than MISSED_TICK_THRESHOLD seconds is skipped with a warning, never backfilled, and the occurrence due now, if any, fires in its place.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ Poller

Returns a new instance of Poller.



704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
 
# File 'lib/wurk/cron.rb', line 704

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
  @done = false
  @woken = false
  @mutex = ::Mutex.new
  @sleeper = ::ConditionVariable.new
  @on_leader = method(:wake_for_leadership)
  @unknown_reported = ::Set.new
  @derived = {}
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, marks = nil) ⇒ 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.

A slot older than the missed-tick threshold is not backfilled (ent §2.6: no backfill on restart, missed ticks across a leader gap are dropped): the marks still advance past it, but nothing is pushed.

One claim, one push at most, whichever slot it pushes for: the occurrence due now stands in for a stale one under the stale one's mark, so a poller that read the same mark loses the CAS for both.

marks is the raw [lf, nf] pair the caller just read (a tick's LoopSet read); without it they are read here.



786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
 
# File 'lib/wurk/cron.rb', line 786

def enqueue_if_due(loop_obj, marks = nil)
  return if loop_obj.paused?

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

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

  skip_missed(loop_obj, slot, now) unless firing == slot
  return if firing.nil?

  fire_claimed(loop_obj, now)
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.



807
808
809
810
811
812
 
# File 'lib/wurk/cron.rb', line 807

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

The loop waits one interval before its first tick unless this process wins the lock first: let a process that never leads, or exits at once, exit without ticking at all.



723
724
725
726
727
728
729
 
# File 'lib/wurk/cron.rb', line 723

def start
  return @thread if @thread

  @mutex.synchronize { @done = false }
  leader_hooks << @on_leader
  @thread = safe_thread('cron-poller') { run_loop }
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 clearing @done, which would un-terminate the loop it is still inside and leave two tick threads double-enqueuing the same loops.



739
740
741
742
743
744
745
746
 
# File 'lib/wurk/cron.rb', line 739

def terminate
  leader_hooks.delete(@on_leader)
  @mutex.synchronize do
    @done = true
    @sleeper.signal
  end
  @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.

Rescued per loop: the LoopSet order is stable, so one loop that raises on every tick (a push its middleware refuses, a class that no longer loads) would otherwise starve every loop after it, forever.



756
757
758
759
760
761
762
763
764
765
766
767
768
769
 
# File 'lib/wurk/cron.rb', line 756

def tick
  return unless leader?

  seen = ::Set.new
  LoopSet.new(@config).each do |lp|
    seen << derived_key(lp)
    enqueue_if_due(lp, lp.fire_marks)
  rescue StandardError => e
    handle_exception(e, { context: 'cron-poller', lid: lp.lid, klass: lp.klass })
  end
  @derived.select! { |key, _| seen.include?(key) }
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.