feature symfony#53508 [Console][Messenger] Asynchronously notify transports which messages are still being processed (HypeMC)

This PR was merged into the 7.2 branch.

Discussion
----------

[Console][Messenger] Asynchronously notify transports which messages are still being processed

| Q             | A
| ------------- | ---
| Branch?       | 7.2
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| Issues        | -
| License       | MIT

Certain transports, such as Beanstalkd, have a timeout that determines how long a certain message is considered as "in progress". If the message is not acknowledged or rejected within that time period, it is returned to the "ready" queue.

This could cause potential issues when it takes a handler longer to finish processing the message. In those instances, the message being processed could be returned to the ready queue and taken by another instance of the receiver. This would result in the message being processed multiple times, or even worse, in an infinite loop.

Usually, transports that have a timeout also have the option to be notified that the message is still being processed, so the timeout can be postponed.

The idea of this PR is to utilize the `SignalableCommandInterface` and `pcntl_alarm()` to notify transports which messages are still being processed. Since the `SignalRegistry` has [asynchronous signals enabled](https://github.com/symfony/symfony/blob/34915f6e16f04537eb18d9d2c303ec375e63cc4b/src/Symfony/Component/Console/SignalRegistry/SignalRegistry.php#L21) by default, the whole process would happen asynchronously. I've named this feature "keepalive" for lack of a better name.

Currently, I've added this option only to the Beanstalkd transport since that's the one I'm familiar with and use, but as far as I was able to gather, at least one other transport supports this feature. Amazon SQS has a visibility timeout which can be [increased](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html#changing-message-visibility-timeout). This visibility timeout seems to be the same thing as Beanstalkd's TTR. (Disclaimer: I've never used Amazon SQS, so if I got something wrong, please let me know.)

I've split the PR into two commits so it would (hopefully) be easier to review:

1. I've extracted the first commit into a separate PR: symfony#53533
2. The second commit adds the keepalive feature to the Messenger component.

Commits
-------

57b556a [Messenger] Notify transports which messages are still being processed, using `pcntl_alarm()`

Author: chalasr

src/Symfony/Component/Messenger/Bridge/Beanstalkd/CHANGELOG.md

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+7.2
+---
+
+ * Implement the `KeepaliveReceiverInterface` to enable asynchronously notifying Beanstalkd that the job is still being processed, in order to avoid timeouts
+
 5.2.0
 -----
 

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdReceiverTest.php

@@ -16,6 +16,7 @@
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdReceivedStamp;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdReceiver;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
+use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\Serializer;
@@ -78,6 +79,17 @@ public function testItRejectTheMessageIfThereIsAMessageDecodingFailedException()
         $receiver->get();
     }
 
+    public function testKeepalive()
+    {
+        $serializer = $this->createSerializer();
+
+        $connection = $this->createMock(Connection::class);
+        $connection->expects($this->once())->method('keepalive')->with(1);
+
+        $receiver = new BeanstalkdReceiver($connection, $serializer);
+        $receiver->keepalive(new Envelope(new DummyMessage('foo'), [new BeanstalkdReceivedStamp(1, 'bar')]));
+    }
+
     private function createBeanstalkdEnvelope(): array
     {
         return [

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/BeanstalkdTransportTest.php

@@ -13,6 +13,7 @@
 
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Fixtures\DummyMessage;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdReceivedStamp;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdTransport;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
 use Symfony\Component\Messenger\Envelope;
@@ -50,6 +51,18 @@ public function testReceivesMessages()
         $this->assertSame($decodedMessage, $envelopes[0]->getMessage());
     }
 
+    public function testKeepalive()
+    {
+        $transport = $this->getTransport(
+            null,
+            $connection = $this->createMock(Connection::class),
+        );
+
+        $connection->expects($this->once())->method('keepalive')->with(1);
+
+        $transport->keepalive(new Envelope(new DummyMessage('foo'), [new BeanstalkdReceivedStamp(1, 'bar')]));
+    }
+
     private function getTransport(?SerializerInterface $serializer = null, ?Connection $connection = null): BeanstalkdTransport
     {
         $serializer ??= $this->createMock(SerializerInterface::class);

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Tests/Transport/ConnectionTest.php

@@ -330,4 +330,37 @@ public function testSendWhenABeanstalkdExceptionOccurs()
 
         $connection->send($body, $headers, $delay);
     }
+
+    public function testKeepalive()
+    {
+        $id = 123456;
+
+        $tube = 'baz';
+
+        $client = $this->createMock(PheanstalkInterface::class);
+        $client->expects($this->once())->method('useTube')->with($tube)->willReturn($client);
+        $client->expects($this->once())->method('touch')->with($this->callback(fn (JobId $jobId): bool => $jobId->getId() === $id));
+
+        $connection = new Connection(['tube_name' => $tube], $client);
+
+        $connection->keepalive((string) $id);
+    }
+
+    public function testKeepaliveWhenABeanstalkdExceptionOccurs()
+    {
+        $id = 123456;
+
+        $tube = 'baz123';
+
+        $exception = new ServerException('baz error');
+
+        $client = $this->createMock(PheanstalkInterface::class);
+        $client->expects($this->once())->method('useTube')->with($tube)->willReturn($client);
+        $client->expects($this->once())->method('touch')->with($this->callback(fn (JobId $jobId): bool => $jobId->getId() === $id))->willThrowException($exception);
+
+        $connection = new Connection(['tube_name' => $tube], $client);
+
+        $this->expectExceptionObject(new TransportException($exception->getMessage(), 0, $exception));
+        $connection->keepalive((string) $id);
+    }
 }

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/BeanstalkdReceiver.php

@@ -14,15 +14,15 @@
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\LogicException;
 use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
+use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
-use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 
 /**
  * @author Antonio Pauletich <[email protected]>
  */
-class BeanstalkdReceiver implements ReceiverInterface, MessageCountAwareInterface
+class BeanstalkdReceiver implements KeepaliveReceiverInterface, MessageCountAwareInterface
 {
     private SerializerInterface $serializer;
 
@@ -65,6 +65,11 @@ public function reject(Envelope $envelope): void
         $this->connection->reject($this->findBeanstalkdReceivedStamp($envelope)->getId());
     }
 
+    public function keepalive(Envelope $envelope): void
+    {
+        $this->connection->keepalive($this->findBeanstalkdReceivedStamp($envelope)->getId());
+    }
+
     public function getMessageCount(): int
     {
         return $this->connection->getMessageCount();

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/BeanstalkdTransport.php

@@ -12,6 +12,7 @@
 namespace Symfony\Component\Messenger\Bridge\Beanstalkd\Transport;
 
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -20,7 +21,7 @@
 /**
  * @author Antonio Pauletich <[email protected]>
  */
-class BeanstalkdTransport implements TransportInterface, MessageCountAwareInterface
+class BeanstalkdTransport implements TransportInterface, KeepaliveReceiverInterface, MessageCountAwareInterface
 {
     private SerializerInterface $serializer;
     private BeanstalkdReceiver $receiver;
@@ -48,6 +49,11 @@ public function reject(Envelope $envelope): void
         $this->getReceiver()->reject($envelope);
     }
 
+    public function keepalive(Envelope $envelope): void
+    {
+        $this->getReceiver()->keepalive($envelope);
+    }
+
     public function getMessageCount(): int
     {
         return $this->getReceiver()->getMessageCount();

src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/Connection.php

@@ -180,6 +180,15 @@ public function reject(string $id): void
         }
     }
 
+    public function keepalive(string $id): void
+    {
+        try {
+            $this->client->useTube($this->tube)->touch(new JobId((int) $id));
+        } catch (Exception $exception) {
+            throw new TransportException($exception->getMessage(), 0, $exception);
+        }
+    }
+
     public function getMessageCount(): int
     {
         try {

src/Symfony/Component/Messenger/Bridge/Beanstalkd/composer.json

@@ -14,7 +14,7 @@
     "require": {
         "php": ">=8.2",
         "pda/pheanstalk": "^4.0",
-        "symfony/messenger": "^6.4|^7.0"
+        "symfony/messenger": "^7.2"
     },
     "require-dev": {
         "symfony/property-access": "^6.4|^7.0",

src/Symfony/Component/Messenger/CHANGELOG.md

@@ -10,6 +10,7 @@ CHANGELOG
  * Add `--format` option to the `messenger:stats` command
  * Add `getRetryDelay()` method to `RecoverableExceptionInterface`
  * Add `skip` option to `messenger:failed:retry` command when run interactively to skip message and requeue it
+ * Add the ability to asynchronously notify transports about which messages are still being processed by the worker, using `pcntl_alarm()`
 
 7.1
 ---

src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php

@@ -43,6 +43,8 @@
 #[AsCommand(name: 'messenger:consume', description: 'Consume messages')]
 class ConsumeMessagesCommand extends Command implements SignalableCommandInterface
 {
+    private const DEFAULT_KEEPALIVE_INTERVAL = 5;
+
     private ?Worker $worker = null;
 
     public function __construct(
@@ -75,6 +77,7 @@ protected function configure(): void
                 new InputOption('queues', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Limit receivers to only consume from the specified queues'),
                 new InputOption('no-reset', null, InputOption::VALUE_NONE, 'Do not reset container services after each message'),
                 new InputOption('all', null, InputOption::VALUE_NONE, 'Consume messages from all receivers'),
+                new InputOption('keepalive', null, InputOption::VALUE_OPTIONAL, 'Whether to use the transport\'s keepalive mechanism if implemented', self::DEFAULT_KEEPALIVE_INTERVAL),
             ])
             ->setHelp(<<<'EOF'
 The <info>%command.name%</info> command consumes messages and dispatches them to the message bus.
@@ -124,6 +127,13 @@ protected function configure(): void
         ;
     }
 
+    protected function initialize(InputInterface $input, OutputInterface $output): void
+    {
+        if ($input->hasParameterOption('--keepalive')) {
+            $this->getApplication()->setAlarmInterval((int) ($input->getOption('keepalive') ?? self::DEFAULT_KEEPALIVE_INTERVAL));
+        }
+    }
+
     protected function interact(InputInterface $input, OutputInterface $output): void
     {
         $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
@@ -264,7 +274,7 @@ public function complete(CompletionInput $input, CompletionSuggestions $suggesti
 
     public function getSubscribedSignals(): array
     {
-        return $this->signals ?? (\extension_loaded('pcntl') ? [\SIGTERM, \SIGINT, \SIGQUIT] : []);
+        return $this->signals ?? (\extension_loaded('pcntl') ? [\SIGTERM, \SIGINT, \SIGQUIT, \SIGALRM] : []);
     }
 
     public function handleSignal(int $signal, int|false $previousExitCode = 0): int|false
@@ -273,6 +283,14 @@ public function handleSignal(int $signal, int|false $previousExitCode = 0): int|
             return false;
         }
 
+        if (\SIGALRM === $signal) {
+            $this->logger?->info('Sending keepalive request.', ['transport_names' => $this->worker->getMetadata()->getTransportNames()]);
+
+            $this->worker->keepalive();
+
+            return false;
+        }
+
         $this->logger?->info('Received signal {signal}.', ['signal' => $signal, 'transport_names' => $this->worker->getMetadata()->getTransportNames()]);
 
         $this->worker->stop();