brooq

Что это

Очередь заданий без внешнего брокера: задания лежат в PostgreSQL, воркеры берут их конкурентно (FOR UPDATE SKIP LOCKED), исход пишется в ту же строку. Один бинарник, один VPS, наблюдаемость с первого дня.

brooq — библиотека: она компилируется в процесс потребителя и делит с ним транзакцию и пул соединений. Отдельный процесс-воркер — следствие, а не суть: cmd/brooq использует ту же библиотеку.

Три обещания разной силы:

• взятие — не более одного раза в каждый момент;
• исполнение — хотя бы один раз (обработчик обязан быть идемпотентным);
• запись исхода — ровно один раз; судьба задания читается одним SELECT.

Зачем

Уведомления и фоновые действия в моём личном проекте шли через push-контур «api → notification-service → транспорт»: состояние задания жило в PostgreSQL, а исполнение — в памяти процесса, из-за чего доставка была плохо отслеживаемой и необслуживаемой. brooq выносит очередь в отдельный проект: Postgres — точка правды, воркеры берут строки сами, всё видно SQL-запросом и метриками.

Быстрый старт

make bin-deps                 # go mod download + зафиксированные версии инструментов
make test-db-up               # локальная PostgreSQL для интеграционных тестов (порт 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

Зелёный make test без TEST_DATABASE_URL не доказывает, что миграции применяются: интеграционные тесты и e2e тогда пропускаются.

Локальный запуск юнита: скопируйте .env.example в .env, задайте DATABASE_URL, затем make migrate-up && make run. Служебная поверхность — 127.0.0.1:8090 (/healthz, /metrics).

Как библиотека

import (
    "github.com/thefish/brooq/pkg/brooq"    // движок
    "github.com/thefish/brooq/pkg/postgres" // SQL-адаптер
    "github.com/thefish/brooq/pkg/queue"    // словарь: Job, Claim, Handler
)

store := postgres.New(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()

// Постановка через пул: id и признак «строка создана этим вызовом».
id, created, err := engine.Enqueue(ctx, queue.Job{Topic: "email", Payload: payload, IdempotencyKey: "order-42"})
if err != nil {
    log.Error("постановка задания", "err", err)
}
// created == false — задание с таким ключом уже было, id — его идентификатор.
log.Info("задание поставлено", "id", id, "created", created)

// Постановка в своей транзакции: своя запись и задание фиксируются вместе.
tx, err := pool.Begin(ctx)                       // 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("постановка задания", "err", err)
}
// ... свои записи тем же tx ...
_ = tx.Commit(ctx)

// Обратное направление: транзакция очереди → свой код.
qtx, err := engine.Begin(ctx)                    // queue.Tx
pgxTx, ok := postgres.AsPgx(qtx)                 // false, если это не *postgres.Tx

postgres.WrapTx заворачивает транзакцию потребителя в порт queue.Tx, postgres.AsPgx делает обратное — отдаёт транзакцию драйвера из порта. Обе стороны — одна и та же транзакция, а не копия.

Подробное руководство с примерами — docs/queue-engine.md.

Документы

Устройство

Пакеты и правила импортов подробно — в docs/architecture.md; коротко:

pkg/queue      словарь: Job, Claim, State, Tx (только stdlib)
pkg/brooq      движок: воркеры, Reaper, heartbeat, метрики
pkg/postgres   SQL-адаптер — единственное место, где живёт PostgreSQL
cmd/brooq      композиция: собрать зависимости, запустить

Схема — только миграции migrations/, накатываются шагом деплоя, никогда из бинарника при старте.

Состояние

Движок готов и используется: постановка (с идентификатором и признаком created), взятие, запись исхода, повтор с backoff, Reaper, heartbeat, отмена в полёте, watchdog, ретенция и одиннадцать метрик brooq_*. Постановка работает и через пул, и в транзакции потребителя — в обе стороны (postgres.WrapTx / postgres.AsPgx).

Отложено осознанно:

Одно ограничение стоит знать заранее: терминальные задания удаляются по TTL (30 дней, QUEUE_RETENTION_TTL), и ключ идемпотентности живёт вместе со строкой. После удаления повтор с тем же ключом создаст новое задание — если повтор должен отсекаться дольше, увеличьте TTL или держите окно в самом потребителе.

Статус: словарь зафиксирован первой сессией доменного моделирования. Слова отсюда используют спеки, ADR и идентификаторы кода (ADR-0005). Правится только в сессии доменного моделирования.

brooq — библиотека очереди заданий: она компилируется в процесс потребителя и делит с ним транзакцию и пул соединений. Отдельный процесс-воркер — следствие, а не суть: он использует ту же библиотеку.

Словарь

Написание, которого избегаем: cancelled (две l) — пишем canceled, как context.Canceled. state — не status (у состояния нет отдельного «статуса»; это одно и то же слово). kind — не синоним темы. payload — не data и не body.

Состояния и события

Состояния — lowercase, события — past-tense PascalCase (ADR-0005). Смысл состояний:

События: JobClaimed, JobDone, JobFailed, JobDead, JobCanceled, JobRequeued (возврат в оборот сборщиком).

failed — не «провалено навсегда», а «ждёт следующей попытки»: задание берётся из pending и из failed, поэтому это состояние ожидания, а не отложенный приговор.

Инварианты

1. Задание в processing всегда имеет аренду: идентификатор воркера и срок аренды заполнены одновременно.
2. Задание не может быть закреплено за двумя воркерами в один момент: захват — одна SQL-строка с FOR UPDATE SKIP LOCKED (ADR-0001, ADR-0010). Это обещание про закрепление, а не про исполнение.
3. Терминальные состояния — done, dead, canceled — не перезаписываются.
4. Исход задания пишется в ту же строку, что и само задание: отдельного «журнала доставок» нет.
5. Очередь не знает домена потребителя: тело задания — непрозрачный JSON.
6. Попытка списывается при взятии: attempts — сколько раз задание уже брали.
7. dead достигается только через записанный сбой: успешное исполнение пишет done безусловно. Истечение аренды попыткой не считается и к dead не приближает — это признак падения процесса, а не сбоя исполнения.
8. Идемпотентность — на паре «тема + ключ»: не более одного задания на пару за всё время жизни строки, включая терминальные состояния. Ключ описывает конкретную попытку доставки, поэтому повтор с другим ключом — новое задание. Задания без ключа не дедуплицируются.
9. Очередь не владеет процессом: она не запускает горутины и не открывает соединения без явного вызова. Всё, что она делает, она делает в транзакции вызывающего или по его команде.
10. Порядок исполнения — по времени готовности, а при равенстве — по порядку постановки. Между процессами порядок не гарантируется.
11. Гарантии разной силы и не смешиваются: взятие — не более одного раза в каждый момент; исполнение — хотя бы один раз; запись исхода — ровно один раз. Повторное исполнение штатно, а не аварийно: оно происходит без чьего-либо решения, когда аренда истекла или запись исхода не прошла.

