Repository: PrefectHQ/prefect
Issue category: Bug Report

Async tasks stays in Pending state until Completed on Windows Client
### Bug summary

This issue is directly related to #15607 . After few hours of trial and error I managed to bring more light to it, at least i hope so.

**My findings indicate that Running state had the same timestamp as Pending state.** 

Could there be an issue with `datetime.datetime.now()` (`pendulum.now()`) having lower resolution on Windows (5-15ms)? 

More precisely, time between emitting task run pending state in [AsyncTaskRunEngine.initialize_run](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1176) and emitting running state in following [AsyncTaskRunEngine.begin_run](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L904) is less than 15ms resulting in equal timestamp.

For Sync tasks this issue does not appear, probably due to wrapping `Task.create_local_run` in `run_coro_as_sync` [here](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L660) adding overhead and increasing time between Pending and Running state.

My current setup is Prefect Server with PostgreSQL in docker and a client on Windows 10 machine. Client on Ubuntu 22 is not causing any problems.

The following code should reproduce the duplicate key value violation on a database. **Notice the submission of async tasks**, as for sync tasks the issue wont trigger.

```python
from prefect import flow, task


@flow
def test_flow():
    tasks = [test_task.submit() for i in range(4)]
    [task.wait() for task in tasks]
    return [task.state for task in tasks]


@task
async def test_task():
    subtasks = [test_subtask.submit() for _ in range(10)]
    [task.wait() for task in subtasks]
    return [task.state for task in subtasks]


@task
async def test_subtask():
    return 42


if __name__ == "__main__":
    test_flow()

```

The partial log from database about the violation:
```
2024-11-25 00:14:31.474 UTC [82] ERROR:  23505: duplicate key value violates unique constraint "uq_task_run_state__task_run_id_timestamp_desc"
2024-11-25 00:14:31.474 UTC [82] DETAIL:  Key (task_run_id, "timestamp")=(a230f4b0-8e76-4540-8698-349b60c1f47b, 2024-11-25 00:14:30.432075+00) already exists.
2024-11-25 00:14:31.474 UTC [82] CONTEXT:  unnamed portal with parameters: $1 = 'a230f4b0-8e76-4540-8698-349b60c1f47b', $2 = 'RUNNING', $3 = '2024-11-25 00:14:30.432075+00', $4 = 'Running', $5 = '', $6 = '{"deferred": false, "cache_key": null, "pause_key": null, "retriable": null, "flow_run_id": "e75cf6e7-3318-42cb-9f0a-a5fc32eee727", "task_run_id": "a230f4b0-8e76-4540-8698-349b60c1f47b", "pause_timeout": null, "refresh_cache": null, "transition_id": null, "scheduled_time": null, "cache_expiration": null, "pause_reschedule": false, "run_input_keyset": null, "child_flow_run_id": null, "task_parameters_id": null, "untrackable_result": false}', $7 = 'null', $8 = '3a8232be-baaa-4e83-9c6f-3a1b75e436bc', $9 = '2024-11-25 00:14:31.472665+00', $10 = '2024-11-25 00:14:31.47371+00'
2024-11-25 00:14:31.474 UTC [82] LOCATION:  _bt_check_unique, nbtinsert.c:666
2024-11-25 00:14:31.474 UTC [82] STATEMENT:  INSERT INTO task_run_state (task_run_id, type, timestamp, name, message, state_details, data, id, created, updated) VALUES ($1::UUID, $2::state_type, $3::TIMESTAMP WITH TIME ZONE, $4::VARCHAR, $5::VARCHAR, $6, $7::JSON, $8::UUID, $9::TIMESTAMP WITH TIME ZONE, $10::TIMESTAMP WITH TIME ZONE) ON CONFLICT (id) DO NOTHING

```

