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 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:

-- +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:

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 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.

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.

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:

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:

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.

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.

#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.

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 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.

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
		})
	}
}

#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.