Repository: PrefectHQ/prefect

Issue:
duration parameter is not taken into account from task_pause
## Description
<!-- A clear description of the bug -->
`duration` parameter in `prefect.utilities.tasks.pause_task` seems to be ignored. Based on documentation, if `duration` is specified, then, the task should be resumed after a period of time



## Expected Behavior
<!-- What did you expect to happen instead? -->
Resume task run after specified period of time.



## Reproduction
<!-- A minimal example that exhibits the behavior. -->
```
import datetime

from prefect import Flow, task
from prefect.utilities.tasks import pause_task


@task
def say_hello():
    pause_task("wait or approve..", datetime.timedelta(seconds=10))
    print('hello')


with Flow("My Flow") as f:
    say_hello()
```

When I executed this flow from the UI, I noticed that the task stays paused for ever (well, for 10 years :P )

![image](https://user-images.githubusercontent.com/8322266/105976777-42170600-6099-11eb-8cff-b4bcab172657.png)

**NOTE**: Every time the duration time is reached, the task's state changes like: Paused -> Resumed -> Paused.

## Environment
<!-- Any additional information about your environment

Optionally run `prefect diagnostics` from the command line and paste the information here. -->
```
{
  "config_overrides": {},
  "env_vars": [
    "PREFECT__CONTEXT__SECRETS__XXxx",
    "PREFECT__CONTEXT__SECRETS__XXxx"
  ],
  "system_information": {
    "platform": "Linux-5.8.0-7630-generic-x86_64-with-debian-bullseye-sid",
    "prefect_backend": "server",
    "prefect_version": "0.14.4",
    "python_version": "3.7.9"
  }
}
```


Discussion:

Patch:
diff --git a/changes/issue4031.yaml b/changes/issue4031.yaml
new file mode 100644
index 0000000000..f521836e2c
--- /dev/null
+++ b/changes/issue4031.yaml
@@ -0,0 +1,2 @@
+fix:
+  - "Fix issue with fixed duration Paused states not resuming properly - [#4031](https://github.com/PrefectHQ/prefect/issues/4031)"
diff --git a/src/prefect/engine/task_runner.py b/src/prefect/engine/task_runner.py
index c1642076d6..66950ccf4a 100644
--- a/src/prefect/engine/task_runner.py
+++ b/src/prefect/engine/task_runner.py
@@ -25,6 +25,7 @@ from prefect.engine.state import (
     Failed,
     Looped,
     Mapped,
+    Paused,
     Pending,
     Resume,
     Retrying,
@@ -141,7 +142,13 @@ class TaskRunner(Runner):
         else:
             run_count = state.context.get("task_run_count", 1)
 
-        if isinstance(state, Resume):
+        # detect if currently Paused with a recent start_time
+        should_resume = (
+            isinstance(state, Paused)
+            and state.start_time
+            and state.start_time <= pendulum.now("utc")  # type: ignore
+        )
+        if isinstance(state, Resume) or should_resume:
             context.update(resume=True)
 
         if "_loop_count" in state.context:


Test patch:
diff --git a/tests/engine/test_task_runner.py b/tests/engine/test_task_runner.py
index 534537c29f..9be34ec8fd 100644
--- a/tests/engine/test_task_runner.py
+++ b/tests/engine/test_task_runner.py
@@ -429,6 +429,23 @@ class TestInitializeRun:
             result = TaskRunner(Task()).initialize_run(state=Resume(), context=ctx)
             assert result.context.resume is True
 
+    def test_task_runner_puts_resume_in_context_if_paused_start_time_elapsed(self):
+        with prefect.context() as ctx:
+            assert "resume" not in ctx
+            result = TaskRunner(Task()).initialize_run(
+                state=Paused(start_time=pendulum.now("utc")), context=ctx
+            )
+            assert result.context.resume is True
+
+    def test_task_runner_ignores_resume_in_context_if_paused_start_time_in_future(self):
+        with prefect.context() as ctx:
+            assert "resume" not in ctx
+            result = TaskRunner(Task()).initialize_run(
+                state=Paused(start_time=pendulum.now("utc").add(seconds=10)),
+                context=ctx,
+            )
+            assert "resume" not in ctx
+
     def test_task_runner_puts_checkpointing_in_context(self):
         with prefect.context() as ctx:
             assert "checkpointing" not in ctx