[JOBS] AMQP redialer does not resume pipeline (sometimes?)

[algirdasci]

Hello,

I'm using RR (via Spiral framework) mainly for RabbitMQ message processing. Problem is, that network connection is not very stable, as RR is running in multiple (100+) servers across the globe. I've configured high timeout values, so RR would eventually reconnect to RabbitMQ server and continue to process messages without intervention. Currently each server consumes at least 4 different queues. The pipelines for those queues are created dynamically:

    public function declareQueue(string $queueName, array $options = []): void
    {
        $jobs = new Jobs(RPC::create(RPCConst::CONNECTION));
        if (!$this->doesQueueExist($jobs, $queueName)) {
            $jobs->create(self::getAMPQCreateInfo($queueName, $options));
        }
    }

    /**
     * @throws JobsException
     */
    private function doesQueueExist(Jobs $jobs, string $queueName): bool
    {
        foreach ($jobs->getIterator() as $queue) {
            if ($queue->getName() === $queueName) {
                return true;
            }
        }

        return false;
    }

    public static function getAMPQCreateInfo(string $queueName, array $options = []): AMQPCreateInfo
    {
        $nodeQueueName = sprintf('%d.%s', env('SERVER_ID'), $queueName);

        return new AMQPCreateInfo(
            name: $queueName, // @phpstan-ignore argument.type
            priority: $options['priority'] ?? CreateInfo::PRIORITY_DEFAULT_VALUE,
            prefetch: $options['prefetch'] ?? AMQPCreateInfo::PREFETCH_DEFAULT_VALUE,
            queue: $options['queue_name'] ?? $nodeQueueName,
            exchange: QueueConst::EXCHANGE_NAME,
            exchangeType: ExchangeType::Topics,
            routingKey: $options['queue_name'] ?? $nodeQueueName,
            durable: $options['durable'] ?? true,
            exchangeDurable: true,
            consumeAll: $options['consume_all'] ?? false,
            queueHeaders: array_merge($options['queue_headers'] ?? [], [
                'x-queue-type' => 'quorum'
            ]),
            redialTimeout: 2 * 24 * 3600
        );
    }

Everything is working ok, but there are some cases, when redialer restores connection to RabbitMQ and leaves the pipeline paused, which means that message processing stops and RR needs to be restarted or job queue resumed, to continue working. Also, I cannot reproduce it locally - when breaking connection to RabbitMQ, RR always resumes pipeline. What is strange that, when this case occurs, last message of log claims, that redialer exited. Is this expected behaviour? Is there something I can do, to automatically resume pipeline, once RR restores connection? Is there a difference if pipelines are declared dynamically (as currently) or statically, in .rr.yaml file?

{"level":"debug","ts":1727514286185636176,"logger":"amqp","msg":"delivery channel was closed, leaving the AMQP listener"}
{"level":"debug","ts":1727514286185643119,"logger":"amqp","msg":"number of listeners","listeners":0}
{"level":"error","ts":1727514286185707689,"logger":"amqp","msg":"consume channel close","error":"Exception (504) Reason: \"channel/connection is not open\""}
{"level":"error","ts":1727514286185767441,"logger":"amqp","msg":"amqp connection closed","error":"Exception (504) Reason: \"channel/connection is not open\""}
{"level":"error","ts":1727514286185779093,"logger":"amqp","msg":"pipeline connection was closed, redialing","error":"Exception (501) Reason: \"read tcp <redacted>:38128-><redacted>:5672: read: connection reset by peer\"","pipeline":"action","driver":"amqp","start":1727514286185774725}
{"level":"info","ts":1727514286228254674,"logger":"amqp","msg":"amqp dial was succeed. trying to redeclare queues and subscribers"}
{"level":"info","ts":1727514286280420208,"logger":"amqp","msg":"queues and subscribers was redeclared successfully"}
{"level":"info","ts":1727514286280465312,"logger":"amqp","msg":"connection was successfully restored","pipeline":"action","driver":"amqp","start":1727514286185774725,"elapsed":94}
{"level":"info","ts":1727514286280484528,"logger":"amqp","msg":"redialer restarted"}
{"level":"debug","ts":1727514286280503032,"logger":"amqp","msg":"exited from redialer"}

$ ./rr workers
Workers of [jobs]:
+---------+-----------+---------+---------+---------+--------------------+
|   PID   |  STATUS   |  EXECS  | MEMORY  |  CPU%   |      CREATED       |
+---------+-----------+---------+---------+---------+--------------------+
| 2719733 | ready     |      23 | 42 MB   |    0.00 | 8 hours ago        |
| 2719734 | ready     |      22 | 42 MB   |    0.00 | 8 hours ago        |
| 2719735 | ready     |      22 | 43 MB   |    0.00 | 8 hours ago        |
| 2719736 | ready     |      23 | 44 MB   |    0.00 | 8 hours ago        |
| 2719737 | ready     |      23 | 40 MB   |    0.00 | 8 hours ago        |
| 2719738 | ready     |      23 | 42 MB   |    0.00 | 8 hours ago        |
| 2719739 | ready     |      23 | 42 MB   |    0.00 | 8 hours ago        |
| 2719740 | ready     |      23 | 43 MB   |    0.00 | 8 hours ago        |
+---------+-----------+---------+---------+---------+--------------------+
Jobs of [jobs]:
+----------------+----------------+--------+---------------------+--------+---------+----------+
|     STATUS     |    PIPELINE    | DRIVER |        QUEUE        | ACTIVE | DELAYED | RESERVED |
+----------------+----------------+--------+---------------------+--------+---------+----------+
| PAUSED/STOPPED | action         | amqp   | 1306.action         | 0      | 0       | 0        |
| PAUSED/STOPPED | callback       | amqp   | callback            | 0      | 0       | 0        |
| READY          | flow           | amqp   | 1306.flow           | 0      | 0       | 0        |
+----------------+----------------+--------+---------------------+--------+---------+----------+

.rr.yaml:

version: '3'
rpc:
  listen: 'tcp://127.0.0.1:6001'
jobs:
  timeout: 172800
  pool:
    num_workers: 1
  consume:
    - flow
    - action
server:
  command: 'php app.php'
  relay: pipes
  env:
    - XDEBUG_SESSION: '1'
metrics:
  address: '127.0.0.1:2112'
amqp:
  addr: 'amqp://<redacted>:<redacted>@<redacted>:5672/'
endure:
  log_level: error
  watchdog_sec: 60

[rustatian]

Hey @algirdasci 👋
Interesting case... Could you please post here a RR version as well?

[algirdasci]

Sure! I'm using latest RR build:

$ ./rr --version
rr version 2024.2.1 (build time: 2024-09-12T16:25:04+0000, go1.23.1), OS: linux, arch: amd64

[rustatian]

I'll double-check that, need to write an additional test for that case.

[rustatian]

I did not forget about this case @algirdasci, just need some more time to find the problem (friendly note 😃 )

[algirdasci]

all good, no rush :) thanks for your work!

[rustatian]

I think I found the reason. There is might be a case, when the channel for messages delivery could be closed by the RabbitMQ server, without connection interruption. In that case, RR will close the delivery channel but does not create a new one, since the trigger for redial is broken connection.
I wrote a new Redialer which can be applied to channels + connections and recreate a delivery channel in that case. Expected release: 2024.3.1. Could you please create a bug ticket for that?

[algirdasci]

Great find, @rustatian! I've created bug ticket for it.