Reapply "fix(hybridcloud) Continue to track down write volume from sentryapps v2 (#96657)

This reverts commit
51eed24.

Redo of #96587 which was
reverted because of semantic conflicts.

Author: markstory

src/sentry/hybridcloud/models/cacheversion.py

@@ -1,6 +1,3 @@
-import logging
-import random
-
 from django.db import models, router, transaction
 from django.db.models import F
 
@@ -9,8 +6,6 @@
 from sentry.db.postgres.transactions import enforce_constraints
 from sentry.options.rollout import in_random_rollout
 
-logger = logging.getLogger(__name__)
-
 
 class CacheVersionBase(Model):
     class Meta:
@@ -23,8 +18,6 @@ class Meta:
     def incr_version(cls, key: str) -> int:
         if in_random_rollout("sentry.hybridcloud.cacheversion.rollout"):
             with enforce_constraints(transaction.atomic(router.db_for_write(cls))):
-                if random.random() < 0.01:
-                    logger.info("cacheversion.incr_version", extra={"key": key})
                 obj, created = cls.objects.select_for_update().get_or_create(
                     key=key, defaults=dict(version=1)
                 )
@@ -36,8 +29,6 @@ def incr_version(cls, key: str) -> int:
                 return obj.version
 
         with enforce_constraints(transaction.atomic(router.db_for_write(cls))):
-            if random.random() < 0.01:
-                logger.info("cacheversion.incr_version", extra={"key": key})
             cls.objects.create_or_update(
                 key=key, defaults=dict(version=1), values=dict(version=F("version") + 1)
             )

src/sentry/hybridcloud/rpc/caching/service.py

@@ -3,10 +3,12 @@
 # in modules such as this one where hybrid cloud data models or service classes are
 # defined, because we want to reflect on type annotations and avoid forward references.
 import abc
+import random
 from collections.abc import Callable, Generator, Mapping
 from typing import TYPE_CHECKING, Generic, TypeVar
 
 import pydantic
+import sentry_sdk
 
 from sentry.hybridcloud.rpc.resolvers import ByRegionName
 from sentry.hybridcloud.rpc.service import RpcService, regional_rpc_method, rpc_method
@@ -162,7 +164,18 @@ def resolve_from(
             try:
                 metrics.incr("hybridcloud.caching.list.cached", tags={"base_key": self.base_key})
                 return [self.type_(**item) for item in json.loads(value)]
-            except (pydantic.ValidationError, JSONDecodeError, TypeError):
+            except (pydantic.ValidationError, JSONDecodeError, TypeError) as err:
+                metrics.incr(
+                    "hybridcloud.caching.list.failed_read",
+                    tags={
+                        "base_key": self.base_key,
+                        "err": type(err).__name__,
+                    },
+                )
+                if random.random() < 0.001:
+                    with sentry_sdk.isolation_scope() as scope:
+                        scope.set_level("warning")
+                        sentry_sdk.capture_exception(err)
                 version = yield from _delete_cache(key, self.silo_mode)
         else:
             version = value