Fixed SocketStream close race on asyncio

docs/versionhistory.rst

@@ -17,6 +17,9 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   module name. (The default name for a task spawned with ``TaskGroup.start_soon`` or
   ``TaskGroup.start`` typically includes the module name.)
   (`#1234 <https://github.com/agronholm/anyio/pull/1234>`_; PR by @gschaffner)
+- Fixed ``SocketStream.aclose()`` raising an ``AttributeError`` on the asyncio backend
+  when the write buffer drained between closing and aborting the transport
+  (`#1250 <https://github.com/agronholm/anyio/issues/1250>`_; PR by @subotac)
 - Fixed free-threading compatibility issues arising from the fact that on Python 3.14
   free-threading builds, newly created threads inherit the current context by default,
   causing AnyIO to behave erroneously in relation to ``start_blocking_portal()`` and

src/anyio/_backends/_asyncio.py

@@ -1250,6 +1250,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()
@@ -1259,6 +1260,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None:
         cast(asyncio.Transport, transport).set_write_buffer_limits(0)
 
     def connection_lost(self, exc: Exception | None) -> None:
+        self.is_connection_lost = True
         if exc:
             self.exception = exc
 
@@ -1401,7 +1403,8 @@ async def aclose(self) -> None:
 
             self._transport.close()
             await sleep(0)
-            self._transport.abort()
+            if not self._protocol.is_connection_lost:
+                self._transport.abort()
 
 
 class _RawSocketMixin:

tests/test_sockets.py

@@ -506,6 +506,28 @@ async def interrupt() -> None:
                 with pytest.raises(ClosedResourceError):
                     await stream.receive()
 
+    @pytest.mark.parametrize("anyio_backend", asyncio_params[:1])
+    async def test_close_after_connection_lost(
+        self, server_addr: tuple[str, int], mocker: MockerFixture
+    ) -> None:
+        stream = cast(Any, await connect_tcp(*server_addr))
+        original_close = stream._transport.close
+
+        def close() -> None:
+            original_close()
+            stream._protocol.connection_lost(None)
+
+        mocker.patch.object(stream._transport, "close", side_effect=close)
+        abort = mocker.patch.object(
+            stream._transport,
+            "abort",
+            side_effect=AttributeError(
+                "'NoneType' object has no attribute 'call_soon'"
+            ),
+        )
+        await stream.aclose()
+        abort.assert_not_called()
+
     async def test_receive_after_close(self, server_addr: tuple[str, int]) -> None:
         stream = await connect_tcp(*server_addr)
         await stream.aclose()