fix(uptime): Defensive error handling in deletion cascade for billing seats

src/sentry/deletions/defaults/detector.py

@@ -1,6 +1,10 @@
+import logging
+
 from sentry.deletions.base import BaseRelation, ModelDeletionTask, ModelRelation
 from sentry.workflow_engine.models.detector import Detector
 
+logger = logging.getLogger(__name__)
+
 
 class DetectorDeletionTask(ModelDeletionTask[Detector]):
     manager_name = "objects_for_deletion"
@@ -28,3 +32,19 @@ def get_child_relations(self, instance: Detector) -> list[BaseRelation]:
             )
 
         return model_relations
+
+    def delete_instance(self, instance: Detector) -> None:
+        from sentry.uptime.subscriptions.subscriptions import remove_uptime_seat
+        from sentry.uptime.types import GROUP_TYPE_UPTIME_DOMAIN_CHECK_FAILURE
+
+        if instance.type == GROUP_TYPE_UPTIME_DOMAIN_CHECK_FAILURE:
+            try:
+                remove_uptime_seat(instance)
+            except Exception:
+                logger.warning(
+                    "detector.deletion.remove_uptime_seat_failed",
+                    extra={"detector_id": instance.id},
+                    exc_info=True,
+                )
+
+        super().delete_instance(instance)

src/sentry/deletions/defaults/uptime_subscription.py

@@ -1,24 +1,35 @@
+import logging
+
 from sentry.deletions.base import ModelDeletionTask
 from sentry.uptime.models import UptimeSubscription, get_detector
 from sentry.uptime.subscriptions.subscriptions import delete_uptime_subscription, remove_uptime_seat
+from sentry.workflow_engine.models.detector import Detector
+
+logger = logging.getLogger(__name__)
 
 
 class UptimeSubscriptionDeletionTask(ModelDeletionTask[UptimeSubscription]):
     def delete_instance(self, instance: UptimeSubscription) -> None:
-        detector = get_detector(instance)
-
         # XXX: Typically quota updates would be handled by the
         # delete_uptime_detector function exposed in the
         # uptime.subscriptions.subscriptions module. However if a Detector is
-        # deleted without using this, we need to still ensure the billing east
+        # deleted without using this, we need to still ensure the billing seat
         # is revoked. This should never happen.
         #
         # Since the delete_uptime_detector function is already scheduling the
         # detector for deletion, you may think we could remove the quota
         # remove_seat call there, since it will happen here. But this would
         # mean the customers quota is not freed up _immediately_ when the
         # detector is deleted using that method.
-        remove_uptime_seat(detector)
+        try:
+            detector = get_detector(instance)
+        except Detector.DoesNotExist:
+            logger.warning(
+                "uptime.subscription_deletion.detector_not_found",
+                extra={"uptime_subscription_id": instance.id},
+            )
+        else:
+            remove_uptime_seat(detector)
 
         # Ensure the remote subscription is removed if it wasn't already (again
         # it should have been as part of delete_uptime_detector)

tests/sentry/deletions/test_detector.py

@@ -1,3 +1,5 @@
+from unittest import mock
+
 import pytest
 
 from sentry.constants import ObjectStatus
@@ -135,3 +137,62 @@ def test_delete_uptime_detector(self) -> None:
             detector.refresh_from_db()
         with pytest.raises(UptimeSubscription.DoesNotExist):
             uptime_sub.refresh_from_db()
