Scheduling

Import the scheduler subpackage to define cron schedules and jobs in code:

import "github.com/gp-system/events/scheduler"

scheduler lets you define your schedule in code, not a crontab: you build a Schedule fluently and the worker runs it. Running on a single server among replicas is the default behavior here, not an opt-in: among N worker replicas, a Valkey-lease leader election decides which one fires, with no extra flag and no cache lock to configure.

You can schedule two kinds of things:

• an event: a fire is a plain dispatch, the event fans out to all of its listeners;
• a job: a single named handler, no fan-out.

Building the schedule

Every module's generated jobs/register.go provides a Register(sched, deps); the worker composes them into a single schedule. The shop's nightly report was generated by add job shop nightlySalesReport --cron "0 3 * * *":

package jobs

import (
    "github.com/acme/shop/internal/modules/shop"

    "github.com/gp-system/gpsystem/scheduler"
    // gpsystem:job-imports
)

func Register(sched *scheduler.Schedule, deps shop.Dependencies) {
    sched.Job("shop.nightlySalesReport", "0 3 * * *", nightlySalesReport(deps))
    // gpsystem:jobs
}

The job's handler stub lives in its own file; the body is yours:

package jobs

import (
    "context"

    "github.com/acme/shop/internal/modules/shop"
)

func nightlySalesReport(deps shop.Dependencies) func(context.Context) error {
    return func(ctx context.Context) error {
        report, err := buildSalesReport(ctx, deps.DB)
        if err != nil {
            return err
        }
        return deps.Mailer.Send(ctx, salesReportMessage(report))
    }
}

A returned error triggers a retry (up to the default budget of 25, or the entry's own queue.MaxRetry); a transient DB error doesn't cost you the report.

Event schedules vs job schedules

The Schedule fluent API: every method returns *Schedule, so calls chain.

Which one when? An event schedule when the fire needs several independently retried reactions, or when the same event is also dispatched at runtime: the listeners don't know and don't need to know whether a cron or a user action dispatched it. A job schedule when it is one named piece of work, like the shop's report.

// Event schedule: every fire dispatches a fresh event that fans out
// to all of its listeners, exactly like a runtime Dispatch.
sched.Every(15*time.Minute, shopevents.SyncInventory{})
sched.DailyAt("06:30", shopevents.DigestDue{})

// Job schedule: one handler, no fan-out; the task type is "job:<name>".
sched.Job("shop.nightlySalesReport", "0 3 * * *", nightlySalesReport(deps))

The opts... are the usual enqueue options (e.g. queue.OnQueue("reports"), queue.Timeout(10*time.Minute)), and apply to the task submitted on each fire.

One technical detail that provides the guarantees: when a scheduled event fires, the worker uses the trigger task's id as the envelope id. So every fire is a new dispatch (every listener receives it again), but a retry of the same fire is idempotent: it doesn't double the fan-out.

Cron syntax

The spec is the standard 5-field cron (minute hour day-of-month month day-of-week), plus asynq's descriptors:

"0 3 * * *"     every day at 03:00
"*/10 * * * *"  every 10 minutes
"0 8 * * 1"     Mondays at 08:00
"@every 30m"    every 30 minutes (not on the clock grid, counted from start)
"@daily"        at midnight

Expressions are evaluated in SCHEDULER_TIMEZONE (an IANA name, e.g. Europe/Budapest); the default is UTC, so the shop's "0 3 * * *" is 03:00 UTC.

Replica safety: the leader election

The schedule is executed by the Runner, which you never construct by hand: worker.Run starts it automatically when the schedule is non-empty. Inside, asynq's Scheduler does the work, guarded by a Valkey lease:

1. Every worker replica tries to acquire the lease key (SETNX, expiring after SCHEDULER_LEASE_TTL).
2. Only the lease holder registers and runs the cron entries; it renews the lease every LEASE_TTL/3 (a Lua compare-and-act, so a stale old leader cannot clobber the new one).
3. If the leader dies or loses the lease, its firing stops, and another replica takes over after at most LEASE_TTL.

A brief overlap is possible on handover (the old leader issued one more tick, the new one did too), but the rest of the system absorbs it: delivery is at-least-once anyway, and listeners are idempotent. When several applications (or environments) share one Valkey, separate their lease keys with SCHEDULER_NAMESPACE; otherwise they share each other's leader election.

// "run on one server" is not an option here, it's the only mode of operation
sched.Job("shop.nightlySalesReport", "0 3 * * *", nightlySalesReport(deps))

What runs where

An important difference from a crontab: firing and execution are decoupled. The leader only enqueues: the schedule:<event> or job:<name> task lands on the queue, and any worker replica may process it, under the usual concurrency and queue weights. A heavy report therefore doesn't load "the scheduler machine"; it loads the pool.

Configuration

scheduler.Config arrives embedded in worker.Config, under the SCHEDULER_ prefix:

With the kit

The worker chassis is what actually constructs and runs the Runner: worker.Run starts it automatically whenever the Schedule your RegisterFunc builds is non-empty, and stops it in the same shutdown sequence as the asynq server and the outbox relay. Nothing here is kit-specific, though: a standalone consumer builds a Schedule, opens a queue.Config, and calls scheduler.NewRunner(ctx, cfg, queueCfg, sched).Run(ctx) directly.

Patterns used

Valkey-lease leader election (SETNX to acquire, a Lua compare-and-renew script to renew): see Design patterns.