Repository: agronholm/anyio

Issue:
Cancelling a TaskGroup in which a task is starting all also cancel the TaskGroup in which it is due to run
### Things to check first

- [X] I have searched the existing issues and didn't find my bug already reported there

- [X] I have checked that my bug is still present in the latest release


### AnyIO version

4.2

### Python version

3.10.13

### What happened?

This is related to, but slightly different from, issue #517 

Consider this sequence of steps with anyio (asyncio backend) in this way:

1. I start an outer TaskGroup
2. Within that, I start an inner TaskGroup
3. I start a task in the inner TaskGroup, due to run in the outer TaskGroup (e.g., calling `await outer_tg.start(foo)` in the context of the inner TaskGroup)
4. Before the routine calls `task_status.started()` and gets reparented to the outer TaskGroup, cancel the inner task group (and therefore this partially started task).

If you do all that, then the outer TaskGroup will be cancelled as well, including any other tasks that were running in it. Somehow the partially started task has leaked the cancellation from one TaskGroup to the other.

### How can we reproduce the bug?

One example of how this would look:

```python
async def issue_demo1():
    async with anyio.create_task_group() as run_tg:
        run_tg.start_soon(sleep_only)
        async with anyio.create_task_group() as start_tg:
            start_tg.start_soon(run_tg.start, sleep_twice)
            start_tg.cancel_scope.cancel()
```

If using `TaskGroup.start_soon()` on `TaskGroup.start()` looks a bit suspicious, another way would be to do cancellation in the background, for example with a time out:

```python
async def issue_demo2():
    async with anyio.create_task_group() as run_tg:
        run_tg.start_soon(sleep_only)
        async with anyio.create_task_group() as start_tg:
            start_tg.cancel_scope.deadline = anyio.current_time() + 0.5
            await run_tg.start(sleep_twice)
```

Here are definitions of the coroutines used in the examples:

```python
async def sleep_twice(task_status=anyio.TASK_STATUS_IGNORED):
    try:
        print("sleep_twice(): at beginning")
        await anyio.sleep(1)
        task_status.started()
        print("sleep_twice(): started() called")
        await anyio.sleep(1)
        print("sleep_twice(): completed")
    except BaseException as be:
        print(f"sleep_twice(): got exception {be!r}")
        raise

async def sleep_only():
    try:
        print("sleep_only(): at beginning")
        await anyio.sleep(100)
        print("sleep_only(): completed")
    except BaseException as be:
        print(f"sleep_only(): got exception {be!r}")
        raise
```

Here is an example output:

```
sleep_only(): at beginning
sleep_twice(): at beginning
sleep_twice(): got exception CancelledError()
sleep_only(): got exception CancelledError('Cancelled by cancel scope 27ba061c850')
```


Discussion:
arthur-tacca: I tried this out. I did `await tg_destination.start(foo)` in `tg_first`  (just giving the task groups names here).

If I cancel `tg_first` before `foo` gets to `task_status.started()`, then previously it would cancel both `tg_first` (obviously), and the routine `foo()` immediately, and the `tg_destination`. That's the original bug. After this change, it cancels `tg_first` and `foo()`, but not `tg_destination`. Bug fixed, hurrah! This also matches trio.

