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.

Author: gaelreyrol

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

src/DependencyInjection/OpenTelemetryExtension.php

@@ -283,6 +283,7 @@ 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');
         }

src/Instrumentation/Symfony/Messenger/AddStampForPropagationMiddleware.php

@@ -11,12 +11,14 @@
 use Symfony\Component\Messenger\Middleware\StackInterface;
 
 /**
- * Be aware the app consuming the message must be able to denormalize the stamp.
+ * Injects the current trace context into the Messenger envelope as a TraceStamp
+ * so that trace propagation is maintained across asynchronous message boundaries.
  */
-readonly class AddStampForPropagationMiddleware implements MiddlewareInterface
+final readonly class AddStampForPropagationMiddleware implements MiddlewareInterface
 {
     public function __construct(
         private MultiTextMapPropagator $propagator,
+        private TraceStampPropagator $traceStampPropagator,
         private ?LoggerInterface $logger = null,
     ) {
     }
@@ -32,7 +34,9 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
         $scope = Context::storage()->scope();
 
         if (null !== $scope) {
-            $this->propagator->inject($envelope, new TraceStampPropagator($this->logger), Context::getCurrent());
+            // 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);

src/Instrumentation/Symfony/Messenger/TraceStamp.php

@@ -5,11 +5,9 @@
 use Symfony\Component\Messenger\Stamp\StampInterface;
 
 /**
- * @doc: https://www.w3.org/TR/trace-context/
- *
- * You can see how the trace parent generated here: https://github.com/open-telemetry/opentelemetry-php/blob/main/src/API/Trace/Propagation/TraceContextPropagator.php
+ * @see https://www.w3.org/TR/trace-context/
  */
-readonly class TraceStamp implements StampInterface
+final readonly class TraceStamp implements StampInterface
 {
     public function __construct(
         private string $traceParent,

src/Instrumentation/Symfony/Messenger/TraceableMessengerStack.php

@@ -33,18 +33,6 @@ public function next(): MiddlewareInterface
             $this->logger?->debug('No active scope');
         }
 
-        /*        if (null !== $scope) {
-                    $span = Span::fromContext($scope->context());
-
-                    if ($span->isRecording()) {
-                        $scope->detach();
-
-                        $span->setStatus(StatusCode::STATUS_OK);
-                        $this->logger?->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
-                        $span->end();
-                    }
-                }*/
-
         $spanBuilder = $this->tracer
             ->spanBuilder('messenger.middleware')
             ->setSpanKind(SpanKind::KIND_INTERNAL)

src/Instrumentation/Symfony/Messenger/WorkerMessageEventSubscriber.php

@@ -23,16 +23,22 @@
 use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
 use Symfony\Contracts\Service\ServiceSubscriberInterface;
 
-class WorkerMessageEventSubscriber implements EventSubscriberInterface, ServiceSubscriberInterface, InstrumentationTypeInterface
+/**
+ * Creates consumer-side spans for messages processed by the Symfony Messenger worker,
+ * with support for both auto and attribute-based instrumentation modes,
+ * trace context propagation from the producer, and retry/failure metadata.
+ */
+final class WorkerMessageEventSubscriber implements EventSubscriberInterface, ServiceSubscriberInterface, InstrumentationTypeInterface
 {
-    private ?InstrumentationTypeEnum $instrumentationType = null;
+    private InstrumentationTypeEnum $instrumentationType = InstrumentationTypeEnum::Auto;
 
     public function __construct(
         private readonly MultiTextMapPropagator $propagator,
         private readonly TracerInterface $tracer,
         /** @var ServiceLocator<TracerInterface> */
         private readonly ServiceLocator $tracerLocator,
-        private readonly LoggerInterface $logger,
+        private readonly TraceStampPropagator $traceStampPropagator,
+        private readonly ?LoggerInterface $logger = null,
     ) {
     }
 
@@ -56,36 +62,43 @@ public static function getSubscribedEvents(): array
         ];
     }
 
+    /**
+     * @return class-string[]
+     */
     public static function getSubscribedServices(): array
     {
-        return [];
+        return [TracerInterface::class];
     }
 
     public function startSpan(WorkerMessageReceivedEvent $event): void
     {
         $message = $event->getEnvelope()->getMessage();
+        $traceable = $this->parseAttribute($message);
+
+        if (!$this->isAutoTraceable() && !$this->isAttributeTraceable($traceable)) {
+            $this->logger?->debug(sprintf('Message "%s" is not traceable, skipping span creation', get_class($message)));
 
-        if (false === $this->isAutoTraceable() && false === $this->isAttributeTraceable($message)) {
             return;
         }
 
         // Clean up any lingering scope from a previous message that was not
-        // properly ended (e.g. worker killed, unhandled error in another subscriber).
+        // properly ended (e.g. an exception in another high-priority subscriber
+        // prevented the handled/failed event from firing).
         $previousScope = Context::storage()->scope();
         if (null !== $previousScope) {
             $previousScope->detach();
             $orphanedSpan = Span::fromContext($previousScope->context());
             $orphanedSpan->setStatus(StatusCode::STATUS_ERROR, 'Span was not properly ended');
             $orphanedSpan->end();
-            $this->logger->warning(sprintf('Cleaned up orphaned span "%s"', $orphanedSpan->getContext()->getSpanId()));
+            $this->logger?->warning(sprintf('Cleaned up orphaned span "%s"', $orphanedSpan->getContext()->getSpanId()));
         }
 
         // ensure propagation from incoming trace
-        $context = $this->propagator->extract($event->getEnvelope(), new TraceStampPropagator($this->logger));
+        $context = $this->propagator->extract($event->getEnvelope(), $this->traceStampPropagator);
 
         $messageClass = get_class($message);
 
-        $span = $this->getTracer($message)
+        $span = $this->getTracer($traceable)
             ->spanBuilder(sprintf('%s %s', $event->getReceiverName(), $messageClass))
             ->setParent($context)
             ->setSpanKind(SpanKind::KIND_CONSUMER)
@@ -99,7 +112,7 @@ public function startSpan(WorkerMessageReceivedEvent $event): void
             $span->setAttribute('symfony.messenger.bus.name', $busNameStamp->getBusName());
         }
 
-        $this->logger->debug(sprintf('Starting span "%s"', $span->getContext()->getSpanId()));
+        $this->logger?->debug(sprintf('Starting span "%s"', $span->getContext()->getSpanId()));
 
         Context::storage()
             ->attach(
@@ -113,14 +126,16 @@ public function endSpanWithSuccess(WorkerMessageHandledEvent $event): void
         $scope = Context::storage()->scope();
 
         if (null === $scope) {
+            $this->logger?->debug('No active scope');
+
             return;
         }
 
         $scope->detach();
 
         $span = Span::fromContext($scope->context());
         $span->setStatus(StatusCode::STATUS_OK);
-        $this->logger->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
+        $this->logger?->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
         $span->end();
     }
 
@@ -129,6 +144,8 @@ public function endSpanOnError(WorkerMessageFailedEvent $event): void
         $scope = Context::storage()->scope();
 
         if (null === $scope) {
+            $this->logger?->debug('No active scope');
+
             return;
         }
 
@@ -155,7 +172,7 @@ public function endSpanOnError(WorkerMessageFailedEvent $event): void
 
         $span->setStatus(StatusCode::STATUS_ERROR, $exception->getMessage());
 
-        $this->logger->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
+        $this->logger?->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
         $span->end();
     }
 
@@ -167,11 +184,15 @@ private function parseAttribute(object $message): ?Traceable
         return $attribute?->newInstance();
     }
 
-    private function getTracer(object $message): TracerInterface
+    private function getTracer(?Traceable $traceable): TracerInterface
     {
-        $traceable = $this->parseAttribute($message);
-
         if (null !== $traceable?->tracer) {
+            if (!$this->tracerLocator->has($traceable->tracer)) {
+                $this->logger?->warning(sprintf('Tracer "%s" not found in service locator, using default tracer', $traceable->tracer));
+
+                return $this->tracer;
+            }
+
             return $this->tracerLocator->get($traceable->tracer);
         }
 
@@ -183,9 +204,9 @@ private function isAutoTraceable(): bool
         return InstrumentationTypeEnum::Auto === $this->instrumentationType;
     }
 
-    private function isAttributeTraceable(object $message): bool
+    private function isAttributeTraceable(?Traceable $traceable): bool
     {
         return InstrumentationTypeEnum::Attribute === $this->instrumentationType
-            && $this->parseAttribute($message) instanceof Traceable;
+            && null !== $traceable;
     }
 }

