go / job queues

updated August 31, 2026

I ported my Ruby job queue to Go. It uses the same Postgres table in a production jobs service:

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.

Modest needs

The service has ~40 queues. Most invoke rate-limited third-party APIs: Apollo, GitHub, Discord, Slack, Anthropic, Pitchbook. I do not need high throughput or parallelism within a queue. Processing one job at a time keeps each queue inside its provider's rate limit. It also prevents a slow job from causing a thundering herd.

The table

The table uses the schema from the Ruby version:

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

The Ruby version used a PollWorker base class. Go has no inheritance, so the shared mechanics live in one Poller struct. Each queue supplies two functions: Dispatch routes a job name to a handler, and an optional Throttle sets the sleep duration 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 poll loop recovers stale jobs on boot. It then fetches and processes pending rows until the context is cancelled:

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

	if err := p.DB.Exec(ctx, qInterruptStartedJobsInQueue, p.Queue); err != nil {
		log.Printf("queue=%s interrupt stale jobs error: %v", p.Queue, err)
	}

	log.Printf("queue=%s poll=%s", p.Queue, interval)

	for {
		select {
		case <-ctx.Done():
			return
		default:
		}

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

		rows, err := p.DB.Query(ctx, qFetchPendingJobs, p.Queue)
		// ... collect rows, then:
		for _, job := range jobs {
			p.WorkOnce(ctx, job)
			if ctx.Err() != nil {
				return
			}
		}
	}
}

WorkOnce claims a job, runs the dispatch within a per-job deadline, and finalizes the job in a deferred block that also catches panics. It writes the status string, latency, and duration to one greppable log line:

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)
			captureSentry(job, fmt.Errorf("panic: %v", r))
		}
		if err := p.DB.Exec(ctx, qFinalizeJob, status, job.ID); err != nil {
			log.Printf("queue=%s job=%s id=%d finalize error: %v", p.Queue, job.Name, job.ID, err)
		}
		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) {
			captureSentry(job, workErr)
		}
		status = "err: " + workErr.Error()
	}
}

WorkOnce is exported so tests can run one job without starting the poll goroutine.

The claim uses a conditional update, so a worker cannot claim a re-interrupted job twice after a crash:

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 worker

Each queue package exports NewWorker, which returns a configured Poller. The Anthropic queue, the caches queue, and the other queues each define their own dispatch switch. 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 is a generic function that decodes the JSONB payload and adds a prefix to any parse error:

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 you can test rate-limit policies without a clock. The Apollo queue targets 150 jobs a minute. It backs off when a status reports an exhausted header:

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 sets Throttle to nil and runs jobs continuously. For example, the caches queue recomputes materialized data:

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

The Ruby version used fork to start one process per worker. The Go version runs one goroutine per poller in a single process. The main function stores the workers as a list of constructors. Adding a new queue requires one line. The composition root configures dependencies without managing control flow:

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 maximum size of the Postgres pool to one connection per poller plus a small buffer. Each poller then runs in its own goroutine:

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

signal.NotifyContext cancels the context on SIGTERM. Each poll loop finishes its active job and returns, which allows clean deploys. If a hard kill leaves a job in the started state, the poller resets it to err: interrupted on the next boot.

Enqueuing

Insert writes one row with ON CONFLICT DO NOTHING and records the file and line number of the caller 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 nil, 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)
	}

	rows, err := db.Query(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 ids
}

The callsite path is relative to the module root. The package computes this path once from the build-time path of insert.go. It outputs apollo/refresh_person.go:88 regardless of which worktree built the binary.

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 data, and each cadence is a predicate over time.Time:

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, which prevents duplicate jobs if a previous job is still pending.

Maintenance

A scheduled jobs.CleanUpQueues job deletes old rows to limit the size of the table and its indexes. Old rows also provide idempotency: a handler can query 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

The return signature defines the entire protocol. A handler returns (status, nil) for an expected terminal outcome. It returns ("", err) for an unexpected failure to send to Sentry. The poller converts the error into an err: ... status. It skips Sentry capture for expected errors, such as a cancelled context or a known long-job timeout.

← All articles