> ## Documentation Index
> Fetch the complete documentation index at: https://omniflow-vpn.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Задачи и очереди

> Долговременная очередь и изменения состояния, которые её наполняют, виды задач, представление недоставленных задач, периодические проходы и циклы, остающиеся внутри своего процесса, проходы очистки по срокам хранения и outbox

Omniflow снимает работу с пути запроса двумя разными способами, и эта разница важна, когда что-то застревает.

Работа, которая **касается одной записи** и должна пережить перезапуск, попадает в долговременную очередь: это River, работающий на той же базе PostgreSQL, причём строка задачи вставляется в той же транзакции, что и изменение состояния, которое эту задачу требует. Работа, которая представляет собой **проход по всей таблице** — просрочка заказов, зачистка корзин, обновление блок-листов, удаление вчерашних сессий, — выполняется как периодическая задача River на воркере, поэтому она выполняется один раз за интервал на всю установку, сколько бы реплик воркера ни работало. Несколько циклов, которые касаются только своего процесса, остаются обычными внутрипроцессными таймерами.

Задачи выполняет только процесс `worker`. API и бот каждый создают клиент River для *вставки* и никогда не регистрируют воркера: закрытие заказа в любом из них — вебхук провайдера в API, оплата Telegram Stars или заказ, покрытый кошельком, в боте — вставляет задание выдачи доступа в платёжной транзакции.

## Долговременная очередь

Существует три очереди с фиксированной, вкомпилированной параллельностью:

