fix(l1): at-fork hook repairs every cache even when its drop warning raises (LAB-6370) (#392)

Author: 27Bslash6

docs/free-threading.md

@@ -56,7 +56,7 @@ the ticket, plus what the free-threaded CI lane surfaced and sites added since:
 | `backends/cachekitio/session.py` header cache | `threading.local` | Safe — per-thread state; its only cross-thread hazard was the session-identity publication above |
 | `decorators/stats_context.py` | `contextvars.ContextVar` | Safe by construction |
 | `decorators/wrapper.py` L1/SWR + L2/SWR single-flight (`_l1_swr_*`, `_l2_swr_*`) | `BoundedSemaphore` slots + in-flight `set` + PID-owner swap | Safe. The check-then-add on the in-flight set was already documented as benign (worst case one duplicate refresh, absorbed by last-write-wins / the backend lease); builtin `set`/`dict` single ops are atomic under free-threading's per-object locking. The fork-detection wholesale swap races are the same documented-benign shape |
-| `l1_cache.py` `L1CacheManager._take_over_if_forked` + `_reset_cache_locks_after_fork` (LAB-4772) | Owner-PID fast path; on a PID change one thread takes over under a lock fetched by `dict.setdefault(pid, Lock())`. An `os.register_at_fork` hook repairs cache locks orphaned at fork before the child's first `get()`; fork mechanics and the uWSGI limit are in the `_take_over_if_forked` docstring | Safe. `setdefault` is atomic under per-object locking, so every thread in a new child gets the same fresh lock, which no parent thread can have held at fork; the rest block on it and re-check. The fast path gates on `_owner_pid` alone, so a weak-memory reader may still see the inherited manager lock once: the same benign fork-swap race as the SWR row, since that lock is unheld unless a parent thread held it at fork. The hook runs while the child is single-threaded, so its non-blocking probe (`timeout=0`) is exact, and it starts no thread. Once it has run, the take-over skips its own timed cache-lock probe, which could otherwise clear a cache under a live child thread |
+| `l1_cache.py` `L1CacheManager._take_over_if_forked` + `_reset_cache_locks_after_fork` (LAB-4772) | Owner-PID fast path; on a PID change one thread takes over under a lock fetched by `dict.setdefault(pid, Lock())`. An `os.register_at_fork` hook repairs cache locks orphaned at fork before the child's first `get()`; fork mechanics and the uWSGI limit are in the `_take_over_if_forked` docstring | Safe. `setdefault` is atomic under per-object locking, so every thread in a new child gets the same fresh lock, which no parent thread can have held at fork; the rest block on it and re-check. The fast path gates on `_owner_pid` alone, so a weak-memory reader may still see the inherited manager lock once: the same benign fork-swap race as the SWR row, since that lock is unheld unless a parent thread held it at fork. The hook runs while the child is single-threaded, so its non-blocking probe (`timeout=0`) is exact, and it starts no thread. It repairs a cache before logging that cache's drop warning, and carries on past a log call that raises, so every cache is repaired. Once it has run, the take-over skips its own timed cache-lock probe, which could otherwise clear a cache under a live child thread |
 | `decorators/wrapper.py` `_cached_keys` | Builtin `set`, snapshot-copied before iteration in invalidation | Safe — single ops atomic; a copy racing an add can only miss a concurrently-written key, which invalidate-all semantics tolerate |
 | `object_cache.py` `ObjectCache` | `RLock` on every public method | Safe |
 | `reliability/metrics_collection.py` `get_async_metrics_collector` | Double-checked module-global singleton | Safe — single-assignment publication of a fully-constructed object; worst case a benign duplicate worker-restart check |

src/cachekit/l1_cache.py

@@ -260,16 +260,18 @@ def _reset_lock_after_fork(self, timeout: float = 1.0) -> None:
         if not lock._is_owned() and lock.acquire(timeout=timeout):  # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
             lock.release()
             return
-        logger.warning(
-            "L1Cache %s: dropped %d entries (%d bytes) after fork; its lock was unavailable",
-            self.namespace,
-            len(self._cache),
-            self._current_memory_bytes,
-        )
+        dropped, dropped_bytes = len(self._cache), self._current_memory_bytes
         # Clear before publishing the new lock: the orphaned one still shuts every other thread out.
         self._cache.clear()
         self._current_memory_bytes = 0
         self._lock = threading.RLock()
+        # Log only once repaired: a raising logging Filter escapes Logger.handle.
+        logger.warning(
+            "L1Cache %s: dropped %d entries (%d bytes) after fork; its lock was unavailable",
+            self.namespace,
+            dropped,
+            dropped_bytes,
+        )
 
     def _remove_entry(self, key: str) -> None:
         """Remove entry from cache and update memory tracking.
@@ -451,6 +453,8 @@ def _take_over_if_forked(self) -> None:
         they put(), so _reset_cache_locks_after_fork repairs orphaned cache locks before that
         first get() on os.fork() servers; without at-fork hooks (uWSGI unless
         --py-call-osafterfork) a get() on an orphaned cache lock before the first put still hangs.
+        The take-over resets cache locks only when that hook did not run in this PID
+        (_locks_reset_pid): after it, a held cache lock belongs to a live child thread.
         """
         pid = os.getpid()
         if self._owner_pid == pid:
@@ -594,10 +598,16 @@ def _reset_cache_locks_after_fork() -> None:
     Only the locks: starting the cleanup thread stays with _take_over_if_forked, outside the hook.
     """
     pid = os.getpid()
+    error: Optional[Exception] = None
     for manager in list(_managers):
         for cache in list(manager._caches.values()):
-            cache._reset_lock_after_fork(timeout=0)
+            try:
+                cache._reset_lock_after_fork(timeout=0)
+            except Exception as e:  # its drop warning, logged after the repair: repair the rest too
+                error = error or e
         manager._locks_reset_pid = pid
+    if error is not None:
+        raise error  # only now; Python reports an at-fork hook's exception and carries on
 
 
 if hasattr(os, "register_at_fork"):

tests/unit/test_l1_memory_bounds.py

@@ -359,7 +359,7 @@ def hold() -> None:
 
                 holder = threading.Thread(target=hold)
                 holder.start()
-                held.wait(5)
+                assert held.wait(5)  # else the take-over below never meets a live holder and the test proves nothing
                 other.put("k", b"v")  # first put runs the take-over while a live child thread holds busy-ns
                 release.set()
                 holder.join(5)
@@ -379,6 +379,98 @@ def hold() -> None:
         finally:
             manager.stop_background_cleanup()
 
+    @pytest.mark.skipif(not hasattr(os, "fork"), reason="fork() not available on this platform")
+    @pytest.mark.parametrize("two_managers", [False, True], ids=["one-manager", "two-managers"])
+    def test_at_fork_hook_repairs_every_cache_when_its_warning_raises(self, two_managers):
+        import multiprocessing
+        import queue as queue_mod
+
+        from cachekit import l1_cache
+
+        first = L1CacheManager(default_max_memory_mb=10)
+        second = L1CacheManager(default_max_memory_mb=10) if two_managers else first
+        caches = [first.get_cache("raising-ns"), second.get_cache("later-ns")]
+
+        class RaiseOnDropWarning(logging.Filter):
+            def filter(self, record: logging.LogRecord) -> bool:
+                # Only this test's first cache: a drop warning from any other live manager passes.
+                if record.args and record.args[0] == "raising-ns":
+                    raise RuntimeError("filter failed")
+                return True
+
+        raising = RaiseOnDropWarning()
+        l1_cache.logger.addFilter(raising)
+        held, release = [threading.Event() for _ in caches], threading.Event()
+
+        def hold(cache: L1Cache, done: threading.Event) -> None:
+            with cache._lock:
+                done.set()
+                release.wait()
+
+        holders = [threading.Thread(target=hold, args=pair, daemon=True) for pair in zip(caches, held, strict=True)]
+        for holder in holders:
+            holder.start()
+        assert all(event.wait(5) for event in held)
+        try:
+            ctx = multiprocessing.get_context("fork")
+            queue = ctx.Queue()
+
+            def child(q) -> None:
+                # From the child's main thread: a lock the hook left orphaned hangs here for good.
+                found = [cache.get("k")[0] for cache in caches]
+                q.put({"found": found, "reset": [m._locks_reset_pid == os.getpid() for m in (first, second)]})
+
+            process = ctx.Process(target=child, args=(queue,))
+            process.start()
+            try:
+                outcome = queue.get(timeout=20)
+            except queue_mod.Empty:
+                outcome = "child hung on a cache lock the at-fork hook left orphaned"
+            finally:
+                process.join(timeout=10)
+                if process.is_alive():  # a hung child would otherwise block pytest's exit
+                    process.kill()
+
+            assert outcome == {"found": [False, False], "reset": [True, True]}
+            assert process.exitcode == 0
+        finally:
+            l1_cache.logger.removeFilter(raising)
+            release.set()
+            for holder in holders:
+                holder.join(5)
+
+    def test_at_fork_hook_reraises_only_after_repairing_every_cache(self, monkeypatch):
+        import weakref
+
+        from cachekit import l1_cache
+
+        manager = L1CacheManager(default_max_memory_mb=10)
+        monkeypatch.setattr(l1_cache, "_managers", weakref.WeakSet([manager]))  # leave the global manager alone
+        caches = [manager.get_cache("raising-ns"), manager.get_cache("later-ns")]
+        orphaned = []
+        for cache in caches:
+            cache.put("pre-fork", b"v")
+            cache._lock.acquire()  # _is_owned(): the hook resets it, as for a hold orphaned at fork
+            orphaned.append(cache._lock)
+
+        class RaiseOnDropWarning(logging.Filter):
+            def filter(self, record: logging.LogRecord) -> bool:
+                if record.args and record.args[0] == "raising-ns":
+                    raise RuntimeError("filter failed")
+                return True
+
+        raising = RaiseOnDropWarning()
+        l1_cache.logger.addFilter(raising)
+        try:
+            with pytest.raises(RuntimeError, match="filter failed"):  # reported, not swallowed
+                l1_cache._reset_cache_locks_after_fork()
+        finally:
+            l1_cache.logger.removeFilter(raising)
+
+        assert [cache._lock is lock for cache, lock in zip(caches, orphaned, strict=True)] == [False, False]
+        assert [cache.get("pre-fork")[0] for cache in caches] == [False, False]
+        assert manager._locks_reset_pid == os.getpid()
+
     def test_cleanup_stopped_in_parent_stays_stopped(self):
         manager = L1CacheManager(default_max_memory_mb=10)
         cache = manager.get_cache("stopped-ns")
@@ -393,6 +485,7 @@ def test_orphaned_cache_lock_drop_is_exclusive_and_logged(self, caplog):
         manager = L1CacheManager(default_max_memory_mb=10)
         cache = manager.get_cache("orphan-ns")
         cache.put("pre-fork", b"v")
+        held_bytes = cache._current_memory_bytes
         _as_if_forked(manager, parent_ran_cleanup=False)
         orphaned = cache._lock
         orphaned.acquire()  # _is_owned(): how a child thread reusing the dead holder's ident sees the hold
@@ -407,7 +500,8 @@ def clear(self) -> None:
             cache.put("k", b"v")
 
         assert not cache.get("pre-fork")[0] and cache.get("k")[0]
-        assert any("dropped 1 entries" in r.message for r in caplog.records)
+        # Logged after the clear, but with what the cache held before it.
+        assert any(f"dropped 1 entries ({held_bytes} bytes)" in r.message for r in caplog.records)
 
     def test_restart_failure_is_logged_once_and_put_still_stores(self, monkeypatch, caplog):
         manager = L1CacheManager(default_max_memory_mb=10)