fix(hybrid-cloud): Adds support for cross database tombstones (#69480)

 Adds cross-database tombstone deletion support

The current tombstone code performs a join between the model and
tombstone tables to construct a minimal set of IDs to update or delete
from the target table.

Because this isn't possible when the target model lives in a different
database (e.g. crons monitors), we have to take a 2 step approach:
1. Query the tombstone table for the current IDs list matching the
watermarking batch.
2. Query the model table for all rows with a matching identifier on the
target field

There are possible performance concerns with the model-side query, so
I've placed this behind a new option named
`hybrid_cloud.allow_cross_db_tombstones`. If the flag is disabled and a
cross db tombstone cleanup is attempted, an explicit exception is raised
noting why the cleanup failed.

Author: GabeVillalobos

src/sentry/options/defaults.py

@@ -1726,6 +1726,7 @@
 register(
     "hybrid_cloud.disable_relative_upload_urls", default=False, flags=FLAG_AUTOMATOR_MODIFIABLE
 )
+register("hybrid_cloud.allow_cross_db_tombstones", default=False, flags=FLAG_AUTOMATOR_MODIFIABLE)
 
 # Retry controls
 register("hybridcloud.regionsiloclient.retries", default=5, flags=FLAG_AUTOMATOR_MODIFIABLE)

src/sentry/tasks/deletion/hybrid_cloud.py

@@ -22,6 +22,8 @@
 from django.db.models.manager import BaseManager
 from django.utils import timezone
 
+from sentry import options
+from sentry.db.models import Model
 from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
 from sentry.models.tombstone import TombstoneBase
 from sentry.silo.base import SiloMode
@@ -247,30 +249,14 @@ def _process_tombstone_reconciliation(
         prefix, field, watermark_manager, batch_size=get_batch_size()
     )
     has_more = watermark_batch.has_more
-    to_delete_ids: list[int] = []
-
     if watermark_batch.low < watermark_batch.up:
-        oldest_seen: datetime.datetime = timezone.now()
-
-        with connections[router.db_for_read(model)].cursor() as conn:
-            conn.execute(
-                f"""
-                SELECT r.id, t.created_at
-                FROM {model._meta.db_table} r
-                JOIN {tombstone_cls._meta.db_table} t
-                    ON t.table_name = %(table_name)s AND t.object_identifier = r.{field.name}
-                WHERE {watermark_target}.id > %(low)s AND {watermark_target}.id <= %(up)s
-            """,
-                {
-                    "table_name": field.foreign_table_name,
-                    "low": watermark_batch.low,
-                    "up": watermark_batch.up,
-                },
-            )
-
-            for (row_id, tomb_created) in conn.fetchall():
-                to_delete_ids.append(row_id)
-                oldest_seen = min(oldest_seen, tomb_created)
+        to_delete_ids, oldest_seen = _get_model_ids_for_tombstone_cascade(
+            tombstone_cls=tombstone_cls,
+            model=model,
+            field=field,
+            watermark_batch=watermark_batch,
+            watermark_target=watermark_target,
+        )
 
         if field.on_delete == "CASCADE":
             task = deletions.get(
@@ -306,3 +292,83 @@ def _process_tombstone_reconciliation(
         )
 
     return has_more
+
+
+def _get_model_ids_for_tombstone_cascade(
+    tombstone_cls: type[TombstoneBase],
+    model: type[Model],
+    field: HybridCloudForeignKey,
+    watermark_target,
+    watermark_batch: WatermarkBatch,
+) -> tuple[list[int], datetime.datetime]:
+    """
+    Queries the database or databases if spanning multiple, and returns
+     a tuple with a list of row IDs to delete, and the oldest
+     tombstone timestamp for the batch.
+
+    :param tombstone_cls: Either a RegionTombstone or ControlTombstone, depending on
+     which silo the tombstone process is running.
+    :param model: The model with a HybridCloudForeignKey.
+    :param field: The HybridCloudForeignKey field from the model to process.
+    :return:
+    """
+
+    to_delete_ids = []
+    oldest_seen = timezone.now()
+    tombstone_and_model_in_same_db = router.db_for_read(model) == router.db_for_read(tombstone_cls)
+
+    if tombstone_and_model_in_same_db:
+        with connections[router.db_for_read(model)].cursor() as conn:
+            conn.execute(
+                f"""
+                    SELECT r.id, t.created_at
+                    FROM {model._meta.db_table} r
+                    JOIN {tombstone_cls._meta.db_table} t
+                        ON t.table_name = %(table_name)s AND t.object_identifier = r.{field.name}
+                    WHERE {watermark_target}.id > %(low)s AND {watermark_target}.id <= %(up)s
+                """,
+                {
+                    "table_name": field.foreign_table_name,
+                    "low": watermark_batch.low,
+                    "up": watermark_batch.up,
+                },
+            )
+
+            for (row_id, tomb_created) in conn.fetchall():
+                to_delete_ids.append(row_id)
+                oldest_seen = min(oldest_seen, tomb_created)
+
+            return to_delete_ids, oldest_seen
+
+    return get_ids_for_tombstone_cascade_cross_db(
+        tombstone_cls=tombstone_cls, model=model, field=field, watermark_batch=watermark_batch
+    )
+
+
+def get_ids_for_tombstone_cascade_cross_db(
+    tombstone_cls: type[TombstoneBase],
+    model: type[Model],
+    field: HybridCloudForeignKey,
+    watermark_batch: WatermarkBatch,
+) -> tuple[list[int], datetime.datetime]:
+    if not options.get("hybrid_cloud.allow_cross_db_tombstones"):
+        raise Exception("Cannot process tombstones due to model living in separate database.")
+
+    oldest_seen = timezone.now()
+    # Because tombstones can span multiple databases, we can't always rely on
+    # the join code above. Instead, we have to manually query the tombstone IDs
+    # first, then query the model for matching rows.
+    tombstone_entries = tombstone_cls.objects.filter(
+        id__lte=watermark_batch.up, id__gt=watermark_batch.low, table_name=field.foreign_table_name
+    ).values_list("object_identifier", "created_at")
+
+    ids_to_check = []
+    for object_id, created_at in tombstone_entries:
+        ids_to_check.append(object_id)
+        oldest_seen = min(oldest_seen, created_at)
+
+    field_name = f"{field.name}__in"
+    query_kwargs = {field_name: ids_to_check}
+    affected_rows = list(model.objects.filter(**query_kwargs).values_list("id", flat=True))
+
+    return affected_rows, oldest_seen

tests/sentry/tasks/deletion/test_hybrid_cloud.py

@@ -12,19 +12,29 @@
 from sentry.models.integrations.integration import Integration
 from sentry.models.outbox import ControlOutbox, OutboxScope, outbox_context
 from sentry.models.savedsearch import SavedSearch
+from sentry.models.tombstone import RegionTombstone
 from sentry.models.user import User
+from sentry.monitors.models import Monitor
 from sentry.silo.base import SiloMode
 from sentry.tasks.deletion.hybrid_cloud import (
+    WatermarkBatch,
+    get_ids_for_tombstone_cascade_cross_db,
     get_watermark,
     schedule_hybrid_cloud_foreign_key_jobs,
     schedule_hybrid_cloud_foreign_key_jobs_control,
     set_watermark,
 )
 from sentry.testutils.factories import Factories
+from sentry.testutils.helpers import override_options
 from sentry.testutils.helpers.task_runner import BurstTaskRunner
 from sentry.testutils.outbox import outbox_runner
 from sentry.testutils.pytest.fixtures import django_db_all
-from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
+from sentry.testutils.silo import (
+    assume_test_silo_mode,
+    assume_test_silo_mode_of,
+    control_silo_test,
+    region_silo_test,
+)
 from sentry.types.region import find_regions_for_user
 
 
@@ -274,3 +284,183 @@ def test_set_null_deletion_behavior(task_runner):
     # Deletion set field to null
     saved_query = DiscoverSavedQuery.objects.get(id=saved_query.id)
     assert saved_query.created_by_id is None
+
+
+def setup_cross_db_deletion_data():
+    user = Factories.create_user()
+    organization = Factories.create_organization(owner=user, name="Delete Me")
+    project = Factories.create_project(organization=organization)
+    group = Factories.create_group(project=project)
+    with assume_test_silo_mode_of(DiscoverSavedQuery, Monitor):
+        saved_query = DiscoverSavedQuery.objects.create(
+            name="disco-query",
+            organization=organization,
+            created_by_id=user.id,
+        )
+        monitor = Monitor.objects.create(
+            organization_id=organization.id,
+            project_id=project.id,
+            slug="test-monitor",
+            name="Test Monitor",
+            owner_user_id=user.id,
+        )
+
+    return dict(
+        user=user,
+        organization=organization,
+        project=project,
+        monitor=monitor,
+        group=group,
+        saved_query=saved_query,
+    )
+
+
+@django_db_all
+@region_silo_test
+def test_get_ids_for_tombstone_cascade_cross_db(task_runner):
+    data = setup_cross_db_deletion_data()
+
+    unaffected_data = []
+    for i in range(3):
+        unaffected_data.append(setup_cross_db_deletion_data())
+
+    user = data["user"]
+    user_id = user.id
+    monitor = data["monitor"]
+    with assume_test_silo_mode_of(User), outbox_runner():
+        user.delete()
+
+    tombstone = RegionTombstone.objects.get(
+        object_identifier=user_id, table_name=User._meta.db_table
+    )
+
+    highest_tombstone_id = RegionTombstone.objects.aggregate(Max("id"))
+    monitor_owner_field = Monitor._meta.get_field("owner_user_id")
+
+    with override_options({"hybrid_cloud.allow_cross_db_tombstones": True}):
+        ids, oldest_obj = get_ids_for_tombstone_cascade_cross_db(
+            tombstone_cls=RegionTombstone,
+            model=Monitor,
+            field=monitor_owner_field,
+            watermark_batch=WatermarkBatch(
+                low=0,
+                up=highest_tombstone_id["id__max"] + 1,
+                has_more=False,
+                transaction_id="foobar",
+            ),
+        )
+        assert ids == [monitor.id]
+        assert oldest_obj == tombstone.created_at
+
+
+@django_db_all
+@region_silo_test
+def test_get_ids_for_tombstone_cascade_cross_db_watermark_bounds(task_runner):
+    cascade_data = []
+    for i in range(3):
+        cascade_data.append(setup_cross_db_deletion_data())
+
+    unaffected_data = []
+    for i in range(3):
+        unaffected_data.append(setup_cross_db_deletion_data())
+
+    in_order_tombstones = []
+    for data in cascade_data:
+        user = data["user"]
+        user_id = user.id
+        with assume_test_silo_mode_of(User), outbox_runner():
+            user.delete()
+
+        in_order_tombstones.append(
+            RegionTombstone.objects.get(object_identifier=user_id, table_name=User._meta.db_table)
+        )
+
+    bounds_with_expected_results = [
+        (
+            {"low": 0, "up": in_order_tombstones[1].id},
+            [cascade_data[0]["monitor"].id, cascade_data[1]["monitor"].id],
+        ),
+        (
+            {"low": in_order_tombstones[1].id, "up": in_order_tombstones[2].id},
+            [cascade_data[2]["monitor"].id],
+        ),
+        (
+            {"low": 0, "up": in_order_tombstones[0].id - 1},
+            [],
+        ),
+        (
+            {"low": in_order_tombstones[2].id + 1, "up": in_order_tombstones[2].id + 5},
+            [],
+        ),
+        (
+            {"low": -1, "up": in_order_tombstones[2].id + 1},
+            [
+                cascade_data[0]["monitor"].id,
+                cascade_data[1]["monitor"].id,
+                cascade_data[2]["monitor"].id,
+            ],
+        ),
+    ]
+
+    for bounds, bounds_with_expected_results in bounds_with_expected_results:
+        monitor_owner_field = Monitor._meta.get_field("owner_user_id")
+
+        with override_options({"hybrid_cloud.allow_cross_db_tombstones": True}):
+            ids, oldest_obj = get_ids_for_tombstone_cascade_cross_db(
+                tombstone_cls=RegionTombstone,
+                model=Monitor,
+                field=monitor_owner_field,
+                watermark_batch=WatermarkBatch(
+                    low=bounds["low"],
+                    up=bounds["up"],
+                    has_more=False,
+                    transaction_id="foobar",
+                ),
+            )
+            assert ids == bounds_with_expected_results
+
+
+@django_db_all
+@region_silo_test
+def test_get_ids_for_tombstone_cascade_cross_db_with_multiple_tombstone_types():
+    data = setup_cross_db_deletion_data()
+
+    unaffected_data = []
+    for i in range(3):
+        unaffected_data.append(setup_cross_db_deletion_data())
+
+    # Pollute the tombstone data with references to relationships in other
+    # tables matching other User IDs just to ensure we are filtering on the
+    # correct table name.
+    for udata in unaffected_data:
+        unaffected_user = udata["user"]
+        RegionTombstone.objects.create(
+            table_name="something_table", object_identifier=unaffected_user.id
+        )
+
+    user = data["user"]
+    user_id = user.id
+    monitor = data["monitor"]
+    with assume_test_silo_mode_of(User), outbox_runner():
+        user.delete()
+
+    tombstone = RegionTombstone.objects.get(
+        object_identifier=user_id, table_name=User._meta.db_table
+    )
+
+    highest_tombstone_id = RegionTombstone.objects.aggregate(Max("id"))
+
+    with override_options({"hybrid_cloud.allow_cross_db_tombstones": True}):
+        ids, oldest_obj = get_ids_for_tombstone_cascade_cross_db(
+            tombstone_cls=RegionTombstone,
+            model=Monitor,
+            field=Monitor._meta.get_field("owner_user_id"),
+            watermark_batch=WatermarkBatch(
+                low=0,
+                up=highest_tombstone_id["id__max"] + 1,
+                has_more=False,
+                transaction_id="foobar",
+            ),
+        )
+        assert ids == [monitor.id]
+        assert oldest_obj == tombstone.created_at