| Очередь | Максимум воркеров | Что несёт |
| - | - | - |
| `critical` | 16 | `fulfillment` и `goods_delivery` — к моменту запуска любой из них клиент уже заплатил |
| `default` | 8 | `periodic_task` — [периодические проходы](#периодические-проходы-и-внутрипроцессные-циклы) |
| `bulk` | 4 | Сегодня ничего |

Итого 28 задач одновременно, и пул подключений воркера к базе рассчитан под это — см. [Развёртывание](https://omniflow-vpn.mintlify.site/ru/operations/deployment.md#подключения-к-базе-данных).

### Виды задач

| Вид | Аргументы | Очередь | Максимум попыток | Уникальность |
| - | - | - | - | - |
| `fulfillment` | `{"operationId": …}` | `critical` | 12 | Уникальна по аргументам |
| `goods_delivery` | `{"orderId": …}` | `critical` | 12 | Уникальна по аргументам |
| `periodic_task` | `{"task": "<name>"}` | `default` | 1 | Уникальна по аргументам, пока находится в состоянии available, pending, retryable, running или scheduled |

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

### Почему задача вставляется в оплачивающей транзакции

Правило, которому следует код, таково: долговременная задача ставится в очередь в той же транзакции PostgreSQL, что и изменение состояния, которое её требует. Расчёт заказа и вставка его задачи выдачи доступа коммитятся вместе — именно это делает истинным утверждение «у оплаченного заказа всегда есть ожидающая его задача». Альтернатива — рассчитать, зафиксировать, а потом поставить в очередь — оставляет окно, в котором процесс может умереть между двумя действиями, и тогда остаются заплативший клиент и очередь, которая о нём никогда не слышала.

Это же свойство делает безопасными повторы. Операция выдачи доступа несёт уникальный `idempotency_key`, поэтому доставка River «хотя бы один раз» не может применить её дважды, а повтор из панели ставит задачу в очередь под тем же самым ключом, а не создаёт вторую операцию.

### Схема очереди поставляется вместе с миграциями

River держит собственные таблицы (`river_job`, `river_leader`, `river_queue`, `river_notification`, `river_migration`). Их создаёт миграция Atlas `20260905130000_river_queue.sql`: это собственная линия миграций `main` подключённой версии River, скопированная без изменений, за которой следуют строки `river_migration`, которые записал бы `rivermigrate` самого River. Поэтому сервис `migrate` в compose оставляет базу, на которой воркер запускается, а API и бот могут вставлять задачи, — без шагов вне описанного пути развёртывания.

Установка, которая уже создала эти таблицы вручную через CLI River, сохраняет их: миграция пропускает всё своё тело, если `river_migration` уже существует.

Интеграционный набор после каждого прогона миграций проверяет, что `rivermigrate` не видит неприменённых версий River, поэтому обновление River до релиза с новой версией схемы падает в CI, пока не закоммичена соответствующая миграция.

## Задачи выдачи доступа

Задача `fulfillment` применяет одно желаемое состояние права на услугу к Remnawave. Это единственный путь, которым Omniflow пишет в панель, — поэтому каждый повтор, каждое удержание и каждый отказ ниже записывается в запись, которую оператор может прочитать потом. Коммерческая сторона этого описана в [Права на услугу и выдача доступа](https://omniflow-vpn.mintlify.site/ru/commerce/fulfillment.md).

Воркер открывает сериализуемую транзакцию, блокирует строку операции, а затем:

```mermaid theme={null}
stateDiagram-v2
    [*] --> Locked
    Locked --> Done: статус succeeded или cancelled
    Locked --> Failed: желаемое состояние не разбирается
    Locked --> Snoozed: effectiveAt в будущем
    Locked --> Held: активен режим обслуживания
    Locked --> Running: иначе
    Running --> Succeeded: Remnawave принял
    Running --> Retrying: Remnawave отказал
    Retrying --> Running: после отсрочки
    Held --> Running: через минуту, если обслуживание снято
```

* **Неразбираемое желаемое состояние** помечает операцию как `failed` с кодом `invalid_desired_state`, уведомляет операторов и отменяет задачу. Повтор провалился бы точно так же двенадцать раз подряд.
* **`effectiveAt` в будущем** откладывает задачу до этого момента, вместо того чтобы жечь попытки на запланированное изменение, срок которого ещё не наступил.
* **Режим обслуживания** паркует операцию в состоянии `retrying` с `last_error_code = maintenance_active` и откладывает задачу на минуту. Счётчик попыток при этом сначала уменьшается обратно, поэтому удержание не стоит ни одной единицы бюджета повторов, сколько бы окно ни длилось. См. [Режим обслуживания](https://omniflow-vpn.mintlify.site/ru/operations/maintenance-mode.md).
* **Сбой провайдера** сохраняет классифицированный код ошибки и `next_attempt_at`, вычисленный как `5s × 2^attempt` с ограничением на восьмой попытке, — так что отсрочка идёт 10 с, 20 с, 40 с … вплоть до 1280 секунд между попытками.

Поддерживаемые операции — это `create`, `reconcile`, `extend`, `set_limits`, `set_squads`, `enable`, `disable` и `reset_traffic`. Всё остальное отменяет задачу, а не повторяет её.

После **трёх** неудачных попыток — и сразу же при постоянном сбое — воркер ставит в очередь уведомление оператору в топик `fulfillment_failure`, дедуплицированное по `operation:<id>`, так что одно мучающееся право на услугу порождает одно сообщение, сколько бы раз оно ни повторялось. Уведомление несёт идентификатор права на услугу, операцию, классифицированный код ошибки и статус — и никогда тело ответа Remnawave. См. [Уведомления оператора](https://omniflow-vpn.mintlify.site/ru/operations/operator-notifications.md).

Операция `reconcile` дополнительно сравнивает локальное желаемое состояние с тем, что сообщает Remnawave, и записывает любое несовпадение как расхождение права на услугу (`missing_remote`, `expiry`, `traffic`, `device_limit`, `squads`, `status`). Успешное применение закрывает открытые расхождения по этому праву на услугу.

## Задачи доставки цифровых товаров

Задача `goods_delivery` передаёт один оплаченный заказ магазина его провайдеру. Гарантия, которую она обязана держать, — что оплаченный заказ отправляется **не более одного раза**, и форма воркера и есть эта гарантия:

1. Короткая транзакция блокирует строку доставки, отказывает во всём уже завершённом и заявляет следующую попытку — увеличивая счётчик и отодвигая крайний срок повтора *до* того, как будет вызван провайдер. Эта транзакция коммитится.
2. Вызов провайдера происходит вне какой-либо транзакции. Умерший здесь воркер оставляет строку, уже помеченную как отправленная, со сдвинутым сроком, поэтому ничто не подхватит её немедленно и не купит товар повторно.
3. Вторая транзакция записывает исход.

Источник истины о том, что причитается, — строка `goods_deliveries`, а не очередь. Двенадцать попыток River покрывают сбои, которые до провайдера так и не дошли, — короткий отказ базы данных, потерянное соединение, — тогда как собственный счётчик попыток строки управляет тем, сколько раз провайдера действительно спрашивают. Поэтому больший бюджет River не может превратиться в лишние покупки. Адаптер, возвращающий ошибку вместо классифицированного исхода, трактуется как неоднозначный сбой с кодом `adapter_error`, потому что «мы не знаем, состоялась ли покупка» — это не то же самое, что «она не состоялась».

Этот воркер существует, только когда на воркере задан `APP_DATA_ENCRYPTION_KEY`, поскольку без него учётные данные провайдера не распечатать. Без ключа воркер логирует `digital goods delivery disabled` и выполняет всё остальное.

> [!NOTE]
>
> Доставка цифровых товаров не инструментирована: она не открывает трассировочный спан и не пишет метрику задачи, поэтому `omniflow_jobs_total{kind="goods_delivery"}` не появляется никогда. Следите за ней через экраны магазина в панели, а не через Prometheus. См. [Наблюдаемость](https://omniflow-vpn.mintlify.site/ru/operations/observability.md).

## Очередь недоставленных задач, повторы и отмена

Операторских поверхностей две, и показывают они разное. Берите ту, которая нужна.

### Строки River, по bearer-токену

`GET /v1/admin/jobs` перечисляет реальные строки очереди и является единственным местом, где видна задача, которая так и не породила операцию выдачи доступа.

| Параметр | Значения |
| - | - |
| `state` | `dead_letter` (по умолчанию; discarded плюс cancelled), `retryable`, `running`, `available`, `scheduled`, `completed`. На всё остальное отвечает `400 invalid_state` |
| `limit` | 1–200, по умолчанию 50 |

Каждая строка возвращает идентификатор, вид, очередь, состояние, номер попытки и максимум попыток, время создания и последней попытки, счётчик ошибок и **последнюю ошибку, пропущенную через редактор**. Аргументы задачи резюмируются, а не воспроизводятся дословно, потому что они несут идентификаторы, место которым — за теми записями, на которые они указывают.

* `POST /v1/admin/jobs/{jobID}/retry` возвращает задачу в состояние available.
* `POST /v1/admin/jobs/{jobID}/cancel` отменяет её.

Любой из них отвечает `422 job_mutation_rejected`, когда текущее состояние задачи запрещает изменение, и оба отвечают `503 jobs_unavailable` в процессе, у которого нет клиента задач. Все три маршрута аутентифицируются через `APP_OPERATOR_TOKEN`; см. [Административный API](https://omniflow-vpn.mintlify.site/ru/integrations/api.md).

### Операции выдачи доступа, в панели

`/admin/system` → **Jobs** (право доступа `system.read`) жёстко отфильтровано по неудавшимся операциям выдачи доступа — это и есть то представление недоставленных задач, которое оператору действительно нужно: время создания, операция, число попыток, код ошибки и идентификатор корреляции. Оно никогда не показывает ни полезную нагрузку провайдера, ни тело ответа Remnawave: статус, классификация и идентификатор корреляции — это ровно то, что нужно, чтобы решить, повторять ли, и ничто из этого не может вынести ссылку на подписку на экран оператора.

С правом `system.write`:

* **Retry** ставит ту же операцию в очередь заново под **тем же ключом идемпотентности**, так что воркер по-прежнему выполнит её ровно один раз.
* **Cancel** бросает операцию, которая не завершилась успехом.

**Успешную** операцию нельзя ни повторить, ни отменить, и этот отказ — `409 state_conflict`, обеспеченный в SQL, а не проверяемый в Go. Отзыв завершённой выдачи доступа кликом в панели оставил бы право на услугу и Remnawave в разногласии без записи о том, почему.

Каждая мутация в панели требует заголовка `X-Operator-Reason` и пишет событие аудита — `panel.fulfillment.retried` или `panel.fulfillment.cancelled`, категория `system`. История попыток по операции доступна по `GET /v1/panel/system/jobs/{operationID}/history` со статусом, идентификатором корреляции, кодом ошибки и временем каждой попытки; сохранённые сводки запроса и ответа намеренно не возвращаются никогда.

| Маршрут | Право доступа |
| - | - |
| `GET /v1/panel/system/jobs` | `system.read` |
| `GET /v1/panel/system/jobs/{operationID}/history` | `system.read` |
| `POST /v1/panel/system/jobs/{operationID}/retry` | `system.write` |
| `POST /v1/panel/system/jobs/{operationID}/cancel` | `system.write` |

## Периодические проходы и внутрипроцессные циклы

### Периодические проходы на воркере

Каждый проход по всей таблице, меняющий состояние, — это периодическая задача River вида `periodic_task`, которую выполняет воркер. Ставит их в очередь только выбранный River лидер среди работающих клиентов воркера: каждый проход выполняется один раз при старте лидера, затем раз в интервал, поэтому он выполняется один раз за интервал на всю установку, сколько бы реплик воркера ни работало. Уникальность не даёт поставить проход, переживший свой интервал, в очередь второй раз рядом с ним самим; завершившийся проход её освобождает.

У прохода одна попытка. Сбой логируется как `periodic task failed` с именем задачи, задача завершается в состоянии discarded — её видно в `GET /v1/admin/jobs?state=dead_letter`, — а следующий интервал запускает новый проход по тем же строкам. Выполнение прерывается по истечении интервала, если в таблице не указано иное.

| Задача | Интервал | Таймаут выполнения | Что делает |
| - | - | - | - |
| `commerce_maintenance` | 1 мин | интервал | Просрочивает ожидающие заказы, гасит истёкшие начисления кошелька транзакцией двойной записи `expiration`, зачищает сохранённые корзины |
| `payment_reconcile` | 1 мин | интервал | Переопрашивает до 100 незавершённых платёжных намерений |
| `renewal` | 1 мин | 5 мин | Списывает наступившие продления пакетами по 100 |
| `fulfillment_reconcile` | 15 мин | интервал | Создаёт операции `reconcile` не более чем для 500 прав на услугу с ключом `reconcile:<entitlementID>:<15-minute bucket>` |
| `fulfillment_revive` | 5 мин | интервал | Заново ставит в очередь операции выдачи доступа, чья задача так и не запустилась или была отброшена |
| `lifecycle_sweep` | 5 мин | интервал | Просрочивает подарки и персональные предложения, обновляет до 20 источников блок-листов |
| `anomaly_evaluation` | 15 мин | интервал | Вычисляет правила аномалий |
| `campaign_expansion` | 1 мин | 5 мин | Раскрывает аудитории кампаний, пакет 5 000 |
| `bulk_actions` | 15 с | 5 мин | Применяет предпросмотренные массовые операции по одному элементу, пакет 50 |
| `retention` | `APP_RETENTION_INTERVAL_HOURS` (1 ч) | интервал | Семь шагов очистки, ниже |
| `goods_delivery_schedule` | 30 с | интервал | Ставит в очередь до 200 доставок, срок которых наступил |
| `backup` | `APP_BACKUP_INTERVAL_HOURS` (24 ч) | нет | Зашифрованный `pg_dump` плюс удаление старых копий |

Бакетированный ключ идемпотентности у прохода сверки выдачи доступа стоит понять: два прохода внутри одного и того же пятнадцатиминутного бакета дают один и тот же ключ, поэтому перезапущенный воркер не может создать вторую операцию сверки для того же права на услугу в том же окне.

`goods_delivery_schedule` регистрируется, только когда у воркера есть `APP_DATA_ENCRYPTION_KEY`, а `backup` — только когда включено резервное копирование по расписанию. Обновлению блок-листов, вычислению аномалий и массовым операциям тоже нужен `APP_DATA_ENCRYPTION_KEY` на воркере, потому что они читают запечатанные учётные данные; просрочка подарков и предложений работает и без него. При нескольких репликах воркера резервная копия оказывается на той реплике, которая выполнила проход, поэтому `APP_BACKUP_DIR` должен быть общим томом — см. [Резервное копирование и восстановление](https://omniflow-vpn.mintlify.site/ru/operations/backup-restore.md).

### Циклы, которые остаются внутри своего процесса

Это горутины-таймеры, и каждая реплика запускает собственную копию. Ни один из них не меняет общее состояние так, чтобы вторая копия могла его продублировать:

| Цикл | Процесс | Интервал | Настраивается | Что делает |
| - | - | - | - | - |
| Контроллер режима обслуживания | api | `APP_MAINTENANCE_PROBE_SECONDS` (30 с) | да | Снимает пробы состояния зависимостей и двигает переключатель обслуживания. Каждая реплика снимает собственные пробы и читает текущее состояние, прежде чем его менять, поэтому вторая реплика застаёт переход уже сделанным |
| Телеметрический сигнал | api | 24 ч плюс один при старте | только вкл/выкл | Отправляет `installation.heartbeat` |
| Очистка коллектора | api (режим коллектора) | 24 ч | нет | Удаляет собранную телеметрию старше 180 дней; удаление идёт по возрасту, поэтому повтор безвреден |
| Членство в каналах | bot | 10 мин | нет | Перепроверяет обязательное членство в каналах, пакет 100 |
| Уведомитель операторов | bot | 10 с | нет | Привязывает топики форума, затем разбирает до 50 ожидающих уведомлений |
| Уведомления жизненного цикла клиентов | bot | 15 мин | нет | Тестовые отправки оператора, уведомления о неудачных списаниях, затем уведомления жизненного цикла каждого достижимого клиента, по 200 на страницу |
| Доставка рассылок и объявлений | bot | непрерывно, пока есть очередь; 30 с в простое | нет | 200 получателей рассылок и 200 объявлений за пачку в темпе `APP_TELEGRAM_DELIVERY_RATE` (25/с) |
| Push ответов поддержки | bot | 30 с | нет | Отправляет до 200 ожидающих ответов операторов |

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

> [!NOTE]
>
> Поддерживаемая топология — сколько угодно реплик `api` и `worker` и ровно один `bot`. См. [Развёртывание](https://omniflow-vpn.mintlify.site/ru/operations/deployment.md#поддерживаемая-топология).

## Проходы очистки по срокам хранения

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

| Шаг | Что удаляет | Окно |
| - | - | - |
| `bot_sessions` | Истёкшие сессии бота | `expires_at` на уровне строки |
| `bot_shop_reviews` | Экраны проверки покупки в магазине старше суток | `expires_at` на уровне строки |
| `bot_checkout_sessions` | Истёкшие сессии оформления заказа, так и не ставшие заказом | `expires_at` на уровне строки |
| `provider_webhook_events` | Строки приёма вебхуков после их отметки хранения | `retain_until` на уровне строки |
| `support_attachments` | Истёкшие строки вложений, а также файлы за ними | на уровне строки |
| `outbox_events` | Опубликованные события | `APP_RETENTION_OUTBOX_DAYS` (7) |
| `telemetry_events` | Собранная телеметрия, по `received_at` | `APP_RETENTION_TELEMETRY_DAYS` (30) |
| `entitlement_drifts` | Расхождения, чей статус не `open`, по `resolved_at` | `APP_RETENTION_DRIFT_DAYS` (90) |

Шаг, который удалил строки, логирует `retention sweep removed rows` с таблицей и количеством.

Шаг с вложениями — тот, у которого есть особенность. Файлы вложений именуются по хешу их содержимого, поэтому один файл может стоять за несколькими строками, и только код, удаляющий строки, может определить, на какие файлы больше ничто не ссылается. Проход очистки читает ключи кандидатов *внутри* удаляющей транзакции, удаляет истёкшие строки, а затем убирает только те файлы, на которые не ссылается ни одна уцелевшая строка. Файл, который убрать не удалось, логируется и пропускается, а не проваливает весь проход. Без подключённого уборщика вложений — то есть в установке только с Telegram — воркер всё равно удаляет строки, потому что эти вложения являются ссылками на байты, которые держит Telegram.

> [!WARNING]
>
> Если `APP_SUPPORT_ATTACHMENT_DIR` у воркера отличается от такового у API, очистка удалит строки и оставит байты на диске навсегда. Задайте обоим один и тот же общий путь и дайте ему том — см. [Развёртывание](https://omniflow-vpn.mintlify.site/ru/operations/deployment.md).

Очистку по срокам хранения выполняет только воркер. Установка, где работают лишь `api` и `bot`, не убирает ничего из перечисленного.

## Транзакционный outbox

Три доменных события записываются в `outbox_events` в той же транзакции, что и изменение, которое они описывают, — так что у зафиксированного изменения состояния всегда есть его событие, а у откатившегося его никогда нет:

| Топик | Когда записывается |
| - | - |
| `order.paid` | Заказ рассчитан и создано его право на услугу |
| `wallet.credited` | Рассчитано пополнение кошелька |
| `addon.purchased` | Заказ дополнения рассчитан на право на услугу |

Отставание читают две поверхности. `GET /v1/admin/outbox` возвращает `{pendingCount, oldestAgeSeconds}`; вкладка **System → Outbox** в панели (право доступа `system.read`) перечисляет топик и возраст каждого ожидающего события и ничего больше, потому что полезная нагрузка outbox — это доменное событие, которое может называть клиента и сумму, а вопросу «работает ли публикатор» не нужно ни то, ни другое.

> [!WARNING]
>
> **Outbox никто не рассылает.** Ни один код в Omniflow не выставляет `outbox_events.published_at`, и нет ни процесса-публикатора, ни подписчика, ни цели доставки. Отсюда прямо следуют два вывода. Число ожидающих событий растёт в течение всей жизни установки и никогда не падает, поэтому растущее число здесь — нормальное установившееся состояние, а не инцидент. И поскольку шаг очистки удаляет только строки, у которых `published_at` не null, `APP_RETENTION_OUTBOX_DAYS` не удаляет вообще ничего — таблица растёт неограниченно вместе со счётчиком.

`omniflow_outbox_pending_events` объявлена, но никогда не выставляется, поэтому этот датчик не является альтернативным представлением того же самого. Если хотите следить за размером таблицы, читайте счётчик из административного маршрута или из вкладки в панели.

## Связанные страницы

> [card] [Права на услугу и выдача доступа](https://omniflow-vpn.mintlify.site/ru/commerce/fulfillment.md)
>
> Что такое операция выдачи доступа и как работает восстановление после расхождений.

> [card] [Наблюдаемость](https://omniflow-vpn.mintlify.site/ru/operations/observability.md)
>
> Какие из этих задач дают метрику, а какие нет.

> [card] [Режим обслуживания](https://omniflow-vpn.mintlify.site/ru/operations/maintenance-mode.md)
>
> Почему задача выдачи доступа может быть отложена, а не провалена.

> [card] [Операции администратора](https://omniflow-vpn.mintlify.site/ru/operations/admin-operations.md)
>
> Остальные экраны System: вебхуки, расхождения и диагностика.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.