Log warning when a Concurrent, Dask, or Ray versions of PrefectFuture are garbage collection before resolution

src/integrations/prefect-dask/prefect_dask/task_runners.py

@@ -79,13 +79,16 @@ def count_to(highest_number):
 
 from prefect.client.schemas.objects import State, TaskRunInput
 from prefect.futures import PrefectFuture, PrefectWrappedFuture
+from prefect.logging.loggers import get_logger, get_run_logger
 from prefect.task_runners import TaskRunner
 from prefect.tasks import Task
 from prefect.utilities.asyncutils import run_coro_as_sync
 from prefect.utilities.collections import visit_collection
 from prefect.utilities.importtools import from_qualified_name, to_qualified_name
 from prefect_dask.client import PrefectDaskClient
 
+logger = get_logger(__name__)
+
 
 class PrefectDaskFuture(PrefectWrappedFuture[distributed.Future]):
     """
@@ -129,6 +132,18 @@ def result(
             _result = run_coro_as_sync(_result)
         return _result
 
+    def __del__(self):
+        if self._final_state or self._wrapped_future.done():
+            return
+        try:
+            local_logger = get_run_logger()
+        except Exception:
+            local_logger = logger
+        local_logger.warning(
+            "A future was garbage collected before it resolved."
+            " Please call `.wait()` or `.result()` on futures to ensure they resolve.",
+        )
+
 
 class DaskTaskRunner(TaskRunner):
     """

src/integrations/prefect-dask/tests/test_task_runners.py

@@ -319,6 +319,39 @@ async def adapt(self, **kwargs):
             with task_runner:
                 assert task_runner._cluster._adapt_called
 
+    def test_warns_if_future_garbage_collection_before_resolving(
+        self, caplog, task_runner
+    ):
+        @task
+        def test_task():
+            return 42
+
+        @flow(task_runner=task_runner)
+        def test_flow():
+            for _ in range(10):
+                test_task.submit()
+
+        test_flow()
+
+        assert "A future was garbage collected before it resolved" in caplog.text
+
+    def test_does_not_warn_if_future_resolved_when_garbage_collected(
+        self, task_runner, caplog
+    ):
+        @task
+        def test_task():
+            return 42
+
+        @flow(task_runner=task_runner)
+        def test_flow():
+            futures = [test_task.submit() for _ in range(10)]
+            for future in futures:
+                future.wait()
+
+        test_flow()
+
+        assert "A future was garbage collected before it resolved" not in caplog.text
+
     class TestInputArguments:
         async def test_dataclasses_can_be_passed_to_task_runners(self, task_runner):
             """

src/integrations/prefect-ray/prefect_ray/task_runners.py

@@ -82,6 +82,7 @@ def count_to(highest_number):
 from prefect.client.schemas.objects import TaskRunInput
 from prefect.context import serialize_context
 from prefect.futures import PrefectFuture, PrefectWrappedFuture
+from prefect.logging.loggers import get_logger, get_run_logger
 from prefect.states import State, exception_to_crashed_state
 from prefect.task_engine import run_task_async, run_task_sync
 from prefect.task_runners import TaskRunner
@@ -90,6 +91,8 @@ def count_to(highest_number):
 from prefect.utilities.collections import visit_collection
 from prefect_ray.context import RemoteOptionsContext
 
+logger = get_logger(__name__)
+
 
 class PrefectRayFuture(PrefectWrappedFuture[ray.ObjectRef]):
     def wait(self, timeout: Optional[float] = None) -> None:
@@ -129,6 +132,24 @@ def result(
             _result = run_coro_as_sync(_result)
         return _result
 
+    def __del__(self):
+        if self._final_state:
+            return
+        try:
+            ray.get(self.wrapped_future, timeout=0)
+            return
+        except GetTimeoutError:
+            pass
+
+        try:
+            local_logger = get_run_logger()
+        except Exception:
+            local_logger = logger
+        local_logger.warning(
+            "A future was garbage collected before it resolved."
+            " Please call `.wait()` or `.result()` on futures to ensure they resolve.",
+        )
+
 
 class RayTaskRunner(TaskRunner[PrefectRayFuture]):
     """

src/integrations/prefect-ray/tests/test_task_runners.py

@@ -456,3 +456,36 @@ def flow_with_dependent_tasks():
                 e.submit(wait_for=[b_future])
 
         flow_with_dependent_tasks()
+
+    def test_warns_if_future_garbage_collection_before_resolving(
+        self, caplog, task_runner
+    ):
+        @task
+        def test_task():
+            return 42
+
+        @flow(task_runner=task_runner)
+        def test_flow():
+            for _ in range(10):
+                test_task.submit()
+
+        test_flow()
+
+        assert "A future was garbage collected before it resolved" in caplog.text
+
+    def test_does_not_warn_if_future_resolved_when_garbage_collected(
+        self, task_runner, caplog
+    ):
+        @task
+        def test_task():
+            return 42
+
+        @flow(task_runner=task_runner)
+        def test_flow():
+            futures = [test_task.submit() for _ in range(10)]
+            for future in futures:
+                future.wait()
+
+        test_flow()
+
+        assert "A future was garbage collected before it resolved" not in caplog.text

src/prefect/futures.py

@@ -10,7 +10,7 @@
 from prefect.client.orchestration import get_client
 from prefect.client.schemas.objects import TaskRun
 from prefect.exceptions import ObjectNotFound
-from prefect.logging.loggers import get_logger
+from prefect.logging.loggers import get_logger, get_run_logger
 from prefect.states import Pending, State
 from prefect.task_runs import TaskRunWaiter
 from prefect.utilities.annotations import quote
@@ -143,6 +143,18 @@ def result(
             _result = run_coro_as_sync(_result)
         return _result
 
+    def __del__(self):
+        if self._final_state or self._wrapped_future.done():
+            return
+        try:
+            local_logger = get_run_logger()
+        except Exception:
+            local_logger = logger
+        local_logger.warning(
+            "A future was garbage collected before it resolved."
+            " Please call `.wait()` or `.result()` on futures to ensure they resolve.",
+        )
+
 
 class PrefectDistributedFuture(PrefectFuture):
     """

tests/test_futures.py

@@ -80,6 +80,18 @@ def test_result_with_final_state_and_raise_on_failure(self):
         with pytest.raises(ValueError, match="oops"):
             future.result(raise_on_failure=True)
 
+    def test_warns_if_not_resolved_when_garbage_collected(self, caplog):
+        PrefectConcurrentFuture(uuid.uuid4(), Future())
+
+        assert "A future was garbage collected before it resolved" in caplog.text
+
+    def test_does_not_warn_if_resolved_when_garbage_collected(self, caplog):
+        wrapped_future = Future()
+        wrapped_future.set_result(Completed())
+        PrefectConcurrentFuture(uuid.uuid4(), wrapped_future)
+
+        assert "A future was garbage collected before it resolved" not in caplog.text
+
 
 class TestResolveFuturesToStates:
     async def test_resolve_futures_transforms_future(self):