Skip to content

Avoid sending duplicate channel.open #114

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
michalovits merged 2 commits into from
Nov 12, 2026
Merged

Avoid sending duplicate channel.open #114

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4bcfa02
Fix cases of duplicate channel open
michalovits Nov 9, 2026
e68261b
Skipped retry logs -> INFO
michalovits Nov 12, 2026
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
56 changes: 40 additions & 16 deletions src/rmq/rmqamqp/rmqamqp_channel.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,9 +145,19 @@ class Channel::ChannelMethodProcessor {
BALL_LOG_INFO << "Received ChannelClose from server: " << closeMethod;

rmqamqpt::ChannelCloseOk closeOkMethod;
d_channel.writeMessage(
Message(rmqamqpt::Method(rmqamqpt::ChannelMethod(closeOkMethod))),
CLOSED);
const Message closeOk =
Message(rmqamqpt::Method(rmqamqpt::ChannelMethod(closeOkMethod)));

if (d_channel.d_state == CHANNEL_CLOSE_SENT) {
d_channel.writeMessage(closeOk, &noopWriteHandler);
}
else {
d_channel.updateState(CLOSED);
d_channel.writeMessage(
closeOk,
bdlf::BindUtil::bind(&Channel::onCloseOkWritten,
d_channel.weak_from_this()));
}

if (closeMethod.classId() || closeMethod.methodId()) {
(d_channel.d_retryHandler->errorCallback())(
Expand All @@ -160,6 +170,12 @@ class Channel::ChannelMethodProcessor {

ResultType operator()(const rmqamqpt::ChannelCloseOk&) const
{
if (d_channel.d_state == CHANNEL_OPEN_SENT) {
BALL_LOG_INFO << "Ignoring channel CLOSE-OK from before the "
"channel was reopened";
return KEEP;
}

if (d_channel.d_state != CHANNEL_CLOSE_SENT) {
BALL_LOG_ERROR << "Received channel CLOSE-OK method from server "
"without sending CLOSE method";
Expand Down Expand Up @@ -443,19 +459,16 @@ void Channel::processMessage(const rmqt::Message&)
BALL_LOG_ERROR << "Throwing Content Away";
}

void Channel::onWriteComplete(const bsl::weak_ptr<Channel>& weakSelf)
void Channel::onCloseOkWritten(const bsl::weak_ptr<Channel>& weakSelf)
{
bsl::shared_ptr<Channel> self = weakSelf.lock();
if (!self) {
BALL_LOG_TRACE
<< "Write completion handler called after channel destroyed";
BALL_LOG_TRACE << "Close-Ok write completed after channel destroyed";
return;
}

if (self->d_state == CLOSED) {
const bool startRetry = !self->d_permanentlyClosing;
self->reset(startRetry);
}
const bool startRetry = !self->d_permanentlyClosing;
self->reset(startRetry);
}

Channel::CleanupIndicator Channel::reset(bool startRetry)
Expand All @@ -466,23 +479,31 @@ Channel::CleanupIndicator Channel::reset(bool startRetry)

onReset();
if (startRetry) {
d_retryHandler->retry(
bdlf::BindUtil::bind(&Channel::retry, weak_from_this()));
d_retryHandler->retry(bdlf::BindUtil::bind(
&Channel::retry, weak_from_this(), lifetimeId()));
}
d_metricPublisher->publishCounter(
"channel_disconnected", 1, getVHostAndChannelTags());

return d_permanentlyClosing ? CLEANUP : KEEP;
}

void Channel::retry(const bsl::weak_ptr<Channel>& weakSelf)
void Channel::retry(const bsl::weak_ptr<Channel>& weakSelf,
size_t scheduledLifetimeId)
{
bsl::shared_ptr<Channel> self = weakSelf.lock();
if (!self) {
BALL_LOG_TRACE << "Retry handler called during shutdown";
return;
}

if (self->lifetimeId() != scheduledLifetimeId) {
BALL_LOG_INFO << "Ignoring retry scheduled in channel lifetime "
<< scheduledLifetimeId << ", current lifetime is "
<< self->lifetimeId();
return;
}

self->open();
}

Expand Down Expand Up @@ -513,6 +534,11 @@ void Channel::open()
return;
}

if (d_state != CLOSED) {
BALL_LOG_INFO << "Not opening a channel in state " << d_state;
return;
}

using bdlf::PlaceHolders::_1;
d_hungProgressTimer->start(
bdlf::BindUtil::bind(&Channel::channelHung, weak_from_this(), _1));
Expand All @@ -526,9 +552,7 @@ void Channel::writeMessage(const rmqamqp::Message& message, State newState)
{
updateState(newState);
BALL_LOG_TRACE << "State now set to: " << d_state;
d_onAsyncWrite(
bsl::make_shared<Message>(message),
bdlf::BindUtil::bind(&Channel::onWriteComplete, weak_from_this()));
d_onAsyncWrite(bsl::make_shared<Message>(message), &noopWriteHandler);
}

void Channel::writeMessage(
Expand Down
5 changes: 3 additions & 2 deletions src/rmq/rmqamqp/rmqamqp_channel.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,15 +152,16 @@ class Channel : public bsl::enable_shared_from_this<Channel> {
void ready();

// Callback for writes
static void onWriteComplete(const bsl::weak_ptr<Channel>& weakSelf);
static void onCloseOkWritten(const bsl::weak_ptr<Channel>& weakSelf);

void writeMessage(const rmqamqp::Message& message, State newState);

void writeMessage(const rmqamqp::Message& message,
const rmqio::Connection::SuccessWriteCallback& callBack);

/// Callback for re-opening channel
static void retry(const bsl::weak_ptr<Channel>& weakSelf);
static void retry(const bsl::weak_ptr<Channel>& weakSelf,
size_t scheduledLifetimeId);

const bsl::string& vhostName() const;

Expand Down
98 changes: 93 additions & 5 deletions src/tests/rmqamqp/rmqamqp_channel.t.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,6 +199,24 @@ TEST_F(ChannelTests, OpenChannel)
bsl::shared_ptr<ChannelTestImpl> channel = makeOpeningChannel(d_topology);
}

TEST_F(ChannelTests, OpenWhileOpeningSendsNothing)
{
bsl::shared_ptr<ChannelTestImpl> channel = makeOpeningChannel(d_topology);

EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0);
channel->open();
}

TEST_F(ChannelTests, CloseOkWhileOpeningSendsNothing)
{
bsl::shared_ptr<ChannelTestImpl> channel = makeOpeningChannel(d_topology);

EXPECT_CALL(*d_retryHandler, retry(_)).WillRepeatedly(InvokeArgument<0>());
EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0);
channel->processReceived(rmqamqp::Message(
rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk()))));
}

