go / job queues

updated October 21, 2026

My job queue runs in a production jobs service on one Postgres table:

• Each queue runs one job at a time, First In, First Out.
• A job is a func(ctx, db, ...) (status string, err error).
• The only dependencies are Go, Postgres, and a pgx pool wrapper.

Queues

The service has ~40 queues. Most call a rate-limited API: Apollo, GitHub, Discord, Slack, Anthropic, Pitchbook. One job at a time keeps a queue inside its rate limit.

The table

CREATE TABLE jobs (
  id SERIAL,
  queue text NOT NULL,
  name text NOT NULL,
  args jsonb DEFAULT '{}' NOT NULL,
  status text DEFAULT 'pending'::text NOT NULL,
  callsite text,
  created_at timestamp DEFAULT now() NOT NULL,
  started_at timestamp,
  finished_at timestamp
);

The shared poller

One Poller struct holds the shared code. Each queue supplies Dispatch, which routes a job name to a handler, and an optional Throttle, which sets the sleep between jobs.

type ThrottleFunc func(status string, err error, elapsed time.Duration) time.Duration

type DispatchFunc func(ctx context.Context, job PendingJob) (string, error)

type PendingJob struct {
	ID   int64  `db:"id"`
	Name string `db:"name"`
	Args []byte `db:"args"`
}

type Poller struct {
	Queue        string
	DB           *pgdb.DB
	Dispatch     DispatchFunc
	JobTimeout   time.Duration
	Throttle     ThrottleFunc
	PollInterval time.Duration
}

The loop marks stale jobs on boot, then works pending rows until the context ends. An enqueue wakes it through Wake (see go / wakeups), and the interval is the fallback:

func (p *Poller) Poll(ctx context.Context) {
	interval := p.PollInterval
	if interval <= 0 {
		interval = defaultPollInterval
	}

	// Every started row on this queue belongs to the process that
	// died, so the startup sweep takes them whatever their age.
	p.interruptStarted(ctx, 0)
	lastSweep := time.Now()

	// reported holds whether the last fetch failure reached sockeye.
	reported := false

	for {
		if time.Since(lastSweep) >= staleSweepInterval {
			p.interruptStarted(ctx, p.jobTimeout()+staleSweepMargin)
			lastSweep = time.Now()
		}

		jobs, err := p.fetch(ctx)
		if ctx.Err() != nil {
			return
		}
		if err != nil {
			log.Printf("queue=%s fetch error: %v", p.Queue, err)
			if !reported {
				errs.CaptureException(errs.Wrap(err, "fetch pending jobs"))
				reported = true
			}
		} else {
			reported = false
		}

		for _, job := range jobs {
			p.WorkOnce(ctx, job)
			if ctx.Err() != nil {
				return
			}
		}
		if len(jobs) > 0 {
			continue
		}

		select {
		case <-p.Wake:
		case <-time.After(interval):
		case <-ctx.Done():
			return
		}
	}
}

A database that nobody can read fails every cycle, on every queue. One error event for each run of failures says as much as eighty a minute, so reported resets only after a fetch succeeds.

WorkOnce claims a job, dispatches it with a deadline, and finalizes it in a deferred block that also catches a panic. It writes one log line per job:

func (p *Poller) WorkOnce(ctx context.Context, job PendingJob) {
	var latency float64
	err := p.DB.QueryRow(ctx, qClaimJob, job.ID).Scan(&latency)
	if err != nil {
		log.Printf("queue=%s job=%s id=%d claim error: %v", p.Queue, job.Name, job.ID, err)
		return
	}

	start := time.Now()
	var status string
	var workErr error

	defer func() {
		if r := recover(); r != nil {
			status = fmt.Sprintf("err: panic: %v", r)
			captureErr(job, fmt.Errorf("panic: %v", r))
		}
		p.finalize(ctx, job, status)
		elapsed := time.Since(start)
		log.Printf("queue=%s job=%s id=%d status=%q latency=%.2fs duration=%.2fs",
			p.Queue, job.Name, job.ID, status, latency, elapsed.Seconds())

		if p.Throttle != nil {
			if delay := p.Throttle(status, workErr, elapsed); delay > 0 {
				select {
				case <-time.After(delay):
				case <-ctx.Done():
				}
			}
		}
	}()

	jobCtx, cancel := context.WithTimeout(ctx, p.jobTimeout())
	defer cancel()

	status, workErr = p.Dispatch(jobCtx, job)
	if workErr != nil {
		if shouldCaptureDispatchError(job, workErr) {
			captureErr(job, workErr)
		}
		status = "err: " + workErr.Error()
	}
}

I export WorkOnce so a test runs one job without the poll loop.

The claim is a conditional update, so a worker cannot claim a job twice:

UPDATE jobs
SET started_at = now(), status = 'started'
WHERE id = $1 AND status = 'pending'
RETURNING coalesce(extract(EPOCH FROM now() - created_at)::float8, 0) AS latency;

A claim that finds no row is the ordinary answer: another poller got there first. Any other claim error goes to sockeye.

A lost connection

