Repository: PrefectHQ/prefect
Pull request: Adds circuit breaker when a task depends on itself
<!-- 
Thanks for opening a pull request to Prefect! We've got a few requests to help us review contributions:

- Make sure that your title neatly summarizes the proposed changes.
- Provide a short overview of the change and the value it adds.
- Share an example to help us understand the change in user experience.
- Confirm that you've done common tasks so we can give a timely review.
- Review our contribution guidelines: https://docs.prefect.io/latest/contributing/overview/

Happy engineering!
-->

<!-- Include an overview here -->
Adds a circuit breaker to raise when a task depends on itself which would otherwise cause a flow to hang indefinitely.

Closes https://github.com/PrefectHQ/prefect/issues/13874

### Example
<!-- 
Share an example of the change in action.

A code blurb is best. Changes to features should include an example that is executable by a new user.
If changing documentation, a link to a preview of the page is great.
 -->
This flow sneakily has a task that depends on itself by updating an array in place:
```python
from prefect import flow, task

@task
def say_hello(name):
    return f"Hello {name}!"

@flow
def my_flow():
    greeting_queue = []
    for i in range(2):
        if greeting_queue:
            wait_for = greeting_queue
        else:
            wait_for = []
        future = say_hello.submit(name=f"Person {i}", wait_for=wait_for)
        greeting_queue.append(future)
    
    for fut in greeting_queue:
        print(fut.result())
```
Running this flow with this PR will raise an error like this:
```
ValueError: Discovered a task depending on itself. Raising to avoid a deadlock. Please inspect the inputs and dependencies of say_hello.
```

### Checklist
<!-- These boxes may be checked after opening the pull request. -->

- [ ] This pull request references any related issue by including "closes `<link to issue>`"
 	- If no issue exists and your change is not a small fix, please [create an issue](https://github.com/PrefectHQ/prefect/issues/new/choose) first.
- [x] If this pull request adds new functionality, it includes unit tests that cover the changes
- [x] This pull request includes a label categorizing the change e.g. `maintenance`, `fix`, `feature`, `enhancement`, `docs`.
  <!-- If you do not have permission to add a label, a maintainer will add one for you -->

For documentation changes:

- [ ] This pull request includes redirect settings in `mint.json` for files that are removed or renamed.

For new functions or classes in the Python SDK:

- [ ] This pull request includes helpful docstrings.
- [ ] If a new Python file was added, this pull request contains a stub page in the Python SDK docs and an entry in `docs/mint.json` navigation.


Problem statement:
Strange hanging behavior with `ThreadPoolTaskRunner` and `wait_for`
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar issue and didn't find it.
- [X] I searched the Prefect documentation for this issue.
- [X] I checked that this issue is related to Prefect and not one of its dependencies.

### Bug summary

When using the default task runner and `wait_for`, flow runs hang unexpectedly.

The flow below runs as expected with `DaskTaskRunner`.

### Reproduction

```python3
from prefect import flow, task

@task
def say_hello(name):
    return f"Hello {name}!"

@flow
def my_flow():
    # This code works fine
    # future_1 = say_hello.submit(name=f"Person 1", wait_for=[])
    # future_2 = say_hello.submit(name="Person 2", wait_for=[future_1])
    # print(future_1.result())
    # print(future_2.result())

    # This code hangs indefinitely
    greeting_queue = []
    for i in range(2):
        if greeting_queue:
            wait_for = greeting_queue
        else:
            wait_for = []
        future = say_hello.submit(name=f"Person {i}", wait_for=wait_for)
        greeting_queue.append(future)
    
    for fut in greeting_queue:
        print(fut.result())
```


### Error

```python3
13:42:45.621 | INFO    | prefect.engine - Created flow run 'blue-guppy' for flow 'my-flow'
13:42:45.651 | INFO    | Flow run 'blue-guppy' - Submitting task say_hello to thread pool executor...
13:42:45.653 | INFO    | Flow run 'blue-guppy' - Submitting task say_hello to thread pool executor...
13:42:45.694 | INFO    | prefect.engine - Created task run 'say_hello-0' for task 'say_hello'
13:42:45.712 | INFO    | prefect.engine - Created task run 'say_hello-1' for task 'say_hello'
13:42:45.784 | INFO    | Task run 'say_hello-0' - Finished in state Completed()
Hello Person 0!
```


### Versions

```Text
Version:             3.0.0rc2+11.g8bb6a81892
API version:         0.8.4
Python version:      3.10.4
Git commit:          8bb6a818
Built:               Fri, Jun 7, 2024 12:33 PM
OS/Arch:             darwin/arm64
Profile:             tmp
Server type:         ephemeral/cloud
Pydantic version:    2.7.1
Server:
  Database:          sqlite
  SQLite version:    3.38.3
```


### Additional context

_No response_

Hints from issue discussion:
potentially related to https://github.com/PrefectHQ/nebula/issues/7925

Commit message: Adds circuit breaker when a task depends on itself

Patch to review:
diff --git a/src/prefect/task_engine.py b/src/prefect/task_engine.py
index ba03ff2a3777..5862b642aa90 100644
--- a/src/prefect/task_engine.py
+++ b/src/prefect/task_engine.py
@@ -219,7 +219,7 @@ def _wait_for_dependencies(self):
             return_data=False,
             max_depth=-1,
             remove_annotations=True,
-            context={},
+            context={"current_task_run": self.task_run, "current_task": self.task},
         )
 
     def begin_run(self):
