brooq

What it is

A job queue without an external broker: jobs live in PostgreSQL, workers claim them concurrently (FOR UPDATE SKIP LOCKED), and the outcome is written into the same row. One binary, one VPS, observability from day one.

brooq is a library: it compiles into the consumer's process and shares its transaction and connection pool with it. A separate worker process is a consequence, not the essence: cmd/brooq uses the same library.

Three promises of different strength:

• claim — at most once at any moment;
• execution — at least once (the handler must be idempotent);
• outcome write — exactly once; the fate of a job is read with a single SELECT.

Why

Notifications and background actions in my personal project went through an "api → notification-service → transport" push circuit: the job state lived in PostgreSQL while execution lived in process memory, which made delivery poorly traceable and hard to maintain. brooq moves the queue into a separate project: Postgres is the source of truth, workers claim rows themselves, everything is visible through a SQL query and metrics.

Quick start

make bin-deps                 # go mod download + pinned tool versions
make test-db-up               # local PostgreSQL for integration tests (port 55433)
TEST_DATABASE_URL='postgres://brooq:brooq@localhost:55433/brooq_test?sslmode=disable' make test
make build                    # bin/brooq
make lint
make spec-validate            # openspec validate --all --strict

A green make test without TEST_DATABASE_URL does not prove that migrations are applied: integration tests and e2e are skipped in that case.

Running the unit locally: copy .env.example to .env, set DATABASE_URL, then make migrate-up && make run. The service surface is 127.0.0.1:8090 (/healthz, /metrics).

As a library

import (
    "github.com/thefish/brooq/pkg/brooq"    // engine
    "github.com/thefish/brooq/pkg/postgres" // SQL adapter
    "github.com/thefish/brooq/pkg/queue"    // vocabulary: Job, Claim, Handler
)

store := postgres.New(pool)          // the consumer opens the pool
engine := brooq.New(store, log, brooq.Config{Workers: 4, Lease: 30 * time.Second})
_ = engine.RegisterKind(queue.Topic("email"), handler{}, queue.KindOptions{
    MaxAttempts: 5, AttemptTimeout: time.Minute, MaxExecutionTimeout: 5 * time.Minute,
    BackoffBase: time.Second, BackoffMax: time.Minute,
})
engine.Start()
defer engine.Close()

// Enqueue through the pool: the id and a flag saying the row was created by this call.
id, created, err := engine.Enqueue(ctx, queue.Job{Topic: "email", Payload: payload, IdempotencyKey: "order-42"})
if err != nil {
    log.Error("enqueueing job", "err", err)
}
// created == false means a job with this key already existed; id is its identifier.
log.Info("job enqueued", "id", id, "created", created)

// Enqueue in your own transaction: your write and the job commit together.
tx, err := pool.Begin(ctx)                       // the consumer's pgx.Tx
defer func() { _ = tx.Rollback(ctx) }()
if _, _, err := store.EnqueueTx(ctx, postgres.WrapTx(tx), queue.Job{
    Topic: "email", Payload: payload, IdempotencyKey: "order-42",
}); err != nil {
    log.Error("enqueueing job", "err", err)
}
// ... your own writes with the same tx ...
_ = tx.Commit(ctx)

// The reverse direction: the queue transaction → your code.
qtx, err := engine.Begin(ctx)                    // queue.Tx
pgxTx, ok := postgres.AsPgx(qtx)                 // false if this is not *postgres.Tx

postgres.WrapTx wraps the consumer's transaction in the queue.Tx port, postgres.AsPgx does the reverse — it hands the driver transaction out of the port. Both sides are the same transaction, not a copy.

A detailed guide with examples is in queue-engine.md.

Documents

Status

The engine is fully functional: enqueue (with the job identifier and the created flag), claim, outcome write, retry with backoff, Reaper, heartbeat, in-flight cancel, watchdog, and nine brooq_* metrics. Enqueue works both through the pool and inside the consumer's transaction — in both directions (postgres.WrapTx / postgres.AsPgx).

What is not there yet:

• an HTTP API for the queue — cancel exists (Queue.Cancel), there is no endpoint;
• priorities — ordering is strictly FIFO by available_at, id;
• partitioning — at very large volumes, splitting by time instead of deleting rows will be needed.

Terminal jobs (done, dead, canceled) are deleted by TTL — 30 days by default (QUEUE_RETENTION_TTL), in batches, with autovacuum tuned for it. The idempotency key lives as long as the row: the TTL sets the window after which a retry with the same key creates a new job.

Package layout and import rules are in architecture.md:

pkg/queue      vocabulary (stdlib only)
pkg/brooq      engine: workers, Reaper, heartbeat, metrics
pkg/postgres   SQL adapter (the only place where PostgreSQL lives)
cmd/brooq      composition: wire dependencies, start

Schema changes go only through the migrations/ migrations, applied as a deployment step, never from the binary at startup.

Status: the vocabulary was fixed by the first domain modelling session. The words here are used by specs, ADRs and code identifiers (ADR-0005). It is edited only in a domain modelling session.

brooq is a job queue library: it is compiled into the consumer's process and shares its transaction and connection pool. A separate worker process is a consequence, not the essence: it uses the same library.

Goal (the end scenario)

The producer put a job → a worker claimed it → the outcome is recorded and visible. Behind this stand three promises of different strength:

• Claim — "no more than once at any moment": at no moment is a job pinned to two workers.
• Execution — "at least once": the handler may be run again, including without anyone's decision (the lease expired — the job returned to circulation). The handler must be idempotent. "Exactly once" is achievable only for effects inside the same database, not for external ones (email, HTTP) — see ADR-0010.
• Outcome recording — "exactly once": terminal states are not overwritten, a job's fate is read with a single SQL query.

Everything else in the project exists so that this scenario is reliable, observable and maintainable: a job is not lost when a worker crashes, is not executed twice at the same moment, and its fate is read with a single SQL query and metrics.

Vocabulary

