go / job queues

updated July 17, 2026

This is the Go port of my Ruby job queue, running the same Postgres table in an EDS 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 don't need high throughput or parallelism within a queue. Processing one at a time keeps each queue inside its provider's rate limit and makes a slow job impossible to interleave into a thundering herd.

The table

The same schema the Ruby version uses:

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

Ruby had a PollWorker base class. Go has no inheritance, so the shared mechanics live in one Poller struct and each queue fills in two functions: a Dispatch that routes a job name to a handler, and an optional Throttle that decides 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 poll loop recovers stale jobs on boot, then fetches and works 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 inside a per-job deadline, and finalizes in a deferred block that also covers panics. The status string, latency, and duration land on 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 a queue's tests can drive one job without spinning up the poll goroutine.

The claim is a conditional update, so a crashed job that a boot re-interrupts cannot be claimed 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 a NewWorker that returns a configured Poller. The Anthropic queue, the caches queue, and the rest each own their dispatch switch. Dependencies (an API client) are captured in the closure:

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 that decodes the JSONB payload and wraps the parse error with a stable prefix:

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
}

The Throttle is a pure function, so a queue's rate-limit policy is unit-testable without a clock. Apollo aims for 150 jobs a minute and backs off the matching window 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 leaves Throttle nil and runs back to back. 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 forked one process per worker. Go runs one goroutine per poller inside a single process. main holds the workers as a list of constructors, so a new queue adds one line and the composition root wires dependencies, not 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 sizes the Postgres pool cap at one connection per poller plus a small buffer, then each poller runs on 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 in-flight job and returns, so a deploy drains cleanly. Any job that a hard kill leaves started is reset to err: interrupted when its poller boots again.

Enqueuing

Insert writes one row with ON CONFLICT DO NOTHING 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 resolved relative to the module root, computed once from the build-time path of insert.go, so it reads apollo/refresh_person.go:88 regardless of which worktree built the binary.

Scheduling

A Clock ticks every minute and inserts any 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, so 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 uses the same ON CONFLICT DO NOTHING, so a job still pending from the last tick is not duplicated.

Maintenance

A scheduled jobs.CleanUpQueues prunes old rows to keep the table and its indexes bounded. The old rows double as an idempotency guard: a handler can query for recent work before calling 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 carries the whole protocol. A handler returns (status, nil) for an expected terminal outcome that should finalize, and ("", err) for an unexpected failure worth capturing to Sentry. The poller turns the error into an err: ... status and skips the capture for expected cases like a cancelled context or a known long-job timeout.

← All articles