Fix automation deletion deadlock by enforcing consistent lock ordering (#19369)

Author: zzstoatzz

src/prefect/server/events/models/automations.py

@@ -233,6 +233,25 @@ async def delete_automation(
     if not automation:
         return False
 
+    # Delete child tables in a consistent order to prevent deadlocks
+    # when multiple automations are deleted concurrently
+    await session.execute(
+        sa.delete(db.AutomationBucket).where(
+            db.AutomationBucket.automation_id == automation_id,
+        )
+    )
+    await session.execute(
+        sa.delete(db.AutomationRelatedResource).where(
+            db.AutomationRelatedResource.automation_id == automation_id,
+        )
+    )
+    await session.execute(
+        sa.delete(db.CompositeTriggerChildFiring).where(
+            db.CompositeTriggerChildFiring.automation_id == automation_id,
+        )
+    )
+
+    # Now delete the parent automation
     await session.execute(
         sa.delete(db.Automation).where(
             db.Automation.id == automation_id,
@@ -252,6 +271,14 @@ async def delete_automations_for_workspace(
     automations = await read_automations_for_workspace(
         session,
     )
+
+    # Delete child tables in a consistent order to prevent deadlocks
+    # when multiple workspace deletions occur concurrently
+    await session.execute(sa.delete(db.AutomationBucket))
+    await session.execute(sa.delete(db.AutomationRelatedResource))
+    await session.execute(sa.delete(db.CompositeTriggerChildFiring))
+
+    # Now delete all automations
     result = await session.execute(sa.delete(db.Automation))
     for automation in automations:
         await _notify(session, automation, "deleted")

tests/test_automations.py

@@ -237,3 +237,31 @@ async def test_find_automation(automation: Automation):
     # Test finding nonexistent name returns None
     found = await client.find_automation("nonexistent_name")
     assert found is None
+
+
+async def test_concurrent_automation_deletion():
+    """Test that concurrent automation deletions don't deadlock."""
+    import asyncio
+
+    automations_to_create = [
+        Automation(
+            name=f"concurrent-test-{i}",
+            trigger=EventTrigger(
+                expect={"prefect.flow-run.Completed"},
+                posture=Posture.Reactive,
+                threshold=1,
+            ),
+            actions=[DoNothing()],
+        )
+        for i in range(10)
+    ]
+
+    created_automations = await asyncio.gather(
+        *[automation.create() for automation in automations_to_create]
+    )
+
+    await asyncio.gather(*[automation.adelete() for automation in created_automations])
+
+    for automation in created_automations:
+        with pytest.raises(ValueError):
+            await Automation.read(id=automation.id)