@Marvin task stucks in RUNNING state and even with timeout is not going our of this state

It seems like you are experiencing an issue where tasks get stuck in the RUNNING state despite having a timeout set. This behavior has been documented in several Prefect GitHub issues. Here are a couple of scenarios that might match your problem: 1. Sub-deployments stuck in Canceling state: A race condition may occur when a parent deployment calls a child deployment, and the parent is canceled, leaving the child in a canceling state indefinitely. This is discussed in this GitHub issue. 2. Flows exceeding timeout with subflows in a Crashed state: If a subflow crashes, the parent flow may hit the timeout but remain in the RUNNING state, as reported in this GitHub issue. 3. Agent process interruption: If the

prefect agent start

process is interrupted (e.g., using CTRL+C), the flow can remain stuck in the RUNNING state. Here's a related GitHub issue. To help you better, could you please specify which version of Prefect you are using? The resolution and debugging steps can differ between Prefect 2.x and 3.x.

@Marvin I use prefect 3.0.0

Here are some insights from the Prefect 3.x documentation regarding tasks getting stuck in the RUNNING state and issues related to timeouts: 1. Flow and Task States: - Task runs can be in different states like

COMPLETED

,

FAILED

, or

CRASHED

. The final state depends on factors such as code execution and external signals like

KeyboardInterrupt

or

SIGTERM

. If a task encounters an infrastructure issue, it might enter a

CRASHED

state instead of transitioning out of

RUNNING

. - Resource: Managing States in Prefect 3.0 2. Manual Cancellation and Timeouts: - You can cancel a flow run using the

prefect flow-run cancel

CLI command or through the UI. This can help in cases where a flow is stuck. - Flow timeouts are enforced using the

timeout_seconds

keyword. If a flow exceeds this duration, it should get marked as

TimedOut

and fail. - Example:

python
     from prefect import flow
     import time

     @flow(timeout_seconds=1, log_prints=True)
     def show_timeouts():
         print("I will execute")
         time.sleep(5)
         print("I will not execute")

- Resource: Timeouts in Prefect 3.0 For your specific issue of tasks not transitioning out of the RUNNING state despite a timeout, verify the following: - The timeout is set correctly at both the task and flow levels. - There are no underlying infrastructure issues causing tasks to hang, leading to a

CRASHED

state. - Check Prefect logs for any clues or irregularities. If the issue persists, refer to the examples above to ensure your implementation aligns with recommended practices, and consider creating a new GitHub issue with detailed information if needed.

@Marvin how to remove flow or task in a stucked state?

To remove or cancel a flow or task that is stuck in the RUNNING state in Prefect 3.x, you can follow these methods: 1. Cancel Flow Runs Using the CLI: You can cancel a flow run directly through the Prefect CLI by passing the flow run ID.

sh
   prefect flow-run cancel 'your-flow-run-id'

Example:

sh
   prefect flow-run cancel 'a55a4804-9e3c-4042-8b59-b3b6b7618736'

2. Cancel Flow Runs Programmatically: You can use the Prefect client to cancel flow runs programmatically. Below is a code example demonstrating how to cancel flow runs in specific states (Pending, Running, Scheduled, Late):

python
   from prefect.client.orchestration import get_client
   from prefect.states import StateType
   import anyio

   async def list_flow_runs_with_states(states):
       async with get_client() as client:
           flow_runs = await client.read_flow_runs(state_types=states)
       return flow_runs

   async def cancel_flow_runs(flow_runs):
       async with get_client() as client:
           for idx, flow_run in enumerate(flow_runs):
               print(f"[{idx + 1}] Cancelling flow run '{flow_run.name}' with ID '{flow_run.id}'")
               state_updates = {
                   "name": "Cancelled",
                   "type": StateType.CANCELLED
               }
               state = flow_run.state.copy(update=state_updates)
               await client.set_flow_run_state(flow_run.id, state, force=True)

   async def bulk_cancel_flow_runs():
       states = [StateType.PENDING, StateType.RUNNING, StateType.SCHEDULED, StateType.LATE]
       flow_runs = await list_flow_runs_with_states(states)
       while len(flow_runs) > 0:
           print(f"Cancelling {len(flow_runs)} flow runs\n")
           await cancel_flow_runs(flow_runs)
           flow_runs = await list_flow_runs_with_states(states)
       print("Done!")

   if __name__ == "__main__":
       anyio.run(bulk_cancel_flow_runs)

These techniques should help you manage and cancel stuck flow or task runs. For further details, consult the Prefect 3.x documentation on interacting with the API and managing states.

Are you using task run concurrency limits? Do the tasks actually run in a way you can observe and just our record of the run remains in running state, or do the tasks actually hang during execution?

@Andrew Brookins, no, I don't use concurrency limits. I think the problem is that flow itself is hanging, not the task. If a worker that executes this flow (using .serve) fails (like crashes unexpectedly) flow remains in RUNNING state, as well as all containing tasks, even when worker is up again after some time. I want to implement custom heartbeat functionality with background process to cancel or even completely delete the flow if executing worker is down (not sure why isn't it implemented in prefect) Record of the run remains in running state, however task is dead with worker