refactor(jsonrpc2): harden resource lifecycle without __del__

- remove __del__-based resource cleanup from server, debugger client, and protocol
- move shutdown to explicit close/close_async paths
- make JsonRPCServer.close wait for server shutdown when safe
- add best-effort weakref finalizers for forgotten cleanup without noisy warnings
- unify runtime and finalizer bookkeeping into shared state objects

Author: d-biehl

packages/debugger/src/robotcode/debugger/launcher/client.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import asyncio
+import logging
+import weakref
 from typing import Any, Optional, Sequence
 
 from robotcode.core.event import event
@@ -38,6 +40,17 @@ class DAPClientError(Exception):
     pass
 
 
+class _DAPClientState:
+    __slots__ = ("closed", "logger", "loop", "protocol", "transport")
+
+    def __init__(self, logger: logging.Logger) -> None:
+        self.logger = logger
+        self.loop: Optional[asyncio.AbstractEventLoop] = None
+        self.protocol: Optional[DAPClientProtocol] = None
+        self.transport: Optional[asyncio.BaseTransport] = None
+        self.closed = False
+
+
 class DAPClient:
     _logger = LoggingDescriptor()
 
@@ -48,8 +61,43 @@ def __init__(
     ) -> None:
         self.parent = parent
         self.tcp_params = tcp_params
-        self._protocol: Optional[DAPClientProtocol] = None
-        self._transport: Optional[asyncio.BaseTransport] = None
+        self._state = _DAPClientState(
+            logging.getLogger(f"{type(self).__module__}.{type(self).__qualname__}"),
+        )
+        self._finalizer = weakref.finalize(self, DAPClient._finalize_resources, self._state)
+
+    @staticmethod
+    def _finalize_resources(state: _DAPClientState) -> None:
+        if state.closed or state.transport is None:
+            return
+
+        try:
+            if state.loop is not None and not state.loop.is_closed() and state.loop.is_running():
+                state.loop.call_soon_threadsafe(state.transport.close)
+            else:
+                state.transport.close()
+        except BaseException:
+            pass
+
+        state.logger.debug(
+            "DAPClient was garbage collected without calling close(); the transport was closed best-effort only.",
+        )
+
+    @property
+    def _protocol(self) -> Optional[DAPClientProtocol]:
+        return self._state.protocol
+
+    @_protocol.setter
+    def _protocol(self, value: Optional[DAPClientProtocol]) -> None:
+        self._state.protocol = value
+
+    @property
+    def _transport(self) -> Optional[asyncio.BaseTransport]:
+        return self._state.transport
+
+    @_transport.setter
+    def _transport(self, value: Optional[asyncio.BaseTransport]) -> None:
+        self._state.transport = value
 
     @event
     def on_closed(sender) -> None: ...
@@ -61,11 +109,10 @@ def close(self) -> None:
             self._transport = None
             self._protocol = None
 
+        self._state.closed = True
+        self._finalizer.detach()
         self.on_closed(self)
 
-    def __del__(self) -> None:
-        self.close()
-
     @_logger.call
     def on_connection_lost(self, sender: Any, exc: Optional[BaseException]) -> None:
         if sender == self._protocol:
@@ -76,6 +123,8 @@ async def connect(self, timeout: float = 5) -> DAPClientProtocol:
         async def wait() -> None:
             while self._protocol is None:
                 try:
+                    current_loop = asyncio.get_running_loop()
+                    self._state.loop = current_loop
                     if self.tcp_params.host is not None:
                         if isinstance(self.tcp_params.host, Sequence):
                             host = self.tcp_params.host[0]
