gh-156512: Fix asyncio calling connection_lost() twice from resume_writing()

Lib/asyncio/selector_events.py

@@ -1127,7 +1127,9 @@ def _write_sendmsg(self):
                 self._loop._remove_writer(self._sock_fd)
                 if self._empty_waiter is not None:
                     self._empty_waiter.set_result(None)
-                if self._closing:
+                # gh-156512: don't let _call_connection_lost be called twice
+                if self._closing and not self._conn_lost:
+                    self._conn_lost += 1
                     self._call_connection_lost(None)
                 elif self._eof:
                     self._sock.shutdown(socket.SHUT_WR)
@@ -1173,7 +1175,9 @@ def _write_send(self):
                 self._loop._remove_writer(self._sock_fd)
                 if self._empty_waiter is not None:
                     self._empty_waiter.set_result(None)
-                if self._closing:
+                # gh-156512: don't let _call_connection_lost be called twice
+                if self._closing and not self._conn_lost:
+                    self._conn_lost += 1
                     self._call_connection_lost(None)
                 elif self._eof:
                     self._sock.shutdown(socket.SHUT_WR)

Lib/test/test_asyncio/test_selector_events.py

@@ -1195,6 +1195,49 @@ def _resume_writing():
         self.assertEqual(transport.get_write_buffer_size(), 0)
         self.assertTrue(self.protocol.connection_lost.called)
 
+    def test_write_ready_resume_writing_closes(self):
+        # gh-156512: closing from resume_writing() must not lose the connection twice
+        self.sock.send.return_value = 2
+
+        def _resume_writing():
+            transport.close()
+
+        self.protocol.resume_writing.side_effect = _resume_writing
+        self.loop.call_exception_handler = mock.Mock()
+
+        transport = self.socket_transport()
+        transport.set_write_buffer_limits(high=1, low=0)
+        transport.write(b'data')
+
+        self.loop.writers[7]._run()
+        test_utils.run_briefly(self.loop)
+
+        self.assertEqual(self.protocol.connection_lost.call_count, 1)
+        self.loop.call_exception_handler.assert_not_called()
+
+    @unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
+    def test_write_sendmsg_resume_writing_closes(self):
+        # gh-156512: same as above, for the sendmsg write path
+        self.sock.send.return_value = 2
+        self.sock.sendmsg.return_value = 2
+
+        def _resume_writing():
+            transport.close()
+
+        self.protocol.resume_writing.side_effect = _resume_writing
+        self.loop.call_exception_handler = mock.Mock()
+
+        transport = self.socket_transport(sendmsg=True)
+        transport.set_write_buffer_limits(high=1, low=0)
+        transport.write(b'data')
+
+        self.loop.writers[7]._run()
+        test_utils.run_briefly(self.loop)
+
+        self.assertEqual(self.protocol.connection_lost.call_count, 1)
+        self.loop.call_exception_handler.assert_not_called()
+
+
 class SelectorSocketTransportBufferedProtocolTests(test_utils.TestCase):
 
     def setUp(self):

Misc/NEWS.d/next/Library/2026-08-30-13-25-53.gh-issue-156512.tVTOIV.rst

@@ -0,0 +1,2 @@
+Fix :mod:`asyncio` losing a connection twice when ``resume_writing()``
+closes the transport.