Repository: agronholm/anyio

Issue:
Opening a CancelScope with shield=True also shields sibling tasks on asyncio backend
### 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.1.0

### Python version

3.11

### What happened?

I want to run an asyncio task from inside of an anyio task group. Anyio enforces level triggered cancelation. But, in this case, I want to run some asyncio code that may not be prepared to deal with level triggered cancellation. So, my plan is to run the asyncio code that wants edge triggered cancellation inside of a shielded CancelScope. And then I want to run a sibling task that just wants for a cancellation signal, and then sends a single cancellation into the asyncio code when that occurs.

The problem that i'm running into, is that when I open a shielded CancelScope on the main task, the sibling task also seems to be shielded when running on the asyncio backend.

### How can we reproduce the bug?

When I run this code, it hangs forever:

```python
import anyio


async def func() -> None:
    cancelled = anyio.Event()

    async with anyio.create_task_group() as task_group:
        async def wait_cancel() -> None:
            try:
                await anyio.sleep_forever()
            except anyio.get_cancelled_exc_class():
                cancelled.set()
                raise

        task_group.start_soon(wait_cancel)

        with anyio.CancelScope(shield=True):
            await cancelled.wait()


async def main() -> None:
    async with anyio.create_task_group() as task_group:
        task_group.start_soon(func)
        await anyio.sleep(0.1)
        task_group.cancel_scope.cancel()


if __name__ == "__main__":
    anyio.run(main, backend="asyncio")
```

However, if you switch it to use the trio backend, it then completes as expected.

Additionally, if you change the function to detect cancellation in the main task and put the shielded CancelScope in the sibling task using the asyncio backend, it also completes as expected:


```python
async def func() -> None:
    cancelled = anyio.Event()

    async with anyio.create_task_group() as task_group:
        async def wait_cancel() -> None:
            with anyio.CancelScope(shield=True):
                await cancelled.wait()

        task_group.start_soon(wait_cancel)

        try:
            await anyio.sleep_forever()
        except anyio.get_cancelled_exc_class():
            cancelled.set()
            raise
```

Discussion:
agronholm: All ready for merging then?
agronholm: > might be good to add [0214c34](https://github.com/agronholm/anyio/commit/0214c3498fed76c711d828351c70cd7e6b7127f6), as that bug almost slipped by.

Done.
agronholm: No worries! Would you mind approving the PR if it looks good?
agronholm: Hm, that didn't do the trick. It doesn't look like you're on the collaborator list. Would you like to be?

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 78ef510..94eaf15 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -13,6 +13,9 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   from Egor Blagov)
 - Fixed ``loop_factory`` and ``use_uvloop`` options not being used on the asyncio
   backend (`#643 <https://github.com/agronholm/anyio/issues/643>`_)
+- Fixed cancellation propagating on asyncio from a task group to child tasks if the task
+  hosting the task group is in a shielded cancel scope
+  (`#642 <https://github.com/agronholm/anyio/issues/642>`_)
 
 **4.1.0**
 
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index bdc0aa8..95b8e55 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -343,6 +343,7 @@ class CancelScope(BaseCancelScope):
         self._deadline = deadline
         self._shield = shield
         self._parent_scope: CancelScope | None = None
+        self._child_scopes: set[CancelScope] = set()
         self._cancel_called = False
         self._cancelled_caught = False
         self._active = False
@@ -369,6 +370,9 @@ class CancelScope(BaseCancelScope):
         else:
             self._parent_scope = task_state.cancel_scope
             task_state.cancel_scope = self
+            if self._parent_scope is not None:
+                self._parent_scope._child_scopes.add(self)
+                self._parent_scope._tasks.remove(host_task)
 
         self._timeout()
         self._active = True
@@ -377,7 +381,7 @@ class CancelScope(BaseCancelScope):
 
         # Start cancelling the host task if the scope was cancelled before entering
         if self._cancel_called:
-            self._deliver_cancellation()
+            self._deliver_cancellation(self)
 
         return self
 
@@ -409,13 +413,15 @@ class CancelScope(BaseCancelScope):
             self._timeout_handle = None
 
         self._tasks.remove(self._host_task)
+        if self._parent_scope is not None:
+            self._parent_scope._child_scopes.remove(self)
+            self._parent_scope._tasks.add(self._host_task)
 
         host_task_state.cancel_scope = self._parent_scope
 
-        # Restart the cancellation effort in the farthest directly cancelled parent
+        # Restart the cancellation effort in the closest directly cancelled parent
         # scope if this one was shielded
