Module: Wurk::Cron

Defined in:
lib/wurk/cron.rb

Overview

Sidekiq Enterprise periodic jobs. Pure leader-driven cron — only the elected leader enqueues per tick; followers run nothing. No backfill on restart. DST-aware via per-loop timezone. In-tree crontab parser (no fugit dependency) supporting 5-field expressions plus the standard @hourly / @daily / @weekly / @monthly / @yearly aliases.

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

Layout:

* `Cron::Parser` — crontab → wall-clock match + `next_fire_at`. Walks
forward minute-by-minute in the loop's TZ; DST gaps are skipped
naturally because the wall-clock components advance past them.
* `Cron::Loop` — one registered job. Identity = SHA1(schedule+klass+opts)
so a re-registration of the same loop is idempotent.
* `Cron::Manager` — registration DSL. `mgr.register(cron, klass, **opts)`
with `tz=` mass-setter. Writes to Redis (`periodic` SET + `loops:{lid}`
HASH) and prunes superseded loops for the classes it registers; the
`config.periodic` one defers both until the server boots.
* `Cron::LoopSet` — Enumerable view (`each`/`size`/`fetch(lid)`).
* `Cron::ConfigTester` — boot-time validator. Verifies cron syntax and
that every worker class constant resolves.
* `Cron::Poller` — once-per-minute tick loop. Only the cluster leader
(`Component#leader?` / `dear-leader`) enqueues; non-leaders return
early without iterating loops.

