feat(Instrumentation/Messenger): implement worker and middleware instrumentation (#173)

* chore(messenger): add trace parent in dispatched messages (AMQP)

* chore(messenger): use Messenger events to start/end span for instrumentation

* remove strict type, merge start/end span into 1 subscriber, close span on error and on message handled

* propagation not related to AMQP + read incoming trace (async context)

* rename event subscriber

* chore(messenger): use Messenger events to start/end span for instrumentation

* chore(messenger): add trace parent in dispatched messages (AMQP)

* remove strict type, merge start/end span into 1 subscriber, close span on error and on message handled

* propagation not related to AMQP + read incoming trace (async context)

* rename event subscriber

* fix(messenger): clean up worker subscriber and add functional tests

Fix several issues in the WorkerMessageEventSubscriber introduced by PR #173:
- Replace SDK Span import with API Span to respect API/SDK separation
- Implement InstrumentationTypeInterface for consistency with other subscribers
- Add event priorities (10000/-10000) to wrap all other processing
- Add messaging semantic convention attributes (operation.type, destination.name)
- Include message class name in span name for better trace readability
- Remove stale imports and duplicate propagation middleware service definition
- Clean up propagation middleware when messenger tracing is disabled
- Add PHPStan baseline entries for untyped $carrier interface params
- Add functional tests for worker message handled, failed, and attribute mode
- Reorganize messenger tests into Messenger/ subdirectory
- Disable retry on test transport to isolate worker span assertions

* test(messenger): add tests for transport tracing and propagation middleware

Cover TraceableMessengerTransport (get/ack/reject spans + TransportException
error recording) and AddStampForPropagationMiddleware (stamp skip, no-scope
passthrough, active-scope injection).

* fix(messenger): use recordException() instead of manual attributes

Align WorkerMessageEventSubscriber.endSpanOnError() with every other
subscriber in the codebase (HttpKernel, Console, Doctrine, Mailer,
TransportTracer) by using $span->recordException() which creates a
proper OTel exception event with exception.type, exception.message,
and exception.stacktrace per the OTel semantic conventions.

Update tests to assert on span events rather than span attributes.

* fix(messenger): unwrap HandlerFailedException in error spans

When Symfony Messenger handlers fail, the exception is wrapped in a
HandlerFailedException (or other WrappedExceptionsInterface implementors
like DelayedMessageHandlingException). Without unwrapping, the span
records the generic wrapper message instead of the actual root causes.

Now each nested exception is recorded as a separate span event via
recordException(), matching how Sentry handles these wrapped exceptions.

* fix(console): make messenger:consume exclusion configurable

Replace the hardcoded $notSupportedCommands array and isNotSupported()
method with a default value on the existing exclude_commands config node.

messenger:consume is now the default entry in console.tracing.exclude_commands,
following the same pattern as Sentry's configurable excluded_commands. Users
can override the list or clear it entirely via configuration.

* test(messenger): add unit test for TraceStampPropagator

Cover all branches of set(), get(), and keys() methods:
- Valid envelope carrier for set/get/keys
- Invalid carrier throws InvalidArgumentException
- Non-traceparent key is ignored (set) or returns null (get)
- Missing TraceStamp returns null from get()

Closes the 34.78% patch coverage gap reported by Codecov.

* fix(messenger): propagate tracestate alongside traceparent

The W3C Trace Context spec defines two headers: traceparent and
tracestate. Previously only traceparent was propagated, silently
dropping vendor-specific trace context (Datadog, AWS X-Ray, etc.).

Add optional traceState property to TraceStamp and update
TraceStampPropagator to handle both keys in set(), get(), and keys().
When tracestate arrives after traceparent, the existing stamp is
replaced with one carrying both values.

* fix(messenger): namespace bus.name attribute as symfony.messenger.bus.name

bus.name is not a standard OTel semantic convention. Rename to
symfony.messenger.bus.name to follow the existing codebase pattern
(symfony.console.*, symfony.kernel.*).

Only changes the worker subscriber attribute; the pre-existing
bus.name in TraceableMessengerStack is left unchanged (out of scope).

* fix(messenger): add scope leak safety in worker subscriber

If startSpan fires but neither endSpanWithSuccess nor endSpanOnError
fires (e.g. worker killed, unhandled error in another subscriber),
the OTel context scope leaks into subsequent messages.

Now startSpan checks for a lingering scope at the beginning of each
message and cleans it up: detaches the scope, marks the orphaned span
as ERROR, and ends it. This prevents context pollution across messages.

* feat(messenger): add retry awareness, attribute instrumentation, and namespace span attributes

- Namespace middleware span attributes (bus.name, event.category, event.current)
  under symfony.messenger.* prefix for consistency with worker subscriber
- Add symfony.messenger.will_retry and symfony.messenger.retry_count attributes
  to failure spans so operators can distinguish retriable from terminal failures
- Support #[Traceable] attribute on message classes for attribute-based
  instrumentation mode, with custom tracer selection via TracerLocatorPass

* refactor(messenger): improve type safety, observability, and test coverage

Mark all new types as final for consistency with existing codebase.
Register TraceStampPropagator as a shared DI service instead of inline
construction. Default instrumentationType to Auto, add safe tracer
locator fallback, nullable logger consistency, debug logging on skipped
messages and null scopes, and remove dead commented-out code. Add
end-to-end propagation test, WorkerMessageEventSubscriber unit tests,
and PropagatorFactory test.

* fix(http-client): use lowercase header key for Content-Length lookup

Symfony's getHeaders() normalizes header names to lowercase, so the
Content-Length check was unreachable. Access the first array element
since headers are returned as arrays.

* fix(messenger): use explicit scope tracking and record errors on middleware spans

TraceableMessengerStack now tracks its own scope, span, and parent context
instead of probing global context storage — fixing scope leaks where earlier
middleware spans were never ended. TraceableMessengerMiddleware catches
exceptions and records them with STATUS_ERROR on the active span, aligning
with every other instrumentation in the bundle.

* fix(messenger): remove erroneous kernel.reset tag from transport factory

TraceableMessengerTransportFactory was tagged `kernel.reset` with
method `reset`, but the class has no such method. Symfony's
ServicesResetter invokes the declared method on every tagged service
between worker iterations, so `messenger:consume` threw as soon as a
retry (or any post-handler reset) ran:

    Attempted to call an undefined method named "reset" of class
    "...\TraceableMessengerTransportFactory".

The tag was introduced with the original messenger tracing (6f3c032)
and never had a corresponding method — likely speculative, possibly
confused with messenger transports that sometimes implement
ResetInterface.

Removal is preferred over adding an empty reset(): the factory's
dependencies (inner TransportFactory, TracerInterface, LoggerInterface)
are immutable DI services, createTransport() returns a fresh
TraceableMessengerTransport without retaining a reference, and
supports() is pure — there is no per-message state to clear. An empty
reset() would silence the error but imply a ResetInterface-style
contract the class does not fulfill and mislead anyone later adding
mutable state.

---------

Co-authored-by: Gaël Reyrol <me@gaelreyrol.dev>

Author: jvocampings

.gitignore

@@ -25,4 +25,3 @@ CLAUDE.md
 .claude
 GEMINI.md
 .gemini
-

composer.json

@@ -62,6 +62,7 @@
         "phpunit/phpunit": "^13.0",
         "pyrech/composer-changelogs": "^2.2",
         "roave/security-advisories": "dev-master",
+        "symfony/amqp-messenger": "^7.4",
         "symfony/browser-kit": "^7.4",
         "symfony/cache": "^7.4",
         "symfony/framework-bundle": "^7.4",
@@ -127,9 +128,13 @@
     },
     "scripts": {
         "check-reqs": "@php tools/composer-require-checker/vendor/bin/composer-require-checker check",
-        "coverage": [
+        "coverage:html": [
             "@putenv XDEBUG_MODE=coverage",
-            "@phpunit --coverage-html=coverage"
+            "@phpunit --coverage-html=coverage/html"
+        ],
+        "coverage:xml": [
+            "@putenv XDEBUG_MODE=coverage",
+            "@phpunit --coverage-clover=coverage/coverage.xml"
         ],
         "format": [
             "@php-cs-fixer:fix",

phpstan-baseline.neon

@@ -12,6 +12,12 @@ parameters:
 			count: 1
 			path: src/Instrumentation/Doctrine/Middleware/TraceableConnectionV4.php
 
+		-
+			message: '#^Property FriendsOfOpenTelemetry\\OpenTelemetryBundle\\Instrumentation\\Symfony\\Messenger\\AddStampForPropagationMiddleware\:\:\$logger is never read, only written\.$#'
+			identifier: property.onlyWritten
+			count: 1
+			path: src/Instrumentation/Symfony/Messenger/AddStampForPropagationMiddleware.php
+
 		-
 			message: '#^Property FriendsOfOpenTelemetry\\OpenTelemetryBundle\\Instrumentation\\Doctrine\\Middleware\\TraceableStatementV4\:\:\$logger is never read, only written\.$#'
 			identifier: property.onlyWritten
@@ -90,6 +96,24 @@ parameters:
 			count: 1
 			path: tests/Unit/OpenTelemetry/Trace/TracerProvider/NoopTracerProviderFactoryTest.php
 
+		-
+			message: '#^Method FriendsOfOpenTelemetry\\OpenTelemetryBundle\\OpenTelemetry\\Context\\Propagator\\TraceStampPropagator\:\:set\(\) has parameter \$carrier with no type specified\.$#'
+			identifier: missingType.parameter
+			count: 1
+			path: src/OpenTelemetry/Context/Propagator/TraceStampPropagator.php
+
+		-
+			message: '#^Method FriendsOfOpenTelemetry\\OpenTelemetryBundle\\OpenTelemetry\\Context\\Propagator\\TraceStampPropagator\:\:keys\(\) has parameter \$carrier with no type specified\.$#'
+			identifier: missingType.parameter
+			count: 1
+			path: src/OpenTelemetry/Context/Propagator/TraceStampPropagator.php
+
+		-
+			message: '#^Method FriendsOfOpenTelemetry\\OpenTelemetryBundle\\OpenTelemetry\\Context\\Propagator\\TraceStampPropagator\:\:get\(\) has parameter \$carrier with no type specified\.$#'
+			identifier: missingType.parameter
+			count: 1
+			path: src/OpenTelemetry/Context/Propagator/TraceStampPropagator.php
+
 		-
 			message: '#^Call to an undefined static method ReflectionMethod\:\:createFromMethodName\(\)\.$#'
 			identifier: staticMethod.notFound

src/DependencyInjection/Compiler/SetInstrumentationTypePass.php

@@ -27,7 +27,10 @@ public function process(ContainerBuilder $container): void
 
         if ($container->hasParameter('open_telemetry.instrumentation.messenger.type')) {
             $messengerInstrumentationType = $container->getParameter('open_telemetry.instrumentation.messenger.type');
-            if ($container->hasDefinition('open_telemetry.instrumentation.http_kernel.trace.event_subscriber')) {
+
+            if ($container->hasDefinition('open_telemetry.instrumentation.messenger.worker')) {
+                $container->getDefinition('open_telemetry.instrumentation.messenger.worker')
+                    ->addMethodCall('setInstrumentationType', [$messengerInstrumentationType]);
             }
         }
     }

src/DependencyInjection/Compiler/TracerLocatorPass.php

@@ -23,6 +23,10 @@ public function process(ContainerBuilder $container): void
                 $traceableHttpKernelEventSubscriber = $container->getDefinition('open_telemetry.instrumentation.http_kernel.trace.event_subscriber');
                 $this->setTracerLocatorArgument($container, $traceableHttpKernelEventSubscriber, $tracers);
             }
+            if ($container->hasDefinition('open_telemetry.instrumentation.messenger.worker')) {
+                $messengerWorkerSubscriber = $container->getDefinition('open_telemetry.instrumentation.messenger.worker');
+                $this->setTracerLocatorArgument($container, $messengerWorkerSubscriber, $tracers);
+            }
         }
     }
 

