feat(aci): Move workflow engine scheduling to new task (#99535)

Creates a new task that (if options allow it) schedules batch workflow
processing.

Workflow scheduling and Delayed rule scheduling run on different redis
clusters with different client types and
thus some different constraints. We mostly don't want to touch rule
processing until we delete it, whereas workflow engine is under active
development. By splitting the scheduling, we can make changes for
workflow engine without needing to worry about maintaining or modifying
rules processing behavior.

The deployment/cutover plans is:
1. Deploy new task, verify it is running and doing nothing.
2. Set up a PR to set `delayed_workflow.rollout` to False and another
setting setting it to True and
`workflow_engine.use_process_pending_batch` to False.
3. Merge the first PR, then the second immediately after the first is
deployed. This should disable the workflow engine publishing, allow it
to finish, then re-enable it in the new task.

Author: kcons

src/sentry/conf/server.py

@@ -1208,6 +1208,12 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
         "schedule": crontab(minute="*/1"),
         "options": {"expires": 10, "queue": "buffers.process_pending_batch"},
     },
+    "flush-delayed-workflows": {
+        "task": "sentry.tasks.process_buffer.schedule_delayed_workflows",
+        # Run every 1 minute
+        "schedule": crontab(minute="*/1"),
+        "options": {"expires": 10, "queue": "workflow_engine.process_workflows"},
+    },
     "sync-options": {
         "task": "sentry.tasks.options.sync_options",
         # Run every 10 seconds
@@ -1634,6 +1640,10 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
         "task": "buffer:sentry.tasks.process_buffer.process_pending_batch",
         "schedule": task_crontab("*/1", "*", "*", "*", "*"),
     },
