Fix aclose_forcefully() returning before the socket is released

docs/versionhistory.rst

@@ -23,6 +23,14 @@ 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 ``aclose_forcefully()`` (and, more generally, concurrent calls to
+  ``SocketStream.aclose()``) returning before the underlying TCP socket was actually
+  released on asyncio: a checkpoint hit while the cancel scope used by
+  ``aclose_forcefully()`` was already cancelled could skip the transport's
+  ``abort()`` call, and a second, concurrent ``aclose()`` call could see the
+  transport already marked as closing and return immediately without waiting for
+  the socket to actually be released
+  (`#1273 <https://github.com/agronholm/anyio/issues/1273>`_)
 - 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

@@ -1323,6 +1323,7 @@ def __init__(self, transport: asyncio.Transport, protocol: StreamProtocol):
         self._receive_guard = ResourceGuard("reading from")
         self._send_guard = ResourceGuard("writing to")
         self._closed = False
+        self._aclose_event: asyncio.Event | None = None
 
     @property
     def _raw_socket(self) -> socket.socket:
@@ -1393,15 +1394,31 @@ async def send_eof(self) -> None:
 
     async def aclose(self) -> None:
         self._closed = True
-        if not self._transport.is_closing():
-            try:
-                self._transport.write_eof()
-            except OSError:
-                pass
+        if self._aclose_event is not None:
+            # Another task is already closing (or has already closed) the
+            # transport. Wait for it to actually finish, shielded from our own
+            # cancellation, so that we never return before the socket is
+            # really released (see #1273).
+            with CancelScope(shield=True):
+                await self._aclose_event.wait()
 
-            self._transport.close()
-            await sleep(0)
-            self._transport.abort()
+            return
+
+        self._aclose_event = event = asyncio.Event()
+        try:
+            if not self._transport.is_closing():
+                try:
+                    self._transport.write_eof()
+                except OSError:
+                    pass
+
+                try:
+                    self._transport.close()
+                    await sleep(0)
+                finally:
+                    self._transport.abort()
+        finally:
+            event.set()
 
 
 class _RawSocketMixin:

tests/test_sockets.py

@@ -43,6 +43,7 @@
     TCPConnectable,
     TypedAttributeLookupError,
     UNIXConnectable,
+    aclose_forcefully,
     as_connectable,
     connect_tcp,
     connect_unix,
@@ -507,6 +508,23 @@ async def interrupt() -> None:
                 with pytest.raises(ClosedResourceError):
                     await stream.receive()
 
+    async def test_concurrent_aclose_forcefully_returns_before_socket_closes(
+        self, server_addr: tuple[str, int]
+    ) -> None:
+        stream = await connect_tcp(*server_addr)
+        raw_socket = stream.extra(SocketAttribute.raw_socket)
+        fds: list[int] = []
+
+        async def do_aclose() -> None:
+            await aclose_forcefully(stream)
+            fds.append(raw_socket.fileno())
+
+        async with create_task_group() as tg:
+            tg.start_soon(do_aclose)
+            tg.start_soon(do_aclose)
+
+        assert fds == [-1, -1]
+
     async def test_receive_after_close(self, server_addr: tuple[str, int]) -> None:
         stream = await connect_tcp(*server_addr)
         await stream.aclose()