Skip to content

Fix[mqbnet::InitialConnectionContext]: propagate close to authn #1639

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
678098 merged 1 commit into from
Jul 27, 2026
Merged

Fix[mqbnet::InitialConnectionContext]: propagate close to authn #1639

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
2 changes: 1 addition & 1 deletion src/groups/mqb/mqba/mqba_authenticator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@
bsl::shared_ptr<mqbnet::AuthenticationContext> authenticationContext =
bsl::allocate_shared<mqbnet::AuthenticationContext>(
d_allocator_p,
d_scheduler_p,
context_p, // initialConnectionContext
authenticationMsg.authenticationRequest()
.mechanism(), // mechanism
Expand Down Expand Up @@ -353,7 +354,6 @@
bmqu::MemOutStream scheduleErrStream;
const int scheduleRc = context_sp->setAuthenticatedAndScheduleReauthn(
scheduleErrStream,
d_scheduler_p,
result->lifetimeMs(),
channel);
if (scheduleRc != 0) {
Expand Down Expand Up @@ -404,7 +404,7 @@
, d_threadPool(bslmt::ThreadAttributes(),
mqbcfg::BrokerConfig::get().authentication().minThreads(),
mqbcfg::BrokerConfig::get().authentication().maxThreads(),
bsls::TimeInterval(120).totalMilliseconds(), // idle time

Check warning on line 407 in src/groups/mqb/mqba/mqba_authenticator.cpp

View workflow job for this annotation

GitHub Actions / Build [ubuntu] / Build [ubuntu, cpp23] c147d2be7cfa54757a2dc3d779c849cd861168a4 bmqbrkr bmqtool bmqstoragetool all.it 

conversion from ‘BloombergLP::bsls::Types::Int64’ {aka ‘long long int’} to ‘int’ may change value [-Wconversion]

Check warning on line 407 in src/groups/mqb/mqba/mqba_authenticator.cpp

View workflow job for this annotation

GitHub Actions / Build [ubuntu, C++03] / Build [ubuntu, cpp03] c147d2be7cfa54757a2dc3d779c849cd861168a4 bmqbrkr bmqtool bmq.t mqb.t 

conversion from ‘BloombergLP::bsls::Types::Int64’ {aka ‘long long int’} to ‘int’ may change value [-Wconversion]

Check warning on line 407 in src/groups/mqb/mqba/mqba_authenticator.cpp

View workflow job for this annotation

GitHub Actions / UT [c++] / Build [ubuntu, cpp23] c147d2be7cfa54757a2dc3d779c849cd861168a4 all.t 

conversion from ‘BloombergLP::bsls::Types::Int64’ {aka ‘long long int’} to ‘int’ may change value [-Wconversion]
allocator)
, d_blobSpPool_p(blobSpPool)
, d_scheduler_p(scheduler)
Expand Down
18 changes: 8 additions & 10 deletions src/groups/mqb/mqbnet/mqbnet_authenticationcontext.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,13 +100,15 @@ bool AuthenticationState::fromAscii(AuthenticationState::Enum* out,
// ---------------------------

AuthenticationContext::AuthenticationContext(
bdlmt::EventScheduler* scheduler,
InitialConnectionContext* initialConnectionContext,
bsl::string_view mechanism,
const bmqp_ctrlmsg::AuthenticationMessage& authenticationMessage,
bmqp::EncodingType::Enum authenticationEncodingType,
AuthenticationState::Enum state,
bslma::Allocator* allocator)
: d_allocator_p(allocator)
, d_scheduler_p(scheduler)
, d_self(this) // use default allocator
, d_mutex()
, d_authenticationResultSp()
Expand All @@ -117,7 +119,8 @@ AuthenticationContext::AuthenticationContext(
, d_authenticationMessage(authenticationMessage)
, d_encodingType(authenticationEncodingType)
{
// NOTHING
// PRECONDITION
BSLS_ASSERT_SAFE(d_scheduler_p);

@678098 678098 Jul 24, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Cached single scheduler d_scheduler_p so we have a guarantee that the same scheduler is used for both scheduling/cancelling events.
Before, the API has weakness, allowing providing different schedulers

}

void AuthenticationContext::setAuthenticationResult(
Expand Down Expand Up @@ -153,14 +156,12 @@ void AuthenticationContext::resetAuthenticationMessage()

int AuthenticationContext::setAuthenticatedAndScheduleReauthn(
bsl::ostream& errorDescription,
bdlmt::EventScheduler* scheduler_p,
const bsl::optional<bsls::Types::Uint64>& lifetimeMs,
const bsl::shared_ptr<bmqio::Channel>& channel_sp)
{
// executed by an *AUTHENTICATION* thread

// PRECONDITION
BSLS_ASSERT_SAFE(scheduler_p);
BSLS_ASSERT_SAFE(channel_sp);

bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex); // LOCKED
Expand All @@ -175,14 +176,14 @@ int AuthenticationContext::setAuthenticatedAndScheduleReauthn(

d_state = AuthenticationState::e_AUTHENTICATED;

scheduler_p->cancelEventAndWait(&d_timeoutHandle);
d_scheduler_p->cancelEventAndWait(&d_timeoutHandle);

if (!lifetimeMs.has_value()) {
return 0;
}
const bsls::Types::Uint64 lifetime = lifetimeMs.value();

scheduler_p->scheduleEvent(
d_scheduler_p->scheduleEvent(
&d_timeoutHandle,
bsls::TimeInterval(bmqu::Time::nowMonotonicClock())
.addMilliseconds(lifetime),
Expand Down Expand Up @@ -250,21 +251,18 @@ void AuthenticationContext::onReauthenticationError(
channel_sp->close(status);
}

void AuthenticationContext::onClose(bdlmt::EventScheduler* scheduler_p)
void AuthenticationContext::onClose()
{
// executed by *ANY* thread

// PRECONDITIONS
BSLS_ASSERT_SAFE(scheduler_p);

@678098 678098 Jul 24, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not null due to invariant check on construction


bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex); // LOCKED

if (d_state == AuthenticationState::e_CLOSED) {
return; // idempotent
}
d_state = AuthenticationState::e_CLOSED;

scheduler_p->cancelEventAndWait(&d_timeoutHandle);
d_scheduler_p->cancelEventAndWait(&d_timeoutHandle);
}

bool AuthenticationContext::tryStartReauthentication()
Expand Down
21 changes: 12 additions & 9 deletions src/groups/mqb/mqbnet/mqbnet_authenticationcontext.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,9 @@ class AuthenticationContext {
/// Allocator to use.
bslma::Allocator* d_allocator_p;

/// Scheduler used to schedule and cancel the reauthentication timer.
bdlmt::EventScheduler* d_scheduler_p;

/// Used to make sure no callback is invoked on a destroyed object.
bmqu::SharedResource<AuthenticationContext> d_self;

Expand Down Expand Up @@ -179,6 +182,7 @@ class AuthenticationContext {
bslma::UsesBslmaAllocator)
// CREATORS
AuthenticationContext(
bdlmt::EventScheduler* scheduler,
InitialConnectionContext* initialConnectionContext,
bsl::string_view mechanism,
const bmqp_ctrlmsg::AuthenticationMessage& authenticationMessage,
Expand All @@ -201,14 +205,13 @@ class AuthenticationContext {

void resetAuthenticationMessage();

/// Schedule a reauthentication timer using the specified `scheduler_p`
/// with the specified `lifetimeMs`. The specified `channel_sp` is used to
/// close the connection in case of reauthentication timeout or error.
/// Return 0 on success, and a non-zero value populating the specified
/// `errorDescription` with details on failure.
/// @brief Mark as authenticated and schedule a reauthentication timer.
/// @param[out] errorDescription Populated with details on failure.
/// @param lifetimeMs Duration after which reauthentication is required.
/// @param channel_sp Channel closed on reauthentication timeout or error.
/// @return 0 on success, and a non-zero value on failure.
int setAuthenticatedAndScheduleReauthn(
bsl::ostream& errorDescription,
bdlmt::EventScheduler* scheduler_p,
const bsl::optional<bsls::Types::Uint64>& lifetimeMs,
const bsl::shared_ptr<bmqio::Channel>& channel_sp);

Expand All @@ -226,9 +229,9 @@ class AuthenticationContext {
int errorCode,
const bsl::string& errorDescription);

/// Called when a channel is closing. Cancel any outstanding
/// reauthentication timer using the specified `scheduler_p`.
void onClose(bdlmt::EventScheduler* scheduler_p);
/// @brief Cancel any outstanding reauthentication timer when the channel
/// is closing.
void onClose();

/// Attempt to begin reauthentication by transitioning the state from
/// AUTHENTICATED to AUTHENTICATING.
Expand Down
21 changes: 7 additions & 14 deletions src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,7 @@
bmqp_ctrlmsg::AuthenticationMessage authnMsg;
return bsl::allocate_shared<mqbnet::AuthenticationContext>(
d_allocator_p,
&d_scheduler,
static_cast<mqbnet::InitialConnectionContext*>(0),
"testMechanism",
authnMsg,
Expand Down Expand Up @@ -149,13 +150,12 @@
bmqu::MemOutStream errStream(alloc);
bsl::optional<bsls::Types::Uint64> noLifetime;

int rc = ctx->setAuthenticatedAndScheduleReauthn(errStream,

Check failure on line 153 in src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp

View workflow job for this annotation

GitHub Actions / C++ Linter Check

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp:153:9 [cppcoreguidelines-init-variables] 

variable 'rc' is not initialized
&tb.d_scheduler,
noLifetime,
tb.d_channel);
BMQTST_ASSERT_EQ(rc, 0);

ctx->onClose(&tb.d_scheduler);
ctx->onClose();
}

static void test2_zeroLifetimeTimeout()
Expand Down Expand Up @@ -187,8 +187,7 @@
bsl::optional<bsls::Types::Uint64> lifetime(lifetimeMs);

bmqu::MemOutStream errStream(alloc);
int rc = ctx->setAuthenticatedAndScheduleReauthn(errStream,

Check failure on line 190 in src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp

View workflow job for this annotation

GitHub Actions / C++ Linter Check

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp:190:24 [cppcoreguidelines-init-variables] 

variable 'rc' is not initialized
&tb.d_scheduler,
lifetime,
tb.d_channel);
BMQTST_ASSERT_EQ(rc, 0);
Expand All @@ -208,7 +207,7 @@
".*Reauthentication timeout.*",
alloc));

ctx->onClose(&tb.d_scheduler);
ctx->onClose();
}

static void test3_reauthenticationTimeout()
Expand Down Expand Up @@ -246,8 +245,7 @@
bsl::optional<bsls::Types::Uint64> lifetime(lifetimeMs);

bmqu::MemOutStream errStream(alloc);
int rc = ctx->setAuthenticatedAndScheduleReauthn(errStream,

Check failure on line 248 in src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp

View workflow job for this annotation

GitHub Actions / C++ Linter Check

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp:248:24 [cppcoreguidelines-init-variables] 

variable 'rc' is not initialized
&tb.d_scheduler,
lifetime,
tb.d_channel);
BMQTST_ASSERT_EQ(rc, 0);
Expand Down Expand Up @@ -280,7 +278,7 @@
".*Reauthentication timeout.*",
alloc));

ctx->onClose(&tb.d_scheduler);
ctx->onClose();
}

static void test4_reauthenticationBeforeTimeout()
Expand Down Expand Up @@ -315,8 +313,7 @@
bmqu::MemOutStream errStream(alloc);

// 1) Initial authentication with lifetime
int rc = ctx->setAuthenticatedAndScheduleReauthn(errStream,

Check failure on line 316 in src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp

View workflow job for this annotation

GitHub Actions / C++ Linter Check

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp:316:9 [cppcoreguidelines-init-variables] 

variable 'rc' is not initialized
&tb.d_scheduler,
lifetime,
tb.d_channel);
BMQTST_ASSERT_EQ(rc, 0);
Expand All @@ -338,7 +335,6 @@
bsl::optional<bsls::Types::Uint64> newLifetime(newLifetimeMs);

rc = ctx->setAuthenticatedAndScheduleReauthn(errStream,
&tb.d_scheduler,
newLifetime,
tb.d_channel);
BMQTST_ASSERT_EQ(rc, 0);
Expand All @@ -353,7 +349,7 @@
// 7) Now the new timer fires and closes the channel
BMQTST_ASSERT_EQ(tb.d_channel->numCloseCalls(), 1u);

ctx->onClose(&tb.d_scheduler);
ctx->onClose();
}

static void test5_noLifetimeNoTimer()
Expand All @@ -380,8 +376,7 @@
bmqu::MemOutStream errStream(alloc);
bsl::optional<bsls::Types::Uint64> noLifetime;

int rc = ctx->setAuthenticatedAndScheduleReauthn(errStream,

Check failure on line 379 in src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp

View workflow job for this annotation

GitHub Actions / C++ Linter Check

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp:379:9 [cppcoreguidelines-init-variables] 

variable 'rc' is not initialized
&tb.d_scheduler,
noLifetime,
tb.d_channel);
BMQTST_ASSERT_EQ(rc, 0);
Expand All @@ -391,7 +386,7 @@

BMQTST_ASSERT_EQ(tb.d_channel->numCloseCalls(), 0u);

ctx->onClose(&tb.d_scheduler);
ctx->onClose();
}

static void test6_onCloseBeforeTimeout()
Expand Down Expand Up @@ -423,14 +418,13 @@
bsl::optional<bsls::Types::Uint64> lifetime(lifetimeMs);

bmqu::MemOutStream errStream(alloc);
int rc = ctx->setAuthenticatedAndScheduleReauthn(errStream,

Check failure on line 421 in src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp

View workflow job for this annotation

GitHub Actions / C++ Linter Check

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp:421:24 [cppcoreguidelines-init-variables] 

variable 'rc' is not initialized
&tb.d_scheduler,
lifetime,
tb.d_channel);
BMQTST_ASSERT_EQ(rc, 0);

// 2) Close before timeout
ctx->onClose(&tb.d_scheduler);
ctx->onClose();

// 3) Advance time past the lifetime
tb.d_testClock.d_timeSource.advanceTime(
Expand Down Expand Up @@ -470,8 +464,7 @@
bsl::optional<bsls::Types::Uint64> lifetime(lifetimeMs);

bmqu::MemOutStream errStream(alloc);
int rc = ctx->setAuthenticatedAndScheduleReauthn(errStream,

Check failure on line 467 in src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp

View workflow job for this annotation

GitHub Actions / C++ Linter Check

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.t.cpp:467:13 [cppcoreguidelines-init-variables] 

variable 'rc' is not initialized
&tb.d_scheduler,
lifetime,
tb.d_channel);
BMQTST_ASSERT_EQ(rc, 0);
Expand Down
29 changes: 17 additions & 12 deletions src/groups/mqb/mqbnet/mqbnet_initialconnectioncontext.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,9 +213,9 @@ InitialConnectionContext::InitialConnectionContext(
, d_negotiator_p(negotiator)
, d_resultState_p(resultState)
, d_userData_sp(userData)
, d_channelSp(channel)
, d_authenticationCtxSp()
, d_negotiationCtxSp()
, d_channel_sp(channel)
, d_authenticationCtx_sp()
, d_negotiationCtx_sp()
, d_initialConnectionCompleteCb(initialConnectionCompleteCb)
, d_authenticationEncodingType(bmqp::EncodingType::e_BER)
, d_state(InitialConnectionState::e_INITIAL)
Expand All @@ -236,7 +236,7 @@ void InitialConnectionContext::setState(InitialConnectionState::Enum value,
InitialConnectionEvent::Enum event)
{
BALL_LOG_DEBUG << "State transition: " << d_state << " -> (" << event
<< ") -> " << value << " [peer: " << d_channelSp.get()
<< ") -> " << value << " [peer: " << d_channel_sp.get()
<< "]";
d_state = value;
}
Expand Down Expand Up @@ -402,11 +402,11 @@ int InitialConnectionContext::decodeInitialConnectionMessage(

void InitialConnectionContext::createNegotiationContext()
{
if (d_negotiationCtxSp) {
if (d_negotiationCtx_sp) {
return; // RETURN
}

d_negotiationCtxSp = bsl::allocate_shared<mqbnet::NegotiationContext>(
d_negotiationCtx_sp = bsl::allocate_shared<mqbnet::NegotiationContext>(
d_allocator_p,
this // initialConnectionContext
);
Expand Down Expand Up @@ -455,14 +455,19 @@ void InitialConnectionContext::setAuthenticationContext(
const bsl::shared_ptr<AuthenticationContext>& value)
{
// PRECONDITIONS
BSLS_ASSERT_SAFE(!d_authenticationCtxSp);
BSLS_ASSERT_SAFE(!d_authenticationCtx_sp);

d_authenticationCtxSp = value;
d_authenticationCtx_sp = value;
}

void InitialConnectionContext::onClose()
{
d_isClosed = true;

// Propagate close
if (d_authenticationCtx_sp) {
d_authenticationCtx_sp->onClose();
}
}

void InitialConnectionContext::readCallback(const bmqio::Status& status,
Expand Down Expand Up @@ -539,7 +544,7 @@ void InitialConnectionContext::handleEvent(

BALL_LOG_DEBUG << "Enter InitialConnectionContext::handleEvent: "
<< "state = " << d_state << ", event = " << event
<< " [peer: " << d_channelSp.get() << "]";
<< " [peer: " << d_channel_sp.get() << "]";

InitialConnectionState::Enum oldState = d_state;

Expand Down Expand Up @@ -731,7 +736,7 @@ void* InitialConnectionContext::resultState() const
const bsl::shared_ptr<bmqio::Channel>&
InitialConnectionContext::channel() const
{
return d_channelSp;
return d_channel_sp;
}

bmqp::EncodingType::Enum
Expand All @@ -743,13 +748,13 @@ InitialConnectionContext::authenticationEncodingType() const
const bsl::shared_ptr<AuthenticationContext>&
InitialConnectionContext::authenticationContext() const
{
return d_authenticationCtxSp;
return d_authenticationCtx_sp;
}

const bsl::shared_ptr<NegotiationContext>&
InitialConnectionContext::negotiationContext() const
{
return d_negotiationCtxSp;
return d_negotiationCtx_sp;
}

InitialConnectionState::Enum InitialConnectionContext::state() const
Expand Down
8 changes: 4 additions & 4 deletions src/groups/mqb/mqbnet/mqbnet_initialconnectioncontext.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,14 +278,14 @@ class InitialConnectionContext {
bsl::shared_ptr<NegotiationUserData> d_userData_sp;

/// The channel to use for the initial connection.
bsl::shared_ptr<bmqio::Channel> d_channelSp;
bsl::shared_ptr<bmqio::Channel> d_channel_sp;

/// The AuthenticationContext updated upon receiving an
/// authentication message.
bsl::shared_ptr<AuthenticationContext> d_authenticationCtxSp;
bsl::shared_ptr<AuthenticationContext> d_authenticationCtx_sp;

/// The NegotiationContext updated upon receiving a negotiation message.
bsl::shared_ptr<NegotiationContext> d_negotiationCtxSp;
bsl::shared_ptr<NegotiationContext> d_negotiationCtx_sp;

@678098 678098 Jul 24, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

BDE convention to have _sp


/// The callback to invoke to notify of the status of the initial
/// connection.
Expand Down Expand Up @@ -394,7 +394,7 @@ class InitialConnectionContext {
void setAuthenticationContext(
const bsl::shared_ptr<AuthenticationContext>& value);

/// Called by the IO upon `onClose` signal
/// @brief Called by the IO upon `onClose` signal.
void onClose();

/// Read callback invoked when data is available on the channel.
Expand Down
Loading
Loading