Skip to main content
Omniflow снимает работу с пути запроса двумя разными способами, и эта разница важна, когда что-то застревает. Работа, которая касается одной записи и должна пережить перезапуск, попадает в долговременную очередь: это River, работающий на той же базе PostgreSQL, причём строка задачи вставляется в той же транзакции, что и изменение состояния, которое эту задачу требует. Работа, которая представляет собой проход по всей таблице — просрочка заказов, зачистка корзин, обновление блок-листов, удаление вчерашних сессий, — выполняется как периодическая задача River на воркере, поэтому она выполняется один раз за интервал на всю установку, сколько бы реплик воркера ни работало. Несколько циклов, которые касаются только своего процесса, остаются обычными внутрипроцессными таймерами. Задачи выполняет только процесс worker. API и бот каждый создают клиент River для вставки и никогда не регистрируют воркера: закрытие заказа в любом из них — вебхук провайдера в API, оплата Telegram Stars или заказ, покрытый кошельком, в боте — вставляет задание выдачи доступа в платёжной транзакции.

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

Существует три очереди с фиксированной, вкомпилированной параллельностью: Итого 28 задач одновременно, и пул подключений воркера к базе рассчитан под это — см. Развёртывание.

​
Виды задач

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

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

Правило, которому следует код, таково: долговременная задача ставится в очередь в той же транзакции 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 пишет в панель, — поэтому каждый повтор, каждое удержание и каждый отказ ниже записывается в запись, которую оператор может прочитать потом. Коммерческая сторона этого описана в Права на услугу и выдача доступа. Воркер открывает сериализуемую транзакцию, блокирует строку операции, а затем:
  • Неразбираемое желаемое состояние помечает операцию как failed с кодом invalid_desired_state, уведомляет операторов и отменяет задачу. Повтор провалился бы точно так же двенадцать раз подряд.
  • effectiveAt в будущем откладывает задачу до этого момента, вместо того чтобы жечь попытки на запланированное изменение, срок которого ещё не наступил.
  • Режим обслуживания паркует операцию в состоянии retrying с last_error_code = maintenance_active и откладывает задачу на минуту. Счётчик попыток при этом сначала уменьшается обратно, поэтому удержание не стоит ни одной единицы бюджета повторов, сколько бы окно ни длилось. См. Режим обслуживания.
  • Сбой провайдера сохраняет классифицированный код ошибки и 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. См. Уведомления оператора. Операция 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 и выполняет всё остальное.
Доставка цифровых товаров не инструментирована: она не открывает трассировочный спан и не пишет метрику задачи, поэтому omniflow_jobs_total{kind="goods_delivery"} не появляется никогда. Следите за ней через экраны магазина в панели, а не через Prometheus. См. Наблюдаемость.

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

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

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

GET /v1/admin/jobs перечисляет реальные строки очереди и является единственным местом, где видна задача, которая так и не породила операцию выдачи доступа. Каждая строка возвращает идентификатор, вид, очередь, состояние, номер попытки и максимум попыток, время создания и последней попытки, счётчик ошибок и последнюю ошибку, пропущенную через редактор. Аргументы задачи резюмируются, а не воспроизводятся дословно, потому что они несут идентификаторы, место которым — за теми записями, на которые они указывают.
  • POST /v1/admin/jobs/{jobID}/retry возвращает задачу в состояние available.
  • POST /v1/admin/jobs/{jobID}/cancel отменяет её.
Любой из них отвечает 422 job_mutation_rejected, когда текущее состояние задачи запрещает изменение, и оба отвечают 503 jobs_unavailable в процессе, у которого нет клиента задач. Все три маршрута аутентифицируются через APP_OPERATOR_TOKEN; см. Административный API.

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

/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 со статусом, идентификатором корреляции, кодом ошибки и временем каждой попытки; сохранённые сводки запроса и ответа намеренно не возвращаются никогда.

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

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

