Background jobs

apistock.dev/modules/jobs runs background jobs on PostgreSQL with River. Decision: ADR-0033. Admin endpoints: ops API reference.

Concepts#

Changing a job's code always needs a deploy. Its configuration never does.

Adding a job#

Generate a job with the CLI, interactively or with flags (CLI guide):

aps gen job CleanupSessions                                            # asks the rest
aps gen job CleanupSessions --schedule "0 3 * * *" --timeout 5m --yes  # no questions

It creates these files (shown here with the worker filled in):

internal/jobs/cleanupsessions/cleanupsessions.go

package cleanupsessions

const Name = "cleanup_sessions" // public API: never rename

type Args struct{} // stored as JSON with each job; no personal data

func (Args) Kind() string { return Name }

type Worker struct {
	river.WorkerDefaults[Args]
	sessions SessionPurger
}

func NewWorker(sessions SessionPurger) *Worker { return &Worker{sessions: sessions} }

// Work returns an error to retry; it must respect ctx.
func (w *Worker) Work(ctx context.Context, job *river.Job[Args]) error {
	return w.sessions.PurgeExpired(ctx)
}

internal/app/job_cleanup_sessions.go

func defineCleanupSessionsJob(defs *jobs.Definitions, deps jobDeps) {
	jobs.Define(defs, jobs.Definition[cleanupsessions.Args]{
		Name:        cleanupsessions.Name,
		Description: "Deletes expired sessions.",
		Worker:      cleanupsessions.NewWorker(deps.sessions),
		NewArgs:     func() cleanupsessions.Args { return cleanupsessions.Args{} },
		Enabled:     true,
		Schedule:    "0 3 * * *",
		Timeout:     5 * time.Minute,
		MaxAttempts: 5,
	})
}

and adds defineCleanupSessionsJob(defs, deps) below //aps:anchor jobs in internal/app/jobs.go. Without the CLI, create the same files by hand.

Define panics at startup when the name isn't lowercase snake_case, Args.Kind() doesn't equal the name, the worker or NewArgs is missing, or the defaults are out of bounds. Zero Timeout, MaxAttempts, Queue and Priority become 1 minute, 25, default and 1.

Configuration#

Queue, priority and max attempts are applied to every job of a defined kind, whoever enqueues it. Scheduled jobs run once across all instances because only River's elected leader inserts them. The admin panel's next run time is approximate: the leader keeps the real timer in memory.

Wiring#

defs := jobs.NewDefinitions()
defineJobs(defs, jobDeps{logger: logger})

client, err := jobs.New(pool, nil,
	jobs.WithQueues(map[string]river.QueueConfig{river.QueueDefault: {MaxWorkers: 10}}),
	jobs.WithDefinitions(defs),
	jobs.WithLogger(logger),
	jobs.WithTracerProvider(tel.TracerProvider()),
)
manager, err := jobs.NewManager(ctx, pool, client, recorder, jobs.WithManagerLogger(logger))

runners := []app.Runner{server, client, manager}

Enqueueing from code#

res, err := client.Insert(ctx, cleanupsessions.Args{}, nil)

err = postgres.InTx(ctx, pool, func(tx pgx.Tx) error {
	// ... write rows with tx ...
	_, err := client.InsertTx(ctx, tx, welcome.Args{UserID: id}, nil) // only runs if tx commits
	return err
})

What a worker sees#

Authorise work when enqueuing: a job never runs with a user's permissions.

Failures, retries and shutdown#

• Return an error to retry with River's backoff; river.JobCancel(err) stops retrying; river.JobSnooze(d) retries later without using an attempt.
• Each failed attempt is logged once: a warning while attempts remain, an error on the last one. Panics are logged with their stack.
• On shutdown no new jobs are fetched; running jobs finish or are cancelled after the stop timeout.

Email#

workers := river.NewWorkers()
_ = jobs.AddMailWorker(workers, sender) // works "apistock.mail.send" jobs; sender is Resend, SMTP or Mailpit
client, err := jobs.New(pool, workers, jobs.WithQueues(jobs.DefaultQueues()))
mailer := mail.WithDefaults(jobs.AsyncSender(client), senderSettings) // validates, fills the sender, enqueues

Delivery is retried up to 8 times; each job's ID becomes the provider idempotency key (job-<id>) unless the message sets one, so retries never send twice. A send that fails with mail.ErrRejected (an unverified domain, a refused address) is cancelled at once instead of retried, and the run keeps the reason. Choosing the provider and the sender settings: email guide.

Managing jobs in Go#

Rules: writes need an authenticated actor; Update and Reset need the current version; disabling a job or changing an enabled job's schedule needs a reason; moving a job to a queue requires that queue to be active. Audit actions: jobs.definition.changed, jobs.definition.run_requested, jobs.run.retried, jobs.run.cancelled, jobs.queue.paused, jobs.queue.resumed.

Migrations and storage#

Testing#

• Worker logic: call Work directly with a &river.Job[Args]{JobRow: &rivertype.JobRow{ID: 1}}.
• Integration: pgtest.New(t, pgtest.WithMigrations(jobs.Migrations)), then jobs.Migrate, then run the client; wait with client.River().Subscribe(river.EventKindJobCompleted).