Skip to content

Treat consumer pause and resume as the desired state #102

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
willhoy merged 1 commit into from
Oct 29, 2026
Merged

Treat consumer pause and resume as the desired state #102

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 71 additions & 0 deletions docs/consumer-pause-resume.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
# Consumer Pause and Resume

Consumers created with `consumeOnlyFromHealthyHost` are paused and resumed
by the `HostHealthMonitor`. On every poll that returns a health result
(not a retry) it tells each health-aware consumer to pause (host unhealthy)
or resume (host healthy).

Pausing sends `basic.cancel` and resuming sends `basic.consume` with the
same consumer tag. The consumer object, and its message callback, are kept
while paused.

## Pause and resume set the desired state

`pause()` and `resume()` tell the channel what state the consumer should be
in. The channel makes it so at the next opportunity: straight away if it
can, otherwise once a reply in flight arrives or the channel reopens. The
latest request wins.

A request is only refused if it repeats the state the consumer is already
in, or if it can't be carried out:

- Pausing a paused consumer, or resuming a consuming (or starting) one,
resolves with an error and changes nothing.
- Resuming or pausing a cancelled consumer, or one being cancelled, resolves
with an error. Cancel is final.
- Resuming or pausing with no consumer resolves with an error. A consumer
added later starts in the requested state.
- Resuming or pausing a channel that is permanently closing
(`gracefulClose()`) resolves with an error, since it won't reopen and the
broker discards anything sent after `channel.close`.

Repeating a request that is still in flight returns the same future.

## When the future resolves

| Request | Resolves |
|---|---|
| Pause, consumer consuming | On `basic.cancel-ok` |
| Pause, consumer starting | On `basic.cancel-ok` for the `basic.cancel` sent once `basic.consume-ok` arrives |
| Pause, nothing consuming on the broker (e.g. channel reopening) | Immediately; the channel reopens paused |
| Resume, channel ready | On `basic.consume-ok` |
| Resume, channel closed or reopening | On `basic.consume-ok` after the channel reopens |
| Resume, pause in flight | On `basic.consume-ok` for the `basic.consume` sent once the pause's `basic.cancel-ok` arrives |

A request overtaken by the opposite one before it completes resolves with
an error:

- A pending resume fails as soon as a pause or cancel is requested.
- A pending pause fails when its `basic.cancel-ok` arrives, if a resume was
requested meanwhile; the resume's `basic.consume` follows straight away.

Futures still waiting at a channel reset resolve with an error, but the
latest request stands: the reopened channel carries it out.

## Reopening a channel

A channel that resets reopens in the state the consumer was last asked to
be in:

| Latest request at reset | Reopens |
|---|---|
| Resume (or none since consuming) | Consuming |
| Pause | Paused |
| Cancel | Consumer removed |

## Cancelling

Cancelling a paused consumer, or one waiting for the channel to reopen,
completes immediately, since nothing is consuming on the broker. Cancelling
while a pause is in flight turns the pause into a cancel; the future
resolves when `basic.cancel-ok` arrives.
4 changes: 4 additions & 0 deletions src/rmq/rmqamqp/rmqamqp_channel.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,10 @@ class Channel : public bsl::enable_shared_from_this<Channel> {

const bsl::string& vhostName() const;

/// Return true once gracefulClose() has been called: the channel will
/// not reopen
bool isPermanentlyClosing() const { return d_permanentlyClosing; }

virtual const char* channelType() const = 0;

virtual bsl::vector<bsl::pair<bsl::string, bsl::string> >
Expand Down
160 changes: 125 additions & 35 deletions src/rmq/rmqamqp/rmqamqp_receivechannel.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,12 @@ class ReceiveChannel::Consumer {
const bsl::string& tag,
const bool channelPausedOnOpen);

/// True unless NOT_CONSUMING. A channel reset returns a consumer to
/// NOT_CONSUMING, and one created paused starts PAUSED, so this does not
/// mean the consumer has ever consumed.
bool isStarted() const;
bool isNotConsuming() const;
bool isStarting() const;
bool isActive() const;
bool isPaused() const;
bool isPausing() const;
Expand Down Expand Up @@ -183,6 +188,16 @@ bool ReceiveChannel::Consumer::isStarted() const
return d_state != NOT_CONSUMING;
}

bool ReceiveChannel::Consumer::isNotConsuming() const
{
return d_state == NOT_CONSUMING;
}

bool ReceiveChannel::Consumer::isStarting() const
{
return d_state == STARTING;
}

bool ReceiveChannel::Consumer::isActive() const { return d_state == CONSUMING; }

bool ReceiveChannel::Consumer::isPaused() const { return d_state == PAUSED; }
Expand Down Expand Up @@ -298,6 +313,15 @@ ReceiveChannel::Consumer::cancel(const bool pause)
d_state = CANCELLING;
}