Spelling we avoid: cancelled (two l) — we write canceled, as in context.Canceled. state — not status (a state has no separate "status"; it is the same word). kind — not a synonym for topic. payload — not data and not body.

States and events

States are lowercase, events are past-tense PascalCase (ADR-0005). The meaning of the states:

Events: JobClaimed, JobDone, JobFailed, JobDead, JobCanceled, JobRequeued (returned to circulation by the reaper).

failed is not "failed forever" but "waiting for the next attempt": a job is claimed both from pending and from failed, so this is a waiting state, not a deferred verdict.

Invariants

1. A job in processing always has a lease: the worker identifier and the lease deadline are filled in at the same time.
2. A job cannot be pinned to two workers at one moment: the capture is a single SQL statement with FOR UPDATE SKIP LOCKED (ADR-0001, ADR-0010). This is a promise about pinning, not about execution.
3. Terminal states — done, dead, canceled — are not overwritten.
4. A job's outcome is written into the same row as the job itself: there is no separate "delivery log".
5. The queue does not know the consumer's domain: the job payload is opaque JSON.
6. An attempt is charged at claim time: attempts is how many times the job has already been claimed.
7. dead is reached only through a recorded failure: a successful execution writes done unconditionally. A lease expiry does not count as an attempt and does not bring a job closer to dead — it is a sign of a process crash, not of an execution failure.
8. Idempotency is on the pair "topic + key": at most one job per pair for the whole lifetime of the row, including terminal states. The key describes a specific delivery attempt, so a re-enqueue with a different key is a new job. Jobs without a key are not deduplicated.
9. The queue does not own the process: it does not start goroutines and does not open connections without an explicit call. Everything it does, it does inside the caller's transaction or on the caller's command.
10. The execution order is by readiness time, and on a tie by enqueue order. Between processes the order is not guaranteed.
11. The guarantees are of different strength and are not mixed: claim — no more than once at any moment; execution — at least once; outcome recording — exactly once. Repeated execution is normal, not an emergency: it happens without anyone's decision when the lease has expired or the outcome recording did not succeed.

How the queue is built and what you can rely on without reading the code. Examples, configuration and operational practices are in queue-engine.md.

Overview

brooq is a job queue that lives entirely in PostgreSQL. Jobs sit in the job table in the same database as the consumer's data; workers claim rows concurrently; the outcome of an attempt is written into the same row that holds the job itself. There is no external broker, no second state store, no separate service for observation.

It ships as a library plus a binary. The library is compiled into the consumer's process and shares its transaction and connection pool: that is the main thing it offers beyond "put some JSON in a table". The cmd/brooq binary is built from the same packages and runs a worker as a separate process.

The end scenario: the producer put a job → a worker claimed it → the outcome is recorded and visible.

Jobs and states

A job is a row of the job table: topic, payload, state, attempt counter, availability time, lease. The payload is opaque JSONB: only the consumer knows its schema, the queue never looks inside.

Six states:

failed is not a verdict but a wait: a job is claimed from pending and from failed alike. There are exactly three terminal states — done, dead, canceled; they are never overwritten.

Cancel comes from two places: the handler's signal that "the job is no longer relevant" and an external cancel by an operator, including a job in flight. A job in flight is stopped through loss of the lease, without LISTEN/NOTIFY: the worker renews the lease every Lease/3 with a query conditioned on "the row is still processing and I am the one holding it"; after the cancel the condition does not hold, the update returns 0 rows, and the worker shuts the handler down without writing anything into the row — the outcome has already been recorded by the canceler. Reaction latency is up to Lease/3.

Guarantees

Three promises of different strength, and they must not be mixed:

• claim — at most once at any moment: a job is never pinned to two workers simultaneously;
• execution — at least once: the handler may be run again, including without anyone's decision (the lease expired, the process crashed, the outcome write failed);
• outcome write — exactly once: the fate of a job is read with a single SELECT.

Hence the main requirement on the consumer: the handler must be idempotent. "Exactly once" is achievable only for effects inside the same database; for external effects — an email, an HTTP call, a payment — only "at least once". Two windows of repeated execution are unavoidable by design:

• the effect was made but the done write failed — the job comes back into circulation and the effect repeats;
• the lease expired while the handler was still running — a second worker claimed the job, two handlers run simultaneously.

That is why the guarantee is stated per place of effect:

Storage

Jobs live in the job table in the consumer's database. Workers claim rows with a single SELECT … FOR UPDATE SKIP LOCKED, so a taken row does not block the others: the second worker claims the next job instead of waiting.

A job's state exists only in that row. There are no second state stores — no Redis, no in-memory maps, no files. It follows that a worker remembers nothing between runs: a crash at any moment means "the lease will expire and the job will come back into circulation".

A cache is acceptable only as an accelerator and never as the source of an answer: if the cache and the database disagree, the truth is in the database.

Why not a broker: the fate of a job is read with ordinary SQL rather than a dedicated UI; one backup and one set of migrations instead of two; "write the order and enqueue the job" is one transaction instead of a dual write with an outbox table. A volume of tens of thousands of jobs per day is far from the limits where a Postgres queue starts to get in the way.

Schema and migrations

The schema is described only by migrations/*.sql files (goose). There is no second truth: neither CREATE TABLE IF NOT EXISTS in the code at startup, nor a schema assembled by hand in production.

An applied file is never edited — a mistake is cured by a new migration. Migrations are applied by a deployment step and never by the binary at startup: otherwise a restart of several instances turns into a DDL race. Tests apply the same directory through goose, so the divergence "the schema in tests is one thing, in production another" does not happen. A rollback (+goose Down) is mandatory and is verified on a clean database.

Retention of terminal jobs

Terminal jobs (done, dead, canceled) are deleted by TTL: by default 30 days after the transition to a terminal state. Without cleanup the table and its indexes grow without bound, and the idempotency key is never released.

The age is computed from updated_at: it is set by every outcome write operator, so "older than the TTL" and "terminal for longer than the TTL" are the same thing. Deletion goes in batches rather than in one query: a large DELETE holds locks and produces a spike of dead rows, whereas a batch is committed in its own transaction and repeated while it comes back full. The subquery carries FOR UPDATE SKIP LOCKED — several processes or replicas remove different rows instead of waiting for each other.

Active states (pending, processing, failed) are never deleted, even if their updated_at is very old: only terminal ones are in the selection. A zero TTL means "cleanup disabled" and never "delete everything".

For cleanup there is a partial index on updated_at for the three terminal states: it does not grow with the work queue and does not get in the way of the hot path. For the table itself the autovacuum threshold is lowered: at the default 20 % dead rows a queue with a constant flow of inserts and deletes gets rare and very long passes, while frequent short ones are cheaper.

Consequence for idempotency: the TTL sets the deduplication window. While the row is alive, a repeat with the same key does not create a job; after cleanup it does. The key describes a delivery attempt, not an event in the consumer's domain, so this is expected behaviour rather than a defect.

Partitioning by time is the next step if the volume becomes large: then old partitions are dropped wholesale, without DELETE.

Packages and dependencies

cmd/brooq      composition: assemble dependencies, start up
pkg/brooq      engine: workers, Reaper, heartbeat, metrics, backoff
pkg/queue      vocabulary: Job, State, Claim, Lease, Tx, Handler
pkg/postgres   SQL adapter: the only place where PostgreSQL lives
                (and the transaction bridge WrapTx / Pgx / AsPgx)
pkg/{config,db,logger}   reusable packages with no domain knowledge
internal/{archgate,health,loopback,observability}   infrastructure of the unit

Imports go downwards only, and this is checked by a test against the actual dependencies rather than by agreement:

• pkg/* does not import internal/*;
• pkg/queue knows neither SQL nor transport: stdlib only. The producer's transaction is passed as the structural interface queue.Tx, which the driver's transaction satisfies without importing the driver;
• pkg/brooq does not know PostgreSQL: it works through its own Repo port, so it is testable without a database and the storage is replaced by a fake;
• SQL lives only in pkg/postgres.

The engine and the adapter are public because the internal/ directory cannot be imported from another module: with the packages in internal/ this would not be a library. Only the unit's own infrastructure stays in internal/.

Transactions

Every write goes through a transaction: "claim the job and record the attempt", "confirm and schedule a retry" are atomic operations. Reads — lists, statistics, service queries — go straight through the pool: a transaction for the sake of one SELECT holds a snapshot and gets in the way of vacuum.

A write through the bare pool is a defect. If a method does not accept a transaction, that is visible in review and must be justified.

Transaction bridge

The queue.Tx port abstracts the transaction, so at the junction of the two worlds both sides are needed: the consumer with its pgx layer and the queue with its port. The bridge lives in the adapter, because only the adapter is allowed to know the driver:

func WrapTx(tx pgx.Tx) *Tx             // pgx.Tx → queue.Tx (the same object)
func (t *Tx) Pgx() pgx.Tx              // *Tx → pgx.Tx
func AsPgx(tx queue.Tx) (pgx.Tx, bool) // queue.Tx → pgx.Tx, safely

WrapTx does not open a new transaction and does not copy it: enqueueing through the wrapper goes into the consumer's very transaction, so a rollback leaves no job behind, and a commit makes it visible to workers. Returning the concrete *Tx rather than the port — so that the reverse direction stays available without a type assertion: *Tx already satisfies queue.Tx.

AsPgx is needed where the port came from the engine — from Queue.Begin or from Claim.OnSucceeded: on a port value the Pgx method cannot be called. The second result is false for a foreign implementation of the port, nil and an empty wrapper: the adapter does not take it upon itself to unwrap foreign transactions.

Whoever opened the transaction closes it. The wrapper merely translates it between representations: Commit or Rollback through either side closes it for everyone who shares it.

Service contours

The service surface (/healthz, /metrics) and any internal endpoints listen on 127.0.0.1 only. Internal endpoints additionally require a service bearer token, compared in constant time. A public surface appears only through nginx and only deliberately. /healthz is deliberately not behind authentication: otherwise monitoring goes blind exactly when it is needed.

Secrets

Secrets — a DSN with a password, provider keys, bearer tokens — do not live in the repository, in the code, in migrations or in the database in plain text. Locally that is .env (gitignored) or environment variables; on the server — EnvironmentFile for non-secret settings and LoadCredential= for secrets, so that the value does not end up in /proc/<pid>/environ or in systemctl show. An empty variable value counts as unset: KEY= must not turn a default into an empty string.

Language and naming

Comments, doc comments, commit messages, error texts and logs are in Russian. Identifiers are English; a Russian term in the code is acceptable only as a data value, not as a name. User-facing texts are not written inline: they come from a message catalogue by a symbolic key.

States and other enumeration values are lowercase (pending). Events are past-tense PascalCase (JobClaimed). Job topics are lowercase with a hyphen or a dot (email.send). Tables are singular (job), columns are snake_case (available_at), Go fields are PascalCase, JSON tags are snake_case.

A state is where a thing stands, an event is a fact. When they are written the same way, the query "what happened" and the query "where is it now" start to blur together. The engine currently produces no events: a cancel can be observed through state and metrics.

Why a custom engine

Ready-made queue libraries on the same principle would have to be taken together with their schema and their migrator. For a project where the schema is described only by goose migrations and is applied by a single deployment step, a second migration stream breaks the single process of deployment and testing on a clean database.

The second constraint is the lease model. Some ready-made implementations have no explicit "claimed" state: the lease is encoded in a deferred availability time, and the question "what is in flight right now and who holds it" is inexpressible, while cancel creates a race — the worker's success overwrites the cancel.

Instead of a dependency, techniques were taken rather than code: a backoff formula with overflow protection, catching a panic in the handler, a watchdog for a hung handler, limiting the fetch by the number of free workers, enqueueing the next job in the transaction that closes the current one. Renewing the lease at short intervals with an ownership check absorbs the need for LISTEN/NOTIFY.

Operations and limits

One host, one systemd unit, PostgreSQL alongside. The binary is static, configuration comes from .env, secrets through LoadCredential, migrations are a deployment step. The service surface listens on loopback. Such a deployment is maintained by one person: a rollback is restoring the previous binary, diagnostics are systemctl status and journalctl. Scaling beyond the limits of a host requires no changes in the queue: it is enough to run workers on several hosts.

What is worth knowing before production:

• The TTL sets the idempotency window. Terminal rows are deleted after QUEUE_RETENTION_TTL (30 days by default); after deletion the key is released, and a repeat with the same key will create a new job. If repeats must be filtered out for longer, increase the TTL or keep the window in the consumer. Cleanup can be disabled entirely: QUEUE_RETENTION_TTL=0; nothing is released, the table grows. Retention and partitioning are separate work that will be needed as volume grows;
• a handler that ignores the context cannot be killed. Lease renewal stops after a cancel, so the reaper will return the job into circulation after Lease, while the hung goroutine remains a memory leak until the process restarts. The leak is visible in brooq_hung_workers_total;
• there is no absolute ceiling without a hang. A handler that kills the process writes no failure, and dead is not reached: the job will be claimed again. An external detector is needed — growth of attempts with an empty last_error — plus a manual transition of the row to dead;
• there is no HTTP API for the queue. Cancel is available from the consumer's code and from psql; there are no priorities, the order is strictly FIFO by availability time.

A practical guide to the engine. How it is built, guarantees and limitations — in architecture.md.

1. Quick start

Five steps from an empty database to a claimed job.

Step 1. Apply the migrations

make test-db-up                    # local PostgreSQL for experiments
export DATABASE_URL='postgres://brooq:brooq@localhost:55433/brooq_test?sslmode=disable'
make migrate-up
make migrate-status

Expected output: 00001_job and 00002_job_idempotency applied. A repeated make migrate-up is a no-op. Migrations are applied as a deployment step, never from the binary at startup.

Step 2. Set the settings

Everything through the environment or .env; the defaults already work:

QUEUE_WORKERS=2
QUEUE_POLL_INTERVAL=1s
QUEUE_LEASE=2m
QUEUE_ATTEMPT_TIMEOUT=1m
QUEUE_MAX_EXECUTION_TIMEOUT=5m
QUEUE_BACKOFF_BASE=1s
QUEUE_BACKOFF_MAX=1m
QUEUE_MAX_ATTEMPTS=5

Invalid values (zero, negative, AttemptTimeout > Lease) are rejected at startup with all the problems listed at once.

Step 3. Write a handler

package main

import (
    "context"
    "encoding/json"
    "log/slog"

    "github.com/thefish/brooq/pkg/queue"
)

// notification — the job payload; only the consumer knows the schema.
type notification struct {
    UserID string `json:"user_id"`
    Text   string `json:"text"`
}

type notifyHandler struct {
    log *slog.Logger
}

// Handle executes one job. The handler must be idempotent:
// execution may repeat (the lease expired, the outcome write did not go through).
func (h notifyHandler) Handle(ctx context.Context, job queue.Claim) error {
    var msg notification
    if err := json.Unmarshal(job.Payload, &msg); err != nil {
        // Permanent error: a retry will not help, but the job cannot be
        // canceled either — let the attempts run out and the job become dead.
        return err
    }

    if err := send(ctx, msg); err != nil {
        // Temporary error of an external API: retry later without spending
        // an attempt (the attempt was already charged at claim — give it back).
        return queue.ErrMustPreserveAttempt
    }

    h.log.Info("уведомление отправлено", "job", job.ID, "attempt", job.Attempt)
    return nil
}

Step 4. Assemble the engine

// The store is the public SQL adapter; the consumer opens the pool
// (pkg/db is a convenient helper, but any *pgxpool.Pool will do).
store := postgres.New(pool)

// The engine is assembled from the store, the logger and the settings.
engine := brooq.New(store, log, brooq.Config{
    Workers:      2,
    PollInterval: time.Second,
    Lease:        2 * time.Minute,
})

// Topic registration — before Start: settings errors are visible at startup.
if err := engine.RegisterKind(queue.Topic("notification"), notifyHandler{log}, queue.KindOptions{
    MaxAttempts:         5,
    AttemptTimeout:      time.Minute,
    MaxExecutionTimeout: 5 * time.Minute,
    BackoffBase:         time.Second,
    BackoffMax:          time.Minute,
}); err != nil {
    return err
}

engine.Start()
defer engine.Close()

Step 5. Put a job and see the outcome

// A job is enqueued through the facade — the SQL adapter needs no direct access.
// id is known immediately, created == false — a job with this key already existed.
id, created, err := engine.Enqueue(ctx, queue.Job{
    Topic:   queue.Topic("notification"),
    Payload: []byte(`{"user_id":"u-42","text":"привет"}`),
    // The idempotency key is optional: with it a repeated enqueue
    // will not create a second job.
    IdempotencyKey: "notify-u-42-2026-10-20",
})
if err != nil {
    return err
}
log.Info("задание поставлено", "id", id, "created", created)

A worker will claim the job and record the outcome in the same row; attempts will become 1, state — done.

Consumer imports — public packages only:

import (
    "github.com/thefish/brooq/pkg/brooq"    // engine: New, Config, Queue
    "github.com/thefish/brooq/pkg/postgres" // SQL adapter for PostgreSQL
    "github.com/thefish/brooq/pkg/queue"    // vocabulary: Job, Claim, Handler, signals
)

Not a single internal/* is needed: that is the point of the public facade.

A complete ready-made example is cmd/brooq/main.go (it has the demo topic and a demonstration handler).

Checking it live:

make build
DATABASE_URL='postgres://brooq:brooq@localhost:55433/brooq_test?sslmode=disable' ./bin/brooq &
sleep 2
psql "$DATABASE_URL" -c "INSERT INTO job (id, topic, payload, idempotency_key)
  VALUES (gen_random_uuid(), 'demo', '{\"msg\":\"привет\"}'::jsonb, 'demo-1')"
sleep 2
psql "$DATABASE_URL" -c "SELECT state, attempts FROM job WHERE idempotency_key='demo-1'"
#  state | attempts
# -------+----------
#  done  |        1

2. Public surface

Three public packages — and the consumer needs nothing more:

The direction of imports is downwards only: pkg/brooq does not know PostgreSQL (it works through the Repo port), pkg/queue knows neither. That is why the engine is tested without a database, and the store is swappable.

The engine facade:

type Queue struct{ /* ... */ }

