test(background_tasks): fix two CI flakes in eager outbox and real-Redis invalidation tests

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,23 @@
 _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). ``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)
+
+
 async def _await_delivery(seen: list[Invalidation], count: int = 1) -> None:
     """Wait for the listener task to have delivered *count* messages.
 

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()