fix: restore tenant context for error channel in multi-tenant setup

packages/Dbal/tests/Integration/Recoverability/MultiTenantDeadLetterTest.php

@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Integration\Recoverability;
+
+use Ecotone\Dbal\DbalBackedMessageChannelBuilder;
+use Ecotone\Dbal\MultiTenant\MultiTenantConfiguration;
+use Ecotone\Dbal\Recoverability\DbalDeadLetterHandler;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Messaging\Attribute\Asynchronous;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use Interop\Queue\ConnectionFactory;
+use RuntimeException;
+use Test\Ecotone\Dbal\DbalMessagingTestCase;
+
+/**
+ * licence Apache-2.0
+ * @internal
+ */
+final class MultiTenantDeadLetterTest extends DbalMessagingTestCase
+{
+    public function test_failed_message_from_polling_consumer_is_stored_in_dead_letter_of_related_tenant(): void
+    {
+        $orderService = new class () {
+            #[Asynchronous('async')]
+            #[CommandHandler('order.place', endpointId: 'placeOrderEndpoint')]
+            public function placeOrder(string $order): void
+            {
+                throw new RuntimeException('Order processing has failed');
+            }
+        };
+
+        $ecotoneLite = EcotoneLite::bootstrapFlowTesting(
+            [$orderService::class],
+            [
+                $orderService,
+                'tenant_a_connection' => $this->connectionForTenantA(),
+                'tenant_b_connection' => $this->connectionForTenantB(),
+            ],
+            ServiceConfiguration::createWithDefaults()
+                ->withDefaultErrorChannel('dbal_dead_letter')
+                ->withExtensionObjects([
+                    DbalBackedMessageChannelBuilder::create('async'),
+                    MultiTenantConfiguration::create(
+                        'tenant',
+                        ['tenant_a' => 'tenant_a_connection', 'tenant_b' => 'tenant_b_connection'],
+                    ),
+                ])
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::ASYNCHRONOUS_PACKAGE, ModulePackageList::DBAL_PACKAGE])),
+        );
+
+        $ecotoneLite->sendCommandWithRoutingKey('order.place', 'milk', metadata: ['tenant' => 'tenant_a']);
+
+        $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup(amountOfMessagesToHandle: 2, maxExecutionTimeInMilliseconds: 1000, failAtError: false));
+
+        $this->assertSame(1, $this->amountOfDeadLetterMessagesIn($this->connectionForTenantA()));
+        $this->assertSame(0, $this->amountOfDeadLetterMessagesIn($this->connectionForTenantB()));
+    }
+
+    private function amountOfDeadLetterMessagesIn(ConnectionFactory $connectionFactory): int
+    {
+        $connection = $connectionFactory->createContext()->getDbalConnection();
+
+        if (! self::checkIfTableExists($connection, DbalDeadLetterHandler::DEFAULT_DEAD_LETTER_TABLE)) {
+            return 0;
+        }
+
+        return (int) $connection
+            ->executeQuery(sprintf('SELECT COUNT(*) FROM %s', DbalDeadLetterHandler::DEFAULT_DEAD_LETTER_TABLE))
+            ->fetchOne();
+    }
+}

packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/BasicMessagingModule.php

@@ -133,6 +133,7 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO
                     MediaType::parseMediaType($serviceConfiguration->getDefaultSerializationMediaType()),
                 ]),
                 Reference::to(ConversionService::REFERENCE_NAME),
+                Reference::to(MessageHeadersPropagatorInterceptor::class),
             ])
         );
 

packages/Ecotone/src/Messaging/Handler/Gateway/ErrorChannelService.php

@@ -12,6 +12,7 @@
 use Ecotone\Messaging\MessageHeaders;
 use Ecotone\Messaging\Support\ErrorMessage;
 use Ecotone\Messaging\Support\MessageBuilder;
+use Ecotone\Modelling\MessageHandling\MetadataPropagator\MessageHeadersPropagatorInterceptor;
 use Throwable;
 
 /**
@@ -23,6 +24,7 @@ public function __construct(
         private LoggingGateway $loggingGateway,
         private OutboundMessageConverter $outboundMessageConverter,
         private ConversionService $conversionService,
+        private MessageHeadersPropagatorInterceptor $messageHeadersPropagator,
     ) {
     }
 
@@ -51,11 +53,14 @@ public function handle(
             $messageBuilder = $messageBuilder->prependRoutingSlip([$routingSlip]);
         }
 
-        $errorChannel->send(
-            ErrorMessage::create(
-                $messageBuilder->build(),
-                $cause
-            )
+        $this->messageHeadersPropagator->storeHeaders(
+            fn () => $errorChannel->send(
+                ErrorMessage::create(
+                    $messageBuilder->build(),
+                    $cause
+                )
+            ),
+            $requestMessage,
         );
 
         $this->loggingGateway->info(