A worker pool is easy to demonstrate and surprisingly hard to operate. Start a few goroutines, feed them jobs, and the happy path looks finished. Production introduces the uncomfortable questions: What happens when a process dies halfway through a job? Can shutdown accidentally start fresh work? Will a retry repeat an irreversible effect? Who owns a job after its lease expires?

This tutorial builds a PostgreSQL-backed Go worker pool that answers those questions explicitly. It uses short reservations, renewable leases, bounded contexts, exponential retries, ownership checks, and a database-enforced idempotency boundary. Its delivery contract is at least once. A job may execute again after uncertainty, but its committed effect is applied once per idempotency key.

Prerequisites and project layout

You need Go 1.22 or newer, PostgreSQL 14 or newer, and a database role allowed to read and modify the application tables. The only Go dependency is github.com/jackc/pgx/v5; this example pins version 5.7.2 and is intended for the compatible 5.7.x line.

reliable-worker/
├── go.mod
├── main.go
├── schema.sql
└── deploy/
    └── reliable-worker.service

Create go.mod:

module example.com/reliable-worker

go 1.22

require github.com/jackc/pgx/v5 v5.7.2

Architecture: separate ownership from execution

Workers claim jobs with one atomic PostgreSQL statement using FOR UPDATE SKIP LOCKED. The statement commits before business processing begins, so no database transaction remains open while a job runs.

A claimed row records a unique worker owner and a lease deadline. A heartbeat extends that deadline. If the process disappears, another worker may reclaim the job after expiration. Every completion, retry, release, and renewal includes the owner in its predicate. This fencing check prevents a worker with an expired lease from modifying a job now owned elsewhere.

The important transitions are:

• ready to running when claimed, incrementing the attempt count.
• running to succeeded when the effect and completion commit together.
• running to ready after a failure, with a delayed available_at.
• An expired running job back to running under a new owner.
• An exhausted job to dead.

Create the durable queue

The idempotency key is unique in both the queue and effect tables. Here the business effect is deliberately simple: recording a processed message. In a real system, the effect table could represent an invoice, notification request, account mutation, or outbox entry.

CREATE TABLE jobs (
    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    idempotency_key text NOT NULL UNIQUE,
    payload         jsonb NOT NULL,
    state           text NOT NULL DEFAULT 'ready'
                    CHECK (state IN ('ready', 'running', 'succeeded', 'dead')),
    attempts        integer NOT NULL DEFAULT 0 CHECK (attempts >= 0),
    max_attempts     integer NOT NULL DEFAULT 5 CHECK (max_attempts > 0),
    available_at    timestamptz NOT NULL DEFAULT now(),
    lease_owner     text,
    lease_until     timestamptz,
    last_error      text,
    created_at      timestamptz NOT NULL DEFAULT now(),
    completed_at    timestamptz
);

CREATE INDEX jobs_claimable_idx
    ON jobs (available_at, id)
    WHERE state = 'ready';

CREATE INDEX jobs_expired_idx
    ON jobs (lease_until)
    WHERE state = 'running';

CREATE TABLE processed_messages (
    idempotency_key text PRIMARY KEY,
    message         text NOT NULL,
    processed_at    timestamptz NOT NULL DEFAULT now()
);

Apply the schema through an existing, explicitly selected database connection:

cd reliable-worker
export DATABASE_URL='postgres://worker_user:[email protected]:5432/workerdb?sslmode=require'
psql "$DATABASE_URL" -v ON_ERROR_STOP=1 -f schema.sql
go mod tidy

Use TLS verification appropriate to your environment. sslmode=require encrypts transport but does not provide the same server identity verification as verify-full with a trusted CA.

Implement the worker pool

The implementation below uses a 30-second lease, renews every 10 seconds, limits claims to two seconds, and limits database statements to three seconds. These bounds are comfortably below the lease. The connection timeout is configured separately because it does not limit query execution.

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"strconv"
	"sync"
	"syscall"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var errLeaseLost = errors.New("lease lost")

type config struct {
	Workers, MaxConns int
	Lease, ClaimTimeout, QueryTimeout, JobTimeout time.Duration
}

type job struct {
	ID int64
	Key string
	Payload []byte
	Attempts, MaxAttempts int
}

