199 lines
18 KiB
Markdown
199 lines
18 KiB
Markdown
# 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 = <job_id>:<attempt>`. После успешной публикации задание 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 с в `files.tabular_uploaded` появляется сообщение, проходящее схему запроса, с `job_id`,
|
||
`file_id`, `s3.key` исходника, `reply_deadline`, `attempt = 1`, а строка переходит в `published` с заполненным `request_msg_id`
|
||
|
||
#### Scenario: NATS недоступен
|
||
- **WHEN** публикация завершается ошибкой соединения
|
||
- **THEN** задание остаётся `queued`, ошибка залогирована, следующий цикл повторяет публикацию
|
||
|
||
### Requirement: Ack сервиса переводит задание в `processing`
|
||
Получив в `files.tabular_parsed` конверт со `status = processing` для задания в `published`, ядро SHALL установить
|
||
`status = processing`, `started_at`, `service_name`, `service_version` и перевести файл задания в
|
||
`stored_file.status = PROCESSING`, если `current_recognition_job_id` файла равен этому заданию. Повторный ack для
|
||
задания в `processing` SHALL игнорироваться; ack для задания в другом статусе SHALL логироваться и не менять задание.
|
||
|
||
#### Scenario: Сервис подтвердил
|
||
- **WHEN** сервис опубликовал конверт `processing` с `job_id` опубликованного задания
|
||
- **THEN** задание имеет `status = processing`, `started_at` заполнен, `service_name = tabular-parser`, файл — `PROCESSING`
|
||
|
||
### 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 = <service.name>`; только при наличии `result_s3`), обновить задание (`status`, `service_name`,
|
||
`service_version`, `finished_at`, `metrics`, `error_code`, `error_message`, `result_schema_version`, `result_s3_object_id`,
|
||
`result_msg_id`) и зафиксировать транзакцию. При `status = error` файл задания (если `current_recognition_job_id` равен
|
||
заданию) SHALL получить `stored_file.status = ERROR`; при `done`/`needs_check` файл остаётся `PROCESSING` и SHALL быть
|
||
обработан циклом маппинга (спека `osv-result-mapping`); inbox MUST NOT читать файл результата. Неизвестный `job_id` SHALL
|
||
быть залогирован. Результат для задания, уже находящегося в финальном статусе (`done`, `needs_check`, `error`,
|
||
`timeout`, `dlq`), SHALL логироваться и не применяться.
|
||
|
||
#### Scenario: Успешный результат
|
||
- **WHEN** сервис публикует конверт `status = done` с `result_s3 = {arm-ka, results/<file_id>/<job_id>/result.json}` и
|
||
`Nats-Msg-Id = result:<job_id>:1`
|
||
- **THEN** задание получает `status = done`, `result_msg_id = result:<job_id>:1`, `result_s3_object_id` указывает на новую
|
||
строку `s3_object` с этим ключом, `finished_at` заполнен; файл остаётся `PROCESSING` до маппинга
|
||
|
||
#### Scenario: Повторная доставка
|
||
- **WHEN** то же сообщение получено второй раз
|
||
- **THEN** строка `recognition_job` и `s3_object` не меняются
|
||
|
||
#### Scenario: Ошибка сервиса
|
||
- **WHEN** конверт содержит `status = error`, `error = {UNSUPPORTED_FORMAT, …}` без `result_s3`
|
||
- **THEN** задание получает `status = error`, `error_code = UNSUPPORTED_FORMAT`, `error_message` из конверта,
|
||
`result_s3_object_id` пуст; файл — `ERROR`
|
||
|
||
#### Scenario: Результат пришёл после таймаута
|
||
- **WHEN** задание уже `timeout`, и приходит конверт `done`
|
||
- **THEN** задание не меняется, событие залогировано
|
||
|
||
### Requirement: Таймаут ожидания результата
|
||
Фоновый вахтёр SHALL с периодом 30 с переводить задания `processing` и `published` с `reply_deadline < now()` в
|
||
`status = timeout`, `error_code = TIMEOUT`, `finished_at = now()`, а их файлы (по `current_recognition_job_id`) — в
|
||
`stored_file.status = ERROR`. Публикатор при переводе задания в `dlq` SHALL так же переводить файл в `ERROR`. Ручной повтор после TIMEOUT/NO_ACK SHALL создавать новое задание по требованию «Новая попытка обработки файла».
|
||
|
||
#### Scenario: Сервис взял, но не ответил
|
||
- **WHEN** задание в `processing` и `reply_deadline` прошёл без результата
|
||
- **THEN** не позднее чем через 30 с задание имеет `status = timeout`, файл — `ERROR` с `error_code = TIMEOUT`
|
||
|
||
### Requirement: Соединение с NATS по шаблону, без JetStream
|
||
Ядро SHALL подключаться к NATS с nkeys-seed и JWT, если заданы `OPENBAO__NATS__NATS_SEED/NATS_JWT`, и без учётных данных,
|
||
если они пусты; подписываться на subject ответа `NATS_RESULT_SUBJECT` (по умолчанию `files.tabular_parsed`) в 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`
|
||
|
||
### Requirement: Subject'ы и учётные данные NATS из настроек окружения
|
||
Ядро SHALL брать subject'ы обмена из `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT`, `NATS_IMAGE_REQUEST_SUBJECT`,
|
||
`NATS_RESULT_SUBJECT` (умолчания `files.tabular_uploaded`, `files.img_uploaded`, `files.tabular_parsed`): публикует запросы в `tabular_request` и `image_request`, подписан на `result` (queue group `arm-core`).
|
||
При старте SHALL проверяться все три имени (непустое, без пробельных символов, без `*` и `>`, без пустых токенов,
|
||
попарно различны); при нарушении приложение MUST NOT стартовать. Значения `NATS_JWT` и `NATS_SEED` MUST NOT попадать в
|
||
логи, `repr` настроек и тексты ошибок валидации. Используемые subject'ы SHALL писаться в лог при старте и отдаваться в
|
||
`/healthcheck` объектом `subjects` с ключами `tabular_request`, `image_request`, `result` при любом значении `nats` (`ok | down | off`).
|
||
|
||
#### Scenario: Умолчания
|
||
- **WHEN** приложение стартует без ключей `OPENBAO__NATS__NATS_*_SUBJECT`
|
||
- **THEN** `/healthcheck.subjects` содержит имена `files.*` по умолчанию, обмен идёт по ним
|
||
|
||
#### Scenario: Subject'ы окружения
|
||
- **WHEN** заданы `OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT=test.files.tabular_uploaded`,
|
||
`OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT=test.files.img_uploaded`,
|
||
`OPENBAO__NATS__NATS_RESULT_SUBJECT=test.files.tabular_parsed`
|
||
- **THEN** подписки и публикации идут только по этим subject'ам, `/healthcheck.subjects` их показывает
|
||
|
||
#### Scenario: Недопустимый subject
|
||
- **WHEN** `OPENBAO__NATS__NATS_RESULT_SUBJECT` совпадает с subject'ом запроса или содержит `>`
|
||
- **THEN** приложение не стартует, ошибка называет ключ
|
||
|
||
#### Scenario: JWT не раскрывается
|
||
- **WHEN** заданы `NATS_JWT` и `NATS_SEED`, а валидация настроек падает на другом поле
|
||
- **THEN** текст ошибки и лог не содержат значений JWT и seed; `repr` настроек показывает их скрытыми
|
||
|
||
### Requirement: Приём конвертов и результатов обоих MAJOR контракта
|
||
Inbox SHALL принимать конверты (включая ack `processing`) и файлы результата и `MAJOR` 1, и `MAJOR` 2 («сначала ядро»,
|
||
`recognition-contract`): версия определяется по `schema_version`, конверт валидируется моделью своего `MAJOR`, неизвестный
|
||
`MAJOR` (например, `3.0`) отбрасывается с предупреждением в лог (задание закроется по сроку). Файл результата
|
||
обрабатывается маппером своего `MAJOR` (`2.0` — маппер результата `2.0`, `1.x` — маппер `1.x`); файл с неизвестным
|
||
`MAJOR` получает `ERROR` с `mapping_error`. Ядро SHALL держать копии схем и примеров обоих `MAJOR` (`api/contracts/`),
|
||
сверяемые с `docs/contracts/` общим правилом. Предупреждения уровня файла SHALL записываться в
|
||
`recognition_job.metrics.warnings` структурами `{code, severity, message, locator?}`: у `2.0` — как пришли, у `1.x` —
|
||
каждая строка как `{code: "LEGACY_WARNING", severity: "info", message: <строка>, locator: null}`.
|
||
`recognition_job.statements_version` SHALL сохранять `service.statements_version` результата `2.0` (у `1.x` — `NULL`).
|
||
|
||
#### Scenario: Ack нового сервиса
|
||
- **WHEN** сервис публикует ack `processing` со `schema_version = "2.0"`
|
||
- **THEN** задание переходит в `processing`, как для ack `1.0`
|
||
|
||
#### Scenario: Предупреждения 1.x оборачиваются
|
||
- **WHEN** конверт `1.0` несёт `metrics.warnings = ["sheet=ОСВ 60 строки без родителя"]`
|
||
- **THEN** в `recognition_job.metrics.warnings` — `{code: "LEGACY_WARNING", severity: "info", message: "sheet=ОСВ 60 строки без родителя", locator: null}`
|
||
|
||
#### Scenario: Предупреждения 2.0 сохраняются как пришли
|
||
- **WHEN** конверт `2.0` несёт предупреждение `JSON_FAILED` уровня `error` с `locator`
|
||
- **THEN** структура сохранена без изменений
|
||
|
||
#### Scenario: Неизвестный MAJOR
|
||
- **WHEN** приходит конверт со `schema_version = "3.0"`
|
||
- **THEN** конверт отбрасывается с предупреждением в лог, задание не меняется
|
||
|
||
#### Scenario: Результат 2.0 разбирает маппер 2.0
|
||
- **WHEN** inbox принял файл результата со `schema_version = "2.0"`
|
||
- **THEN** документы создаёт маппер результата `2.0`; результат `1.x` по-прежнему разбирает маппер `1.x`
|
||
|
||
### Requirement: Новая попытка обработки файла
|
||
Ручной повтор SHALL создать recognition_job с новым id, attempt = прежний attempt + 1, тем же file_id,
|
||
pipeline и max_attempts, статусом queued. Предыдущая запись SHALL сохраняться без изменений.
|
||
Файл SHALL ссылаться на новое задание и получить UPLOADED. Ссылка S3 на исходник и consolidation_id сохраняются;
|
||
новых документов, файлов и копий исходника SHALL NOT создаваться.
|
||
|
||
#### Scenario: Повтор после NO_ACK
|
||
- **WHEN** tabular-задание ОСВ РБ завершилось dlq/NO_ACK с attempt = 1
|
||
- **THEN** новое задание queued имеет attempt = 2, файл UPLOADED с прежним id и объектом S3,
|
||
current_recognition_job_id указывает на новый id
|
||
|
||
### Requirement: Изоляция новой попытки
|
||
Новое задание SHALL начинаться без ack, результата, ошибок, метрик и времён старого задания, со свежим
|
||
requested_at. Публикатор SHALL вычислить новый reply_deadline и Nats-Msg-Id для нового job_id/attempt.
|
||
Запоздавшие ack/результаты прежней финальной попытки MUST NOT менять файл, создавать документы или
|
||
мешать обработке нового задания.
|
||
|
||
#### Scenario: Старый ответ
|
||
- **WHEN** после создания попытки 2 пришли ack и done для попытки 1 в timeout
|
||
- **THEN** текущее задание, статус файла и документы не меняются; ответ логируется и игнорируется
|
||
|
||
#### Scenario: Публикация после недоступности NATS
|
||
- **WHEN** ручной повтор принят, а NATS временно недоступен
|
||
- **THEN** задание остаётся queued; после восстановления публикуется со свежим сроком и новым Nats-Msg-Id
|
||
|
||
### Requirement: Конкурентные запросы повтора
|
||
Проверка текущего задания и лимита, создание новой попытки и переключение файла SHALL быть атомарны.
|
||
Два запроса к одной неуспешной попытке MUST NOT создавать два новых задания. При rollback новое задание,
|
||
переключение файла и аудит SHALL откатываться вместе.
|
||
|
||
#### Scenario: Двойное нажатие
|
||
- **WHEN** два параллельных запроса повторяют attempt = 1
|
||
- **THEN** создан ровно один job attempt = 2; второй одиночный запрос получает 409 FILE_NOT_RETRYABLE,
|
||
а пакетный отмечает файл skipped; attempt = 3 не создаётся
|