+
+    @mock.patch("sentry.quotas.backend.remove_seat")
+    def test_delete_uptime_detector_calls_remove_seat(
+        self, mock_remove_seat: mock.MagicMock
+    ) -> None:
+        """Verify remove_seat is called when an uptime detector is deleted."""
+        detector = self.create_uptime_detector()
+        self.ScheduledDeletion.schedule(instance=detector, days=0)
+
+        with self.tasks():
+            run_scheduled_deletions()
+
+        assert not Detector.objects.filter(id=detector.id).exists()
+        assert mock_remove_seat.call_count >= 1
+
+    @mock.patch("sentry.deletions.defaults.uptime_subscription.remove_uptime_seat")
+    @mock.patch("sentry.uptime.subscriptions.subscriptions.remove_uptime_seat")
+    def test_delete_uptime_detector_succeeds_when_remove_seat_fails(
+        self,
+        mock_remove_seat_subscriptions: mock.MagicMock,
+        mock_remove_seat_deletion: mock.MagicMock,
+    ) -> None:
+        """Detector deletion succeeds even if remove_uptime_seat raises in DetectorDeletionTask."""
+        # DetectorDeletionTask.delete_instance does a lazy import from
+        # sentry.uptime.subscriptions.subscriptions, so it picks up this mock.
+        mock_remove_seat_subscriptions.side_effect = Exception("seat error")
+        # UptimeSubscriptionDeletionTask uses a top-level import bound at module
+        # load time, so we mock at the import target separately (default no-op).
+        detector = self.create_uptime_detector()
+        detector_id = detector.id
+        self.ScheduledDeletion.schedule(instance=detector, days=0)
+
+        with self.tasks():
+            run_scheduled_deletions()
+
+        assert not Detector.objects.filter(id=detector_id).exists()
+        # Verify the error path in DetectorDeletionTask was actually exercised.
+        mock_remove_seat_subscriptions.assert_called_once()
+
+    def test_delete_uptime_subscription_without_detector(self) -> None:
+        """UptimeSubscription deletion proceeds when the detector no longer exists."""
+        detector = self.create_uptime_detector()
+        uptime_sub = get_uptime_subscription(detector)
+        uptime_sub_id = uptime_sub.id
+
+        # Delete the detector and its data sources directly so the
+        # UptimeSubscription is orphaned (no detector to find via get_detector).
+        DataSourceDetector.objects.filter(detector=detector).delete()
+        DataSource.objects.filter(
+            source_id=str(uptime_sub.id),
+        ).delete()
+        detector.delete()
+
+        self.ScheduledDeletion.schedule(instance=uptime_sub, days=0)
+
+        with self.tasks():
+            run_scheduled_deletions()
+
+        assert not UptimeSubscription.objects.filter(id=uptime_sub_id).exists()

tests/sentry/deletions/test_project.py

@@ -236,6 +236,7 @@ def test_delete_error_events(self) -> None:
     def test_delete_with_uptime_monitors(self, mock_remove_seat: mock.MagicMock) -> None:
         project = self.create_project(name="test")
         detector = self.create_uptime_detector(project=project)
+        detector_id = detector.id
         uptime_subscription = get_uptime_subscription(detector)
 
         self.ScheduledDeletion.schedule(instance=project, days=0)
@@ -244,9 +245,11 @@ def test_delete_with_uptime_monitors(self, mock_remove_seat: mock.MagicMock) ->
             run_scheduled_deletions()
 
         assert not Project.objects.filter(id=project.id).exists()
-        assert not Detector.objects.filter(id=detector.id).exists()
+        assert not Detector.objects.filter(id=detector_id).exists()
         assert not UptimeSubscription.objects.filter(id=uptime_subscription.id).exists()
-        mock_remove_seat.assert_called_with(seat_object=detector)
+        # remove_seat is called from both DetectorDeletionTask and
+        # UptimeSubscriptionDeletionTask as belt-and-suspenders.
+        assert mock_remove_seat.call_count == 2
 
 
 class DeleteWorkflowEngineModelsTest(DeleteProjectTest):

tests/sentry/uptime/subscriptions/test_subscriptions.py

@@ -874,7 +874,11 @@ def test_other_subscriptions(self, mock_remove_seat: mock.MagicMock) -> None:
             detector.refresh_from_db()
 
         assert UptimeSubscription.objects.filter(id=other_uptime_subscription.id).exists()
-        mock_remove_seat.assert_called_with(seat_object=detector)
+        # Use assert_any_call because remove_seat is called from multiple
+        # places (delete_uptime_detector, UptimeSubscriptionDeletionTask, and
+        # DetectorDeletionTask). The last call's reference gets mutated when
+        # instance.delete() sets pk=None, so assert_called_with would fail.
+        mock_remove_seat.assert_any_call(seat_object=detector)
 
     @mock.patch("sentry.quotas.backend.remove_seat")
     def test_single_subscriptions(self, mock_remove_seat: mock.MagicMock) -> None:
@@ -897,7 +901,11 @@ def test_single_subscriptions(self, mock_remove_seat: mock.MagicMock) -> None:
 
         with pytest.raises(UptimeSubscription.DoesNotExist):
             uptime_subscription.refresh_from_db()
-        mock_remove_seat.assert_called_with(seat_object=detector)
+        # Use assert_any_call because remove_seat is called from multiple
+        # places (delete_uptime_detector, UptimeSubscriptionDeletionTask, and
+        # DetectorDeletionTask). The last call's reference gets mutated when
+        # instance.delete() sets pk=None, so assert_called_with would fail.
+        mock_remove_seat.assert_any_call(seat_object=detector)
 
 
 class IsUrlMonitoredForProjectTest(UptimeTestCase):