fix: prevent OpenTelemetry scope leaks masking original exceptions

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;
         }
     }
 

packages/OpenTelemetry/src/ChannelSendSpan.php

@@ -0,0 +1,49 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\OpenTelemetry;
+
+use OpenTelemetry\API\Trace\SpanInterface;
+use OpenTelemetry\API\Trace\StatusCode;
+use OpenTelemetry\Context\ScopeInterface;
+use Throwable;
+
+/**
+ * licence Apache-2.0
+ */
+final class ChannelSendSpan
+{
+    private bool $isClosed = false;
+
+    public function __construct(private SpanInterface $span, private ScopeInterface $scope)
+    {
+    }
+
+    public function closeWithSuccess(): void
+    {
+        $this->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();
+    }
+}

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();

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;
     }