Queues

// app/jobs/send_invoice.go (vukem make:job SendInvoice)
type SendInvoice struct{ OrderID int64 }

func (SendInvoice) Kind() string { return "send_invoice" }

// HandleSendInvoice builds the handler from the app: database, clock, mailer.
func HandleSendInvoice(app *vukem.App) func(ctx context.Context, j SendInvoice) error {
	return func(ctx context.Context, j SendInvoice) error {
		db, err := app.DB()
		// ...
	}
}

// app/jobs/registry.go, passed to vukem.WithJobsFunc by bootstrap
func All(app *vukem.App) []queue.Registration {
	return []queue.Registration{
		queue.Handle(HandleSendInvoice(app), queue.Retries(5), queue.Timeout(time.Minute)),
	}
}

vukem.WithJobsFunc(jobs.All) calls All once, when the queue is first built, so handlers get the app's services without globals. It runs inside the first app.Queue() call (a panic becomes that call's error). All may get app.DB(), app.Mailer(), app.Crypt and app.Now(), but must not dispatch: app.Queue(), app.Events.Dispatch with queued listeners, SendLater/MailLater or running scheduled jobs deadlock there (a handler may, when it runs). vukem.WithJobs(regs...) still registers handlers that need nothing from the app.

q, err := app.Queue()
err = q.Dispatch(ctx, jobs.SendInvoice{OrderID: 1}, queue.OnQueue("billing"), queue.Delay(5*time.Minute))
err = q.DispatchTx(ctx, tx, jobs.SendInvoice{OrderID: 1}) // only if tx commits

• Payloads are JSON; Kind() names the handler and must stay stable across deploys. Registering a kind twice is a boot error.
• Defaults: queue default, 3 retries, exponential backoff from 1s (capped at 1h), 1-minute timeout per attempt. Options on Handle set defaults; options on Dispatch override them. Backoffs: queue.Exponential(base), queue.Constant(d), or any func(attempt int) time.Duration.
• Workers recover panics, retry with backoff and then move the job to the failed jobs with its last error. On shutdown they stop reserving and give running jobs SHUTDOWN_TIMEOUT to finish, then cancel their ctx: a job that fails with that cancellation goes back to the queue without using an attempt, up to three times in a row (queue.MaxFreeInterruptions), then interruptions use attempts; any other error is retried or failed as usual (queue.WorkOptions.StopTimeout; 0 waits for each job's own timeout). An interrupted job runs again from the start, so handlers must be idempotent and pass ctx to their queries and calls. Settling a job (ack, release, fail) is bounded by 10 seconds, so a hanging database cannot hold queue:work.
• Delays, retry backoff and the database and Redis drivers' availability checks use the app clock (vukem.WithClock, the fake clock of vukemtest), so advancing a fake clock makes delayed jobs due. app.Queue() sets it on every driver implementing queue.Clock (SetClock); queue.Queue.SetClock sets it outside the app.
• DispatchTx(ctx, tx, job) works on every driver: the database driver inserts the job inside the transaction; sync, memory, queue.Fake and Redis dispatch it right after the commit (database.Tx.AfterCommit), and never for rolled-back work. On sync the job runs after the commit, so its error is logged instead of returned.
• queue:work -once and -stop-when-empty exit with an error when the store fails (instead of treating it as an empty queue); a long-running worker logs the error and keeps polling.
• On the database driver, jobs dispatched in the same millisecond pop in dispatch order within one process (a seq column, seeded by the wall clock, breaks the tie); jobs from several processes in the same millisecond interleave roughly by real time. CreateTables (run at boot) adds seq to tables created by older releases; a jobs table you create with your own migration needs seq BIGINT NOT NULL DEFAULT 0.
• sync cannot delay a job: Dispatch(ctx, j, queue.Delay(d)) returns queue.ErrInlineDelay instead of running it early. Tests that dispatch delayed jobs use queue.NewFake().

Drivers#

The database and Redis drivers reserve each job atomically (Postgres uses FOR UPDATE SKIP LOCKED). A job whose worker died is taken over after its timeout plus one minute, so delivery is at least once: make handlers idempotent. If the first worker was only slow (not dead) and finishes after the takeover, its result is dropped with queue.ErrReservationLost instead of deleting, retrying or failing the job the second worker now owns.

Running workers#

serve runs a queue worker next to the HTTP server, so a single-process deploy (one Railway, Fly or Render service) still sends queued mail and runs jobs: always for memory, and for database and custom drivers such as Redis unless QUEUE_IN_SERVE=false. sync needs no worker. The worker polls QUEUE_NAMES with QUEUE_CONCURRENCY workers and stops with the server (running jobs get SHUTDOWN_TIMEOUT).

To process jobs only in dedicated processes, set QUEUE_IN_SERVE=false on the web service and run queue:work (below). Leaving the in-serve worker on next to queue:work processes is safe: each reservation is atomic and owned by one worker, so the extra worker never processes a job a second time at once; it only adds workers, and their load, to the web process. On SQLite every worker shares the single writer connection, so keep QUEUE_CONCURRENCY low there.

Commands#

vukem queue:work                                  # until Ctrl+C (drains running jobs within SHUTDOWN_TIMEOUT)
vukem queue:work -queue high,default -concurrency 8
vukem queue:work -once                            # a single job
vukem queue:work -stop-when-empty
vukem queue:failed [-json]
vukem queue:retry <id…|all>
vukem queue:forget <id…|all>

Queued mail and listeners#

err := mailer.SendLater(ctx, mail.Welcome{User: u}) // or app.MailLater: rendered now, sent from a retried job
err = app.MailLaterTx(ctx, tx, mail.Receipt{Order: o}) // or mailer.SendLaterTx: queued only if tx commits

// app/listeners/registry.go (vukem make:listener UpdateStock --event OrderShipped --queued)
events.OnQueued("update_stock", UpdateStock)

SendLater keeps the message ID across retries, so providers that deduplicate by it send once, and stores the message encrypted with app.Crypt. SendLaterTx/MailLaterTx tie the mail to a transaction like DispatchTx, so a rolled-back payment never emails the buyer. A queued listener's event is stored as JSON and run by a worker; Dispatch only reports enqueue errors.

Testing#

fake := queue.NewFake()
app := vukemtest.New(t, bootstrap.New, vukemtest.Options{
	AppOptions: []vukem.Option{vukem.WithQueueDriver(fake)},
})
// ... exercise the app ...
invoices := queue.Dispatched[jobs.SendInvoice](fake)

q, _ := app.Queue()
_ = q.Work(t.Context(), queue.WorkOptions{StopWhenEmpty: true}) // run them when you want

Custom drivers implement queue.Driver (and optionally queue.FailedStore and queue.TxPusher) and should pass queue/queuetest.Run.