feat(uptime): Detect whole-store config loss with sentinel keys

fixtures/stubs-for-mypy/rediscluster/nodemanager.pyi

@@ -0,0 +1,5 @@
+from typing import Any
+
+class NodeManager:
+    def __init__(self, startup_nodes: list[dict[str, Any]], **kwargs: object) -> None: ...
+    def keyslot(self, key: str | bytes) -> int: ...

src/sentry/conf/server.py

@@ -1139,6 +1139,10 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
         "task": "uptime:sentry.uptime.tasks.config_drift_dispatcher",
         "schedule": crontab("0", "*/1", "*", "*", "*"),
     },
+    "uptime-config-sentinel-checker": {
+        "task": "uptime:sentry.uptime.tasks.check_config_sentinels",
+        "schedule": crontab("*/1", "*", "*", "*", "*"),
+    },
     "poll_tempest": {
         "task": "tempest:sentry.tempest.tasks.poll_tempest",
         "schedule": crontab("*/1", "*", "*", "*", "*"),

src/sentry/options/defaults.py

@@ -3725,6 +3725,15 @@
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
 
+# Kill switch for sentinel-triggered whole-store repair. Defaults to True (repair off) so the
+# option is never left permanently True with no way to remove it.
+register(
+    "uptime.config-drift.sentinel-repair-disabled",
+    type=Bool,
+    default=True,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 # Controls whether uptime monitoring automatically detects hostnames from error events.
 register(
     "uptime.automatic-hostname-detection",

src/sentry/uptime/config_drift.py

@@ -20,6 +20,8 @@
 SUBSCRIPTION_ID_PREFIX_BUCKETS = 256
 # Bounds the IN list when a partition holds many configs.
 IN_CHUNK_SIZE = 1000
+# Refreshed by every run of a pass; if it lapses, the pass restarts from the first missing id.
+REPAIR_CURSOR_TTL = 3600
 
 
 @dataclass(frozen=True)
@@ -78,7 +80,7 @@ def _live_region_rows() -> QuerySet[UptimeSubscriptionRegion]:
 
 
 def find_missing_configs(store: ConfigStore, subscription_id_prefix: str) -> DriftResult:
-    cluster = redis.redis_clusters.get_binary(store.cluster)
+    cluster = redis.redis_clusters.get(store.cluster)
     pipe = cluster.pipeline()
     subscription_ids: list[str] = []
     # One row per (subscription, region); slugs sharing a store must be checked once.
@@ -103,10 +105,8 @@ def find_missing_configs(store: ConfigStore, subscription_id_prefix: str) -> Dri
 
 
 def find_orphaned_configs(store: ConfigStore, partition: int) -> DriftResult:
-    cluster = redis.redis_clusters.get_binary(store.cluster)
-    stored = {
-        field.decode() for field in cluster.hkeys(get_config_key(store.key_prefix, partition))
-    }
+    cluster = redis.redis_clusters.get(store.cluster)
+    stored = set(cluster.hkeys(get_config_key(store.key_prefix, partition)))
     live: set[str | None] = set()
     for chunk in batched(stored, IN_CHUNK_SIZE):
         live.update(
@@ -118,3 +118,53 @@ def find_orphaned_configs(store: ConfigStore, partition: int) -> DriftResult:
             .values_list("uptime_subscription__subscription_id", flat=True)
         )
     return DriftResult(checked=len(stored), drifted_ids=frozenset(stored - live))
+
+
+def find_missing_configs_for_store(store: ConfigStore) -> DriftResult:
+    """
+    Whole-store form of find_missing_configs: reads every partition's field names in one
+    pipeline instead of probing one id at a time, since here every row is checked anyway.
+    """
+    # Postgres first: a row ACTIVE now had its config written before now, so a config
+    # published between the two reads can't show up as missing.
+    rows = (
+        _active_region_rows()
+        .filter(region_slug__in=store.region_slugs)
+        .values_list("uptime_subscription__subscription_id", flat=True)
+    )
+    live = {subscription_id for subscription_id in rows if subscription_id is not None}
+    cluster = redis.redis_clusters.get(store.cluster)
+    pipe = cluster.pipeline()
+    for partition in range(settings.UPTIME_CONFIG_PARTITIONS):
+        pipe.hkeys(get_config_key(store.key_prefix, partition))
+    stored = {field for fields in pipe.execute() for field in fields}
+    return DriftResult(checked=len(live), drifted_ids=frozenset(live - stored))
+
+
+def get_sentinel_key(key_prefix: str, partition: int) -> str:
+    # Redis hash tag: lands in the config key's slot, so a node that loses it loses this too.
+    return f"{{{get_config_key(key_prefix, partition)}}}:sentinel"
+
+
+def get_repair_cursor_key(key_prefix: str) -> str:
+    return f"{key_prefix}uptime:configs:repair-cursor"
+
+
+def clear_repair_cursor(store: ConfigStore) -> None:
+    redis.redis_clusters.get(store.cluster).delete(get_repair_cursor_key(store.key_prefix))
+
+
+def find_missing_sentinels(store: ConfigStore) -> list[int]:
+    cluster = redis.redis_clusters.get(store.cluster)
+    pipe = cluster.pipeline()
+    for partition in range(settings.UPTIME_CONFIG_PARTITIONS):
+        pipe.exists(get_sentinel_key(store.key_prefix, partition))
+    return [partition for partition, present in enumerate(pipe.execute()) if not present]
+
+
+def write_sentinels(store: ConfigStore) -> None:
+    cluster = redis.redis_clusters.get(store.cluster)
+    pipe = cluster.pipeline()
+    for partition in range(settings.UPTIME_CONFIG_PARTITIONS):
+        pipe.set(get_sentinel_key(store.key_prefix, partition), "1")
+    pipe.execute()

src/sentry/uptime/subscriptions/tasks.py

@@ -15,13 +15,19 @@
 from sentry.tasks.base import instrumented_task
 from sentry.taskworker.namespaces import uptime_tasks
 from sentry.uptime.config_drift import (
+    REPAIR_CURSOR_TTL,
     SUBSCRIPTION_ID_PREFIX_BUCKETS,
     SWEEP_RUN_INTERVAL,
     ConfigStore,
+    clear_repair_cursor,
     find_missing_configs,
+    find_missing_configs_for_store,
+    find_missing_sentinels,
     find_orphaned_configs,
     get_config_stores,
+    get_repair_cursor_key,
     sweep_slice,
+    write_sentinels,
 )
 from sentry.uptime.config_producer import produce_config, produce_config_removal
 from sentry.uptime.models import (
@@ -30,7 +36,7 @@
     UptimeSubscriptionRegion,
 )
 from sentry.uptime.types import CheckConfig
-from sentry.utils import metrics
+from sentry.utils import metrics, redis
 from sentry.utils.audit import create_system_audit_entry
 from sentry.utils.query import RangeQuerySetWrapper
 
@@ -371,6 +377,88 @@ def check_orphaned_configs(cluster: str, key_prefix: str, partition: int, **kwar
         )
 
 
+@instrumented_task(
+    name="sentry.uptime.tasks.check_config_sentinels",
+    namespace=uptime_tasks,
+    processing_deadline_duration=60,
+)
+def check_config_sentinels(**kwargs):
+    """
+    Checks each config store's partition sentinels every minute; a missing sentinel means the
+    store lost data, so its whole-store comparison is queued unless sentinel repair is disabled.
+    """
+    if not options.get("uptime.config-drift.enabled"):
+        return
+
+    for store in get_config_stores():
+        # The store that failed is likely the one being lost; keep checking the others.
+        try:
+            missing = find_missing_sentinels(store)
+        except Exception:
+            logger.exception("uptime.config_drift.sentinel_check_failed")
+            continue
+        metrics.gauge(
+            "uptime.config_drift.sentinel_missing",
+            len(missing),
+            tags={"cluster": store.cluster},
+            sample_rate=1.0,
+        )
+        if not missing:
+            # No pass runs while every sentinel is present, so a cursor here was left by an
+            # overlapping run or an old worker and must not carry over into the next loss.
+            try:
+                clear_repair_cursor(store)
+            except Exception:
+                logger.exception("uptime.config_drift.repair_cursor_clear_failed")
+        if missing and not options.get("uptime.config-drift.sentinel-repair-disabled"):
+            logger.warning(
+                "uptime.config_drift.sentinel_missing",
+                extra={"cluster": store.cluster, "count": len(missing), "partitions": missing},
+            )
+            repair_config_store.delay(cluster=store.cluster, key_prefix=store.key_prefix)
+
+
+@instrumented_task(
+    name="sentry.uptime.tasks.repair_config_store",
+    namespace=uptime_tasks,
+    processing_deadline_duration=60,
+    expires=60,
+)
+def repair_config_store(cluster: str, key_prefix: str, **kwargs):
+    """
+    One step of a pass over the store's missing configs after a sentinel went missing.
+    """
+    store = _find_store(cluster, key_prefix)
+    if store is None:
+        return
+
+    client = redis.redis_clusters.get(store.cluster)
+    cursor_key = get_repair_cursor_key(store.key_prefix)
+    cursor = client.get(cursor_key) or ""
+    result = find_missing_configs_for_store(store)
+    # Each missing id is queued once per pass, so ids that can't be published neither hold
+    # the sentinels back nor crowd out the ids after them.
+    pending = sorted(i for i in result.drifted_ids if i > cursor)
+    batch = pending[:CONFIG_REPAIR_MAX_TASKS]
+    logger.info(
+        "uptime.config_drift.store_repair",
+        extra={
+            "cluster": store.cluster,
+            "checked": result.checked,
+            "missing": len(result.drifted_ids),
+            "pending": len(pending),
+        },
+    )
+    if batch:
+        repair_missing_configs(store, batch, limit=CONFIG_REPAIR_MAX_TASKS)
+    if len(pending) <= CONFIG_REPAIR_MAX_TASKS:
+        # Cleared first, so a failed sentinel write restarts the pass rather than skipping ids.
+        client.delete(cursor_key)
+        write_sentinels(store)
+    else:
+        client.set(cursor_key, batch[-1], ex=REPAIR_CURSOR_TTL)
+
+
 def repair_missing_configs(
     store: ConfigStore, subscription_ids: Collection[str], *, limit: int = CONFIG_REPAIR_MAX_TASKS
 ) -> int:

tests/sentry/uptime/test_config_drift.py

@@ -8,11 +8,13 @@
 from django.db import connections, router
 from django.test import override_settings
 from django.test.utils import CaptureQueriesContext
+from rediscluster.nodemanager import NodeManager
 
 from sentry.conf.types.uptime import UptimeRegionConfig
 from sentry.testutils.cases import UptimeTestCase
 from sentry.testutils.helpers import override_options
 from sentry.testutils.helpers.datetime import freeze_time
+from sentry.uptime.config_drift import get_repair_cursor_key, get_sentinel_key
 from sentry.uptime.config_producer import (
     get_config_key,
     get_partition_from_subscription_id,
@@ -21,9 +23,11 @@
 from sentry.uptime.models import UptimeSubscription, UptimeSubscriptionRegion
 from sentry.uptime.subscriptions import tasks
 from sentry.uptime.subscriptions.tasks import (
+    check_config_sentinels,
     check_missing_configs,
     check_orphaned_configs,
     config_drift_dispatcher,
+    repair_config_store,
     update_remote_uptime_subscription,
     uptime_subscription_to_check_config,
 )
@@ -342,3 +346,129 @@ def test_tasks_write_nothing_to_postgres(self) -> None:
             if q["sql"].startswith(("INSERT", "UPDATE", "DELETE"))
         ]
         assert writes == []
+
+
+def test_sentinel_shares_slot_with_partition_hash() -> None:
+    # Offline slot math; the node is never contacted.
+    keyslot = NodeManager(startup_nodes=[{"host": "localhost", "port": 1}]).keyslot
+    for key_prefix in ("", "a"):
+        for partition in range(128):
+            assert keyslot(get_sentinel_key(key_prefix, partition)) == keyslot(
+                get_config_key(key_prefix, partition)
+            )
+
+
+@override_settings(UPTIME_REGIONS=REGIONS)
+class CheckConfigSentinelsTest(ConfigPusherTestMixin):
+    def setUp(self) -> None:
+        super().setUp()
+        self.enterContext(override_options({"uptime.config-drift.enabled": True}))
+
+    def test_sentinel_repair_disabled_emits_metric_only(self) -> None:
+        cluster = redis.redis_clusters.get_binary("default")
+        keys_before = set(cluster.keys())
+
+        with (
+            mock.patch.object(tasks, "metrics") as metrics,
+            mock.patch.object(repair_config_store, "delay") as delay,
+        ):
+            check_config_sentinels()
+
+        # One gauge per store, both on the test cluster.
+        all_missing = mock.call(
+            "uptime.config_drift.sentinel_missing",
+            128,
+            tags={"cluster": "default"},
+            sample_rate=1.0,
+        )
+        assert metrics.gauge.mock_calls == [all_missing, all_missing]
+        assert metrics.incr.mock_calls == []
+        assert not delay.called
+        assert set(cluster.keys()) == keys_before
+
+    @override_options({"uptime.config-drift.sentinel-repair-disabled": False})
+    def test_sentinel_write_touches_no_config_hash(self) -> None:
+        cluster = redis.redis_clusters.get_binary("default")
+
+        with self.tasks():
+            check_config_sentinels()
+
+        for key_prefix in "ab":
+            for partition in range(128):
+                assert cluster.type(get_sentinel_key(key_prefix, partition)) == b"string"
+                assert not cluster.exists(get_config_key(key_prefix, partition))
+                assert not cluster.exists(f"{key_prefix}uptime:updates:{partition}")
+
+    def _seed_lost_on_b(self, count: int = 1) -> tuple[list[UptimeSubscription], str]:
+        """
+        Writes every sentinel, then creates ``count`` subscriptions never published to store B
+        and drops one of their partitions' sentinel there. Returns them and the lost sentinel key.
+        """
+        with self.tasks():
+            check_config_sentinels()
+        subscriptions = []
+        for _ in range(count):
+            subscription = self.create_uptime_subscription(
+                subscription_id=_subscription_id(), region_slugs=["a1", "b1"]
+            )
+            _publish(subscription, ["a1"])
+            subscriptions.append(subscription)
+        assert subscription.subscription_id is not None
+        partition = get_partition_from_subscription_id(UUID(subscription.subscription_id))
+        sentinel = get_sentinel_key("b", partition)
+        redis.redis_clusters.get_binary("default").delete(sentinel)
+        return subscriptions, sentinel
+
+    @override_options({"uptime.config-drift.sentinel-repair-disabled": False})
+    def test_missing_sentinel_repairs_only_that_store(self) -> None:
+        [subscription], _ = self._seed_lost_on_b()
+
+        with mock.patch.object(repair_config_store, "delay") as delay:
+            check_config_sentinels()
+        delay.assert_called_once_with(cluster="default", key_prefix="b")
+
+        with self.tasks():
+            repair_config_store(cluster="default", key_prefix="b")
+
+        self.assert_redis_config(
+            "b1", subscription, "upsert", UptimeSubscriptionRegion.RegionMode.ACTIVE
+        )
+
+    @override_options({"uptime.config-drift.sentinel-repair-disabled": False})
+    @mock.patch.object(tasks, "CONFIG_REPAIR_MAX_TASKS", 1)
+    def test_unpublishable_configs_end_the_pass(self) -> None:
+        lost, sentinel = self._seed_lost_on_b(count=2)
+        first, second = sorted(lost, key=lambda subscription: subscription.subscription_id or "")
+        cluster = redis.redis_clusters.get_binary("default")
+
+        # The republishes never land, so both stay missing on every run.
+        with mock.patch.object(update_remote_uptime_subscription, "delay") as delay:
+            repair_config_store(cluster="default", key_prefix="b")
+            assert not cluster.exists(sentinel)
+            repair_config_store(cluster="default", key_prefix="b")
+
+        assert delay.call_args_list == [
+            mock.call(uptime_subscription_id=first.id, region_slugs=["b1"]),
+            mock.call(uptime_subscription_id=second.id, region_slugs=["b1"]),
+        ]
+        assert cluster.exists(sentinel)
+
+        with mock.patch.object(repair_config_store, "delay") as repair:
+            check_config_sentinels()
+        assert not repair.called
+
+    @override_options({"uptime.config-drift.sentinel-repair-disabled": False})
+    def test_leftover_cursor_does_not_skip_the_next_loss(self) -> None:
+        with self.tasks():
+            check_config_sentinels()
+        # Left by an overlapping run after the sentinels came back, past every id.
+        redis.redis_clusters.get("default").set(get_repair_cursor_key("b"), "f" * 32, ex=60)
+
+        # Its first check sees every sentinel present, which is where the cursor is cleared.
+        [subscription], _ = self._seed_lost_on_b()
+        with self.tasks():
+            check_config_sentinels()
+
+        self.assert_redis_config(
+            "b1", subscription, "upsert", UptimeSubscriptionRegion.RegionMode.ACTIVE
+        )