func New(repo Repo, log *slog.Logger, cfg Config) *Queue
func (q *Queue) RegisterKind(topic queue.Topic, h queue.Handler, opts queue.KindOptions) error

// Enqueue — no need to touch the SQL adapter directly.
func (q *Queue) Enqueue(ctx context.Context, job queue.Job) (id string, created bool, err error)
func (q *Queue) EnqueueTx(ctx context.Context, tx queue.Tx, job queue.Job) (id string, created bool, err error)
func (q *Queue) Begin(ctx context.Context) (queue.Tx, error)

// Administrative cancel (D9, D14).
func (q *Queue) Cancel(ctx context.Context, id string) error

func (q *Queue) Start()
func (q *Queue) Stop()
func (q *Queue) Close() error

The vocabulary (pkg/queue) — below.

// A job to enqueue.
type Job struct {
    Topic          Topic
    Payload        Payload         // []byte, opaque JSON
    IdempotencyKey string          // "" = no key
    AvailableAt    time.Time       // zero = now
}

// A claimed job as the handler sees it.
type Claim struct {
    ID      string
    Topic   Topic
    Payload Payload
    Attempt int                    // the number of the completed attempt, from 1
    Lease   Lease                  // who holds it and until when
    // OnSucceeded — a callback for enqueuing the next job in the same
    // transaction that closes the current one. The engine fills it in before
    // calling the handler, so in Handle it is never nil; tx is the
    // queue.Tx port, the driver is obtained from it with postgres.AsPgx.
    OnSucceeded func(ctx context.Context, tx Tx) error
}

