Fix SocketStream.aclose() abort() race on the asyncio backend

docs/versionhistory.rst

@@ -84,6 +84,10 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Fixed asyncio task groups leaking unawaited coroutines when a custom task constructor
   fails; default task creation is unaffected
   (`#1274 <https://github.com/agronholm/anyio/issues/1274>`_; PR by @dsfaccini)
+- Fixed ``SocketStream.aclose()`` on the asyncio backend raising ``AttributeError``
+  when the connection is lost while the close is suspended at its checkpoint between
+  ``transport.close()`` and ``transport.abort()``
+  (`#1250 <https://github.com/agronholm/anyio/issues/1250>`_; PR by @afonsojanu)
 
 **4.14.2**
 

src/anyio/_backends/_asyncio.py

@@ -1265,6 +1265,7 @@ class StreamProtocol(asyncio.Protocol):
     write_event: asyncio.Event
     exception: Exception | None = None
     is_at_eof: bool = False
+    is_connection_lost: bool = False
 
     def connection_made(self, transport: asyncio.BaseTransport) -> None:
         self.read_queue = deque()
@@ -1277,6 +1278,7 @@ def connection_lost(self, exc: Exception | None) -> None:
         if exc:
             self.exception = exc
 
+        self.is_connection_lost = True
         self.read_event.set()
         self.write_event.set()
 
@@ -1416,7 +1418,13 @@ async def aclose(self) -> None:
 
             self._transport.close()
             await sleep(0)
-            self._transport.abort()
+
+            # If connection_lost() has already fired by the time we get here (e.g.
+            # a buffered write drained and completed the close during that checkpoint),
+            # the transport has detached itself from the event loop, so calling
+            # abort() on it would raise AttributeError instead of being a no-op.
+            if not self._protocol.is_connection_lost:
+                self._transport.abort()
 
 
 class _RawSocketMixin:

tests/test_sockets.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import array
+import asyncio
 import errno
 import gc
 import io
@@ -65,6 +66,8 @@
     wait_socket_writable,
     wait_writable,
 )
+from anyio._backends._asyncio import SocketStream as AsyncioSocketStream
+from anyio._backends._asyncio import StreamProtocol
 from anyio._core._eventloop import get_async_backend
 from anyio.abc import (
     AnyByteStream,
@@ -725,6 +728,51 @@ def serve() -> None:
             )
             assert not caplog_text
 
+    @pytest.mark.parametrize("anyio_backend", asyncio_params)
+    async def test_aclose_after_connection_already_lost(self) -> None:
+        """
+        Regression test for #1250: if connection_lost() fires while
+        SocketStream.aclose() is suspended at its checkpoint between
+        transport.close() and transport.abort() (which happens when a buffered
+        write finishes draining during that window), the transport has already
+        detached from the event loop, so aclose() must not call abort() on it.
+        Calling abort() at that point raises AttributeError on the real
+        asyncio transport.
+        """
+
+        class DetachingTransport(asyncio.Transport):
+            def __init__(self) -> None:
+                self.closed = False
+                self.aborted = False
+
+            def is_closing(self) -> bool:
+                return self.closed
+
+            def write_eof(self) -> None:
+                pass
+
+            def set_write_buffer_limits(
+                self, high: int | None = None, low: int | None = None
+            ) -> None:
+                pass
+
+            def close(self) -> None:
+                self.closed = True
+
+            def abort(self) -> None:
+                self.aborted = True
+
+        transport = DetachingTransport()
+        protocol = StreamProtocol()
+        protocol.connection_made(transport)
+        stream = AsyncioSocketStream(transport, protocol)
+
+        loop = asyncio.get_running_loop()
+        loop.call_soon(protocol.connection_lost, None)
+        await stream.aclose()
+
+        assert not transport.aborted
+
     async def test_from_socket(
         self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol
     ) -> None: