go / job queues

updated September 02, 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.

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

The schema is the Ruby version's:

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:

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

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 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 is relative to the module root, so it reads apollo/refresh_person.go:88 whatever 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 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 a job still pending is not enqueued twice.

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 Sentry and stores it as an err: ... status. It skips Sentry for a cancelled context.

← All articles