diff --git a/CHANGELOG.md b/CHANGELOG.md index fc3b000de..5472360f6 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/docs/superpowers/plans/2026-11-07-invalidation-warm-publish.md b/docs/superpowers/plans/2026-11-07-invalidation-warm-publish.md new file mode 100644 index 000000000..532638c5b --- /dev/null +++ b/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. diff --git a/docs/superpowers/specs/2026-11-07-invalidation-warm-publish-design.md b/docs/superpowers/specs/2026-11-07-invalidation-warm-publish-design.md new file mode 100644 index 000000000..b7ab492ec --- /dev/null +++ b/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. diff --git a/modules/background_tasks/background_tasks/core/invalidation.py b/modules/background_tasks/background_tasks/core/invalidation.py index 85b170df0..f022f7277 100644 --- a/modules/background_tasks/background_tasks/core/invalidation.py +++ b/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: 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 6e496f30c..e460a0814 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 @@ -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, diff --git a/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_warmup.py b/modules/background_tasks/tests/core/invalidation/test_background_tasks_invalidation_warmup.py new file mode 100644 index 000000000..6fcc24c68 --- /dev/null +++ b/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"