Repository: PrefectHQ/prefect
Pull request: Avoid deleting runs where execution has been attempted or are past their scheduled start date
This PR updates the patch deployment API to avoid deleting runs in `AwaitingConcurrencySlot` or have passed their scheduled start time. The latter condition means that `Late` runs would not be deleted by updating a deployment. 

Closes https://github.com/PrefectHQ/prefect/issues/17764

Problem statement:
late flow runs get deleted on new prefect deploy
### Bug summary

flow runs that are late or awaitingconcurrencyslot in a given deployment are deleted and not rescheduled when a fresh `prefect deploy` is issued for the deployment.

steps to reproduce:
1. deploy a simple flow like below to a process workpool
2. make sure schedule is active + flow runs are being kicked off
3. kill the process worker temporarily, let a few of the scheduled flow runs stack up as late
4. bring the process worker back up + run `prefect deploy`
5. all the late runs seem to have disappeared - and are replaced with the new deployment's runs, if applicable.

flow code:
```
import time
from prefect import flow, task


@task(log_prints=True)
def generic_task(name: str, sleep_time: int):
    print(f"Task {name} Started")
    print(f"sleeping for {sleep_time} seconds")
    time.sleep(sleep_time)
    print(f"Task {name} Completed")

@flow(log_prints=True)
def generic_flow():
    print("Starting Flow")
    a = generic_task.submit("a",25)
    print(a.result())
```

prefect.yaml contents:
```
deployments:
- name: default
  parameters: {}
  parameter_openapi_schema: {}
  entrypoint: simple_flow.py:generic_flow
  schedule:
    interval: 10 # seconds
    slug: "default-schedule"
    timezone: "America/New_York"
    active: true
```




### Version info

```Text
Version:             3.3.3
API version:         0.8.4
Python version:      3.12.9
Git commit:          4100d4ea
Built:               Sat, Apr 05, 2025 01:44 AM
OS/Arch:             darwin/arm64
Profile:             sbx-rowdy-cloud
Server type:         cloud
Pydantic version:    2.11.2
Integrations:
  prefect-databricks: 0.3.0
```

### Additional context

alternatively, set a `concurrency_limit` of 1 on the deployment and the task sleep to an amount of time greater than the interval on the schedule. let a few runs stack up, re-run `prefect deploy` and all the awaitingconcurrencyslot flow runs disappear.

Hints from issue discussion:
When updating a deployment, the server will delete all flow runs in a `SCHEDULED` state for the deployment. Since `AwaitingConcurrencySlot` is a `SCHEDULED` state type, those runs get deleted. I suspect the same thing would happen with `AwaitingRetry` as well.

We could either exclude `SCHEDULED` states named `AwaitingConcurrencySlot` or `AwaitingRetry` from the flow run deletion or create a new state type. Updating the flow run deletion is the safer change, so I'll go that route now to give us some time to think about a new state type.
We're seeing similar behavior where during deployment if a flow run is LATE that flow run is being deleted, and that schedule is essentially skipped.
Looks like I focused in too closely on only one part of the issue! I've updated https://github.com/PrefectHQ/prefect/pull/17769 to account for runs past their scheduled start date too.

Commit message: Fix failing test

Patch to review:
diff --git a/src/prefect/server/api/deployments.py b/src/prefect/server/api/deployments.py
index 9fd876d53e8f..c883e0a08d5c 100644
--- a/src/prefect/server/api/deployments.py
+++ b/src/prefect/server/api/deployments.py
@@ -911,6 +911,7 @@ async def update_deployment_schedule(
             session=session,
             deployment_id=deployment_id,
             auto_scheduled_only=True,
+            future_only=True,
         )
 
 
diff --git a/src/prefect/server/models/deployments.py b/src/prefect/server/models/deployments.py
index 44dfae037c51..16350bcdbc0b 100644
--- a/src/prefect/server/models/deployments.py
+++ b/src/prefect/server/models/deployments.py
@@ -43,6 +43,7 @@ async def _delete_scheduled_runs(
     session: AsyncSession,
     deployment_id: UUID,
     auto_scheduled_only: bool = False,
+    future_only: bool = False,
 ) -> None:
     """
     This utility function deletes all of a deployment's runs that are in a Scheduled state
@@ -52,10 +53,13 @@ async def _delete_scheduled_runs(
     Args:
         deployment_id: the deployment for which we should delete runs.
         auto_scheduled_only: if True, only delete auto scheduled runs. Defaults to `False`.
+        future_only: if True, only delete runs that are scheduled to run in the future.
+            Defaults to `False`.
     """
     delete_query = sa.delete(db.FlowRun).where(
         db.FlowRun.deployment_id == deployment_id,
         db.FlowRun.state_type == schemas.states.StateType.SCHEDULED.value,
+        db.FlowRun.state_name != schemas.states.AwaitingConcurrencySlot().name,
         db.FlowRun.run_count == 0,
     )
 
@@ -64,6 +68,11 @@ async def _delete_scheduled_runs(
             db.FlowRun.auto_scheduled.is_(True),
         )
 
+    if future_only:
+        delete_query = delete_query.where(
+            db.FlowRun.next_scheduled_start_time > now("UTC"),
+        )
+
     await session.execute(delete_query)
 
 
