Schedule writer.close on loop thread in Connection.close best-effort arm

connection.py:1441-1448 — Connection.close()'s finally block called
writer.close() directly on the calling thread while self._loop is
still running on a different thread (loop.stop is queued via
call_soon_threadsafe several lines later). The sibling
force_close_transport at line 1554-1555 already schedules the same
call via loop.call_soon_threadsafe(_safe_writer_close, writer)
precisely because StreamWriter.close() mutates transport state
(_remove_reader on the selector) and is not thread-safe per stdlib.

Asymmetric discipline within the same file. The current
contextlib.suppress(Exception) masked the symptoms of the race.

Switch to call_soon_threadsafe(_safe_writer_close) when the loop
is alive; fall back to the synchronous call only when the loop
is already closed / unavailable. FIFO ready-queue discipline with
the subsequent loop.stop queue ensures FIN goes out before
run_forever exits — exactly the pattern force_close_transport
already uses.

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

Author: antoineleclair

src/dqlitedbapi/connection.py

@@ -1442,7 +1442,27 @@ def close(self) -> None:
                     inner = self._async_conn
                     proto = getattr(inner, "_protocol", None)
                     writer = getattr(proto, "_writer", None) if proto is not None else None
-                    if writer is not None:
+                    if writer is not None and self._loop is not None and not self._loop.is_closed():
+                        # ``StreamWriter.close()`` mutates transport
+                        # state (calls ``self._loop._remove_reader(...)``
+                        # under the hood) and is documented as not
+                        # thread-safe by stdlib asyncio — touching
+                        # the selector from the calling thread while
+                        # the loop is still running on its own thread
+                        # races with the selector's transport-state
+                        # bookkeeping. Schedule via
+                        # ``call_soon_threadsafe`` like the sibling
+                        # ``force_close_transport`` does (see
+                        # ``connection.py:1554-1555``). FIFO discipline
+                        # of the ready queue with the subsequent
+                        # ``loop.stop`` queue ensures FIN goes out
+                        # before ``run_forever`` exits.
+                        with contextlib.suppress(RuntimeError):
+                            self._loop.call_soon_threadsafe(_safe_writer_close, writer)
+                    elif writer is not None:
+                        # Loop is closed / unavailable — the
+                        # ``_safe_writer_close`` synchronous call
+                        # is the best we can do to flush FIN.
                         with contextlib.suppress(Exception):
                             writer.close()
                     self._async_conn = None