# Background jobs

> Move slow work out of requests with a PostgreSQL-backed job queue: SKIP LOCKED workers, retries with backoff, idempotency and scheduled jobs.

Backend with Go roadmap · Data · 18 min read · https://levelcode.org/roadmaps/go-backend/background-jobs

Some work doesn't belong in a request. Sending a welcome email, resizing an upload, calling a slow partner API or generating a monthly report can take seconds, can fail and be retried, and the user doesn't need to wait for it. A **background job** records that the work must happen, returns to the user immediately, and lets a worker do it later. This lesson builds a reliable job queue on the PostgreSQL database you already run, then covers retries, at-least-once delivery, shutdown and scheduled jobs.

## Why not just start a goroutine?

The tempting version is one line:

```go
go sendWelcomeEmail(user.Email) // don't do this for work that matters
```

It works in development and loses work in production:

- **Deploys and crashes lose it.** A goroutine lives in memory. When the process restarts mid-send, the email is never sent and nothing records that it should have been.
- **No retries.** If the mail provider returns a 503, the error is logged (if you're lucky) and the work is gone.
- **No limits.** A burst of 10,000 sign-ups starts 10,000 concurrent sends and gets you rate-limited.
- **No visibility.** You can't answer "how many emails are waiting?" or "which ones failed?".

A job queue fixes all four by storing the job **durably** before acknowledging the request, and running it with bounded concurrency, retries and a record of what happened.

## A jobs table

If you already run PostgreSQL, it makes a good queue for most services: it's durable, transactional and needs no new infrastructure. Add a migration:

```sql
-- +goose Up
CREATE TABLE jobs (
    id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    kind         text        NOT NULL,
    payload      jsonb       NOT NULL,
    status       text        NOT NULL DEFAULT 'pending'
                             CHECK (status IN ('pending', 'running', 'done', 'failed')),
    attempts     int         NOT NULL DEFAULT 0,
    max_attempts int         NOT NULL DEFAULT 8,
    run_at       timestamptz NOT NULL DEFAULT now(),
    locked_until timestamptz,
    last_error   text,
    unique_key   text,
    created_at   timestamptz NOT NULL DEFAULT now()
);

-- Workers only ever look for pending jobs that are due.
CREATE INDEX jobs_ready ON jobs (run_at) WHERE status = 'pending';
-- Optional de-duplication for scheduled jobs (see below).
CREATE UNIQUE INDEX jobs_unique ON jobs (kind, unique_key) WHERE unique_key IS NOT NULL;

-- +goose Down
DROP TABLE jobs;
```

The partial index stays small however many finished jobs accumulate, because it only contains pending rows.

## Enqueue in the same transaction

The biggest advantage of a database queue is that you can enqueue a job **in the same transaction** as the change that caused it. Either both happen or neither does:

```go
type WelcomeEmail struct {
	UserID int64  `json:"user_id"`
	Email  string `json:"email"`
}

func (s *Store) CreateUser(ctx context.Context, email, hash string) (int64, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback(ctx) // no-op after Commit

	var id int64
	err = tx.QueryRow(ctx,
		`INSERT INTO users (email, password_hash) VALUES ($1, $2) RETURNING id`, email, hash,
	).Scan(&id)
	if err != nil {
		return 0, err
	}

	payload, _ := json.Marshal(WelcomeEmail{UserID: id, Email: email})
	if _, err := tx.Exec(ctx,
		`INSERT INTO jobs (kind, payload) VALUES ('welcome_email', $1)`, payload,
	); err != nil {
		return 0, err
	}
	return id, tx.Commit(ctx)
}
```

With a separate queue such as Redis or SQS, you'd write the user, then publish the job, and a crash between the two leaves a user with no email (or, the other way round, an email for a user who doesn't exist). That gap is the dual-write problem, and the [transactional outbox](https://levelcode.org/concepts/transactional-outbox) is the standard fix when the queue lives elsewhere.

## Claiming jobs safely with SKIP LOCKED

Several workers, perhaps in several processes, poll the same table. Two of them must never run the same job at once. `SELECT ... FOR UPDATE SKIP LOCKED` (PostgreSQL 9.5+) solves this: each worker locks a row, and other workers skip locked rows instead of waiting for them.

```go
type Job struct {
	ID          int64
	Kind        string
	Payload     []byte
	Attempts    int
	MaxAttempts int
}

const claimSQL = `
UPDATE jobs
SET status = 'running', attempts = attempts + 1, locked_until = now() + make_interval(secs => $1)
WHERE id = (
    SELECT id FROM jobs
    WHERE status = 'pending' AND run_at <= now()
    ORDER BY run_at
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
RETURNING id, kind, payload, attempts, max_attempts`

func (q *Queue) claim(ctx context.Context) (*Job, error) {
	var j Job
	err := q.pool.QueryRow(ctx, claimSQL, q.lease.Seconds()).
		Scan(&j.ID, &j.Kind, &j.Payload, &j.Attempts, &j.MaxAttempts)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil // nothing to do
	}
	return &j, err
}
```

The `UPDATE` marks the job `running` and commits immediately, so the row lock is held only for milliseconds, not for the length of the job. The **lease** (`locked_until`) handles workers that die mid-job: a periodic reaper returns expired running jobs to the queue.

```go
const reapSQL = `
UPDATE jobs SET status = 'pending', run_at = now()
WHERE status = 'running' AND locked_until < now()`
```

Set the lease comfortably longer than your slowest job, and give each job a timeout shorter than the lease.

## Running, retrying and failing

A worker loop claims a job, runs the handler registered for its kind, and records the outcome:

```go
type Handler func(ctx context.Context, payload []byte) error

type Queue struct {
	pool     *pgxpool.Pool
	handlers map[string]Handler
	lease    time.Duration
	timeout  time.Duration
}

func (q *Queue) work(ctx context.Context) {
	for ctx.Err() == nil {
		job, err := q.claim(ctx)
		if err != nil {
			slog.ErrorContext(ctx, "claim failed", "err", err)
		}
		if job == nil {
			select { // idle: poll again shortly
			case <-ctx.Done():
			case <-time.After(time.Second):
			}
			continue
		}
		q.run(ctx, job)
	}
}

func (q *Queue) run(ctx context.Context, job *Job) {
	// Let an in-flight job finish even if shutdown cancels ctx, but never past its timeout.
	jobCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout)
	defer cancel()

	h, ok := q.handlers[job.Kind]
	err := fmt.Errorf("no handler for kind %q", job.Kind)
	if ok {
		err = h(jobCtx, job.Payload)
	}
	if err == nil {
		_, err = q.pool.Exec(jobCtx, `UPDATE jobs SET status = 'done', locked_until = NULL WHERE id = $1`, job.ID)
		if err != nil {
			slog.Error("mark done failed", "job", job.ID, "err", err)
		}
		return
	}

	slog.Warn("job failed", "job", job.ID, "kind", job.Kind, "attempt", job.Attempts, "err", err)
	if job.Attempts >= job.MaxAttempts {
		q.pool.Exec(jobCtx, `UPDATE jobs SET status = 'failed', last_error = $2, locked_until = NULL WHERE id = $1`,
			job.ID, err.Error())
		return
	}
	q.pool.Exec(jobCtx, `UPDATE jobs SET status = 'pending', last_error = $2, locked_until = NULL,
		run_at = now() + make_interval(secs => $3) WHERE id = $1`,
		job.ID, err.Error(), backoff(job.Attempts).Seconds())
}

// backoff: 2s, 4s, 8s ... capped at 1h, with up to 50% random jitter.
func backoff(attempt int) time.Duration {
	d := min(time.Second<<attempt, time.Hour)
	return d/2 + rand.N(d/2+1)
}
```

Start a fixed number of workers, which bounds concurrency:

```go
func (q *Queue) Start(ctx context.Context, workers int) *sync.WaitGroup {
	var wg sync.WaitGroup
	for range workers {
		wg.Go(func() { q.work(ctx) })
	}
	wg.Go(func() { // reaper
		t := time.NewTicker(30 * time.Second)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				if _, err := q.pool.Exec(ctx, reapSQL); err != nil {
					slog.Error("reap failed", "err", err)
				}
			}
		}
	})
	return &wg
}
```

`sync.WaitGroup.Go` is new in Go 1.25; on older versions use `wg.Add(1)` and `defer wg.Done()` inside a `go func`. On shutdown, cancel `ctx` and `Wait`: workers stop claiming new jobs, and in-flight jobs finish within their timeout, which pairs with the graceful shutdown from the [deployment lesson](https://levelcode.org/roadmaps/go-backend/deploying-go).

A `failed` job is a dead letter: it stays in the table for you to inspect, fix and re-queue with `UPDATE jobs SET status = 'pending', attempts = 0 WHERE id = ...`. Alert when the count of failed jobs grows.

## At-least-once means handlers must be idempotent

Consider a worker that sends the email and then crashes before marking the job `done`. The lease expires, the reaper re-queues the job, and the email is sent twice. No queue can fully prevent this: between "did the work" and "recorded that I did it" there is always a gap. Queues therefore promise **at-least-once** execution, and your handlers must be **idempotent**: running twice has the same effect as running once.

Practical techniques:

- **Pass an idempotency key to external APIs.** Many payment and email providers accept one; use the job ID, so a retry is recognised as the same request.
- **Check before acting.** "Mark invoice sent" can be `UPDATE invoices SET sent_at = now() WHERE id = $1 AND sent_at IS NULL`, and the handler skips the send if zero rows changed. Do the check and the record in a transaction where possible.
- **Make the effect naturally idempotent.** "Set the thumbnail for image 42" can be repeated; "append a thumbnail" can't.

> [!IMPORTANT]
> Keep payloads small and stable: IDs, not whole records. A job that runs an hour later should load the current data, and a payload format that changes between deploys breaks jobs that were enqueued by the old version.

## Scheduled and recurring jobs

Two kinds of scheduling come up:

- **Delayed**: "remind the user in 24 hours". Insert with `run_at = now() + interval '24 hours'`.
- **Recurring**: "build the daily report at 02:00". If every replica runs a ticker that enqueues the job, you get one report per replica.

The unique index from the migration de-duplicates recurring jobs: every replica tries to enqueue the job for the same period, and only the first insert wins.

```go
func enqueueDailyReport(ctx context.Context, pool *pgxpool.Pool, day time.Time) error {
	_, err := pool.Exec(ctx, `
		INSERT INTO jobs (kind, payload, unique_key, run_at)
		VALUES ('daily_report', $1, $2, $3)
		ON CONFLICT (kind, unique_key) WHERE unique_key IS NOT NULL DO NOTHING`,
		[]byte(`{}`), day.Format(time.DateOnly), day.Add(2*time.Hour))
	return err
}
```

Finally, clean up: delete `done` jobs older than a few days so the table doesn't grow forever.

## When to outgrow it

A PostgreSQL queue comfortably handles hundreds of jobs per second, which covers most products. Production-grade Go libraries such as River implement this same design with LISTEN/NOTIFY for instant pickup, unique jobs, a UI and periodic jobs, and are worth adopting instead of maintaining your own. Move to a dedicated broker (SQS, RabbitMQ, Kafka) when job volume would load your primary database, when other teams' services must consume the same events, or when you need streaming replay. The [System Design roadmap's queues lesson](https://levelcode.org/roadmaps/system-design/queues-and-streams) compares the options.

## Practice

Build the queue from this lesson in a small service:

1. Add the migration and the `Queue` type with `Enqueue(ctx, tx, kind, payload)`, `Start` and the reaper.
2. Register a `welcome_email` handler that "sends" by logging, and fails randomly 50% of the time. Enqueue 20 jobs and watch them retry and complete.
3. Make the handler idempotent with an `emails_sent (user_id bigint PRIMARY KEY)` table: insert the row and "send" in one transaction, skipping the send if the row already exists.
4. Kill the process with `kill -9` while jobs are running, restart it, and confirm the reaper recovers them once the lease expires.

**Solution**

```go
func (q *Queue) Enqueue(ctx context.Context, tx pgx.Tx, kind string, payload any) error {
	b, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `INSERT INTO jobs (kind, payload) VALUES ($1, $2)`, kind, b)
	return err
}

func welcomeEmail(pool *pgxpool.Pool) Handler {
	return func(ctx context.Context, payload []byte) error {
		var p WelcomeEmail
		if err := json.Unmarshal(payload, &p); err != nil {
			return fmt.Errorf("bad payload: %w", err) // will exhaust retries; fine to fail
		}
		if rand.IntN(2) == 0 {
			return errors.New("simulated provider 503")
		}
		return pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
			tag, err := tx.Exec(ctx,
				`INSERT INTO emails_sent (user_id) VALUES ($1) ON CONFLICT DO NOTHING`, p.UserID)
			if err != nil {
				return err
			}
			if tag.RowsAffected() == 0 {
				slog.Info("already sent, skipping", "user", p.UserID)
				return nil
			}
			slog.Info("sending welcome email", "to", p.Email)
			return nil
		})
	}
}
```

Keying `emails_sent` on the user (one welcome email per user) is even better than keying on the job ID: it also stops duplicates if the same job were somehow enqueued twice. With a real provider, the "send" happens outside the database, so pass the user ID as the provider's idempotency key as well; the table then guards your side and the key guards theirs.

After `kill -9`, jobs left in `running` stay there until `locked_until` passes; the reaper then resets them to `pending` and they run again, which is exactly why the handler has to be idempotent.

## Key takeaways

- A bare goroutine loses work on restart and has no retries, limits or visibility; durable jobs fix all four.
- PostgreSQL makes a solid queue: enqueue in the same transaction as the change, and claim with `FOR UPDATE SKIP LOCKED`.
- Use a lease plus a reaper for crashed workers, exponential backoff with jitter for retries, and a `failed` state as your dead-letter queue.
- Execution is at-least-once, so handlers must be idempotent; idempotency keys and "insert if absent" records are the usual tools.
- De-duplicate recurring jobs with a unique key per period, and adopt a library such as River before hand-maintaining the details.

## Quiz

1. Why enqueue the welcome-email job inside the same transaction that inserts the user?

   A. It's faster than two statements
   B. Either both the user and the job are committed, or neither is, so no user is missed and no orphan email is sent
   C. Postgres requires it for jsonb
   D. It prevents the job from being retried

   **Answer: B.** A single transaction removes the dual-write gap: a crash can't leave a user without the job, or a job for a user that was rolled back.

2. What does FOR UPDATE SKIP LOCKED achieve when several workers claim jobs?

   A. Each worker skips rows already locked by another worker, so no two workers claim the same job and none of them wait
   B. It deletes locked rows
   C. It makes the claim query read-only
   D. It guarantees jobs run exactly once

   **Answer: A.** Locked rows are skipped instead of blocking, which gives non-overlapping claims with high concurrency. It doesn't guarantee exactly-once execution; crashes can still cause re-runs.

3. A worker sends an email, then crashes before marking the job done. What should you expect?

   A. The job is lost
   B. The job stays running forever
   C. After the lease expires the job is re-queued and the email may be sent again, so the handler must be idempotent
   D. Postgres rolls back the email

   **Answer: C.** This is why queues are at-least-once. The reaper recovers the job, and idempotency (keys or insert-if-absent records) prevents a duplicate effect.

4. Three replicas each run a ticker that enqueues the 02:00 daily report. How do you make sure only one report job exists per day?

   A. Only run the ticker in one replica by hostname
   B. Use a longer ticker interval
   C. Use a goroutine instead of a job
   D. Insert with a unique key for the day and ON CONFLICT DO NOTHING

   **Answer: D.** A unique index on (kind, unique_key) lets every replica try to insert the same day's job; only the first succeeds. Pinning to one hostname breaks when that replica is down.