Recover from wf failure manually how to test properly

Community Support
1

I am trying to implement the following behavior.
Workflow has 3 attempts(RetryPolicy) to do the work, if it fails, I want to run another wf that will do other logic instead.

try:
       result = await workflow.execute_child_workflow(
            RetryWorkflow.run,
            params,
            id="wf_id",
            task_queue="queue",
            retry_policy=RetryPolicy(maximum_attempts=3)
        )
except WorkflowFailureError as err:
            if isinstance(err.value.cause, ApplicationError) and str(err.value.cause) == "attempt 3":
                  # other stuff goes here

How can I fail wf after 3 attempts to mock the flow, raise with the error seems not to be working as expected. The code inside except is unreachable and wf continues running.

2

Can you clarify a bit here? Are you saying the child workflow is not running 3 times and then throwing? Throwing a different error? Only in test cases? What type of test environment/server?

3

I am doing this inside the wf
raise ApplicationError("attempt 3")

I just want wf to fail properly, like it will do IRL, after 3 runs, how to achieve that?

4

Just checked by hand seems the raise is fine for the test. But when I raise the error my parent wf is also terminating. How to fix this?
I have the following setup:
Parent_WF → Child_WF(here I am going to catch the error) → Child_of_child_WF(here I expect the error)

5

I whipped up a little example demonstrating that max_attempts works as expected on children:

import asyncio
import logging
from uuid import uuid4

from temporalio import workflow
from temporalio.common import RetryPolicy
from temporalio.exceptions import ApplicationError
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker


@workflow.defn
class AlwaysFailWorkflow:
    @workflow.run
    async def run(self) -> None:
        workflow.logger.info("Running child, attempt %s" % workflow.info().attempt)
        raise ApplicationError("Intentional failure")


@workflow.defn
class RetryChildWorkflow:
    @workflow.run
    async def run(self) -> None:
        try:
            await workflow.execute_child_workflow(
                AlwaysFailWorkflow.run, retry_policy=RetryPolicy(maximum_attempts=3)
            )
        except:
            workflow.logger.exception("Caught error")
            raise


async def main():
    logging.basicConfig(level=logging.INFO)

    async with await WorkflowEnvironment.start_local() as env:
        logging.info("Starting worker")
        task_queue = f"tq-{uuid4()}"
        async with Worker(
            env.client,
            task_queue=task_queue,
            workflows=[RetryChildWorkflow, AlwaysFailWorkflow],
        ):
            logging.info("Running workflow")
            result = await env.client.execute_workflow(
                RetryChildWorkflow.run,
                id=f"wf-{uuid4()}",
                task_queue=task_queue,
            )
            print(f"Result: {result}")


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

This outputs:

INFO:root:Starting worker
INFO:root:Running workflow
INFO:temporalio.workflow:Running child, attempt 1 ({'attempt': 1, 'namespace': 'default', 'run_id': '746b9366-7ecc-49ee-89b3-f7055e660d52', 'task_queue': 'tq-b41af022-ca79-4f5c-b1e7-d3f9431b320a', 'workflow_id': '46335f27-f0ad-4e5a-a12f-338057c82bb8', 'workflow_type': 'AlwaysFailWorkflow'})
INFO:temporalio.workflow:Running child, attempt 2 ({'attempt': 2, 'namespace': 'default', 'run_id': 'a91238cf-adbe-4098-8e9e-bd2a3398d757', 'task_queue': 'tq-b41af022-ca79-4f5c-b1e7-d3f9431b320a', 'workflow_id': '46335f27-f0ad-4e5a-a12f-338057c82bb8', 'workflow_type': 'AlwaysFailWorkflow'})
INFO:temporalio.workflow:Running child, attempt 3 ({'attempt': 3, 'namespace': 'default', 'run_id': 'e497ccd2-e496-49fe-81d9-60057bfa3c8f', 'task_queue': 'tq-b41af022-ca79-4f5c-b1e7-d3f9431b320a', 'workflow_id': '46335f27-f0ad-4e5a-a12f-338057c82bb8', 'workflow_type': 'AlwaysFailWorkflow'})
ERROR:temporalio.workflow:Caught error ({'attempt': 1, 'namespace': 'default', 'run_id': 'ff600866-65fb-4244-b549-13cb90ed2011', 'task_queue': 'tq-b41af022-ca79-4f5c-b1e7-d3f9431b320a', 'workflow_id': 'wf-f959c547-8a57-4a9f-b2be-175c4aad39db', 'workflow_type': 'RetryChildWorkflow'})
temporalio.exceptions.ApplicationError: Intentional failure

The above exception was the direct cause of the following exception:

[...bunch of exception lines snipped...]
6

I think you need to catch temporalio.exceptions.ChildWorkflowError instead.