type Handler interface {
    Handle(ctx context.Context, job Claim) error
}

// Topic settings: set in code, not as a table column.
type KindOptions struct {
    MaxAttempts         int
    AttemptTimeout      time.Duration
    MaxExecutionTimeout time.Duration
    BackoffBase         time.Duration
    BackoffMax          time.Duration
}

3. States

                   ┌──────────── done  (success, unconditionally)
                   │
pending ──claim───▶ processing ──┬──────── dead  (failure on the last attempt)
   ▲                   │         │
   │                   │         └──────── canceled (cancel signal / external cancel)
   │                   │
   └── Reaper ◀────────┤ lease expired  → pending (§ 10)
                       │
                       └── failed ──(available_at)──▶ claimed again

The key point: failed is a waiting queue, not a verdict. A job is claimed from pending and failed once available_at has come.

Terminal states are not overwritten: all outcome writes are conditional UPDATEs with the guard WHERE state = 'processing' AND locked_by = $me, and "zero rows affected" is a normal result, not an error.

4. Retry policy

The attempt is charged at claim

attempts is incremented in the same UPDATE that pins the lease. So Claim.Attempt is already the number of the completed attempt, starting at 1.

Backoff

d = min(BackoffBase · 2^(attempts-1), BackoffMax)
actual backoff = uniformly from [d/2, d]

The jitter spreads out workers that would otherwise hit an external API simultaneously. The overflow guard sits before the exponentiation: for a large attempt number the result equals the ceiling, without Inf.

Example with BackoffBase=1s, BackoffMax=1m:

Three handler signals

import "github.com/thefish/brooq/pkg/queue"

// 1. "Retry later, in this much time" — does not spend an attempt beyond
//    the one already charged and does not decrement the counter. The backoff
//    is clamped to BackoffMax. On the last attempt the job becomes dead anyway.
return queue.MustDelay(30 * time.Second)

// 2. "The failure must not spend an attempt" — the job returns to failed,
//    attempts is decremented back (but not below zero), dead is unreachable.
return queue.ErrMustPreserveAttempt

// 3. "The job has become irrelevant" — canceled regardless of the remaining
//    attempts, processing stops.
return queue.ErrMustCancelTask

Errors are wrappable — fmt.Errorf("отправка: %w", queue.ErrMustCancelTask) works the same. To check the backoff in your own code:

if d, ok := queue.DelayOf(err); ok {
    log.Info("обработчик попросил отсрочку", "delay", d)
}

An ordinary error

Returning any other error is a failed attempt: failed + backoff, or dead if attempts are exhausted. The reason is written to last_error (no secrets — that is your responsibility).

Panic

A handler panic is recovered: the job gets the failure outcome, the worker keeps running, the process does not exit. The stack is written to the log at ERROR level.

5. Reasons for context interruption

The engine distinguishes a timeout from a cancel via context.WithCancelCause (D15). The handler does not need to parse the cause — returning ctx.Err() is enough.

func (h handler) Handle(ctx context.Context, job queue.Claim) error {
    if err := h.doWork(ctx); err != nil {
        if ctx.Err() != nil {
            // Cancel or timeout: the engine determines the outcome from context.Cause.
            return ctx.Err()
        }
        return err
    }
    return nil
}

ErrMaxExecutionExceeded does not make the job dead: dead is reached only by exhausting attempts.

6. Producer

An ordinary enqueue

id, created, err := repo.Enqueue(ctx, queue.Job{
    Topic:   queue.Topic("notification"),
    Payload: payload,
})
if err != nil {
    return err
}
log.Info("задание поставлено", "id", id, "created", created)

In the producer's transaction (the main point of a "library")

The job is committed together with your changes — atomically:

tx, err := repo.Begin(ctx)
if err != nil {
    return err
}
defer tx.Rollback(ctx) // safe: a no-op after Commit

