gh-130141: clean up asyncio._SelectorTransport in __del__

__del__ closed the socket but stopped there: the file descriptor stayed
registered with the event loop and self._sock_fd kept pointing at a
number the OS is now free to hand out to anything else.  So the loop
could go on polling a descriptor the transport no longer owned, and a
transport resurrected during garbage collection could later call
close() and unregister an unrelated descriptor from the selector,
hanging whoever was waiting on it.

Leave the transport in the state that _force_close() and
_call_connection_lost() would instead: clear the buffer and drop the
writer, mark the transport as closing and drop the reader, invalidate
_sock_fd before closing the socket, then unlink the protocol, the loop
and the server.  All of it stays under "if self._sock is not None", so
a finalizer running on a transport that has already lost its socket
still does nothing.  That matters for the server link especially:
_call_connection_lost() calls server._detach() before it clears
self._server, so a _detach() that raises leaves _server set on a
transport whose socket is gone, and an unguarded finalizer would then
detach a second time.

The ResourceWarning is issued first, while repr() can still report the
fd and the selector state.  _call_connection_lost() also sets _sock_fd
to -1 now, so the "_sock_fd is valid only while _sock is open"
invariant holds on every path that closes the socket; as a result
repr() of a closed transport reports fd=-1.

This is the patch Guido van Rossum worked out in the issue thread,
rebased, with the cleanup kept inside the socket guard and extended
with the _buffer_size reset that current main needs, plus the unit
tests the earlier attempt was missing.

Author: v0ropaev

Lib/asyncio/selector_events.py

@@ -878,11 +878,39 @@ def close(self):
             self._call_soon(self._call_connection_lost, None)
 
     def __del__(self, _warn=warnings.warn):
+        # The transport can be resurrected after this runs, so leave it in the
+        # state _force_close() and _call_connection_lost() would: the fd is
+        # gone, and a stale self._sock_fd would later be used to unregister a
+        # descriptor the OS has handed out to somebody else.
         if self._sock is not None:
-            _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
+            # Warn before cleaning up: the message embeds repr(self), which
+            # reports the fd number.
+            if self._protocol_connected:
+                self._protocol_connected = False
+                _warn(f"unclosed transport {self!r}", ResourceWarning,
+                      source=self)
+
+            if self._buffer:
+                self._buffer.clear()
+                self._buffer_size = 0
+                self._loop._remove_writer(self._sock_fd)
+
+            if not self._closing:
+                self._closing = True
+                self._loop._remove_reader(self._sock_fd)
+
+            self._conn_lost += 1
+
+            self._sock_fd = -1
             self._sock.close()
-            if self._server is not None:
-                self._server._detach(self)
+            self._sock = None
+            self._protocol = None
+            self._loop = None
+
+            server = self._server
+            if server is not None:
+                self._server = None
+                server._detach(self)
 
     def _fatal_error(self, exc, message='Fatal error on transport'):
         # Should be called from exception handler only.
@@ -914,8 +942,10 @@ def _force_close(self, exc):
     def _call_connection_lost(self, exc):
         try:
             if self._protocol_connected:
+                self._protocol_connected = False
                 self._protocol.connection_lost(exc)
         finally:
+            self._sock_fd = -1
             self._sock.close()
             self._sock = None
             self._protocol = None

Lib/test/test_asyncio/test_selector_events.py

@@ -542,6 +542,75 @@ def test_force_close(self):
         self.assertFalse(self.loop.readers)
         self.assertEqual(1, self.loop.remove_reader_count[7])
 
