gh-109564: check server._sockets before attaching a new transport

Lib/asyncio/selector_events.py

@@ -225,6 +225,16 @@ async def _accept_connection2(
         protocol = None
         transport = None
         try:
+            # gh-109564: create_task() defers this coroutine's first
+            # step by at least one loop iteration, so the server can
+            # legitimately close() in the gap between "connection
+            # accepted" and "this task actually runs". Attaching to an
+            # already-closed server crashes deep inside transport
+            # construction (Server._attach()'s assert); check up front
+            # instead and drop the already-accepted connection cleanly.
+            if server is not None and server._sockets is None:
+                conn.close()
+                return
             protocol = protocol_factory()
             waiter = self.create_future()
             if sslcontext:

Lib/test/test_asyncio/test_selector_events.py

@@ -421,6 +421,45 @@ def test_accept_connection_reschedules_once_on_resource_error(self):
         self.assertEqual(self.loop.call_exception_handler.call_count, 1)
         self.assertEqual(self.loop.call_later.call_count, 1)
 
+    def test_accept_connection2_server_already_closed(self):
+        # gh-109564: Server.close() sets Server._sockets = None
+        # synchronously, but _accept_connection2() only starts running at
+        # least one event-loop iteration after it's scheduled (it's a
+        # Task). If close() lands in that gap, attaching a transport to
+        # the now-closed server used to raise an uncatchable
+        # AssertionError deep inside transport construction instead of
+        # just dropping the already-accepted connection.
+        conn = mock.Mock()
+        protocol_factory = mock.Mock()
+        server = mock.Mock()
+        server._sockets = None
+
+        coro = self.loop._accept_connection2(
+            protocol_factory, conn, {}, server=server)
+        self.loop.run_until_complete(coro)
+
+        conn.close.assert_called_with()
+        protocol_factory.assert_not_called()
+
+    def test_accept_connection2_no_server(self):
+        # _accept_connection2's server parameter defaults to None (no
+        # Server object to check), which must not itself raise -- only
+        # an actual closed Server should short-circuit.
+        conn = mock.Mock()
+        protocol = mock.Mock()
+        protocol_factory = mock.Mock(return_value=protocol)
+        waiter = self.loop.create_future()
+        waiter.set_result(None)
+        self.loop.create_future = mock.Mock(return_value=waiter)
+        self.loop._make_socket_transport = mock.Mock()
+
+        coro = self.loop._accept_connection2(
+            protocol_factory, conn, {}, server=None)
+        self.loop.run_until_complete(coro)
+
+        protocol_factory.assert_called_with()
+        conn.close.assert_not_called()
+
 class SelectorTransportTests(test_utils.TestCase):
 
     def setUp(self):

Misc/NEWS.d/next/Library/2026-07-29-08-01-59.gh-issue-109564.PWM3XU.rst

@@ -0,0 +1,4 @@
+Fix a race in :class:`asyncio.Server` where calling :meth:`~asyncio.Server.close`
+while a connection was in the process of being accepted could raise an
+uncatchable :exc:`AssertionError` deep inside transport creation instead of
+simply dropping the connection.