// 1. Change our own state.
if _, err := tx.Exec(ctx, `UPDATE account SET notified = true WHERE id = $1`, id); err != nil {
    return err
}

// 2. Put the job in the same transaction.
// The identifier is known before the commit, but is valid only after Commit.
jobID, created, err := repo.EnqueueTx(ctx, tx, queue.Job{
    Topic:          queue.Topic("notification"),
    Payload:        payload,
    IdempotencyKey: "welcome-" + id,
})
if err != nil {
    return err
}

// 3. Always write the meta, idempotently by job_id: created == false may
// refer to an already finished job — then the meta is simply appended.
if _, err := tx.Exec(ctx,
    `INSERT INTO outbox (order_id, job_id) VALUES ($1, $2) ON CONFLICT (job_id) DO NOTHING`,
    id, jobID); err != nil {
    return err
}
log.Info("задание поставлено в транзакции", "id", jobID, "created", created)

// 4. Commit: either both changes are visible or neither.
return tx.Commit(ctx)

If the transaction rolls back, there is no job. If it commits, the worker will see it.

Both sides of one transaction

The boundary between consumer code and the queue is crossed in both directions. postgres.WrapTx wraps the driver transaction into the queue.Tx port, and (*Tx).Pgx and postgres.AsPgx return the driver transaction back.

// own transaction → queue
tx, err := pool.Begin(ctx)                       // the consumer's pgx.Tx
defer func() { _ = tx.Rollback(ctx) }()
if _, _, err := repo.EnqueueTx(ctx, postgres.WrapTx(tx), job); err != nil { ... }
// ... own writes with the same tx ...
err = tx.Commit(ctx)

// queue transaction → own code
qtx, err := engine.Begin(ctx)                    // queue.Tx
pgxTx, ok := postgres.AsPgx(qtx)                 // false if it is not *postgres.Tx

WrapTx returns a concrete *postgres.Tx — which is also a queue.Tx, so it fits everywhere the engine expects the port. Pgx and AsPgx give the very same transaction, not a copy: a Commit or Rollback through it closes it for everyone who shares it. Its owner closes it — the one who opened it. The same technique works in Claim.OnSucceeded(ctx, tx): the engine hands out a queue.Tx, and AsPgx turns it into the driver transaction for your own writes in the same transaction.

Idempotency

// A repeat with the same key → no second job (ON CONFLICT DO NOTHING).
id, created, err := repo.Enqueue(ctx, queue.Job{Topic: "notification", Payload: p, IdempotencyKey: "order-42-paid"})
// One row in the table; the first call: created == true, id — the new row.
id2, created2, err := repo.Enqueue(ctx, queue.Job{Topic: "notification", Payload: p, IdempotencyKey: "order-42-paid"})
// The second call: the same id (id2 == id) and created2 == false — no job appeared.

The created flag answers "is this a new row", not "is this a new effect": created == false may refer to an already finished job (the producer enqueued the job and crashed before writing the meta, the worker brought it to done, the producer retried). That is why the consumer's meta is written always, made idempotent by job_id — in the same transaction as EnqueueTx:

INSERT INTO outbox (order_id, job_id) VALUES ($1, $2)
ON CONFLICT (job_id) DO NOTHING

Then a repeat with created == false appends the missing meta and does not duplicate the existing one. The reverse order ("meta first, then enqueue") does not work: id is not known yet.

Rules:

• the key is unique within a topic: the same key in different topics means two different jobs;
• the empty string "" means "no key" (it becomes NULL), and such jobs are not deduplicated;
• a whitespace-only key " " is a real key;
• the key is not released after a terminal state: a repeat in a month with the same key will not create a job. The key describes a specific delivery attempt — a new delivery needs a new key.

7. Canceling a job

// Cancels both a waiting and a running job.
err := engine.Cancel(ctx, jobID)

From psql — the same thing:

UPDATE job SET state = 'canceled', updated_at = now()
WHERE id = '...' AND state IN ('pending', 'failed', 'processing');

How a cancel reaches a running handler

The mechanism is lease loss, without LISTEN/NOTIFY (D14):

1. the canceling side moves the row to canceled;
2. the worker's next heartbeat does not pass the state='processing' guard and returns 0 rows;
3. the worker calls cancelCause(ErrLeaseLost), the handler gets ctx.Done(), the worker writes nothing to the row — the outcome has already been recorded by the canceling side.

The reaction delay is no longer than Lease/3 (the renewal period).

cancel ──▶ row canceled ──▶ heartbeat: 0 rows ──▶ ErrLeaseLost ──▶ ctx.Done()
                                      (≤ Lease/3)                        │
                                                                         ▼
                                                     the worker does NOT write the outcome

A handler that respects ctx will finish on its own. A handler that ignores ctx keeps running (a goroutine in Go cannot be killed) — then the watchdog fires: ERROR in the log and brooq_hung_workers_total. Its outcome write gets 0 rows and does not overwrite the cancel.

8. Settings

Startup validation rejects:

• QUEUE_WORKERS < 1, QUEUE_MAX_ATTEMPTS < 1, QUEUE_RETENTION_BATCH < 1;
• any period <= 0 — except QUEUE_RETENTION_PERIOD, where 0 means "computed";
• a negative QUEUE_RETENTION_TTL; zero is allowed and means "cleanup disabled";
• AttemptTimeout > MaxExecutionTimeout;
• any period <= 0;
• AttemptTimeout > MaxExecutionTimeout;
• AttemptTimeout > Lease — the restriction is lifted: the lease is renewed by the heartbeat, so long work no longer loses the lease. Consequence for operations: the shorter the Lease, the faster the Reaper returns a crashed worker's job, but the more frequent the renewals.

Bare numbers without units (QUEUE_LEASE=120) are rejected: time.Duration requires a suffix (120s).

How to choose Lease

Lease no longer limits the duration of work: while the handler is alive, the worker renews the lease every Lease/3. So Lease is the recovery speed:

