_deliver_cancellation re-arms for tasks skipped via _must_cancel, for which the retry can never do anything

[ZacharyHampton]

Things to check first

• I have searched the existing issues and didn't find my bug already reported there
• I have checked that my bug is still present in the latest release

AnyIO version

4.14.2 (also current master)

Python version

3.12.11

What happened?

Split out of #1257 as the standalone, conforming-parts core.

In _deliver_cancellation, a task with task._must_cancel set arms should_retry before being skipped:

should_retry = True
if task._must_cancel:
    continue

The retry can never do anything for that task. _must_cancel means CPython will raise CancelledError into it at its next step with no help from anyio, and anyio's only action (task.cancel()) is skipped anyway. So while such a task sits in a cancelled scope, the scope reschedules itself via call_soon every loop iteration until the task steps.

The state is reachable with stock primitives. Future.cancel() returns False when the future is already done, and a task's waiter can be done before its __wakeup has run. A Task.cancel() in that window takes CPython's _must_cancel branch.

On a fair loop the pending __wakeup runs promptly, so the cost here is a few wasted callbacks per occurrence. Anything that delays or prevents that step turns the same code path into a busy-spin (see #1257).

Suggested direction: when skipping a task on _must_cancel, don't drive the retry by polling; re-arm from task._fut_waiter.add_done_callback(...) when a waiter exists (in this repro the waiter is already done, so behavior is identical), and keep the retry armed in some form — a task may legally swallow the pending CancelledError and re-park, and redelivery must still work.

How can we reproduce the bug?

import asyncio
import anyio
from anyio._backends._asyncio import CancelScope

passes = []
_orig = CancelScope._deliver_cancellation

def counting(self, origin):  # observation only
    cur = asyncio.current_task()
    would_deliver = mc_skips = 0
    for t in self._tasks:
        if t.done():
            continue
        if t._must_cancel:
            mc_skips += 1
            continue
        if t is not cur:
            w = t._fut_waiter
            if not isinstance(w, asyncio.Future) or not w.done():
                would_deliver += 1
    ret = _orig(self, origin)
    passes.append((would_deliver, mc_skips, ret))
    return ret

CancelScope._deliver_cancellation = counting

async def main():
    fut = asyncio.get_running_loop().create_future()
    parked_task = None

    async def parked():
        nonlocal parked_task
        parked_task = asyncio.current_task()
        await fut

    async with anyio.create_task_group() as tg:
        tg.start_soon(parked)
        await anyio.sleep(0.01)

        fut.set_result("ok")                  # waiter done, __wakeup still pending
        assert parked_task.cancel() is True   # Future.cancel() -> False (done): _must_cancel set
        assert parked_task._must_cancel is True

        tg.cancel_scope.cancel()

    print("(would_deliver, must_cancel_skips, should_retry) per pass:")
    for p in passes:
        print(f"  {p}")

asyncio.run(main())

Output on 4.14.2 and master:

(would_deliver, must_cancel_skips, should_retry) per pass:
  (0, 1, True)
  (0, 0, True)
  (1, 0, True)
  (0, 0, True)

The first pass delivers nothing, skips the task on _must_cancel, and arms a retry that cannot contribute anything to that task's cancellation.