fix(invalidation): throttle the listener's reconnect WARNING to one a minute (LAB-7638)

docs/configuration.md

@@ -98,13 +98,19 @@ CACHEKIT_ENCRYPTION_FAIL_CLOSED=false
 # Fallback: REDIS_URL also supported (lower priority)
 REDIS_URL=redis://localhost:6379/0
 
+# Evict this process's L1 copies when another process invalidates them (default: false).
+# Tenant-scoped Redis backend only; see the link below this block.
+CACHEKIT_INVALIDATION_LISTENER_ENABLED=false
+
 # Logging
 LOG_LEVEL=INFO
 
 # Performance Testing
 REQUESTS_CA_BUNDLE=  # Unset to avoid SSL issues
 ```
 
+`CACHEKIT_INVALIDATION_LISTENER_ENABLED` runs one invalidation listener per process; see [Cross-Process L1 Eviction](features/l1-invalidation.md#cross-process-l1-eviction) for what it does and what it needs from Redis.
+
 ---
 
 ## CachekitIO Configuration

docs/features/l1-invalidation.md

@@ -246,7 +246,7 @@ export CACHEKIT_INVALIDATION_LISTENER_ENABLED=true
 The process then runs one invalidation listener: a background thread with its own Redis connection, subscribed to the channel that every invalidation is [announced on](#whole-function-invalidation). It starts on the process's first cache operation that reaches Redis. When another process's `invalidate_cache(args)` deletes a key, the listener evicts that key from this process's L1. When another process's no-args `invalidate_cache()` drains a function, the listener evicts every key this process cached for that function. Eviction is tenant-blind, like L1 itself: an invalidation under any tenant evicts the one shared L1 entry.
 
 - **Off by default.** No preset and no decorator argument turns it on. With the variable unset, a process opens no thread and no extra connection. Every process on this backend announces its own invalidations either way, so a cron job or an operator shell that calls `invalidate_cache()` needs no setting for listening workers to evict.
-- **Delivery.** At most once. A subscribed process usually evicts within milliseconds, and cachekit's tests hold it to one second. An event sent while the listener is not subscribed is lost, and that L1 copy expires by its L1 TTL as before: before the process's first cache operation that reaches Redis, while the listener starts, and while it reconnects. The start waits for Redis to confirm the subscription, so a sync cache operation that starts the listener reads L2 only once it is subscribed. An async operation starts it without waiting, and other threads' operations go on while a start is in progress, so a value they read from L2 just before the subscription can outlive an invalidation sent then, until its L1 TTL. A connection that Redis closes, on a restart, a failover or `CLIENT KILL`, is noticed at once: the listener reconnects and subscribes again on its own, logging a WARNING `Invalidation listener error` while Redis is unreachable. The connection PINGs Redis after 10 idle seconds, which keeps idle-timeout proxies and NAT gateways from dropping it, but the listener does not wait for the reply: a connection that dies silently, with packets dropped and no reset, is noticed only when TCP gives up on it, which can take minutes. Events sent meanwhile are lost.
+- **Delivery.** At most once. A subscribed process usually evicts within milliseconds, and cachekit's tests hold it to one second. An event sent while the listener is not subscribed is lost, and that L1 copy expires by its L1 TTL as before: before the process's first cache operation that reaches Redis, while the listener starts, and while it reconnects. The start waits for Redis to confirm the subscription, so a sync cache operation that starts the listener reads L2 only once it is subscribed. An async operation starts it without waiting, and other threads' operations go on while a start is in progress, so a value they read from L2 just before the subscription can outlive an invalidation sent then, until its L1 TTL. A connection that Redis closes, on a restart, a failover or `CLIENT KILL`, is noticed at once: the listener reconnects and subscribes again on its own, logging a WARNING `Invalidation listener error (errors since the last warning: N)` while Redis is unreachable, at most once a minute per process, carrying the count of errors since the last one; each error in between logs at DEBUG. The connection PINGs Redis after 10 idle seconds, which keeps idle-timeout proxies and NAT gateways from dropping it, but the listener does not wait for the reply: a connection that dies silently, with packets dropped and no reset, is noticed only when TCP gives up on it, which can take minutes. Events sent meanwhile are lost.
 - **Backends.** Only the tenant-scoped Redis backend carries events (`CACHEKIT_REDIS_URL` / `REDIS_URL` without `backend=`, or a backend from `RedisBackendProvider`). With the variable set, a function cached on any other backend, such as a `RedisBackend` passed as `backend=`, File, Memcached or CachekitIO, gets no cross-process eviction, and cachekit logs one WARNING per process, `... does not carry invalidation events`. A process has one listener, on the Redis server of the first cache operation that starts it, so give every function in a listening process the same Redis.
 - **Redis permissions.** To listen, a restricted ACL user needs the `subscribe` and `ping` commands (`+@connection` includes `ping`) and access to the channel (`&cachekit:py:invalidate:v1`), besides `publish` on it to announce. From Redis 7 a new ACL user gets no channels unless granted, so `~* +@all` alone is refused. The start waits up to 5 seconds for Redis to confirm the subscription and answer a PING. A refusal, or no answer, fails the start with a WARNING `Invalidation listener failed to start`, and the next cache operation that reaches Redis (an L1 miss) retries it, no sooner than a minute later: a process that serves only L1 hits does not retry. Once the listener runs, a refusal, after an ACL change for instance, is a WARNING `Invalidation listener refused by Redis`, and the listener subscribes again every minute on its own, with no restart and no cache operation.
 - **One Redis, several apps.** Redis delivers pub/sub messages across database numbers, so apps on different databases of one Redis server see each other's events. An event names its function by namespace and `module.qualname`, so the most it costs another app is an extra L1 miss on a function with the same name and namespace.
@@ -419,9 +419,9 @@ For typical workloads (1000s of keys), overhead is <1MB.
 - `... does not carry invalidation events`: the function's backend is not the tenant-scoped Redis backend, so no event reaches it.
 - `uWSGI forks its workers without running Python's at-fork hooks`: the workers run no listener; set one of the uWSGI options under [Forked Processes](#forked-processes).
 - `Invalidation listener failed to start`: Redis was unreachable, refused the listener's SUBSCRIBE or PING, or did not answer them within 5 seconds. The message ends with the error type: `NoPermissionError` for a refusal, usually for the ACL (see **Redis permissions** under [Cross-Process L1 Eviction](#cross-process-l1-eviction)), `TimeoutError` for no answer. The next cache operation that reaches Redis (an L1 miss) retries it, no sooner than a minute later. Events sent meanwhile are lost, and those L1 copies expire by TTL.
