feat(preprod): Read snapshot objectstore content from both usecases

src/sentry/objectstore/__init__.py

@@ -89,6 +89,7 @@ class UsecaseId(Enum):
     DEBUG_FILES = "debug_files"
     PROFILE_ATTACHMENTS = "profile_attachments"
     PREPROD = "preprod"
+    PREPROD_SNAPSHOTS = "preprod_snapshots"
 
     def create(self) -> ObjectstoreClientUsecase:
         match self:
@@ -108,7 +109,7 @@ def create(self) -> ObjectstoreClientUsecase:
                     self.value,
                     expiration_policy=TimeToLive(timedelta(days=default_attachment_retention())),
                 )
-            case UsecaseId.PREPROD:
+            case UsecaseId.PREPROD | UsecaseId.PREPROD_SNAPSHOTS:
                 return ObjectstoreClientUsecase(
                     self.value,
                     expiration_policy=TimeToIdle(timedelta(days=30)),

src/sentry/objectstore/types.py

@@ -5,6 +5,7 @@
 
 class ObjectstoreUploadOptions(TypedDict):
     url: str
+    usecase: str
     scopes: list[tuple[str, str]]
     authToken: str | None
     expirationPolicy: str

src/sentry/preprod/api/endpoints/project_preprod_artifact_image.py

@@ -14,7 +14,7 @@
 from sentry.api.base import cell_silo_endpoint
 from sentry.api.bases.project import ProjectEndpoint
 from sentry.models.project import Project
-from sentry.objectstore import UsecaseId, get_session
+from sentry.preprod.snapshots.storage import get_snapshot_storage
 from sentry.ratelimits.config import RateLimitConfig
 from sentry.types.ratelimit import RateLimit, RateLimitCategory
 
@@ -69,7 +69,7 @@ def get(
         project_id = project.id
 
         object_key = f"{organization_id}/{project_id}/{image_id}"
-        session = get_session(UsecaseId.PREPROD, project)
+        session = get_snapshot_storage(project)
 
         try:
             result = session.get(object_key)

src/sentry/preprod/api/endpoints/project_preprod_upload_options.py

@@ -28,8 +28,12 @@ class ProjectPreprodUploadOptionsEndpoint(ProjectEndpoint):
     permission_classes = (ProjectReleasePermission,)
 
     def get(self, request: Request, project: Project) -> Response:
+        if request.GET.get("usecase") not in (None, "auto"):
+            return Response({"detail": "Invalid usecase"}, status=400)
+        usecase = UsecaseId.PREPROD
+
         organization = project.organization
-        session = get_session(UsecaseId.PREPROD, project)
+        session = get_session(usecase, project)
 
         path = reverse(
             "sentry-api-0-organization-objectstore",
@@ -44,6 +48,7 @@ def get(self, request: Request, project: Project) -> Response:
 
         options = ObjectstoreUploadOptions(
             url=url,
+            usecase=usecase.value,
             scopes=[
                 ("org", str(organization.id)),
                 ("project", str(project.id)),

src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot.py

@@ -34,7 +34,6 @@
 from sentry.models.commitcomparison import CommitComparison
 from sentry.models.organization import Organization
 from sentry.models.project import Project
-from sentry.objectstore import UsecaseId, get_session
 from sentry.preprod.analytics import (
     PreprodArtifactApiDeleteEvent,
     PreprodArtifactApiGetSnapshotDetailsEvent,
@@ -79,6 +78,7 @@
     load_precomputed_head_images,
     refresh_manifest_expiration,
 )
+from sentry.preprod.snapshots.storage import get_snapshot_storage
 from sentry.preprod.snapshots.tasks import compare_snapshots
 from sentry.preprod.snapshots.utils import (
     find_base_snapshot_artifact,
@@ -355,7 +355,7 @@ def get(
             return Response({"detail": "Snapshot metrics not found"}, status=404)
 
         extras = snapshot_metrics.extras or {}
-        session = get_session(UsecaseId.PREPROD, artifact.project)
+        session = get_snapshot_storage(artifact.project)
 
         image_list: list[SnapshotImageResponseDict]
         precomputed = load_precomputed_head_images(session, extras.get("head_images_key"))
@@ -844,7 +844,7 @@ def post(
 
             # Write manifest inside the transaction so that a failed objectstore
             # write rolls back the DB records, ensuring both succeed or neither does.
-            session = get_session(UsecaseId.PREPROD, project)
+            session = get_snapshot_storage(project)
             manifest_bytes = manifest.json(exclude_none=True).encode()
             manifest_size_bytes = len(manifest_bytes)
             session.put(manifest_bytes, key=manifest_key)

src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot_archive.py

@@ -18,10 +18,10 @@
 from sentry.auth.staff import is_active_staff
 from sentry.issues.action_log import resolve_action_source
 from sentry.models.organization import Organization
-from sentry.objectstore import UsecaseId, get_session
 from sentry.preprod.analytics import PreprodArtifactApiSnapshotArchiveDownloadEvent
 from sentry.preprod.models import PreprodArtifact
 from sentry.preprod.snapshots.models import PreprodSnapshotMetrics
+from sentry.preprod.snapshots.storage import get_snapshot_storage
 from sentry.preprod.snapshots.zip_builder import archive_exists, archive_object_key
 from sentry.preprod.snapshots.zip_tasks import build_snapshot_images_zip
 from sentry.ratelimits.config import RateLimitConfig
@@ -87,7 +87,7 @@ def _resolve(
         return artifact, metrics
 
     def _download(self, artifact: PreprodArtifact) -> HttpResponseBase:
-        session = get_session(UsecaseId.PREPROD, artifact.project)
+        session = get_snapshot_storage(artifact.project)
         result = session.get(archive_object_key(artifact.id))
         if result is None:
             return Response({"detail": "Download not ready"}, status=409)
@@ -104,7 +104,7 @@ def _download(self, artifact: PreprodArtifact) -> HttpResponseBase:
         return response
 
     def _archive_exists(self, artifact: PreprodArtifact) -> bool:
-        session = get_session(UsecaseId.PREPROD, artifact.project)
+        session = get_snapshot_storage(artifact.project)
         try:
             return archive_exists(session, archive_object_key(artifact.id))
         except RequestError:

src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot_image_detail.py

@@ -24,7 +24,6 @@
 from sentry.auth.staff import is_active_staff
 from sentry.issues.action_log import resolve_action_source
 from sentry.models.organization import Organization
-from sentry.objectstore import UsecaseId, get_session
 from sentry.preprod.analytics import PreprodArtifactApiGetSnapshotImageEvent
 from sentry.preprod.api.models.public.snapshots import SnapshotImageDetailResponseDict
 from sentry.preprod.api.models.snapshots.project_preprod_snapshot_models import (
@@ -40,6 +39,7 @@
     image_metadata_extras,
 )
 from sentry.preprod.snapshots.models import PreprodSnapshotComparison, PreprodSnapshotMetrics
+from sentry.preprod.snapshots.storage import get_snapshot_storage
 
 logger = logging.getLogger(__name__)
 
@@ -216,7 +216,7 @@ def get(
             return Response({"detail": "Manifest key not found"}, status=404)
 
         try:
-            session = get_session(UsecaseId.PREPROD, artifact.project)
+            session = get_snapshot_storage(artifact.project)
             response = session.get(manifest_key)
             if response is None:
                 raise FileNotFoundError("Manifest does not exist in objectstore")

src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot_latest_base.py

@@ -27,7 +27,6 @@
 from sentry.constants import ObjectStatus
 from sentry.issues.action_log import resolve_action_source
 from sentry.models.organization import Organization
-from sentry.objectstore import UsecaseId, get_session
 from sentry.preprod.analytics import PreprodArtifactApiGetLatestBaseSnapshotEvent
 from sentry.preprod.api.endpoints.snapshots.preprod_artifact_snapshot import (
     _strip_to_compact,
@@ -38,6 +37,7 @@
 )
 from sentry.preprod.models import PreprodArtifact
 from sentry.preprod.snapshots.image_serialization import build_head_image_dict
+from sentry.preprod.snapshots.storage import get_snapshot_storage
 
 logger = logging.getLogger(__name__)
 
@@ -201,7 +201,7 @@ def get(
             return Response({"detail": "Manifest key not found"}, status=404)
 
         try:
-            session = get_session(UsecaseId.PREPROD, artifact.project)
+            session = get_snapshot_storage(artifact.project)
             response = session.get(manifest_key)
             if response is None:
                 raise FileNotFoundError("Manifest does not exist in objectstore")

src/sentry/preprod/helpers/deletion.py

@@ -17,11 +17,11 @@
 from sentry_protos.snuba.v1.trace_item_filter_pb2 import ComparisonFilter, TraceItemFilter
 
 from sentry.models.files.file import File
-from sentry.objectstore import UsecaseId, get_session
 from sentry.preprod.eap.constants import get_preprod_trace_id
 from sentry.preprod.models import PreprodArtifact, PreprodArtifactSizeMetrics
 from sentry.preprod.snapshots.manifest import ComparisonManifest
 from sentry.preprod.snapshots.models import PreprodSnapshotComparison, PreprodSnapshotMetrics
+from sentry.preprod.snapshots.storage import get_snapshot_storage
 from sentry.utils import snuba_rpc
 from sentry.utils.concurrent import ContextPropagatingThreadPoolExecutor
 
@@ -150,7 +150,7 @@ def _collect_snapshot_objectstore_keys(
 
         keys.append((org_id, project_id, comparison_key))
         try:
-            session = get_session(UsecaseId.PREPROD, project_id, org=org_id)
+            session = get_snapshot_storage(project_id, org=org_id)
             response = session.get(comparison_key)
             if response is None:
                 raise FileNotFoundError("Comparison manifest does not exist in objectstore")
@@ -170,7 +170,7 @@ def _collect_snapshot_objectstore_keys(
 def _delete_objectstore_key(args: tuple[int, int, str]) -> bool:
     org_id, project_id, key = args
     try:
-        get_session(UsecaseId.PREPROD, project_id, org=org_id).delete(key)
+        get_snapshot_storage(project_id, org=org_id).delete(key)
         return True
     except Exception:
         logger.exception("preprod.cleanup.objectstore_delete_failed", extra={"key": key})

src/sentry/preprod/snapshots/precompute.py

@@ -5,10 +5,10 @@
 from typing import Any, TypedDict
 
 import orjson
-from objectstore_client import Session
 
 from sentry.preprod.api.models.public.snapshots import SnapshotImageResponseDict
 from sentry.preprod.snapshots.image_serialization import build_head_image_list
+from sentry.preprod.snapshots.storage import SnapshotStorage
 from sentry.utils.tracing import set_span_data, start_span
 
 logger = logging.getLogger(__name__)
@@ -36,7 +36,7 @@ def build_head_images_payload(
     }
 
 
-def refresh_manifest_expiration(session: Session, manifest_key: str | None) -> None:
+def refresh_manifest_expiration(session: SnapshotStorage, manifest_key: str | None) -> None:
     if not manifest_key:
         return
     try:
@@ -46,7 +46,7 @@ def refresh_manifest_expiration(session: Session, manifest_key: str | None) -> N
 
 
 def load_precomputed_head_images(
-    session: Session, key: str | None
+    session: SnapshotStorage, key: str | None
 ) -> tuple[list[SnapshotImageResponseDict], float | None] | None:
     if not key:
         return None

src/sentry/preprod/snapshots/reconstruction.py

@@ -5,12 +5,13 @@
 from typing import NamedTuple
 
 import orjson
-from objectstore_client import RequestError, Session
+from objectstore_client import RequestError
 from pydantic import ValidationError
 
 from sentry.preprod.models import PreprodArtifact
 from sentry.preprod.snapshots.categorize import categorize_image_sets
 from sentry.preprod.snapshots.manifest import ImageMetadata, SnapshotManifest
+from sentry.preprod.snapshots.storage import SnapshotStorage
 from sentry.utils.concurrent import ContextPropagatingThreadPoolExecutor
 
 MAX_CHAIN_DEPTH = 50
@@ -63,7 +64,7 @@ def _manifest_key(artifact: PreprodArtifact) -> str | None:
     return (metrics.extras or {}).get("manifest_key")
 
 
-def _read_manifest_by_key(session: Session, key: str) -> _ManifestRead:
+def _read_manifest_by_key(session: SnapshotStorage, key: str) -> _ManifestRead:
     """Fetch and parse a manifest. Pure objectstore I/O — no ORM access, so this is safe to
     run from a worker thread (see _fetch_manifests)."""
     try:
@@ -134,7 +135,7 @@ def _collect_chain(base_artifact: PreprodArtifact) -> tuple[list[PreprodArtifact
 
 
 def _fetch_manifests(
-    session: Session, artifacts: list[PreprodArtifact]
+    session: SnapshotStorage, artifacts: list[PreprodArtifact]
 ) -> dict[int, _ManifestRead]:
     """Read each artifact's manifest, fanning the objectstore gets across a thread pool.
 
@@ -167,7 +168,7 @@ def _fetch_manifests(
 
 
 def reconstruct_base_manifest(
-    base_artifact: PreprodArtifact, session: Session
+    base_artifact: PreprodArtifact, session: SnapshotStorage
 ) -> ReconstructionResult:
     """Reconstruct the complete manifest for base_artifact by folding the ancestry chain.
 

src/sentry/preprod/snapshots/storage.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from typing import IO, Literal
+
+from objectstore_client import Compression, GetResponse, Metadata, RequestError, Session
+from objectstore_client.multipart import MultipartUpload
+from urllib3.exceptions import HTTPError
+
+from sentry.models.project import Project
+from sentry.objectstore import UsecaseId, get_session
+from sentry.utils import metrics
+
+
+# TODO: On January 1, 2027, remove the preprod fallback and use preprod_snapshots exclusively.
+class SnapshotStorage:
+    def __init__(self, primary: Session, fallback: Session) -> None:
+        self._primary = primary
+        self._fallback = fallback
+
+    def get(self, key: str) -> GetResponse | None:
+        response = self._primary.get(key)
+        if response is None:
+            response = self._fallback.get(key)
+            self._record_fallback("get", response is not None)
+        return response
+
+    def head(self, key: str) -> Metadata | None:
+        metadata = self._primary.head(key)
+        if metadata is None:
+            metadata = self._fallback.head(key)
+            self._record_fallback("head", metadata is not None)
+        return metadata
+
+    def _record_fallback(self, op: str, found: bool) -> None:
+        metrics.incr(
+            "preprod.snapshot_storage.legacy_fallback",
+            tags={"op": op, "found": str(found).lower()},
+        )
+
+    def put(self, contents: bytes | IO[bytes], *, key: str, content_type: str | None = None) -> str:
+        return self._primary.put(contents, key=key, content_type=content_type)
+
+    def initiate_multipart_upload(
+        self, *, key: str, compression: Compression | Literal["none"], content_type: str
+    ) -> MultipartUpload:
+        return self._primary.initiate_multipart_upload(
+            key=key, compression=compression, content_type=content_type
+        )
+
+    def delete(self, key: str) -> None:
+        error: RequestError | HTTPError | None = None
+        for session in (self._primary, self._fallback):
+            try:
+                session.delete(key)
+            except (RequestError, HTTPError) as caught_error:
+                if error is None and (
+                    not isinstance(caught_error, RequestError) or caught_error.status != 404
+                ):
+                    error = caught_error
+        if error is not None:
+            raise error
+
+
+def get_snapshot_storage(project: Project | int, *, org: int | None = None) -> SnapshotStorage:
+    return SnapshotStorage(
+        primary=get_session(UsecaseId.PREPROD, project, org=org),
+        fallback=get_session(UsecaseId.PREPROD_SNAPSHOTS, project, org=org),
+    )