+    def test_del(self):
+        tr = self.create_transport()
+        self.loop._add_reader(7, mock.sentinel)
+
+        with self.assertWarns(ResourceWarning):
+            tr.__del__()
+
+        # The socket is closed, so fd 7 may be handed out to an unrelated
+        # file at any moment: the loop must not be left polling it, and the
+        # cached fd must no longer look valid.
+        self.assertFalse(self.loop.readers)
+        self.assertEqual(1, self.loop.remove_reader_count[7])
+        self.assertEqual(-1, tr._sock_fd)
+        self.sock.close.assert_called_with()
+        self.assertIsNone(tr._sock)
+        self.assertIsNone(tr._protocol)
+        self.assertIsNone(tr._loop)
+        self.assertTrue(tr.is_closing())
+
+    def test_del_write_buffer(self):
+        tr = self.create_transport()
+        tr._buffer.extend(b'data')
+        tr._buffer_size = 4
+        self.loop._add_reader(7, mock.sentinel)
+        self.loop._add_writer(7, mock.sentinel)
+
+        with self.assertWarns(ResourceWarning):
+            tr.__del__()
+
+        self.assertFalse(self.loop.readers)
+        self.assertFalse(self.loop.writers)
+        self.assertEqual(1, self.loop.remove_writer_count[7])
+        self.assertEqual(tr._buffer, list_to_buffer())
+        self.assertEqual(0, tr.get_write_buffer_size())
+
+    def test_del_warning_names_the_fd(self):
+        # The warning has to be issued before the cleanup, otherwise its
+        # repr(self) reports a transport that is already closed and the fd
+        # number, the only actionable part of the message, is lost.
+        tr = self.create_transport()
+
+        with self.assertWarns(ResourceWarning) as cm:
+            tr.__del__()
+
+        self.assertIn('fd=7', str(cm.warning))
+
+    def test_del_then_close_leaves_reused_fd_alone(self):
+        tr = self.create_transport()
+        self.loop._add_reader(7, mock.sentinel)
+
+        with self.assertWarns(ResourceWarning):
+            tr.__del__()
+
+        # Something else in the process now owns fd 7 and waits on it.
+        self.loop._add_reader(7, mock.sentinel.other)
+        self.loop._add_writer(7, mock.sentinel.other)
+        self.loop.reset_counters()
+
+        # The transport got resurrected during garbage collection and is
+        # closed properly by its new owner.  It no longer owns fd 7, so it
+        # must keep its hands off the loop.
+        tr.close()
+        tr.abort()
+
+        self.assertEqual(0, self.loop.remove_reader_count[7])
+        self.assertEqual(0, self.loop.remove_writer_count[7])
+        self.assertIs(mock.sentinel.other, self.loop.readers[7]._callback)
+        self.assertIs(mock.sentinel.other, self.loop.writers[7]._callback)
+
     @mock.patch('asyncio.log.logger.error')
     def test_fatal_error(self, m_exc):
         exc = OSError()
@@ -599,6 +668,31 @@ def test__add_reader(self):
         self.assertFalse(self.loop.readers)
 
 
+class SelectorTransportDelTests(test_utils.TestCase):
+    """__del__ against a real event loop and a real socket."""
+
+    def test_del_unregisters_fd_from_the_selector(self):
+        loop = asyncio.SelectorEventLoop()
+        self.set_event_loop(loop)
+        rsock, wsock = socket.socketpair()
+        self.addCleanup(wsock.close)
+        self.addCleanup(rsock.close)
+
+        protocol = test_utils.make_test_protocol(asyncio.Protocol)
+        tr = _SelectorSocketTransport(loop, rsock, protocol)
+        test_utils.run_briefly(loop)  # let connection_made() and _add_reader()
+        fd = tr._sock_fd
+        self.assertIn(fd, loop._selector.get_map())
+
+        with self.assertWarns(ResourceWarning):
+            tr.__del__()
+
+        # Leaving the fd registered here would make the loop poll a
+        # descriptor owned by whatever opens a file next.
+        self.assertNotIn(fd, loop._selector.get_map())
+        self.assertEqual(-1, tr._sock_fd)
+
+
 class SelectorSocketTransportTests(test_utils.TestCase):
 
     def setUp(self):

Misc/NEWS.d/next/Library/2026-09-25-09-29-05.gh-issue-130141.eg0cBm.rst

@@ -0,0 +1,5 @@
+Fix :mod:`asyncio` selector transports unregistering a file descriptor they
+no longer own.  ``_SelectorTransport.__del__`` closed the socket but left the
+descriptor registered with the event loop, so a transport resurrected during
+garbage collection could later remove an unrelated descriptor from the
+selector.  ``repr()`` of a closed transport now reports ``fd=-1``.