-        if self._shield:
-            self._deliver_cancellation_to_parent()
+        self._restart_cancellation_in_parent()
 
         if self._cancel_called and exc_val is not None:
             for exc in iterate_exceptions(exc_val):
@@ -451,12 +457,16 @@ class CancelScope(BaseCancelScope):
             else:
                 self._timeout_handle = loop.call_at(self._deadline, self._timeout)
 
-    def _deliver_cancellation(self) -> None:
+    def _deliver_cancellation(self, origin: CancelScope) -> bool:
         """
         Deliver cancellation to directly contained tasks and nested cancel scopes.
 
         Schedule another run at the end if we still have tasks eligible for
         cancellation.
+
+        :param origin: the cancel scope that originated the cancellation
+        :return: ``True`` if the delivery needs to be retried on the next cycle
+
         """
         should_retry = False
         current = current_task()
@@ -464,42 +474,46 @@ class CancelScope(BaseCancelScope):
             if task._must_cancel:  # type: ignore[attr-defined]
                 continue
 
-            # The task is eligible for cancellation if it has started and is not in a
-            # cancel scope shielded from this one
-            cancel_scope = _task_states[task].cancel_scope
-            while cancel_scope is not self:
-                if cancel_scope is None or cancel_scope._shield:
-                    break
-                else:
-                    cancel_scope = cancel_scope._parent_scope
-            else:
-                should_retry = True
-                if task is not current and (
-                    task is self._host_task or _task_started(task)
-                ):
-                    waiter = task._fut_waiter  # type: ignore[attr-defined]
-                    if not isinstance(waiter, asyncio.Future) or not waiter.done():
-                        self._cancel_calls += 1
-                        if sys.version_info >= (3, 9):
-                            task.cancel(f"Cancelled by cancel scope {id(self):x}")
-                        else:
-                            task.cancel()
+            # The task is eligible for cancellation if it has started
+            should_retry = True
+            if task is not current and (task is self._host_task or _task_started(task)):
+                waiter = task._fut_waiter  # type: ignore[attr-defined]
+                if not isinstance(waiter, asyncio.Future) or not waiter.done():
+                    self._cancel_calls += 1
+                    if sys.version_info >= (3, 9):
+                        task.cancel(f"Cancelled by cancel scope {id(origin):x}")
+                    else:
+                        task.cancel()
+
+        # Deliver cancellation to child scopes that aren't shielded or running their own
+        # cancellation callbacks
+        for scope in self._child_scopes:
+            if not scope._shield and not scope.cancel_called:
+                should_retry = scope._deliver_cancellation(origin) or should_retry
 
         # Schedule another callback if there are still tasks left
-        if should_retry:
-            self._cancel_handle = get_running_loop().call_soon(
-                self._deliver_cancellation
-            )
-        else:
-            self._cancel_handle = None
+        if origin is self:
+            if should_retry:
+                self._cancel_handle = get_running_loop().call_soon(
+                    self._deliver_cancellation, origin
+                )
+            else:
+                self._cancel_handle = None
+
+        return should_retry
+
+    def _restart_cancellation_in_parent(self) -> None:
+        """
+        Restart the cancellation effort in the closest directly cancelled parent scope.
 
-    def _deliver_cancellation_to_parent(self) -> None:
-        """Start cancellation effort in the farthest directly cancelled parent scope"""
+        """
         scope = self._parent_scope
-        scope_to_cancel: CancelScope | None = None
         while scope is not None:
-            if scope._cancel_called and scope._cancel_handle is None:
-                scope_to_cancel = scope
+            if scope._cancel_called:
+                if scope._cancel_handle is None:
+                    scope._deliver_cancellation(scope)
+
+                break
 
             # No point in looking beyond any shielded scope
             if scope._shield:
@@ -507,9 +521,6 @@ class CancelScope(BaseCancelScope):
 
             scope = scope._parent_scope
 
-        if scope_to_cancel is not None:
-            scope_to_cancel._deliver_cancellation()
-
     def _parent_cancelled(self) -> bool:
         # Check whether any parent has been cancelled
         cancel_scope = self._parent_scope
@@ -529,7 +540,7 @@ class CancelScope(BaseCancelScope):
 
             self._cancel_called = True
             if self._host_task is not None:
-                self._deliver_cancellation()
+                self._deliver_cancellation(self)
 
     @property
     def deadline(self) -> float:
@@ -562,7 +573,7 @@ class CancelScope(BaseCancelScope):
         if self._shield != value:
             self._shield = value
             if not value:
-                self._deliver_cancellation_to_parent()
+                self._restart_cancellation_in_parent()
 
 
 #
