diff --git a/composer.json b/composer.json index 6865dce..8efcda2 100644 --- a/composer.json +++ b/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" }, diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon index 593d27d..777d5d5 100644 --- a/phpstan-baseline.neon +++ b/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 \: 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 \: 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\ but returns array\.$#' identifier: return.type diff --git a/src/bundle/DependencyInjection/IbexaMessengerExtension.php b/src/bundle/DependencyInjection/IbexaMessengerExtension.php index 37fb942..63c2a8a 100644 --- a/src/bundle/DependencyInjection/IbexaMessengerExtension.php +++ b/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, + ); } } diff --git a/src/bundle/EventListener/ReleaseDeduplicationLockOnFailureListener.php b/src/bundle/EventListener/ReleaseDeduplicationLockOnFailureListener.php new file mode 100644 index 0000000..bd4f91a --- /dev/null +++ b/src/bundle/EventListener/ReleaseDeduplicationLockOnFailureListener.php @@ -0,0 +1,55 @@ += 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(); + } +} diff --git a/src/bundle/Middleware/DeduplicateMiddleware.php b/src/bundle/Middleware/DeduplicateMiddleware.php deleted file mode 100644 index 2e51fad..0000000 --- a/src/bundle/Middleware/DeduplicateMiddleware.php +++ /dev/null @@ -1,67 +0,0 @@ -. - * - * Original code: https://github.com/symfony/symfony/blob/7.3/src/Symfony/Component/Messenger/Middleware/DeduplicateMiddleware.php - */ -final class DeduplicateMiddleware implements MiddlewareInterface -{ - private LockFactory $lockFactory; - - public function __construct(LockFactory $lockFactory) - { - $this->lockFactory = $lockFactory; - } - - public function handle( - Envelope $envelope, - StackInterface $stack - ): Envelope { - $deduplicateStamp = $envelope->last(DeduplicateStamp::class); - if ($deduplicateStamp === null) { - return $stack->next()->handle($envelope, $stack); - } - - $receivedStamp = $envelope->last(ReceivedStamp::class); - if ($receivedStamp === null) { - $lock = $this->lockFactory->createLockFromKey( - $deduplicateStamp->getKey(), - $deduplicateStamp->getTtl(), - false, - ); - - if (!$lock->acquire()) { - return $envelope; - } - } elseif ($deduplicateStamp->onlyDeduplicateInQueue()) { - $this->lockFactory->createLockFromKey($deduplicateStamp->getKey())->release(); - } - - try { - $envelope = $stack->next()->handle($envelope, $stack); - } finally { - $receivedStamp = $envelope->last(ReceivedStamp::class); - if ($receivedStamp !== null && !$deduplicateStamp->onlyDeduplicateInQueue()) { - $this->lockFactory->createLockFromKey($deduplicateStamp->getKey())->release(); - } - } - - return $envelope; - } -} diff --git a/src/bundle/Resources/config/services/buses.yaml b/src/bundle/Resources/config/services/buses.yaml index 5cd61c3..f245984 100644 --- a/src/bundle/Resources/config/services/buses.yaml +++ b/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' diff --git a/src/bundle/Resources/config/services/serializer/normalizers.yaml b/src/bundle/Resources/config/services/serializer/normalizers.yaml index 4e4096b..faee30f 100644 --- a/src/bundle/Resources/config/services/serializer/normalizers.yaml +++ b/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 } diff --git a/src/bundle/Serializer/Normalizer/DeduplicateStampNormalizer.php b/src/bundle/Serializer/Normalizer/DeduplicateStampNormalizer.php deleted file mode 100644 index 4ff1e15..0000000 --- a/src/bundle/Serializer/Normalizer/DeduplicateStampNormalizer.php +++ /dev/null @@ -1,99 +0,0 @@ - true, - ]; - } - - /** - * @phpstan-param TData $data - */ - public function denormalize( - $data, - string $type, - ?string $format = null, - array $context = [] - ): DeduplicateStamp { - $stamp = (new ReflectionClass(DeduplicateStamp::class))->newInstanceWithoutConstructor(); - - $key = $this->denormalizer->denormalize($data['key'], Key::class, $format, $context); - - Closure::bind(function () use ($data, $key): void { - $this->key = $key; - $this->ttl = $data['ttl'] ?? 300.0; - $this->onlyDeduplicateInQueue = $data['only_deduplicate_in_queue'] ?? false; - }, $stamp, DeduplicateStamp::class)(); - - return $stamp; - } - - public function supportsDenormalization( - $data, - string $type, - ?string $format = null, - array $context = [] - ): bool { - return $type === DeduplicateStamp::class; - } - - /** - * @phpstan-return TData - */ - public function normalize( - mixed $object, - ?string $format = null, - array $context = [] - ): array { - assert($object instanceof DeduplicateStamp); - - return [ - 'key' => $this->normalizer->normalize($object->getKey(), $format, $context), - 'ttl' => $object->getTtl(), - 'only_deduplicate_in_queue' => $object->onlyDeduplicateInQueue(), - ]; - } - - public function supportsNormalization( - $data, - ?string $format = null, - array $context = [] - ): bool { - return $data instanceof DeduplicateStamp; - } -} diff --git a/src/bundle/Serializer/Normalizer/LockKeyNormalizer.php b/src/bundle/Serializer/Normalizer/LockKeyNormalizer.php deleted file mode 100644 index 7f2a161..0000000 --- a/src/bundle/Serializer/Normalizer/LockKeyNormalizer.php +++ /dev/null @@ -1,82 +0,0 @@ - true, - ]; - } - - /** - * @return array - */ - public function normalize( - $data, - ?string $format = null, - array $context = [] - ): array { - assert($data instanceof Key); - - return $data->__serialize(); - } - - public function supportsNormalization( - $data, - ?string $format = null, - array $context = [] - ): bool { - return $data instanceof Key; - } - - /** - * @throws \ReflectionException - */ - public function denormalize( - $data, - string $type, - ?string $format = null, - array $context = [] - ): Key { - $key = (new ReflectionClass(Key::class))->newInstanceWithoutConstructor(); - $setter = Closure::bind( - function (string $field) use ($data): void { - $this->$field = $data[$field]; - }, - $key, - Key::class, - ); - foreach (['resource', 'expiringTime', 'state'] as $serializedField) { - $setter($serializedField); - } - - return $key; - } - - public function supportsDenormalization( - $data, - string $type, - ?string $format = null, - array $context = [] - ): bool { - return $type === Key::class; - } -} diff --git a/src/contracts/Stamp/DeduplicateStamp.php b/src/contracts/Stamp/DeduplicateStamp.php deleted file mode 100644 index 273a059..0000000 --- a/src/contracts/Stamp/DeduplicateStamp.php +++ /dev/null @@ -1,72 +0,0 @@ -. - * - * Original code: https://github.com/symfony/symfony/blob/7.3/src/Symfony/Component/Messenger/Stamp/DeduplicateStamp.php - * - * @deprecated since Ibexa 5.0.10. Starting from Ibexa 6.0, the native {@see SymfonyDeduplicateStamp} - * will be used instead. Ibexa 5.0 is not prepared to handle the Symfony stamp yet, so keep using - * this class until you upgrade. - */ -final class DeduplicateStamp implements StampInterface -{ - private Key $key; - - private ?float $ttl; - - private bool $onlyDeduplicateInQueue; - - public function __construct( - string $key, - ?float $ttl = 300.0, - bool $onlyDeduplicateInQueue = false - ) { - trigger_deprecation( - 'ibexa/messenger', - '5.0.10', - 'The "%s" class is deprecated, starting from Ibexa 6.0 the native "%s" will be used instead.', - self::class, - SymfonyDeduplicateStamp::class, - ); - - if (!class_exists(Key::class)) { - throw new LogicException(sprintf( - 'You cannot use the "%s" as the Lock component is not installed. Try running "composer require symfony/lock".', - self::class, - )); - } - - $this->key = new Key($key); - $this->ttl = $ttl; - $this->onlyDeduplicateInQueue = $onlyDeduplicateInQueue; - } - - public function onlyDeduplicateInQueue(): bool - { - return $this->onlyDeduplicateInQueue; - } - - public function getKey(): Key - { - return $this->key; - } - - public function getTtl(): ?float - { - return $this->ttl; - } -} diff --git a/tests/bundle/DependencyInjection/IbexaMessengerExtensionTest.php b/tests/bundle/DependencyInjection/IbexaMessengerExtensionTest.php index e7c710d..02e15c9 100644 --- a/tests/bundle/DependencyInjection/IbexaMessengerExtensionTest.php +++ b/tests/bundle/DependencyInjection/IbexaMessengerExtensionTest.php @@ -9,7 +9,6 @@ namespace Ibexa\Tests\Bundle\Messenger\DependencyInjection; use Ibexa\Bundle\Messenger\DependencyInjection\IbexaMessengerExtension; -use Ibexa\Bundle\Messenger\Middleware\DeduplicateMiddleware; use Ibexa\Bundle\Messenger\Middleware\SudoMiddleware; use Ibexa\Bundle\Messenger\Middleware\UserPermissionMiddleware; use Matthias\SymfonyDependencyInjectionTest\PhpUnit\AbstractExtensionTestCase; @@ -37,7 +36,8 @@ public function testConfigurationDefaults(): void self::assertTrue($this->container->hasDefinition(SudoMiddleware::class)); self::assertTrue($this->container->hasDefinition(UserPermissionMiddleware::class)); - self::assertTrue($this->container->hasDefinition(DeduplicateMiddleware::class)); + self::assertTrue($this->container->hasDefinition('ibexa.messenger.deduplicate_middleware')); + self::assertTrue($this->container->hasDefinition('ibexa.messenger.release_deduplication_lock_on_failure_listener')); self::assertTrue($this->container->hasDefinition('ibexa.messenger.lock_factory')); self::assertTrue($this->container->hasDefinition('ibexa.messenger.lock_store.dbal')); } @@ -52,7 +52,8 @@ public function testConfigurationWithDisabledLocks(): void self::assertTrue($this->container->hasDefinition(SudoMiddleware::class)); self::assertTrue($this->container->hasDefinition(UserPermissionMiddleware::class)); - self::assertFalse($this->container->hasDefinition(DeduplicateMiddleware::class)); + self::assertFalse($this->container->hasDefinition('ibexa.messenger.deduplicate_middleware')); + self::assertFalse($this->container->hasDefinition('ibexa.messenger.release_deduplication_lock_on_failure_listener')); self::assertFalse($this->container->hasDefinition('ibexa.messenger.lock_factory')); self::assertFalse($this->container->hasDefinition('ibexa.messenger.lock_store.dbal')); } diff --git a/tests/bundle/Serializer/Normalizer/DeduplicateStampNormalizerTest.php b/tests/bundle/Serializer/Normalizer/DeduplicateStampNormalizerTest.php deleted file mode 100644 index 439ce0f..0000000 --- a/tests/bundle/Serializer/Normalizer/DeduplicateStampNormalizerTest.php +++ /dev/null @@ -1,119 +0,0 @@ -normalizer = new DeduplicateStampNormalizer(); - $this->subNormalizer = $this->createMock(NormalizerInterface::class); - $this->subDenormalizer = $this->createMock(DenormalizerInterface::class); - - $this->normalizer->setNormalizer($this->subNormalizer); - $this->normalizer->setDenormalizer($this->subDenormalizer); - } - - /** - * @param array{ - * key: \ArrayObject, - * ttl: float, - * only_deduplicate_in_queue: bool, - * } $expected - */ - #[DataProvider('provideForTest')] - public function testNormalize( - DeduplicateStamp $stamp, - array $expected - ): void { - $this->subDenormalizer->expects(self::never())->method(self::anything()); - - $this->subNormalizer->expects(self::once()) - ->method('normalize') - ->with(self::isInstanceOf(Key::class)) - ->willReturn($expected['key']); - - $normalized = $this->normalizer->normalize($stamp); - - self::assertSame($expected, $normalized); - } - - /** - * @param array{ - * key: \ArrayObject, - * ttl: float, - * only_deduplicate_in_queue: bool, - * } $data - */ - #[DataProvider('provideForTest')] - public function testDenormalize( - DeduplicateStamp $expectedStamp, - array $data - ): void { - $this->subNormalizer->expects(self::never())->method(self::anything()); - - $this->subDenormalizer - ->expects(self::once()) - ->method('denormalize') - ->with(self::identicalTo($data['key']), Key::class) - ->willReturn($expectedStamp->getKey()); - - $denormalized = $this->normalizer->denormalize($data, DeduplicateStamp::class); - - self::assertSame($expectedStamp->getKey(), $denormalized->getKey()); - self::assertSame($expectedStamp->getTtl(), $denormalized->getTtl()); - self::assertSame($expectedStamp->onlyDeduplicateInQueue(), $denormalized->onlyDeduplicateInQueue()); - } - - /** - * @return iterable, - * ttl: float, - * only_deduplicate_in_queue: bool, - * }, - * }> - */ - public static function provideForTest(): iterable - { - yield [ - new DeduplicateStamp('foo_key'), - [ - 'key' => new \ArrayObject(), - 'ttl' => 300.0, - 'only_deduplicate_in_queue' => false, - ], - ]; - - yield [ - new DeduplicateStamp('foo_key', 600.0, true), - [ - 'key' => new \ArrayObject(), - 'ttl' => 600.0, - 'only_deduplicate_in_queue' => true, - ], - ]; - } -} diff --git a/tests/bundle/Serializer/Normalizer/LockKeyNormalizerTest.php b/tests/bundle/Serializer/Normalizer/LockKeyNormalizerTest.php deleted file mode 100644 index c6e64e8..0000000 --- a/tests/bundle/Serializer/Normalizer/LockKeyNormalizerTest.php +++ /dev/null @@ -1,114 +0,0 @@ -normalizer = new LockKeyNormalizer(); - } - - /** - * @param array{ - * resource: string, - * expiringTime: float|null, - * state: array, - * } $expected - */ - #[DataProvider('provideForTest')] - public function testNormalize( - Key $data, - array $expected - ): void { - $normalized = $this->normalizer->normalize($data); - - self::assertSame($expected, $normalized); - } - - /** - * @param array{ - * resource: string, - * expiringTime: float|null, - * state: array, - * } $data - */ - #[DataProvider('provideForTest')] - public function testDenormalize( - Key $expectedKey, - array $data - ): void { - $denormalized = $this->normalizer->denormalize($data, Key::class); - - self::assertSame($data['resource'], (string)$denormalized); - - if ($data['expiringTime'] === null) { - self::assertNull($denormalized->getRemainingLifetime()); - } else { - // Some time will pass - self::assertNotNull($denormalized->getRemainingLifetime()); - self::assertLessThan($data['expiringTime'], $denormalized->getRemainingLifetime()); - } - self::assertFalse($denormalized->hasState('_non_existent_state_')); - - foreach ($data['state'] as $key => $value) { - self::assertTrue($denormalized->hasState($key)); - self::assertSame($value, $denormalized->getState($key)); - } - } - - /** - * @return iterable, - * }, - * }> - */ - public static function provideForTest(): iterable - { - $currentTime = time(); - $key = new Key('bar'); - \Closure::bind(function () use ($currentTime): void { - $this->expiringTime = $currentTime + 300; - }, $key, Key::class)(); - - yield [ - $key, - [ - 'resource' => 'bar', - 'expiringTime' => (float)($currentTime + 300), - 'state' => [], - ], - ]; - - $key = new Key('foo'); - $key->setState('foo', 'bar'); - - yield [ - $key, - [ - 'resource' => 'foo', - 'expiringTime' => null, - 'state' => [ - 'foo' => 'bar', - ], - ], - ]; - } -} diff --git a/tests/integration/MessageBusTest.php b/tests/integration/MessageBusTest.php index 99d0cc3..feefcb4 100644 --- a/tests/integration/MessageBusTest.php +++ b/tests/integration/MessageBusTest.php @@ -8,7 +8,6 @@ namespace Ibexa\Tests\Integration\Messenger; -use Ibexa\Contracts\Messenger\Stamp\DeduplicateStamp; use Ibexa\Contracts\Test\Core\IbexaKernelTestCase; use Ibexa\Tests\Integration\Messenger\Stubs\FooMessage; use Ibexa\Tests\Integration\Messenger\Stubs\FooMessageHandler; @@ -17,6 +16,7 @@ use Symfony\Component\Console\Output\NullOutput; use Symfony\Component\Messenger\Envelope; use Symfony\Component\Messenger\MessageBusInterface; +use Symfony\Component\Messenger\Stamp\DeduplicateStamp; use Symfony\Component\Messenger\Stamp\ReceivedStamp; use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;