Based on my research, I can explain the behavior you're seeing and provide a solution: When you use

return_state=True

, you're telling Prefect to return a State object instead of the raw result. This creates an explicit state dependency between tasks, which means downstream tasks will wait for the complete state transition of the upstream task before proceeding. This can introduce delays because: 1. The state transition process is more involved than simple result passing 2. The task must fully complete its state lifecycle before downstream tasks can begin Here are two ways to handle this: 1. Avoid using

return_state=True

unless necessary:

python
   @task
   def upstream_task():
       return "result"  # Returns raw result
   
   @task
   def downstream_task(data):
       # Will start as soon as upstream_task's result is available
       process(data)

2. Use state handling only where needed: If you need state information for specific error handling or control flow, consider using it only at those specific points rather than throughout your entire pipeline:

python
   @task(return_state=True)
   def critical_task():
       # Only use return_state where you need state information
       return "result"
   
   @task
   def normal_task(data):
       # Regular tasks can process the results directly
       process(data)

If you absolutely need state information but want to minimize delays, you can also use asynchronous execution with

task_runner=AsyncTaskRunner()

in your flow definition, which can help manage state transitions more efficiently. Remember that in Prefect 3.x, the default behavior (without

return_state=True

) already provides good error handling and observability without introducing unnecessary delays in task execution.