@@ -86,7 +135,7 @@ async def wait() -> None:
                     (
                         self._transport,
                         protocol,
-                    ) = await asyncio.get_running_loop().create_connection(
+                    ) = await current_loop.create_connection(
                         self._create_protocol,
                         host=host,
                         port=self.tcp_params.port,

packages/debugger/src/robotcode/debugger/launcher/server.py

@@ -85,6 +85,11 @@ def client(self, value: DAPClient) -> None:
     def connected(self) -> bool:
         return self._client is not None and self._client.connected
 
+    def close(self) -> None:
+        if self._client is not None:
+            self._client.close()
+            self._client = None
+
     @rpc_method(name="initialize", param_type=InitializeRequestArguments)
     async def _initialize(self, arguments: InitializeRequestArguments, *args: Any, **kwargs: Any) -> Capabilities:
         self._initialize_arguments = arguments
@@ -352,3 +357,9 @@ def create_protocol(self) -> LauncherDebugAdapterProtocol:
         self.protocol = LauncherDebugAdapterProtocol(debugger_script=self.debugger_script)
 
         return self.protocol
+
+    def _close(self) -> None:
+        if self.protocol is not None:
+            self.protocol.close()
+
+        super()._close()

packages/jsonrpc2/src/robotcode/jsonrpc2/server.py

@@ -1,8 +1,10 @@
 import abc
 import asyncio
 import io
+import logging
 import sys
 import threading
+import weakref
 from concurrent.futures import ThreadPoolExecutor
 from types import TracebackType
 from typing import (
@@ -50,6 +52,18 @@ def write(self, data: "bytes | bytearray | memoryview[Any]") -> None:
         self.wfile.flush()
 
 
+class _JsonRPCServerState:
+    __slots__ = ("closed", "in_closing", "logger", "loop", "server", "stdio_stop_event")
+
+    def __init__(self, loop: asyncio.AbstractEventLoop, logger: logging.Logger) -> None:
+        self.logger = logger
+        self.loop = loop
+        self.server: Optional[asyncio.AbstractServer] = None
+        self.stdio_stop_event: Optional[threading.Event] = None
+        self.in_closing = False
+        self.closed = False
+
+
 class JsonRPCServer(Generic[TProtocol], abc.ABC):
     _logger = LoggingDescriptor()
 
@@ -65,24 +79,78 @@ def __init__(
 
         self._run_func: Optional[Callable[[], None]] = None
         self._serve_func: Optional[Callable[[], Coroutine[None, None, None]]] = None
-        self._server: Optional[asyncio.AbstractServer] = None
-
-        self._stdio_stop_event: Optional[threading.Event] = None
-
-        self._in_closing = False
-        self._closed = False
 
         try:
-            self.loop = asyncio.get_event_loop()
+            loop = asyncio.get_event_loop()
         except RuntimeError:
-            self.loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(self.loop)
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+        self._state = _JsonRPCServerState(
+            loop,
+            logging.getLogger(f"{type(self).__module__}.{type(self).__qualname__}"),
+        )
+        self._finalizer = weakref.finalize(self, JsonRPCServer._finalize_resources, self._state)
 
         if self.loop is not None:
             self.loop.slow_callback_duration = 10
 
-    def __del__(self) -> None:
-        self.close()
+    @staticmethod
+    def _finalize_resources(state: _JsonRPCServerState) -> None:
+        if state.closed:
+            return
+
+        if state.stdio_stop_event is not None:
+            state.stdio_stop_event.set()
+
+        if state.server is not None:
+            try:
+                if not state.loop.is_closed() and state.loop.is_running():
+                    state.loop.call_soon_threadsafe(state.server.close)
+                else:
+                    state.server.close()
+            except BaseException:
+                pass
+
+        state.logger.debug(
+            "JsonRPCServer was garbage collected without calling close(); resources were cleaned up best-effort only.",
+        )
+
+    @property
+    def loop(self) -> asyncio.AbstractEventLoop:
+        return self._state.loop
+
+    @property
+    def _server(self) -> Optional[asyncio.AbstractServer]:
+        return self._state.server
+
+    @_server.setter
+    def _server(self, value: Optional[asyncio.AbstractServer]) -> None:
+        self._state.server = value
+
+    @property
+    def _stdio_stop_event(self) -> Optional[threading.Event]:
+        return self._state.stdio_stop_event
+
+    @_stdio_stop_event.setter
+    def _stdio_stop_event(self, value: Optional[threading.Event]) -> None:
+        self._state.stdio_stop_event = value
+
+    @property
+    def _in_closing(self) -> bool:
+        return self._state.in_closing
+
+    @_in_closing.setter
+    def _in_closing(self, value: bool) -> None:
+        self._state.in_closing = value
+
+    @property
+    def _closed(self) -> bool:
+        return self._state.closed
+
+    @_closed.setter
+    def _closed(self, value: bool) -> None:
+        self._state.closed = value
 
     @_logger.call
     def start(self) -> None:
@@ -122,17 +190,49 @@ def _close(self) -> None:
         if self._server and self._server.is_serving():
             self._server.close()
 
+    def _close_on_loop_thread(self) -> None:
+        if self.loop.is_running() and getattr(self.loop, "_thread_id", None) != threading.get_ident():
+            closed_event = threading.Event()
+
+            def close_on_loop() -> None:
+                try:
+                    self._close()
+                finally:
+                    closed_event.set()
+
+            self.loop.call_soon_threadsafe(close_on_loop)
+            closed_event.wait()
+            return
+
+        self._close()
+
+    def _wait_closed_sync(self) -> None:
+        if self._server is None or self.loop.is_closed():
+            return
+
+        if self.loop.is_running():
+            if getattr(self.loop, "_thread_id", None) == threading.get_ident():
+                asyncio.create_task(self._server.wait_closed(), name=f"{type(self).__name__}.wait_closed")
+                return
+
+            asyncio.run_coroutine_threadsafe(self._server.wait_closed(), self.loop).result()
+            return
+
+        self.loop.run_until_complete(self._server.wait_closed())
+
     @_logger.call
     def close(self) -> None:
         if self._in_closing or self._closed:
             return
 
         self._in_closing = True
         try:
-            self._close()
+            self._close_on_loop_thread()
+            self._wait_closed_sync()
         finally:
             self._in_closing = False
             self._closed = True
+            self._finalizer.detach()
 
     @_logger.call
     async def close_async(self) -> None:
@@ -147,6 +247,7 @@ async def close_async(self) -> None:
         finally:
             self._in_closing = False
             self._closed = True
+            self._finalizer.detach()
 
     async def __aenter__(self) -> Self:
         await self.start_async()

packages/language_server/src/robotcode/language_server/common/protocol.py

@@ -161,9 +161,6 @@ def __init__(self, server: JsonRPCServer[Any]):
 
         self.is_initialized = Event()
 
-    def __del__(self) -> None:
-        self.trace = TraceValues.OFF
-
     @event
     def on_shutdown(sender) -> None:  # pragma: no cover, NOSONAR
         ...