Class: Honker::Scheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/honker/scheduler.rb

Overview

Time-trigger scheduler. Register tasks with add; tick fires all boundaries that have elapsed since the last tick and enqueues the resulting jobs. run(owner:, stop:) drives the loop under a leader-elected advisory lock.

Constant Summary collapse

LEADER_LOCK =

Lock name used for leader election in run. Constant so all processes contending for leader share a single lock row.

"honker-scheduler"
LOCK_TTL_S =

TTL on the leader lock. Refreshed from run every HEARTBEAT_S; a leader whose refresh fails drops out of the loop so a standby can pick up without waiting the full TTL.

60
HEARTBEAT_S =

Refresh cadence. Balance: too small and every tick is a lock write; too large and a standby waits longer than necessary after a crash. Matches the Rust binding.

20
UPDATE_POLL_S = 
0.05

Instance Method Summary collapse

Constructor Details

#initialize(db) ⇒ Scheduler

Returns a new instance of Scheduler.



33
34
35
 
# File 'lib/honker/scheduler.rb', line 33

def initialize(db)
  @db = db
end

Instance Method Details

#add(name:, queue:, cron: nil, schedule: nil, payload:, priority: 0, expires_s: nil, max_attempts: 3) ⇒ Object

Register a scheduled task. cron: is kept for backward compatibility; schedule: is the clearer name and can hold:

  • 5-field cron
  • 6-field cron
  • @every <n><unit> like @every 1s

Idempotent by name; registering the same name twice replaces the previous row.

Raises:

  • (ArgumentError) 


46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/honker/scheduler.rb', line 46

def add(name:, queue:, cron: nil, schedule: nil, payload:, priority: 0, expires_s: nil, max_attempts: 3)
  expr = schedule || cron
  raise ArgumentError, "must provide cron: or schedule:" if expr.nil? || expr.empty?

  @db.db.get_first_row(
    "SELECT honker_scheduler_register(?, ?, ?, ?, ?, ?, ?)",
    [name, queue, expr, JSON.dump(payload), priority, expires_s, max_attempts],
  )
  @db.mark_updated
  nil
end

#list ⇒ Object

Return every registered schedule with current state. Each entry is a Hash with: name, queue, cron_expr, payload (JSON string), priority, expires_s, next_fire_at, enabled, max_attempts.



109
110
111
112
113
114
 
# File 'lib/honker/scheduler.rb', line 109

def list
  raw = @db.db.get_first_row("SELECT honker_scheduler_list()")[0]
  return [] if raw.nil? || raw.empty?

  JSON.parse(raw)
end

#pause(name) ⇒ Object

Pause a registered schedule. Returns true if a row was paused; false if missing or already paused. Idempotent.



89
90
91
92
93
94
95
 
# File 'lib/honker/scheduler.rb', line 89

def pause(name)
  n = @db.db.get_first_row(
    "SELECT honker_scheduler_pause(?)", [name],
  )[0]
  @db.mark_updated if n.positive?
  n.positive?
end

#remove(name) ⇒ Object

Remove a registered task by name. Returns the count deleted (0 or 1).



60
61
62
63
64
65
 
# File 'lib/honker/scheduler.rb', line 60

def remove(name)
  @db.db.get_first_row(
    "SELECT honker_scheduler_unregister(?)",
    [name],
  )[0].tap { @db.mark_updated }
end

#resume(name) ⇒ Object

Resume a paused schedule. Returns true if a row was resumed.



98
99
100
101
102
103
104
 
# File 'lib/honker/scheduler.rb', line 98

def resume(name)
  n = @db.db.get_first_row(
    "SELECT honker_scheduler_resume(?)", [name],
  )[0]
  @db.mark_updated if n.positive?
  n.positive?
end

#run(owner:, stop:) ⇒ Object

Run the scheduler loop with leader election. Blocks until stop signals. stop is any object that responds to call (returning truthy to stop) — a common choice is a lambda backed by a Mutex- guarded flag, or an AtomicBoolean-like wrapper.

Only the process holding the "honker-scheduler" advisory lock fires. Standbys sleep 5s and retry. The leader heartbeats every 20s; if the refresh fails (returns 0), we break out of the leader loop immediately so we don't double-fire alongside a new leader that acquired the lock after our TTL elapsed.

owner distinguishes processes — typically a hostname + pid. On tick error, the lock is released before re-raising so a standby can pick up without waiting the full TTL.



159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/honker/scheduler.rb', line 159

def run(owner:, stop:)
  stop_fn = normalize_stop(stop)
  until stop_fn.call
    acquired = lock_try_acquire(LEADER_LOCK, owner, LOCK_TTL_S)
    unless acquired
      wait_for_update_or_timeout(5, stop_fn)
      next
    end

    begin
      leader_loop(owner, stop_fn)
    ensure
      lock_release(LEADER_LOCK, owner)
    end
  end
  nil
end

#soonest ⇒ Object

Soonest next_fire_at across all tasks, or 0 if no tasks.



78
79
80
 
# File 'lib/honker/scheduler.rb', line 78

def soonest
  @db.db.get_first_row("SELECT honker_scheduler_soonest()")[0]
end

#tick(now = Time.now.to_i) ⇒ Object

Fire all due boundaries at now. Returns an array of ScheduledFire — one per enqueued job.



69
70
71
72
73
74
75
 
# File 'lib/honker/scheduler.rb', line 69

def tick(now = Time.now.to_i)
  rows_json = @db.db.get_first_row(
    "SELECT honker_scheduler_tick(?)",
    [now],
  )[0]
  JSON.parse(rows_json).map { |r| ScheduledFire.from_row(r) }
end

#update(name, schedule: UNSET, cron: UNSET, payload: UNSET, priority: UNSET, expires_s: UNSET, max_attempts: UNSET) ⇒ Object

Mutate fields in place. Pass only the kwargs you want changed (omitting a kwarg leaves the field alone). payload: nil writes JSON null; max_attempts: nil resets to default 3. Cron change recomputes next_fire_at from now. Returns true iff a row was updated.



121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/honker/scheduler.rb', line 121

def update(name, schedule: UNSET, cron: UNSET, payload: UNSET, priority: UNSET, expires_s: UNSET, max_attempts: UNSET)
  expr = nil
  expr = schedule if schedule != UNSET
  expr = cron if expr.nil? && cron != UNSET

  payload_arg = (payload == UNSET) ? nil : JSON.dump(payload)
  priority_arg = (priority == UNSET) ? nil : priority
  touch_expires = (expires_s == UNSET) ? 0 : 1
  expires_arg = (expires_s == UNSET) ? nil : expires_s
  touch_max_attempts = (max_attempts == UNSET) ? 0 : 1
  max_attempts_arg = (max_attempts == UNSET) ? nil : max_attempts

  any_field = !expr.nil? || payload != UNSET || priority != UNSET || expires_s != UNSET || max_attempts != UNSET
  return false unless any_field

  n = @db.db.get_first_row(
    "SELECT honker_scheduler_update(?, ?, ?, ?, ?, ?, ?, ?)",
    [name, expr, payload_arg, priority_arg, expires_arg, touch_expires,
     max_attempts_arg, touch_max_attempts],
  )[0]
  @db.mark_updated if n.positive?
  n.positive?
end