feat(replays): Delete videos on replay delete request (#68463)

If we did not delete this the videos would remain in storage until the
retention-period expires.

Relates to: #63255

Author: cmanallen

src/sentry/replays/tasks.py

@@ -3,8 +3,10 @@
 import concurrent.futures as cf
 from typing import Any
 
+from google.cloud.exceptions import NotFound
+
 from sentry.replays.lib.kafka import initialize_replays_publisher
-from sentry.replays.lib.storage import filestore, storage
+from sentry.replays.lib.storage import filestore, make_video_filename, storage, storage_kv
 from sentry.replays.models import ReplayRecordingSegment
 from sentry.replays.usecases.events import archive_event
 from sentry.replays.usecases.reader import fetch_segments_metadata
@@ -54,15 +56,18 @@ def delete_replay_recording(project_id: int, replay_id: str) -> None:
     # Filestore and direct storage segments are split into two different delete operations.
     direct_storage_segments = []
     filestore_segments = []
+    video_filenames = []
     for segment in segments_from_metadata:
+        video_filenames.append(make_video_filename(segment))
         if segment.file_id:
             filestore_segments.append(segment)
         else:
             direct_storage_segments.append(segment)
 
     # Issue concurrent delete requests when interacting with a remote service provider.
-    if direct_storage_segments:
-        with cf.ThreadPoolExecutor(max_workers=100) as pool:
+    with cf.ThreadPoolExecutor(max_workers=100) as pool:
+        pool.map(_delete_if_exists, video_filenames)
+        if direct_storage_segments:
             pool.map(storage.delete, direct_storage_segments)
 
     # This will only run if "filestore" was used to store the files. This hasn't been the
@@ -81,3 +86,11 @@ def archive_replay(publisher: KafkaPublisher, project_id: int, replay_id: str) -
     # We publish manually here because we sometimes provide a managed Kafka
     # publisher interface which has its own setup and teardown behavior.
     publisher.publish("ingest-replay-events", message)
+
+
+def _delete_if_exists(filename: str) -> None:
+    """Delete the blob if it exists or silence the 404."""
+    try:
+        storage_kv.delete(filename)
+    except NotFound:
+        pass

tests/sentry/replays/test_project_replay_details.py

@@ -7,7 +7,12 @@
 
 from sentry.models.files.file import File
 from sentry.replays.lib import kafka
-from sentry.replays.lib.storage import RecordingSegmentStorageMeta, storage
+from sentry.replays.lib.storage import (
+    RecordingSegmentStorageMeta,
+    make_video_filename,
+    storage,
+    storage_kv,
+)
 from sentry.replays.models import ReplayRecordingSegment
 from sentry.replays.testutils import assert_expected_response, mock_expected_response, mock_replay
 from sentry.testutils.cases import APITestCase, ReplaysSnubaTestCase
@@ -210,6 +215,7 @@ def test_delete_replay_from_clickhouse_data(self):
             file_id=None,
         )
         storage.set(metadata1, b"hello, world!")
+        storage_kv.set(make_video_filename(metadata1), b"hello, world!")
 
         metadata2 = RecordingSegmentStorageMeta(
             project_id=self.project.id,
@@ -219,6 +225,8 @@ def test_delete_replay_from_clickhouse_data(self):
             file_id=None,
         )
         storage.set(metadata2, b"hello, world!")
+        # Intentionally not written.
+        # storage_kv.set(make_video_filename(metadata2), b"hello, world!")
 
         metadata3 = RecordingSegmentStorageMeta(
             project_id=self.project.id,
@@ -228,6 +236,7 @@ def test_delete_replay_from_clickhouse_data(self):
             file_id=None,
         )
         storage.set(metadata3, b"hello, world!")
+        storage_kv.set(make_video_filename(metadata3), b"hello, world!")
 
         with self.feature(REPLAYS_FEATURES):
             with TaskRunner():
@@ -237,3 +246,6 @@ def test_delete_replay_from_clickhouse_data(self):
         assert storage.get(metadata1) is None
         assert storage.get(metadata2) is None
         assert storage.get(metadata3) is not None
+        assert storage_kv.get(make_video_filename(metadata1)) is None
+        assert storage_kv.get(make_video_filename(metadata2)) is None
+        assert storage_kv.get(make_video_filename(metadata3)) is not None