chore(core): deprecate core.dispatch_with_results

.sg/rules/core-dispatch-with-results.yml

@@ -0,0 +1,67 @@
+id: core-dispatch-with-results
+message: Avoid using `dispatch_with_results()` - use `core.dispatch()` instead
+severity: warning
+language: python
+ignores:
+  - "ddtrace/internal/core/__init__.py"
+  - "ddtrace/internal/core/event_hub.py"
+  - "tests/internal/test_context_events_api.py"
+rule:
+  any:
+    # Match function calls to dispatch_with_results
+    - pattern: core.dispatch_with_results($$$ARGS)
+    - pattern: dispatch_with_results($$$ARGS)
+    - pattern: event_hub.dispatch_with_results($$$ARGS)
+    # Match imports of dispatch_with_results
+    - pattern: from ddtrace.internal.core import dispatch_with_results
+    - pattern: from ddtrace.internal.core import dispatch_with_results as $ALIAS
+    - pattern: from ddtrace.internal.core import $$$IMPORTS, dispatch_with_results
+    - pattern: from ddtrace.internal.core import $$$IMPORTS, dispatch_with_results as $ALIAS
+    - pattern: from ddtrace.internal.core import dispatch_with_results, $$$IMPORTS
+    - pattern: from ddtrace.internal.core import dispatch_with_results as $ALIAS, $$$IMPORTS
+    - pattern: from ddtrace.internal.core.event_hub import dispatch_with_results
+    - pattern: from ddtrace.internal.core.event_hub import dispatch_with_results as $ALIAS
+    - pattern: from ddtrace.internal.core.event_hub import $$$IMPORTS, dispatch_with_results
+    - pattern: from ddtrace.internal.core.event_hub import $$$IMPORTS, dispatch_with_results as $ALIAS
+    - pattern: from ddtrace.internal.core.event_hub import dispatch_with_results, $$$IMPORTS
+    - pattern: from ddtrace.internal.core.event_hub import dispatch_with_results as $ALIAS, $$$IMPORTS
+constraints:
+  # Exclude the implementation files where it's defined or exported
+  ALIAS:
+    regex: ".*"
+note: |
+  Avoid using `dispatch_with_results()` to prevent tight coupling between dispatched events
+  and potential listeners. This function creates explicit dependencies that make the code
+  harder to maintain and understand.
+
+  Prefer managing state/context via core.ExecutionContext using `core.set_item()` and
+  `core.get_item()`:
+
+  Before:
+    result = core.dispatch_with_results("my.event", (arg1, arg2)).result_key.value
+
+  After:
+    core.dispatch("my.event", (arg1, arg2))
+    result = core.get_item("result_key")
+
+  Alternatively, pass mutable state in the dispatch call:
+
+    data = {}
+    core.dispatch("my.event", (data,))
+    if data.get("result"):
+        # do something with data["result"]
+
+  To interrupt application flow, use try/except around dispatch calls:
+
+    try:
+        core.dispatch("my.event", (args,))
+    except MyException:
+        # handle interruption
+        pass
+
+  Note: try/except will block all other listeners from running after the exception is raised.
+
+  This rule does not apply to:
+  - The implementation in ddtrace/internal/core/event_hub.py
+  - The export in ddtrace/internal/core/__init__.py
+  - Test files and benchmarks

.sg/tests/__snapshots__/core-dispatch-with-results-snapshot.yml

