IBX-11974: Replaced backported message deduplication with Symfony native implementation and lock release listener

composer.json

@@ -15,6 +15,7 @@
         "symfony/event-dispatcher": "^7.4",
         "symfony/http-foundation": "^7.4",
         "symfony/http-kernel": "^7.4",
+        "symfony/lock": "^7.4",
         "symfony/messenger": "^7.4",
         "symfony/yaml": "^7.4"
     },

phpstan-baseline.neon

@@ -48,36 +48,6 @@ parameters:
 			count: 1
 			path: src/bundle/Lock/Store/DoctrineDbalStore.php
 
-		-
-			message: '#^Parameter \#1 \$data \(array\{key\: mixed, ttl\?\: float\|null, only_deduplicate_in_queue\?\: bool\|null\}\) of method Ibexa\\Bundle\\Messenger\\Serializer\\Normalizer\\DeduplicateStampNormalizer\:\:denormalize\(\) should be contravariant with parameter \$data \(mixed\) of method Symfony\\Component\\Serializer\\Normalizer\\DenormalizerInterface\:\:denormalize\(\)$#'
-			identifier: method.childParameterType
-			count: 1
-			path: src/bundle/Serializer/Normalizer/DeduplicateStampNormalizer.php
-
-		-
-			message: '#^Return type \(Ibexa\\Contracts\\Messenger\\Stamp\\DeduplicateStamp\) of method Ibexa\\Bundle\\Messenger\\Serializer\\Normalizer\\DeduplicateStampNormalizer\:\:denormalize\(\) should be covariant with return type \(\(\$type is class\-string\<object\> \? object \: mixed\)\) of method Symfony\\Component\\Serializer\\Normalizer\\DenormalizerInterface\:\:denormalize\(\)$#'
-			identifier: method.childReturnType
-			count: 1
-			path: src/bundle/Serializer/Normalizer/DeduplicateStampNormalizer.php
-
-		-
-			message: '#^Cannot access offset string on mixed\.$#'
-			identifier: offsetAccess.nonOffsetAccessible
-			count: 1
-			path: src/bundle/Serializer/Normalizer/LockKeyNormalizer.php
-
-		-
-			message: '#^Return type \(Symfony\\Component\\Lock\\Key\) of method Ibexa\\Bundle\\Messenger\\Serializer\\Normalizer\\LockKeyNormalizer\:\:denormalize\(\) should be covariant with return type \(\(\$type is class\-string\<object\> \? object \: mixed\)\) of method Symfony\\Component\\Serializer\\Normalizer\\DenormalizerInterface\:\:denormalize\(\)$#'
-			identifier: method.childReturnType
-			count: 1
-			path: src/bundle/Serializer/Normalizer/LockKeyNormalizer.php
-
-		-
-			message: '#^Variable property access on Symfony\\Component\\Lock\\Key\.$#'
-			identifier: property.dynamicName
-			count: 1
-			path: src/bundle/Serializer/Normalizer/LockKeyNormalizer.php
-
 		-
 			message: '#^Method Ibexa\\Bundle\\Messenger\\Transport\\Sender\\SendersLocator\:\:listTypes\(\) should return array\<class\-string\> but returns array\.$#'
 			identifier: return.type

src/bundle/DependencyInjection/IbexaMessengerExtension.php

@@ -8,10 +8,9 @@
 
 namespace Ibexa\Bundle\Messenger\DependencyInjection;
 
-use Ibexa\Bundle\Messenger\Middleware\DeduplicateMiddleware;
+use Ibexa\Bundle\Messenger\EventListener\ReleaseDeduplicationLockOnFailureListener;
 use Ibexa\Bundle\Messenger\Middleware\SudoMiddleware;
 use Ibexa\Bundle\Messenger\Middleware\UserPermissionMiddleware;
-use Ibexa\Bundle\Messenger\Serializer\Normalizer\LockKeyNormalizer;
 use Ibexa\Contracts\Messenger\Transport\MessageProviderInterface;
 use LogicException;
 use Symfony\Component\Config\FileLocator;
@@ -22,7 +21,6 @@
 use Symfony\Component\DependencyInjection\Reference;
 use Symfony\Component\HttpKernel\DependencyInjection\ConfigurableExtension;
 use Symfony\Component\Lock\PersistingStoreInterface;
-use Symfony\Component\Lock\Serializer\LockNormalizer as SymfonyLockNormalizer;
 use Symfony\Component\Lock\Store\StoreFactory;
 use Symfony\Component\Stopwatch\Stopwatch;
 