One night, two jobs lost their database connections in the same second. The job query reported conn closed, and the finalize after it reported write: software caused connection abort. The pool pings an idle connection before it hands it out. The dialer sends a TCP keepalive every 15 seconds. So an idle connection was not the cause. I never found the cause, so the loop now recovers from a lost connection and reports what it cannot recover.

finalize runs the write once more when the failure is the connection, not Postgres:

func (p *Poller) finalize(ctx context.Context, job PendingJob, status string) {
	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), finalizeTimeout)
	defer cancel()

	err := p.DB.Exec(ctx, qFinalizeJob, status, job.ID)
	if err != nil && pgdb.IsConnFailure(err) {
		err = p.DB.Exec(ctx, qFinalizeJob, status, job.ID)
	}
	if err != nil {
		captureErr(job, errs.Wrap(err, "finalize job"))
	}
}

The update sets two columns by primary key, so a second run is safe whether or not the first reached the server. The pool discards the connection that failed, so the second attempt runs on a connection that answered. finalize detaches from ctx, because a deploy cancels ctx in the middle of a job.

IsConnFailure reads any error that did not come from the server as a lost connection. A cancelled context is the caller's own deadline, so it is not a reason to run the statement again:

func IsConnFailure(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return false
	}
	_, fromServer := errors.AsType[*pgconn.PgError](err)
	return !fromServer
}

A finalize that fails twice leaves the row in started. That row is more than a wrong reading on an admin page. Some jobs dedupe with a unique partial index over (queue, name, args) where the status is pending or started. One stuck row then blocks every later enqueue of that job. Before, only the startup sweep cleared it, so the fix cost a deploy. Now the poller sweeps every five minutes while it runs:

UPDATE jobs
SET status = $2, finished_at = now()
WHERE queue = $1
  AND status = 'started'
  AND (
    started_at IS NULL
    OR started_at < now() - make_interval(secs => $3)
  );

The sweep in the loop passes the queue's job timeout plus a minute, because a younger row is the job that the poller runs now.

A worker

Each queue package exports NewWorker, which returns a Poller. The closure captures dependencies such as an API client:

func NewWorker(db *pgdb.DB, client API) *jobs.Poller {
	return &jobs.Poller{
		Queue:    "apollo",
		DB:       db,
		Throttle: throttle,
		Dispatch: func(ctx context.Context, job jobs.PendingJob) (string, error) {
			switch job.Name {
			case "apollo.IngestCompany":
				args, err := jobs.UnmarshalArgs[IngestCompanyArgs](job.Args)
				if err != nil {
					return "", err
				}
				return IngestCompany(ctx, db, client, args)
			case "apollo.RefreshPerson":
				args, err := jobs.UnmarshalArgs[RefreshPersonArgs](job.Args)
				if err != nil {
					return "", err
				}
				return RefreshPerson(ctx, db, client, args)
			default:
				return "", fmt.Errorf("unknown job %q for queue apollo", job.Name)
			}
		},
	}
}

UnmarshalArgs decodes the JSONB payload:

func UnmarshalArgs[T any](raw []byte) (T, error) {
	var args T
	if err := json.Unmarshal(raw, &args); err != nil {
		return args, fmt.Errorf("unmarshal args: %w", err)
	}
	return args, nil
}

Throttle is a pure function, so a test needs no clock. The Apollo queue targets 150 jobs a minute and backs off when a status reports an exhausted rate limit:

func throttle(status string, _ error, elapsed time.Duration) time.Duration {
	minJobTime := time.Minute / maxJobsPerMinute
	for _, p := range rateLimitStatusPrefixes {
		if strings.HasPrefix(status, p.prefix) {
			minJobTime = p.backoff
			break
		}
	}
	if elapsed >= minJobTime {
		return 0
	}
	return minJobTime - elapsed
}

No throttle

A queue with no external API leaves Throttle nil:

func NewWorker(db *pgdb.DB) *jobs.Poller {
	return &jobs.Poller{
		Queue: "caches",
		DB:    db,
		Dispatch: func(ctx context.Context, job jobs.PendingJob) (string, error) {
			switch job.Name {
			case "caches.Age":
				return Age(ctx, db)
			// ...
			default:
				return "", fmt.Errorf("unknown job %q for queue caches", job.Name)
			}
		},
	}
}

The registry

One process runs one goroutine per poller. main holds a list of constructors, so a new queue is one line:

registry = []func(*pgdb.DB) *jobs.Poller{
	func(db *pgdb.DB) *jobs.Poller { return caches.NewWorker(db) },
	func(db *pgdb.DB) *jobs.Poller { return github.NewWorker(db, githubClient) },
	func(db *pgdb.DB) *jobs.Poller { return apollo.NewWorker(db, apolloClient) },
	func(db *pgdb.DB) *jobs.Poller { return anthropic.NewWorker(db, anthropicClient) },
	// ... one line per queue
}

The list length sets the Postgres pool size, one connection per poller plus a small buffer:

pollers := make([]*jobs.Poller, len(registry))
for i, newPoller := range registry {
	pollers[i] = newPoller(db)
}

ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
defer stop()

