bug #58763 [Messenger][RateLimiter] fix additional message handled wh…

…en using a rate limiter (Jean-Beru)

This PR was merged into the 6.4 branch.

Discussion
----------

[Messenger][RateLimiter] fix additional message handled when using a rate limiter

| Q             | A
| ------------- | ---
| Branch?       | 6.4
| Bug fix?      | yes
| New feature?  | no
| Deprecations? | no
| Issues        | Fix #57230
| License       | MIT

Fix additional message handled by Messenger when using a rate limiter. A token was reserved but not consumed. See #57230

With the following configuration:
```yaml
framework:
  rate_limiter:
    test:
      policy: 'fixed_window'
      limit: 1
      interval: '10 seconds'

  messenger:
    transports:
      test:
        dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
        rate_limiter: 'test'
    routing:
      'App\Messenger\DoSomething': 'test'
```

Log generated by the MessageHandler:
```bash
$ bin/console messenger:consume test

 [OK] Consuming messages from transport "test".

 // The worker will automatically exit once it has received a stop signal via the messenger:stop-workers command.

 // Quit the worker with CONTROL-C.

 // Re-run the command with a -vv option to see logs about consumed messages.

09:13:48 WARNING   [app] Message handled
09:13:58 WARNING   [app] Message handled
09:13:58 WARNING   [app] Message handled # Duplicated
09:14:08 WARNING   [app] Message handled
09:14:08 WARNING   [app] Message handled # Duplicated
09:14:18 WARNING   [app] Message handled
09:14:18 WARNING   [app] Message handled # Duplicated
```

After fix:
```bash
bin/console messenger:consume test

 [OK] Consuming messages from transport "test".

 // The worker will automatically exit once it has received a stop signal via the messenger:stop-workers command.

 // Quit the worker with CONTROL-C.

 // Re-run the command with a -vv option to see logs about consumed messages.

09:18:54 WARNING   [app] Message handled
09:19:04 WARNING   [app] Message handled
09:19:14 WARNING   [app] Message handled
09:19:24 WARNING   [app] Message handled
```

Commits
-------

ec1b999b812 [Messenger][RateLimiter] fix additional message handled when using a rate limiter

Tests/WorkerTest.php

@@ -14,6 +14,7 @@
 use PHPUnit\Framework\TestCase;
 use Psr\EventDispatcher\EventDispatcherInterface;
 use Psr\Log\LoggerInterface;
+use Symfony\Bridge\PhpUnit\ClockMock;
 use Symfony\Component\Clock\MockClock;
 use Symfony\Component\EventDispatcher\EventDispatcher;
 use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
@@ -47,8 +48,8 @@
 use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
 use Symfony\Component\Messenger\Worker;
 use Symfony\Component\RateLimiter\RateLimiterFactory;
+use Symfony\Component\RateLimiter\Reservation;
 use Symfony\Component\RateLimiter\Storage\InMemoryStorage;
-use Symfony\Contracts\Service\ResetInterface;
 
 /**
  * @group time-sensitive
@@ -73,7 +74,7 @@ public function testWorkerDispatchTheReceivedMessage()
                 return $envelopes[] = $envelope;
             });
 
-        $dispatcher = new class() implements EventDispatcherInterface {
+        $dispatcher = new class implements EventDispatcherInterface {
             private StopWorkerOnMessageLimitListener $listener;
 
             public function __construct()
@@ -403,7 +404,7 @@ public function testWorkerLimitQueuesUnsupported()
 
         $worker = new Worker(['transport1' => $receiver1, 'transport2' => $receiver2], $bus, clock: new MockClock());
         $this->expectException(RuntimeException::class);
-        $this->expectExceptionMessage(sprintf('Receiver for "transport2" does not implement "%s".', QueueReceiverInterface::class));
+        $this->expectExceptionMessage(\sprintf('Receiver for "transport2" does not implement "%s".', QueueReceiverInterface::class));
         $worker->run(['queues' => ['foo']]);
     }
 
@@ -418,7 +419,7 @@ public function testWorkerMessageReceivedEventMutability()
         $eventDispatcher = new EventDispatcher();
         $eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1));
 
-        $stamp = new class() implements StampInterface {
+        $stamp = new class implements StampInterface {
         };
         $listener = function (WorkerMessageReceivedEvent $event) use ($stamp) {
             $event->addStamps($stamp);
@@ -438,21 +439,21 @@ public function testWorkerRateLimitMessages()
         $envelope = [
             new Envelope(new DummyMessage('message1')),
             new Envelope(new DummyMessage('message2')),
+            new Envelope(new DummyMessage('message3')),
+            new Envelope(new DummyMessage('message4')),
         ];
         $receiver = new DummyReceiver([$envelope]);
 
         $bus = $this->createMock(MessageBusInterface::class);
         $bus->method('dispatch')->willReturnArgument(0);
 
         $eventDispatcher = new EventDispatcher();
-        $eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(2));
+        $eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(4));
 
         $rateLimitCount = 0;
-        $listener = function (WorkerRateLimitedEvent $event) use (&$rateLimitCount) {
+        $eventDispatcher->addListener(WorkerRateLimitedEvent::class, static function () use (&$rateLimitCount) {
             ++$rateLimitCount;
-            $event->getLimiter()->reset(); // Reset limiter to continue test
-        };
-        $eventDispatcher->addListener(WorkerRateLimitedEvent::class, $listener);
+        });
 
         $rateLimitFactory = new RateLimiterFactory([
             'id' => 'bus',
@@ -461,11 +462,14 @@ public function testWorkerRateLimitMessages()
             'interval' => '1 minute',
         ], new InMemoryStorage());
 
+        ClockMock::register(Reservation::class);
+        ClockMock::register(InMemoryStorage::class);
+
         $worker = new Worker(['bus' => $receiver], $bus, $eventDispatcher, null, ['bus' => $rateLimitFactory], new MockClock());
         $worker->run();
 
-        $this->assertCount(2, $receiver->getAcknowledgedEnvelopes());
-        $this->assertEquals(1, $rateLimitCount);
+        $this->assertSame(4, $receiver->getAcknowledgeCount());
+        $this->assertSame(3, $rateLimitCount);
     }
 
     public function testWorkerShouldLogOnStop()

Worker.php

@@ -86,7 +86,7 @@ public function run(array $options = []): void
             // if queue names are specified, all receivers must implement the QueueReceiverInterface
             foreach ($this->receivers as $transportName => $receiver) {
                 if (!$receiver instanceof QueueReceiverInterface) {
-                    throw new RuntimeException(sprintf('Receiver for "%s" does not implement "%s".', $transportName, QueueReceiverInterface::class));
+                    throw new RuntimeException(\sprintf('Receiver for "%s" does not implement "%s".', $transportName, QueueReceiverInterface::class));
                 }
             }
         }
@@ -242,6 +242,7 @@ private function rateLimit(string $transportName): void
 
         $this->eventDispatcher?->dispatch(new WorkerRateLimitedEvent($rateLimiter, $transportName));
         $rateLimiter->reserve()->wait();
+        $rateLimiter->consume();
     }
 
     private function flush(bool $force): bool