@@ -147,7 +156,10 @@ async def create_deployment(
     # schedules and any runs from the old deployment.
 
     await _delete_scheduled_runs(
-        session=session, deployment_id=deployment_id, auto_scheduled_only=True
+        session=session,
+        deployment_id=deployment_id,
+        auto_scheduled_only=True,
+        future_only=True,
     )
 
     await delete_schedules_for_deployment(session=session, deployment_id=deployment_id)
@@ -263,7 +275,10 @@ async def update_deployment(
 
     # delete any auto scheduled runs that would have reflected the old deployment config
     await _delete_scheduled_runs(
-        session=session, deployment_id=deployment_id, auto_scheduled_only=True
+        session=session,
+        deployment_id=deployment_id,
+        auto_scheduled_only=True,
+        future_only=True,
     )
 
     if should_update_schedules:
diff --git a/src/prefect/server/schemas/states.py b/src/prefect/server/schemas/states.py
index 610ae841c076..a1633c8f8b1d 100644
--- a/src/prefect/server/schemas/states.py
+++ b/src/prefect/server/schemas/states.py
@@ -468,6 +468,24 @@ def AwaitingRetry(
     )
 
 
+def AwaitingConcurrencySlot(
+    cls: type[_State] = State,
+    scheduled_time: Optional[DateTime] = None,
+    **kwargs: Any,
+) -> _State:
+    """Convenience function for creating `AwaitingConcurrencySlot` states.
+
+    Returns:
+        State: a AwaitingConcurrencySlot state
+    """
+    return Scheduled(
+        cls=cls,
+        scheduled_time=scheduled_time,
+        name="AwaitingConcurrencySlot",
+        **kwargs,
+    )
+
+
 def Retrying(cls: type[_State] = State, **kwargs: Any) -> _State:
     """Convenience function for creating `Retrying` states.
 
diff --git a/tests/server/models/test_deployments.py b/tests/server/models/test_deployments.py
index c27b3e7fb289..e39746c5a978 100644
--- a/tests/server/models/test_deployments.py
+++ b/tests/server/models/test_deployments.py
@@ -1347,7 +1347,9 @@ async def test_update_deployment_deletes_autoscheduled_flow_runs_in_scheduled(
             flow_run=schemas.core.FlowRun(
                 flow_id=flow.id,
                 deployment_id=deployment.id,
-                state=states.Scheduled(),
+                state=states.Scheduled(
+                    scheduled_time=now("UTC") + datetime.timedelta(minutes=1)
+                ),
                 auto_scheduled=True,
             ),
         )
@@ -1393,6 +1395,64 @@ async def test_update_deployment_does_not_delete_autoscheduled_flow_reruns_in_sc
 
         assert run is not None
 
+    async def test_update_deployment_does_not_delete_runs_waiting_for_a_concurrency_slot(
+        self,
+        session: AsyncSession,
+        deployment: orm_models.Deployment,
+        flow: orm_models.Flow,
+    ):
+        flow_run = await models.flow_runs.create_flow_run(
+            session=session,
+            flow_run=schemas.core.FlowRun(
+                flow_id=flow.id,
+                deployment_id=deployment.id,
+                state=states.AwaitingConcurrencySlot(),
+                auto_scheduled=True,
+            ),
+        )
+
+        await models.deployments.update_deployment(
+            session=session,
+            deployment_id=deployment.id,
+            deployment=schemas.actions.DeploymentUpdate(),
+        )
+
+        run = await models.flow_runs.read_flow_run(
+            session=session, flow_run_id=flow_run.id
+        )
+
+        assert run is not None
+
+    async def test_update_deployment_does_not_delete_runs_past_their_scheduled_start_time(
+        self,
+        session,
+        deployment,
+        flow,
+    ):
+        flow_run = await models.flow_runs.create_flow_run(
+            session=session,
+            flow_run=schemas.core.FlowRun(
+                flow_id=flow.id,
+                deployment_id=deployment.id,
+                state=states.Scheduled(
+                    scheduled_time=now("UTC") - datetime.timedelta(minutes=1)
+                ),
+                auto_scheduled=True,
+            ),
+        )
+
+        await models.deployments.update_deployment(
+            session=session,
+            deployment_id=deployment.id,
+            deployment=schemas.actions.DeploymentUpdate(),
+        )
+
+        run = await models.flow_runs.read_flow_run(
+            session=session, flow_run_id=flow_run.id
+        )
+
+        assert run is not None
+
 
 @pytest.fixture
 async def deployment_schedules(
diff --git a/tests/server/orchestration/api/test_deployment_schedules.py b/tests/server/orchestration/api/test_deployment_schedules.py
index fe42a5cc88f2..d573df713902 100644
--- a/tests/server/orchestration/api/test_deployment_schedules.py
+++ b/tests/server/orchestration/api/test_deployment_schedules.py
@@ -76,7 +76,9 @@ async def scheduled_flow_runs(
                     state=schemas.states.State(
                         type=schemas.states.StateType.SCHEDULED,
                         timestamp=now("UTC"),
-                        state_details={"scheduled_time": now("UTC")},
+                        state_details={
+                            "scheduled_time": now("UTC") + timedelta(minutes=1)
+                        },
                     ),
                 ),
             )


Code review comments:
diff_hunk: @@ -468,6 +468,24 @@ def AwaitingRetry(
     )
 
 
+def AwaitingConcurrencySlot(
+    cls: type[_State] = State,
+    scheduled_time: Optional[DateTime] = None,
+    **kwargs: Any,
+) -> _State:
+    """Convenience function for creating `AwaitingConcurrencySlot` states.
+
+    Returns:
+        State: a AwaitingConcurrencySlot state
path: src/prefect/server/schemas/states.py
text: @user1:
```suggestion
        State: an AwaitingConcurrencySlot state
```