Как устроена очередь и на что можно опираться, не читая код. Примеры, настройки и приёмы эксплуатации — в queue-engine.md.

Обзор

brooq — очередь заданий, которая живёт целиком в PostgreSQL. Задания лежат в таблице job в той же базе, что и данные потребителя; воркеры берут строки конкурентно; исход попытки пишется в ту же строку, где лежит само задание. Внешнего брокера нет, второго хранилища состояния нет, отдельного сервиса для наблюдения нет.

Поставляется как библиотека плюс бинарник. Библиотека компилируется в процесс потребителя и делит с ним транзакцию и пул соединений: это главное, что она даёт сверх «положить JSON в таблицу». Бинарник cmd/brooq собирается из тех же пакетов и запускает воркер отдельным процессом.

Конечный сценарий: производитель положил задание → воркер взял его → исход записан и виден.

Задания и состояния

Задание — строка таблицы job: тема, тело, состояние, счётчик попыток, время готовности, аренда. Тело — непрозрачный JSONB: его схему знает только потребитель, очередь в него не заглядывает.

Шесть состояний:

failed — не приговор, а ожидание: задание берётся и из pending, и из failed. Терминальных состояний ровно три — done, dead, canceled; они не перезаписываются.

Отмена приходит из двух мест: сигнал обработчика «задание стало неактуальным» и внешняя отмена оператором, включая задание в полёте. Задание в полёте останавливается через потерю аренды, без LISTEN/NOTIFY: воркер продлевает аренду каждые Lease/3 запросом с условием «строка всё ещё в processing и держит её именно я»; после отмены условие не выполняется, обновление возвращает 0 строк, и воркер гасит обработчик, ничего не записывая в строку — исход уже записан отменяющим. Задержка реакции — до Lease/3.

Гарантии

Три обещания разной силы, и смешивать их нельзя:

• взятие — не более одного раза в каждый момент: задание никогда не закреплено за двумя воркерами одновременно;
• исполнение — хотя бы один раз: обработчик может быть запущен повторно, в том числе без чьего-либо решения (аренда истекла, процесс упал, запись исхода не прошла);
• запись исхода — ровно один раз: судьба задания читается одним SELECT.

Отсюда главное требование к потребителю: обработчик обязан быть идемпотентным. «Ровно один раз» достижимо только для эффектов внутри той же базы; для внешних эффектов — письма, HTTP, платежа — только «хотя бы один раз». Два окна повторного исполнения неустранимы по устройству:

• эффект сделан, а запись done не прошла — задание вернётся в оборот, и эффект повторится;
• аренда истекла, пока обработчик ещё работал — задание взял второй воркер, два обработчика работают одновременно.

Поэтому гарантия формулируется по месту эффекта:

Хранилище

Задания живут в таблице job в базе потребителя. Воркеры берут строки одной конструкцией SELECT … FOR UPDATE SKIP LOCKED, поэтому занятая строка не блокирует остальные: второй воркер берёт следующее задание, а не ждёт.

Состояние задания существует только в этой строке. Вторых хранилищ состояния нет — ни Redis, ни карт в памяти, ни файлов. Из этого следует, что воркер не помнит ничего между запусками: падение в любой момент означает «аренда истечёт, задание вернётся в оборот».

Кэш допустим только как ускорение и никогда как источник ответа: если кэш и база разошлись, правда — в базе.

Почему не брокер: судьба задания читается обычным SQL, а не отдельным UI; одна резервная копия и один набор миграций вместо двух; «записать заказ и поставить задание» — одна транзакция вместо dual-write с outbox-таблицей. Объём в десятки тысяч заданий в сутки далёк от пределов, где Postgres-очередь начинает мешать.

Схема и миграции