@@ -60,7 +58,6 @@ protected function loadInternal(
         $loader->load('services.yaml');
 
         $this->configureLockStorage($mergedConfig['deduplication_lock_storage'], $container);
-        $this->configureLockNormalizerBackport($container);
         $this->registerMessengerConfiguration($mergedConfig, $container);
 
         if ($this->shouldLoadTestServices($container)) {
@@ -128,7 +125,7 @@ private function registerMessengerConfiguration(
         ];
 
         if ($mergedConfig['deduplication_lock_storage']['enabled'] === true) {
-            $middleware[] = ['id' => DeduplicateMiddleware::class];
+            $middleware[] = ['id' => 'ibexa.messenger.deduplicate_middleware'];
         }
 
         $middleware = array_merge(
@@ -153,13 +150,15 @@ private function configureLockStorage(
         ContainerBuilder $container
     ): void {
         if ($lockStorageConfig['enabled'] === false) {
-            $container->removeDefinition(DeduplicateMiddleware::class);
+            $container->removeDefinition('ibexa.messenger.deduplicate_middleware');
             $container->removeDefinition('ibexa.messenger.lock_factory');
             $container->removeDefinition('ibexa.messenger.lock_store.dbal');
 
             return;
         }
 
+        $this->registerDeduplicationFailureListener($container);
+
         $lockStorageType = $lockStorageConfig['type'];
         if ($lockStorageType === 'doctrine') {
             $storeDefinition = new Reference('ibexa.messenger.lock_store.dbal');
@@ -200,14 +199,25 @@ private function configureLockStorage(
         ));
     }
 
-    private function configureLockNormalizerBackport(ContainerBuilder $container): void
+    private function registerDeduplicationFailureListener(ContainerBuilder $container): void
     {
-        // Symfony 7.4 contains proper implementation
-        if (class_exists(SymfonyLockNormalizer::class)) {
-            $container->removeDefinition(LockKeyNormalizer::class);
-            $definition = new Definition(SymfonyLockNormalizer::class);
-            $definition->addTag('ibexa.messenger.serializer.normalizer', ['priority' => -60]);
-            $container->setDefinition('ibexa.messenger.lock_normalizer', $definition);
-        }
+        // Release the deduplication lock when a message definitively fails, so a new dispatch of
+        // the same key is not blocked until the lock's TTL expires. Prefer Symfony's native listener
+        // (available since Symfony 8.1); fall back to our backport otherwise. Either way it must be
+        // wired to our lock factory, because Symfony's own native listener uses the default
+        // "lock.factory" and would not release locks stored in the "ibexa_messenger_lock_keys" table.
+        // @todo Remove the backport class and this fallback once the minimum Symfony version is >= 8.1.
+        $nativeListenerClass = 'Symfony\Component\Messenger\EventListener\ReleaseDeduplicationLockOnFailureListener';
+        $listenerClass = class_exists($nativeListenerClass)
+            ? $nativeListenerClass
+            : ReleaseDeduplicationLockOnFailureListener::class;
+
+        $definition = new Definition($listenerClass);
+        $definition->setArgument(0, new Reference('ibexa.messenger.lock_factory'));
+        $definition->addTag('kernel.event_subscriber');
+        $container->setDefinition(
+            'ibexa.messenger.release_deduplication_lock_on_failure_listener',
+            $definition,
+        );
     }
 }

src/bundle/EventListener/ReleaseDeduplicationLockOnFailureListener.php

@@ -0,0 +1,55 @@
+<?php
+
+/**
+ * @copyright Copyright (C) Ibexa AS. All rights reserved.
+ * @license For full copyright and license information view LICENSE file distributed with this source code.
+ */
+declare(strict_types=1);
+
+namespace Ibexa\Bundle\Messenger\EventListener;
+
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use Symfony\Component\Messenger\Stamp\DeduplicateStamp;
+
+/**
+ * Backport of Symfony's ReleaseDeduplicationLockOnFailureListener (introduced in Symfony 8.1).
+ * The DeduplicateMiddleware keeps the lock held when a handler throws, so a message still being
+ * retried cannot be enqueued again. Once the retry flow has decided not to retry, the lock must
+ * be released to unblock future messages sharing the same key.
+ *
+ * Original code: https://github.com/symfony/symfony/blob/8.1/src/Symfony/Component/Messenger/EventListener/ReleaseDeduplicationLockOnFailureListener.php
+ *
+ * @todo Remove this backport once the minimum Symfony version is >= 8.1.
+ */
+final class ReleaseDeduplicationLockOnFailureListener implements EventSubscriberInterface
+{
+    public function __construct(private LockFactory $lockFactory) {}
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            // Must have lower priority than SendFailedMessageForRetryListener (100) so willRetry() is already set.
+            WorkerMessageFailedEvent::class => ['onMessageFailed', -10],
+        ];
+    }
+
+    public function onMessageFailed(WorkerMessageFailedEvent $event): void
+    {
+        if ($event->willRetry()) {
+            return;
+        }
+
+        $stamp = $event->getEnvelope()->last(DeduplicateStamp::class);
+        if ($stamp === null) {
+            return;
+        }
+
+        if ($stamp->onlyDeduplicateInQueue()) {
+            return;
+        }
+
+        $this->lockFactory->createLockFromKey($stamp->getKey())->release();
+    }
+}

src/bundle/Resources/config/services/buses.yaml

@@ -25,7 +25,8 @@ services:
 
     Ibexa\Bundle\Messenger\Middleware\SudoMiddleware: ~
 
-    Ibexa\Bundle\Messenger\Middleware\DeduplicateMiddleware:
+    ibexa.messenger.deduplicate_middleware:
+        class: Symfony\Component\Messenger\Middleware\DeduplicateMiddleware
         arguments:
             - '@ibexa.messenger.lock_factory'
 

src/bundle/Resources/config/services/serializer/normalizers.yaml

@@ -4,11 +4,8 @@ services:
         autoconfigure: false
         public: false
 
-    Ibexa\Bundle\Messenger\Serializer\Normalizer\DeduplicateStampNormalizer:
-        tags:
-            - { name: ibexa.messenger.serializer.normalizer, priority: -60 }
-
-    Ibexa\Bundle\Messenger\Serializer\Normalizer\LockKeyNormalizer:
+    ibexa.messenger.serializer.normalizer.lock_key_normalizer:
+        class: Symfony\Component\Lock\Serializer\LockKeyNormalizer
         tags:
             - { name: ibexa.messenger.serializer.normalizer, priority: -60 }