feat(aci): Separate Buffer for Workflows

getsentry · kcons · Aug 9, 2025 · Aug 9, 2025 · ef15d429bda7f8fca4d98adda136c2928d29e0e8
commit ef15d429bda7f8fca4d98adda136c2928d29e0e8
diff --git a/src/sentry/buffer/base.py b/src/sentry/buffer/base.py
@@ -30,7 +30,6 @@ class Buffer(Service):
         "incr",
         "process",
         "process_pending",
-        "process_batch",
         "validate",
         "push_to_sorted_set",
         "push_to_hash",
@@ -140,9 +139,6 @@ def incr(
     def process_pending(self) -> None:
         return
 
-    def process_batch(self) -> None:
-        return
-
     def process(
         self,
         model: type[models.Model] | None,

diff --git a/src/sentry/buffer/redis.py b/src/sentry/buffer/redis.py
@@ -56,32 +56,6 @@ def _validate_json_roundtrip(value: dict[str, Any], model: type[models.Model]) -
             logger.exception("buffer.invalid_value", extra={"value": value, "model": model})
 
 
-class BufferHookEvent(Enum):
-    FLUSH = "flush"
-
-
-class BufferHookRegistry:
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
-        self._registry: dict[BufferHookEvent, Callable[..., Any]] = {}
-
-    def add_handler(self, key: BufferHookEvent, func: Callable[..., Any]) -> None:
-        self._registry[key] = func
-
-    def has(self, key: BufferHookEvent) -> bool:
-        return self._registry.get(key) is not None
-
-    def callback(self, buffer_hook_event: BufferHookEvent) -> bool:
-        try:
-            callback = self._registry[buffer_hook_event]
-        except KeyError:
-            logger.exception("buffer_hook_event.missing")
-
-        return callback()
-
-
-redis_buffer_registry = BufferHookRegistry()
-
-
 # Callable to get the queue name for the given model_key.
 # May return None to not assign a queue for the given model_key.
 ChooseQueueFunction = Callable[[str], str | None]
@@ -445,12 +419,6 @@ def get_hash_length(self, model: type[models.Model], field: dict[str, BufferFiel
         key = self._make_key(model, field)
         return self._execute_redis_operation(key, RedisOperation.HASH_LENGTH)
 
-    def process_batch(self) -> None:
-        try:
-            redis_buffer_registry.callback(BufferHookEvent.FLUSH)
-        except Exception:
-            logger.exception("process_batch.error")
-
     def incr(
         self,
         model: type[models.Model],

diff --git a/src/sentry/conf/server.py b/src/sentry/conf/server.py
@@ -2088,6 +2088,10 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
 SENTRY_BUFFER = "sentry.buffer.Buffer"
 SENTRY_BUFFER_OPTIONS: dict[str, str] = {}
 
+# Workflow Buffer backend
+SENTRY_WORKFLOW_BUFFER = "sentry.buffer.Buffer"
+SENTRY_WORKFLOW_BUFFER_OPTIONS: dict[str, str] = {}
+
 # Cache backend
 # XXX: We explicitly require the cache to be configured as its not optional
 # and causes serious confusion with the default django cache

diff --git a/src/sentry/options/defaults.py b/src/sentry/options/defaults.py
@@ -3147,6 +3147,12 @@
     default=[],
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+register(
+    "workflow_engine.buffer.use_new_buffer",
+    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

diff --git a/src/sentry/rules/processing/buffer_processing.py b/src/sentry/rules/processing/buffer_processing.py
@@ -9,11 +9,11 @@
 
 from celery import Task
 
-from sentry import buffer, options
-from sentry.buffer.base import BufferField
-from sentry.buffer.redis import BufferHookEvent, redis_buffer_registry
+from sentry import options
+from sentry.buffer.base import Buffer, BufferField
 from sentry.db import models
 from sentry.utils import metrics
+from sentry.utils.lazy_service_wrapper import LazyServiceWrapper
 from sentry.utils.registry import NoRegistrationExistsError, Registry
 
 logger = logging.getLogger("sentry.delayed_processing")
@@ -47,12 +47,19 @@ def hash_args(self) -> BufferHashKeys:
     def processing_task(self) -> Task:
         raise NotImplementedError
 
+    @staticmethod
+    def buffer_backend() -> LazyServiceWrapper[Buffer]:
+        raise NotImplementedError
+
 
 delayed_processing_registry = Registry[type[DelayedProcessingBase]]()
 
 
 def fetch_group_to_event_data(
-    project_id: int, model: type[models.Model], batch_key: str | None = None
+    buffer: LazyServiceWrapper[Buffer],
+    project_id: int,
+    model: type[models.Model],
+    batch_key: str | None = None,
 ) -> dict[str, str]:
     field: dict[str, models.Model | int | str] = {
         "project_id": project_id,
@@ -61,7 +68,7 @@ def fetch_group_to_event_data(
     if batch_key:
         field["batch_key"] = batch_key
 
-    return buffer.backend.get_hash(model=model, field=field)
+    return buffer.get_hash(model=model, field=field)
 
 
 def bucket_num_groups(num_groups: int) -> str:
@@ -71,7 +78,9 @@ def bucket_num_groups(num_groups: int) -> str:
     return "1"
 
 
-def process_in_batches(project_id: int, processing_type: str) -> None:
+def process_in_batches(
+    buffer: LazyServiceWrapper[Buffer], project_id: int, processing_type: str
+) -> None:
     """
     This will check the number of alertgroup_to_event_data items in the Redis buffer for a project.
 
@@ -100,7 +109,7 @@ def process_in_batches(project_id: int, processing_type: str) -> None:
     task = processing_info.processing_task
     filters: dict[str, BufferField] = asdict(hash_args.filters)
 
-    event_count = buffer.backend.get_hash_length(model=hash_args.model, field=filters)
+    event_count = buffer.get_hash_length(model=hash_args.model, field=filters)
     metrics.incr(
         f"{processing_type}.num_groups", tags={"num_groups": bucket_num_groups(event_count)}
     )
@@ -118,22 +127,22 @@ def process_in_batches(project_id: int, processing_type: str) -> None:
         )
 
     # if the dictionary is large, get the items and chunk them.
-    alertgroup_to_event_data = fetch_group_to_event_data(project_id, hash_args.model)
+    alertgroup_to_event_data = fetch_group_to_event_data(buffer, project_id, hash_args.model)
 
     with metrics.timer(f"{processing_type}.process_batch.duration"):
         items = iter(alertgroup_to_event_data.items())
 
         while batch := dict(islice(items, batch_size)):
             batch_key = str(uuid.uuid4())
 
-            buffer.backend.push_to_hash_bulk(
+            buffer.push_to_hash_bulk(
                 model=hash_args.model,
                 filters={**filters, "batch_key": batch_key},
                 data=batch,
             )
 
             # remove the batched items from the project alertgroup_to_event_data
-            buffer.backend.delete_hash(**asdict(hash_args), fields=list(batch.keys()))
+            buffer.delete_hash(**asdict(hash_args), fields=list(batch.keys()))
 
             task.apply_async(
                 kwargs={"project_id": project_id, "batch_key": batch_key},
@@ -150,14 +159,16 @@ def process_buffer() -> None:
             logger.info(log_name, extra={"option": handler.option})
             continue
 
+        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)
-            project_ids = buffer.backend.get_sorted_set(
+            project_ids = buffer.get_sorted_set(
                 handler.buffer_key, min=0, max=fetch_time.timestamp()
             )
             if should_emit_logs:
@@ -168,10 +179,6 @@ def process_buffer() -> None:
                 logger.info(log_name, extra={"project_ids": log_str})
 
             for project_id, _ in project_ids:
-                process_in_batches(project_id, processing_type)
-
-            buffer.backend.delete_key(handler.buffer_key, min=0, max=fetch_time.timestamp())
-
+                process_in_batches(buffer, project_id, processing_type)
 
-if not redis_buffer_registry.has(BufferHookEvent.FLUSH):
-    redis_buffer_registry.add_handler(BufferHookEvent.FLUSH, process_buffer)
+            buffer.delete_key(handler.buffer_key, min=0, max=fetch_time.timestamp())
@@ -12,7 +12,7 @@
 from django.db.models import OuterRef, Subquery
 
 from sentry import buffer, features, nodestore
-from sentry.buffer.base import BufferField
+from sentry.buffer.base import Buffer, BufferField
 from sentry.db import models
 from sentry.eventstore.models import Event, GroupEvent
 from sentry.issues.issue_occurrence import IssueOccurrence
@@ -52,6 +52,7 @@
 from sentry.utils import json, metrics
 from sentry.utils.dates import ensure_aware
 from sentry.utils.iterators import chunked
+from sentry.utils.lazy_service_wrapper import LazyServiceWrapper
 from sentry.utils.retries import ConditionalRetryPolicy, exponential_delay
 from sentry.utils.safe import safe_execute
 from sentry.workflow_engine.processors.log_util import track_batch_performance
@@ -787,3 +788,7 @@ def hash_args(self) -> BufferHashKeys:
     @property
     def processing_task(self) -> Task:
         return apply_delayed
+
+    @staticmethod
+    def buffer_backend() -> LazyServiceWrapper[Buffer]:
+        return buffer.backend
diff --git a/src/sentry/tasks/process_buffer.py b/src/sentry/tasks/process_buffer.py
@@ -52,13 +52,13 @@ def process_pending_batch() -> None:
     """
     Process pending buffers in a batch.
     """
-    from sentry import buffer
+    from sentry.rules.processing.buffer_processing import process_buffer
 
     lock = get_process_lock("process_pending_batch")
 
     try:
         with lock.acquire():
-            buffer.backend.process_batch()
+            process_buffer()
     except UnableToAcquireLock as error:
         logger.warning("process_pending_batch.fail", extra={"error": error})
 

@@ -0,0 +1,21 @@
+from django.conf import settings
+
+import sentry.buffer as old_buffer
+from sentry import options
+from sentry.buffer.base import Buffer
+from sentry.utils.services import LazyServiceWrapper
+
+_backend = LazyServiceWrapper(
+    Buffer, settings.SENTRY_WORKFLOW_BUFFER, settings.SENTRY_WORKFLOW_BUFFER_OPTIONS
+)
+
+
+def validate_new_backend() -> None:
+    pass
+
+
+def get_backend() -> LazyServiceWrapper[Buffer]:
+    if options.get("workflow_engine.buffer.use_new_buffer"):
+        return _backend
+    else:
+        return old_buffer.backend
@@ -11,7 +11,8 @@
 from django.utils import timezone
 from pydantic import BaseModel, validator
 
-from sentry import buffer, features, nodestore, options
+import sentry.workflow_engine.buffer as buffer
+from sentry import features, nodestore, options
 from sentry.buffer.base import BufferField
 from sentry.db import models
 from sentry.eventstore.models import Event, GroupEvent
@@ -310,7 +311,7 @@ def fetch_group_to_event_data(
     if batch_key:
         field["batch_key"] = batch_key
 
-    return buffer.backend.get_hash(model=model, field=field)
+    return buffer.get_backend().get_hash(model=model, field=field)
 
 
 def fetch_workflows_envs(
@@ -771,7 +772,7 @@ def cleanup_redis_buffer(
     if batch_key:
         filters["batch_key"] = batch_key
 
-    buffer.backend.delete_hash(model=Workflow, filters=filters, fields=hashes_to_delete)
+    buffer.get_backend().delete_hash(model=Workflow, filters=filters, fields=hashes_to_delete)
 
 
 def repr_keys[T, V](d: dict[T, V]) -> dict[str, V]:

@@ -8,11 +8,12 @@
 from django.db.models import Q
 from django.utils import timezone
 
-from sentry import buffer, features
+from sentry import features
 from sentry.eventstore.models import GroupEvent
 from sentry.models.activity import Activity
 from sentry.models.environment import Environment
 from sentry.utils import json
+from sentry.workflow_engine import buffer
 from sentry.workflow_engine.models import Action, DataConditionGroup, Detector, Workflow
 from sentry.workflow_engine.models.workflow_data_condition_group import WorkflowDataConditionGroup
 from sentry.workflow_engine.processors.action import filter_recently_fired_workflow_actions
@@ -114,8 +115,9 @@ def enqueue_workflows(
         sentry_sdk.set_tag("delayed_workflow_items", items)
         return
 
+    backend = buffer.get_backend()
     for project_id, queue_items in items_by_project_id.items():
-        buffer.backend.push_to_hash_bulk(
+        backend.push_to_hash_bulk(
             model=Workflow,
             filters={"project_id": project_id},
             data={queue_item.buffer_key(): queue_item.buffer_value() for queue_item in queue_items},
@@ -125,7 +127,7 @@ def enqueue_workflows(
 
     sentry_sdk.set_tag("delayed_workflow_items", items)
 
-    buffer.backend.push_to_sorted_set(
+    backend.push_to_sorted_set(
         key=WORKFLOW_ENGINE_BUFFER_LIST_KEY, value=list(items_by_project_id.keys())
     )
 

@@ -4,7 +4,9 @@
 
 from celery import Task
 
+import sentry.workflow_engine.buffer as buffer
 from sentry import options
+from sentry.buffer.base import Buffer
 from sentry.rules.processing.buffer_processing import (
     BufferHashKeys,
     DelayedProcessingBase,
@@ -16,6 +18,7 @@
 from sentry.taskworker.config import TaskworkerConfig
 from sentry.taskworker.namespaces import workflow_engine_tasks
 from sentry.taskworker.retry import Retry
+from sentry.utils.lazy_service_wrapper import LazyServiceWrapper
 from sentry.workflow_engine.models import Workflow
 from sentry.workflow_engine.processors.workflow import WORKFLOW_ENGINE_BUFFER_LIST_KEY
 from sentry.workflow_engine.utils import log_context
@@ -66,3 +69,7 @@ def processing_task(self) -> Task:
         if options.get("delayed_workflow.use_workflow_engine_pool"):
             return process_delayed_workflows_shim
         return process_delayed_workflows
+
+    @staticmethod
+    def buffer_backend() -> LazyServiceWrapper[Buffer]:
+        return buffer.get_backend()