diff --git a/src/prefect/utilities/engine.py b/src/prefect/utilities/engine.py
index 7bec5432d799..cdaea9e51001 100644
--- a/src/prefect/utilities/engine.py
+++ b/src/prefect/utilities/engine.py
@@ -786,6 +786,17 @@ def resolve_to_final_result(expr, context):
         raise StopVisiting()
 
     if isinstance(expr, NewPrefectFuture):
+        upstream_task_run = context.get("current_task_run")
+        upstream_task = context.get("current_task")
+        if (
+            upstream_task
+            and upstream_task_run
+            and expr.task_run_id == upstream_task_run.id
+        ):
+            raise ValueError(
+                f"Discovered a task depending on itself. Raising to avoid a deadlock. Please inspect the inputs and dependencies of {upstream_task.name}."
+            )
+
         expr.wait()
         state = expr.state
     elif isinstance(expr, State):
diff --git a/tests/test_tasks.py b/tests/test_tasks.py
index 1fe66c7ba916..725936cbda75 100644
--- a/tests/test_tasks.py
+++ b/tests/test_tasks.py
@@ -640,6 +640,28 @@ def test_flow():
             assert y == i + 1
             assert exceptions_equal(z, ValueError("Fail task"))
 
+    async def test_raises_if_depends_on_itself(self):
+        @task
+        def say_hello(name):
+            return f"Hello {name}!"
+
+        @flow
+        def my_flow():
+            greeting_queue = []
+            for i in range(3):
+                if greeting_queue:
+                    wait_for = greeting_queue
+                else:
+                    wait_for = []
+                future = say_hello.submit(name=f"Person {i}", wait_for=wait_for)
+                greeting_queue.append(future)
+
+            for fut in greeting_queue:
+                print(fut.result())
+
+        with pytest.raises(ValueError):
+            my_flow()
+
 
 class TestTaskStates:
     @pytest.mark.parametrize("error", [ValueError("Hello"), None])


Code review comments:
diff_hunk: @@ -640,6 +640,28 @@ def test_flow():
             assert y == i + 1
             assert exceptions_equal(z, ValueError("Fail task"))
 
+    async def test_raises_if_depends_on_itself(self):
+        @task
+        def say_hello(name):
+            return f"Hello {name}!"
+
+        @flow
+        def my_flow():
+            greeting_queue = []
+            for i in range(3):
+                if greeting_queue:
+                    wait_for = greeting_queue
+                else:
+                    wait_for = []
+                future = say_hello.submit(name=f"Person {i}", wait_for=wait_for)
+                greeting_queue.append(future)
+
+            for fut in greeting_queue:
+                print(fut.result())
+
+        with pytest.raises(ValueError):
path: tests/test_tasks.py
text: @user1:
```suggestion
        with pytest.raises(ValueError, match="deadlock"):
```