@@ -623,6 +634,7 @@ class TaskGroup(abc.TaskGroup):
         self.cancel_scope: CancelScope = CancelScope()
         self._active = False
         self._exceptions: list[BaseException] = []
+        self._tasks: set[asyncio.Task] = set()
 
     async def __aenter__(self) -> TaskGroup:
         self.cancel_scope.__enter__()
@@ -642,9 +654,9 @@ class TaskGroup(abc.TaskGroup):
                 self._exceptions.append(exc_val)
 
         cancelled_exc_while_waiting_tasks: CancelledError | None = None
-        while self.cancel_scope._tasks:
+        while self._tasks:
             try:
-                await asyncio.wait(self.cancel_scope._tasks)
+                await asyncio.wait(self._tasks)
             except CancelledError as exc:
                 # This task was cancelled natively; reraise the CancelledError later
                 # unless this task was already interrupted by another exception
@@ -676,8 +688,11 @@ class TaskGroup(abc.TaskGroup):
         task_status_future: asyncio.Future | None = None,
     ) -> asyncio.Task:
         def task_done(_task: asyncio.Task) -> None:
-            assert _task in self.cancel_scope._tasks
-            self.cancel_scope._tasks.remove(_task)
+            task_state = _task_states[_task]
+            assert task_state.cancel_scope is not None
+            assert _task in task_state.cancel_scope._tasks
+            task_state.cancel_scope._tasks.remove(_task)
+            self._tasks.remove(task)
             del _task_states[_task]
 
             try:
@@ -693,7 +708,8 @@ class TaskGroup(abc.TaskGroup):
                     if not isinstance(exc, CancelledError):
                         self._exceptions.append(exc)
 
-                    self.cancel_scope.cancel()
+                    if not self.cancel_scope._parent_cancelled():
+                        self.cancel_scope.cancel()
                 else:
                     task_status_future.set_exception(exc)
             elif task_status_future is not None and not task_status_future.done():
@@ -732,6 +748,7 @@ class TaskGroup(abc.TaskGroup):
             parent_id=parent_id, cancel_scope=self.cancel_scope
         )
         self.cancel_scope._tasks.add(task)
+        self._tasks.add(task)
         return task
 
     def start_soon(


Test patch:
diff --git a/tests/test_taskgroups.py b/tests/test_taskgroups.py
index 89e9764..f4c87b3 100644
--- a/tests/test_taskgroups.py
+++ b/tests/test_taskgroups.py
@@ -844,7 +844,7 @@ async def test_nested_shield() -> None:
     assert isinstance(exc.value.exceptions[0], TimeoutError)
 
 
-async def test_triple_nested_shield() -> None:
+async def test_triple_nested_shield_checkpoint_in_outer() -> None:
     """Regression test for #370."""
 
     got_past_checkpoint = False
@@ -867,6 +867,26 @@ async def test_triple_nested_shield() -> None:
     assert not got_past_checkpoint
 
 
+async def test_triple_nested_shield_checkpoint_in_middle() -> None:
+    got_past_checkpoint = False
+
+    async def taskfunc() -> None:
+        nonlocal got_past_checkpoint
+
+        with CancelScope() as scope1:
+            with CancelScope():
+                with CancelScope(shield=True):
+                    scope1.cancel()
+
+                await checkpoint()
+                got_past_checkpoint = True
+
+    async with create_task_group() as tg:
+        tg.start_soon(taskfunc)
+
+    assert not got_past_checkpoint
+
+
 def test_task_group_in_generator(
     anyio_backend_name: str, anyio_backend_options: dict[str, Any]
 ) -> None:
@@ -1293,6 +1313,29 @@ async def test_reraise_cancelled_in_excgroup() -> None:
             await anyio.sleep_forever()
 
 
+async def test_cancel_child_task_when_host_is_shielded() -> None:
+    # Regression test for #642
+    # Tests that cancellation propagates to a child task even if the host task is within
+    # a shielded cancel scope.
+    cancelled = anyio.Event()
+
+    async def wait_cancel() -> None:
+        try:
+            await anyio.sleep_forever()
+        except anyio.get_cancelled_exc_class():
+            cancelled.set()
+            raise
+
+    with CancelScope() as parent_scope:
+        async with anyio.create_task_group() as task_group:
+            task_group.start_soon(wait_cancel)
+            await wait_all_tasks_blocked()
+
+            with CancelScope(shield=True), fail_after(1):
+                parent_scope.cancel()
+                await cancelled.wait()
+
+
 class TestTaskStatusTyping:
     """
     These tests do not do anything at run time, but since the test suite is also checked