How to handle error messages without ACK nor infinite retry

[yannicktian]

Hi everyone,

This is my current rabbtimq configuration

gamma_ex
      |
      `--> lambda_q <--------------------- lambda_ex <-------------------------+
              |                                                                |
        OK <--o--> KO --> lambda_on_error_ex --> lambda_on_error_q --> [TTL] --+

old_sound_rabbit_mq:
    producers:
        # only here for dead-letter exchange pattern, do not use directly
        lambda:
            connection: default
            exchange_options: {name: lambda_ex, type: direct}

    consumers:
        lambda:
            connection: default
            exchange_options: {name: gamma_ex, type: fanout } # or direct
            queue_options:
                name: lambda_q
                arguments:
                    'x-dead-letter-exchange': [S, lambda_on_error_ex]
                    'x-dead-letter-routing-key': [S, '']
            callback: app.consumer.lambda_consumer

        # only here for dead-letter exchange pattern, do not use directly
        lambda_on_error:
            connection: default
            exchange_options: {name: lambda_on_error_ex, type: direct}
            queue_options:
                name: lambda_on_error_q
                arguments:
                    'x-dead-letter-exchange': [S, lambda_ex]
                    'x-message-ttl': ['I', 10000]
            callback: app.consumer.lambda_consumer  # wont be used but the key must be present :/

    bindings:
        - {exchange: lambda_ex, destination: lambda_q}

In my app.consumer.lambda_consumer, I do a request to a third party. If the third party server returns an error (500) I ConsumerInterface::MSG_REJECT the message and the dead-letter system retries and publishes it via the lambda_ex exchange after the TTL.
When the number of retry max is reached, say 10 for example, I ConsumerInterface::MSG_ACK the message and it is lost forever.

Instead of losing the message, I want to save it "somewhere" and be able to retry when the third party server comes back on. Is there a simple way to do it with RabbitMq ?

I was thinking of

• publishing the message to new queue retry_over_limit_q.
• the consumer associated to this queue is set to republish the message to the initial exchange
• start the consumer when the third party returns 200 again
• stop the consumer when the queue is empty

But I can't find a suitable way to start the consumer asynchronously within Symfony app...
Maybe using http://jmsyst.com/bundles/JMSJobQueueBundle ? or https://symfony.com/doc/current/components/process.html#running-processes-asynchronously ?

Thank you for your help,

Yannick