src/DependencyInjection/Configuration.php

@@ -104,6 +104,7 @@ private function addInstrumentationSection(ArrayNodeDefinition $node): void
                                 (new ArrayNodeDefinition('exclude_commands'))
                                     ->info('Exclude commands from auto instrumentation')
                                     ->scalarPrototype()->cannotBeEmpty()->end()
+                                    ->defaultValue(['messenger:consume'])
                             ))
                             ->append($this->getMeteringInstrumentationNode())
                         ->end()

src/DependencyInjection/OpenTelemetryExtension.php

@@ -283,6 +283,9 @@ private function registerMessengerTracingInstrumentationConfiguration(ContainerB
             $container->removeDefinition('open_telemetry.instrumentation.messenger.trace.transport');
             $container->removeDefinition('open_telemetry.instrumentation.messenger.trace.transport_factory');
             $container->removeDefinition('open_telemetry.instrumentation.messenger.trace.middleware');
+            $container->removeDefinition('open_telemetry.instrumentation.messenger.trace_stamp_propagator');
+            $container->removeDefinition('open_telemetry.instrumentation.messenger.propagation.middleware');
+            $container->removeDefinition('open_telemetry.instrumentation.messenger.worker');
         }
 
         $this->setTracingInstrumentationParams($container, 'messenger', $config, $isConfigEnabled);