type payload struct {
	Message string `json:"message"`
	SleepMS int    `json:"sleep_ms"`
}

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

	cfg := config{
		Workers: envInt("WORKERS", 4),
		Lease: 30 * time.Second,
		ClaimTimeout: 2 * time.Second,
		QueryTimeout: 3 * time.Second,
		JobTimeout: 20 * time.Second,
	}
	cfg.MaxConns = cfg.Workers + 2

	poolCfg, err := pgxpool.ParseConfig(mustEnv("DATABASE_URL"))
	if err != nil {
		panic(err)
	}
	poolCfg.MaxConns = int32(cfg.MaxConns)
	poolCfg.MinConns = 1
	poolCfg.MaxConnLifetime = 30 * time.Minute
	poolCfg.MaxConnIdleTime = 5 * time.Minute
	poolCfg.HealthCheckPeriod = 30 * time.Second
	poolCfg.ConnConfig.ConnectTimeout = 3 * time.Second
	poolCfg.AfterConnect = func(ctx context.Context, c *pgx.Conn) error {
		for _, q := range []string{
			"SET statement_timeout = '3s'",
			"SET lock_timeout = '1s'",
			"SET idle_in_transaction_session_timeout = '5s'",
		} {
			if _, err := c.Exec(ctx, q); err != nil {
				return err
			}
		}
		return nil
	}

	pool, err := pgxpool.NewWithConfig(ctx, poolCfg)
	if err != nil {
		panic(err)
	}
	defer pool.Close()

	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
	processID := randomID()

	var wg sync.WaitGroup
	for i := 0; i < cfg.Workers; i++ {
		wg.Add(1)
		owner := fmt.Sprintf("%s-%d", processID, i)
		go func() {
			defer wg.Done()
			worker(ctx, pool, logger, cfg, owner)
		}()
	}

	wg.Add(1)
	go func() {
		defer wg.Done()
		reaper(ctx, pool, logger, cfg.QueryTimeout)
	}()

	<-ctx.Done()
	logger.Info("shutdown_started")
	wg.Wait()
	logger.Info("shutdown_complete")
}

func worker(ctx context.Context, db *pgxpool.Pool, log *slog.Logger,
	cfg config, owner string) {
	for {
		claimCtx, cancel := context.WithTimeout(ctx, cfg.ClaimTimeout)
		j, err := claim(claimCtx, db, owner, cfg.Lease)
		cancel()

		// Shutdown may have arrived while the blocking claim committed.
		if ctx.Err() != nil {
			if err == nil {
				release(db, j.ID, owner, cfg.QueryTimeout)
			}
			return
		}
		if errors.Is(err, pgx.ErrNoRows) {
			if !wait(ctx, 500*time.Millisecond) {
				return
			}
			continue
		}
		if err != nil {
			log.Error("claim_failed", "owner", owner, "error", err)
			if !wait(ctx, time.Second) {
				return
			}
			continue
		}

		jobCtx, timeoutCancel := context.WithTimeout(ctx, cfg.JobTimeout)
		runCtx, causeCancel := context.WithCancelCause(jobCtx)
		renewDone := make(chan struct{})
		stopRenew := make(chan struct{})
		go renew(runCtx, db, j.ID, owner, cfg, causeCancel,
			stopRenew, renewDone)

		err = handle(runCtx, db, j, owner, cfg.QueryTimeout)
		close(stopRenew)
		<-renewDone
		cause := context.Cause(runCtx)
		causeCancel(nil)
		timeoutCancel()

		switch {
		case err == nil:
			log.Info("job_succeeded", "job_id", j.ID,
				"attempt", j.Attempts)
		case errors.Is(cause, errLeaseLost):
			log.Warn("job_abandoned", "job_id", j.ID,
				"reason", "lease_lost")
		case ctx.Err() != nil:
			release(db, j.ID, owner, cfg.QueryTimeout)
			return
		default:
			if failErr := fail(db, j, owner, err, cfg.QueryTimeout); failErr != nil {
				log.Error("failure_update_failed", "job_id", j.ID,
					"error", failErr)
			}
			log.Warn("job_failed", "job_id", j.ID,
				"attempt", j.Attempts, "error", err)
		}
	}
}

func claim(ctx context.Context, db *pgxpool.Pool, owner string,
	lease time.Duration) (job, error) {
	const q = `
WITH candidate AS (
	SELECT id
	FROM jobs
	WHERE attempts < max_attempts
	  AND (
		(state = 'ready' AND available_at <= now())
		OR (state = 'running' AND lease_until < now())
	  )
	ORDER BY available_at, id
	FOR UPDATE SKIP LOCKED
	LIMIT 1
)
UPDATE jobs AS j
SET state = 'running',
    lease_owner = $1,
    lease_until = now() + ($2 * interval '1 millisecond'),
    attempts = attempts + 1
FROM candidate
WHERE j.id = candidate.id
RETURNING j.id, j.idempotency_key, j.payload,
          j.attempts, j.max_attempts`
	var j job
	err := db.QueryRow(ctx, q, owner, lease.Milliseconds()).Scan(
		&j.ID, &j.Key, &j.Payload, &j.Attempts, &j.MaxAttempts)
	return j, err
}