TEST_F(ChannelTests, GracefulCloseOnClosedChannelNeverReopens)
{
bsl::shared_ptr<ChannelTestImpl> mockChannel = makeChannel(d_topology);
Expand Down Expand Up @@ -306,6 +324,73 @@ TEST_F(ChannelTests, ReceiveCloseMethodReOpens)
EXPECT_THAT(mockChannel->state(), Eq(rmqamqp::Channel::CHANNEL_OPEN_SENT));
}

TEST_F(ChannelTests, ServerCloseRetriesOnceWithWritesInFlight)
{
bsl::shared_ptr<ChannelTestImpl> mockChannel =
makeOpeningChannel(d_topology);

bsl::function<void()> queueDeclareWritten;
EXPECT_CALL(d_callback, onAsyncWrite(_, _))
.WillOnce(SaveArg<1>(&queueDeclareWritten));
mockChannel->processReceived(rmqamqp::Message(
rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelOpenOk()))));

bsl::function<void()> closeOkWritten;
EXPECT_CALL(d_callback, onAsyncWrite(_, _))
.WillOnce(SaveArg<1>(&closeOkWritten));
mockChannel->processReceived(rmqamqp::Message(
rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose(
rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel")))));

EXPECT_CALL(*d_retryHandler, retry(_));
queueDeclareWritten();
closeOkWritten();
}

TEST_F(ChannelTests, ServerCloseCrossingGracefulCloseCleansUpOnCloseOk)
{
bsl::shared_ptr<ChannelTestImpl> mockChannel =
makeOpeningChannel(d_topology);
mockChannel->setState(rmqamqp::Channel::READY);

EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>());
mockChannel->gracefulClose();

EXPECT_CALL(d_callback,
onAsyncWrite(
Pointee(MessageEq(rmqamqp::Message(rmqamqpt::Method(
rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk()))))),
_))
.WillOnce(InvokeArgument<1>());
mockChannel->processReceived(rmqamqp::Message(
rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose(
rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel")))));

EXPECT_THAT(mockChannel->processReceived(rmqamqp::Message(rmqamqpt::Method(
rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk())))),
Eq(rmqamqp::Channel::CLEANUP));
}

TEST_F(ChannelTests, ServerCloseCrossingOurCloseReopensOnlyAfterCloseOk)
{
bsl::shared_ptr<ChannelTestImpl> mockChannel =
makeOpeningChannel(d_topology);
mockChannel->setState(rmqamqp::Channel::READY);

EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>());
mockChannel->close(rmqamqpt::Constants::CHANNEL_ERROR, "Channel error");

EXPECT_CALL(*d_retryHandler, retry(_)).Times(0);
EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>());
mockChannel->processReceived(rmqamqp::Message(
rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose(
rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel")))));

EXPECT_CALL(*d_retryHandler, retry(_));
mockChannel->processReceived(rmqamqp::Message(
rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk()))));
}

TEST_F(ChannelTests, ReceiveOpenOkMethod)
{
rmqt::Topology emptyTopology;
Expand Down Expand Up @@ -973,19 +1058,22 @@ TEST_F(ChannelTests, sendBindAndUnbind)
EXPECT_TRUE(future2.tryResult());
}

TEST_F(ChannelTests, onWriteCompleteCbValidity)
TEST_F(ChannelTests, onCloseOkWrittenCbValidity)
{
// Ensure that the onWriteComplete callback can be invoked after the Channel
// is destructed.
// Ensure that the onCloseOkWritten callback can be invoked after the
// Channel is destructed.

bsl::function<void()> callbackFn;

{
bsl::shared_ptr<ChannelTestImpl> channel = makeChannel(d_topology);
bsl::shared_ptr<ChannelTestImpl> channel =
makeOpeningChannel(d_topology);

EXPECT_CALL(d_callback, onAsyncWrite(_, _))
.WillOnce(SaveArg<1>(&callbackFn));
channel->open();
channel->processReceived(rmqamqp::Message(
rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose(
rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel")))));
}

// This shouldn't segfault
Expand Down
19 changes: 19 additions & 0 deletions src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2264,6 +2264,25 @@ TEST_F(ReceiveChannelTests, ServerCancelAfterChannelCloseSentIsDiscarded)
Eq(rmqamqp::Channel::CLEANUP));
}

TEST_F(ReceiveChannelTests, ConsumeOkCrossingGracefulCloseStillCleansUp)
{
bsl::shared_ptr<ReceiveChannel> receiveChannel = makeReceiveChannel(1);

makeReady(*receiveChannel);
EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>());
setupConsumerNoReply(*receiveChannel, d_consumerTag);

expectChannelClose();
receiveChannel->gracefulClose();

consumerReply(*receiveChannel, d_consumerTag);

EXPECT_THAT(
receiveChannel->processReceived(rmqamqp::Message(rmqamqpt::Method(
rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk())))),
Eq(rmqamqp::Channel::CLEANUP));
}

TEST_F(ReceiveChannelTests, ResumeRefusedWhilstClosingForGood)
{
bsl::shared_ptr<ReceiveChannel> receiveChannel = makeReceiveChannel(1);
Expand Down
18 changes: 18 additions & 0 deletions src/tests/rmqamqp/rmqamqp_sendchannel.t.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -515,6 +515,24 @@ TEST_F(SendChannelTests, MessagesArePublishedAfterReconnecting)
startupExpectations(*d_sendChannel);
}

TEST_F(SendChannelTests, RetryScheduledBeforeResetDoesNotReopen)
{
startupExpectations(*d_sendChannel);

rmqio::RetryHandler::RetryCallback scheduledRetry;
EXPECT_CALL(*d_retryHandler, retry(_))
.WillOnce(SaveArg<0>(&scheduledRetry));
EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>());
d_sendChannel->processReceived(rmqamqp::Message(
rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose(
rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel")))));

d_sendChannel->reset();

EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0);
scheduledRetry();
}

TEST_F(SendChannelTests, MessagesArePublishedAfterFlow)
{
rmqt::Message message;
Expand Down
Loading