ref(replays): Retry deletes on failure and record failed bulk delete jobs as failed (#93996)

Deletes may sometimes fail due to service flakes. We should make an
attempt to retry. If we can't delete the replay go into the failed
state. We should offer retry logic in a follow-up PR. Or a beat task to
automate the retry. But if the error is deterministic then we'll just
spin.

Author: cmanallen

src/sentry/filestore/gcs.py

@@ -393,22 +393,28 @@ def _try(self, callable: Callable[[], None]) -> None:
 class GoogleCloudStorageWithReplayUploadPolicy(GoogleCloudStorage):
     """Google cloud storage class with replay upload policy."""
 
-    # "try_del" and "try_get" inherit the default behavior. We don't want to exponentially
-    # wait in those contexts. We're maintaining the status-quo for now but in the future we
-    # can add policies for these methods or use no policy at all and implement retries at a
-    # higher, more contextual level.
+    # "try_get" inherits the default behavior. We don't want to exponentially wait in that
+    # context. We're maintaining the status-quo for now but in the future we can add policies for
+    # these methods or use no policy at all and implement retries at a higher, more contextual
+    # level.
     #
-    # def try_del(self, callable: Callable[[], None]) -> None:
     # def try_get(self, callable: Callable[[], None]) -> None:
 
-    def try_set(self, callable: Callable[[], None]) -> None:
-        """Upload a blob with exponential delay for a maximum of five attempts."""
+    def create_retry_policy(self):
+        """Retry an action with sigmoid delay for a maximum of five attempts."""
 
         def should_retry(attempt: int, e: Exception) -> bool:
             """Retry gateway timeout exceptions up to the limit."""
             return attempt <= REPLAY_GCS_RETRIES and isinstance(e, GCS_RETRYABLE_ERRORS)
 
         # Retry cadence: After a brief period of fast retries the function will retry once
         # per second for two minutes.
-        policy = ConditionalRetryPolicy(should_retry, sigmoid_delay())
+        return ConditionalRetryPolicy(should_retry, sigmoid_delay())
+
+    def try_set(self, callable: Callable[[], None]) -> None:
+        policy = self.create_retry_policy()
+        policy(callable)
+
+    def try_del(self, callable: Callable[[], None]) -> None:
+        policy = self.create_retry_policy()
         policy(callable)

src/sentry/replays/models.py

@@ -19,6 +19,7 @@ class DeletionJobStatus(models.TextChoices):
     PENDING = "pending", gettext_lazy("Pending")
     IN_PROGRESS = "in-progress", gettext_lazy("In Progress")
     COMPLETED = "completed", gettext_lazy("Completed")
+    FAILED = "failed", gettext_lazy("Failed")
 
 
 @region_silo_model

src/sentry/replays/tasks.py

@@ -186,25 +186,34 @@ def run_bulk_replay_delete_job(replay_delete_job_id: int, offset: int, limit: in
     job = ReplayDeletionJobModel.objects.get(id=replay_delete_job_id)
 
     # If this is the first run of the task we set the model to in-progress.
-    if offset == 0:
+    if job.status == DeletionJobStatus.PENDING:
         job.status = DeletionJobStatus.IN_PROGRESS
         job.save()
 
-    # Delete the replays within a limited range. If more replays exist an incremented offset value
-    # is returned.
-    results = fetch_rows_matching_pattern(
-        project_id=job.project_id,
-        start=job.range_start,
-        end=job.range_end,
-        query=job.query,
-        environment=job.environments,
-        limit=limit,
-        offset=offset,
-    )
+    # Exit if the job status is failed or completed.
+    if job.status != DeletionJobStatus.IN_PROGRESS:
+        return None
+
+    try:
+        # Delete the replays within a limited range. If more replays exist an incremented offset value
+        # is returned.
+        results = fetch_rows_matching_pattern(
+            project_id=job.project_id,
+            start=job.range_start,
+            end=job.range_end,
+            query=job.query,
+            environment=job.environments,
+            limit=limit,
+            offset=offset,
+        )
 
-    # Delete the matched rows if any rows were returned.
-    if len(results["rows"]) > 0:
-        delete_matched_rows(job.project_id, results["rows"])
+        # Delete the matched rows if any rows were returned.
+        if len(results["rows"]) > 0:
+            delete_matched_rows(job.project_id, results["rows"])
+    except Exception:
+        job.status = DeletionJobStatus.FAILED
+        job.save()
+        raise
 
     # Compute the next offset to start from. If no further processing is required then this serves
     # as a count of replays deleted.

tests/sentry/replays/tasks/test_delete_replays_bulk.py

@@ -4,7 +4,7 @@
 import uuid
 from unittest.mock import patch
 
-from sentry.replays.models import ReplayDeletionJobModel
+from sentry.replays.models import DeletionJobStatus, ReplayDeletionJobModel
 from sentry.replays.tasks import run_bulk_replay_delete_job
 from sentry.replays.testutils import mock_replay
 from sentry.testutils.cases import APITestCase, ReplaysSnubaTestCase
@@ -180,6 +180,24 @@ def test_run_bulk_replay_delete_job_chained_runs(self):
         assert self.job.status == "completed"
         assert self.job.offset == 2
 
+    def test_run_bulk_replay_delete_job_already_failed(self):
+        t1 = datetime.datetime.now() - datetime.timedelta(seconds=10)
+        replay_id1 = uuid.uuid4().hex
+        self.store_replays(
+            mock_replay(t1, self.project.id, replay_id1, segment_id=0, environment="prod")
+        )
+
+        self.job.status = DeletionJobStatus.FAILED
+        self.job.save()
+
+        with TaskRunner():
+            run_bulk_replay_delete_job.delay(self.job.id, offset=0, limit=0)
+
+        # Runs were chained.
+        self.job.refresh_from_db()
+        assert self.job.status == "failed"
+        assert self.job.offset == 0
+
     def test_run_bulk_replay_delete_job_no_matches(self):
         with TaskRunner():
             run_bulk_replay_delete_job.delay(self.job.id, offset=0)