-- `Invalidation listener error`: a running listener lost its connection; it reconnects and subscribes again on its own. Events sent meanwhile are lost.
+- `Invalidation listener error (errors since the last warning: N)`: a running listener lost its connection; it reconnects and subscribes again on its own, retrying every second. The WARNING repeats at most once a minute per process while Redis stays unreachable, and N counts the errors since the last one. Events sent meanwhile are lost.
 - `Invalidation listener refused by Redis`: Redis refused a running listener's subscription or PING, usually after an ACL change; see **Redis permissions** under [Cross-Process L1 Eviction](#cross-process-l1-eviction). The listener tries again every minute on its own, with no cache operation.
-- No WARNING: the event was published before the pod's listener subscribed, which happens during the pod's first cache operation that reaches Redis; or the invalidating process published nothing, because its L2 delete or drain failed (an ERROR or WARNING in that process), its publish failed (a WARNING `Invalidation announcement failed` there), or the function's namespace is over 1000 bytes, too long for an event (a WARNING `Invalidation not announced` there, at most once a minute with the count since the last one); or the pod was reading the old value from L2 when the eviction arrived, and stored it in L1 just after. That copy expires by its L1 TTL.
+- No WARNING: the event was published before the pod's listener subscribed, which happens during the pod's first cache operation that reaches Redis; or the invalidating process published nothing, because its L2 delete or drain failed (an ERROR or WARNING in that process), its publish failed (a WARNING `Invalidation announcement failed` there), or the function's namespace is over 1000 bytes, too long for an event (a WARNING `Invalidation not announced` there, at most once a minute with the count since the last one); or the pod was reading the old value from L2 when the eviction arrived, and stored it in L1 just after; or the listener's connection died silently, with packets dropped and no reset, so the listener has not noticed yet (see **Delivery** under [Cross-Process L1 Eviction](#cross-process-l1-eviction)). That copy expires by its L1 TTL.
 
 ### Problem: SWR refresh failing
 

src/cachekit/invalidation.py

@@ -66,6 +66,8 @@
 _too_long_warn = _WarnThrottle()
 # A forged or foreign publisher controls how many bad events arrive: one WARNING a minute here too.
 _dropped_event_warn = _WarnThrottle()
+# A listener that cannot reach Redis fails once a second while it retries: one WARNING a minute.
+_listener_error_warn = _WarnThrottle()
 
 
 def encode_event(registry_id: str, key: Optional[str]) -> Optional[bytes]:
@@ -415,7 +417,15 @@ def _on_listener_error(error: BaseException, pubsub: Any, thread: Any) -> None:
         if connection is not None:
             connection.disconnect()
         return
-    logger.warning("Invalidation listener error; retrying in 1 s: %s", redact_error_for_log(error))
+    errors = _listener_error_warn.claim()
+    if not errors:
+        logger.debug("Invalidation listener error; retrying in 1 s: %s", redact_error_for_log(error))
+    else:
+        logger.warning(
+            "Invalidation listener error (errors since the last warning: %d); retrying every second. Latest: %s",
+            errors,
+            redact_error_for_log(error),
+        )
     time.sleep(1.0)
 
 

tests/integration/test_invalidation_channel_redis.py

@@ -178,6 +178,7 @@ def test_publish_denied_by_acl_leaves_the_drain_standing(
         client.acl_setuser(
             user, enabled=True, passwords=[f"+{password}"], keys=["*"], channels=["*"], commands=["+@all", "-publish"]
         )
+        restricted = None
         try:
             restricted = _client_as(client, user, password)
             fn = cache(backend=PerRequestRedisBackend(restricted, "default"), ttl=300, namespace="chan_acl")(worker.lookup)
@@ -190,8 +191,9 @@ def test_publish_denied_by_acl_leaves_the_drain_standing(
             assert "invalidating local keys only" not in caplog.text
             assert "Invalidation announcement failed" in caplog.text and "NoPermissionError" in caplog.text
             assert _received(subscriber) == []
-            restricted.close()
         finally:
+            if restricted is not None:
+                restricted.close()
             client.acl_deluser(user)
 
 
@@ -525,6 +527,7 @@ def test_refused_subscription_fails_the_start_and_a_cache_operation_after_the_wi
         monkeypatch.setattr(invalidation, "_START_RETRY_SECONDS", 2.0)
         user, password = "ck-chan-nosub", "ck-chan-nosub-pw"  # pragma: allowlist secret - throwaway ACL user
         client.acl_setuser(user, enabled=True, passwords=[f"+{password}"], keys=["*"], commands=["+@all"], reset_channels=True)
+        restricted = None
         try:
             restricted = _client_as(client, user, password)
             ns = "chan_acl_sub"
@@ -546,8 +549,9 @@ def test_refused_subscription_fails_the_start_and_a_cache_operation_after_the_wi
             _run_peer(client, f"invalidate([1], {ns!r})")
             assert _wait_for(lambda: len(evictions) == 1, timeout=5)
             invalidation._stop_listener()
-            restricted.close()
         finally:
+            if restricted is not None:
+                restricted.close()
             client.acl_deluser(user)
 
     def test_acl_without_ping_fails_the_start(
@@ -564,6 +568,7 @@ def test_acl_without_ping_fails_the_start(
             channels=[invalidation.CHANNEL],
             commands=["+@all", "-ping"],
         )
+        restricted = None
         try:
             restricted = _client_as(client, user, password)
             fn = worker.cached_lookup(restricted, namespace="chan_acl_noping")
@@ -573,8 +578,9 @@ def test_acl_without_ping_fails_the_start(
             assert warning.startswith("Invalidation listener failed to start") and warning.endswith(": NoPermissionError")
             assert invalidation._listener_pid is None and invalidation._listener is None
             assert _subscribers(client) == 0  # the refused listener's connection is closed
-            restricted.close()
         finally:
+            if restricted is not None:
+                restricted.close()
             client.acl_deluser(user)
 
     def test_subscription_revoked_while_running_comes_back_when_granted_again(
@@ -583,6 +589,7 @@ def test_subscription_revoked_while_running_comes_back_when_granted_again(
         monkeypatch.setattr(invalidation, "_RESUBSCRIBE_SECONDS", 0.2)
         user, password = "ck-chan-revoked", "ck-chan-revoked-pw"  # pragma: allowlist secret - throwaway ACL user
         client.acl_setuser(user, enabled=True, passwords=[f"+{password}"], keys=["*"], channels=["*"], commands=["+@all"])
+        restricted = None
         try:
             restricted = _client_as(client, user, password)
             fn = worker.cached_lookup(restricted, namespace="chan_acl_revoked")
@@ -596,6 +603,7 @@ def test_subscription_revoked_while_running_comes_back_when_granted_again(
             client.acl_setuser(user, enabled=True, channels=[invalidation.CHANNEL])
             assert _wait_for(lambda: _subscribers(client) == 1, timeout=10)  # back, with no cache operation
             invalidation._stop_listener()
-            restricted.close()
         finally:
+            if restricted is not None:
+                restricted.close()
             client.acl_deluser(user)