var wg sync.WaitGroup
for _, p := range pollers {
	wg.Go(func() { p.Poll(ctx) })
}
wg.Wait()

On SIGTERM, each loop finishes its active job and returns. A hard kill leaves a job in started, and the poller resets it to err: interrupted on the next boot.

A test checks the registry against every writer of a job row. A scheduled job, a job name in a SQL file, and a job that an Insert call names must each have a handler. The Insert check parses each Go file with go/ast, reads the queue and name arguments, and fails on a pair that no worker holds. Without it, such a row stays pending, and only a person who looks sees it.

Enqueuing

Insert writes one row and records the caller's file and line as the callsite:

func Insert(ctx context.Context, db *pgdb.DB, queue string, name string, args any) (int64, error) {
	argsJSON, err := json.Marshal(args)
	if err != nil {
		return 0, fmt.Errorf("marshal args: %w", err)
	}

	callsite := "go:0"
	if _, file, line, ok := runtime.Caller(1); ok {
		short := trimToModulePath(file)
		callsite = fmt.Sprintf("%s:%d", short, line)
	}

	var id int64
	err = db.QueryRow(ctx, `
		INSERT INTO jobs (queue, name, callsite, args)
		VALUES ($1, $2, $3, $4::jsonb)
		ON CONFLICT DO NOTHING
		RETURNING id
	`, queue, name, callsite, string(argsJSON)).Scan(&id)
	if errors.Is(err, pgx.ErrNoRows) {
		return 0, nil
	}
	return id, err
}

It returns 0 when the conflict clause inserts nothing.

The callsite is relative to the module root, so it reads apollo/refresh_person.go:88 whatever worktree built the binary.

Every single-row enqueue goes through Insert. I deleted dozens of SQL files that each spelled out the same INSERT, because each one was a place to copy the next enqueue from. A query that enqueues a set of rows from a SELECT stays a SQL file.

A write and the jobs that follow from it share a transaction. A retry then cannot leave the data changed without its follow-up work:

err := db.Transaction(ctx, func(tx *pgdb.DB) error {
	if err := tx.Exec(ctx, qUpdateCompany, id, domain); err != nil {
		return err
	}
	_, err := jobs.Insert(ctx, tx, "apollo", "apollo.IngestCompany", IngestCompanyArgs{ID: id})
	return err
})

args is part of the conflict key. So an enqueue carries only the keys it sets: a blank field would make a second row, not a duplicate.

Scheduling

A Clock ticks every minute and enqueues each scheduled job whose predicate matches the current UTC time:

func (c *Clock) tick(ctx context.Context, t time.Time) {
	for _, job := range schedule {
		if job.at(t) {
			if err := c.db.Exec(ctx, qInsertScheduledJob, job.queue, job.name); err != nil {
				errs.CaptureException(fmt.Errorf("clock job=%s queue=%s: %w", job.name, job.queue, err))
				log.Printf("clock job=%s queue=%s err: %v", job.name, job.queue, err)
				continue
			}
			log.Printf("clock job=%s queue=%s", job.name, job.queue)
		}
	}
}

The schedule is a list, and each cadence is a predicate:

type scheduledJob struct {
	queue string
	name  string
	at    func(time.Time) bool
}

var schedule = []scheduledJob{
	{queue: "caches", name: "caches.Age", at: func(t time.Time) bool {
		return t.Minute()%15 == 0 // every 15m
	}},
	{queue: "pitchbook", name: "pitchbook.IngestRecentDeals", at: func(t time.Time) bool {
		return t.Minute() == 30 && t.Hour()%6 == 0 // 3x/day
	}},
	// ...
}

insert_scheduled_job.sql also uses ON CONFLICT DO NOTHING, so the Clock does not enqueue a second copy of a job that is still pending.

Maintenance

A scheduled jobs.CleanUpQueues job deletes old rows. The rows it keeps let a handler check for recent work before it calls a paid API.

func CleanUpQueues(ctx context.Context, db *pgdb.DB) (string, error) {
	if err := db.Exec(ctx, qDeleteOldJobs); err != nil {
		return "", errs.Wrap(err, "delete old jobs")
	}
	return "ok", nil
}

Status semantics

A handler returns (status, nil) for an expected outcome and ("", err) for a failure. The poller sends the error to sockeye with errs and stores it as an err: ... status. It skips sockeye for a cancelled context.

A small package builds each status string, so each prefix has one definition:

• fail: The job ran and reached a negative answer: no match, a duplicate, or a rejection that a retry does not change.
• skip: There was nothing to attempt.
• err: A fault. A handler returns an error instead, so the fault reaches sockeye.

One function reads the prefix back, and every count of outcomes uses it. Before, some jobs recorded an expected answer as err:, which inflated the error count. Other jobs recorded a real fault as err: and returned no error, so nobody saw it. An expired API token was one such fault.

A client returns an unhandled HTTP status as a typed error. On a 401 or a 403, the worker pauses its queue for one hour. An expired token then sends one error event an hour, not one for each job. The restart after a token fix ends the pause.

← All articles