+    "flush-delayed-workflows": {
+        "task": "workflow_engine:sentry.tasks.process_buffer.schedule_delayed_workflows",
+        "schedule": task_crontab("*/1", "*", "*", "*", "*"),
+    },
     "sync-options": {
         "task": "options:sentry.tasks.options.sync_options",
         "schedule": timedelta(seconds=10),

src/sentry/options/defaults.py

@@ -3143,6 +3143,17 @@
     default=True,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+# Control whether delayed workflow engine evaluation is done
+# via the old process_pending_batch task with rules delayed processing, or
+# via the new schedule_delayed_workflows task.
+# NB: These tasks are allowed to run concurrently, so a naive switch here
+# may result in duplicate processing.
+register(
+    "workflow_engine.use_new_scheduling_task",
+    type=Bool,
+    default=False,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
 
 # Restrict uptime issue creation for specific host provider identifiers. Items
 # in this list map to the `host_provider_id` column in the UptimeSubscription

src/sentry/rules/processing/buffer_processing.py

@@ -159,45 +159,59 @@ def process_in_batches(
             )
 
 
-def process_buffer() -> None:
+def process_buffer_for_type(processing_type: str, handler: type[DelayedProcessingBase]) -> None:
+    """
+    Process buffers for a specific processing type and handler.
+    """
     should_emit_logs = options.get("delayed_processing.emit_logs")
 
-    for processing_type, handler in delayed_processing_registry.registrations.items():
-        if handler.option and not options.get(handler.option):
-            log_name = f"{processing_type}.disabled"
-            logger.info(log_name, extra={"option": handler.option})
-            continue
+    if handler.option and not options.get(handler.option):
+        log_name = f"{processing_type}.disabled"
+        logger.info(log_name, extra={"option": handler.option})
+        return
 
-        buffer = handler.buffer_backend()
-
-        with metrics.timer(f"{processing_type}.process_all_conditions.duration"):
-            # We need to use a very fresh timestamp here; project scores (timestamps) are
-            # updated with each relevant event, and some can be updated every few milliseconds.
-            # The staler this timestamp, the more likely it'll miss some recently updated projects,
-            # and the more likely we'll have frequently updated projects that are never actually
-            # retrieved and processed here.
-            fetch_time = datetime.now(tz=timezone.utc).timestamp()
-            buffer_keys = handler.get_buffer_keys()
-            all_project_ids_and_timestamps = buffer.bulk_get_sorted_set(
-                buffer_keys,
-                min=0,
-                max=fetch_time,
-            )
+    buffer = handler.buffer_backend()
+
+    with metrics.timer(f"{processing_type}.process_all_conditions.duration"):
+        # We need to use a very fresh timestamp here; project scores (timestamps) are
+        # updated with each relevant event, and some can be updated every few milliseconds.
+        # The staler this timestamp, the more likely it'll miss some recently updated projects,
+        # and the more likely we'll have frequently updated projects that are never actually
+        # retrieved and processed here.
+        fetch_time = datetime.now(tz=timezone.utc).timestamp()
+        buffer_keys = handler.get_buffer_keys()
+        all_project_ids_and_timestamps = buffer.bulk_get_sorted_set(
+            buffer_keys,
+            min=0,
+            max=fetch_time,
+        )
 
-            if should_emit_logs:
-                log_str = ", ".join(
-                    f"{project_id}: {timestamps}"
-                    for project_id, timestamps in all_project_ids_and_timestamps.items()
-                )
-                log_name = f"{processing_type}.project_id_list"
-                logger.info(log_name, extra={"project_ids": log_str})
-
-            project_ids = list(all_project_ids_and_timestamps.keys())
-            for project_id in project_ids:
-                process_in_batches(buffer, project_id, processing_type)
-
-            buffer.delete_keys(
-                buffer_keys,
-                min=0,
-                max=fetch_time,
+        if should_emit_logs:
+            log_str = ", ".join(
+                f"{project_id}: {timestamps}"
+                for project_id, timestamps in all_project_ids_and_timestamps.items()
             )
+            log_name = f"{processing_type}.project_id_list"
+            logger.info(log_name, extra={"project_ids": log_str})
+
+        project_ids = list(all_project_ids_and_timestamps.keys())
+        for project_id in project_ids:
+            process_in_batches(buffer, project_id, processing_type)
+
+        buffer.delete_keys(
+            buffer_keys,
+            min=0,
+            max=fetch_time,
+        )
+
+
+def process_buffer() -> None:
+    """
+    Process all registered delayed processing types.
+    """
+    for processing_type, handler in delayed_processing_registry.registrations.items():
+        # If the new scheduling task is enabled and this is delayed_workflow, skip it
+        use_new_scheduling = options.get("workflow_engine.use_new_scheduling_task")
+        if use_new_scheduling and processing_type == "delayed_workflow":
+            continue
+        process_buffer_for_type(processing_type, handler)

src/sentry/tasks/process_buffer.py

@@ -4,8 +4,10 @@
 import sentry_sdk
 from django.apps import apps
 
+from sentry import options
 from sentry.db.models.base import Model
 from sentry.tasks.base import instrumented_task
+from sentry.taskworker import namespaces
 from sentry.taskworker.config import TaskworkerConfig
 from sentry.taskworker.namespaces import buffer_tasks
 from sentry.utils.locking import UnableToAcquireLock
@@ -63,6 +65,37 @@ def process_pending_batch() -> None:
         logger.warning("process_pending_batch.fail", extra={"error": error})
 
 
+@instrumented_task(
+    name="sentry.tasks.process_buffer.schedule_delayed_workflows",
+    queue="workflow_engine.process_workflows",
+    taskworker_config=TaskworkerConfig(
+        namespace=namespaces.workflow_engine_tasks,
+        processing_deadline_duration=40,
+    ),
+)
+def schedule_delayed_workflows() -> None:
+    """
+    Schedule delayed workflow buffers in a batch.
+    """
+    from sentry.rules.processing.buffer_processing import process_buffer_for_type
+    from sentry.workflow_engine.tasks.delayed_workflows import DelayedWorkflow
+
+    lock = get_process_lock("schedule_delayed_workflows")
+
+    try:
+        with lock.acquire():
+            # Only process delayed_workflow type
+            use_new_scheduling = options.get("workflow_engine.use_new_scheduling_task")
+            if not use_new_scheduling:
+                logger.info(
+                    "Configured to use process_pending_batch for delayed_workflow; exiting."
+                )
+                return
+            process_buffer_for_type("delayed_workflow", DelayedWorkflow)
+    except UnableToAcquireLock as error:
+        logger.warning("schedule_delayed_workflows.fail", extra={"error": error})
+
+
 @instrumented_task(
     name="sentry.tasks.process_buffer.process_incr",
     queue="counters-0",

tests/sentry/tasks/test_process_buffer.py

@@ -8,6 +8,7 @@
     process_incr,
     process_pending,
     process_pending_batch,
+    schedule_delayed_workflows,
 )
 from sentry.testutils.cases import TestCase
 
@@ -50,3 +51,71 @@ def test_process_pending_batch_locked_out(self, mock_process_buffer: mock.MagicM
         with self.assertNoLogs("sentry.tasks.process_buffer", level="WARNING"):
             process_pending_batch()
             assert len(mock_process_buffer.mock_calls) == 1
+
+
+class ScheduleDelayedWorkflowsTest(TestCase):
+    @mock.patch("sentry.options.get")
+    @mock.patch("sentry.rules.processing.buffer_processing.process_buffer_for_type")
+    @mock.patch("sentry.workflow_engine.tasks.delayed_workflows.DelayedWorkflow")
+    def test_schedule_delayed_workflows_locked_out(
+        self,
+        mock_delayed_workflow: mock.MagicMock,
+        mock_process_buffer_for_type: mock.MagicMock,
+        mock_options_get: mock.MagicMock,
+    ) -> None:
+        # Mock the config option to return True (using new scheduling task)
+        mock_options_get.return_value = True
+
+        with self.assertLogs("sentry.tasks.process_buffer", level="WARNING") as logger:
+            lock = get_process_lock("schedule_delayed_workflows")
+            with lock.acquire():
+                schedule_delayed_workflows()
+                self.assertEqual(len(logger.output), 1)
+                assert len(mock_process_buffer_for_type.mock_calls) == 0
+
+        with self.assertNoLogs("sentry.tasks.process_buffer", level="WARNING"):
+            schedule_delayed_workflows()
+            assert len(mock_process_buffer_for_type.mock_calls) == 1
+            mock_process_buffer_for_type.assert_called_with(
+                "delayed_workflow", mock_delayed_workflow
+            )
+
+    @mock.patch("sentry.options.get")
+    @mock.patch("sentry.rules.processing.buffer_processing.process_buffer_for_type")
+    @mock.patch("sentry.workflow_engine.tasks.delayed_workflows.DelayedWorkflow")
+    def test_schedule_delayed_workflows_config_option_true(
+        self,
+        mock_delayed_workflow: mock.MagicMock,
+        mock_process_buffer_for_type: mock.MagicMock,
+        mock_options_get: mock.MagicMock,
+    ) -> None:
+        # Mock the config option to return False (not using new scheduling task)
+        mock_options_get.return_value = False
+
+        with self.assertLogs("sentry.tasks.process_buffer", level="INFO") as logger:
+            schedule_delayed_workflows()
+            # Should log and exit without calling process_buffer_for_type
+            assert len(mock_process_buffer_for_type.mock_calls) == 0
+            assert any(
+                "Configured to use process_pending_batch" in output for output in logger.output
+            )
+
+    @mock.patch("sentry.options.get")
+    @mock.patch("sentry.rules.processing.buffer_processing.process_buffer_for_type")
+    @mock.patch("sentry.workflow_engine.tasks.delayed_workflows.DelayedWorkflow")
+    def test_schedule_delayed_workflows_normal_operation(
+        self,
+        mock_delayed_workflow: mock.MagicMock,
+        mock_process_buffer_for_type: mock.MagicMock,
+        mock_options_get: mock.MagicMock,
+    ) -> None:
+        # Mock the config option to return True (using new scheduling task)
+        mock_options_get.return_value = True
+
+        schedule_delayed_workflows()
+
+        # Should only process delayed_workflow
+        assert len(mock_process_buffer_for_type.mock_calls) == 1
+        mock_process_buffer_for_type.assert_called_once_with(
+            "delayed_workflow", mock_delayed_workflow
+        )