@@ -0,0 +1,117 @@
+id: core-dispatch-with-results
+snapshots:
+  ? |
+    # Using dispatch_with_results in a chain
+    value = core.dispatch_with_results("my.event", (x,)).result.value
+    print(value)
+  : labels:
+    - source: core.dispatch_with_results("my.event", (x,))
+      style: primary
+      start: 49
+      end: 93
+  core.dispatch_with_results("my.event", (arg1, arg2)):
+    labels:
+    - source: core.dispatch_with_results("my.event", (arg1, arg2))
+      style: primary
+      start: 0
+      end: 52
+  ? |
+    data = core.dispatch_with_results("my.event", (arg1, arg2))
+    if data.result_key:
+        process(data.result_key.value)
+  : labels:
+    - source: core.dispatch_with_results("my.event", (arg1, arg2))
+      style: primary
+      start: 7
+      end: 59
+  dispatch_with_results("my.event", (arg1,)):
+    labels:
+    - source: dispatch_with_results("my.event", (arg1,))
+      style: primary
+      start: 0
+      end: 42
+  event_hub.dispatch_with_results("my.event", ()):
+    labels:
+    - source: event_hub.dispatch_with_results("my.event", ())
+      style: primary
+      start: 0
+      end: 47
+  ? |
+    from ddtrace.internal import core
+    result = core.dispatch_with_results("my.event", (arg1, arg2))
+  : labels:
+    - source: core.dispatch_with_results("my.event", (arg1, arg2))
+      style: primary
+      start: 43
+      end: 95
+  from ddtrace.internal.core import dispatch, dispatch_with_results:
+    labels:
+    - source: from ddtrace.internal.core import dispatch, dispatch_with_results
+      style: primary
+      start: 0
+      end: 65
+  from ddtrace.internal.core import dispatch, dispatch_with_results, on:
+    labels:
+    - source: from ddtrace.internal.core import dispatch, dispatch_with_results, on
+      style: primary
+      start: 0
+      end: 69
+  from ddtrace.internal.core import dispatch_with_results:
+    labels:
+    - source: from ddtrace.internal.core import dispatch_with_results
+      style: primary
+      start: 0
+      end: 55
+  ? |
+    from ddtrace.internal.core import dispatch_with_results
+    result = dispatch_with_results("my.event", (arg1,))
+  : labels:
+    - source: from ddtrace.internal.core import dispatch_with_results
+      style: primary
+      start: 0
+      end: 55
+  from ddtrace.internal.core import dispatch_with_results as dwr:
+    labels:
+    - source: from ddtrace.internal.core import dispatch_with_results as dwr
+      style: primary
+      start: 0
+      end: 62
+  from ddtrace.internal.core import dispatch_with_results, on:
+    labels:
+    - source: from ddtrace.internal.core import dispatch_with_results, on
+      style: primary
+      start: 0
+      end: 59
+  from ddtrace.internal.core.event_hub import dispatch, dispatch_with_results:
+    labels:
+    - source: from ddtrace.internal.core.event_hub import dispatch, dispatch_with_results
+      style: primary
+      start: 0
+      end: 75
+  from ddtrace.internal.core.event_hub import dispatch_with_results:
+    labels:
+    - source: from ddtrace.internal.core.event_hub import dispatch_with_results
+      style: primary
+      start: 0
+      end: 65
+  ? |
+    from ddtrace.internal.core.event_hub import dispatch_with_results
+    result = dispatch_with_results("my.event", ())
+  : labels:
+    - source: from ddtrace.internal.core.event_hub import dispatch_with_results
+      style: primary
+      start: 0
+      end: 65
+  from ddtrace.internal.core.event_hub import dispatch_with_results as dwr:
+    labels:
+    - source: from ddtrace.internal.core.event_hub import dispatch_with_results as dwr
+      style: primary
+      start: 0
+      end: 72
+  ? |
+    result = core.dispatch_with_results("my.event", (arg1, arg2)).result_key.value
+  : labels:
+    - source: core.dispatch_with_results("my.event", (arg1, arg2))
+      style: primary
+      start: 9
+      end: 61

.sg/tests/core-dispatch-with-results-test.yml