src/OpenTelemetry/Context/Propagator/PropagatorFactory.php

@@ -6,7 +6,7 @@
 use OpenTelemetry\API\Trace\Propagation\TraceContextPropagator;
 use OpenTelemetry\Context\Propagation\MultiTextMapPropagator;
 
-class PropagatorFactory
+final class PropagatorFactory
 {
     /**
      * Default propagators from https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#general-sdk-configuration.

src/OpenTelemetry/Context/Propagator/TraceStampPropagator.php

@@ -9,7 +9,11 @@
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Messenger\Envelope;
 
-readonly class TraceStampPropagator implements PropagationSetterInterface, PropagationGetterInterface
+/**
+ * Bridges OpenTelemetry's propagation interfaces with Symfony Messenger envelopes,
+ * reading and writing trace context via TraceStamp instances attached to the envelope.
+ */
+final readonly class TraceStampPropagator implements PropagationSetterInterface, PropagationGetterInterface
 {
     public function __construct(
         private ?LoggerInterface $logger = null,

src/Resources/config/services_tracing_instrumentation.php

@@ -15,6 +15,7 @@
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceableMessengerTransportFactory;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\WorkerMessageEventSubscriber;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Twig\TraceableTwigExtension;
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator\TraceStampPropagator;
 use Symfony\Component\DependencyInjection\Loader\Configurator\ContainerConfigurator;
 
 use function Symfony\Component\DependencyInjection\Loader\Configurator\service;
@@ -106,14 +107,19 @@
             ->tag('monolog.logger', ['channel' => 'open_telemetry'])
         ->alias('messenger.middleware.open_telemetry_tracer', 'open_telemetry.instrumentation.messenger.trace.middleware')
 
+        ->set('open_telemetry.instrumentation.messenger.trace_stamp_propagator', TraceStampPropagator::class)
+            ->tag('monolog.logger', ['channel' => 'open_telemetry'])
+
         ->set('open_telemetry.instrumentation.messenger.propagation.middleware', AddStampForPropagationMiddleware::class)
             ->arg('$propagator', service('open_telemetry.propagator_text_map.multi'))
+            ->arg('$traceStampPropagator', service('open_telemetry.instrumentation.messenger.trace_stamp_propagator'))
             ->tag('messenger.middleware')
             ->tag('monolog.logger', ['channel' => 'open_telemetry'])
 
         ->set('open_telemetry.instrumentation.messenger.worker', WorkerMessageEventSubscriber::class)
             ->arg('$propagator', service('open_telemetry.propagator_text_map.multi'))
             ->arg('$tracer', service('open_telemetry.traces.default_tracer'))
+            ->arg('$traceStampPropagator', service('open_telemetry.instrumentation.messenger.trace_stamp_propagator'))
             ->tag('kernel.event_subscriber')
             ->tag('container.service_subscriber')
             ->tag('monolog.logger', ['channel' => 'open_telemetry'])

tests/Functional/Instrumentation/Messenger/MessengerPropagationTracingTest.php

@@ -0,0 +1,104 @@
+<?php
+
+namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Tests\Functional\Instrumentation\Messenger;
+
+use App\Kernel;
+use App\Message\DummyMessage;
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceStamp;
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator\TraceStampPropagator;
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\Tests\Functional\TracingTestCaseTrait;
+use OpenTelemetry\API\Trace\SpanKind;
+use OpenTelemetry\API\Trace\TracerInterface;
+use OpenTelemetry\Context\Context;
+use OpenTelemetry\Context\Propagation\MultiTextMapPropagator;
+use OpenTelemetry\SDK\Trace\StatusData;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
+use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
+use Symfony\Component\Messenger\Stamp\BusNameStamp;
+use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
+use Zalas\PHPUnit\Globals\Attribute\Env;
+
+#[Env('KERNEL_CLASS', Kernel::class)]
+#[Env('APP_ENV', 'auto')]
+class MessengerPropagationTracingTest extends KernelTestCase
+{
+    use TracingTestCaseTrait;
+
+    private EventDispatcherInterface $eventDispatcher;
+
+    protected function setUp(): void
+    {
+        self::bootKernel();
+
+        $this->eventDispatcher = self::getContainer()->get('event_dispatcher');
+    }
+
+    public function testWorkerSpanIsChildOfProducerTrace(): void
+    {
+        /** @var TracerInterface $tracer */
+        $tracer = self::getContainer()->get('open_telemetry.traces.tracers.main');
+
+        // Simulate a producer context (e.g. an HTTP request or CLI command)
+        $producerSpan = $tracer->spanBuilder('producer-request')->startSpan();
+        $producerScope = $producerSpan->activate();
+
+        // Inject current trace context into an envelope via the propagator
+        $envelope = new Envelope(new DummyMessage('propagation-test'), [new BusNameStamp('messenger.bus.default')]);
+
+        /** @var MultiTextMapPropagator $propagator */
+        $propagator = self::getContainer()->get('open_telemetry.propagator_text_map.multi');
+
+        /** @var TraceStampPropagator $traceStampPropagator */
+        $traceStampPropagator = self::getContainer()->get('open_telemetry.instrumentation.messenger.trace_stamp_propagator');
+        $propagator->inject($envelope, $traceStampPropagator, Context::getCurrent());
+
+        // End the producer context before worker processing
+        $producerScope->detach();
+        $producerSpan->end();
+
+        // Assert the TraceStamp was injected
+        $traceStamp = $envelope->last(TraceStamp::class);
+        self::assertNotNull($traceStamp, 'TraceStamp should be present on the envelope after injection');
+        self::assertNotEmpty($traceStamp->getTraceParent());
+
+        // Simulate worker processing with the stamped envelope
+        $this->eventDispatcher->dispatch(new WorkerMessageReceivedEvent($envelope, 'main'));
+        $this->eventDispatcher->dispatch(new WorkerMessageHandledEvent($envelope, 'main'));
+
+        // Find the producer and consumer spans
+        $spans = self::getSpans();
+        self::assertCount(2, $spans);
+
+        $producerSpanData = null;
+        $consumerSpanData = null;
+        foreach ($spans as $span) {
+            if ('producer-request' === $span->getName()) {
+                $producerSpanData = $span;
+            }
+            if (SpanKind::KIND_CONSUMER === $span->getKind()) {
+                $consumerSpanData = $span;
+            }
+        }
+
+        self::assertNotNull($producerSpanData, 'Producer span should exist');
+        self::assertNotNull($consumerSpanData, 'Consumer span should exist');
+
+        // The consumer span must belong to the same trace
+        self::assertSame(
+            $producerSpanData->getContext()->getTraceId(),
+            $consumerSpanData->getContext()->getTraceId(),
+            'Worker span should belong to the same trace as the producer span'
+        );
+
+        // The consumer span must be a direct child of the producer span
+        self::assertSame(
+            $producerSpanData->getContext()->getSpanId(),
+            $consumerSpanData->getParentSpanId(),
+            'Worker span should be a child of the producer span'
+        );
+
+        self::assertSpanStatus($consumerSpanData, StatusData::ok());
+    }
+}