fix(proxy): match a reloaded service class by name and dedup warnings per run

A hot service reload made instance_of miss the new class, so a run with no proxy got a false bypass warning.

tests/core/test_run_proxy.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import sys
+import weakref
 
 import pytest
 import requests
@@ -26,12 +27,31 @@ class Svc:
     def fallback(self) -> requests.Session:
         return run_proxy.fallback_session()
 
+    def helper_lookup(self, host: str) -> None:
+        Helper().lookup(host)
+
+    def helper_fallback(self) -> requests.Session:
+        return Helper().fallback()
+
+
+class Helper:
+    """A helper class of the service with its own HTTP session, which has no proxy."""
+
+    def __init__(self):
+        self.session = requests.Session()
+
+    def lookup(self, host: str) -> None:
+        lookup(host)
+
+    def fallback(self) -> requests.Session:
+        return run_proxy.fallback_session()
+
 
 @pytest.fixture(autouse=True)
 def watching(monkeypatch):
     monkeypatch.setattr(run_proxy, "watched", {})
     monkeypatch.setattr(run_proxy, "proxy_hosts", set())
-    monkeypatch.setattr(run_proxy, "warned", set())
+    monkeypatch.setattr(run_proxy, "warned", weakref.WeakKeyDictionary())
     run_proxy.watch(Svc, PROXY)
 
 
@@ -149,6 +169,39 @@ def test_service_instance_uses_the_proxy_of_its_own_run(caplog):
     assert {r.name for r in caplog.records if r.getMessage() in leaks(caplog)} == {"Svc"}
 
 
+def test_a_reloaded_service_class_uses_the_proxy_of_its_own_run(caplog):
+    """A service reload makes a new class from the same directory; the watch still holds the old one."""
+    reloaded = type("Svc", (), {"__init__": Svc.__init__, "lookup": Svc.lookup})
+    with caplog.at_level(logging.WARNING):
+        reloaded().lookup("direct.example")  # a run with no proxy
+        reloaded(PROXY).lookup("leak.example")
+    assert len(leaks(caplog)) == 1
+    assert "leak.example" in leaks(caplog)[0]
+
+
+def test_a_subclass_of_the_service_uses_the_proxy_of_its_own_run(caplog):
+    sub = type("Sub", (Svc,), {})
+    with caplog.at_level(logging.WARNING):
+        sub().lookup("direct.example")
+    assert leaks(caplog) == []
+
+
+def test_a_helper_session_without_proxy_does_not_hide_a_leak(caplog):
+    """A helper class of the service is not the service instance, so the proxy of the run still decides."""
+    with caplog.at_level(logging.WARNING):
+        Svc(PROXY).helper_lookup("leak.example")
+    assert len(leaks(caplog)) == 1
+    assert Svc(PROXY).helper_fallback().proxies == {"all": PROXY}
+
+
+def test_each_run_warns_even_when_a_finished_run_had_the_same_id(caplog):
+    """CPython reuses the id of a freed instance, so the warned hosts belong to the instance itself."""
+    with caplog.at_level(logging.WARNING):
+        for _ in range(2):
+            Svc(PROXY).lookup("leak.example")
+    assert len(leaks(caplog)) == 2
+
+
 def test_fallback_session_takes_the_proxy_of_the_calling_run():
     session = Svc("http://own.example:1").fallback()
     assert session.proxies == {"all": "http://own.example:1"}

unshackle/core/utils/run_proxy.py

@@ -10,6 +10,7 @@ import ipaddress
 import logging
 import os
 import sys
+import weakref
 from dataclasses import dataclass, field
 from types import FrameType
 from typing import Any, Optional
@@ -30,7 +31,7 @@ class Watch:
 
 watched: dict[str, Watch] = {}  # service directory -> Watch
 proxy_hosts: set[str] = set()
-warned: set[tuple[str, str, int]] = set()
+warned: weakref.WeakKeyDictionary[Any, set[str]] = weakref.WeakKeyDictionary()
 hook_installed = False
 
 
@@ -99,10 +100,15 @@ def instance_of(frame: Optional[FrameType], entry: Watch) -> Any:
     """The service instance nearest to ``frame``, which knows the proxy of its own run.
 
     The walk goes on past helper functions of the service, which have no ``self``.
+    It matches the class by module and qualified name, because a service reload makes a new class
+    that ``entry.cls`` is not.
     """
+    name = (entry.cls.__module__, entry.cls.__qualname__)
     while frame is not None:
         inst = frame.f_locals.get("self")
-        if isinstance(inst, entry.cls) and getattr(inst, "session", None) is not None:
+        if getattr(inst, "session", None) is not None and any(
+            (c.__module__, c.__qualname__) == name for c in type(inst).__mro__
+        ):
             return inst
         frame = frame.f_back
     return None
@@ -198,10 +204,10 @@ def check(host: Any, frame: Optional[FrameType]) -> None:
     inst = instance_of(service_frame, entry)
     if inst is not None and not proxy_of(getattr(inst, "session", None)):
         return
-    key = (entry.cls.__name__, host, id(inst))
-    if key in warned:
+    hosts = warned.setdefault(entry.cls if inst is None else inst, set())
+    if host in hosts:
         return
-    warned.add(key)
+    hosts.add(host)
     # a service instance logs through self.log, which a remote session mirrors to its client
     log = getattr(inst, "log", None) or logging.getLogger(entry.cls.__name__)
     where = os.path.relpath(service_frame.f_code.co_filename, os.path.dirname(entry.directory[:-1]))