if (pause && d_state == NOT_CONSUMING) {
// No Basic.Consume has been sent (e.g. the channel is reopening), so
// there is nothing to cancel and no CancelOk will arrive
BALL_LOG_INFO << "Pausing consumer: " << d_tag
<< " which is not consuming, queue: " << queueName();
d_state = PAUSED;
return method;
}

if (d_state != CONSUMING) {
BALL_LOG_WARN << (pause ? "Pause" : "Cancel")
<< " called in state other than CONSUMING. State: "
Expand Down Expand Up @@ -395,6 +419,8 @@ void ReceiveChannel::onOpen()
void ReceiveChannel::onReset()
{
d_nextMessage.reset();
d_pauseStartTime = bsls::TimeInterval();
d_resumeStartTime = bsls::TimeInterval();
d_multipleAckHandler.reset();

// Drop and lock all pending acknowledgement batches
Expand Down Expand Up @@ -483,49 +509,71 @@ rmqt::Result<> ReceiveChannel::consume(const rmqt::QueueHandle& queue,

rmqt::Future<> ReceiveChannel::resume()
{
d_channelPausedOnOpen = false;
if (isPermanentlyClosing()) {
return rmqt::Future<>(rmqt::Result<>(
"Resume called on a channel that is permanently closing"));
}

if (d_consumer && !d_consumer->shouldRestart()) {
return rmqt::Future<>(
rmqt::Result<>("Resume called with a cancel already in flight"));
}

if (d_consumeFuturePair) {
BALL_LOG_WARN << "Resume called with a resume already in flight";
return d_consumeFuturePair->second;
}

if (!d_consumer) {
// A consumer added later starts consuming
d_channelPausedOnOpen = false;
return rmqt::Future<>(
rmqt::Result<>("Resume called with no existing consumer"));
}

if (d_cancelFuturePair) {
return rmqt::Future<>(rmqt::Result<>(
"Resume called with a " +
bsl::string(d_consumer->isPausing() ? "pause" : "cancel") +
" already in flight"));
}

if (!d_consumer->isPaused()) {
if (d_consumer->isActive() || d_consumer->isStarting()) {
return rmqt::Future<>(
rmqt::Result<>("Resume called on unpaused consumer"));
}

if (state() == READY || state() == AWAITING_REPLY) {
// From here the resume is accepted, and takes effect at the next
// opportunity. Only count it if it undoes a pause, rather than landing
// whilst a consumer that was never paused reconnects.
const bool wasPaused = d_channelPausedOnOpen || d_consumer->isPaused() ||
d_consumer->isPausing();
d_channelPausedOnOpen = false;

if (wasPaused) {
// Track resume start time and increment counter
d_resumeStartTime = bsls::SystemTime::nowMonotonicClock();
d_metricPublisher->publishCounter(
Metrics::CONSUMER_RESUME_TOTAL, 1.0, getMetricTags());
}

d_consumeFuturePair =
bslma::ManagedPtrUtil::makeManaged<rmqt::Future<>::Pair>(
rmqt::Future<>::make());
d_consumeFuturePair =
bslma::ManagedPtrUtil::makeManaged<rmqt::Future<>::Pair>(
rmqt::Future<>::make());

if (d_consumer->isPaused()) {
if (state() == READY || state() == AWAITING_REPLY) {
restartConsumers();
}
else {
BALL_LOG_TRACE << "Resume called before channel ready, "
"consumertag: "
<< d_consumer->consumerTag()
<< ", consuming once the channel reopens";
d_consumer->reset();
}
}
else if (d_consumer->isNotConsuming() &&
(state() == READY || state() == AWAITING_REPLY)) {
restartConsumers();

return d_consumeFuturePair->second;
}
// Otherwise the consumer is NOT_CONSUMING, and is started when the channel
// reopens, or PAUSING, and is restarted when CancelOk arrives

BALL_LOG_TRACE << "Resume called before channel ready, consumertag: "
<< d_consumer->consumerTag() << " pending";

return rmqt::Future<>(rmqt::Result<>("Resume called before channel ready"));
return d_consumeFuturePair->second;
}

ReceiveChannel::~ReceiveChannel()
Expand Down Expand Up @@ -599,6 +647,8 @@ bool ReceiveChannel::consumerIsPaused() const

rmqt::Future<> ReceiveChannel::cancel()
{
failPendingResume("Consumer cancelled before it could be started");

if (d_cancelFuturePair) {
if (d_consumer->isPausing()) {
BALL_LOG_WARN << "Cancel called with a pause already in "
Expand Down Expand Up @@ -645,14 +695,20 @@ rmqt::Future<> ReceiveChannel::cancel()

rmqt::Future<> ReceiveChannel::pause()
{
if (isPermanentlyClosing()) {
return rmqt::Future<>(rmqt::Result<>(
"Pause called on a channel that is permanently closing"));
}

if (d_consumer && !d_consumer->shouldRestart()) {
return rmqt::Future<>(
rmqt::Result<>("Pause called with a cancel already in flight"));
}

d_channelPausedOnOpen = true;
failPendingResume("Resume superseded by pause");

if (d_cancelFuturePair) {
if (d_consumer->isCancelling()) {
return rmqt::Future<>(
rmqt::Result<>("Pause called with a cancel already in flight"));
}

BALL_LOG_WARN << "Pause called with a pause already in flight";
return d_cancelFuturePair->second;
}
Expand All @@ -663,27 +719,33 @@ rmqt::Future<> ReceiveChannel::pause()
}

if (d_consumer->canPause()) {
// Track pause start time and increment counter
d_pauseStartTime = bsls::SystemTime::nowMonotonicClock();
d_metricPublisher->publishCounter(
Metrics::CONSUMER_PAUSE_TOTAL, 1.0, getMetricTags());

const bool pause = true;
bsl::optional<rmqamqpt::BasicMethod> method = d_consumer->cancel(pause);

if (d_consumer->isPaused()) {
// Nothing was consuming on the broker, so the pause has already
// taken effect
return rmqt::Future<>(rmqt::Result<>());
}

// Track pause start time
d_pauseStartTime = bsls::SystemTime::nowMonotonicClock();

d_cancelFuturePair =
bslma::ManagedPtrUtil::makeManaged<rmqt::Future<>::Pair>(
rmqt::Future<>::make());

const bool pause = true;
bsl::optional<rmqamqpt::BasicMethod> method = d_consumer->cancel(pause);

if (method) {
writeMessage(Message(rmqamqpt::Method(method.value())),
AWAITING_REPLY);
}
return d_cancelFuturePair->second;
}

BALL_LOG_DEBUG << "Pause called when consumer is already paused";
return rmqt::Future<>(rmqt::Result<>());
return rmqt::Future<>(rmqt::Result<>("Pause called on paused consumer"));
}

rmqt::Future<> ReceiveChannel::drain()
Expand Down Expand Up @@ -754,7 +816,9 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic)
AWAITING_REPLY);
}

if (d_consumeFuturePair) {
// A resume requested after a pause waits for the pause's
// CancelOk and the Basic.Consume that follows it
if (d_consumeFuturePair && d_consumer->isActive()) {
// Publish resume latency
publishLatencyIfTracked(
&d_resumeStartTime,
Expand All @@ -779,7 +843,7 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic)
} break;
case rmqamqpt::BasicQoSOk::METHOD_ID: {
if (d_consumer) {
if (d_channelPausedOnOpen) {
if (d_channelPausedOnOpen && d_consumer->isNotConsuming()) {
d_consumer->setPaused();
}

Expand All @@ -789,11 +853,14 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic)
d_hungProgressTimer->cancel();
ready();
}
else {
else if (d_consumer->isNotConsuming()) {
// we've restarted, we already have a callback so can
// declare consumer
restartConsumers();
}
// Otherwise a resume issued whilst the QoS was in flight has
// already sent Basic.Consume, and its reply will mark the
// channel ready
}
else {
// This is the first connection, or recovery from connection
Expand Down Expand Up @@ -822,6 +889,10 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic)
basic.the<rmqamqpt::BasicCancelOk>().consumerTag()) {
d_consumer->cancelOk();

// A resume requested whilst the pause was in flight
const bool resumeRequested =
d_consumer->isPaused() && !d_channelPausedOnOpen;

if (d_consumer->isPaused()) {
BALL_LOG_INFO << "Pausing Consumer: "
<< d_consumer->consumerTag();
Expand All @@ -834,12 +905,18 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic)
d_consumer.reset();
}

if (d_consumeFuturePair) {
if (d_consumeFuturePair && !resumeRequested) {
d_consumeFuturePair->first(
rmqt::Result<>("Consumer cancelled before it could be "
"started"));
d_consumeFuturePair.reset();
}
if (d_cancelFuturePair && resumeRequested) {
d_pauseStartTime = bsls::TimeInterval();
d_cancelFuturePair->first(
rmqt::Result<>("Pause superseded by resume"));
d_cancelFuturePair.reset();
}
if (d_cancelFuturePair) {
// Publish pause latency
publishLatencyIfTracked(
Expand All @@ -855,6 +932,10 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic)
// Ensure any waitForReady() futures are resolved before we
// enter the fully cancelled state
ready();

if (resumeRequested) {
restartConsumers();
}
}
else {
invalidConsumerError(
Expand Down Expand Up @@ -894,6 +975,15 @@ void ReceiveChannel::processMessage(const rmqt::Message& message)
}
}

void ReceiveChannel::failPendingResume(const bsl::string& reason)
{
if (d_consumeFuturePair) {
d_resumeStartTime = bsls::TimeInterval();
d_consumeFuturePair->first(rmqt::Result<>(reason));
d_consumeFuturePair.reset();
}
}

void ReceiveChannel::restartConsumers()
{
if (d_consumer) {
Expand Down
Loading
Loading