State table list for the task run:
```sql
SELECT * FROM task_run_state WHERE task_run_id = 'a230f4b0-8e76-4540-8698-349b60c1f47b';
                  id                  |            created            |            updated            |   type    |           timestamp           |   name    | message |                                                                                                                                                                                                                       state_details                                                                                                                                                                                                                        |                                                                                                                 data                                                                                                                 |             task_run_id              | result_artifact_id 
--------------------------------------+-------------------------------+-------------------------------+-----------+-------------------------------+-----------+---------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------+--------------------
 4436d06c-f6a2-486f-87b7-b13e399ba3e3 | 2024-11-25 00:14:31.432305+00 | 2024-11-25 00:14:31.433352+00 | PENDING   | 2024-11-25 00:14:30.432075+00 | Pending   |         | {"deferred": false, "cache_key": null, "pause_key": null, "retriable": null, "flow_run_id": "e75cf6e7-3318-42cb-9f0a-a5fc32eee727", "task_run_id": "a230f4b0-8e76-4540-8698-349b60c1f47b", "pause_timeout": null, "refresh_cache": null, "transition_id": null, "scheduled_time": null, "cache_expiration": null, "pause_reschedule": false, "run_input_keyset": null, "child_flow_run_id": null, "task_parameters_id": null, "untrackable_result": false} | null                                                                                                                                                                                                                                 | a230f4b0-8e76-4540-8698-349b60c1f47b | 
 99bbc758-4685-4774-9e54-d5e126ec5188 | 2024-11-25 00:14:31.601826+00 | 2024-11-25 00:14:31.602852+00 | COMPLETED | 2024-11-25 00:14:30.463352+00 | Completed |         | {"deferred": false, "cache_key": null, "pause_key": null, "retriable": null, "flow_run_id": "e75cf6e7-3318-42cb-9f0a-a5fc32eee727", "task_run_id": "a230f4b0-8e76-4540-8698-349b60c1f47b", "pause_timeout": null, "refresh_cache": null, "transition_id": null, "scheduled_time": null, "cache_expiration": null, "pause_reschedule": false, "run_input_keyset": null, "child_flow_run_id": null, "task_parameters_id": null, "untrackable_result": true}  | {"expiration": null, "serializer": {"type": "pickle"}, "storage_key": "*****", "prefect_version": "3.1.4", "storage_block_id": null} | a230f4b0-8e76-4540-8698-349b60c1f47b | 
(2 rows)

```

### Version info

```Text
Version:             3.1.4
API version:         0.8.4
Python version:      3.11.7
Git commit:          78ee41cb
Built:               Wed, Nov 20, 2024 7:37 PM
OS/Arch:             win32/AMD64
Profile:             ephemeral
Server type:         server
Pydantic version:    2.10.1
Integrations:
  prefect-dask:      0.3.2
  prefect-docker:    0.6.2
  prefect-gitlab:    0.3.1
```


### Additional context

PREFECT_DEBUG_MODE=True wont trigger the error as well as running the flow in debugger (due to the slower execution?).


Functions to edit:
src/prefect/task_engine.py:SyncTaskRunEngine.set_state
src/prefect/task_engine.py:AsyncTaskRunEngine.set_state

Patch:
diff --git a/src/prefect/task_engine.py b/src/prefect/task_engine.py
index 43238d1cbc6c..5accda613f34 100644
--- a/src/prefect/task_engine.py
+++ b/src/prefect/task_engine.py
@@ -427,6 +427,11 @@ def set_state(self, state: State, force: bool = False) -> State:
 
         self.task_run.state = new_state = state
 
+        if last_state.timestamp == new_state.timestamp:
+            # Ensure that the state timestamp is unique, or at least not equal to the last state.
+            # This might occur especially on Windows where the timestamp resolution is limited.
+            new_state.timestamp += pendulum.duration(microseconds=1)
+
         # Ensure that the state_details are populated with the current run IDs
         new_state.state_details.task_run_id = self.task_run.id
         new_state.state_details.flow_run_id = self.task_run.flow_run_id
@@ -942,6 +947,11 @@ async def set_state(self, state: State, force: bool = False) -> State:
 
         self.task_run.state = new_state = state
 
+        if last_state.timestamp == new_state.timestamp:
+            # Ensure that the state timestamp is unique, or at least not equal to the last state.
+            # This might occur especially on Windows where the timestamp resolution is limited.
+            new_state.timestamp += pendulum.duration(microseconds=1)
+
         # Ensure that the state_details are populated with the current run IDs
         new_state.state_details.task_run_id = self.task_run.id
         new_state.state_details.flow_run_id = self.task_run.flow_run_id

Test patch: diff --git a/tests/test_task_engine.py b/tests/test_task_engine.py
index c6702f6a11de..185045fb9455 100644
--- a/tests/test_task_engine.py
+++ b/tests/test_task_engine.py
@@ -7,7 +7,7 @@
 from pathlib import Path
 from typing import List, Optional
 from unittest import mock
-from unittest.mock import AsyncMock, MagicMock, call
+from unittest.mock import AsyncMock, MagicMock, call, patch
 from uuid import UUID, uuid4
 
 import anyio
@@ -38,7 +38,7 @@
     PREFECT_TASK_DEFAULT_RETRIES,
     temporary_settings,
 )
