feat(metrics): Reject stale keys in indexer cache (#60283)

### Overview

Reject keys that are too old in the indexer cache to enforce ttl

src/sentry/sentry_metrics/indexer/cache.py

@@ -126,6 +126,7 @@ class StringIndexerCache:
 
         if not self._is_valid_timestamp(timestamp):
             metrics.incr(_INDEXER_CACHE_STALE_KEYS_METRIC)
+            return None
 
         return int(result)
 

tests/sentry/sentry_metrics/test_all_indexers.py

@@ -11,7 +11,11 @@ from typing import Mapping, Set
 import pytest
 
 from sentry.sentry_metrics.indexer.base import FetchType, FetchTypeExt, Metadata
-from sentry.sentry_metrics.indexer.cache import CachingIndexer, StringIndexerCache
+from sentry.sentry_metrics.indexer.cache import (
+    BULK_RECORD_CACHE_NAMESPACE,
+    CachingIndexer,
+    StringIndexerCache,
+)
 from sentry.sentry_metrics.indexer.mock import RawSimpleIndexer
 from sentry.sentry_metrics.indexer.postgres.postgres_v2 import PGStringIndexerV2
 from sentry.sentry_metrics.indexer.strings import SHARED_STRINGS, StaticStringIndexer
@@ -342,6 +346,58 @@ def test_already_created_plus_written_results(indexer, indexer_cache, use_case_i
         )
 
 
+def test_invalid_timestamp_in_indexer_cache(indexer, indexer_cache) -> None:
+    """
+    Test that the caching indexer will incur a miss if the data is staying in the cache for too long
+    """
+    with override_options(
+        {
+            "sentry-metrics.indexer.read-new-cache-namespace": True,
+            "sentry-metrics.indexer.write-new-cache-namespace": True,
+        }
+    ):
+        indexer = CachingIndexer(indexer_cache, indexer)
+
+        use_case_id = UseCaseID.SPANS
+        org_id = 1
+        s = "str"
+
+        res = indexer.bulk_record({use_case_id: {org_id: {s}}})
+        id = res.get_mapped_results()[use_case_id][org_id][s]
+        assert res.get_fetch_metadata()[use_case_id][org_id][s].fetch_type == FetchType.FIRST_SEEN
+
+        indexer.cache.cache.set(
+            indexer.cache._make_namespaced_cache_key(
+                BULK_RECORD_CACHE_NAMESPACE, f"{use_case_id.value}:{org_id}:{s}"
+            ),
+            indexer.cache._make_cache_val(id, 0),
+            version=indexer_cache.version,
+        )
+
+        assert (
+            indexer.bulk_record({use_case_id: {org_id: {s}}})
+            .get_fetch_metadata()[use_case_id][org_id][s]
+            .fetch_type
+            == FetchType.DB_READ
+        )
+
+        assert indexer.cache._validate_result(
+            indexer.cache.cache.get(
+                indexer.cache._make_namespaced_cache_key(
+                    BULK_RECORD_CACHE_NAMESPACE, f"{use_case_id.value}:{org_id}:{s}"
+                ),
+                version=indexer_cache.version,
+            )
+        )
+
+        assert (
+            indexer.bulk_record({use_case_id: {org_id: {s}}})
+            .get_fetch_metadata()[use_case_id][org_id][s]
+            .fetch_type
+            == FetchType.CACHE_HIT
+        )
+
+
 def test_already_cached_plus_read_results(indexer, indexer_cache, use_case_id) -> None:
     """
     Test that we correctly combine cached results with read results

tests/sentry/sentry_metrics/test_indexer_cache.py

@@ -93,6 +93,26 @@ def test_cache(use_case_id: str) -> None:
         assert indexer_cache.get(namespace_2, f"{use_case_id}:1:blah:123") is None
 
 
+def test_cache_validate_stale_timestamp():
+    with override_options(
+        {
+            "sentry-metrics.indexer.read-new-cache-namespace": True,
+            "sentry-metrics.indexer.write-new-cache-namespace": True,
+        }
+    ):
+        namespace = "test"
+        key = "spans:1:key"
+        cache.clear()
+        cache.set(
+            indexer_cache._make_namespaced_cache_key(namespace, key),
+            indexer_cache._make_cache_val(1, 0),
+        )
+        assert indexer_cache.get_many(namespace, [key]) == {key: None}
+
+        indexer_cache.set_many(namespace, {key: 1})
+        assert indexer_cache.get_many(namespace, [key]) == {key: 1}
+
+
 def test_cache_many(use_case_id: str) -> None:
     with override_options(
         {