From 6405f8ca1ee09c95f5120fc901e0dcf566b7162c Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Thu, 2 Jul 2026 21:21:52 +0200 Subject: [PATCH] fix: prevent OpenTelemetry scope leaks masking original exceptions When a message send failed inside a traced flow, the scope activated in TracingChannelInterceptor::preSend was never detached, causing 'unexpected call to Scope::detach()' notices that replaced the original exception under Symfony's error handler. - SendingInterceptorAdapter guarantees exactly one afterSendCompletion per executed preSend on every path (success, failure, preSend failure, message drop), with per-interceptor guarded cleanup - TracingChannelInterceptor tracks open spans in a per-channel LIFO stack instead of a temporary message header - TracerInterceptor uses try/finally for the consumer scope and closes the Distributed Bus span on failure Fixes #419 --- .../Channel/SendingInterceptorAdapter.php | 72 +++- .../OpenTelemetry/src/ChannelSendSpan.php | 49 +++ .../OpenTelemetry/src/TracerInterceptor.php | 21 +- .../src/TracingChannelInterceptor.php | 22 +- .../Integration/TracingScopeCleanupTest.php | 313 ++++++++++++++++++ 5 files changed, 442 insertions(+), 35 deletions(-) create mode 100644 packages/OpenTelemetry/src/ChannelSendSpan.php create mode 100644 packages/OpenTelemetry/tests/Integration/TracingScopeCleanupTest.php diff --git a/packages/Ecotone/src/Messaging/Channel/SendingInterceptorAdapter.php b/packages/Ecotone/src/Messaging/Channel/SendingInterceptorAdapter.php index 7754b1ddc..91a887bd4 100644 --- a/packages/Ecotone/src/Messaging/Channel/SendingInterceptorAdapter.php +++ b/packages/Ecotone/src/Messaging/Channel/SendingInterceptorAdapter.php @@ -49,32 +49,74 @@ public function __construct(MessageChannel $messageChannel, array $sortedChannel public function send(Message $message): void { $messageToSend = $message; - foreach ($this->sortedChannelInterceptors as $channelInterceptor) { - $messageToSend = $channelInterceptor->preSend($messageToSend, $this->messageChannel); + $executedInterceptors = []; + $isMessageDropped = false; + try { + foreach ($this->sortedChannelInterceptors as $channelInterceptor) { + $transformedMessage = $channelInterceptor->preSend($messageToSend, $this->messageChannel); + + if ($transformedMessage === null) { + $isMessageDropped = true; - if ($messageToSend === null) { - return; + break; + } + + $messageToSend = $transformedMessage; + $executedInterceptors[] = $channelInterceptor; } - } - $exception = null; - try { - $this->messageChannel->send($messageToSend); + if (! $isMessageDropped) { + $this->messageChannel->send($messageToSend); + } } catch (Throwable $exception) { - } finally { - $shouldThrow = $exception !== null; - foreach ($this->sortedChannelInterceptors as $channelInterceptor) { - if ($channelInterceptor->afterSendCompletion($messageToSend, $this->messageChannel, $exception)) { - $shouldThrow = false; + $shouldThrow = true; + $firstCleanupFailure = null; + foreach ($executedInterceptors as $channelInterceptor) { + try { + if ($channelInterceptor->afterSendCompletion($messageToSend, $this->messageChannel, $exception)) { + $shouldThrow = false; + } + } catch (Throwable $cleanupFailure) { + $firstCleanupFailure ??= $cleanupFailure; } } if ($shouldThrow) { throw $exception; } + + $this->executePostSend($messageToSend, $executedInterceptors, $firstCleanupFailure); + + return; } - foreach ($this->sortedChannelInterceptors as $channelInterceptor) { - $channelInterceptor->postSend($messageToSend, $this->messageChannel); + + $firstCleanupFailure = null; + foreach ($executedInterceptors as $channelInterceptor) { + try { + $channelInterceptor->afterSendCompletion($messageToSend, $this->messageChannel, null); + } catch (Throwable $cleanupFailure) { + $firstCleanupFailure ??= $cleanupFailure; + } + } + + $this->executePostSend($messageToSend, $executedInterceptors, $firstCleanupFailure); + } + + /** + * @param ChannelInterceptor[] $executedInterceptors + */ + private function executePostSend(Message $messageToSend, array $executedInterceptors, ?Throwable $firstCleanupFailure): void + { + foreach ($executedInterceptors as $channelInterceptor) { + try { + $channelInterceptor->postSend($messageToSend, $this->messageChannel); + } catch (Throwable $cleanupFailure) { + $firstCleanupFailure ??= $cleanupFailure; + } + } + + if ($firstCleanupFailure !== null) { + throw $firstCleanupFailure; } } diff --git a/packages/OpenTelemetry/src/ChannelSendSpan.php b/packages/OpenTelemetry/src/ChannelSendSpan.php new file mode 100644 index 000000000..39ec6868a --- /dev/null +++ b/packages/OpenTelemetry/src/ChannelSendSpan.php @@ -0,0 +1,49 @@ +close(StatusCode::STATUS_OK, null); + } + + public function closeWithFailure(Throwable $exception): void + { + if ($this->isClosed) { + return; + } + + $this->span->recordException($exception); + $this->close(StatusCode::STATUS_ERROR, $exception->getMessage()); + } + + private function close(string $statusCode, ?string $statusDescription): void + { + if ($this->isClosed) { + return; + } + $this->isClosed = true; + + $this->span->setStatus($statusCode, $statusDescription); + $this->scope->detach(); + $this->span->end(); + } +} diff --git a/packages/OpenTelemetry/src/TracerInterceptor.php b/packages/OpenTelemetry/src/TracerInterceptor.php index 3db50d770..d20116316 100644 --- a/packages/OpenTelemetry/src/TracerInterceptor.php +++ b/packages/OpenTelemetry/src/TracerInterceptor.php @@ -40,7 +40,7 @@ public function traceAsynchronousEndpoint(MethodInvocation $methodInvocation, Me $scope = $parentContext->activate(); try { - $trace = $this->trace( + return $this->trace( $message->getHeaders()->containsKey(MessageHeaders::POLLED_CHANNEL_NAME) ? 'Receiving from channel: ' . $message->getHeaders()->get(MessageHeaders::POLLED_CHANNEL_NAME) : ($message->getHeaders()->containsKey(MessageHeaders::INBOUND_REQUEST_CHANNEL) @@ -50,14 +50,9 @@ public function traceAsynchronousEndpoint(MethodInvocation $methodInvocation, Me $message, spanKind: SpanKind::KIND_CONSUMER, ); - } catch (Throwable $exception) { + } finally { $scope->detach(); - - throw $exception; } - $scope->detach(); - - return $trace; } public function provideContextForDistributedBus(Message $message): Message @@ -77,7 +72,14 @@ public function traceDistributedBus(MethodInvocation $methodInvocation, Message ->startSpan(); $spanScope = $span->activate(); - $result = $methodInvocation->proceed(); + try { + $result = $methodInvocation->proceed(); + } catch (Throwable $exception) { + $span->recordException($exception); + $this->closeSpan($span, $spanScope, StatusCode::STATUS_ERROR, $exception->getMessage()); + + throw $exception; + } $this->closeSpan($span, $spanScope, StatusCode::STATUS_OK, null); @@ -179,9 +181,6 @@ public function trace( private function closeSpan(SpanInterface $span, ScopeInterface $spanScope, string $statusCode, ?string $descriptionStatusCode): void { - $currentSpan = Span::getCurrent(); - $currentContext = Context::getCurrent(); - $span->setStatus($statusCode, $descriptionStatusCode); $spanScope->detach(); $span->end(); diff --git a/packages/OpenTelemetry/src/TracingChannelInterceptor.php b/packages/OpenTelemetry/src/TracingChannelInterceptor.php index 5d7008131..e7c741417 100644 --- a/packages/OpenTelemetry/src/TracingChannelInterceptor.php +++ b/packages/OpenTelemetry/src/TracingChannelInterceptor.php @@ -7,7 +7,6 @@ use Ecotone\Messaging\Channel\ChannelInterceptor; use Ecotone\Messaging\Message; use Ecotone\Messaging\MessageChannel; -use Ecotone\Messaging\MessageHeaders; use Ecotone\Messaging\Support\MessageBuilder; use function json_decode; @@ -18,7 +17,6 @@ use OpenTelemetry\API\Trace\StatusCode; use OpenTelemetry\API\Trace\TracerProviderInterface; use OpenTelemetry\Context\Context; -use OpenTelemetry\SDK\Trace\Span; use Throwable; /** @@ -28,6 +26,9 @@ final class TracingChannelInterceptor implements ChannelInterceptor { public const TRACING_CARRIER_HEADER = 'ecotoneTracingCarrier'; + /** @var ChannelSendSpan[] */ + private array $openSpans = []; + public function __construct(private string $channelName, private TracerProviderInterface $tracerProvider) { } @@ -38,28 +39,31 @@ public function preSend(Message $message, MessageChannel $messageChannel): ?Mess ->startSpan(); $scope = $span->activate(); + $this->openSpans[] = new ChannelSendSpan($span, $scope); + $ctx = $span->storeInContext(Context::getCurrent()); $carrier = []; TraceContextPropagator::getInstance()->inject($carrier, null, $ctx); return MessageBuilder::fromMessage($message) ->setHeader(self::TRACING_CARRIER_HEADER, json_encode($carrier)) - ->setHeader(MessageHeaders::TEMPORARY_SPAN_CONTEXT_HEADER, $scope) ->build(); } public function postSend(Message $message, MessageChannel $messageChannel): void { - // @TODO Remove header from message after notice are stopped from OpenTelemetry (https://github.com/open-telemetry/opentelemetry-php/issues/1138) - $currentContext = $message->getHeaders()->get(MessageHeaders::TEMPORARY_SPAN_CONTEXT_HEADER); - // $currentContext = Context::storage()->scope(); - $currentRelatedSpan = Span::getCurrent(); - $currentContext->detach(); - $currentRelatedSpan->end(); } public function afterSendCompletion(Message $message, MessageChannel $messageChannel, ?Throwable $exception): bool { + $channelSendSpan = array_pop($this->openSpans); + + if ($exception !== null) { + $channelSendSpan?->closeWithFailure($exception); + } else { + $channelSendSpan?->closeWithSuccess(); + } + return false; } diff --git a/packages/OpenTelemetry/tests/Integration/TracingScopeCleanupTest.php b/packages/OpenTelemetry/tests/Integration/TracingScopeCleanupTest.php new file mode 100644 index 000000000..753bf8f73 --- /dev/null +++ b/packages/OpenTelemetry/tests/Integration/TracingScopeCleanupTest.php @@ -0,0 +1,313 @@ +bootstrapWithEventForwardedToSecondChannel( + channelsToRegister: [ + SimpleMessageChannelBuilder::createQueueChannel('async_channel'), + ExceptionalQueueChannel::createWithExceptionOnSend('failing_channel'), + ], + ); + + $ecotoneLite->sendCommandWithRoutingKey('user.register', '1'); + + [$caughtException, $scopeNotices] = $this->invokeCapturingScopeNotices( + fn () => $ecotoneLite->run('async_channel', ExecutionPollingMetadata::createWithTestingSetup()) + ); + + $this->assertNotNull($caughtException); + $this->assertStringContainsString('Exception on send', $caughtException->getMessage()); + $this->assertSame([], $scopeNotices); + } + + public function test_no_scope_detach_notices_when_channel_interceptor_fails_before_send(): void + { + $throwingChannelInterceptor = new class () extends AbstractChannelInterceptor { + public function preSend(Message $message, MessageChannel $messageChannel): ?Message + { + throw new RuntimeException('Exception before send'); + } + }; + + $ecotoneLite = $this->bootstrapWithEventForwardedToSecondChannel( + channelsToRegister: [ + SimpleMessageChannelBuilder::createQueueChannel('async_channel'), + SimpleMessageChannelBuilder::createQueueChannel('failing_channel'), + SimpleChannelInterceptorBuilder::create('failing_channel', 'throwingChannelInterceptor'), + ], + servicesToRegister: ['throwingChannelInterceptor' => $throwingChannelInterceptor], + ); + + $ecotoneLite->sendCommandWithRoutingKey('user.register', '1'); + + [$caughtException, $scopeNotices] = $this->invokeCapturingScopeNotices( + fn () => $ecotoneLite->run('async_channel', ExecutionPollingMetadata::createWithTestingSetup()) + ); + + $this->assertNotNull($caughtException); + $this->assertStringContainsString('Exception before send', $caughtException->getMessage()); + $this->assertSame([], $scopeNotices); + } + + public function test_no_scope_detach_notices_when_message_is_filtered_out_after_tracing_started(): void + { + $filteringChannelInterceptor = new class () extends AbstractChannelInterceptor { + public function preSend(Message $message, MessageChannel $messageChannel): ?Message + { + return null; + } + }; + + $ecotoneLite = $this->bootstrapWithEventForwardedToSecondChannel( + channelsToRegister: [ + SimpleMessageChannelBuilder::createQueueChannel('async_channel'), + SimpleMessageChannelBuilder::createQueueChannel('failing_channel'), + SimpleChannelInterceptorBuilder::create('failing_channel', 'filteringChannelInterceptor'), + ], + servicesToRegister: ['filteringChannelInterceptor' => $filteringChannelInterceptor], + ); + + $ecotoneLite->sendCommandWithRoutingKey('user.register', '1'); + + [$caughtException, $scopeNotices] = $this->invokeCapturingScopeNotices( + fn () => $ecotoneLite->run('async_channel', ExecutionPollingMetadata::createWithTestingSetup()) + ); + + $this->assertNull($caughtException); + $this->assertSame([], $scopeNotices); + } + + public function test_no_scope_detach_notices_when_cleanup_of_another_interceptor_fails_after_send_failure(): void + { + $throwingCleanupInterceptor = new class () extends AbstractChannelInterceptor { + public function afterSendCompletion(Message $message, MessageChannel $messageChannel, ?Throwable $exception): bool + { + if ($exception !== null) { + throw new RuntimeException('Exception during cleanup'); + } + + return false; + } + }; + + $ecotoneLite = $this->bootstrapWithEventForwardedToSecondChannel( + channelsToRegister: [ + SimpleMessageChannelBuilder::createQueueChannel('async_channel'), + ExceptionalQueueChannel::createWithExceptionOnSend('failing_channel'), + SimpleChannelInterceptorBuilder::create('failing_channel', 'throwingCleanupInterceptor')->withPrecedence(500), + ], + servicesToRegister: ['throwingCleanupInterceptor' => $throwingCleanupInterceptor], + ); + + $ecotoneLite->sendCommandWithRoutingKey('user.register', '1'); + + [$caughtException, $scopeNotices] = $this->invokeCapturingScopeNotices( + fn () => $ecotoneLite->run('async_channel', ExecutionPollingMetadata::createWithTestingSetup()) + ); + + $this->assertNotNull($caughtException); + $this->assertSame([], $scopeNotices); + } + + public function test_no_scope_detach_notices_when_post_send_of_another_interceptor_fails(): void + { + $throwingPostSendInterceptor = new class () extends AbstractChannelInterceptor { + public function postSend(Message $message, MessageChannel $messageChannel): void + { + throw new RuntimeException('Exception after send'); + } + }; + + $ecotoneLite = $this->bootstrapWithEventForwardedToSecondChannel( + channelsToRegister: [ + SimpleMessageChannelBuilder::createQueueChannel('async_channel'), + SimpleMessageChannelBuilder::createQueueChannel('failing_channel'), + SimpleChannelInterceptorBuilder::create('failing_channel', 'throwingPostSendInterceptor')->withPrecedence(500), + ], + servicesToRegister: ['throwingPostSendInterceptor' => $throwingPostSendInterceptor], + ); + + $ecotoneLite->sendCommandWithRoutingKey('user.register', '1'); + + [$caughtException, $scopeNotices] = $this->invokeCapturingScopeNotices( + fn () => $ecotoneLite->run('async_channel', ExecutionPollingMetadata::createWithTestingSetup()) + ); + + $this->assertNotNull($caughtException); + $this->assertStringContainsString('Exception after send', $caughtException->getMessage()); + $this->assertSame([], $scopeNotices); + } + + public function test_no_scope_detach_notices_when_interceptor_replaces_message_dropping_framework_headers(): void + { + $rebuildingChannelInterceptor = new class () extends AbstractChannelInterceptor { + public function preSend(Message $message, MessageChannel $messageChannel): ?Message + { + return MessageBuilder::withPayload($message->getPayload())->build(); + } + }; + + $ecotoneLite = $this->bootstrapWithEventForwardedToSecondChannel( + channelsToRegister: [ + SimpleMessageChannelBuilder::createQueueChannel('async_channel'), + SimpleMessageChannelBuilder::createQueueChannel('failing_channel'), + SimpleChannelInterceptorBuilder::create('failing_channel', 'rebuildingChannelInterceptor'), + ], + servicesToRegister: ['rebuildingChannelInterceptor' => $rebuildingChannelInterceptor], + ); + + $ecotoneLite->sendCommandWithRoutingKey('user.register', '1'); + + [$caughtException, $scopeNotices] = $this->invokeCapturingScopeNotices( + fn () => $ecotoneLite->run('async_channel', ExecutionPollingMetadata::createWithTestingSetup()) + ); + + $this->assertNull($caughtException); + $this->assertSame([], $scopeNotices); + } + + public function test_no_scope_detach_notices_when_error_channel_send_fails(): void + { + $userService = new class () { + #[Asynchronous('async_channel')] + #[CommandHandler('user.register', endpointId: 'userRegisterEndpoint')] + public function register(string $userId): void + { + throw new RuntimeException('Handler failure'); + } + }; + + $ecotoneLite = EcotoneLite::bootstrapFlowTesting( + [$userService::class], + [$userService, TracerProviderInterface::class => TracingTestCase::prepareTracer(new InMemoryExporter())], + ServiceConfiguration::createWithDefaults() + ->withDefaultErrorChannel('error_channel') + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::TRACING_PACKAGE, ModulePackageList::ASYNCHRONOUS_PACKAGE])) + ->withExtensionObjects([ + SimpleMessageChannelBuilder::createQueueChannel('async_channel'), + ExceptionalQueueChannel::createWithExceptionOnSend('error_channel'), + ]) + ); + + $ecotoneLite->sendCommandWithRoutingKey('user.register', '1'); + + [, $scopeNotices] = $this->invokeCapturingScopeNotices( + fn () => $ecotoneLite->run('async_channel', ExecutionPollingMetadata::createWithTestingSetup(failAtError: false)) + ); + + $this->assertSame([], $scopeNotices); + } + + public function test_no_scope_detach_notices_when_distributed_bus_send_fails(): void + { + $ecotoneLite = EcotoneLite::bootstrapFlowTesting( + [], + [TracerProviderInterface::class => TracingTestCase::prepareTracer(new InMemoryExporter())], + ServiceConfiguration::createWithDefaults() + ->withServiceName('user_service') + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::TRACING_PACKAGE, ModulePackageList::ASYNCHRONOUS_PACKAGE])) + ->withExtensionObjects([ + ExceptionalQueueChannel::createWithExceptionOnSend('distributed_channel'), + DistributedServiceMap::initialize()->withCommandMapping(targetServiceName: 'ticket_service', channelName: 'distributed_channel'), + ]), + licenceKey: LicenceTesting::VALID_LICENCE, + ); + + [$caughtException, $scopeNotices] = $this->invokeCapturingScopeNotices( + fn () => $ecotoneLite->getDistributedBus()->sendCommand('ticket_service', 'ticket.create', 'User changed billing address') + ); + + $this->assertNotNull($caughtException); + $this->assertStringContainsString('Exception on send', $caughtException->getMessage()); + $this->assertSame([], $scopeNotices); + } + + private function bootstrapWithEventForwardedToSecondChannel(array $channelsToRegister, array $servicesToRegister = []): FlowTestSupport + { + $userService = new class () { + #[Asynchronous('async_channel')] + #[CommandHandler('user.register', endpointId: 'userRegisterEndpoint')] + public function register(string $userId, EventBus $eventBus): void + { + $eventBus->publishWithRouting('user.registered', $userId); + } + }; + $userNotifier = new class () { + #[Asynchronous('failing_channel')] + #[EventHandler('user.registered', endpointId: 'userNotifierEndpoint')] + public function notify(string $userId): void + { + } + }; + + return EcotoneLite::bootstrapFlowTesting( + [$userService::class, $userNotifier::class], + array_merge( + [$userService, $userNotifier, TracerProviderInterface::class => TracingTestCase::prepareTracer(new InMemoryExporter())], + $servicesToRegister, + ), + ServiceConfiguration::createWithDefaults() + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::TRACING_PACKAGE, ModulePackageList::ASYNCHRONOUS_PACKAGE])) + ->withExtensionObjects($channelsToRegister) + ); + } + + /** + * @return array{0: ?Throwable, 1: string[]} + */ + private function invokeCapturingScopeNotices(callable $action): array + { + $scopeNotices = []; + set_error_handler(static function (int $errorNumber, string $errorMessage) use (&$scopeNotices): bool { + $scopeNotices[] = $errorMessage; + + return true; + }, E_USER_NOTICE); + + $caughtException = null; + try { + $action(); + } catch (Throwable $exception) { + $caughtException = $exception; + } finally { + restore_error_handler(); + } + + return [$caughtException, $scopeNotices]; + } +}