Skip to content

[Bug] Using TaskGroup for child workflow execution does not correctly detect failure #1020

Closed

Description

@cpnielsen

What are you really trying to do?

I want to make use of asyncio.TaskGroup to execute a number of child workflows in parallel. However, if one of the child workflows fail, it does not cause the parent workflow to fail, but instead continues to run forever.

The expected behavior would be for child workflow execution failure to cause the parent workflow to fail as well, using default configuration.

Describe the bug

When a child workflow fails it raises a ChildWorkflowError. However, when using TaskGroup it is encapsulated in an ExceptionGroup and raised, which does not fail the calling workflow, causing it to run forever as per the documented error logic (https://docs.temporal.io/references/failures#errors-in-workflows).

At a glance it looks like the Temporal code never considers ExceptionGroup or other inner exceptions that are of the FailureError type that triggers an upstream failure.

Note, that if I instead use asyncio.gather it works as expected. This is expected, as asyncio.gather raises the first exception raised directly, passing the raw failure up which correctly triggers an overall failure.

Minimal Reproduction

# workflows.py
import asyncio
from datetime import timedelta
from typing import List

from temporalio import workflow
from temporalio.common import RetryPolicy

# Import activity, passing it through the sandbox without reloading the module
with workflow.unsafe.imports_passed_through():
    from activities import say_hello


@workflow.defn
class SayHello:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            say_hello, name, start_to_close_timeout=timedelta(seconds=5), retry_policy=RetryPolicy(maximum_attempts=3)
        )


@workflow.defn
class MasterWorkflow:
    @workflow.run
    async def run(self, names: List[str]) -> List[str]:
        tasks: List[asyncio.Task[str]] = []
        async with asyncio.TaskGroup() as tg:
            for name in names:
                tasks.append(
                    tg.create_task(
                        workflow.execute_child_workflow(
                            SayHello.run, name, id=f"say-hello-{name}", retry_policy=RetryPolicy(maximum_attempts=1)
                        )
                    )
                )

        greetings = [task.result() for task in tasks]

        return greetings
# activities.py
from temporalio import activity


@activity.defn
async def say_hello(name: str) -> str:
    if name == "fail":
        raise Exception("Uh-oh")
    return f"Hello, {name}!"
# run_workflow.py
import asyncio

from temporalio.client import Client

from run_worker import MasterWorkflow


async def main():
    # Create client connected to server at the given address
    client = await Client.connect("localhost:7233")

    # Execute a workflow
    result = await client.execute_workflow(
        MasterWorkflow.run, ["ok", "fail", "okay", "cool"], id="master-workflow", task_queue="hello-task-queue"
    )

    print(f"Result: {result}")


if __name__ == "__main__":
    asyncio.run(main())
# run_worker.py
import asyncio

from temporalio.client import Client
from temporalio.worker import Worker

from activities import say_hello
from workflows import MasterWorkflow, SayHello


async def main():
    client = await Client.connect("localhost:7233", namespace="default")
    # Run the worker
    worker = Worker(
        client, task_queue="hello-task-queue", workflows=[MasterWorkflow, SayHello], activities=[say_hello]
    )
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

The exception raised from the workflow is of type ExceptionGroup with the message unhandled errors in a TaskGroup (1 sub-exception).

Changing the code to user asyncio.gather instead correctly registers as a "Child Workflow Execution Failure".

Environment/Versions

  • OS and processor: M2 Macbook Pro, x64 Linux
  • Temporal Version: 1.12.0 (client+worker)
  • Tested consistently locally (Macbook), but observed running in kubernetes as well.

As the nature of the bug is "working as intended", I expect it to behave identically across environments.

Additional context

I am happy to provide a PR (with tests) if you can point me to some good starting point. I checked for places where the code handles/raises FailureError, but I don't feel sure enough to just dive in.

Activity

  1. changed the title [-][Bug] Using TaskGroup for child workflow execution does correctly detect failure[/-] [+][Bug] Using TaskGroup for child workflow execution does not correctly detect failure[/+] on Aug 6, 2025
  2. cpnielsen commented on Aug 6, 2025

    @cpnielsen
    Author

    After looking at the code again, I think the following might solve it:

    # temporalio/worker/_workflow_instance.py
    
    # <-- IMPORTS -->
    
    LOG_IGNORE_DURING_DELETE = False
    
    if sys.version_info < (3, 11):
        from exceptiongroup import ExceptionGroup
    
    # <-- MORE CODE -->
    
        def _is_workflow_failure_exception(self, err: BaseException) -> bool:
            # An exception is a failure instead of a task fail if it's already a
            # failure error or if it is a timeout error or if it is an instance of
            # any of the failure types in the worker or workflow-level setting
            wf_failure_exception_types = self._defn.failure_exception_types
            if self._dynamic_failure_exception_types is not None:
                wf_failure_exception_types = self._dynamic_failure_exception_types
            return (
                isinstance(err, temporalio.exceptions.FailureError)
                or self._is_exception_group_failure(err)
                or isinstance(err, asyncio.TimeoutError)
                or any(isinstance(err, typ) for typ in wf_failure_exception_types)
                or any(
                    isinstance(err, typ)
                    for typ in self._worker_level_failure_exception_types
                )
            )
    
        def _is_exception_group_failure(self, err: BaseException) -> bool:
            if isinstance(err, ExceptionGroup):
                return any(self._is_workflow_failure_exception(e) for e in err.exceptions)
    
            return False
    
    # <-- MORE CODE -->
  3. cpnielsen commented on Aug 6, 2025

    @cpnielsen
    Author

    @cretz Could you re-open this please, I accidentally hit a hotkey to close the issue 🤦

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    bugSomething isn't working

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions