From c3a62bb27c7cb0f6efdbad4e5f77749e866cb8f5 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 03:35:24 +0400 Subject: [PATCH] =?UTF-8?q?=D0=A1=D1=80=D0=B5=D0=B7=200=20[5.2]=20T11.1:?= =?UTF-8?q?=20PUBLISH=5FRETRIES=20=E2=80=94=20=D1=87=D0=B8=D1=81=D0=BB?= =?UTF-8?q?=D0=BE=20=D0=BF=D0=BE=D0=B2=D1=82=D0=BE=D1=80=D0=BE=D0=B2=20?= =?UTF-8?q?=D0=BF=D0=BE=D1=81=D0=BB=D0=B5=20=D0=BF=D0=B5=D1=80=D0=B2=D0=BE?= =?UTF-8?q?=D0=B9=20=D0=BF=D0=BE=D0=BF=D1=8B=D1=82=D0=BA=D0=B8=20(=D0=B4?= =?UTF-8?q?=D0=BE=204=20=D0=BF=D0=BE=D0=BF=D1=8B=D1=82=D0=BE=D0=BA),=20R-1?= =?UTF-8?q?3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit publish делает первую попытку и до retries повторов (retries=0 — одна попытка), PublishFailed.attempts — фактическое число попыток. Комментарий PUBLISH_RETRIES в config.py уточнён. Докстринги: PublishFailed не значит «не доставлено» (T11.3), колбэк подписки должен возвращаться быстро (T11.4). Co-Authored-By: Claude Fable 5.1 --- api/src/core/config.py | 2 +- api/src/core/nats.py | 28 ++++++++++++++++--------- api/tests/unit/test_nats.py | 41 +++++++++++++++++++++++++++++-------- 3 files changed, 51 insertions(+), 20 deletions(-) diff --git a/api/src/core/config.py b/api/src/core/config.py index b54822a..df53a8b 100644 --- a/api/src/core/config.py +++ b/api/src/core/config.py @@ -60,7 +60,7 @@ class Settings(BaseSettings): # NATS: подписка на recognition.image.request и публикация. Выключено — воркер не запускается (dev без NATS) NATS_ENABLED: bool = True - PUBLISH_RETRIES: PositiveInt = 3 # попытки публикации результата/ack + PUBLISH_RETRIES: PositiveInt = 3 # повторы публикации результата/ack после первой попытки (всего до 4) DONE_CACHE_SIZE: PositiveInt = 1000 # размер LRU обработанных job_id (дедупликация повторов) # Горячие папки ИС Сканирование: local — обычный каталог (dev, тесты), smb — SMB-шара платформы diff --git a/api/src/core/nats.py b/api/src/core/nats.py index f72608a..af5507d 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -12,8 +12,8 @@ инстансов сообщение получает ровно один. `cb` получает исходное сообщение nats-py (`msg.data`, `msg.headers`); исключения `cb` логируются, подписка живёт. - `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush` (подтверждение сервером) с - повторами на временных ошибках; исчерпаны — `PublishFailed`. Воркер логирует error и идёт дальше: ядро закроет - задание по `reply_deadline` (D-7). + повторами на временных ошибках (первая попытка + до `PUBLISH_RETRIES` повторов); исчерпаны — `PublishFailed`. + Воркер логирует error и идёт дальше: ядро закроет задание по `reply_deadline` (D-7). - Stream'ов и consumer'ов JetStream нет: `jetstream()` не вызывается. """ @@ -79,7 +79,9 @@ class NatsNotConnectedError(NatsError): class PublishFailed(NatsError): - """Публикация не удалась: попытки исчерпаны или ошибка не временная. Причина последней попытки — в `__cause__`.""" + """Публикация не подтверждена: повторы исчерпаны или ошибка не временная. Причина последней попытки — в + `__cause__`; `attempts` — сколько попыток фактически сделано (0 — соединения не было, при исчерпанных повторах — + `retries + 1`).""" def __init__(self, message: str, *, subject: str, msg_id: str, attempts: int) -> None: super().__init__(message) @@ -211,6 +213,8 @@ class NatsGateway: подтверждения сервером (`flush`). `cb` получает исходное `Msg` (`data`, `headers`); его исключения логируются, подписка не падает. + `cb` должен возвращаться быстро: nats-py вызывает колбэки одной подписки последовательно, долгая обработка + (ожидание ИС) задерживает следующие запросы — её выносят в отдельную задачу. """ if self._client is None: raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {IMAGE_REQUEST_SUBJECT} невозможна") @@ -237,14 +241,18 @@ class NatsGateway: async def publish(self, subject: str, payload: bytes, msg_id: str, *, retries: int | None = None) -> None: """Публикует `payload` с заголовком `Nats-Msg-Id: ` и ждёт подтверждения сервера (`flush`). - `retries` — число попыток **всего** (по умолчанию `PUBLISH_RETRIES` = 3: первая и до двух повторов), между - попытками — пауза `retry_delay`. Повторяются только временные ошибки (`TRANSIENT_PUBLISH_ERRORS`). Попытки - исчерпаны, ошибка не временная или соединения не было вовсе (`attempts = 0`) — `PublishFailed`. + `retries` — число **повторов** после первой попытки (спека: «повторить до трёх раз»; по умолчанию + `PUBLISH_RETRIES` = 3, т.е. не более 4 попыток; `retries = 0` — ровно одна попытка), перед каждым повтором — + пауза `retry_delay`. Повторяются только временные ошибки (`TRANSIENT_PUBLISH_ERRORS`). Повторы исчерпаны, + ошибка не временная или соединения не было вовсе (`attempts = 0`) — `PublishFailed`. Дубликаты при повторе безопасны: ядро дедуплицирует по `Nats-Msg-Id`. + `PublishFailed` не означает «не доставлено»: сообщения, буферизованные nats-py во время переподключения, могут + уйти после восстановления соединения; ядро отбросит копии по `Nats-Msg-Id`. """ - attempts_total = self._settings.PUBLISH_RETRIES if retries is None else retries - if attempts_total < 1: - raise ValueError(f"retries должно быть ≥ 1, получено {attempts_total}") + retry_count = self._settings.PUBLISH_RETRIES if retries is None else retries + if retry_count < 0: + raise ValueError(f"retries должно быть ≥ 0, получено {retry_count}") + attempts_total = retry_count + 1 if self._client is None: raise PublishFailed( f"Нет соединения с NATS: {msg_id} в {subject} не опубликовано", @@ -261,7 +269,7 @@ class NatsGateway: except TRANSIENT_PUBLISH_ERRORS as exc: if attempt == attempts_total: raise PublishFailed( - f"Публикация {msg_id} в {subject} не удалась за {attempt} попыток ({type(exc).__name__})", + f"Публикация {msg_id} в {subject} не удалась, попыток: {attempt} ({type(exc).__name__})", subject=subject, msg_id=msg_id, attempts=attempt, diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py index c490ed0..06e3baf 100644 --- a/api/tests/unit/test_nats.py +++ b/api/tests/unit/test_nats.py @@ -298,9 +298,19 @@ async def test_publish_retries_transient_errors_until_success( await gateway.publish(RESULT_SUBJECT, b"payload", "ack:job:1") - assert len(fake.published) == 3 # PUBLISH_RETRIES=3 — три попытки всего + assert len(fake.published) == 3 # первая попытка + два повтора из PUBLISH_RETRIES=3 assert all(headers == {"Nats-Msg-Id": "ack:job:1"} for _, _, headers in fake.published) - assert sleeps == [0, 0] # пауза между попытками, после последней — нет + assert sleeps == [0, 0] # пауза перед каждым повтором + + +async def test_publish_succeeds_on_last_retry(gateway: NatsGateway, fake: FakeNatsClient, sleeps: list[float]): + await gateway.connect() + fake.publish_errors = [nats_errors.ConnectionClosedError()] * 3 + + await gateway.publish(RESULT_SUBJECT, b"payload", "ack:job:1") + + assert len(fake.published) == 4 # 1 + 3 повтора: последний повтор успешен + assert sleeps == [0, 0, 0] @pytest.mark.parametrize( @@ -315,7 +325,7 @@ async def test_publish_retries_transient_errors_until_success( ConnectionResetError(), ], ) -async def test_publish_raises_publish_failed_after_exactly_retries_attempts( +async def test_publish_raises_publish_failed_after_initial_attempt_and_all_retries( gateway: NatsGateway, fake: FakeNatsClient, error: BaseException ): await gateway.connect() @@ -324,8 +334,8 @@ async def test_publish_raises_publish_failed_after_exactly_retries_attempts( with pytest.raises(PublishFailed) as exc_info: await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1") - assert len(fake.published) == 3 - assert exc_info.value.attempts == 3 + assert len(fake.published) == 4 # PUBLISH_RETRIES=3: первая попытка + 3 повтора + assert exc_info.value.attempts == 4 assert exc_info.value.msg_id == "result:job:1" assert exc_info.value.__cause__ is error @@ -336,7 +346,20 @@ async def test_publish_retries_parameter_overrides_settings(gateway: NatsGateway with pytest.raises(PublishFailed): await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=5) - assert len(fake.published) == 5 + assert len(fake.published) == 6 + + +async def test_publish_zero_retries_makes_exactly_one_attempt( + gateway: NatsGateway, fake: FakeNatsClient, sleeps: list[float] +): + await gateway.connect() + fake.publish_errors = [nats_errors.ConnectionClosedError()] * 10 + + with pytest.raises(PublishFailed) as exc_info: + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=0) + assert len(fake.published) == 1 + assert exc_info.value.attempts == 1 + assert sleeps == [] async def test_publish_uses_publish_retries_from_settings(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient): @@ -347,7 +370,7 @@ async def test_publish_uses_publish_retries_from_settings(monkeypatch: pytest.Mo with pytest.raises(PublishFailed): await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1") - assert len(fake.published) == 2 + assert len(fake.published) == 3 # PUBLISH_RETRIES=2: 1 + 2 повтора @pytest.mark.parametrize("error", [nats_errors.MaxPayloadError(), nats_errors.BadSubjectError(), ValueError("x")]) @@ -371,10 +394,10 @@ async def test_publish_without_connection_fails_without_attempts(gateway: NatsGa assert fake.published == [] -async def test_publish_rejects_non_positive_retries(gateway: NatsGateway): +async def test_publish_rejects_negative_retries(gateway: NatsGateway): await gateway.connect() with pytest.raises(ValueError, match="retries"): - await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=0) + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=-1) # --- публикация конверта ----------------------------------------------------------------------------------------