func renew(ctx context.Context, db *pgxpool.Pool, id int64, owner string,
	cfg config, cancel context.CancelCauseFunc, stop <-chan struct{},
	done chan<- struct{}) {
	defer close(done)
	ticker := time.NewTicker(cfg.Lease / 3)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-stop:
			return
		case <-ticker.C:
			qctx, qcancel := context.WithTimeout(ctx, cfg.QueryTimeout)
			tag, err := db.Exec(qctx, `
UPDATE jobs
SET lease_until = now() + ($3 * interval '1 millisecond')
WHERE id = $1 AND lease_owner = $2 AND state = 'running'`,
				id, owner, cfg.Lease.Milliseconds())
			qcancel()
			if err != nil || tag.RowsAffected() != 1 {
				cancel(errLeaseLost)
				return
			}
		}
	}
}

func handle(ctx context.Context, db *pgxpool.Pool, j job,
	owner string, queryTimeout time.Duration) error {
	var p payload
	if err := json.Unmarshal(j.Payload, &p); err != nil {
		return fmt.Errorf("decode payload: %w", err)
	}
	if p.Message == "" {
		return errors.New("message is required")
	}
	if p.SleepMS < 0 || p.SleepMS > 15000 {
		return errors.New("sleep_ms must be between 0 and 15000")
	}
	if !wait(ctx, time.Duration(p.SleepMS)*time.Millisecond) {
		return context.Cause(ctx)
	}

	qctx, cancel := context.WithTimeout(ctx, queryTimeout)
	defer cancel()
	tx, err := db.Begin(qctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(context.Background())

	_, err = tx.Exec(qctx, `
INSERT INTO processed_messages (idempotency_key, message)
VALUES ($1, $2)
ON CONFLICT (idempotency_key) DO NOTHING`, j.Key, p.Message)
	if err != nil {
		return err
	}

	tag, err := tx.Exec(qctx, `
UPDATE jobs
SET state = 'succeeded', completed_at = now(),
    lease_owner = NULL, lease_until = NULL, last_error = NULL
WHERE id = $1 AND lease_owner = $2 AND state = 'running'`,
		j.ID, owner)
	if err != nil {
		return err
	}
	if tag.RowsAffected() != 1 {
		return errLeaseLost
	}
	return tx.Commit(qctx)
}

func fail(db *pgxpool.Pool, j job, owner string, jobErr error,
	timeout time.Duration) error {
	delay := time.Second << min(j.Attempts-1, 6)
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	_, err := db.Exec(ctx, `
UPDATE jobs
SET state = CASE WHEN attempts >= max_attempts THEN 'dead'
                 ELSE 'ready' END,
    available_at = now() + ($4 * interval '1 millisecond'),
    lease_owner = NULL, lease_until = NULL, last_error = $3
WHERE id = $1 AND lease_owner = $2 AND state = 'running'`,
		j.ID, owner, truncate(jobErr.Error(), 1000), delay.Milliseconds())
	return err
}

func release(db *pgxpool.Pool, id int64, owner string, timeout time.Duration) {
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	_, _ = db.Exec(ctx, `
UPDATE jobs
SET state = 'ready', available_at = now(),
    lease_owner = NULL, lease_until = NULL
WHERE id = $1 AND lease_owner = $2 AND state = 'running'`,
		id, owner)
}

func reaper(ctx context.Context, db *pgxpool.Pool, log *slog.Logger,
	timeout time.Duration) {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	for {
		qctx, cancel := context.WithTimeout(ctx, timeout)
		_, err := db.Exec(qctx, `
UPDATE jobs
SET state = 'dead', lease_owner = NULL, lease_until = NULL,
    last_error = COALESCE(last_error, 'lease expired after final attempt')
WHERE state = 'running' AND lease_until < now()
  AND attempts >= max_attempts`)
		cancel()
		if err != nil && ctx.Err() == nil {
			log.Error("reaper_failed", "error", err)
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func wait(ctx context.Context, d time.Duration) bool {
	timer := time.NewTimer(d)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}

func envInt(name string, fallback int) int {
	if value := os.Getenv(name); value != "" {
		n, err := strconv.Atoi(value)
		if err == nil && n > 0 {
			return n
		}
	}
	return fallback
}

func mustEnv(name string) string {
	value := os.Getenv(name)
	if value == "" {
		panic(name + " is required")
	}
	return value
}

func randomID() string {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n]
}

Why the effect transaction matters

processed_messages and the job completion update commit in one transaction. If the worker no longer owns the lease, the update affects zero rows and the transaction rolls back, including any new effect. If a previous attempt committed the effect but the queue row somehow remained retryable, ON CONFLICT DO NOTHING makes replay harmless.

This pattern only protects effects inside the same database. For an external payment or HTTP API, send the stable idempotency key to a downstream service that supports it, or commit a transactional outbox row and let another idempotent dispatcher perform the network call. A local “processed” flag cannot atomically prove that an unrelated remote side effect occurred.

Test retries, leases, and shutdown

Start with one worker to make transitions easy to inspect:

export DATABASE_URL='postgres://worker_user:[email protected]:5432/workerdb?sslmode=require'
export WORKERS=1
go run .

From another shell, enqueue one successful job and one deterministic failure:

psql "$DATABASE_URL" -v ON_ERROR_STOP=1 <<'SQL'
INSERT INTO jobs (idempotency_key, payload, max_attempts)
VALUES
  ('welcome-1001', '{"message":"welcome","sleep_ms":8000}', 5),
  ('invalid-1002', '{"sleep_ms":0}', 3);
SQL

psql "$DATABASE_URL" -c \
"SELECT id, state, attempts, lease_owner, lease_until, last_error FROM jobs ORDER BY id;"
psql "$DATABASE_URL" -c \
"SELECT idempotency_key, message FROM processed_messages ORDER BY idempotency_key;"

During the eight-second job, send SIGTERM with Ctrl-C. The worker cancels processing and conditionally releases its reservation. Restart it and confirm that the valid job eventually succeeds with one effect row. Its attempt count may exceed one; that is expected under at-least-once delivery. The invalid job retries with bounded exponential delays and becomes dead after its third claim.

To test crash recovery rather than graceful release, terminate the process forcefully in an isolated test environment. The row remains running until its 30-second lease expires, then becomes claimable. Never use a forceful signal as the normal shutdown mechanism.

Observability and performance

The worker emits structured JSON logs containing job IDs, attempts, owners, and errors. Avoid logging complete payloads: they may contain credentials or personal data. Useful database measurements include ready-job count, oldest ready-job age, active leases, expired leases, dead jobs, retry rate, processing duration, and lease-loss count.

Scale workers according to measured database and downstream capacity, not CPU count alone. Each worker can hold one connection during a query, while heartbeats and the reaper need spare capacity; this is why the pool allows workers + 2 connections. Keep job payloads small, archive old succeeded rows, and inspect query plans as the table grows. The partial indexes keep claim scans focused, but they do not replace routine PostgreSQL maintenance.

Security and deployment

Use a dedicated operating-system account and a least-privilege database role. The worker needs no inbound network port, so do not open one in the host firewall. Restrict outbound access to PostgreSQL and genuine downstream dependencies where your firewall architecture supports egress policy. Store the database URL outside the unit file, protect it with root ownership and mode 0600, and rotate credentials without embedding them in the binary.

[Unit]
Description=Reliable Go worker pool
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=reliable-worker
Group=reliable-worker
WorkingDirectory=/opt/reliable-worker
EnvironmentFile=/etc/reliable-worker/worker.env
ExecStart=/opt/reliable-worker/reliable-worker
Restart=on-failure
RestartSec=3s
TimeoutStopSec=15s
KillSignal=SIGTERM
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6

[Install]
WantedBy=multi-user.target

On the host, an administrator can build the binary, place it under /opt/reliable-worker, create the service account and protected environment file, install the unit, then run systemctl daemon-reload and systemctl enable --now reliable-worker. Ensure TimeoutStopSec remains greater than the bounded release query. Containers should use the same unprivileged-user and read-only-filesystem principles and must receive enough termination grace time before forced removal.

Common production failures

• Lease shorter than realistic processing time: renew well before expiration and bound every operation. Long pauses can still lose ownership, so completion must remain fenced.
• Retrying while holding a transaction: claim and commit first. Running business code under a reservation transaction creates locks, connection starvation, and painful recovery.
• Starting work during shutdown: always check cancellation immediately after the blocking claim. If a claim committed during shutdown, release it with an ownership-checked update.
• Assuming connection timeout limits queries: configure connection, statement, lock, and context deadlines independently.
• Calling retries exactly once: leases resolve abandonment, not uncertainty. Idempotency at the effect boundary is what makes repeated delivery safe.

Final verification checklist

• The schema and partial indexes apply without errors.
• Concurrent workers never claim the same live lease.
• A graceful shutdown starts no new business work after a completed claim.
• A crashed worker’s job becomes claimable after lease expiry.
• Heartbeats stop processing when ownership is lost.
• Failed jobs retry with delay and eventually enter dead.
• Repeated delivery produces one committed effect per idempotency key.
• Database and shutdown timeouts remain below the lease budget.
• Logs and queue queries expose backlog, failures, and lease health.

A robust worker pool is not defined by how quickly it consumes a channel. It is defined by what remains true when timing turns hostile: reservations are brief, ownership expires, shutdown is deliberate, retries are bounded, and effects tolerate repetition. Once those invariants are visible in the schema and enforced in every update, failure stops being an exceptional path. It becomes an ordinary state transition the system already knows how to survive.