Fixed task group getting cancelled if start() gets cancelled

This change fixes the problem by special casing the situation where the Future backing `task_status` was cancelled which only happens when the host task is cancelled.

Fixes #685. Fixes #701.

docs/versionhistory.rst

@@ -8,10 +8,15 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Added the ``BlockingPortalProvider`` class to aid with constructing synchronous
   counterparts to asynchronous interfaces that would otherwise require multiple blocking
   portals
-- Fixed erroneous ``RuntimeError: called 'started' twice on the same task status``
-  when cancelling a task in a TaskGroup created with the ``start()`` method before
-  the first checkpoint is reached after calling ``task_status.started()``
-  (`#706 <https://github.com/agronholm/anyio/issues/706>`_; PR by Dominik Schwabe)
+- Fixed two bugs with ``TaskGroup.start()`` on asyncio:
+
+  * Fixed erroneous ``RuntimeError: called 'started' twice on the same task status``
+    when cancelling a task in a TaskGroup created with the ``start()`` method before
+    the first checkpoint is reached after calling ``task_status.started()``
+    (`#706 <https://github.com/agronholm/anyio/issues/706>`_; PR by Dominik Schwabe)
+  * Fixed the entire task group being cancelled if a ``TaskGroup.start()`` call gets
+    cancelled (`#685 <https://github.com/agronholm/anyio/issues/685>`_,
+    `#701 <https://github.com/agronholm/anyio/issues/710>`_)
 - Fixed erroneous ``TypedAttributeLookupError`` if a typed attribute getter raises
   ``KeyError``
 - Fixed the asyncio backend not respecting the ``PYTHONASYNCIODEBUG`` environment

src/anyio/_backends/_asyncio.py

@@ -714,6 +714,10 @@ def task_done(_task: asyncio.Task) -> None:
                 exc = e
 
             if exc is not None:
+                # The future can only be cancelled if the host task was cancelled
+                if task_status_future is not None and task_status_future.cancelled():
+                    return
+
                 if task_status_future is None or task_status_future.done():
                     if not isinstance(exc, CancelledError):
                         self._exceptions.append(exc)

tests/test_taskgroups.py

@@ -22,6 +22,7 @@
     get_current_task,
     move_on_after,
     sleep,
+    sleep_forever,
     wait_all_tasks_blocked,
 )
 from anyio.abc import TaskGroup, TaskStatus
@@ -127,7 +128,6 @@ async def test_no_called_started_twice() -> None:
     async def taskfunc(*, task_status: TaskStatus) -> None:
         task_status.started()
 
-    # anyio>4.3.0 should not raise "RuntimeError: called 'started' twice on the same task status"
     async with create_task_group() as tg:
         coro = tg.start(taskfunc)
         tg.cancel_scope.cancel()
@@ -196,9 +196,6 @@ async def taskfunc(*, task_status: TaskStatus) -> None:
     assert not finished
 
 
-@pytest.mark.xfail(
-    sys.version_info < (3, 9), reason="Requires a way to detect cancellation source"
-)
 @pytest.mark.parametrize("anyio_backend", ["asyncio"])
 async def test_start_native_host_cancelled() -> None:
     started = finished = False
@@ -1347,6 +1344,24 @@ async def wait_cancel() -> None:
                 await cancelled.wait()
 
 
+async def test_start_cancels_parent_scope() -> None:
+    """Regression test for #685 / #710."""
+    started: bool = False
+
+    async def in_task_group(task_status: TaskStatus[None]) -> None:
+        nonlocal started
+        started = True
+        await sleep_forever()
+
+    async with create_task_group() as tg:
+        with CancelScope() as inner_scope:
+            inner_scope.cancel()
+            await tg.start(in_task_group)
+
+    assert started
+    assert not tg.cancel_scope.cancel_called
+
+
 class TestTaskStatusTyping:
     """
     These tests do not do anything at run time, but since the test suite is also checked