fix(api): delegate concurrent deletion deadlocks to TaskBackoffError

api/environments/tasks.py

@@ -1,8 +1,11 @@
+from django.db import IntegrityError, OperationalError, transaction
 from django.db.models import Prefetch, Q
+from django.db.transaction import TransactionManagementError
 from django.utils import timezone
 from task_processor.decorators import (
     register_task_handler,
 )
+from task_processor.exceptions import TaskBackoffError
 from task_processor.models import TaskPriority
 
 from audit.models import AuditLog
@@ -59,7 +62,16 @@ def delete_environment_from_dynamo(api_key: str, environment_id: str):  # type:
 
 @register_task_handler()
 def delete_environment(environment_id: int) -> None:
-    Environment.objects.get(id=environment_id).delete()
+    try:
+        # Wrap in atomic to safely catch DB-level errors
+        with transaction.atomic():
+            Environment.objects.get(id=environment_id).delete()
+    except Environment.DoesNotExist:
+        # If it's already gone, our job is done!
+        pass
+    except (OperationalError, IntegrityError, TransactionManagementError):
+        # Someone else is locking these rows, back off and retry
+        raise TaskBackoffError()
 
 
 @register_task_handler()

api/features/tasks.py

@@ -1,9 +1,12 @@
 import logging
 from typing import Any
 
+from django.db import IntegrityError, OperationalError, transaction
+from django.db.transaction import TransactionManagementError
 from task_processor.decorators import (
     register_task_handler,
 )
+from task_processor.exceptions import TaskBackoffError
 
 from environments.models import Webhook
 from features.models import Feature, FeatureState
@@ -159,4 +162,12 @@ def _get_previous_multivariate_values(
 
 @register_task_handler()
 def delete_feature(feature_id: int) -> None:
-    Feature.objects.get(pk=feature_id).delete()
+    try:
+        with transaction.atomic():
+            Feature.objects.get(pk=feature_id).delete()
+
+    except Feature.DoesNotExist:
+        pass
+
+    except (OperationalError, IntegrityError, TransactionManagementError):
+        raise TaskBackoffError()

api/tests/unit/environments/test_unit_environments_tasks.py

@@ -1,8 +1,12 @@
+import pytest
+from django.db import OperationalError
 from pytest_mock import MockerFixture
+from task_processor.exceptions import TaskBackoffError
 
 from audit.models import AuditLog
 from environments.models import Environment
 from environments.tasks import (
+    delete_environment,
     delete_environment_from_dynamo,
     process_environment_update,
     rebuild_environment_document,
@@ -113,3 +117,44 @@ def test_delete_environment__calls_internal_methods_correctly(
     mocked_identity_wrapper.delete_all_identities.assert_called_once_with(
         environment_api_key
     )
+
+
+def test_delete_environment__environment_does_not_exist__succeeds_silently(
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mock_get_environment = mocker.patch("environments.tasks.Environment.objects.get")
+    mock_get_environment.side_effect = Environment.DoesNotExist
+
+    # When
+    delete_environment(environment_id=1)
+
+    # Then
+    # No exception is raised, confirming silent success
+    mock_get_environment.assert_called_once_with(id=1)
+
+
+def test_delete_environment__database_deadlock__raises_task_backoff_error(
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mock_get_environment = mocker.patch("environments.tasks.Environment.objects.get")
+    mock_get_environment.side_effect = OperationalError
+
+    # When
+    with pytest.raises(TaskBackoffError):
+        delete_environment(environment_id=1)
+
+    # Then
+    # TaskBackoffError is raised to trigger the task-processor retry
+    mock_get_environment.assert_called_once_with(id=1)  # Given
+    mock_get_environment = mocker.patch("environments.tasks.Environment.objects.get")
+    mock_get_environment.side_effect = OperationalError
+
+    # When
+    with pytest.raises(TaskBackoffError):
+        delete_environment(environment_id=1)
+
+    # Then
+    # TaskBackoffError is raised to trigger the task-processor retry
+    mock_get_environment.assert_called_once_with(id=1)

api/tests/unit/features/test_unit_features_tasks.py

@@ -1,11 +1,13 @@
 import pytest
+from django.db import OperationalError
 from pytest_lazyfixture import lazy_fixture  # type: ignore[import-untyped]
 from pytest_mock import MockerFixture
+from task_processor.exceptions import TaskBackoffError
 
 from api_keys.models import MasterAPIKey
 from environments.models import Environment
 from features.models import Feature, FeatureState
-from features.tasks import trigger_feature_state_change_webhooks
+from features.tasks import delete_feature, trigger_feature_state_change_webhooks
 from organisations.models import Organisation
 from projects.models import Project
 from users.models import FFAdminUser
@@ -162,3 +164,34 @@ def test_trigger_feature_state_change_webhooks_for_deleted_flag_uses_fs_instance
 
     assert data["previous_state"]["feature"]["id"] == feature_state.feature.id
     assert event_type == WebhookEventType.FLAG_DELETED.value
+
+
+@pytest.mark.django_db
+def test_delete_feature__feature_does_not_exist__succeeds_silently(
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mock_get_feature = mocker.patch("features.tasks.Feature.objects.get")
+    mock_get_feature.side_effect = Feature.DoesNotExist
+
+    # When
+    delete_feature(feature_id=1)
+
+    # Then
+    mock_get_feature.assert_called_once_with(pk=1)
+
+
+@pytest.mark.django_db
+def test_delete_feature__database_deadlock__raises_task_backoff_error(
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mock_get_feature = mocker.patch("features.tasks.Feature.objects.get")
+    mock_get_feature.side_effect = OperationalError
+
+    # When
+    with pytest.raises(TaskBackoffError):
+        delete_feature(feature_id=1)
+
+    # Then
+    mock_get_feature.assert_called_once_with(pk=1)