Add pause/resume logic in and host health awareness in consumers#85
Merged
Conversation
6fbc28c to
53d6228
Compare
flamble
approved these changes
Jul 10, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem statement
Consumers in rmqcpp keep consuming from a queue regardless of the health of the host they're running on. In environments where a host can become degraded (e.g. losing connectivity to a downstream dependency while still
connected to RabbitMQ), it's desirable for a consumer to stop taking messages off the broker until the host recovers, so that in-flight messages aren't processed on an unhealthy host and can instead be delivered to a healthy consumer.
There was previously no mechanism to pause/resume a consumer's delivery flow based on host health.
Proposed changes
This PR adds opt-in host-health awareness to consumers so they can stop pulling messages while their host is unhealthy and resume once it recovers. So messages aren't processed on a degraded host. It's driven by a periodic health check and enabled per-consumer through a new host-health configuration; when that config isn't set, behavior is completely unchanged. The change also adds metrics for health checks and consumer latency to make the behavior observable.
CI / build fixes (included in this branch)
longer usable).
connectionNamefield and structured log format for sessions amqpprox#120) changed it from<id>: vhost=<vhost>toid=<id> vhost=<vhost>…; the busted Docker cache surfaced this. Updated the regex to accept both formats, and extended the post-force-close assertion timeout to match test_connect_disconnect.Testing
Notes for reviewers