fix(background_tasks): warm the invalidation publish connection from the listener

CHANGELOG.md

@@ -14,6 +14,10 @@ All notable changes to this project are documented in this file. The format is b
 
 ## [Unreleased]
 
+### Fixed
+
+- `background_tasks`: the Redis invalidation listener now opens the connection a publish uses (one `PING` after every successful subscribe), so the first invalidation after boot or after a Redis outage no longer spends part of its 1 s budget connecting. Under load that connect was cancelled and the message was lost; other workers kept the stale value until its TTL (#880).
+
 ### Changed
 
 - Faster tests: `simple_module_test` gives every in-memory SQLite engine made through `init_db` with the same dialect options one shared dialect and compiled-statement cache, so a statement compiled in one test is reused by the next instead of being recompiled for every fresh engine. The databases stay separate. The patch is process-wide; set `SM_TEST_SEPARATE_COMPILE_CACHES=1` to skip it. `make test-py` drops from about 250 s to 232 s on 8 workers (test bodies 566 s to 511 s of worker time).

docs/superpowers/plans/2026-11-07-invalidation-warm-publish.md

@@ -0,0 +1,14 @@
+# Plan: warm the invalidation publish connection (GH #880)
+
+Design: `docs/superpowers/specs/2026-11-07-invalidation-warm-publish-design.md`.
+
+1. **Fake-client tests first.** Create `modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_warmup.py`.
+   - Its own small fake with `ping()`, because the shared fake lives in a 293-line file. Copy only what the tests need.
+   - Cover: a ping after subscribe; another after a reconnect; a failing ping and a hanging ping both leave the listener delivering.
+   - Done when they fail against `main`.
+2. **Implement** `_warm_publish_connection` in `modules/background_tasks/background_tasks/core/invalidation.py` and call it after a successful subscribe. Add one sentence to the `PUBLISH_TIMEOUT_SECONDS` docstring. Done when the step 1 tests pass.
+3. **Real-Redis test** in the same new file: once subscribed, `CLIENT LIST` shows a client whose last command is `ping`, and a publish does not add a client. Done when it passes against `SM_TEST_REDIS_URL`.
+4. **CHANGELOG** `[Unreleased]` entry under Fixed, citing #880.
+5. `make lint`, `make test-py` (with `SM_TEST_REDIS_URL`), `make test-js`.
+
+Tasks 1–3 are sequential and touch the same files; none is marked hard.

docs/superpowers/specs/2026-11-07-invalidation-warm-publish-design.md

@@ -0,0 +1,90 @@
+# Warm the invalidation publish connection (GH #880)
+
+## Goal and scope
+
+A publish on `RedisInvalidationTransport` runs inside `asyncio.timeout(PUBLISH_TIMEOUT_SECONDS)`
+(1 s), in the response cycle of the request that made the write. The client's command
+connection is opened lazily by the first command, so the first publish after boot, or after
+Redis came back from an outage, also pays the TCP connect and handshake inside that second.
+Under load the connect is cancelled (seen in CI, #865), the bus logs "Invalidation broadcast
+failed", and other workers keep the stale value until its TTL.
+
+In scope: make the command connection already open when a publish needs it, in the cases we
+can see coming. Out of scope: any change to the 1 s budget, the bus's swallow-and-log
+contract, or delivery guarantees. Invalidation stays best-effort over a TTL floor.
+
+## Approach
+
+**Chosen: warm from the listener.** Each time the listener task's SUBSCRIBE succeeds (first
+boot, and every reconnect after an outage), it sends one `PING` on the command client, bounded
+by `SUBSCRIBE_TIMEOUT_SECONDS`. That opens a pooled connection, which the next publish reuses.
+`start()` still never waits on Redis.
+
+Rejected:
+- **A separate connect budget** (say 1 s connect + 1 s publish). It doubles the worst case a
+  request can be held, and it still loses the message when the connect is slower than that.
+- **Publish from a background sender queue with retry.** This takes publish out of the request
+  path entirely. But it changes the transport contract (`publish` would no longer raise for
+  the bus to log), queued messages are lost at shutdown, and it is more machinery than a
+  best-effort accelerator needs.
+
+## Design
+
+In `modules/background_tasks/background_tasks/core/invalidation.py`, right after
+`pubsub.subscribe(...)` succeeds in `_listen`:
+
+```python
+await self._warm_publish_connection()
+```
+
+```python
+async def _warm_publish_connection(self) -> None:
+    """Open the command connection a publish will use, off the request path."""
+    try:
+        async with asyncio.timeout(SUBSCRIBE_TIMEOUT_SECONDS):
+            await self._client.ping()
+    except Exception as exc:  # the listener stays up; a publish just connects cold
+        logger.debug("Invalidation publish connection not warmed: %s", exc)
+```
+
+- A failed warm-up never fails or restarts the listener. The subscription is what keeps this
+  worker receiving; the warm-up only saves the next publish a connect.
+- Cancellation (`stop()`) still propagates, because `CancelledError` is not an `Exception`.
+- The constants and docstrings say why: `PUBLISH_TIMEOUT_SECONDS` gains one sentence on the warm
+  connection.
+
+## Decisions made
+
+- Warm on every successful subscribe, not only the first. A Redis restart drops both
+  connections, and the listener notices it first (health check within ~15 s).
+- Bound the PING by `SUBSCRIBE_TIMEOUT_SECONDS` (5 s), not the publish budget: it is not in a
+  request's way.
+- Failures log at DEBUG. The listener already warns once per outage, and a second warning for
+  the same outage would be noise.
+- No new setting or environment variable.
+- The #881 test fixture that raises the publish timeout in the real-Redis tests stays. Those
+  tests publish right after `start()`, before a warm-up can finish.
+
+## Testing
+
+New file `modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_warmup.py`
+(the existing `test_background_tasks_invalidation.py` is at 293 of the 300-line cap):
+
+- With the fake client: after the listener subscribes, the client receives one `ping`; after a
+  scripted subscribe failure and a reconnect, it receives another.
+- With the fake client: a `ping` that raises, or hangs (bounded by a patched short timeout),
+  leaves the listener subscribed and delivering.
+- With a real `redis-server`: once subscribed, the server reports a second client for this
+  transport whose last command is `ping` (`CLIENT LIST`). A publish then reuses it, and the
+  server's client count does not grow.
+
+The new file carries its own small fake, with an `async def ping()`. Full `make test-py` with
+`SM_TEST_REDIS_URL`, plus `make lint`. No browser QA (`--skip-browser`): nothing user-visible changes.
+
+## Risks
+
+- redis-py might discard idle pool connections, so the warmed connection would not be reused.
+  It doesn't (`ConnectionPool` keeps released connections until `disconnect`); the real-Redis
+  test pins this.
+- The health check on an idle command connection runs inside the next publish's budget, but
+  that is a PING round trip on an open socket, not a connect.

modules/background_tasks/background_tasks/core/invalidation.py

@@ -83,7 +83,9 @@
 One second is far more than a local round trip needs and far less than a user
 waits. Exceeding it raises, which the bus logs and swallows: the local eviction
 has already happened, so the cost of the timeout is the same staleness window
-that existed before this transport.
+that existed before this transport. The listener opens the connection a publish
+uses ahead of time (``_warm_publish_connection``), so the budget normally covers
+a round trip on an open socket, not a connect.
 """
 
 
@@ -174,6 +176,7 @@ async def _listen(self) -> None:
                 pubsub = self._client.pubsub()
                 async with asyncio.timeout(SUBSCRIBE_TIMEOUT_SECONDS):
                     await pubsub.subscribe(self._channel)
+                await self._warm_publish_connection()
                 if self._warned:
                     logger.info("Invalidation listener reconnected to %r", self._channel)
                 self._warned = False
@@ -198,6 +201,22 @@ async def _listen(self) -> None:
                     with contextlib.suppress(Exception):
                         await pubsub.aclose()
 
+    async def _warm_publish_connection(self) -> None:
+        """Open the command connection the next publish will use, off the request path.
+
+        The client opens it lazily, so without this the first publish after boot,
+        or after Redis came back, paid the TCP connect inside its 1 s budget and
+        lost the message under load (GH #880). Runs after every successful
+        SUBSCRIBE because an outage drops both connections. A failure here leaves
+        the listener up: the subscription is what keeps this worker receiving, and
+        a cold publish still connects on its own.
+        """
+        try:
+            async with asyncio.timeout(SUBSCRIBE_TIMEOUT_SECONDS):
+                await self._client.ping()
+        except Exception as exc:
+            logger.debug("Invalidation publish connection not warmed: %s", exc)
+
     def _log_failure(self, exc: Exception) -> None:
         """Warn once per outage, then stay quiet about the same outage."""
         if self._warned:

modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_redis.py

@@ -38,8 +38,9 @@ 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
+    publish may open its command connection. The listener warms it after SUBSCRIBE
+    (GH #880), but these tests publish right after ``start()``, before that can
+    finish. 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,

modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_warmup.py

@@ -0,0 +1,190 @@
+"""The listener warms the connection a publish will use (GH #880).
+
+``publish`` runs under a 1 s budget in the request's response cycle, and the
+client's command connection opens lazily. Without a warm-up the first publish
+after boot, or after an outage, paid the TCP connect inside that second and lost
+the message under load. The listener PINGs the command client after every
+successful SUBSCRIBE, off the request path, so that connection is already open.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+
+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
+
+_CHANNEL = "sm.invalidation.warmup"
+
+
+class _PubSub:
+    def __init__(self, frames, subscribe_error) -> None:
+        self._frames = list(frames)
+        self._subscribe_error = subscribe_error
+        self.subscribed = False
+
+    async def subscribe(self, channel: str) -> None:
+        if self._subscribe_error is not None:
+            raise self._subscribe_error
+        self.subscribed = True
+
+    async def get_message(self, ignore_subscribe_messages: bool = False, timeout=None):
+        if self._frames:
+            return self._frames.pop(0)
+        await asyncio.sleep(timeout or 0)  # intentional-sleep: fake idle-channel poll interval
+        return None
+
+    async def aclose(self) -> None: ...
+
+
+class _Redis:
+    """Records PINGs; the first ``fail_subscribes`` subscriptions raise."""
+
+    def __init__(self, frames=(), fail_subscribes=0, ping_error=None, ping_hangs=False) -> None:
+        self.pings = 0
+        self.pubsubs: list[_PubSub] = []
+        self._frames = frames
+        self._fail_subscribes = fail_subscribes
+        self._ping_error = ping_error
+        self._ping_hangs = ping_hangs
+
+    def pubsub(self) -> _PubSub:
+        error = None
+        if self._fail_subscribes:
+            self._fail_subscribes -= 1
+            error = ConnectionError("redis down")
+        ps = _PubSub(self._frames, error)
+        self.pubsubs.append(ps)
+        return ps
+
+    async def ping(self) -> bool:
+        self.pings += 1
+        if self._ping_hangs:
+            await asyncio.Event().wait()
+        if self._ping_error is not None:
+            raise self._ping_error
+        return True
+
+    async def aclose(self) -> None: ...
+
+
+@pytest.fixture
+def fake_redis(monkeypatch):
+    import redis.asyncio as aioredis
+
+    def install(**kwargs) -> _Redis:
+        fake = _Redis(**kwargs)
+        monkeypatch.setattr(aioredis, "from_url", lambda *a, **kw: fake)
+        return fake
+
+    monkeypatch.setattr(bg_invalidation, "INITIAL_BACKOFF_SECONDS", 0.01)
+    monkeypatch.setattr(bg_invalidation, "LISTEN_POLL_SECONDS", 0.01)
+    return install
+
+
+def _frame(key: str) -> dict:
+    wire = Invalidation(channel=_CHANNEL, key=key, origin="elsewhere").to_wire()
+    return {"type": "message", "data": wire.encode()}
+
+
+class TestWarmUp:
+    async def test_the_listener_pings_once_it_has_subscribed(self, fake_redis):
+        fake = fake_redis()
+        transport = RedisInvalidationTransport(InvalidationBus(), "redis://fake", _CHANNEL)
+        await transport.start()
+        try:
+            await wait_for(lambda: fake.pings == 1)
+            assert fake.pubsubs[0].subscribed
+        finally:
+            await transport.stop()
+
+    async def test_a_reconnect_warms_again(self, fake_redis):
+        """A Redis restart drops both connections; the resubscribe re-warms."""
+        fake = fake_redis(fail_subscribes=1)
+        transport = RedisInvalidationTransport(InvalidationBus(), "redis://fake", _CHANNEL)
+        await transport.start()
+        try:
+            await wait_for(lambda: fake.pings == 1)
+            assert [ps.subscribed for ps in fake.pubsubs] == [False, True]
+        finally:
+            await transport.stop()
+
+    async def test_no_ping_before_a_subscribe_succeeds(self, fake_redis):
+        fake = fake_redis(fail_subscribes=10**6)
+        transport = RedisInvalidationTransport(InvalidationBus(), "redis://fake", _CHANNEL)
+        await transport.start()
+        try:
+            await wait_for(lambda: len(fake.pubsubs) >= 3)
+            assert fake.pings == 0
+        finally:
+            await transport.stop()
+
+
+class TestAFailedWarmUpIsHarmless:
+    @pytest.mark.parametrize("ping", ["raises", "hangs"])
+    async def test_the_listener_still_delivers(self, fake_redis, monkeypatch, ping):
+        monkeypatch.setattr(bg_invalidation, "SUBSCRIBE_TIMEOUT_SECONDS", 0.05)
+        if ping == "raises":
+            fake = fake_redis(frames=[_frame("k")], ping_error=ConnectionError("no"))
+        else:
+            fake = fake_redis(frames=[_frame("k")], ping_hangs=True)
+        bus = InvalidationBus()
+        seen: list[Invalidation] = []
+        bus.subscribe(_CHANNEL, seen.append)
+        transport = RedisInvalidationTransport(bus, "redis://fake", _CHANNEL)
+        await transport.start()
+        try:
+            await wait_for(lambda: bool(seen))
+        finally:
+            await transport.stop()
+
+        assert [i.key for i in seen] == ["k"]
+        assert len(fake.pubsubs) == 1, "a failed warm-up must not restart the listener"
+
+
+class TestAgainstARealServer:
+    async def test_a_publish_reuses_the_warmed_connection(self, redis_server):
+        """The server sees the PING'd connection, and the publish arrives on it.
+
+        Clients are found by a per-test ``client_name`` (parsed from the URL by
+        redis-py): under xdist every worker shares one Redis, so counting all of
+        the server's clients would race the other workers.
+        """
+        import redis.asyncio as aioredis
+
+        name = f"sm-warmup-{os.getpid()}-{uuid.uuid4().hex[:8]}"
+        sep = "&" if "?" in redis_server else "?"
+        transport = RedisInvalidationTransport(
+            InvalidationBus(), f"{redis_server}{sep}client_name={name}", _CHANNEL
+        )
+        probe = aioredis.from_url(redis_server)
+
+        async def ours() -> list[dict]:
+            return [c for c in await probe.client_list() if c.get("name") == name]
+
+        async def warmed() -> list[dict]:
+            clients = await ours()
+            return clients if any(c["cmd"] == "ping" for c in clients) else []
+
+        await transport.start()
+        try:
+            before = await wait_for(warmed, timeout=5.0)
+            await transport.publish(
+                Invalidation(channel=_CHANNEL, key="k", origin="elsewhere").to_wire()
+            )
+            after = await ours()
+        finally:
+            await transport.stop()
+            await probe.aclose()
+
+        assert len(before) == 2, "expected the subscription plus one warmed command connection"
+        warmed_id = next(c["id"] for c in before if c["cmd"] == "ping")
+        assert {c["id"] for c in after} == {c["id"] for c in before}, (
+            "publish opened a new connection"
+        )
+        assert next(c["cmd"] for c in after if c["id"] == warmed_id) == "publish"