Repository: PrefectHQ/prefect

Issue:
Task Run Fails to Read Upstream Result the First Time a Flow is Run on Windows
## Description
<!-- A clear description of the bug -->
![image](https://user-images.githubusercontent.com/2325367/94076056-4faa3f80-fdca-11ea-9269-56edfae373bf.png)
> it seems that the file is either getting cleaned up OR reading from the file has a bug that is specific to windows paths



## Expected Behavior
<!-- What did you expect to happen instead? -->
Result is read properly


## Reproduction
<!-- A minimal example that exhibits the behavior. -->




## Environment
<!-- Any additional information about your environment

Optionally run `prefect diagnostics` from the command line and paste the information here. -->
Windows

Discussion:

Patch:
diff --git a/changes/issue3378.yaml b/changes/issue3378.yaml
new file mode 100644
index 0000000000..6d27ee0f1f
--- /dev/null
+++ b/changes/issue3378.yaml
@@ -0,0 +1,2 @@
+enhancement:
+  - "Don't stop execution if the task runner fails to load a cached result - [#3378](https://github.com/PrefectHQ/prefect/issues/3378)"
diff --git a/src/prefect/engine/cloud/task_runner.py b/src/prefect/engine/cloud/task_runner.py
index 517556e641..c9bf748bb6 100644
--- a/src/prefect/engine/cloud/task_runner.py
+++ b/src/prefect/engine/cloud/task_runner.py
@@ -266,7 +266,16 @@ class CloudTaskRunner(TaskRunner):
                         sanitized_inputs,
                         prefect.context.get("parameters"),
                     ):
-                        return candidate_state.load_result(self.result)
+                        try:
+                            return candidate_state.load_result(self.result)
+                        except Exception:
+                            location = getattr(
+                                candidate_state._result, "location", None
+                            )
+                            self.logger.warning(
+                                f"Failed to load cached state data from {location}.",
+                                exc_info=True,
+                            )
 
                 self.logger.debug(
                     "Task '{name}': can't use cache because no candidate Cached states "


Test patch:
diff --git a/tests/engine/cloud/test_cloud_task_runner.py b/tests/engine/cloud/test_cloud_task_runner.py
index 08e05545a2..e4cc08616a 100644
--- a/tests/engine/cloud/test_cloud_task_runner.py
+++ b/tests/engine/cloud/test_cloud_task_runner.py
@@ -13,7 +13,7 @@ from prefect.core import Edge, Task
 from prefect.engine.cache_validators import all_inputs, duration_only
 from prefect.engine.cloud import CloudTaskRunner
 from prefect.engine.result import NoResult, Result, SafeResult, NoResult
-from prefect.engine.results import PrefectResult, SecretResult
+from prefect.engine.results import PrefectResult, SecretResult, LocalResult
 
 from prefect.engine.result_handlers import JSONResultHandler, ResultHandler
 from prefect.engine.runner import ENDRUN
@@ -253,6 +253,32 @@ def test_task_runner_validates_cached_states_if_task_has_caching(client):
     assert res.result == 42
 
 
+def test_task_runner_treats_unfound_files_as_invalid_caches(client, tmpdir):
+    @prefect.task(
+        cache_for=datetime.timedelta(minutes=1), result_handler=JSONResultHandler()
+    )
+    def cached_task():
+        return 42
+
+    state = Cached(
+        cached_result_expiration=datetime.datetime.utcnow()
+        + datetime.timedelta(minutes=2),
+        result=LocalResult(location=str(tmpdir / "made_up_data.prefect")),
+    )
+    old_state = Cached(
+        cached_result_expiration=datetime.datetime.utcnow()
+        + datetime.timedelta(days=1),
+        result=Result(13, JSONResultHandler()),
+    )
+    client.get_latest_cached_states = MagicMock(return_value=[state, old_state])
+
+    res = CloudTaskRunner(task=cached_task).run()
+    assert client.get_latest_cached_states.called
+    assert res.is_successful()
+    assert res.is_cached()
+    assert res.result == 13
+
+
 class TestCheckTaskCached:
     def test_reads_result_if_cached_valid(self, client):
         result = PrefectResult(location="2")