Wire-compat: periodic, loops:{lid}, loop-history:{lid} per docs/target/sidekiq-ent.md §2.7. Periodic enqueue is gated by the single cluster leader (§6, Component#leader?) rather than a separate cron-leader lock — see the §2.7 divergence note.

Defined Under Namespace

Classes: ConfigTester, Loop, LoopSet, Manager, Parser, Poller, UnknownWorkerClass

Constant Summary collapse

PERIODIC_KEY = 
'periodic'
LOOP_PREFIX = 
'loops:'
HISTORY_PREFIX = 
'loop-history:'
HISTORY_CAP = 
25
DEFAULT_TICK_SECONDS = 
60
MISSED_TICK_THRESHOLD = 
90

Class Method Summary collapse

Class Method Details

.fire!(lid) ⇒ Object

Test/ops helper: fire one registered loop immediately, ignoring the leader gate and the schedule due-check. Records history + advances the fire marks just like a leader tick, so specs can assert on the enqueue and history deterministically without waiting on wall-clock or stubbing leadership. Returns the enqueued jid, or nil for an unknown lid. Aliased as Sidekiq::Periodic.fire!.



1158
1159
1160
1161
1162
1163
 
# File 'lib/wurk/cron.rb', line 1158

def fire!(lid)
  loop_obj = LoopSet.new.fetch(lid)
  return nil if loop_obj.nil?

  Poller.new(Wurk.configuration).fire(loop_obj)
end

.jobs ⇒ Object 



1135
1136
1137
 
# File 'lib/wurk/cron.rb', line 1135

def jobs
  LoopSet.new
end

.lid(schedule, klass, options) ⇒ Object

Stable 16-hex lid from (schedule, klass, options). Re-registering the same triple no-ops because the Redis writes overwrite under the same key.



1036
1037
1038
1039
 
# File 'lib/wurk/cron.rb', line 1036

def lid(schedule, klass, options)
  opts = options.is_a?(Hash) ? options : {}
  ::Digest::SHA1.hexdigest("#{schedule}|#{klass}|#{JSON.dump(opts.sort.to_h)}")[0, 16]
end

.persist(loop_obj) ⇒ Object

paused is written with HSETNX, not HSET: a code-declared paused: option is only the initial value at first registration, after which the field belongs to the runtime (Web UI pause/unpause). Writing it unconditionally un-paused a dashboard-paused loop on the next boot.



1055
1056
1057
1058
 
# File 'lib/wurk/cron.rb', line 1055

def persist(loop_obj)
  persist_all([loop_obj])
  loop_obj
end

.persist_all(loops) ⇒ Object

#persist for many loops in one pipelined round trip: a booting process (each swarm child) writes its whole config.periodic block at once. Nothing here reads, so a pipeline is the same writes in the same order.



1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
 
# File 'lib/wurk/cron.rb', line 1063

def persist_all(loops)
  return loops if loops.empty?

  Wurk.redis do |c|
    c.pipelined do |p|
      loops.each do |loop_obj|
        fields = loop_obj.to_redis_hash
        paused = fields.delete('paused')
        key = "#{LOOP_PREFIX}#{loop_obj.lid}"
        p.call('SADD', PERIODIC_KEY, loop_obj.lid)
        p.call('HSET', key, *fields.flatten)
        p.call('HSETNX', key, 'paused', paused)
      end
    end
  end
  loops
end

.prefork(manager) ⇒ Object

For the swarm parent's pre-fork step (Swarm::Prefork): resolve the IANA zone of every loop manager registered, so tzinfo and the zone data are loaded once and shared copy-on-write rather than loaded by the leader child on its first tick. Opens no socket. erb and yaml need no warming: Manager#load_file runs in the parent, at configuration.



1144
1145
1146
1147
1148
1149
1150
 
# File 'lib/wurk/cron.rb', line 1144

def prefork(manager)
  manager&.loops&.each do |lp|
    name = lp.tz_name
    Parser.resolve_zone(name) if lp.tz.is_a?(String) && !name.empty?
  end
  nil
end

.prune_superseded(klass, keep_lids) ⇒ Object

Drop every registered loop for klass (a name or a list of names) except keep_lids — the loops a booting process just registered for those classes. Editing a schedule or any option changes the lid, so without this the pre-edit loop stays in periodic and keeps firing next to its replacement, forever.

Scoped to the registered classes deliberately: the registry is fleet-wide, so a process that registers a subset of the fleet's loops (or none at all, e.g. a client-only app) must never delete loops for classes it doesn't register. Idempotent and safe to race — two processes booting the same code compute the same candidate set, and SREM/DEL repeat harmlessly.

One SMEMBERS and one pipelined HGET per candidate, however many classes: a boot registering C classes used to read every foreign loop C times.



1095
1096
1097
1098
1099
1100
1101
 
# File 'lib/wurk/cron.rb', line 1095

def prune_superseded(klass, keep_lids)
  klasses = Array(klass)
  keep = Array(keep_lids)
  stale = Wurk.redis { |c| superseded_lids(c, klasses, keep) }
  stale.each { |lid| unregister(lid) }
  stale
end

.register(name, cron, worker_class, args = [], **opts) ⇒ Object

Task-stated convenience signature. name is treated as a label; the lid is still derived from (schedule, klass, opts) so the call is idempotent. Callers that want the Sidekiq DSL should use Manager#register via config.periodic { |mgr| ... }.



1045
1046
1047
1048
1049
 
# File 'lib/wurk/cron.rb', line 1045

def register(name, cron, worker_class, args = [], **opts)
  merged = opts.merge(args: args)
  merged[:label] = name if name
  Loop.new(schedule: cron, klass: worker_class.to_s, options: merged).tap { |lp| persist(lp) }
end

.reset! ⇒ Object

Test helper: wipe every Cron Redis key. Production code must not call this — it removes every registered loop in the cluster.



1125
1126
1127
1128
1129
1130
1131
1132
1133
 
# File 'lib/wurk/cron.rb', line 1125

def reset!
  Wurk.redis do |c|
    lids = c.call('SMEMBERS', PERIODIC_KEY)
    lids.each do |lid|
      c.call('DEL', "#{LOOP_PREFIX}#{lid}", "#{HISTORY_PREFIX}#{lid}")
    end
    c.call('DEL', PERIODIC_KEY)
  end
end

.superseded_lids(conn, klasses, keep) ⇒ Object 



1103
1104
1105
1106
1107
1108
1109
 
# File 'lib/wurk/cron.rb', line 1103

def superseded_lids(conn, klasses, keep)
  candidates = conn.call('SMEMBERS', PERIODIC_KEY) - keep
  return [] if candidates.empty?

  owners = conn.pipelined { |p| candidates.each { |lid| p.call('HGET', "#{LOOP_PREFIX}#{lid}", 'klass') } }
  candidates.zip(owners).filter_map { |lid, owner| lid if klasses.include?(owner) }
end

.unregister(lid) ⇒ Object

Drop a loop entirely: registry membership, hash, and fire history. Public API for retiring a loop whose register(...) line is gone (no process registers its class any more, so prune_superseded cannot see it); also the primitive behind pruning and ConfigTester rollback.



1116
1117
1118
1119
1120
1121
 
# File 'lib/wurk/cron.rb', line 1116

def unregister(lid)
  Wurk.redis do |c|
    c.call('SREM', PERIODIC_KEY, lid)
    c.call('DEL', "#{LOOP_PREFIX}#{lid}", "#{HISTORY_PREFIX}#{lid}")
  end
end