Avoid memory leak with weakref, avoid race condition, log closures (#17)

This suggestion implements my comments on
[1408](interuss#1408) plus a few
additional things:
* Avoids memory leak by using weakref in periodic closure check
* Avoids race condition between closure and new request by adding
_closure_lock
* Informs user when automatic closure occurs via log message, including
differentiating UTMClientSession instances with int ID
* Increases keepalive period to just under max keepalive time likely
used by external infrastructure

Tested in current state and verified a large number of sessions (90+)
and closures. Tested cherrypicking
[1411](interuss#1411) and verified a
smaller number of sessions and closures.

Author: BenjaminPelletier

monitoring/monitorlib/infrastructure.py

@@ -1,14 +1,18 @@
+from __future__ import annotations
+
 import asyncio
 import datetime
 import functools
 import threading
 import time
 import urllib.parse
+import weakref
 from enum import Enum
 
 import jwt
 import requests
 from aiohttp import ClientSession
+from loguru import logger
 
 ALL_SCOPES = [
     "dss.write.identification_service_areas",
@@ -22,7 +26,7 @@
 EPOCH = datetime.datetime.fromtimestamp(0, datetime.UTC)
 TOKEN_REFRESH_MARGIN = datetime.timedelta(seconds=15)
 CLIENT_TIMEOUT = 10  # seconds
-SOCKET_KEEP_ALIVE_LIMIT = 15  # seconds.
+SOCKET_KEEP_ALIVE_LIMIT = 57  # seconds.
 
 
 AuthSpec = str
@@ -89,6 +93,14 @@ class UTMClientSession(requests.Session):
     DSS).
     """
 
+    _next_session_id: int = 1
+    _session_id: int
+    _session_id_lock = threading.Lock()
+
+    _closure_timer: threading.Timer | None = None
+    _closure_lock: threading.Lock
+    _last_used: float | None = None
+
     def __init__(
         self,
         prefix_url: str,
@@ -97,14 +109,64 @@ def __init__(
     ):
         super().__init__()
 
+        with UTMClientSession._session_id_lock:
+            self._session_id = UTMClientSession._next_session_id
+            UTMClientSession._next_session_id += 1
+
         self._prefix_url = prefix_url[0:-1] if prefix_url[-1] == "/" else prefix_url
         self.auth_adapter = auth_adapter
         self.default_scopes: list[str] | None = None
         self.timeout_seconds = timeout_seconds or CLIENT_TIMEOUT
-        self._last_used = None
+        self._closure_lock = threading.Lock()
+
+        UTMClientSession._start_closure_timer(weakref.ref(self))
+
+    @staticmethod
+    def _start_closure_timer(wref: weakref.ReferenceType[UTMClientSession]) -> None:
+        def wrapper():
+            weak_self_final = wref()
+            if weak_self_final is not None:
+                try:
+                    weak_self_final.close_if_idle()
+                finally:
+                    UTMClientSession._start_closure_timer(wref)
+
+        weak_self_initial = wref()
+        if weak_self_initial:
+            weak_self_initial._closure_timer = threading.Timer(
+                weak_self_initial.seconds_until_idle(), wrapper
+            )
+            weak_self_initial._closure_timer.daemon = True
+            weak_self_initial._closure_timer.start()
 
-        t = threading.Thread(target=self._idle_watchdog, daemon=True)
-        t.start()
+    def close_if_idle(self) -> None:
+        with self._closure_lock:
+            if (
+                self._last_used
+                and time.monotonic() - self._last_used > SOCKET_KEEP_ALIVE_LIMIT
+            ):
+                logger.debug(
+                    "Closing idle UTMClientSession {} to {} with default scopes {} last used {} (now {})",
+                    self._session_id,
+                    self._prefix_url,
+                    ", ".join(self.default_scopes) if self.default_scopes else "<none>",
+                    self._last_used,
+                    time.monotonic(),
+                )
+                self.close()
+                self._last_used = None
+
+    def seconds_until_idle(self) -> float:
+        if self._last_used is None:
+            return SOCKET_KEEP_ALIVE_LIMIT
+        else:
+            return max(
+                0.0, SOCKET_KEEP_ALIVE_LIMIT - (time.monotonic() - self._last_used)
+            )
+
+    def __del__(self):
+        if timer := self._closure_timer:
+            timer.cancel()
 
     # Overrides method on requests.Session
     def prepare_request(self, request, **kwargs):
@@ -142,9 +204,10 @@ def request(self, method, url, *args, **kwargs):
         if "auth" not in kwargs:
             kwargs = self.adjust_request_kwargs(kwargs)
 
-        self._last_used = time.monotonic()
-
-        return super().request(method, url, *args, **kwargs)
+        with self._closure_lock:
+            result = super().request(method, url, *args, **kwargs)
+            self._last_used = time.monotonic()
+            return result
 
     def get_prefix_url(self):
         return self._prefix_url
@@ -155,16 +218,6 @@ def get(self, *args, **kwargs):
     def delete(self, *args, **kwargs):
         return super().delete(*args, **kwargs)
 
-    def _idle_watchdog(self):
-        while True:
-            time.sleep(SOCKET_KEEP_ALIVE_LIMIT)
-            if (
-                self._last_used
-                and time.monotonic() - self._last_used > SOCKET_KEEP_ALIVE_LIMIT
-            ):
-                self.close()
-                self._last_used = None
-
 
 class AsyncUTMTestSession:
     """