• a short Lease (10–30s) — a crashed worker's job comes back quickly, but renewals happen more often (more database queries);
• a long Lease (2–5m) — less load, but a crash costs up to Lease of job downtime.

WatchdogGrace — how long to wait for the handler to exit after a cancel before recording a hang. By default it equals Lease.

9. Schema and observability

Columns

Indexes

Useful queries

-- A job's fate in one query — the main promise of the queue.
SELECT state, attempts, last_error, available_at, locked_by, locked_until
FROM job WHERE id = '...';

-- What is in flight right now and who holds it.
SELECT id, topic, locked_by, locked_until, attempts,
       locked_until < now() AS lease_expired
FROM job WHERE state = 'processing' ORDER BY locked_until;

-- Queue depth by topic.
SELECT topic, state, count(*) FROM job GROUP BY topic, state ORDER BY 1, 2;

-- Age of the oldest ready job (queue lag).
SELECT topic, now() - min(available_at) AS oldest_ready
FROM job WHERE state IN ('pending','failed') AND available_at <= now()
GROUP BY topic;

-- Dead jobs: to be sorted out by hand.
SELECT id, topic, attempts, last_error, updated_at
FROM job WHERE state = 'dead' ORDER BY updated_at DESC;

Reaper: what happens to stuck jobs

The Reaper runs inside every process: one goroutine, period max(5s, Lease/4), overridden by QUEUE_REAPER_PERIOD. It does exactly what D8 describes:

UPDATE job
SET state = 'pending', locked_by = NULL, locked_until = NULL,
    available_at = now(), updated_at = now()
WHERE state = 'processing' AND locked_until < now()
RETURNING id, topic;

Why exactly this way:

• attempts does not change: an expired lease is a sign of a crashed process, not of a recorded failure (invariant 7). So a crash does not bring the job closer to dead;
• available_at = now(), without an extra backoff: the lease itself was already the wait — the job could not come back earlier than its term. A second delay would be a double one;
• the state = 'processing' guard makes the return idempotent and does not allow resurrecting terminal rows (they also keep a locked_until from their past lease);
• several replicas are safe: no mutex table is needed, a second concurrent reaper will see the row already in pending and will not return it again.

To see what exactly came back: an Info log entry with count and topics. How many jobs are executing right now and who holds them — the "what is in flight" query above.

If you still need to return a job by hand (for example, the reaper is disabled):

-- WARNING: safe only if the lease has really expired.
-- attempts does not change — an expired lease is not counted as an attempt.
UPDATE job
SET state = 'pending', locked_by = NULL, locked_until = NULL, updated_at = now()
WHERE state = 'processing' AND locked_until < now();

Do not run this in a loop with a short period: a job whose lease has just expired may be running in a live but slow worker. Its outcome write after the return to circulation will get 0 rows (the guard on locked_by), that is, the work will not be counted — the job will be executed again. That is exactly the window the heartbeat exists to eliminate (4.6).

Cleaning up terminal rows (retention)

Terminal jobs (done, dead, canceled) are deleted by TTL: by default 30 days after moving to a terminal state. Why: without cleanup the table and its indexes grow without bound, and the idempotency key is never released.

DELETE FROM job
WHERE id IN (
    SELECT id
    FROM job
    WHERE state IN ('done', 'dead', 'canceled')
      AND updated_at < now() - $1::interval   -- TTL
    ORDER BY updated_at
    LIMIT $2                                  -- batch size
    FOR UPDATE SKIP LOCKED
)
RETURNING id, topic;

Why this way:

• updated_at as the age. Every outcome statement sets it, so "older than the TTL" and "terminal for longer than the TTL" are the same thing;
• in batches, not all at once. One big DELETE holds locks and produces a spike of dead rows. The goroutine deletes up to QUEUE_RETENTION_BATCH rows in one transaction and repeats while a batch comes back full;
• FOR UPDATE SKIP LOCKED. Several processes or replicas clean different rows instead of waiting for each other;
• ORDER BY updated_at. Oldest first, so that a lagging cleanup does not accumulate a tail;
• active rows are untouched — the selection contains only terminal states, and pending/processing/failed are never deleted, even with a very old updated_at.

Disabling. QUEUE_RETENTION_TTL=0 disables cleanup entirely: the goroutine does not start. This is protection against an accidental DELETE of the terminal part of the table — a zero TTL never means "delete everything".

The loop is bounded to a hundred batches per tick. If the accumulation is larger, the remainder goes to the next tick, and a Warn appears in the log — better than one endless pass that holds the load unnoticed.

Autovacuum. Migration 00003 lowers the threshold for the job table (autovacuum_vacuum_scale_factor = 0.02, threshold = 100): with the default 20 % dead rows, a queue with a constant stream of inserts and deletes gets rare and very long passes. Frequent short passes are cheaper.

How much was deleted: the brooq_retention_deleted_total{topic} metric and an Info log entry when the cleanup was non-empty.

To check that exactly the right rows are being deleted:

-- Deletion candidates: how many rows the cleanup will take on the next tick.
SELECT state, count(*), min(updated_at) AS oldest
FROM job
WHERE state IN ('done', 'dead', 'canceled')
  AND updated_at < now() - interval '30 days'
GROUP BY state;

Delete by hand (for example, when cleanup is disabled):

DELETE FROM job
WHERE state IN ('done', 'dead', 'canceled')
  AND updated_at < now() - interval '30 days';

Remember: the idempotency key is released together with the row. If the consumer relied on a repeat with the same key not creating a job, then after the cleanup such a repeat will create a new one — that is the expected behaviour of the TTL, not a defect.

Metrics

