# recognition-job-exchange Specification ## Purpose Единственная точка обмена ядра с сервисами распознавания — таблица `recognition_job` (DECISION-4): outbox публикует запросы в NATS, inbox принимает подтверждения и результаты идемпотентно, потеря сообщений без JetStream компенсируется повторной публикацией и таймаутами, файл результата регистрируется в `s3_object`. ## Requirements ### Requirement: Outbox публикует задания из `queued` Фоновый публикатор SHALL с периодом `OUTBOX_POLL_SECONDS` (по умолчанию 5) выбирать задания `status = queued` (с блокировкой `FOR UPDATE SKIP LOCKED`), формировать сообщение по `recognition-request.schema.json` v1.0 из `recognition_job ⋈ stored_file ⋈ s3_object` и публиковать в subject по `pipeline` с заголовком `Nats-Msg-Id = :`. После успешной публикации задание SHALL получить `status = published`, `published_at`, `request_msg_id`, `request_subject`; при ошибке публикации задание SHALL остаться `queued`. #### Scenario: Публикация запроса парсеру - **WHEN** в `recognition_job` вставлена строка `queued` с `pipeline = tabular`, связанная со `stored_file` (XLSX) и его `s3_object` - **THEN** в течение 10 с в `recognition.tabular.request` появляется сообщение, проходящее схему запроса, с `job_id`, `file_id`, `s3.key` исходника, `reply_deadline`, `attempt = 1`, а строка переходит в `published` с заполненным `request_msg_id` #### Scenario: NATS недоступен - **WHEN** публикация завершается ошибкой соединения - **THEN** задание остаётся `queued`, ошибка залогирована, следующий цикл повторяет публикацию ### Requirement: Ack сервиса переводит задание в `processing` Получив в `recognition.result` конверт со `status = processing` для задания в `published`, ядро SHALL установить `status = processing`, `started_at`, `service_name`, `service_version`. Повторный ack для задания в `processing` SHALL игнорироваться; ack для задания в другом статусе SHALL логироваться и не менять задание. #### Scenario: Сервис подтвердил - **WHEN** сервис опубликовал конверт `processing` с `job_id` опубликованного задания - **THEN** задание имеет `status = processing`, `started_at` заполнен, `service_name = tabular-parser` ### Requirement: Запрос без ack публикуется повторно, затем — `dlq` Если задание находится в `published` дольше `ACK_TIMEOUT_SECONDS` (по умолчанию 30) без ack, публикатор SHALL опубликовать запрос повторно с тем же `Nats-Msg-Id` и тем же `attempt`, обновив `published_at` и увеличив счётчик `metrics.republish_count`. После `OUTBOX_MAX_REPUBLISH` (по умолчанию 3) повторов без ack задание SHALL получить `status = dlq`, `error_code = NO_ACK`, `finished_at`. #### Scenario: Сервис недоступен - **WHEN** ни один инстанс `tabular-parser` не подписан, задание опубликовано - **THEN** запрос публикуется повторно примерно каждые 30 с; после трёх повторов задание имеет `status = dlq`, `error_code = NO_ACK`, `request_msg_id` не изменился #### Scenario: Сервис появился после первого повтора - **WHEN** сервис подписался между первой и второй публикацией и прислал ack - **THEN** задание переходит в `processing`, дальнейшие повторы не выполняются ### Requirement: Inbox принимает результат идемпотентно Для конверта со `status ∈ {done, needs_check, error}` ядро SHALL валидировать его по `recognition-result-message.schema.json`, найти задание по `job_id` и: если `result_msg_id` задания уже равен `Nats-Msg-Id` сообщения — ничего не менять; иначе зарегистрировать файл результата в `s3_object` (`bucket_role = results`, `created_by = `; только при наличии `result_s3`), обновить задание (`status`, `service_name`, `service_version`, `finished_at`, `metrics`, `error_code`, `error_message`, `result_schema_version`, `result_s3_object_id`, `result_msg_id`) и зафиксировать транзакцию. Неизвестный `job_id` SHALL быть залогирован. Результат для задания, уже находящегося в финальном статусе (`done`, `error`, `timeout`, `dlq`), SHALL логироваться и не применяться. #### Scenario: Успешный результат - **WHEN** сервис публикует конверт `status = done` с `result_s3 = {arm-ka, results///result.json}` и `Nats-Msg-Id = result::1` - **THEN** задание получает `status = done`, `result_msg_id = result::1`, `result_s3_object_id` указывает на новую строку `s3_object` с этим ключом, `finished_at` заполнен #### Scenario: Повторная доставка - **WHEN** то же сообщение получено второй раз - **THEN** строка `recognition_job` и `s3_object` не меняются #### Scenario: Ошибка сервиса - **WHEN** конверт содержит `status = error`, `error = {NOT_IMPLEMENTED, …}` без `result_s3` - **THEN** задание получает `status = error`, `error_code = NOT_IMPLEMENTED`, `error_message` из конверта, `result_s3_object_id` пуст #### Scenario: Результат пришёл после таймаута - **WHEN** задание уже `timeout`, и приходит конверт `done` - **THEN** задание не меняется, событие залогировано ### Requirement: Таймаут ожидания результата Фоновый вахтёр SHALL с периодом 30 с переводить задания `processing` и `published` с `reply_deadline < now()` в `status = timeout`, `error_code = TIMEOUT`, `finished_at = now()`. Replay задания (новая попытка, `attempt + 1`) — вне Среза 0. #### Scenario: Сервис взял, но не ответил - **WHEN** задание в `processing` и `reply_deadline` прошёл без результата - **THEN** не позднее чем через 30 с задание имеет `status = timeout` ### Requirement: Соединение с NATS по шаблону, без JetStream Ядро SHALL подключаться к NATS с nkeys-seed и JWT, если заданы `OPENBAO__NATS__NATS_SEED/NATS_JWT`, и без учётных данных, если они пусты; подписываться на `recognition.result` в queue group `arm-core`; переподключаться автоматически. Stream'ы и consumer'ы JetStream MUST NOT требоваться. Фоновые задачи SHALL запускаться при `NATS_ENABLED=true`. #### Scenario: Dev-окружение compose - **WHEN** ядро стартует с `DEBUG=true`, `NATS_ENABLED=true` и пустыми учётными данными против NATS без JetStream - **THEN** подписка активна, `/healthcheck` отдаёт `nats = ok`, в логах нет ошибок подключения #### Scenario: Два инстанса ядра - **WHEN** запущены два инстанса с `NATS_ENABLED=true` и опубликован один результат - **THEN** результат обработан ровно одним инстансом (queue group), строка задания изменена один раз #### Scenario: NATS выключен - **WHEN** `NATS_ENABLED=false` - **THEN** приложение стартует, фоновые задачи не запускаются, `/healthcheck` отвечает 200 с `nats = off`