fix(mqs): redial dropped RabbitMQ connections instead of wedging until restart

[LendritIbrahimi]

This fixes the issue where a dropped RabbitMQ connection permanently wedged the queue: the connection was dialed once via sync.Once, so after a broker restart or network blip every subsequent publish failed against the closed connection until the whole pod was restarted. (explained in this issue #998)

The implementation replaces the sync.Once with a mutex-guarded ensureConnected that lazily redials whenever the current connection is nil or closed, reopening the topic on the new connection (the stale topic is shut down in the background). Publish additionally retries once: if a publish fails and the connection turns out to be lost, it redials and re-attempts before surfacing the error.

If the redial itself fails, the original publish error is returned rather than the dial error, so callers see the real failure. Subscribe goes through the same path, so consumers created after a drop also get a fresh connection.

[alexluong]

Overall LGTM and thanks for raising the issue!

Got a few comments here, would appreciate it you can address or respond here

1: I think there's some formatter issue, can you make sure to run gofmt on the codebase?
2: The redial mechanism is fair but I wonder if there's anything we should consider in case of an outage since every Publish will attempt a redial which may or may not be the desired behavior. I'm okay moving forward with this behavior as is, just want to flag this in case you want to rethink this mechanism.

Overall, great job and appreciate your contribution!

[LendritIbrahimi]

2: The redial mechanism is fair but I wonder if there's anything we should consider in case of an outage since every Publish will attempt a redial which may or may not be the desired behavior. I'm okay moving forward with this behavior as is, just want to flag this in case you want to rethink this mechanism.

I agree, I looked more into this. Assuming RabbitMQ is fully shut down (not a network blip but down for a while), consecutive publish attempts redial in serial order and each one adds delay for the next. So basically callers would not fail fast but get stuck waiting to get a response from the API until it's their turn to redial and fail naturally.

To fix this I've added a short cooldown which makes it so after a failed dial, callers within the next 5 seconds get the last dial error back immediately instead of redialing. Once the cooldown expires, the next publish probes again. Callers fail fast and can retry instead of piling up:

• t=0: publish A dials, hangs up to 30s. Publishes B, C, D arrive during that window and wait on the lock.
• t=30: dial fails. A gets the error, cooldown starts, and B, C, D drain instantly with the cached error.
• t=30 to 35: everyone gets the cached error in microseconds.
• t=35: next publish becomes the new probe, hangs up to 30s, and the cycle repeats.

The one trade-off is that recovery can lag by up to 5 seconds after the broker comes back, but that shouldn't be a big deal.

I'll make a commit with this approach and if you don't want to move in this direction then please let me know :)

[alexluong]

Thanks for the prompt response. This approach works for me!