Fix concurrent force-close on asyncio socket streams

docs/versionhistory.rst

@@ -52,6 +52,9 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   which triggers ``PytestRemovedIn10Warning`` on ``pytest>=9.2`` and crashes pytest at
   startup when ``filterwarnings = error`` is configured
   (`#1271 <https://github.com/agronholm/anyio/issues/1271>`_; PR by @matthewfeickert)
+- Fixed concurrent ``aclose_forcefully()`` calls on asyncio socket streams returning
+  before the underlying socket was closed
+  (`#1273 <https://github.com/agronholm/anyio/issues/1273>`_; PR by @hansu650)
 
 **4.14.2**
 

src/anyio/_backends/_asyncio.py

@@ -1393,15 +1393,19 @@ 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
+        try:
+            if not self._transport.is_closing():
+                try:
+                    self._transport.write_eof()
+                except OSError:
+                    pass
 
-            self._transport.close()
-            await sleep(0)
+                self._transport.close()
+                await sleep(0)
+        finally:
             self._transport.abort()
+            with CancelScope(shield=True):
+                await sleep(0)
 
 
 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_waits_for_socket_close(
+        self, server_addr: tuple[str, int]
+    ) -> None:
+        stream = await connect_tcp(*server_addr)
+        raw_socket = stream.extra(SocketAttribute.raw_socket)
+        file_descriptors: list[int] = []
+
+        async def close_stream() -> None:
+            await aclose_forcefully(stream)
+            file_descriptors.append(raw_socket.fileno())
+
+        async with create_task_group() as task_group:
+            task_group.start_soon(close_stream)
+            task_group.start_soon(close_stream)
+
+        assert file_descriptors == [-1, -1]
+
     async def test_receive_after_close(self, server_addr: tuple[str, int]) -> None:
         stream = await connect_tcp(*server_addr)
         await stream.aclose()