[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

[algirdasci]

Thanks for the advice, I will test that scenario.

[algirdasci]

When I set default timeout values, the pipeline does not get recreated. RR just says pipeline will be recreated but nothing happens. How to reproduce:

1. Start RoadRunner
2. Stop or pause RabbitMQ container
3. Wait until timeout is reached
4. Start or resume RabbitMQ container

2024-10-10T14:03:14+0000        DEBUG   rpc             plugin was started      {"address": "tcp://127.0.0.1:6001", "list of the plugins with RPC methods:": ["lock", "kv", "metrics", "resetter", "informer", "app", "jobs"]}
2024-10-10T14:03:14+0000        DEBUG   amqp            pipeline was started    {"driver": "amqp", "pipeline": "rpc", "start": "2024-10-10T14:03:14+0000", "elapsed": 4}
2024-10-10T14:03:15+0000        DEBUG   server          worker is allocated     {"pid": 84205, "max_execs": 0, "internal_event_name": "EventWorkerConstruct"}
2024-10-10T14:03:44+0000        DEBUG   amqp            delivery channel was closed, leaving the AMQP listener
2024-10-10T14:03:44+0000        DEBUG   amqp            exited from redialer
2024-10-10T14:03:44+0000        DEBUG   amqp            number of listeners     {"listeners": 0}
2024-10-10T14:03:44+0000        ERROR   amqp            amqp connection closed  {"error": "Exception (504) Reason: \"channel/connection is not open\""}
2024-10-10T14:03:44+0000        ERROR   amqp            pipeline connection was closed, redialing       {"error": "Exception (501) Reason: \"read tcp 172.23.0.2:57384->192.168.65.254:5672: i/o timeout\"", "pipeline": "rpc", "driver": "amqp", "start": "2024-10-10T14:03:44+0000"}
2024-10-10T14:04:45+0000        ERROR   amqp            backoff operation failed, pipeline will be recreated    {"error": "amqp_driver_redial: Exception (501) Reason: \"read tcp 172.23.0.2:44474->192.168.65.254:5672: i/o timeout\""}

Meanwhile from another terminal:

root@7756dcc75962:/app# ./rr workers
Workers of [jobs]:
+---------+-----------+---------+---------+---------+--------------------+
|   PID   |  STATUS   |  EXECS  | MEMORY  |  CPU%   |      CREATED       |
+---------+-----------+---------+---------+---------+--------------------+
|   84205 | ready     |       0 | 44 MB   |    0.05 | 14 minutes ago     |
+---------+-----------+---------+---------+---------+--------------------+

root@7756dcc75962:/app# ./rr jobs --resume rpc
jobs_plugin_rmq_queue_declare: Exception (504) Reason: "channel/connection is not open"

For testing I'm using following .rr.yaml. Keep in mind, that rpc pipeline is declared dynamically:

version: '3'
rpc:
  listen: 'tcp://127.0.0.1:6001'
kv:
  local:
    driver: memory
    config:
      interval: 60
jobs:
  pool:
    num_workers: 1
  consume:
    - rpc
server:
  command: 'php app.php'
  relay: pipes
  env:
    - XDEBUG_SESSION: '1'
metrics:
  address: '127.0.0.1:2112'
amqp:
  addr: 'amqp://guest:[email protected]:5672/'
endure:
  log_level: error
  watchdog_sec: 60

[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!