Clear old scheduled runs when updating a flow group schedule

Changelog.md

@@ -16,7 +16,7 @@
 
 ### Fixes
 
-- None
+- Fixes bug where updating a Flow Group schedule didn't clear old scheduled runs - [#9](https://github.com/PrefectHQ/server/pull/9)
 
 ### Breaking Changes
 

src/prefect_server/api/flow_groups.py

@@ -11,8 +11,7 @@
 async def set_flow_group_default_parameters(
     flow_group_id: str, parameters: dict
 ) -> bool:
-    """
-    Sets default value(s) for parameter(s) on a flow group
+    """ Sets default value(s) for parameter(s) on a flow group
 
     Args:
         - flow_group_id (str): the ID of the flow group to update
@@ -58,6 +57,14 @@ async def set_flow_group_schedule(flow_group_id: str, clocks: List[dict]) -> boo
     result = await models.FlowGroup.where(id=flow_group_id).update(
         set=dict(schedule=dict(type="Schedule", clocks=clocks))
     )
+
+    deleted_runs = await models.FlowRun.where(
+        {
+            "flow": {"flow_group_id": {"_eq": flow_group_id}},
+            "state": {"_eq": "Scheduled"},
+            "auto_scheduled": {"_eq": True},
+        }
+    ).delete()
     return bool(result.affected_rows)
 
 
@@ -80,6 +87,15 @@ async def delete_flow_group_schedule(flow_group_id: str) -> bool:
     result = await models.FlowGroup.where(id=flow_group_id).update(
         set=dict(schedule=None)
     )
+
+    deleted_runs = await models.FlowRun.where(
+        {
+            "flow": {"flow_group_id": {"_eq": flow_group_id}},
+            "state": {"_eq": "Scheduled"},
+            "auto_scheduled": {"_eq": True},
+        }
+    ).delete()
+
     return bool(result.affected_rows)
 
 

tests/api/test_flow_groups.py

@@ -111,6 +111,46 @@ async def test_set_flow_group_schedule(self, flow_group_id, clock):
         flow_group = await models.FlowGroup.where(id=flow_group_id).first({"schedule"})
         assert flow_group.schedule["clocks"][0] == clock
 
+    async def test_setting_schedule_deletes_runs(self, flow_id, flow_group_id):
+        """
+        This test takes a flow with a schedule and ensures that updating the Flow Group
+        schedule deletes previously auto-scheduled runs.
+        """
+        flow_group = await models.FlowGroup.where(id=flow_group_id).first({"schedule"})
+        assert flow_group.schedule is None
+
+        # make sure we're starting from a clean slate
+        await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).delete()
+        assert await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).count() == 0
+        await api.flows.set_schedule_active(flow_id=flow_id)
+        assert await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).count() == 10
+        # create one manual run that won't be deleted
+        await api.runs.create_flow_run(flow_id=flow_id)
+
+        # 10 autoscheduled runs and 1 manual run
+        runs = await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).get(
+            {"scheduled_start_time"}
+        )
+        assert len(runs) == 11
+        start_times = {r.scheduled_start_time for r in runs}
+
+        # update the flow group schedule and confirm the runs have been repopulated
+        clock = {"type": "CronClock", "cron": "42 0 0 * * *"}
+
+        success = await api.flow_groups.set_flow_group_schedule(
+            flow_group_id=flow_group_id, clocks=[clock]
+        )
+        assert success is True
+
+        flow_group = await models.FlowGroup.where(id=flow_group_id).first({"schedule"})
+        assert flow_group.schedule["clocks"][0] == clock
+
+        # the only run that should still be scheduled from before is the manually created one
+        new_runs = await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).get(
+            {"scheduled_start_time"}
+        )
+        assert len({r.scheduled_start_time for r in new_runs} & start_times) == 1
+
     async def test_set_schedule_with_two_clocks(self, flow_group_id):
         flow_group = await models.FlowGroup.where(id=flow_group_id).first({"schedule"})
         assert flow_group.schedule is None
@@ -193,20 +233,30 @@ async def test_set_schedule_for_none_flow_group(self):
 
 
 class TestDeleteFlowGroupSchedule:
-    async def test_delete_flow_group_schedule(self, flow_group_id):
-        await models.FlowGroup.where(id=flow_group_id).update(
-            set=dict(schedule={"foo": "bar"})
+    async def test_delete_flow_group_schedule(self, flow_id, flow_group_id):
+
+        # update the flow group schedule and confirm the runs have been repopulated
+        clock = {"type": "CronClock", "cron": "42 0 0 * * *"}
+
+        success = await api.flow_groups.set_flow_group_schedule(
+            flow_group_id=flow_group_id, clocks=[clock]
         )
-        flow_group = await models.FlowGroup.where(id=flow_group_id).first({"schedule"})
-        assert flow_group.schedule == {"foo": "bar"}
+        assert success is True
+
+        await api.flows.set_schedule_active(flow_id=flow_id)
+        assert await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).count() == 10
 
         success = await api.flow_groups.delete_flow_group_schedule(
             flow_group_id=flow_group_id
         )
         assert success is True
+
         flow_group = await models.FlowGroup.where(id=flow_group_id).first({"schedule"})
         assert flow_group.schedule is None
 
+        # ensure old runs were deleted
+        assert await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).count() == 0
+
     async def test_delete_flow_group_schedule_for_invalid_flow_group(self):
         success = await api.flow_groups.delete_flow_group_schedule(
             flow_group_id=str(uuid.uuid4())

tests/api/test_flows.py

@@ -866,7 +866,9 @@ async def test_set_schedule_inactive_twice(self, flow_id):
         flow = await models.Flow.where(id=flow_id).first({"is_schedule_active"})
         assert flow.is_schedule_active is False
 
-    async def test_set_schedule_inactive_deletes_runs(self, flow_id):
+    async def test_set_schedule_inactive_deletes_only_auto_scheduled_runs(
+        self, flow_id
+    ):
         await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).delete()
         assert await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).count() == 0
 
@@ -916,7 +918,7 @@ async def test_set_schedule_inactive_deletes_runs(self, flow_id):
 
     async def test_set_schedule_inactive_deletes_runs_in_utc(self, project_id):
         """
-        Ensures that toggling schedules on and off properly creates new api.runs even if the 
+        Ensures that toggling schedules on and off properly creates new api.runs even if the
         schedule was in local time
         https://github.com/PrefectHQ/cloud/issues/2295
         """