/metrics serves ten engine metrics (plus the surface's HTTP metrics):

Average attempt duration = _sum / _count. There are no histograms on purpose: the registry has no buckets, and the sum/count pair covers the average and the rate.

brooq_ready_oldest_age_seconds resets to zero when there are no ready jobs — so "growing" unambiguously means "the queue cannot keep up", without false alarms after a topic is drained.

Typical alerts:

# The queue is stuck: the oldest ready job has been waiting for more than 5 minutes.
brooq_ready_oldest_age_seconds > 300

# Jobs are dying: dead outcomes are growing.
rate(brooq_settle_total{outcome="dead"}[5m]) > 0

# Handlers are hanging.
increase(brooq_hung_workers_total[15m]) > 0

# All slots are busy — the queue is not being drained by the workers.
brooq_worker_busy == on(topic) brooq_jobs{state="processing"}

Job chain (Chain)

A handler can enqueue the next job in the same transaction that closes the current one (D9). This is the main point of the "library" shape: if the write of the next job fails, the current one is not considered closed — both changes roll back together.

func (h handler) Handle(ctx context.Context, job queue.Claim) error {
    if err := h.step(ctx, job); err != nil {
        return err
    }
    // The next step is committed together with the closing of the current one.
    job.Chain(queue.Job{
        Topic:          queue.Topic("email"),
        Payload:        nextPayload,
        IdempotencyKey: "step-2-" + job.ID,
    })
    return nil
}

The chain fires only on success: on a failure or a cancel the next job is not enqueued.

10. Guarantees

Three promises of different strength — it is important not to mix them:

1. Claim — "no more than once at any moment." At no moment is a job pinned to two workers. Ensured by one SQL statement with FOR NO KEY UPDATE SKIP LOCKED and the locked_by guard on the outcome write.
2. Execution — "at least once." The handler may be started again — including without anyone's decision (the lease expired, the outcome write did not go through). The handler must be idempotent. "Exactly once" is achievable only for effects inside the same database, but not for external ones (email, HTTP).
3. Outcome recording — "exactly once." Terminal states are not overwritten; a job's fate is read with one SQL query.

What the queue does not do

• does not guarantee ordering between processes (within a topic — FIFO by available_at, then id);
• does not provide priorities;
• does not retry internal database queries: a failed query is a failed operation;
• does not delete terminal rows: retention is a separate change, so the idempotency key is not released and the table grows;
• does not start goroutines or open connections without an explicit call.

11. Using it as a library

brooq is imported from another module: the engine and the adapter are public (pkg/brooq, pkg/postgres), only the unit's own infrastructure remains in internal/. The proof is a test in the brooq_test package that assembles the queue from the outside and drives a job to done without importing a single internal/*.

Minimal consumer:

package main

import (
    "context"
    "log/slog"

    "github.com/thefish/brooq/pkg/brooq"
    "github.com/thefish/brooq/pkg/db"
    "github.com/thefish/brooq/pkg/postgres"
    "github.com/thefish/brooq/pkg/queue"
)

func main() {
    ctx := context.Background()
    pool, err := db.Connect(ctx, dsn, db.DefaultParams())
    if err != nil {
        panic(err)
    }
    defer pool.Close()

    engine := brooq.New(postgres.New(pool), slog.Default(), brooq.Config{
        Workers: 4, Lease: 2 * time.Minute,
    })
    if err := engine.RegisterKind(queue.Topic("email"), handler{}, queue.KindOptions{
        MaxAttempts: 5, AttemptTimeout: time.Minute,
        MaxExecutionTimeout: 5 * time.Minute,
        BackoffBase: time.Second, BackoffMax: time.Minute,
    }); err != nil {
        panic(err)
    }

    engine.Start()
    defer engine.Close()
    select {} // or your own loop/signal.NotifyContext
}

What the consumer must provide

• its own job table: migrations are a deployment step, the binary does not apply them. All migrations/*.sql files are part of the schema contract, not an implementation detail;
• an idempotent handler: "at least once" execution is not a slogan but a consequence of the lease and retries (§ 4);
• its own PostgreSQL connection: the pool is opened by the consumer, the engine neither creates nor closes it;
• its own transaction or port: its own pgx transaction is brought into the queue via postgres.WrapTx, and back via postgres.AsPgx (§ 6).

What the facade does not have

• the queue's HTTP API — cancel exists (Queue.Cancel), but there is no endpoint;
• priorities, sharding, isolation by application version;
• schema management: retention and partitioning are separate tasks.

pkg/queue remains a frozen contract: renaming a field in Job or Claim breaks the consumer.

12. Tests

make test                      # without a database: integration and e2e — SKIP
make test-db-up
TEST_DATABASE_URL='postgres://brooq:brooq@localhost:55433/brooq_test?sslmode=disable' make test

A green make test without TEST_DATABASE_URL does not prove that the migrations are applied — pkg/postgres and pkg/brooq are skipped. With a database the tests create an isolated schema, apply goose and remove it.

What is covered:

Two e2e tests — by the owner's decision: the claim race and the happy path. The rest are unit tests in table style, without a database.

The full gate before completion:

make build && make test && make lint && make spec-validate

13. What's next

The engine is closed: all tasks of the 01-queue-engine and 02-enqueue-returns-id changes are done and archived, and so is the retention of terminal rows. What remains is consciously deferred:

1. The queue's HTTP API — cancel exists (Queue.Cancel), but there is no endpoint and no authorization; cancel from psql or from the consumer's code.
2. Priorities — the order is strictly FIFO by available_at, id.
3. Table partitioning — TTL cleanup solves the growth problem, but at very large volumes splitting by time will be needed: then old partitions are dropped whole, without DELETE.
4. Sharding and isolation by application version — not needed at the current scale.

What is worth knowing about operations:

• Lease now defines not "how long work may last" (the lease is renewed) but the recovery speed after a worker crash: a crashed process's job comes back after Lease;
• a hung handler that ignores ctx holds its slot until the process restarts — a goroutine in Go cannot be killed. Observed via brooq_hung_workers_total;
• brooq_ready_oldest_age_seconds is the main SLO: growth means that the queue cannot keep up. The value resets to zero when there are no ready jobs;
• the TTL defines the idempotency window: after the row is deleted the key is released, and a repeated enqueue with the same key will create a new job. If a repeat must be cut off for longer, increase QUEUE_RETENTION_TTL or keep the window in the consumer itself.