diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index b5b8d01b9..61152d819 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -17,6 +17,9 @@ This library adheres to `Semantic Versioning 2.0 `_.
module name. (The default name for a task spawned with ``TaskGroup.start_soon`` or
``TaskGroup.start`` typically includes the module name.)
(`#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 `_; 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
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index d0f4b0126..83878d530 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/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:
diff --git a/tests/test_sockets.py b/tests/test_sockets.py
index 97e1d5768..dfd5a412b 100644
--- a/tests/test_sockets.py
+++ b/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()