Схема описана только файлами migrations/*.sql (goose). Второй правды нет: ни CREATE TABLE IF NOT EXISTS в коде при старте, ни схемы, собранной руками на проде.

Применённый файл не редактируется никогда — ошибка лечится новой миграцией. Миграции накатываются шагом деплоя и никогда бинарником при старте: иначе рестарт нескольких экземпляров превращается в гонку DDL. Тесты применяют тот же каталог через goose, поэтому расхождения «в тестах схема одна, на проде другая» не бывает. Откат (+goose Down) обязателен и проверяется на чистой базе.

Ретенция терминальных заданий

Терминальные задания (done, dead, canceled) удаляются по TTL: по умолчанию через 30 дней после перехода в терминальное состояние. Без уборки таблица и её индексы растут бесконечно, а ключ идемпотентности никогда не освобождается.

Возраст считается по updated_at: его выставляет каждый оператор записи исхода, поэтому «старше TTL» и «терминальное дольше TTL» — одно и то же. Удаление идёт пачками, а не одним запросом: большой DELETE держит блокировки и даёт пик мёртвых строк, тогда как пачка фиксируется отдельной транзакцией и повторяется, пока приходит полной. В подзапросе стоит FOR UPDATE SKIP LOCKED — несколько процессов или реплик убирают разные строки, а не ждут друг друга.

Активные состояния (pending, processing, failed) не удаляются никогда, даже если updated_at у них очень старый: в выборке только терминальные. Ноль в TTL означает «уборка выключена» и никогда не означает «удалить всё».

Под уборку есть частичный индекс по updated_at для трёх терминальных состояний: он не растёт вместе с очередью работы и не мешает hot path. Для самой таблицы понижен порог autovacuum: при дефолтных 20 % мёртвых строк очередь с постоянным потоком вставок и удалений получает редкие и очень долгие проходы, а частые короткие дешевле.

Следствие для идемпотентности: TTL задаёт окно дедупликации. Пока строка жива, повтор с тем же ключом не создаёт задание; после уборки — создаёт. Ключ описывает попытку доставки, а не событие в домене потребителя, поэтому это ожидаемое поведение, а не дефект.

Партиционирование по времени — следующий шаг, если объём станет большим: тогда старые партиции отбрасываются целиком, без DELETE.

Пакеты и зависимости

cmd/brooq      композиция: собрать зависимости, запустить
pkg/brooq      движок: воркеры, Reaper, heartbeat, метрики, backoff
pkg/queue      словарь: Job, State, Claim, Lease, Tx, Handler
pkg/postgres   SQL-адаптер: единственное место, где живёт PostgreSQL
                (и мост транзакций WrapTx / Pgx / AsPgx)
pkg/{config,db,logger}   переиспользуемые пакеты без знания домена
internal/{archgate,health,loopback,observability}   инфраструктура юнита

Импорты идут только вниз, и это проверяется тестом по фактическим зависимостям, а не договорённостью:

• pkg/* не импортирует internal/*;
• pkg/queue не знает ни SQL, ни транспорта: только stdlib. Транзакция производителя передаётся структурным интерфейсом queue.Tx, которому транзакция драйвера удовлетворяет без импорта драйвера;
• pkg/brooq не знает PostgreSQL: работает через собственный порт Repo, поэтому тестируется без базы, а хранилище подменяется фейком;
• SQL живёт только в pkg/postgres.

Движок и адаптер публичны, потому что каталог internal/ нельзя импортировать из чужого модуля: с пакетами в internal/ это была бы не библиотека. В internal/ остаётся только инфраструктура самого юнита.

Транзакции

Любая запись идёт через транзакцию: «взять задание и записать попытку», «подтвердить и запланировать повтор» — атомарные операции. Чтения — списки, статистика, служебные запросы — идут прямо через пул: транзакция ради одного SELECT держит снимок и мешает вакууму.

Запись через голый пул — дефект. Если метод не принимает транзакцию, это видно в ревью и должно быть обосновано.

Мост транзакций

Порт queue.Tx абстрагирует транзакцию, поэтому на стыке двух миров нужны обе стороны: потребитель со своим pgx-слоем и очередь со своим портом. Мост живёт в адаптере, потому что только ему разрешён драйвер:

func WrapTx(tx pgx.Tx) *Tx             // pgx.Tx → queue.Tx (тот же объект)
func (t *Tx) Pgx() pgx.Tx              // *Tx → pgx.Tx
func AsPgx(tx queue.Tx) (pgx.Tx, bool) // queue.Tx → pgx.Tx, безопасно

WrapTx не открывает новую транзакцию и не копирует её: постановка через обёртку уходит в ту самую транзакцию потребителя, поэтому откат не оставляет задания, а фиксация делает его видимым воркерам. Возврат конкретного *Tx, а не порта — чтобы обратное направление оставалось доступным без приведения типа: *Tx и так удовлетворяет queue.Tx.

AsPgx нужен там, где порт пришёл из движка — из Queue.Begin или из Claim.OnSucceeded: на значении порта метод Pgx не вызвать. Второй результат — false для чужой реализации порта, nil и пустой обёртки: разворачивать чужие транзакции адаптер не берётся.

Закрывает транзакцию тот, кто её открыл. Обёртка лишь переводит её между представлениями: Commit или Rollback через любую из сторон закрывает её для всех, кто её делит.

Служебные контуры

Служебная поверхность (/healthz, /metrics) и любые внутренние эндпоинты слушают только 127.0.0.1. Внутренние эндпоинты дополнительно требуют сервисный bearer-токен, сравнение — постоянного времени. Публичная поверхность появляется только через nginx и только осознанно. /healthz не стоит за аутентификацией: иначе мониторинг слепнет ровно тогда, когда нужен.

Секреты

Секреты — DSN с паролем, ключи провайдеров, bearer-токены — не живут в репозитории, в коде, в миграциях и в базе в открытом виде. Локально это .env (gitignored) или переменные окружения; на сервере — EnvironmentFile для несекретных настроек и LoadCredential= для секретов, чтобы значение не попадало в /proc/<pid>/environ и в systemctl show. Пустое значение переменной считается незаданным: KEY= не должен превращать дефолт в пустую строку.

Язык и имена

Комментарии, doc-комментарии, сообщения коммитов, тексты ошибок и логи — по-русски. Идентификаторы — английские; русский термин в коде допустим только как значение-данные, не как имя. Пользовательские тексты не пишутся инлайн: они приходят из каталога сообщений по символическому ключу.

Состояния и прочие значения-перечисления — lowercase (pending). События — past-tense PascalCase (JobClaimed). Темы заданий — lowercase через дефис или точку (email.send). Таблицы — единственное число (job), колонки — snake_case (available_at), Go-поля — PascalCase, JSON-теги — snake_case.

Состояние — это положение вещи, событие — факт. Когда их пишут одинаково, запрос «что случилось» и запрос «где сейчас» начинают путаться. Движок сейчас событий не порождает: наблюдать отмену можно по state и метрикам.

Почему свой движок

Готовые библиотеки очередей на том же принципе пришлось бы брать вместе с их схемой и их мигратором. Для проекта, где схема описана только goose-миграциями и накатывается одним шагом деплоя, вторая лента миграций разрушает единый процесс развёртывания и тестирования на чистой базе.

Второе ограничение — модель аренды. В части готовых реализаций нет явного состояния «взято»: аренда закодирована в отложенном времени готовности, и вопрос «что сейчас в полёте и кто это держит» невыразим, а отмена создаёт гонку — успех воркера перезаписывает отмену.

Вместо зависимости взяты приёмы, а не код: формула backoff с защитой от переполнения, перехват паники обработчика, watchdog на зависший обработчик, ограничение выборки числом свободных воркеров, постановка следующего задания в транзакции закрытия текущего. Продление аренды короткими интервалами с проверкой владения поглощает потребность в LISTEN/NOTIFY.

Эксплуатация и ограничения

Один хост, один systemd-юнит, PostgreSQL рядом. Бинарник статический, конфигурация из .env, секреты через LoadCredential, миграции — шагом деплоя. Служебная поверхность слушает loopback. Такой деплой обслуживается одним человеком: откат — вернуть предыдущий бинарник, диагностика — systemctl status и journalctl. Масштабирование за пределы хоста не требует изменений в очереди: достаточно запустить воркеры на нескольких хостах.

Что стоит знать до продакшена:

• TTL задаёт окно идемпотентности. Терминальные строки удаляются через QUEUE_RETENTION_TTL (по умолчанию 30 дней); после удаления ключ освобождается, и повтор с тем же ключом создаст новое задание. Если повтор должен отсекаться дольше — увеличьте TTL или держите окно в потребителе. Уборку можно выключить целиком: QUEUE_RETENTION_TTL=0; освобождается, таблица растёт. Ретенция и партиционирование — отдельная работа, которая понадобится с объёмом;
• обработчик, игнорирующий контекст, не убить. Продление аренды после отмены прекращается, поэтому сборщик вернёт задание в оборот через Lease, а зависшая горутина останется утечкой памяти до перезапуска процесса. Утечка видна в brooq_hung_workers_total;
• абсолютного потолка без зависания нет. Обработчик, который убивает процесс, сбой не пишет, и dead не достигается: задание будет браться снова. Нужен внешний детектор — рост attempts при пустом last_error — и ручной перевод строки в dead;
• HTTP-API очереди нет. Отмена доступна из кода потребителя и из psql; приоритетов нет, порядок строго FIFO по времени готовности.

Практическое руководство по движку. Устройство, гарантии и ограничения — в architecture.md.

1. Быстрый старт

Пять шагов от пустой базы до пойманного задания.

Шаг 1. Накатить миграции

make test-db-up                    # локальная PostgreSQL для проб
export DATABASE_URL='postgres://brooq:brooq@localhost:55433/brooq_test?sslmode=disable'
make migrate-up
make migrate-status

Ожидаемый вывод: применены 00001_job и 00002_job_idempotency. Повторный make migrate-up — no-op. Миграции накатываются шагом деплоя, никогда из бинарника при старте.

Шаг 2. Задать настройки

Всё через окружение или .env, дефолты уже рабочие:

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

Невалидные значения (ноль, отрицательное, AttemptTimeout > Lease) отклоняются на старте с перечислением всех проблем сразу.

Шаг 3. Написать обработчик

package main

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

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

// notification — тело задания; схему знает только потребитель.
type notification struct {
    UserID string `json:"user_id"`
    Text   string `json:"text"`
}

type notifyHandler struct {
    log *slog.Logger
}

// Handle исполняет одно задание. Обработчик обязан быть идемпотентным:
// исполнение может повториться (аренда истекла, запись исхода не прошла).
func (h notifyHandler) Handle(ctx context.Context, job queue.Claim) error {
    var msg notification
    if err := json.Unmarshal(job.Payload, &msg); err != nil {
        // Постоянная ошибка: повтор не поможет, но и отменить задание
        // нельзя — пусть попытки исчерпаются и задание станет dead.
        return err
    }

    if err := send(ctx, msg); err != nil {
        // Временная ошибка внешнего API: повторить позже, не расходуя
        // попытку (попытка уже списана при взятии — вернём её).
        return queue.ErrMustPreserveAttempt
    }

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

Шаг 4. Собрать движок

// Хранилище — публичный SQL-адаптер; пул открывает потребитель
// (pkg/db — удобный помощник, но подойдёт любой *pgxpool.Pool).
store := postgres.New(pool)

// Движок собирается из хранилища, логгера и настроек.
engine := brooq.New(store, log, brooq.Config{
    Workers:      2,
    PollInterval: time.Second,
    Lease:        2 * time.Minute,
})

// Регистрация темы — до Start: ошибки настроек видны на старте.
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()

Шаг 5. Положить задание и увидеть исход

// Задание ставится через фасад — SQL-адаптер трогать не нужно.
// id известен сразу, created == false — задание с таким ключом уже было.
id, created, err := engine.Enqueue(ctx, queue.Job{
    Topic:   queue.Topic("notification"),
    Payload: []byte(`{"user_id":"u-42","text":"привет"}`),
    // Ключ идемпотентности необязателен: с ним повторная постановка
    // не создаст второе задание.
    IdempotencyKey: "notify-u-42-2026-10-20",
})
if err != nil {
    return err
}
log.Info("задание поставлено", "id", id, "created", created)

Воркер заберёт задание и запишет исход в ту же строку; attempts станет 1, state — done.

Импорты потребителя — только публичные пакеты:

import (
    "github.com/thefish/brooq/pkg/brooq"    // движок: New, Config, Queue
    "github.com/thefish/brooq/pkg/postgres" // SQL-адаптер для PostgreSQL
    "github.com/thefish/brooq/pkg/queue"    // словарь: Job, Claim, Handler, сигналы
)

Ни один internal/* не нужен: это и есть смысл публичного фасада.

Готовый пример целиком — cmd/brooq/main.go (там тема demo и демонстрационный обработчик).

Проверка вживую:

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. Публичная поверхность

Три публичных пакета — и больше ничего не нужно потребителю:

Направление импортов — только вниз: pkg/brooq не знает PostgreSQL (работает через порт Repo), pkg/queue не знает ни того, ни другого. Поэтому движок тестируется без базы, а хранилище подменяется.

Фасад движка:

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

// Постановка — не нужно трогать SQL-адаптер напрямую.
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)

// Административная отмена (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

Словарь (pkg/queue) — ниже.

// Задание для постановки.
type Job struct {
    Topic          Topic
    Payload        Payload         // []byte, непрозрачный JSON
    IdempotencyKey string          // "" = ключа нет
    AvailableAt    time.Time       // нулевое = сейчас
}

// Взятое задание, которое видит обработчик.
type Claim struct {
    ID      string
    Topic   Topic
    Payload Payload
    Attempt int                    // номер состоявшейся попытки, с 1
    Lease   Lease                  // кто держит и до какого момента
    // OnSucceeded — колбэк для постановки следующего задания в той же
    // транзакции закрытия текущего. Заполняет движок перед вызовом
    // обработчика, поэтому в Handle он всегда не nil; tx — порт
    // queue.Tx, драйвер из него достаёт postgres.AsPgx.
    OnSucceeded func(ctx context.Context, tx Tx) error
}

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

// Настройки темы: задаются в коде, а не колонкой таблицы.
type KindOptions struct {
    MaxAttempts         int
    AttemptTimeout      time.Duration
    MaxExecutionTimeout time.Duration
    BackoffBase         time.Duration
    BackoffMax          time.Duration
}

3. Состояния

                   ┌──────────── done  (успех, безусловно)
                   │
pending ──взятие──▶ processing ──┬──────── dead  (сбой на последней попытке)
   ▲                   │         │
   │                   │         └──────── canceled (сигнал отмены / внешняя отмена)
   │                   │
   └── Reaper ◀────────┤ аренда истекла  → pending (§ 10)
                       │
                       └── failed ──(available_at)──▶ берётся снова

Ключевое: failed — это очередь ожидания, а не приговор. Задание берётся из pending и failed, когда наступил available_at.

Терминальные состояния не перезаписываются: все записи исхода идут условным UPDATE с guard-ом WHERE state = 'processing' AND locked_by = $me, и «ноль изменённых строк» — штатный результат, а не ошибка.

4. Политика повторов

Попытка списывается при взятии

attempts увеличивается в том же UPDATE, который закрепляет аренду. Значит Claim.Attempt — уже номер состоявшейся попытки, начиная с 1.

Отсрочка

d = min(BackoffBase · 2^(attempts-1), BackoffMax)
фактическая отсрочка = равномерно из [d/2, d]

Джиттер разводит воркеров, которые иначе ударили бы по внешнему API одновременно. Защита от переполнения стоит до возведения в степень: при большом номере попытки результат равен потолку, без Inf.

Пример при BackoffBase=1s, BackoffMax=1m:

Три сигнала обработчика

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

// 1. «Повтори позже, вот через столько» — не расходует попытку сверх
//    уже списанной и не уменьшает счётчик. Отсрочка клампится к
//    BackoffMax. На последней попытке задание всё равно станет dead.
return queue.MustDelay(30 * time.Second)

// 2. «Сбой не должен расходовать попытку» — задание вернётся в failed,
//    attempts уменьшится обратно (но не ниже нуля), dead недостижим.
return queue.ErrMustPreserveAttempt

// 3. «Задание стало неактуальным» — canceled независимо от остатка
//    попыток, обработка прекращается.
return queue.ErrMustCancelTask

Ошибки оборачиваемы — fmt.Errorf("отправка: %w", queue.ErrMustCancelTask) работает так же. Проверить отсрочку в своём коде:

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

Обычная ошибка

Возврат любой другой ошибки — сбой попытки: failed + backoff, либо dead, если попытки исчерпаны. Причина пишется в last_error (без секретов — это ваша ответственность).

Паника

Паника обработчика перехватывается: задание получает исход сбоя, воркер продолжает работать, процесс не завершается. Стек пишется в лог уровнем ERROR.

5. Причины прерывания контекста

Движок различает таймаут и отмену через context.WithCancelCause (D15). Обработчику не нужно разбирать причину — достаточно возвращать ctx.Err().

func (h handler) Handle(ctx context.Context, job queue.Claim) error {
    if err := h.doWork(ctx); err != nil {
        if ctx.Err() != nil {
            // Отмена или таймаут: исход определит движок по context.Cause.
            return ctx.Err()
        }
        return err
    }
    return nil
}

ErrMaxExecutionExceeded не делает задание мёртвым: dead достигается только исчерпанием попыток.

6. Производитель

Обычная постановка

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)

В транзакции производителя (главный смысл «библиотеки»)

Задание фиксируется вместе с вашими изменениями — атомарно:

tx, err := repo.Begin(ctx)
if err != nil {
    return err
}
defer tx.Rollback(ctx) // безопасно: после Commit — no-op

// 1. Меняем своё состояние.
if _, err := tx.Exec(ctx, `UPDATE account SET notified = true WHERE id = $1`, id); err != nil {
    return err
}

// 2. Кладём задание в той же транзакции.
// Идентификатор известен до фиксации, но действителен только после 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. Мету пишем всегда и идемпотентно по job_id: created == false может
// относиться к уже завершённому заданию — тогда мета просто дописывается.
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. Фиксируем: либо оба изменения видны, либо ни одного.
return tx.Commit(ctx)

Если транзакция откатится — задания не будет. Если зафиксируется — воркер его увидит.

Обе стороны одной транзакции

Граница между кодом потребителя и очередью проходится в обе стороны. postgres.WrapTx заворачивает транзакцию драйвера в порт queue.Tx, а (*Tx).Pgx и postgres.AsPgx возвращают обратно транзакцию драйвера.

// своя транзакция → очередь
tx, err := pool.Begin(ctx)                       // pgx.Tx потребителя
defer func() { _ = tx.Rollback(ctx) }()
if _, _, err := repo.EnqueueTx(ctx, postgres.WrapTx(tx), job); err != nil { ... }
// ... свои записи тем же tx ...
err = tx.Commit(ctx)

// транзакция очереди → свой код
qtx, err := engine.Begin(ctx)                    // queue.Tx
pgxTx, ok := postgres.AsPgx(qtx)                 // false, если это не *postgres.Tx

WrapTx возвращает конкретный *postgres.Tx — он же queue.Tx, поэтому годится всюду, где движок ждёт порт. Pgx и AsPgx дают ту же самую транзакцию, а не копию: Commit или Rollback через неё закрывает её для всех, кто её делит. Закрывает её владелец — тот, кто открыл. Тот же приём работает в Claim.OnSucceeded(ctx, tx): движок отдаёт queue.Tx, а AsPgx превращает его в транзакцию драйвера для своих записей в той же транзакции.

Идемпотентность

// Повтор с тем же ключом → второго задания нет (ON CONFLICT DO NOTHING).
id, created, err := repo.Enqueue(ctx, queue.Job{Topic: "notification", Payload: p, IdempotencyKey: "order-42-paid"})
// В таблице одна строка; первый вызов: created == true, id — новая строка.
id2, created2, err := repo.Enqueue(ctx, queue.Job{Topic: "notification", Payload: p, IdempotencyKey: "order-42-paid"})
// Второй вызов: тот же id (id2 == id) и created2 == false — задания не появилось.

Признак created отвечает на вопрос «новая ли строка», а не «новый ли эффект»: created == false может относиться к уже завершённому заданию (производитель поставил задание и упал до записи меты, воркер довёл его до done, производитель повторил попытку). Поэтому мету потребителя пишут всегда, делая её идемпотентной по job_id, — в той же транзакции, что и EnqueueTx:

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

Тогда повтор при created == false дописывает недостающую мету и не дублирует существующую. Обратный порядок («сначала мета, потом постановка») не годится: id ещё неизвестен.

Правила:

• ключ уникален внутри темы: одинаковый ключ у разных тем — два разных задания;
• пустая строка "" означает «ключа нет» (превращается в NULL), и такие задания не дедуплицируются;
• ключ из пробелов " " — настоящий ключ;
• ключ не освобождается после терминального состояния: повтор через месяц с тем же ключом не создаст задание. Ключ описывает конкретную попытку доставки — для новой доставки нужен новый ключ.

7. Отмена задания

// Отменяет и ждущее, и исполняющееся задание.
err := engine.Cancel(ctx, jobID)

Из psql — то же самое:

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

Как отмена доходит до работающего обработчика

Механизм — потеря аренды, без LISTEN/NOTIFY (D14):

1. отменяющий переводит строку в canceled;
2. следующий heartbeat воркера не проходит guard state='processing' и возвращает 0 строк;
3. воркер вызывает cancelCause(ErrLeaseLost), обработчик получает ctx.Done(), воркер не пишет в строку ничего — исход уже записан отменяющим.

Задержка реакции — не дольше Lease/3 (период продления).

отмена ──▶ строка canceled ──▶ heartbeat: 0 строк ──▶ ErrLeaseLost ──▶ ctx.Done()
                                      (≤ Lease/3)                        │
                                                                         ▼
                                                     воркер НЕ пишет исход

Обработчик, уважающий ctx, завершится сам. Обработчик, игнорирующий ctx, продолжит работу (горутину в Go убить нельзя) — тогда сработает watchdog: ERROR в лог и brooq_hung_workers_total. Его запись исхода получит 0 строк и не перезапишет отмену.

8. Настройки

Валидация на старте отклоняет:

• QUEUE_WORKERS < 1, QUEUE_MAX_ATTEMPTS < 1, QUEUE_RETENTION_BATCH < 1;
• любой период <= 0 — кроме QUEUE_RETENTION_PERIOD, где 0 значит «расчёт»;
• QUEUE_RETENTION_TTL отрицательный; ноль допустим и означает «уборка выключена»;
• AttemptTimeout > MaxExecutionTimeout;
• любой период <= 0;
• AttemptTimeout > MaxExecutionTimeout;
• AttemptTimeout > Lease — ограничение снято: аренда продлевается heartbeat-ом, поэтому длинная работа больше не теряет аренду. Следствие для эксплуатации: чем короче Lease, тем быстрее Reaper вернёт задание упавшего воркера, но тем чаще продления.

Голые числа без единиц (QUEUE_LEASE=120) отклоняются: time.Duration требует суффикс (120s).

Как выбирать Lease

Lease больше не ограничивает длительность работы: пока обработчик жив, воркер продлевает аренду каждые Lease/3. Поэтому Lease — это скорость восстановления:

• короткий Lease (10–30s) — задание упавшего воркера вернётся быстро, но продления идут чаще (больше запросов к БД);
• длинный Lease (2–5m) — меньше нагрузки, но падение стоит до Lease простоя задания.

WatchdogGrace — сколько ждать выхода обработчика после отмены, прежде чем зафиксировать зависание. По умолчанию равен Lease.

9. Схема и наблюдение

Колонки

Индексы

Полезные запросы

-- Судьба задания одним запросом — главное обещание очереди.
SELECT state, attempts, last_error, available_at, locked_by, locked_until
FROM job WHERE id = '...';

-- Что сейчас в полёте и кто держит.
SELECT id, topic, locked_by, locked_until, attempts,
       locked_until < now() AS lease_expired
FROM job WHERE state = 'processing' ORDER BY locked_until;

-- Глубина очереди по темам.
SELECT topic, state, count(*) FROM job GROUP BY topic, state ORDER BY 1, 2;

-- Возраст старейшего готового задания (лаг очереди).
SELECT topic, now() - min(available_at) AS oldest_ready
FROM job WHERE state IN ('pending','failed') AND available_at <= now()
GROUP BY topic;

-- Мёртвые задания: разбирать руками.
SELECT id, topic, attempts, last_error, updated_at
FROM job WHERE state = 'dead' ORDER BY updated_at DESC;

Reaper: что происходит с зависшими заданиями

Reaper работает внутри каждого процесса: одна горутина, период max(5s, Lease/4), переопределяется QUEUE_REAPER_PERIOD. Он делает ровно то, что описано в D8:

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;

Почему именно так:

• attempts не меняется: истечение аренды — признак упавшего процесса, а не записанного сбоя (инвариант 7). Поэтому падение не приближает задание к dead;
• available_at = now(), без дополнительной отсрочки: сама аренда уже была ожиданием — задание не могло вернуться раньше срока. Вторая задержка была бы двойной;
• guard state = 'processing' делает возврат идемпотентным и не даёт воскресить терминальные строки (у них тоже остался locked_until с прошлой аренды);
• несколько реплик безопасны: таблицы-мьютекса не нужно, второй одновременный сборщик увидит строку уже в pending и ничего не вернёт повторно.

Смотреть, что именно вернулось: в логе Info с count и topics. Сколько заданий прямо сейчас исполняется и кто их держит — запрос «что сейчас в полёте» выше.

Если всё же нужно вернуть задание вручную (например, сборщик отключён):

-- ВНИМАНИЕ: безопасно только если аренда действительно истекла.
-- attempts не меняется — истечение аренды попыткой не считается.
UPDATE job
SET state = 'pending', locked_by = NULL, locked_until = NULL, updated_at = now()
WHERE state = 'processing' AND locked_until < now();

Не запускайте это в цикле с малым периодом: задание, чья аренда только что истекла, может исполняться живым, но медленным воркером. Его запись исхода после возврата в оборот получит 0 строк (guard по locked_by), то есть работа не будет засчитана — задание исполнится повторно. Это ровно то окно, ради устранения которого нужен heartbeat (4.6).

Уборка терминальных строк (ретенция)

Терминальные задания (done, dead, canceled) удаляются по TTL: по умолчанию через 30 дней после перехода в терминальное состояние. Зачем: без уборки таблица и её индексы растут бесконечно, а ключ идемпотентности никогда не освобождается.

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                                  -- размер пачки
    FOR UPDATE SKIP LOCKED
)
RETURNING id, topic;

Почему так:

• updated_at как возраст. Каждый оператор исхода выставляет его, поэтому «старше TTL» и «терминальное дольше TTL» — одно и то же;
• пачками, а не всё сразу. Один большой DELETE держит блокировки и даёт пик мёртвых строк. Горутина удаляет до QUEUE_RETENTION_BATCH строк в одной транзакции и повторяет, пока пачка приходит полной;
• FOR UPDATE SKIP LOCKED. Несколько процессов или реплик убирают разные строки, а не ждут друг друга;
• ORDER BY updated_at. Сначала самое старое, чтобы отставшая уборка не копила хвост;
• активные строки не трогаются — в выборке только терминальные состояния, а pending/processing/failed не удаляются никогда, даже с очень старым updated_at.

Выключение. QUEUE_RETENTION_TTL=0 отключает уборку целиком: горутина не запускается. Это защита от случайного DELETE терминальной части таблицы — ноль в TTL никогда не означает «удалить всё».

Цикл ограничен сотней пачек за тик. Если накопление больше, остаток уйдёт следующим тиком, а в лог придёт Warn — лучше, чем один бесконечный проход, который незаметно держит нагрузку.

Autovacuum. Миграция 00003 понижает порог для таблицы job (autovacuum_vacuum_scale_factor = 0.02, threshold = 100): при дефолтных 20 % мёртвых строк очередь с постоянным потоком вставок и удалений получает редкие и очень долгие проходы. Частые короткие проходы дешевле.

Сколько удалено: метрика brooq_retention_deleted_total{topic} и запись Info в логе при непустой уборке.

Проверить, что удаляется именно то, что нужно:

-- Кандидаты на удаление: столько строк уборка заберёт в ближайший тик.
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 FROM job
WHERE state IN ('done', 'dead', 'canceled')
  AND updated_at < now() - interval '30 days';

Помните: ключ идемпотентности освобождается вместе со строкой. Если потребитель полагался на то, что повтор с тем же ключом не создаст задание, то после уборки такой повтор создаст новое — это ожидаемое поведение TTL, а не дефект.

Метрики

/metrics отдаёт одиннадцать метрик движка (плюс HTTP-метрики поверхности):

Средняя длительность попытки = _sum / _count. Гистограмм нет сознательно: реестр без корзин, а пара sum/count покрывает среднее и темп.

brooq_ready_oldest_age_seconds обнуляется, когда готовых заданий нет — поэтому «растёт» однозначно означает «очередь не успевает», без ложных срабатываний после опустошения темы.

Типовые алерты:

# Очередь встала: старейшее готовое ждёт больше 5 минут.
brooq_ready_oldest_age_seconds > 300

# Задания умирают: растут dead-исходы.
rate(brooq_settle_total{outcome="dead"}[5m]) > 0

# Обработчики зависают.
increase(brooq_hung_workers_total[15m]) > 0

# Все слоты заняты — очередь не разгружается воркерами.
brooq_worker_busy == on(topic) brooq_jobs{state="processing"}

Grafana: подключение с нуля

Метрики отдаёт сама библиотека — отдельного экспортёра нет. Наружу поверхность не выставляется (ADR-0009): /metrics слушает loopback (HTTP_ADDR, по умолчанию 127.0.0.1:8090), поэтому сборщик должен стоять на том же хосте или ходить через туннель.

1. Сбор. Prometheus на том же VPS:

# /etc/prometheus/prometheus.yml
scrape_configs:
  - job_name: brooq
    scrape_interval: 15s
    static_configs:
      - targets: ['127.0.0.1:8090']

Если Prometheus живёт отдельно — туннель вместо публикации порта:

# постоянно, как systemd-юнит на стороне Prometheus
ssh -N -L 127.0.0.1:8090:127.0.0.1:8090 thefish@zaar.be

Проверка до Grafana — метрики видны и с правильными типами:

curl -s 127.0.0.1:8090/metrics | grep '^# TYPE brooq_'

2. Источник данных. Grafana → Connections → Data sources → Prometheus → URL http://127.0.0.1:9090 (или адрес своего Prometheus) → Save & test. Если Prometheus не на этом же хосте с Grafana — нужен http://<host>:9090.

3. Панели. Пять запросов закрывают эксплуатацию. Вставляются в панель как PromQL, topic — переменная дашборда (label_values(brooq_jobs, topic)).

# SLO: сколько ждёт старейшее готовое задание (главный график)
max by (topic) (brooq_ready_oldest_age_seconds)

# Глубина очереди по состояниям: растёт pending — не хватает воркеров
sum by (topic, state) (brooq_jobs)

# Темп исходов: сколько заданий закрывается в секунду и чем именно
sum by (topic, outcome) (rate(brooq_settle_total[5m]))

# Средняя длительность попытки (гистограмм нет, считаем из пары sum/count)
  sum by (topic) (rate(brooq_attempt_duration_seconds_sum[5m]))
/ sum by (topic) (rate(brooq_attempt_duration_seconds_count[5m]))

# Занятость пула: сколько слотов работает прямо сейчас
sum by (topic) (brooq_worker_busy)

rate() по счётчикам требует минимум двух скрейпов — на дашборде это означает «первые 30 секунд после старта Prometheus графики пустые». Это нормально, а не дефект.

4. Алерты. Правила — в Prometheus, не в Grafana: тогда они работают без открытого браузера.

# /etc/prometheus/rules/brooq.yml
groups:
  - name: brooq
    rules:
      - alert: BrooqQueueStalled
        expr: max by (topic) (brooq_ready_oldest_age_seconds) > 300
        for: 10m
        labels: { severity: warning }
        annotations:
          summary: "Очередь {{ $labels.topic }} стоит: старейшее задание ждёт {{ $value | humanizeDuration }}"

      - alert: BrooqJobsDying
        expr: sum by (topic) (rate(brooq_settle_total{outcome="dead"}[10m])) > 0
        for: 5m
        labels: { severity: warning }
        annotations:
          summary: "Тема {{ $labels.topic }}: задания умирают (dead)"

      - alert: BrooqHungHandlers
        expr: increase(brooq_hung_workers_total[15m]) > 0
        labels: { severity: critical }
        annotations:
          summary: "Тема {{ $labels.topic }}: обработчик не вышел после отмены"

      - alert: BrooqLeasesLost
        expr: increase(brooq_lease_lost_total[15m]) > 0
        labels: { severity: info }
        annotations:
          summary: "Тема {{ $labels.topic }}: аренда потеряна, возможен повторный запуск"

      - alert: BrooqTargetDown
        expr: up{job="brooq"} == 0
        for: 2m
        labels: { severity: critical }
        annotations:
          summary: "brooq не отдаёт /metrics"

BrooqLeasesLost — уровень info, а не critical: потеря аренды штатна (так работает «хотя бы один раз», ADR-0010), и тревожить по ней стоит только при росте.

Чего в дашборде нет. Гистограмм и квантилей: реестр не хранит корзин, а пара _sum/_count даёт среднее и темп — для SLO этого хватает. Если нужны p95/p99 — это отдельное решение (корзины в реестре или экспорт в prometheus-client у потребителя), а не настройка.

Цепочка заданий (Chain)

Обработчик может поставить следующее задание в той же транзакции, которой закрывается текущее (D9). Это главный смысл формы «библиотека»: если запись следующего задания не удастся, текущее не считается закрытым — оба изменения откатятся вместе.

func (h handler) Handle(ctx context.Context, job queue.Claim) error {
    if err := h.step(ctx, job); err != nil {
        return err
    }
    // Следующий шаг фиксируется вместе с закрытием текущего.
    job.Chain(queue.Job{
        Topic:          queue.Topic("email"),
        Payload:        nextPayload,
        IdempotencyKey: "step-2-" + job.ID,
    })
    return nil
}

Цепочка срабатывает только при успехе: при сбое или отмене следующее задание не ставится.

10. Гарантии

Три обещания разной силы — их важно не смешивать:

1. Взятие — «не более одного раза в каждый момент». Ни в какой момент задание не закреплено за двумя воркерами. Обеспечено одной SQL-строкой с FOR NO KEY UPDATE SKIP LOCKED и guard-ом locked_by при записи исхода.
2. Исполнение — «хотя бы один раз». Обработчик может быть запущен повторно — в том числе без чьего-либо решения (аренда истекла, запись исхода не прошла). Обработчик обязан быть идемпотентным. «Ровно один раз» достижимо только для эффектов внутри той же БД, но не для внешних (письмо, HTTP).
3. Запись исхода — «ровно один раз». Терминальные состояния не перезаписываются; судьба задания читается одним SQL-запросом.

Чего очередь не делает

• не гарантирует порядок между процессами (внутри темы — FIFO по available_at, затем id);
• не даёт приоритетов;
• не ретраит внутренние запросы к БД: сбой запроса — сбой операции;
• не удаляет терминальные строки: ретенция — отдельный change, поэтому ключ идемпотентности не освобождается, а таблица растёт;
• не запускает горутины и не открывает соединения без явного вызова.

11. Использование как библиотеки

brooq импортируется из чужого модуля: движок и адаптер публичны (pkg/brooq, pkg/postgres), в internal/ осталась только инфраструктура самого юнита. Доказательство — тест в пакете brooq_test, который собирает очередь снаружи и проводит задание до done, не импортируя ни одного internal/*.

Минимальный потребитель:

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 {} // или свой цикл/signal.NotifyContext
}

Что от потребителя требуется

• своя таблица job: миграции — шаг деплоя, бинарник их не накатывает. Все файлы migrations/*.sql — часть контракта схемы, а не деталь реализации;
• идемпотентный обработчик: исполнение «хотя бы один раз» — не лозунг, а следствие аренды и повторов (§ 4);
• своё соединение с PostgreSQL: пул открывает потребитель, движок его не создаёт и не закрывает;
• своя транзакция или порт: своя pgx-транзакция заводится в очередь через postgres.WrapTx, обратно — postgres.AsPgx (§ 6).

Чего в фасаде нет

• HTTP-API очереди — отмена есть (Queue.Cancel), но эндпоинта нет;
• приоритетов, шардирования, изоляции по версии приложения;
• управления схемой: ретенция и партиционирование — отдельные задачи.

pkg/queue при этом остаётся замороженным контрактом: переименование поля в Job или Claim ломает потребителя.

12. Тесты

make test                      # без базы: интеграционные и e2e — SKIP
make test-db-up
TEST_DATABASE_URL='postgres://brooq:brooq@localhost:55433/brooq_test?sslmode=disable' make test

Зелёный make test без TEST_DATABASE_URL не доказывает, что миграции применяются — pkg/postgres и pkg/brooq пропускаются. С базой тесты создают изолированную схему, накатывают goose и убирают её.

Что покрыто:

Два e2e — по решению владельца: гонка взятия и happy path. Остальное — unit-тесты в табличном стиле, без базы.

Полный гейт перед завершением:

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

13. Что дальше

Движок закрыт: все задачи change'ов 01-queue-engine и 02-enqueue-returns-id выполнены и заархивированы, ретенция терминальных строк тоже. Осталось осознанно отложенное:

1. HTTP-API очереди — отмена есть (Queue.Cancel), но эндпоинта и авторизации нет; отмена из psql или из кода потребителя.
2. Приоритеты — порядок строго FIFO по available_at, id.
3. Партиционирование таблицы — уборка по TTL решает проблему роста, но при очень больших объёмах понадобится разбиение по времени: тогда старые партиции отбрасываются целиком, без DELETE.
4. Шардирование и изоляция по версии приложения — не нужны на текущем масштабе.

Что стоит знать про эксплуатацию:

• Lease теперь задаёт не «сколько может длиться работа» (аренда продлевается), а скорость восстановления после падения воркера: задание упавшего процесса вернётся через Lease;
• зависший обработчик, игнорирующий ctx, удерживает слот до перезапуска процесса — горутину в Go убить нельзя. Наблюдается через brooq_hung_workers_total;
• brooq_ready_oldest_age_seconds — главный SLO: рост означает, что очередь не успевает. Значение обнуляется, когда готовых заданий нет;
• TTL задаёт окно идемпотентности: после удаления строки ключ освобождается, и повторная постановка с тем же ключом создаст новое задание. Если повтор должен отсекаться дольше, увеличьте QUEUE_RETENTION_TTL или держите окно в самом потребителе.