Каждый проход по всей таблице, меняющий состояние, — это периодическая задача River вида periodic_task, которую выполняет воркер. Ставит их в очередь только выбранный River лидер среди работающих клиентов воркера: каждый проход выполняется один раз при старте лидера, затем раз в интервал, поэтому он выполняется один раз за интервал на всю установку, сколько бы реплик воркера ни работало. Уникальность не даёт поставить проход, переживший свой интервал, в очередь второй раз рядом с ним самим; завершившийся проход её освобождает. У прохода одна попытка. Сбой логируется как periodic task failed с именем задачи, задача завершается в состоянии discarded — её видно в GET /v1/admin/jobs?state=dead_letter, — а следующий интервал запускает новый проход по тем же строкам. Выполнение прерывается по истечении интервала, если в таблице не указано иное. Бакетированный ключ идемпотентности у прохода сверки выдачи доступа стоит понять: два прохода внутри одного и того же пятнадцатиминутного бакета дают один и тот же ключ, поэтому перезапущенный воркер не может создать вторую операцию сверки для того же права на услугу в том же окне. goods_delivery_schedule регистрируется, только когда у воркера есть APP_DATA_ENCRYPTION_KEY, а backup — только когда включено резервное копирование по расписанию. Обновлению блок-листов, вычислению аномалий и массовым операциям тоже нужен APP_DATA_ENCRYPTION_KEY на воркере, потому что они читают запечатанные учётные данные; просрочка подарков и предложений работает и без него. При нескольких репликах воркера резервная копия оказывается на той реплике, которая выполнила проход, поэтому APP_BACKUP_DIR должен быть общим томом — см. Резервное копирование и восстановление.

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

Это горутины-таймеры, и каждая реплика запускает собственную копию. Ни один из них не меняет общее состояние так, чтобы вторая копия могла его продублировать: Циклы бота безопасны, потому что бот работает ровно в одном экземпляре: Telegram доставляет обновления одному потребителю. Цикл членства в каналах отключается, когда не настроен ни один проверяющий членство.
Поддерживаемая топология — сколько угодно реплик api и worker и ровно один bot. См. Развёртывание.

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

Ежечасный проход очистки удаляет одноразовые операционные данные. Он никогда не трогает финансовую историю, историю аудита, согласий, идентичностей и прав на услугу — она доступна только на добавление в течение всей жизни установки. Восемь независимых шагов выполняются по порядку, и провалившийся шаг логируется и пропускается, чтобы одна заблокированная таблица не могла остановить очистку целиком. Шаг, который удалил строки, логирует retention sweep removed rows с таблицей и количеством. Шаг с вложениями — тот, у которого есть особенность. Файлы вложений именуются по хешу их содержимого, поэтому один файл может стоять за несколькими строками, и только код, удаляющий строки, может определить, на какие файлы больше ничто не ссылается. Проход очистки читает ключи кандидатов внутри удаляющей транзакции, удаляет истёкшие строки, а затем убирает только те файлы, на которые не ссылается ни одна уцелевшая строка. Файл, который убрать не удалось, логируется и пропускается, а не проваливает весь проход. Без подключённого уборщика вложений — то есть в установке только с Telegram — воркер всё равно удаляет строки, потому что эти вложения являются ссылками на байты, которые держит Telegram.
Если APP_SUPPORT_ATTACHMENT_DIR у воркера отличается от такового у API, очистка удалит строки и оставит байты на диске навсегда. Задайте обоим один и тот же общий путь и дайте ему том — см. Развёртывание.
Очистку по срокам хранения выполняет только воркер. Установка, где работают лишь api и bot, не убирает ничего из перечисленного.

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

Три доменных события записываются в outbox_events в той же транзакции, что и изменение, которое они описывают, — так что у зафиксированного изменения состояния всегда есть его событие, а у откатившегося его никогда нет: Отставание читают две поверхности. GET /v1/admin/outbox возвращает {pendingCount, oldestAgeSeconds}; вкладка System → Outbox в панели (право доступа system.read) перечисляет топик и возраст каждого ожидающего события и ничего больше, потому что полезная нагрузка outbox — это доменное событие, которое может называть клиента и сумму, а вопросу «работает ли публикатор» не нужно ни то, ни другое.
Outbox никто не рассылает. Ни один код в Omniflow не выставляет outbox_events.published_at, и нет ни процесса-публикатора, ни подписчика, ни цели доставки. Отсюда прямо следуют два вывода. Число ожидающих событий растёт в течение всей жизни установки и никогда не падает, поэтому растущее число здесь — нормальное установившееся состояние, а не инцидент. И поскольку шаг очистки удаляет только строки, у которых published_at не null, APP_RETENTION_OUTBOX_DAYS не удаляет вообще ничего — таблица растёт неограниченно вместе со счётчиком.
omniflow_outbox_pending_events объявлена, но никогда не выставляется, поэтому этот датчик не является альтернативным представлением того же самого. Если хотите следить за размером таблицы, читайте счётчик из административного маршрута или из вкладки в панели.

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

Права на услугу и выдача доступа

Что такое операция выдачи доступа и как работает восстановление после расхождений.

Наблюдаемость

Какие из этих задач дают метрику, а какие нет.

Режим обслуживания

Почему задача выдачи доступа может быть отложена, а не провалена.

Операции администратора

Остальные экраны System: вебхуки, расхождения и диагностика.