Better archiving experience

changes/pr185.yaml

@@ -0,0 +1,2 @@
+enhancement:
+  - "Cancel ad-hoc created flow runs instead of deleting them when archiving a flow - [#185](https://github.com/PrefectHQ/server/pull/185)"

src/prefect_server/api/flows.py

@@ -1,3 +1,4 @@
+import asyncio
 import datetime
 import hashlib
 import json
@@ -332,7 +333,8 @@ async def archive_flow(flow_id: str) -> bool:
     Archives a flow.
 
     Archiving a flow prevents it from scheduling new runs. It also:
-        - deletes any currently scheduled runs
+        - deletes any currently auto-scheduled runs
+        - cancels any ad-hoc created scheduled runs
         - resets the "last scheduled run time" of any schedules
 
     Args:
@@ -355,9 +357,28 @@ async def archive_flow(flow_id: str) -> bool:
 
     # delete scheduled flow runs
     await models.FlowRun.where(
-        {"flow_id": {"_eq": flow_id}, "state": {"_eq": "Scheduled"}}
+        {
+            "flow_id": {"_eq": flow_id},
+            "state": {"_eq": "Scheduled"},
+            "auto_scheduled": {"_eq": True},
+        }
     ).delete()
 
+    # cancel adhoc scheduled flow runs
+    fr_ids = await models.FlowRun.where(
+        {
+            "flow_id": {"_eq": flow_id},
+            "state": {"_eq": "Scheduled"},
+            "auto_scheduled": {"_eq": False},
+        }
+    ).get({"id"})
+    msg = f"Flow {flow_id} was archived."
+    await asyncio.gather(
+        *[
+            api.states.cancel_flow_run(flow_run.id, state_message=msg)
+            for flow_run in fr_ids
+        ]
+    )
     return True
 
 

src/prefect_server/api/states.py

@@ -207,7 +207,9 @@ async def set_task_run_state(
 
 
 @register_api("states.cancel_flow_run")
-async def cancel_flow_run(flow_run_id: str) -> models.FlowRun:
+async def cancel_flow_run(
+    flow_run_id: str, state_message: str = None
+) -> models.FlowRun:
     """
     Cancel a flow run.
 
@@ -231,7 +233,7 @@ async def cancel_flow_run(flow_run_id: str) -> models.FlowRun:
         return flow_run
     else:
         if state.is_running():
-            state = Cancelling("Flow run is cancelling")
+            state = Cancelling(state_message or "Flow run is cancelling")
         else:
-            state = Cancelled("Flow run is cancelled")
+            state = Cancelled(state_message or "Flow run is cancelled")
         return await set_flow_run_state(flow_run_id=flow_run_id, state=state)

tests/api/test_flows.py

@@ -746,7 +746,7 @@ async def test_archive_flow_twice(self, flow_id):
         flow = await models.Flow.where(id=flow_id).first({"archived"})
         assert flow.archived
 
-    async def test_archive_flow_deletes_scheduled_runs(self, flow_id):
+    async def test_archive_flow_deletes_auto_scheduled_runs(self, flow_id):
         # create scheduled api.runs since the fixture doesn't
 
         await api.flows.schedule_flow_runs(flow_id=flow_id)
@@ -765,6 +765,35 @@ async def test_archive_flow_deletes_scheduled_runs(self, flow_id):
             == 0
         )
 
+    async def test_archive_flow_cancels_ad_hoc_runs(self, flow_id):
+        # create one ad hoc run and 10 auto scheduled runs
+        flow_run_id = await api.runs.create_flow_run(flow_id)
+        await api.flows.schedule_flow_runs(flow_id=flow_id)
+
+        scheduled_runs = await models.FlowRun.where(
+            {
+                "flow_id": {"_eq": flow_id},
+                "state": {"_eq": "Scheduled"},
+                "auto_scheduled": {"_eq": True},
+            }
+        ).get({"id"})
+        assert scheduled_runs
+
+        await api.flows.archive_flow(flow_id)
+
+        # all auto scheduled deleted
+        assert (
+            await models.FlowRun.where(
+                {"id": {"_in": [r.id for r in scheduled_runs]}}
+            ).count()
+            == 0
+        )
+        ad_hoc_state = await models.FlowRun.where(id=flow_run_id).first(
+            {"state", "state_message"}
+        )
+        assert ad_hoc_state.state == "Cancelled"
+        assert "archived" in ad_hoc_state.state_message
+
     async def test_archive_flow_with_bad_id(self, flow_id):
         assert not await api.flows.archive_flow(str(uuid.uuid4()))