weakref.finalize cleans up event-loop thread on Connection GC

ISSUE-10 — the dedicated event-loop thread created by _ensure_loop
used to leak forever if the Connection was garbage-collected without
explicit close() (exception during use, forgotten `with`, tests).
Long-lived processes accumulated dozens of daemon threads.

Now _ensure_loop registers a weakref.finalize; the finalizer stops
the loop, joins the thread, and emits ResourceWarning (matches stdlib
sqlite3 convention) so leaks are visible in dev and test. The
finalizer captures a 1-element ``closed_flag`` list that close()
mutates, so explicit close() suppresses the warning and detaches
the finalizer.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqlitedbapi/connection.py

@@ -3,13 +3,58 @@
 import asyncio
 import contextlib
 import threading
+import warnings
+import weakref
 from typing import Any
 
 from dqliteclient import DqliteConnection
 from dqlitedbapi.cursor import Cursor
 from dqlitedbapi.exceptions import InterfaceError, OperationalError, ProgrammingError
 
 
+def _cleanup_loop_thread(
+    loop: asyncio.AbstractEventLoop,
+    thread: threading.Thread,
+    closed_flag: list[bool],
+    address: str,
+) -> None:
+    """Stop the background event loop and join its thread.
+
+    Called from a ``weakref.finalize`` so it must not reference the
+    ``Connection`` instance. ``closed_flag`` is a 1-element list that
+    the Connection mutates when ``close()`` is called — we use that
+    rather than a direct reference to self to decide whether to emit
+    a ``ResourceWarning``.
+    """
+    if closed_flag[0] is False:
+        # User never called close() → leak warning (matches stdlib
+        # sqlite3). Don't crash at interpreter shutdown.
+        try:
+            warnings.warn(
+                f"Connection(address={address!r}) was garbage-collected "
+                f"without close(); cleaning up event-loop thread. Call "
+                f"Connection.close() explicitly to avoid this warning.",
+                ResourceWarning,
+                stacklevel=2,
+            )
+        except Exception:
+            pass
+    try:
+        if not loop.is_closed():
+            loop.call_soon_threadsafe(loop.stop)
+    except Exception:
+        pass
+    try:
+        thread.join(timeout=5)
+    except Exception:
+        pass
+    try:
+        if not loop.is_closed():
+            loop.close()
+    except Exception:
+        pass
+
+
 class Connection:
     """PEP 249 compliant database connection."""
 
@@ -38,6 +83,11 @@ def __init__(
         self._op_lock = threading.Lock()
         self._connect_lock: asyncio.Lock | None = None
         self._creator_thread = threading.get_ident()
+        # 1-element list (mutable, captured by the finalizer) that
+        # close() flips to True. Using a list avoids the finalizer
+        # closing over ``self`` and preventing GC.
+        self._closed_flag: list[bool] = [False]
+        self._finalizer: weakref.finalize | None = None
 
     def _check_thread(self) -> None:
         """Raise ProgrammingError if called from a different thread than the creator."""
@@ -54,6 +104,11 @@ def _ensure_loop(self) -> asyncio.AbstractEventLoop:
 
         This allows sync methods to work even when called from within
         an already-running async context (e.g. uvicorn).
+
+        Registers a ``weakref.finalize`` the first time the loop is
+        created so a Connection that's garbage-collected without an
+        explicit ``close()`` still cleans up its thread. (GC'd connections
+        used to leak daemon threads forever.)
         """
         if self._loop is not None and not self._loop.is_closed():
             return self._loop
@@ -62,6 +117,18 @@ def _ensure_loop(self) -> asyncio.AbstractEventLoop:
                 self._loop = asyncio.new_event_loop()
                 self._thread = threading.Thread(target=self._loop.run_forever, daemon=True)
                 self._thread.start()
+                # Finalizer can't close over self — it'd keep the
+                # Connection alive. Capture primitives only. The
+                # closed-flag list is mutated by close() so the
+                # finalizer knows whether to emit a leak warning.
+                self._finalizer = weakref.finalize(
+                    self,
+                    _cleanup_loop_thread,
+                    self._loop,
+                    self._thread,
+                    self._closed_flag,
+                    self._address,
+                )
         return self._loop
 
     def _run_sync(self, coro: Any) -> Any:
@@ -134,6 +201,14 @@ def close(self) -> None:
         if self._closed:
             return
         self._closed = True
+        # Flip the flag the finalizer reads so it knows this was an
+        # explicit close (no ResourceWarning).
+        self._closed_flag[0] = True
+        # Detach the finalizer — it's about to do nothing useful, and
+        # keeping it registered would double-stop the loop.
+        if self._finalizer is not None:
+            self._finalizer.detach()
+            self._finalizer = None
         try:
             if self._loop is not None and not self._loop.is_closed():
                 with contextlib.suppress(Exception):

tests/test_gc_cleanup.py

@@ -0,0 +1,57 @@
+"""Event-loop thread is cleaned up on Connection GC (ISSUE-10).
+
+Previously Connection spawned a daemon thread via _ensure_loop() and
+only stopped it on explicit close(). A GC'd connection leaked its
+thread forever, which bit long-lived processes (web servers, workers)
+accumulating dozens/hundreds of leaked loops over time.
+
+Now _ensure_loop registers a weakref.finalize that stops the loop and
+joins the thread when the Connection is GC'd, and emits ResourceWarning
+to mirror stdlib sqlite3 conventions.
+"""
+
+import gc
+import threading
+import warnings
+
+import pytest
+
+from dqlitedbapi.connection import Connection
+
+
+class TestGCCleanup:
+    def test_gc_cleans_up_loop_thread(self) -> None:
+        """Connection without explicit close() gets cleaned up on GC."""
+        baseline = threading.active_count()
+        # Create a connection and force the background loop to exist.
+        conn = Connection("localhost:19001", timeout=2.0)
+        conn._ensure_loop()
+        assert threading.active_count() == baseline + 1
+
+        # GC without close(). Expect ResourceWarning.
+        with warnings.catch_warnings(record=True) as captured:
+            warnings.simplefilter("always")
+            del conn
+            gc.collect()
+            # Give the finalizer thread time to join.
+            for t in threading.enumerate():
+                if t.daemon and t is not threading.current_thread():
+                    t.join(timeout=1.0)
+
+        # At least one ResourceWarning about the leak.
+        rw = [w for w in captured if issubclass(w.category, ResourceWarning)]
+        assert rw, f"expected ResourceWarning; got categories={[w.category for w in captured]}"
+
+    def test_explicit_close_suppresses_resourcewarning(self) -> None:
+        """If the user calls close(), finalizer shouldn't emit a warning."""
+        with warnings.catch_warnings(record=True) as captured:
+            warnings.simplefilter("always")
+            conn = Connection("localhost:19001", timeout=2.0)
+            conn._ensure_loop()
+            conn.close()
+            del conn
+            gc.collect()
+
+        # No ResourceWarning since the user did the right thing.
+        rw = [w for w in captured if issubclass(w.category, ResourceWarning)]
+        assert not rw, f"unexpected ResourceWarning: {[str(w.message) for w in rw]}"