-from prefect.states import Running, State
+from prefect.states import Completed, Running, State
 from prefect.task_engine import (
     AsyncTaskRunEngine,
     SyncTaskRunEngine,
@@ -78,6 +78,32 @@ async def test_client_attr_returns_client_after_starting(self):
         with pytest.raises(RuntimeError, match="not started"):
             engine.client
 
+    async def test_set_task_run_state(self):
+        engine = SyncTaskRunEngine(task=foo)
+        with engine.initialize_run():
+            running_state = Running()
+
+            new_state = engine.set_state(running_state)
+            assert new_state == running_state
+
+            completed_state = Completed()
+            new_state = engine.set_state(completed_state)
+            assert new_state == completed_state
+
+    async def test_set_task_run_state_duplicated_timestamp(self):
+        engine = SyncTaskRunEngine(task=foo)
+        with engine.initialize_run():
+            running_state = Running()
+            completed_state = Completed()
+            completed_state.timestamp = running_state.timestamp
+
+            new_state = engine.set_state(running_state)
+            assert new_state == running_state
+
+            new_state = engine.set_state(completed_state)
+            assert new_state == completed_state
+            assert new_state.timestamp > running_state.timestamp
+
 
 class TestAsyncTaskRunEngine:
     async def test_basic_init(self):
@@ -100,6 +126,32 @@ async def test_client_attr_returns_client_after_starting(self):
         with pytest.raises(RuntimeError, match="not started"):
             engine.client
 
+    async def test_set_task_run_state(self):
+        engine = AsyncTaskRunEngine(task=foo)
+        async with engine.initialize_run():
+            running_state = Running()
+
+            new_state = await engine.set_state(running_state)
+            assert new_state == running_state
+
+            completed_state = Completed()
+            new_state = await engine.set_state(completed_state)
+            assert new_state == completed_state
+
+    async def test_set_task_run_state_duplicated_timestamp(self):
+        engine = AsyncTaskRunEngine(task=foo)
+        async with engine.initialize_run():
+            running_state = Running()
+            completed_state = Completed()
+            completed_state.timestamp = running_state.timestamp
+
+            new_state = await engine.set_state(running_state)
+            assert new_state == running_state
+
+            new_state = await engine.set_state(completed_state)
+            assert new_state == completed_state
+            assert new_state.timestamp > running_state.timestamp
+
 
 class TestRunTask:
     def test_run_task_with_client_provided_uuid(
@@ -519,6 +571,56 @@ async def second():
         assert one == 42
         assert two == 42
 
+    async def test_task_run_states(
+        self,
+        prefect_client,
+        events_pipeline,
+    ):
+        @task
+        async def foo():
+            return TaskRunContext.get().task_run.id
+
+        task_run_id = await run_task_async(foo)
+        await events_pipeline.process_events()
+        states = await prefect_client.read_task_run_states(task_run_id)
+
+        state_names = [state.name for state in states]
+        assert state_names == [
+            "Pending",
+            "Running",
+            "Completed",
+        ]
+
+    async def test_task_run_states_with_equal_timestamps(
+        self,
+        prefect_client,
+        events_pipeline,
+    ):
+        @task
+        async def foo():
+            return TaskRunContext.get().task_run.id
+
+        original_set_state = AsyncTaskRunEngine.set_state
+
+        async def alter_new_state_timestamp(engine, state, force=False):
+            """Give the new state the same timestamp as the current state."""
+            state.timestamp = engine.task_run.state.timestamp
+            return await original_set_state(engine, state, force=force)
+
+        with patch.object(AsyncTaskRunEngine, "set_state", alter_new_state_timestamp):
+            task_run_id = await run_task_async(foo)
+
+            await events_pipeline.process_events()
+            states = await prefect_client.read_task_run_states(task_run_id)
+
+            state_names = [state.name for state in states]
+            assert state_names == [
+                "Pending",
+                "Running",
+                "Completed",
+            ]
+            assert states[0].timestamp < states[1].timestamp < states[2].timestamp
+
 
 class TestTaskRunsSync:
     def test_basic(self):
@@ -729,6 +831,56 @@ def second():
         assert one == 42
         assert two == 42
 
+    async def test_task_run_states(
+        self,
+        prefect_client,
+        events_pipeline,
+    ):
+        @task
+        def foo():
+            return TaskRunContext.get().task_run.id
+
+        task_run_id = run_task_sync(foo)
+        await events_pipeline.process_events()
+        states = await prefect_client.read_task_run_states(task_run_id)
+
+        state_names = [state.name for state in states]
+        assert state_names == [
+            "Pending",
+            "Running",
+            "Completed",
+        ]
+
+    async def test_task_run_states_with_equal_timestamps(
+        self,
+        prefect_client,
+        events_pipeline,
+    ):
+        @task
+        def foo():
+            return TaskRunContext.get().task_run.id
+
+        original_set_state = SyncTaskRunEngine.set_state
+
+        def alter_new_state_timestamp(engine, state, force=False):
+            """Give the new state the same timestamp as the current state."""
+            state.timestamp = engine.task_run.state.timestamp
+            return original_set_state(engine, state, force=force)
+
+        with patch.object(SyncTaskRunEngine, "set_state", alter_new_state_timestamp):
+            task_run_id = run_task_sync(foo)
+
+            await events_pipeline.process_events()
+            states = await prefect_client.read_task_run_states(task_run_id)
+
+            state_names = [state.name for state in states]
+            assert state_names == [
+                "Pending",
+                "Running",
+                "Completed",
+            ]
+            assert states[0].timestamp < states[1].timestamp < states[2].timestamp
+
 
 class TestReturnState:
     async def test_return_state(self, prefect_client):