Heartbeats are only used for blocking calls (amqp.consume). Messenger uses basic.get which is non-blocking. So I guess heartsbeats config wouldn't do anything.

https://github.com/pdezwart/php-amqp

heartbeats are limited to blocking calls only, so if there are no any operations on a connection or no active consumer set, connection may be closed by the broker as dead.

See also #32055

As I read, there is no quick "fix" at the moment. How can I auto-ack a message once it's received or ack it manually before executing my python script? This way, if my script throws an exception then I can republish the message for retry. Do you have a code example/configuration?

Thanks

How can I auto-ack a message once it's received or ack it manually before executing my python script

You could try listening on the WorkerMessageReceivedEvent event and then calling $transport->ack($envelope) (the envelope is included in the event).

What's the longer-term fix for this? I'm not an AMQP expert - just someone who accidentally got involved in the Messenger component :). Is the problem basically that we open a connection to AMQP, keep it open while the handler is processing... and eventually that connection fails? If I'm reading the error correctly, it's coming from:

If we're not using amqp.consume (blocking), shouldn't our non-blocking solution not have this problem? I mean, shouldn't we be able to disconnect (e.g. after handling each message) to avoid keeping a long connection?

Valid questions, someone with more experience should help us to answer and clear this out. @Tobion, any thoughts?

I mean, shouldn't we be able to disconnect (e.g. after handling each message) to avoid keeping a long connection?

The delivery tag of an envelope is unique per channel. So I don't think you can disconnect after receiving a message, reconnent and then acknowledge because the channel would be different.

From our experience the easiest solution is to acknowledge messages first when processing would take too long. This can be done with a listener as Ryan explained.
The alternative would be to fix amqp-ext or trying to ping the connection manually from time to time. But there are alot of different approaches and alof of different potential causes of failures so none is really failure-proof. Some links:

• [amqp][lib] Improve heartbeat handling. Introduce heartbeat on tick. Fixes "Invalid frame type 65" and "Broken pipe or closed connection" php-enqueue/enqueue-dev#658
• Heartbeat with long running tasks php-amqp/php-amqp#296
• Support for handling blocked connections. Possible? php-amqp/php-amqp#197

So if pre-ack’ing is the solution, we can’t change that to be the default behavior - it’s not what you usually want (because, for example, if your handler caused php to run out of memory, then you would have ack’ed a message that in fact never processed).

This leads me to a proposed solution: introduce a stamp that you can add to a message to “opt in” to acknowledging messages immediately.

Thoughts?

Yes I agree that we need a way to enable "auto-ack". What we do so far is having a separate queue that we consume with AMQP_AUTOACK.
Using a Stamp for this would also work and be more flexible as you can decide per message if it needs the auto-ack mode and don't need to create a separate transport/queue for this. The only disadvantage is that we cannot use the built-in auto-ack of AMQP but need to emulate it. That because the built-ina auto-ack needs to be enabled before your receive the message, so you can't enable it based on the stamp. But emulating this is fine and easy and transport independent so works without changes in in doctrine transport etc.

Maybe a generic AcknowledgeModeStamp with modes similar to https://docs.spring.io/spring-amqp/api/org/springframework/amqp/core/AcknowledgeMode.html

Hey, thanks for your report!
There has not been a lot of activity here for a while. Is this bug still relevant? Have you managed to find a workaround?

Friendly ping? Should this still be open? I will close if I don't hear anything.

Hey,

I didn't hear anything so I'm going to close it. Feel free to comment if this is still relevant, I can always reopen!

This is still happening for me. any real solution found so far?

I am facing this issue also. It is sometimes occurring but not every time when I am trying to send another message to another queue inside Handler. Any ideas how to fix it?

Same problem here

Any solution, anyone? The problem still persists.

I am also experiencing this issue. Is there a permanent solution yet?

I have learned so far that, in my situation it wasn't about the duration about the processing of the message, decreasing the prefetch number, in my case --limit 1, on the consumer seemed to have solved the issue.

If you're also processing messages coming in not from Symfony, skipping NonSendableStampInterface stamps from serialization also improves the stability.