From 7bc5690a6ee303fb7934a4c8a4beda2e717bde5c Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Sat, 7 Nov 2026 14:16:42 +0100 Subject: [PATCH 1/2] test(background_tasks): fix two CI flakes in eager outbox and real-Redis invalidation tests The eager test now awaits the delivery task with drain() instead of polling the shared in-memory SQLite connection. The real-Redis tests give a publish the delivery wait's budget: in CI the 1 s publish timeout cancelled the publisher's first connect and the message was lost. Closes #865 Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- .../test_background_tasks_invalidation_redis.py | 15 +++++++++++++++ ...test_background_tasks_outbox_eager_platform.py | 13 ++++++++----- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py b/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py index 4bc51db5..8a527fce 100644 --- a/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py +++ b/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py @@ -20,6 +20,7 @@ import os import pytest +from background_tasks.core import invalidation as bg_invalidation from background_tasks.core.invalidation import RedisInvalidationTransport from simple_module_core.invalidation import Invalidation, InvalidationBus from simple_module_test import wait_for @@ -32,6 +33,20 @@ _DELIVERY_TIMEOUT_SECONDS = 5.0 +@pytest.fixture(autouse=True) +def _patient_publish(monkeypatch): + """Give a publish the delivery wait's budget instead of production's 1 s. + + ``PUBLISH_TIMEOUT_SECONDS`` bounds the whole publish, and a publisher's first + publish opens its command connection (the listener's pub/sub connection does + not warm it). In CI that connect was cancelled by the 1 s timeout inside + ``create_connection``, the bus logged "Invalidation broadcast failed", and the + message was lost (GH #865). These tests prove redis-py's wire behaviour; the timeout itself is + covered against a mute broker in ``test_background_tasks_invalidation_degraded.py``. + """ + monkeypatch.setattr(bg_invalidation, "PUBLISH_TIMEOUT_SECONDS", _DELIVERY_TIMEOUT_SECONDS) + + async def _await_delivery(seen: list[Invalidation], count: int = 1) -> None: """Wait for the listener task to have delivered *count* messages. diff --git a/modules/background_tasks/tests/outbox/delivery/test_background_tasks_outbox_eager_platform.py b/modules/background_tasks/tests/outbox/delivery/test_background_tasks_outbox_eager_platform.py index f9dcb3d9..e6f586d0 100644 --- a/modules/background_tasks/tests/outbox/delivery/test_background_tasks_outbox_eager_platform.py +++ b/modules/background_tasks/tests/outbox/delivery/test_background_tasks_outbox_eager_platform.py @@ -22,7 +22,6 @@ from simple_module_db import all_tenants, current_tenant_id, finalize_session, tenant_context from simple_module_db.tenancy import is_all_tenants from simple_module_db.transaction import SESSION_START_KEY -from simple_module_test import wait_for pytestmark = pytest.mark.usefixtures("_stub_celery") @@ -45,11 +44,15 @@ async def handler(event: EagerPing, ctx: DurableContext) -> None: await app.state.sm.event_bus.publish_durable(EagerPing(7), session) await session.commit() - async def delivered() -> bool: - rows = await outbox_rows() - return bool(rows) and all(r.status == OutboxStatus.DELIVERED for r in rows) + # Wait on the delivery task itself, not by polling the table: the test + # database is one in-memory SQLite connection (StaticPool), so a polling + # session ran on the same connection as the delivery's own transaction. + # That poll timed out once in CI with no delivery error logged (GH #865). + async with asyncio.timeout(15.0): + await app.state.sm.event_bus.durable_transport.drain() - await wait_for(delivered, timeout=15.0) + rows = await outbox_rows() + assert rows and all(r.status == OutboxStatus.DELIVERED for r in rows) assert seen == [7] app.state.background_tasks.celery.send_task.assert_not_called() From f3f0aaa5cf922de3a75e0d819f5a3751f483b764 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Sat, 7 Nov 2026 14:20:00 +0100 Subject: [PATCH 2/2] fix: address code review findings (round 1, pass 1) Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- .../test_background_tasks_invalidation_redis.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py b/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py index 8a527fce..6e496f30 100644 --- a/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py +++ b/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py @@ -41,8 +41,11 @@ def _patient_publish(monkeypatch): publish opens its command connection (the listener's pub/sub connection does not warm it). In CI that connect was cancelled by the 1 s timeout inside ``create_connection``, the bus logged "Invalidation broadcast failed", and the - message was lost (GH #865). These tests prove redis-py's wire behaviour; the timeout itself is - covered against a mute broker in ``test_background_tasks_invalidation_degraded.py``. + message was lost (GH #865). ``start()`` also reads it as + ``socket_connect_timeout``, so this fixture must run before a transport starts, + and a test here that needs the production bound has to set it back. These tests + prove redis-py's wire behaviour; the timeout itself is covered against a mute + broker in ``test_background_tasks_invalidation_degraded.py``. """ monkeypatch.setattr(bg_invalidation, "PUBLISH_TIMEOUT_SECONDS", _DELIVERY_TIMEOUT_SECONDS)