src/Instrumentation/Symfony/Console/TraceableConsoleEventSubscriber.php

@@ -56,6 +56,9 @@ public static function getSubscribedEvents(): array
         ];
     }
 
+    /**
+     * @return class-string[]
+     */
     public static function getSubscribedServices(): array
     {
         return [TracerInterface::class];

src/Instrumentation/Symfony/HttpClient/TraceableResponse.php

@@ -138,8 +138,8 @@ private function endSpan(): void
         $statusCode = $this->response->getStatusCode();
         if (0 !== $statusCode && $this->span->isRecording()) {
             $headers = $this->response->getHeaders(false);
-            if (isset($headers['Content-Length'])) {
-                $this->span->setAttribute(HttpIncubatingAttributes::HTTP_RESPONSE_BODY_SIZE, $headers['Content-Length']);
+            if (isset($headers['content-length'][0])) {
+                $this->span->setAttribute(HttpIncubatingAttributes::HTTP_RESPONSE_BODY_SIZE, $headers['content-length'][0]);
             }
 
             $this->span->setAttribute(HttpAttributes::HTTP_RESPONSE_STATUS_CODE, $statusCode);

src/Instrumentation/Symfony/Messenger/AddStampForPropagationMiddleware.php

@@ -0,0 +1,44 @@
+<?php
+
+namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger;
+
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator\TraceStampPropagator;
+use OpenTelemetry\Context\Context;
+use OpenTelemetry\Context\Propagation\MultiTextMapPropagator;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackInterface;
+
+/**
+ * Injects the current trace context into the Messenger envelope as a TraceStamp
+ * so that trace propagation is maintained across asynchronous message boundaries.
+ */
+final readonly class AddStampForPropagationMiddleware implements MiddlewareInterface
+{
+    public function __construct(
+        private MultiTextMapPropagator $propagator,
+        private TraceStampPropagator $traceStampPropagator,
+        private ?LoggerInterface $logger = null,
+    ) {
+    }
+
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        $traceStamp = $envelope->last(TraceStamp::class);
+
+        if (null !== $traceStamp) {
+            return $stack->next()->handle($envelope, $stack);
+        }
+
+        $scope = Context::storage()->scope();
+
+        if (null !== $scope) {
+            // inject() mutates $envelope by reference through the TraceStampPropagator setter,
+            // because Envelope is immutable and with() returns a new instance.
+            $this->propagator->inject($envelope, $this->traceStampPropagator, Context::getCurrent());
+        }
+
+        return $stack->next()->handle($envelope, $stack);
+    }
+}