Clean up after full attachments rollout

This cleans up the codebase after attachmens on objectstore has been fully rolled out.
It assumes that attachments are now fully stored from within relay, and assumes that no attachment chunks will ever be emitted.

Thus, there is no more need for the attachments cache, or the chunk message handlers.
All feature flags related to objectstore rollout are removed and hardcoded.

Author: Swatinem

src/sentry/api/endpoints/event_attachment_details.py

@@ -1,8 +1,5 @@
-import contextlib
 import posixpath
-from typing import IO, ContextManager
 
-import sentry_sdk
 from django.http import StreamingHttpResponse
 from rest_framework.request import Request
 from rest_framework.response import Response
@@ -17,11 +14,10 @@
 from sentry.auth.system import is_system_auth
 from sentry.constants import ATTACHMENTS_ROLE_DEFAULT
 from sentry.models.activity import Activity
-from sentry.models.eventattachment import V1_PREFIX, V2_PREFIX, EventAttachment
+from sentry.models.eventattachment import EventAttachment
 from sentry.models.organizationmember import OrganizationMember
 from sentry.services import eventstore
 from sentry.types.activity import ActivityType
-from sentry.utils import metrics
 
 
 class EventAttachmentDetailsPermission(ProjectPermission):
@@ -65,35 +61,12 @@ def download(self, attachment: EventAttachment):
         name = posixpath.basename(" ".join(attachment.name.split()))
 
         def stream_attachment():
-            attachment_file = attachment.getfile()
-            doublewrite_file: IO[bytes] | ContextManager[None] = contextlib.nullcontext()
-            blob_path = attachment.blob_path or ""
-            blob_path = blob_path.startswith(V1_PREFIX) and blob_path.removeprefix(V1_PREFIX) or ""
-            if blob_path.startswith(V2_PREFIX):
-                try:
-                    # We force the attachment model to use the objectstore backend
-                    # by changing its prefix. Its a big hack, but hey why not.
-                    attachment.blob_path = blob_path
-                    doublewrite_file = attachment.getfile()
-                    metrics.incr("storage.attachments.double_write.read")
-                except Exception:
-                    sentry_sdk.capture_exception()
-
             # TODO: We should pass along the `Accept-Encoding`, so we can avoid
             # decompressing on the API side, and just transfer the already
             # compressed bytes to the client as it indicated it can handle it.
-            with attachment_file as af, doublewrite_file as df:
-                while filestore_chunk := af.read(4096):
-                    if df:
-                        try:
-                            objectstore_chunk = df.read(4096)
-                            assert filestore_chunk == objectstore_chunk
-                        except Exception:
-                            # If we have encountered one error, clear the reference
-                            # to avoid spamming more errors for all the remaining chunks.
-                            df = None
-                            sentry_sdk.capture_exception()
-                    yield filestore_chunk
+            with attachment.getfile() as attachment_file:
+                while chunk := attachment_file.read(4096):
+                    yield chunk
 
         response = StreamingHttpResponse(
             stream_attachment(),

src/sentry/attachments.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+__all__ = [
+    "store_attachments_for_event",
+    "get_attachments_for_event",
+    "delete_cached_and_ratelimited_attachments",
+    "CachedAttachment",
+]
+
+from typing import TYPE_CHECKING, Any
+
+import sentry_sdk
+
+from sentry.objectstore import get_attachments_session
+from sentry.utils.json import prune_empty_keys
+
+if TYPE_CHECKING:
+    from sentry.models.project import Project
+
+
+UNINITIALIZED_DATA = object()
+
+
+class CachedAttachment:
+    def __init__(
+        self,
+        name=None,
+        content_type=None,
+        type=None,
+        data=UNINITIALIZED_DATA,
+        stored_id: str | None = None,
+        rate_limited=None,
+        size=None,
+        **kwargs,
+    ):
+        self.name = name
+        self.content_type = content_type
+        self.type = type or "event.attachment"
+        assert isinstance(self.type, str), self.type
+        self.rate_limited = rate_limited
+
+        if size is not None:
+            self.size = size
+        elif data not in (None, UNINITIALIZED_DATA):
+            self.size = len(data)
+        else:
+            self.size = 0
+
+        self.stored_id = stored_id
+        self._data = data
+
+    def load_data(self, project: Project | None = None) -> bytes:
+        if self.stored_id:
+            assert project
+            session = get_attachments_session(project.organization_id, project.id)
+            return session.get(self.stored_id).payload.read()
+
+        assert self._data is not UNINITIALIZED_DATA
+        return self._data
+
+    def meta(self) -> dict:
+        return prune_empty_keys(
+            {
+                "name": self.name,
+                "rate_limited": self.rate_limited,
+                "content_type": self.content_type,
+                "type": self.type,
+                "size": self.size,
+                "stored_id": self.stored_id,
+            }
+        )
+
+
+@sentry_sdk.trace
+def store_attachments_for_event(project: Project, event: Any, attachments: list[CachedAttachment]):
+    """
+    Stores the given list of `attachments` belonging to `event` for processing.
+
+    The attachment metadata is stored within the `event`, and attachment payloads
+    are stored in `objectstore`.
+    """
+
+    attachments_metadata: list[dict] = []
+    for attachment in attachments:
+        # if the attachment has non-empty data set, we want to store it, overwriting any existing data in case a `stored_id` is set.
+        if attachment._data and attachment._data is not UNINITIALIZED_DATA:
+            session = get_attachments_session(project.organization_id, project.id)
+            attachment.stored_id = session.put(attachment._data, key=attachment.stored_id)
+
+        attachments_metadata.append(attachment.meta())
+
+    event["_attachments"] = attachments_metadata
+
+
+def get_attachments_for_event(event: Any) -> list[CachedAttachment]:
+    """
+    Retrieves the attachments belonging to the given `event`.
+    """
+
+    return [CachedAttachment(**attachment) for attachment in event.get("_attachments", [])]
+
+
+@sentry_sdk.trace
+def delete_cached_and_ratelimited_attachments(
+    project: Project, attachments: list[CachedAttachment]
+):
+    """
+    This deletes all the `rate_limited` attachments from the `objectstore`.
+    Non-ratelimited attachments which are already stored in `objectstore` will
+    be retained there for long-term storage.
+    """
+    for attachment in attachments:
+        # deletes from objectstore if no long-term storage is desired
+        if attachment.rate_limited and attachment.stored_id:
+            get_attachments_session(project.organization_id, project.id).delete(
+                attachment.stored_id
+            )