Module: OpenLoam::Scheduler

Defined in:
lib/open_loam/scheduler.rb

Overview

Per-tenant recurring jobs. Modules self-register default schedules; an app also creates them from the admin. A runner calls OpenLoam::Scheduler.tick periodically (wire open_loam:scheduler:tick to system cron — every minute) and each DUE schedule enqueues its ActiveJob.

NO DOUBLE-FIRE — the whole correctness story. tick ATOMICALLY CLAIMS due jobs so two worker processes never enqueue the same one:

* PostgreSQL: `SELECT ... FOR UPDATE SKIP LOCKED` — one worker wins each row.
* SQLite (demo/harness): a transactional claim; SQLite serializes writers,
which is correct for the single-process prototype.

The mechanism is behind claim_due, chosen per adapter. A claim also stamps locked_until so a crashed worker's rows free themselves after LOCK_TTL.

Defined Under Namespace

Classes: UnknownJobError

Class Method Summary collapse

Class Method Details

.allowed_job_class?(name) ⇒ Boolean

Returns:

  • (Boolean) 


96
97
98
 
# File 'lib/open_loam/scheduler.rb', line 96

def allowed_job_class?(name)
  registered.any? { |default| default[:job_class] == name } || OpenLoam.schedulable_jobs.include?(name)
end

.register(key:, job_class:, schedule:, scope: "tenant", name: nil) ⇒ Object

--- declarative registry (like broadcast_events / feature_defaults) ---



20
21
22
23
24
25
 
# File 'lib/open_loam/scheduler.rb', line 20

def register(key:, job_class:, schedule:, scope: "tenant", name: nil)
  registry[key.to_s] = {
    key: key.to_s, job_class: job_class.to_s, schedule: schedule.to_s,
    scope: scope.to_s, name: (name || key).to_s
  }
end

.registered ⇒ Object 



27
 
# File 'lib/open_loam/scheduler.rb', line 27

def registered = registry.values

.reset_registry! ⇒ Object 



29
30
31
 
# File 'lib/open_loam/scheduler.rb', line 29

def reset_registry!
  @registry = {}
end

.resolve_job_class(name) ⇒ Object

THE code-execution guard: a job_class must resolve to a real ActiveJob subclass AND be ALLOWLISTED — either registered here or listed in OpenLoam.schedulable_jobs. "Any ActiveJob" would let a tenant admin schedule ActiveStorage::PurgeJob, a mailer's delivery job, etc.



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

def resolve_job_class(name)
  name = name.to_s
  return nil unless allowed_job_class?(name)

  klass = name.safe_constantize
  klass if klass.is_a?(Class) && defined?(ActiveJob::Base) && klass < ActiveJob::Base
end

.resolve_job_class!(name) ⇒ Object 



100
101
102
 
# File 'lib/open_loam/scheduler.rb', line 100

def resolve_job_class!(name)
  resolve_job_class(name) || raise(UnknownJobError, "#{name.inspect} is not a known ActiveJob")
end

.run_now(job) ⇒ Object

Enqueue a schedule's job right now without touching its schedule (the admin "run now" button). Re-validates job_class.



80
81
82
 
# File 'lib/open_loam/scheduler.rb', line 80

def run_now(job)
  enqueue_target(job)
end

.run_tick(now) ⇒ Object 



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/open_loam/scheduler.rb', line 62

def run_tick(now)
  fired = 0
  claim_due(now).each do |job|
    begin
      enqueue_target(job)
      job.update_columns(last_run_at: now, next_run_at: job.compute_next_run(now),
                         locked_until: nil, updated_at: now)
      fired += 1
    rescue StandardError => error
      job.update_columns(locked_until: nil, updated_at: now)
      logger&.error("[open_loam scheduler] #{job.key} failed: #{error.class}: #{error.message}")
    end
  end
  fired
end

.sync_tenant(tenant) ⇒ Object

Materialize the registered TENANT-scope schedules as rows in tenant (idempotent — safe from on_tenant_created and open_loam:sync). System-scope schedules are created explicitly by the app (they are not per-tenant).



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/open_loam/scheduler.rb', line 36

def sync_tenant(tenant)
  OpenLoam.as_tenant(tenant) do
    registered.each do |default|
      next unless default[:scope] == "tenant"

      job = OpenLoam::ScheduledJob.find_or_initialize_by(key: default[:key])
      job.name = default[:name]
      job.job_class = default[:job_class]
      job.schedule = default[:schedule]
      job.scope = "tenant"
      job.active = true if job.new_record?
      job.next_run_at ||= job.compute_next_run
      job.save!
    end
  end
end

.tick(now: Time.current) ⇒ Object

Claim and enqueue every due schedule. Returns how many fired. One job's failure never blocks the others (its lock is released so a later tick retries it).



58
59
60
 
# File 'lib/open_loam/scheduler.rb', line 58

def tick(now: Time.current)
  OpenLoam::Telemetry.span("scheduler_tick") { run_tick(now) }
end