feat(action-log): Route recovery through dedicated outbox

src/sentry/conf/server.py

@@ -916,6 +916,7 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.integrations.tasks.update_comment",
     "sentry.integrations.vsts.tasks.kickoff_subscription_check",
     "sentry.integrations.vsts.tasks.subscription_check",
+    "sentry.issues.action_log.tasks",
     "sentry.issues.derived.tasks",
     "sentry.issues.escalating.forecasts",
     "sentry.middleware.integrations.tasks",
@@ -1076,6 +1077,10 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
         "task": "hybridcloud:sentry.tasks.enqueue_outbox_jobs",
         "schedule": crontab("*/1", "*", "*", "*", "*"),
     },
+    "deliver-group-action-log-outbox": {
+        "task": "issues.action_log:sentry.issues.action_log.tasks.enqueue_group_action_log_outbox_jobs",
+        "schedule": crontab("*/1", "*", "*", "*", "*"),
+    },
     "update-user-reports": {
         "task": "issues:sentry.feedback.tasks.update_user_reports",
         "schedule": crontab("*/15", "*", "*", "*", "*"),

src/sentry/hybridcloud/tasks/deliver_from_outbox.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import math
+from collections.abc import Mapping
 from typing import Any
 
 import sentry_sdk
@@ -73,56 +74,15 @@ def schedule_batch(
 ) -> None:
     scheduled_count = 0
 
-    if not concurrency:
-        concurrency = CONCURRENCY
     try:
         for outbox_name in settings.SENTRY_OUTBOX_MODELS[silo_mode.name]:
             outbox_model: type[OutboxBase] = OutboxBase.from_outbox_name(outbox_name)
-
-            aggregates = outbox_model.objects.all().aggregate(Min("id"), Max("id"))
-
-            lo = aggregates["id__min"] or 0
-            hi = aggregates["id__max"] or -1
-            if hi < lo:
-                continue
-
-            scheduled_count += hi - lo + 1
-            batch_size = math.ceil((hi - lo + 1) / concurrency)
-
-            metrics_tags = dict(silo_mode=silo_mode.name, outbox_name=outbox_name)
-            metrics.gauge(
-                "deliver_from_outbox.queued_batch_size",
-                value=batch_size,
-                tags=metrics_tags,
-                sample_rate=1.0,
-            )
-
-            # Notably, when l and h are close, this will result in creating tasks that are processing future ids --
-            # that's totally fine.
-            for i in range(concurrency):
-                drain_task.delay(
-                    outbox_name=outbox_name,
-                    outbox_identifier_low=lo + i * batch_size,
-                    outbox_identifier_hi=lo + (i + 1) * batch_size,
-                )
-
-            deepest_shard_information = outbox_model.get_shard_depths_descending(limit=1)
-            max_shard_depth = (
-                float(deepest_shard_information[0]["depth"]) if deepest_shard_information else 0.0
-            )
-            metrics.gauge(
-                "deliver_from_outbox.maximum_shard_depth",
-                value=max_shard_depth,
-                tags=metrics_tags,
-                sample_rate=1.0,
-            )
-
-            outbox_count = outbox_model.get_total_outbox_count()
-            metrics.gauge(
-                "deliver_from_outbox.total_outbox_count",
-                value=outbox_count,
-                tags=metrics_tags,
-                sample_rate=1.0,
+            scheduled_count += schedule_outbox_model(
+                silo_mode=silo_mode,
+                outbox_model=outbox_model,
+                drain_task=drain_task,
+                concurrency=concurrency,
+                drain_task_kwargs={"outbox_name": outbox_name},
             )
         if process_outbox_backfills:
             backfill_outboxes_for(silo_mode, scheduled_count)
@@ -132,6 +92,63 @@ def schedule_batch(
         raise
 
 
+def schedule_outbox_model(
+    *,
+    silo_mode: SiloMode,
+    outbox_model: type[OutboxBase],
+    drain_task: Task[Any, Any],
+    concurrency: int | None = None,
+    drain_task_kwargs: Mapping[str, Any] | None = None,
+) -> int:
+    if not concurrency:
+        concurrency = CONCURRENCY
+
+    id_range = outbox_model.objects.all().aggregate(Min("id"), Max("id"))
+    identifier_low = id_range["id__min"] or 0
+    identifier_high = id_range["id__max"] or -1
+    if identifier_high < identifier_low:
+        return 0
+
+    scheduled_count = identifier_high - identifier_low + 1
+    batch_size = math.ceil(scheduled_count / concurrency)
+    metrics_tags = dict(silo_mode=silo_mode.name, outbox_name=outbox_model._meta.label)
+    metrics.gauge(
+        "deliver_from_outbox.queued_batch_size",
+        value=batch_size,
+        tags=metrics_tags,
+        sample_rate=1.0,
+    )
+
+    # Notably, when low and high are close, some tasks process future ids. That's fine.
+    task_kwargs = drain_task_kwargs or {}
+    for i in range(concurrency):
+        drain_task.delay(
+            outbox_identifier_low=identifier_low + i * batch_size,
+            outbox_identifier_hi=identifier_low + (i + 1) * batch_size,
+            **task_kwargs,
+        )
+
+    deepest_shard_information = outbox_model.get_shard_depths_descending(limit=1)
+    max_shard_depth = (
+        float(deepest_shard_information[0]["depth"]) if deepest_shard_information else 0.0
+    )
+    metrics.gauge(
+        "deliver_from_outbox.maximum_shard_depth",
+        value=max_shard_depth,
+        tags=metrics_tags,
+        sample_rate=1.0,
+    )
+
+    outbox_count = outbox_model.get_total_outbox_count()
+    metrics.gauge(
+        "deliver_from_outbox.total_outbox_count",
+        value=outbox_count,
+        tags=metrics_tags,
+        sample_rate=1.0,
+    )
+    return scheduled_count
+
+
 @instrumented_task(
     name="sentry.tasks.drain_outbox_shards",
     namespace=hybridcloud_tasks,

src/sentry/issues/action_log/publish.py

@@ -34,7 +34,7 @@
 
 # Group Action Log — tracks who did what to an issue and how.
 #
-# publish_action() writes a CellOutbox entry; the outbox receiver creates the
+# publish_action() writes an outbox entry; the outbox receiver creates the
 # GroupActionLogEntry on the (eventually separate) grouplog database and kicks
 # off derived-data processing.
 #
@@ -102,9 +102,10 @@ def publish_action(
     # load time so it can be imported from models without creating cycles.
     from django.db import router, transaction
 
-    from sentry import features
+    from sentry import features, options
     from sentry.hybridcloud.models.outbox import CellOutbox, outbox_context
     from sentry.hybridcloud.outbox.category import OutboxCategory, OutboxScope
+    from sentry.issues.models.groupactionlogoutbox import GroupActionLogOutbox
     from sentry.utils import metrics
 
     for callback in _publish_callbacks.get():
@@ -141,6 +142,13 @@ def publish_action(
     if not write_to_db:
         return
 
+    use_dedicated_outbox = options.get("issues.action_log.use_dedicated_outbox")
+    outbox_model = GroupActionLogOutbox if use_dedicated_outbox else CellOutbox
+    metrics.incr(
+        "issues.action_log.outbox_write",
+        tags={"route": "dedicated" if use_dedicated_outbox else "shared"},
+    )
+
     payload: GroupActionLogPayload = {
         "group_id": group_id,
         "project_id": project.id,
@@ -155,15 +163,15 @@ def publish_action(
     if idempotency_key is not None:
         payload["idempotency_key"] = idempotency_key
 
-    outbox = CellOutbox(
+    outbox = outbox_model(
         shard_scope=OutboxScope.GROUP_SCOPE,
         shard_identifier=group_id,
         category=OutboxCategory.GROUP_ACTION_LOG_EVENT,
-        object_identifier=CellOutbox.next_object_identifier(),
+        object_identifier=outbox_model.next_object_identifier(),
         payload=payload,
     )
     # Flush on commit by default; callers can wrap in outbox_context(flush=False) to defer.
-    with outbox_context(transaction.atomic(router.db_for_write(CellOutbox))):
+    with outbox_context(transaction.atomic(router.db_for_write(outbox_model))):
         outbox.save()
 
 

src/sentry/issues/action_log/tasks.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import Any
+
+import sentry_sdk
+
+from sentry.hybridcloud.tasks.deliver_from_outbox import (
+    process_outbox_batch,
+    schedule_outbox_model,
+)
+from sentry.issues.models.groupactionlogoutbox import GroupActionLogOutbox
+from sentry.silo.base import SiloMode
+from sentry.tasks.base import instrumented_task
+from sentry.taskworker.namespaces import issues_action_log_tasks
+
+
+@instrumented_task(
+    name="sentry.issues.action_log.tasks.enqueue_group_action_log_outbox_jobs",
+    namespace=issues_action_log_tasks,
+    silo_mode=SiloMode.CELL,
+    processing_deadline_duration=30,
+)
+def enqueue_group_action_log_outbox_jobs(concurrency: int | None = None, **kwargs: Any) -> None:
+    try:
+        schedule_outbox_model(
+            silo_mode=SiloMode.CELL,
+            outbox_model=GroupActionLogOutbox,
+            drain_task=drain_group_action_log_outbox_shards,
+            concurrency=concurrency,
+        )
+    except Exception:
+        sentry_sdk.capture_exception()
+        raise
+
+
+@instrumented_task(
+    name="sentry.issues.action_log.tasks.drain_group_action_log_outbox_shards",
+    namespace=issues_action_log_tasks,
+    silo_mode=SiloMode.CELL,
+    processing_deadline_duration=90,
+)
+def drain_group_action_log_outbox_shards(
+    outbox_identifier_low: int = 0,
+    outbox_identifier_hi: int = 0,
+) -> None:
+    try:
+        process_outbox_batch(
+            outbox_identifier_hi=outbox_identifier_hi,
+            outbox_identifier_low=outbox_identifier_low,
+            outbox_model=GroupActionLogOutbox,
+        )
+    except Exception:
+        sentry_sdk.capture_exception()
+        raise

src/sentry/options/defaults.py

@@ -1315,6 +1315,11 @@
     default=0,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+register(
+    "issues.action_log.use_dedicated_outbox",
+    default=False,
+    flags=FLAG_MODIFIABLE_BOOL | FLAG_AUTOMATOR_MODIFIABLE,
+)
 register(
     "issues.backfill_group_action_log.killswitch",
     type=Bool,

src/sentry/taskworker/namespaces.py

@@ -166,6 +166,11 @@
     app_feature="issueplatform",
 )
 
+issues_action_log_tasks = app.taskregistry.create_namespace(
+    "issues.action_log",
+    app_feature="issueplatform",
+)
+
 issues_merge_tasks = app.taskregistry.create_namespace(
     "issues.merge",
     app_feature="issueplatform",

src/sentry/testutils/outbox.py

@@ -13,6 +13,8 @@
     enqueue_outbox_jobs,
     enqueue_outbox_jobs_control,
 )
+from sentry.issues.action_log.tasks import enqueue_group_action_log_outbox_jobs
+from sentry.issues.models.groupactionlogoutbox import GroupActionLogOutbox
 from sentry.silo.base import SiloMode
 from sentry.testutils.silo import assume_test_silo_mode
 
@@ -41,16 +43,20 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
     yield
     from sentry.testutils.helpers.task_runner import TaskRunner
 
+    outbox_models = [
+        OutboxBase.from_outbox_name(outbox_name)
+        for outbox_names in settings.SENTRY_OUTBOX_MODELS.values()
+        for outbox_name in outbox_names
+    ]
+    outbox_models.append(GroupActionLogOutbox)
+
     with TaskRunner(), assume_test_silo_mode(SiloMode.MONOLITH):
-        for i in range(10):
+        for _ in range(10):
             enqueue_outbox_jobs(concurrency=1, process_outbox_backfills=False)
             enqueue_outbox_jobs_control(concurrency=1, process_outbox_backfills=False)
+            enqueue_group_action_log_outbox_jobs(concurrency=1)
 
-            if not any(
-                OutboxBase.from_outbox_name(outbox_name).find_scheduled_shards()
-                for outbox_names in settings.SENTRY_OUTBOX_MODELS.values()
-                for outbox_name in outbox_names
-            ):
+            if not any(outbox_model.find_scheduled_shards() for outbox_model in outbox_models):
                 break
         else:
             raise OutboxRecursionLimitError