Repository: agronholm/anyio

Issue:
Raising an error inside a task of a task group can produce: `RuntimeError: called 'started' twice on the same task status`
### 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.3.0

### Python version

3.12.2

### What happened?

Calling task_status.started() can raise `RuntimeError: called 'started' twice on the same task status` under certain circumstances (see examples), even though it was not called twice.
The reason for this is that the underlying future gets awaited in the start() method. When the start() method gets canceled, the underlying future gets also canceled which makes calling set_result on the future later fail and raise the erroneous RuntimeError.
This can be fixed by using asyncio.shield when awaiting the future.
https://github.com/agronholm/anyio/blob/05a3d02ed383f996ad93e8a7a4f6dcce153a06b6/src/anyio/_backends/_asyncio.py#L782

### How can we reproduce the bug?

```python
from anyio import create_task_group, run
from anyio.abc import TaskStatus


async def taskfunc(*, task_status: TaskStatus):
    task_status.started()
    raise ValueError


async def main():
    async with create_task_group() as tg:
        await tg.start(taskfunc)
        await tg.start(taskfunc)


run(main)
```
or
```python
from anyio import create_task_group, run
from anyio.abc import TaskStatus


async def taskfunc(*, task_status: TaskStatus):
    task_status.started()


async def main():
    async with create_task_group() as tg:
        coro = tg.start(taskfunc)
        tg.cancel_scope.cancel()
        await coro


run(main)
```

Discussion:
agronholm: I'm reviewing this now. Would you mind adding a changelog entry?
dominik-schwabe: I updated the changelog, but I wasn't happy with it and changed it with a force push and I think this removed you commit https://github.com/agronholm/anyio/commit/6704b9ed0bfd2f651d04a8c7871073d268cfcfa2 .
agronholm: Yeah, just don't force push your changes :) I think my revisions made the entry more accurate, yes?
dominik-schwabe: Sorry for this, I didn't expect you to be so fast.

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index f07c289..f0dcfdb 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -5,6 +5,10 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 **UNRELEASED**
 
+- 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 erroneous ``TypedAttributeLookupError`` if a typed attribute getter raises
   ``KeyError``
 - Fixed the asyncio backend not respecting the ``PYTHONASYNCIODEBUG`` environment
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index d64e553..fdf1cf4 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -620,9 +620,10 @@ class _AsyncioTaskStatus(abc.TaskStatus):
         try:
             self._future.set_result(value)
         except asyncio.InvalidStateError:
-            raise RuntimeError(
-                "called 'started' twice on the same task status"
-            ) from None
+            if not self._future.cancelled():
+                raise RuntimeError(
+                    "called 'started' twice on the same task status"
+                ) from None
 
         task = cast(asyncio.Task, current_task())
         _task_states[task].parent_id = self._parent_id


Test patch:
diff --git a/tests/test_taskgroups.py b/tests/test_taskgroups.py
index db662a7..d54115f 100644
--- a/tests/test_taskgroups.py
+++ b/tests/test_taskgroups.py
@@ -123,6 +123,17 @@ async def test_start_called_twice() -> None:
         assert value is None
 
 
+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()
+        await coro
+
+
 async def test_start_with_value() -> None:
     async def taskfunc(*, task_status: TaskStatus) -> None:
         task_status.started("foo")