But if I cancel `tg_destination` before `foo` gets to `task_status.started()` then that cancels the routine even though it's not supposed to currently be running in that task group. It also cancels `tg_first` and weirdly even the outer task group containing both `tg_first` and `tg_destination`. This happens with both the original and modified code. In fairness, this wasn't mentioned in the original report (sorry). Trio, on the other hand, only cancels `tg_destination`, and `foo()` continues to run until the first await after `task_status.started()`.
arthur-tacca: BTW in Trio, in that second case, by the time the routine `foo()` calls `task_status.started()`, the second task group has not only has cancel called but already completed (exited the `async with` block) - it has to be that way otherwise an arbitrary starting task that is not cancelled could hold up its cancellation for an arbitrary length of time. So perhaps there is extra magic to inject cancellation into `foo()` and catch it even though it is in a cancel scope that is already exited. But what if it converts that to a different exception? Is that lost? Surely Trio never loses exceptions. I think I have missed something but I don't have time to figure it out, sorry (and sorry for the ramble).
agronholm: Can you at least provide a reproducing snippet of what isn't working?
arthur-tacca: Sorry yes, the fold below contains a repro (use of aioresult isn't necessary but I found it easier to write this way).

I think the thing I'm missing about Trio is that it actually can have bugs / incomplete design decisions! This other issue is arguably a second bug (I actually found it by looking at your changing and thinking ... how does the destination nursery find out when the new task is reparented to it? The answer is: it already was treated as a child.)

Another way to interpret my comment: this change fixes the original bug and doesn't (to my knowledge) introduce any new bugs.

<details>
<summary>Code to reproduce odd behaviours</summary>

```python

# Test for https://github.com/agronholm/anyio/pull/717


import anyio
import aioresult


futures = [None, None]


async def startable(task_status = anyio.TASK_STATUS_IGNORED):
    print("startable(): starting")
    try:
        await anyio.sleep(1)
    except BaseException as e:
        print(f"startable(): during first sleep {e!r}")
        raise
    print("startable(): after first sleep")

    try:
        task_status.started("x")
    except BaseException as e:
        print(f"startable(): during started {e!r}")

    print("startable(): about to sleep again")
    try:
        await anyio.sleep(1)
    except BaseException as e:
        print(f"startable(): during second sleep {e!r}")
        raise
    print("startable(): after second sleep")



async def run_first():
    async with anyio.create_task_group() as tg:
        futures[0].set_result(tg)

        await futures[1].wait_done()
        other_tg = futures[1].result()
        await other_tg.start(startable)

        await anyio.sleep(1)

async def run_second():
    async with anyio.create_task_group() as tg:
        futures[1].set_result(tg)

        await anyio.sleep(2)


async def run_test(i, which_tg_cancel, when_cancel):
    futures[0] = aioresult.Future()
    futures[1] = aioresult.Future()

    async with anyio.create_task_group() as tg:
        tg.start_soon(run_first)
        tg.start_soon(run_second)

        await aioresult.wait_all(futures)
        await anyio.sleep(when_cancel)
        futures[which_tg_cancel].result().cancel_scope.cancel()

        print("before sleep")
        await anyio.sleep(0.2)
        print("after sleep")
        for i in range(2):
            c : anyio.CancelScope = futures[i].result().cancel_scope
            print(f"tg {i}: cancelled: {c.cancel_called}, caught: {c.cancelled_caught}")
    c: anyio.CancelScope = tg.cancel_scope
    print(f"tg outer: cancelled: {c.cancel_called}, caught: {c.cancelled_caught}")
    print("test complete")


async def run_all():
    print("\n\n\ntest 1: cancel 1st TG early")
    await run_test(1, 0, 0.5)

    print("\n\n\ntest 2: cancel 2nd TG early")
    await run_test(2, 1, 0.5)

    print("\n\n\ntest 3: cancel 1st TG late")
    await run_test(3, 0, 1.5)

    print("\n\n\ntest 4: cancel 2nd TG late")
    await run_test(4, 1, 1.5)

anyio.run(run_all, backend="asyncio")
```
</details>
arthur-tacca: Also, a nitpick: I found the comment confusing

> The future can only be cancelled if the host task was cancelled

I assumed this referred to the reason *why* we're performing this test at all. So we're doing this because the host task was cancelled? Hmm.

In fact, it's assuming you already know why we're performing this test and is actually explaining *how* it's testing for it. Perhaps it could be preceded by a comment that says why we're performing the test (e.g., something like "check whether this task was cancelled while still waiting for it to start in `TaskGroup.start()` before reaching `task_status.started()`"). But I'm not familiar with this code base so perhaps this isn't valid for those that are.
agronholm: > Also, a nitpick: I found the comment confusing
> 
> > The future can only be cancelled if the host task was cancelled
> 
> I assumed this referred to the reason _why_ we're performing this test at all. So we're doing this because the host task was cancelled? Hmm.
> 
> In fact, it's assuming you already know why we're performing this test and is actually explaining _how_ it's testing for it. Perhaps it could be preceded by a comment that says why we're performing the test (e.g., something like "check whether this task was cancelled while still waiting for it to start in `TaskGroup.start()` before reaching `task_status.started()`"). But I'm not familiar with this code base so perhaps this isn't valid for those that are.

I've clarified the comment now, hopefully it's okay now.
agronholm: > Sorry yes, the fold below contains a repro (use of aioresult isn't necessary but I found it easier to write this way).
> 
> I think the thing I'm missing about Trio is that it actually can have bugs / incomplete design decisions! This other issue is arguably a second bug (I actually found it by looking at your changing and thinking ... how does the destination nursery find out when the new task is reparented to it? The answer is: it already was treated as a child.)
> 
> Another way to interpret my comment: this change fixes the original bug and doesn't (to my knowledge) introduce any new bugs.
> 
> Code to reproduce odd behaviours
> ```python
> # Test for https://github.com/agronholm/anyio/pull/717
> 
> 
> import anyio
> import aioresult
> 
> 
> futures = [None, None]
> 
> 
> async def startable(task_status = anyio.TASK_STATUS_IGNORED):
>     print("startable(): starting")
>     try:
>         await anyio.sleep(1)
>     except BaseException as e:
>         print(f"startable(): during first sleep {e!r}")
>         raise
>     print("startable(): after first sleep")
> 
>     try:
>         task_status.started("x")
>     except BaseException as e:
>         print(f"startable(): during started {e!r}")
> 
>     print("startable(): about to sleep again")
>     try:
>         await anyio.sleep(1)
>     except BaseException as e:
>         print(f"startable(): during second sleep {e!r}")
>         raise
>     print("startable(): after second sleep")
> 
> 
> 
> async def run_first():
>     async with anyio.create_task_group() as tg:
>         futures[0].set_result(tg)
> 
>         await futures[1].wait_done()
>         other_tg = futures[1].result()
>         await other_tg.start(startable)
> 
>         await anyio.sleep(1)
> 
> async def run_second():
>     async with anyio.create_task_group() as tg:
>         futures[1].set_result(tg)
> 
>         await anyio.sleep(2)
> 
> 
> async def run_test(i, which_tg_cancel, when_cancel):
>     futures[0] = aioresult.Future()
>     futures[1] = aioresult.Future()
> 
>     async with anyio.create_task_group() as tg:
>         tg.start_soon(run_first)
>         tg.start_soon(run_second)
> 
>         await aioresult.wait_all(futures)
>         await anyio.sleep(when_cancel)
>         futures[which_tg_cancel].result().cancel_scope.cancel()
> 
>         print("before sleep")
>         await anyio.sleep(0.2)
>         print("after sleep")
>         for i in range(2):
>             c : anyio.CancelScope = futures[i].result().cancel_scope
>             print(f"tg {i}: cancelled: {c.cancel_called}, caught: {c.cancelled_caught}")
>     c: anyio.CancelScope = tg.cancel_scope
>     print(f"tg outer: cancelled: {c.cancel_called}, caught: {c.cancelled_caught}")
>     print("test complete")
> 
> 
> async def run_all():
>     print("\n\n\ntest 1: cancel 1st TG early")
>     await run_test(1, 0, 0.5)
> 
>     print("\n\n\ntest 2: cancel 2nd TG early")
>     await run_test(2, 1, 0.5)
> 
>     print("\n\n\ntest 3: cancel 1st TG late")
>     await run_test(3, 0, 1.5)
> 
>     print("\n\n\ntest 4: cancel 2nd TG late")
>     await run_test(4, 1, 1.5)
> 
> anyio.run(run_all, backend="asyncio")
> ```

It's not clear to me what the _expected_ output of this script is.
agronholm: @Zac-HD would you mind rubber stamping this PR? @arthur-tacca said above that it fixes the original bug and doesn't introduce any new ones.

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 0102869..2ecec4b 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -10,14 +10,19 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   portals
 - Added ``__slots__`` to ``AsyncResource`` so that child classes can use ``__slots__``
   (`#733 <https://github.com/agronholm/anyio/pull/733>`_; PR by Justin Su)
+- 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>`_,
+    `#710 <https://github.com/agronholm/anyio/issues/710>`_)
 - Fixed a race condition that caused crashes when multiple event loops of the same
   backend were running in separate threads and simultaneously attempted to use AnyIO for
   their first time (`#425 <https://github.com/agronholm/anyio/issues/425>`_; PR by David
   Jiricek and Ganden Schaffner)
-- 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 cancellation delivery on asyncio incrementing the wrong cancel scope's
   cancellation counter when cascading a cancel operation to a child scope, thus failing
   to uncancel the host task (`#716 <https://github.com/agronholm/anyio/issues/716>`_)
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index ecd865a..b69665c 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -714,6 +714,12 @@ class TaskGroup(abc.TaskGroup):
                 exc = e
 
             if exc is not None:
+                # The future can only be in the cancelled state if the host task was
+                # cancelled, so return immediately instead of adding one more
+                # CancelledError to the exceptions list
+                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)


Test patch:
diff --git a/tests/test_taskgroups.py b/tests/test_taskgroups.py
index da84f65..11329df 100644
--- a/tests/test_taskgroups.py
+++ b/tests/test_taskgroups.py
@@ -22,6 +22,7 @@ from anyio import (
     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 test_start_cancelled() -> 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
@@ -1360,6 +1357,24 @@ async def test_cancel_child_task_when_host_is_shielded() -> 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