wagoe.jobs.shell.recurring — com.wagoe/wagoe-jobs 1.1.0-rc-1

Recurring jobs on cron schedules, and enqueue-once by idempotency key.

Every replica may run a ticker; the one holding the lease in IRecurringState enqueues. Each tick is claimed in the idempotency store under (job type, scheduled instant) before it is enqueued, so two tickers that overlap after a lease handover still enqueue it once.

Ticks missed while no ticker ran collapse into one: after downtime a schedule fires its latest missed tick once and the earlier ones never. A schedule's first tick is the first after it was registered.

The lease holder warns once about a scheduled job type with no handler here. Not refused at init: handlers may be registered after boot, or live only on worker nodes.

create-scheduler clj

(create-scheduler {:keys [queue state idempotency registry watch]} & [opts])

A scheduler enqueueing on queue, coordinating through state (IRecurringState) and idempotency (IIdempotencyStore). With registry (IJobRegistry), a scheduled job type it has no handler for is logged as a warning. With watch, (fn [now]), the lease holder calls it every :watch-ms (the staleness watch). opts as defaults; :zone is the default for schedules that name none.

default-key-ttl-ms clj

How long a claimed idempotency key holds by default: a week.

defaults clj

enqueue-once! clj

(enqueue-once! queue
               idempotency
               queue-name
               job
               key
               &
               [{:keys [ttl-ms] :or {ttl-ms default-key-ttl-ms}}])

Enqueue job on queue-name unless (its job type, key) was claimed in idempotency within :ttl-ms. Returns the job id, or nil for a duplicate. A failed or refused enqueue (a held :unique-key) releases the claim, so a retry can succeed; a process that dies between claim and enqueue loses the job (at most once).

schedule-recurring! clj

(schedule-recurring! {:keys [schedules next-due opts]}
                     job-type
                     cron-expr
                     &
                     [job-opts])

Enqueue a job-type job on every tick of cron expression cron-expr. opts: :args, :queue, :priority, :max-retries, and :zone, a zone id like "Europe/Amsterdam". Calling it again for a job type replaces its schedule. Throws :validation-error for an expression that is malformed or never fires.

start! clj

(start! scheduler)

Run tick! every :tick-ms. Returns the scheduler with its :executor.

stop! clj

(stop! {:keys [state owner] :as scheduler})

Stop ticking and hand the lease to the next ticker.

tick! clj

(tick! {:keys [state schedules owner opts held? next-due watch] :as scheduler}
       now)

One ticker pass at now. Does nothing unless this scheduler holds the lease. Returns the job types it enqueued.