@@ -0,0 +1,65 @@
+id: core-dispatch-with-results
+valid:
+  # These should NOT trigger the rule (valid code)
+  - core.dispatch("my.event", (arg1, arg2))
+  - |
+    from ddtrace.internal.core import dispatch
+    dispatch("my.event", (arg1,))
+  - |
+    from ddtrace.internal.core import core
+    core.dispatch("my.event", ())
+  - |
+    # Using dispatch with mutable state is OK
+    data = {}
+    core.dispatch("my.event", (data,))
+    if data.get("result"):
+        process(data["result"])
+  - |
+    # Using set_item/get_item is the preferred pattern
+    core.dispatch("my.event", (arg1, arg2))
+    result = core.get_item("result_key")
+  - |
+    # Using try/except for flow control is OK
+    try:
+        core.dispatch("my.event", (args,))
+    except MyException:
+        pass
+  - |
+    # Other functions with similar names are OK
+    my_module.dispatch_with_results(args)
+    dispatch_with_other_results(args)
+
+invalid:
+  # These should trigger the rule (warnings)
+  # Direct function calls
+  - core.dispatch_with_results("my.event", (arg1, arg2))
+  - dispatch_with_results("my.event", (arg1,))
+  - event_hub.dispatch_with_results("my.event", ())
+  - |
+    result = core.dispatch_with_results("my.event", (arg1, arg2)).result_key.value
+  - |
+    data = core.dispatch_with_results("my.event", (arg1, arg2))
+    if data.result_key:
+        process(data.result_key.value)
+  - |
+    # Using dispatch_with_results in a chain
+    value = core.dispatch_with_results("my.event", (x,)).result.value
+    print(value)
+  # Import statements
+  - from ddtrace.internal.core import dispatch_with_results
+  - from ddtrace.internal.core import dispatch_with_results as dwr
+  - from ddtrace.internal.core import dispatch, dispatch_with_results
+  - from ddtrace.internal.core import dispatch_with_results, on
+  - from ddtrace.internal.core import dispatch, dispatch_with_results, on
+  - from ddtrace.internal.core.event_hub import dispatch_with_results
+  - from ddtrace.internal.core.event_hub import dispatch_with_results as dwr
+  - from ddtrace.internal.core.event_hub import dispatch, dispatch_with_results
+  - |
+    from ddtrace.internal.core import dispatch_with_results
+    result = dispatch_with_results("my.event", (arg1,))
+  - |
+    from ddtrace.internal.core.event_hub import dispatch_with_results
+    result = dispatch_with_results("my.event", ())
+  - |
+    from ddtrace.internal import core
+    result = core.dispatch_with_results("my.event", (arg1, arg2))

benchmarks/core_api/scenario.py

@@ -36,7 +36,9 @@ def core_dispatch(loops):
         def core_dispatch_with_results(loops):
             """Measure the cost to dispatch an event on the hub"""
             for _ in range(loops):
-                core.dispatch_with_results(CUSTOM_EVENT_NAME, (5, 6, 7, 8))
+                core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+                    CUSTOM_EVENT_NAME, (5, 6, 7, 8)
+                )
 
         def context_with_data(loops):
             """Measure the cost of creating and ending a new context"""

ddtrace/contrib/dbapi.py

@@ -106,7 +106,7 @@ def _trace_method(self, method, name, resource, extra_tags, dbm_propagator, *arg
             # dispatch DBM
             if dbm_propagator:
                 # this check is necessary to prevent fetch methods from trying to add dbm propagation
-                result = core.dispatch_with_results(
+                result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
                     f"{self._self_config.integration_name}.execute", (self._self_config, s, args, kwargs)
                 ).result
                 if result:

ddtrace/contrib/dbapi_async.py

@@ -81,7 +81,7 @@ async def _trace_method(self, method, name, resource, extra_tags, dbm_propagator
             # dispatch DBM
             if dbm_propagator:
                 # this check is necessary to prevent fetch methods from trying to add dbm propagation
-                result = core.dispatch_with_results(
+                result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
                     f"{self._self_config.integration_name}.execute", (self._self_config, s, args, kwargs)
                 ).result
                 if result:

ddtrace/contrib/internal/aiomysql/patch.py

@@ -93,7 +93,9 @@ async def _trace_method(self, method, resource, extra_tags, *args, **kwargs):
             s.set_tags(extra_tags)
 
             # dispatch DBM
-            result = core.dispatch_with_results("aiomysql.execute", (config.aiomysql, s, args, kwargs)).result
+            result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+                "aiomysql.execute", (config.aiomysql, s, args, kwargs)
+            ).result
             if result:
                 s, args, kwargs = result.value
 

ddtrace/contrib/internal/asgi/middleware.py

@@ -261,7 +261,9 @@ async def __call__(self, scope: Mapping[str, Any], receive: Callable, send: Call
             if not self.integration_config.trace_query_string:
                 query_string = None
             body = None
-            result = core.dispatch_with_results("asgi.request.parse.body", (receive, headers)).await_receive_and_body
+            result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+                "asgi.request.parse.body", (receive, headers)
+            ).await_receive_and_body
             if result:
                 receive, body = await result.value
 
@@ -421,7 +423,9 @@ async def wrapped_send(message: Mapping[str, Any]):
                         span.finish()
 
             async def wrapped_blocked_send(message: Mapping[str, Any]):
-                result = core.dispatch_with_results("asgi.block.started", (ctx, url)).status_headers_content
+                result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+                    "asgi.block.started", (ctx, url)
+                ).status_headers_content
                 if result:
                     status, headers, content = result.value
                 else:

ddtrace/contrib/internal/asyncpg/patch.py

@@ -142,7 +142,9 @@ async def _traced_query(pin, method, query, args, kwargs):
         span.set_tags(pin.tags)
 
         # dispatch DBM
-        result = core.dispatch_with_results("asyncpg.execute", (config.asyncpg, method, span, args, kwargs)).result
+        result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+            "asyncpg.execute", (config.asyncpg, method, span, args, kwargs)
+        ).result
         if result:
             span, args, kwargs = result.value
 

ddtrace/contrib/internal/django/patch.py

@@ -374,7 +374,7 @@ def traced_authenticate(django, pin, func, instance, args, kwargs):
     if mode == "disabled":
         return result_user
     try:
-        result = core.dispatch_with_results(
+        result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
             "django.auth",
             (result_user, mode, kwargs, pin, _DjangoUserInfoRetriever(result_user, credentials=kwargs), config_django),
         ).user
@@ -485,7 +485,9 @@ def _(m):
 
 
 def wrap_wsgi_environ(wrapped, _instance, args, kwargs):
-    result = core.dispatch_with_results("django.wsgi_environ", (wrapped, _instance, args, kwargs)).wrapped_result
+    result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+        "django.wsgi_environ", (wrapped, _instance, args, kwargs)
+    ).wrapped_result
     # if the callback is registered and runs, return the result
     if result:
         return result.value

ddtrace/contrib/internal/django/utils.py

@@ -264,7 +264,9 @@ def _extract_body(request):
     if request.method in _BODY_METHODS:
         req_body = None
         content_type = request.content_type if hasattr(request, "content_type") else request.META.get("CONTENT_TYPE")
-        headers = core.dispatch_with_results("django.extract_body").headers.value
+        headers = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+            "django.extract_body"
+        ).headers.value
         try:
             if content_type == "application/x-www-form-urlencoded":
                 req_body = parse_form_params(request.body.decode("UTF-8", errors="ignore"))
@@ -372,7 +374,9 @@ def _after_request_tags(pin, span: Span, request, response):
 
             url = get_request_uri(request)
 
-            request_headers = core.dispatch_with_results("django.after_request_headers").headers.value
+            request_headers = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+                "django.after_request_headers"
+            ).headers.value
             if not request_headers:
                 request_headers = _get_request_headers(request)
 

ddtrace/contrib/internal/flask/patch.py

@@ -127,7 +127,9 @@ def _wrapped_start_response(self, start_response, ctx, status_code, headers, exc
             core.dispatch("flask.start_response", ("Flask",))
             if get_blocked():
                 # response code must be set here, or it will be too late
-                result_content = core.dispatch_with_results("flask.block.request.content", ()).block_requested
+                result_content = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+                    "flask.block.request.content", ()
+                ).block_requested
                 if result_content:
                     _, status, response_headers = result_content.value
                     result = start_response(str(status), response_headers)
@@ -155,7 +157,7 @@ def _request_call_modifier(self, ctx, parsed_headers=None):
         request = _RequestType(environ)
 
         req_body = None
-        result = core.dispatch_with_results(
+        result = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
             "flask.request_call_modifier",
             (
                 ctx,

ddtrace/contrib/internal/flask/wrappers.py

@@ -45,7 +45,9 @@ def _wrap_call(
         tags=tags,
     ) as ctx, ctx.span:
         if do_dispatch:
-            dispatch = core.dispatch_with_results("flask.wrapped_view", (kwargs,))
+            dispatch = core.dispatch_with_results(  # ast-grep-ignore: core-dispatch-with-results
+                "flask.wrapped_view", (kwargs,)
+            )
 
             # Appsec blocks the request
             result = dispatch.callbacks