From 13b5419f5d8724046f525605830f555eec57c3e8 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Mon, 10 Aug 2026 15:04:18 -0400 Subject: [PATCH 1/7] feat(core-amqp): refresh CBS tokens before they expire The connection cached an access token for each audience and never looked at it again. On a cache hit it returned the stored token with no expiry test, and it stored tokens with emplace, which does not overwrite. There was no refresh timer. So a client that outlived one token lifetime got an unauthorized detach and could not recover without a process restart. The cache now refuses a token that is about to die and authenticates the audience again. A refresh thread also replaces each cached token seven minutes before the token expires, which follows the .NET client. The refresh thread is specific to uAMQP. It releases the token mutex for the credential call and for the CBS operation, because both go to the network. Holding that mutex there would block every caller that opens a link, and it would stop a shutdown from cancelling the work. The thread keeps a weak pointer to the session, so it never extends the life of a session, and it releases that pointer before it takes the mutex again. It drops the cache entry on any failure, so the next link open authenticates again. A result is discarded when a caller stored a newer token while the mutex was free. Shutdown cancels the refresh context before it takes the mutex, so a cancel can stop an operation that is in flight. The connection stops the thread from Close and from the destructor. The expiry rules live in a new header as pure functions, so tests can drive them without a service. They compare in the Azure::DateTime domain, because the cast to a system clock time point throws for a value outside the range of that clock, and a credential can return any value. Refs: #7254 --- sdk/core/azure-core-amqp/CHANGELOG.md | 3 + .../azure-core-amqp/src/amqp/connection.cpp | 341 ++++++++++++++++-- .../src/amqp/private/token_refresh.hpp | 63 ++++ .../src/impl/uamqp/amqp/connection.cpp | 12 + .../uamqp/amqp/private/connection_impl.hpp | 25 ++ .../test/ut/connection_tests.cpp | 90 +++++ .../test/ut/producer_client_test.cpp | 286 +++++++++++++++ 7 files changed, 786 insertions(+), 34 deletions(-) create mode 100644 sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index c6522a625c..124b62e02d 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/sdk/core/azure-core-amqp/CHANGELOG.md @@ -10,6 +10,9 @@ ### Bugs Fixed +- The connection no longer returns a cached CBS token that is at or near its expiry. It authenticates the audience again instead. [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) +- The connection now replaces each cached CBS token before that token expires, so a client that runs for longer than one token lifetime keeps working. This refresh applies to the uAMQP transport. [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) + ### Other Changes ## 1.0.0-beta.12 (2026-05-14) diff --git a/sdk/core/azure-core-amqp/src/amqp/connection.cpp b/sdk/core/azure-core-amqp/src/amqp/connection.cpp index 625af7523b..6a95b170e5 100644 --- a/sdk/core/azure-core-amqp/src/amqp/connection.cpp +++ b/sdk/core/azure-core-amqp/src/amqp/connection.cpp @@ -9,12 +9,15 @@ #include "azure/core/amqp/models/amqp_value.hpp" #include "claims_based_security_impl.hpp" #include "connection_impl.hpp" +#include "private/token_refresh.hpp" #include "session_impl.hpp" #include #include +#include #include +#include using namespace Azure::Core::Diagnostics::_internal; using namespace Azure::Core::Diagnostics; @@ -127,6 +130,48 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal { }}}} // namespace Azure::Core::Amqp::_internal namespace Azure { namespace Core { namespace Amqp { namespace _detail { + + namespace { + // Put a token for one audience on the wire. Both the first authentication + // and the proactive refresh use this function, so the two paths stay the + // same. The caller holds the token mutex. + void PutTokenForAudience( + std::shared_ptr session, + CbsTokenType tokenType, + std::string const& audienceUrl, + std::string const& token, + Azure::DateTime const& expiresOn, + Azure::Core::Context const& context) + { + auto claimsBasedSecurity = std::make_shared(session); + auto cbsOpenStatus = claimsBasedSecurity->Open(context); + if (cbsOpenStatus != CbsOpenResult::Ok) + { + throw std::runtime_error("Could not open Claims Based Security object."); + } + + try + { + auto result + = claimsBasedSecurity->PutToken(tokenType, audienceUrl, token, expiresOn, context); + if (std::get<0>(result) != CbsOperationResult::Ok) + { + throw Azure::Core::Credentials::AuthenticationException( + "Could not authenticate client. Error Status: " + std::to_string(std::get<1>(result)) + + " reason: " + std::get<2>(result)); + } + Log::Stream(Logger::Level::Verbose) << "Close CBS object"; + claimsBasedSecurity->Close(context); + } + catch (...) + { + // Ensure that the claims based security object is closed before we leave this scope. + claimsBasedSecurity->Close(context); + throw; + } + } + } // namespace + bool ConnectionImpl::IsSasCredential() const { if (GetCredential()) @@ -172,14 +217,32 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { std::unique_lock lock(m_tokenMutex); // If we have authenticated this audience, we're done and can return success. + // A cached token is only good while it has enough life left to use. A token + // that is at or near its expiry is discarded here, so the audience is + // authenticated again below. auto token = m_tokenStore.find(audienceUrl); if (token != m_tokenStore.end()) { + if (IsCachedTokenUsable(token->second, std::chrono::system_clock::now())) + { + if (m_options.EnableTrace) + { + Log::Stream(Logger::Level::Verbose) << "Using cached token for " << audienceUrl; + } +#if ENABLE_UAMQP + // Point the refresh thread at a session that is in use now. The + // session that first authenticated this audience can be gone while + // another session still uses the token. + m_tokenSessions[audienceUrl] = session; +#endif + return token->second; + } if (m_options.EnableTrace) { - Log::Stream(Logger::Level::Verbose) << "Using cached token for " << audienceUrl; + Log::Stream(Logger::Level::Verbose) << "Cached token for " << audienceUrl + << " is at or near expiry, authenticating again."; } - return token->second; + m_tokenStore.erase(token); } // We've not authenticated this audience. // Authenticate it with the server @@ -190,55 +253,265 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { << "No cached token for " << audienceUrl << ", Authenticating."; } - auto claimsBasedSecurity = std::make_shared(session); - auto cbsOpenStatus = claimsBasedSecurity->Open(context); - if (cbsOpenStatus != CbsOpenResult::Ok) + Credentials::TokenRequestContext requestContext; + + requestContext.Scopes = m_options.AuthenticationScopes; + auto accessToken{GetCredential()->GetToken(requestContext, context)}; + + PutTokenForAudience( + session, + (IsSasCredential() ? CbsTokenType::Sas : CbsTokenType::Jwt), + audienceUrl, + accessToken.Token, + accessToken.ExpiresOn, + context); + + if (m_options.EnableTrace) { - throw std::runtime_error("Could not open Claims Based Security object."); + Log::Stream(Logger::Level::Verbose) + << "Authenticated connection for audience " << audienceUrl << " successfully."; } - try + // Assign, do not emplace. A refreshed token must replace the token that is + // already in the cache. + m_tokenStore[audienceUrl] = accessToken; +#if ENABLE_UAMQP + // Remember the session that authenticated this audience, so the refresh + // thread can put a new token on the same session. The pointer is weak, so + // the refresh thread never keeps a session alive. + m_tokenSessions[audienceUrl] = session; + StartTokenRefresh(); +#endif + return accessToken; + } + else + { + Log::Stream(Logger::Level::Verbose) << "No credential, returning empty token."; + // If the connection is unauthenticated, then just return an empty access token. + return {}; + } + } + +#if ENABLE_UAMQP + // Start the refresh thread if it is not running yet. The caller holds the + // token mutex. + void ConnectionImpl::StartTokenRefresh() + { + if (m_tokenRefreshStop) + { + return; + } + if (!m_tokenRefreshThread.joinable()) + { + m_tokenRefreshThread = std::thread([this]() { TokenRefreshThread(); }); + } + m_tokenRefreshCv.notify_all(); + } + + void ConnectionImpl::StopTokenRefresh() + { + // Cancel before the lock, not after it. The refresh thread does its network + // work without the token mutex, but it can hold that mutex at other times. + // A cancel that waits for the mutex could not stop an operation that is in + // flight, which is the operation this call must stop. Context::Cancel is + // safe to call from any thread. + m_tokenRefreshContext.Cancel(); + + std::thread threadToJoin; + { + std::unique_lock lock(m_tokenMutex); + m_tokenRefreshStop = true; + m_tokenRefreshCv.notify_all(); + threadToJoin = std::move(m_tokenRefreshThread); + } + + // Join outside the lock, because the refresh thread takes the token mutex. + if (threadToJoin.joinable()) + { + if (threadToJoin.get_id() == std::this_thread::get_id()) + { + // The refresh thread is running this call, which means it released the + // last reference to this connection. It cannot join itself. + threadToJoin.detach(); + } + else { - Credentials::TokenRequestContext requestContext; + threadToJoin.join(); + } + } + } - requestContext.Scopes = m_options.AuthenticationScopes; - auto accessToken{GetCredential()->GetToken(requestContext, context)}; + void ConnectionImpl::TokenRefreshThread() + { + // A background thread must not let an exception escape, because that ends + // the process. Every failure of one refresh is handled inside + // RefreshTokenForAudience, so this catch is for the unexpected. + try + { + std::unique_lock lock(m_tokenMutex); + while (!m_tokenRefreshStop) + { + auto const now = std::chrono::system_clock::now(); - auto result = claimsBasedSecurity->PutToken( - (IsSasCredential() ? CbsTokenType::Sas : CbsTokenType::Jwt), - audienceUrl, - accessToken.Token, - accessToken.ExpiresOn, - context); - if (std::get<0>(result) != CbsOperationResult::Ok) + // Idle wake time when no token is close to its expiry. + auto nextWake = now + IdleTokenRefreshPoll; + std::vector dueAudiences; + for (auto const& entry : m_tokenStore) { - throw Azure::Core::Credentials::AuthenticationException( - "Could not authenticate client. Error Status: " + std::to_string(std::get<1>(result)) - + " reason: " + std::get<2>(result)); + if (IsTokenRefreshDue(entry.second, now)) + { + dueAudiences.push_back(entry.first); + } + else if (IsTokenRefreshDue(entry.second, now + IdleTokenRefreshPoll)) + { + // This token is due inside the idle poll time, so wake for it. The + // expiry is close to now, which makes the cast safe. + auto const expiry + = static_cast(entry.second.ExpiresOn); + nextWake = (std::min)(nextWake, expiry - TokenRefreshBuffer); + } } - Log::Stream(Logger::Level::Verbose) << "Close CBS object"; - claimsBasedSecurity->Close(context); - if (m_options.EnableTrace) + + for (auto const& audience : dueAudiences) { - Log::Stream(Logger::Level::Verbose) - << "Authenticated connection for audience " << audienceUrl << " successfully."; + if (m_tokenRefreshStop) + { + break; + } + RefreshTokenForAudience(audience, lock); + } + + if (!dueAudiences.empty()) + { + // Keep a minimum time between two refresh passes. A token with a + // lifetime shorter than the refresh buffer is due as soon as it + // arrives, and this wait stops the thread from spinning on it. + nextWake = std::chrono::system_clock::now() + MinimumTokenRefreshInterval; } - m_tokenStore.emplace(audienceUrl, accessToken); - return accessToken; + if (m_tokenStore.empty()) + { + // Nothing to refresh. Wait until an audience is authenticated, or + // until shutdown. + m_tokenRefreshCv.wait( + lock, [this]() { return m_tokenRefreshStop || !m_tokenStore.empty(); }); + } + else + { + m_tokenRefreshCv.wait_until(lock, nextWake, [this]() { return m_tokenRefreshStop; }); + } } - catch (...) + } + catch (std::exception const& e) + { + Log::Stream(Logger::Level::Error) + << "The token refresh thread stopped on an error: " << e.what() + << ". Tokens are refreshed on use instead."; + } + catch (...) + { + Log::Stream(Logger::Level::Error) + << "The token refresh thread stopped on an unknown error. Tokens are refreshed on use " + "instead."; + } + } + + // Replace the token for one audience. + // + // The caller holds the token mutex through `lock`. This function releases that + // mutex for the credential call and for the CBS operation, because both go to + // the network. Holding the mutex there would block every caller that opens a + // link, and it would stop a shutdown from cancelling this work. + void ConnectionImpl::RefreshTokenForAudience( + std::string const& audienceUrl, + std::unique_lock& lock) + { + std::shared_ptr session; + auto sessionEntry = m_tokenSessions.find(audienceUrl); + if (sessionEntry != m_tokenSessions.end()) + { + session = sessionEntry->second.lock(); + } + if (!session) + { + // The session that authenticated this audience is gone. Drop the entry, + // and the next link open authenticates the audience again. + m_tokenStore.erase(audienceUrl); + m_tokenSessions.erase(audienceUrl); + return; + } + + // Remember the token that this refresh replaces, so the result does not + // overwrite a newer token that a caller stored while the mutex was free. + auto const previousEntry = m_tokenStore.find(audienceUrl); + if (previousEntry == m_tokenStore.end()) + { + return; + } + std::string const previousToken{previousEntry->second.Token}; + + auto const tokenType = (IsSasCredential() ? CbsTokenType::Sas : CbsTokenType::Jwt); + auto const credential = GetCredential(); + auto const scopes = m_options.AuthenticationScopes; + auto const traceEnabled = m_options.EnableTrace; + auto const parentContext = m_tokenRefreshContext; + + Credentials::AccessToken accessToken; + bool refreshed{false}; + std::string failureMessage; + + lock.unlock(); + try + { + auto context = parentContext.WithDeadline( + std::chrono::system_clock::now() + TokenRefreshOperationTimeout); + + Credentials::TokenRequestContext requestContext; + requestContext.Scopes = scopes; + accessToken = credential->GetToken(requestContext, context); + + PutTokenForAudience( + session, tokenType, audienceUrl, accessToken.Token, accessToken.ExpiresOn, context); + refreshed = true; + } + catch (std::exception const& e) + { + failureMessage = e.what(); + } + catch (...) + { + failureMessage = "unknown error"; + } + // Release the session before the mutex is taken again. This reference can be + // the last one, and a session destructor takes other locks. + session.reset(); + lock.lock(); + + auto currentEntry = m_tokenStore.find(audienceUrl); + if (currentEntry == m_tokenStore.end() || currentEntry->second.Token != previousToken) + { + // A caller replaced or dropped this token while the mutex was free. That + // token is newer than this one, so keep it. + return; + } + + if (refreshed) + { + currentEntry->second = accessToken; + if (traceEnabled) { - // Ensure that the claims based security object is closed before we leave this scope. - claimsBasedSecurity->Close(context); - throw; + Log::Stream(Logger::Level::Verbose) + << "Refreshed the token for audience " << audienceUrl << " before its expiry."; } } else { - Log::Stream(Logger::Level::Verbose) << "No credential, returning empty token."; - // If the connection is unauthenticated, then just return an empty access token. - return {}; + // Drop the cache entry, so the next link open authenticates again. + Log::Stream(Logger::Level::Warning) + << "Could not refresh the token for audience " << audienceUrl << ": " << failureMessage; + m_tokenStore.erase(currentEntry); + m_tokenSessions.erase(audienceUrl); } } +#endif // ENABLE_UAMQP }}}} // namespace Azure::Core::Amqp::_detail diff --git a/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp b/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp new file mode 100644 index 0000000000..3ba85857b9 --- /dev/null +++ b/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp @@ -0,0 +1,63 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#pragma once + +#include +#include + +#include + +namespace Azure { namespace Core { namespace Amqp { namespace _detail { + + // The .NET client refreshes a CBS token seven minutes before the token + // expires. See AmqpConnectionScope.cs in Azure.Messaging.EventHubs. Use the + // same buffer here. + constexpr std::chrono::minutes TokenRefreshBuffer{7}; + + // The smallest time between two refresh passes. A service that issues tokens + // with a lifetime shorter than the buffer makes every token due immediately. + // This interval stops the refresh thread from spinning in that case. + constexpr std::chrono::seconds MinimumTokenRefreshInterval{20}; + + // The refresh thread does the early refresh. The cache only has to stop + // itself from giving a caller a token that is about to die, so its margin is + // small. A larger margin here would make the cache authenticate again on + // every call for a token with a short lifetime. + constexpr std::chrono::seconds MinimumTokenLifetimeToUse{30}; + + // How long the refresh thread sleeps when no token is close to its expiry. + constexpr std::chrono::minutes IdleTokenRefreshPoll{1}; + + // The deadline for one refresh, which covers the credential call and the CBS + // operation. + constexpr std::chrono::seconds TokenRefreshOperationTimeout{60}; + + // These functions compare in the Azure::DateTime domain on purpose. The cast + // from Azure::DateTime to std::chrono::system_clock::time_point is explicit + // and it throws when the value is outside the range of the system clock. A + // credential can return any value in ExpiresOn, and a default constructed + // ExpiresOn is year 1, so a cast here could throw on a caller thread or on + // the refresh thread. The conversion in the other direction does not throw. + + // Return true while the cached token has enough life left to give to a + // caller. + inline bool IsCachedTokenUsable( + Azure::Core::Credentials::AccessToken const& token, + std::chrono::system_clock::time_point now) + { + return token.ExpiresOn > Azure::DateTime(now + MinimumTokenLifetimeToUse); + } + + // Return true when the refresh thread must replace this token by the given + // time. The refresh is due one buffer before the token expires, so the test + // adds the buffer to the time instead of subtracting it from the expiry. + // Adding avoids an underflow for a token that reports a very early expiry. + inline bool IsTokenRefreshDue( + Azure::Core::Credentials::AccessToken const& token, + std::chrono::system_clock::time_point now) + { + return token.ExpiresOn <= Azure::DateTime(now + TokenRefreshBuffer); + } + +}}}} // namespace Azure::Core::Amqp::_detail diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp index 08fe657e1a..15a932562b 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp @@ -148,6 +148,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { ConnectionImpl::~ConnectionImpl() { + // Stop the refresh thread first. It uses this connection, so it must not be + // running while the connection is torn down. + StopTokenRefresh(); + std::unique_lock lock(m_amqpMutex); if (m_openCount.load() != 0) { @@ -435,6 +439,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { throw std::logic_error("Connection not opened."); } + // Stop the refresh thread before polling stops, because a CBS operation + // needs the connection to poll. + StopTokenRefresh(); + // Stop polling on this connection, we're shutting it down. EnableAsyncOperation(false); @@ -463,6 +471,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { throw std::logic_error("Connection not opened."); } + // Stop the refresh thread before polling stops, because a CBS operation + // needs the connection to poll. + StopTokenRefresh(); + // Stop polling on this connection, we're shutting it down. EnableAsyncOperation(false); diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp index 26be189e63..3a5d51da66 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp @@ -8,14 +8,19 @@ #include "azure/core/amqp/internal/connection.hpp" #include "azure/core/amqp/internal/network/transport.hpp" +#include #include #include #include #include +#include +#include #include +#include #include +#include #if defined(_MSC_VER) #define _azure_ACQUIRES_LOCK(...) _Acquires_exclusive_lock_(__VA_ARGS__) @@ -132,6 +137,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { std::string const& audience, Azure::Core::Context const& context); + // Stop the token refresh thread. This is safe to call more than once, and + // the connection calls it from Close and from the destructor. + void StopTokenRefresh(); + using LockType = std::recursive_mutex; _azure_ACQUIRES_LOCK(m_amqpMutex) std::unique_lock Lock() @@ -164,6 +173,22 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { std::shared_ptr m_credential{}; std::map m_tokenStore; + // The session that authenticated each audience. The pointer is weak, so the + // refresh thread never keeps a session alive. + std::map> m_tokenSessions; + // The thread that replaces each cached token before the token expires. + std::thread m_tokenRefreshThread; + std::condition_variable m_tokenRefreshCv; + bool m_tokenRefreshStop{false}; + // Cancelled on shutdown, to stop a CBS operation that is in flight. + Azure::Core::Context m_tokenRefreshContext; + + void StartTokenRefresh(); + void TokenRefreshThread(); + void RefreshTokenForAudience( + std::string const& audienceUrl, + std::unique_lock& lock); + ConnectionImpl( _internal::ConnectionEvents* eventHandler, _internal::ConnectionOptions const& options); diff --git a/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp b/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp index 33d08d51cb..0ce236ef14 100644 --- a/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp +++ b/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +#include "../../src/amqp/private/token_refresh.hpp" #include "azure/core/amqp/internal/common/async_operation_queue.hpp" #include "azure/core/amqp/internal/connection.hpp" #include "azure/core/amqp/internal/message_receiver.hpp" @@ -17,6 +18,7 @@ #include #include +#include #include #include @@ -31,6 +33,94 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { void TearDown() override {} }; + // Tests for the rules that decide when a cached CBS token is still good, and + // when the connection must replace it. These rules are pure functions, so they + // run on every platform and need no service. + class TestTokenRefresh : public testing::Test { + protected: + static Azure::Core::Credentials::AccessToken TokenExpiringIn(std::chrono::seconds lifetime) + { + Azure::Core::Credentials::AccessToken token; + token.Token = "TestToken"; + token.ExpiresOn = std::chrono::system_clock::now() + lifetime; + return token; + } + }; + + TEST_F(TestTokenRefresh, CachedTokenIsUsableWhileItHasLifeLeft) + { + auto const now = std::chrono::system_clock::now(); + EXPECT_TRUE( + Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::hours(1)), now)); + EXPECT_TRUE( + Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::minutes(2)), now)); + } + + TEST_F(TestTokenRefresh, CachedTokenIsNotUsableNearOrAfterExpiry) + { + auto const now = std::chrono::system_clock::now(); + // Inside the minimum lifetime that a caller may use. + EXPECT_FALSE( + Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::seconds(10)), now)); + // Already expired. + EXPECT_FALSE( + Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::seconds(-30)), now)); + } + + TEST_F(TestTokenRefresh, RefreshIsDueOneBufferBeforeExpiry) + { + auto const now = std::chrono::system_clock::now(); + // A normal token has a long life, so no refresh is due yet. + EXPECT_FALSE( + Azure::Core::Amqp::_detail::IsTokenRefreshDue( + TokenExpiringIn(std::chrono::minutes(90)), now)); + // Inside the buffer, so the refresh thread must replace the token. + EXPECT_TRUE( + Azure::Core::Amqp::_detail::IsTokenRefreshDue( + TokenExpiringIn(std::chrono::minutes(6)), now)); + } + + // A credential can put any value in ExpiresOn, and the cast from + // Azure::DateTime to a system clock time point throws outside the range of + // that clock. A default constructed ExpiresOn is year 1. These rules run on + // the refresh thread, where an exception would end the process, so they must + // not throw for any value. + TEST_F(TestTokenRefresh, ExtremeExpiryValuesDoNotThrow) + { + auto const now = std::chrono::system_clock::now(); + + // A default constructed token reports year 1. + Azure::Core::Credentials::AccessToken defaultToken; + EXPECT_NO_THROW({ + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsCachedTokenUsable(defaultToken, now)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsTokenRefreshDue(defaultToken, now)); + }); + + // A token that reports a year past the range of the system clock. + Azure::Core::Credentials::AccessToken farFutureToken; + farFutureToken.Token = "TestToken"; + farFutureToken.ExpiresOn = Azure::DateTime(9999, 12, 31); + EXPECT_NO_THROW({ + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable(farFutureToken, now)); + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsTokenRefreshDue(farFutureToken, now)); + }); + } + + // A token with a lifetime shorter than the buffer is due as soon as it + // arrives. The connection must still hand it to a caller, because refusing it + // would make every call authenticate again. + TEST_F(TestTokenRefresh, ShortLivedTokenIsDueButStillUsable) + { + auto const now = std::chrono::system_clock::now(); + auto const token = TokenExpiringIn(std::chrono::seconds(80)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsTokenRefreshDue(token, now)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable(token, now)); + } + #if !defined(AZ_PLATFORM_MAC) TEST_F(TestConnections, SimpleConnection) { diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp index 24a73d24f1..df8dab65de 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp @@ -5,12 +5,17 @@ #include "eventhubs_test_base.hpp" +#include #include +#include #include #include #include +#include +#include #include +#include #include @@ -356,6 +361,287 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test { } } +#if ENABLE_UAMQP + namespace { + // Counts the tokens that the test asked for, and can report a shorter life + // for each token than the real life. + // + // A real Microsoft Entra token lives for 60 to 90 minutes, so a test that + // waits for a true expiry would run for hours. A shorter reported life makes + // the connection refresh the token in seconds. The token itself is a real + // token and stays valid, so the service still accepts it. + class CountingTokenCredential final : public Azure::Core::Credentials::TokenCredential { + public: + CountingTokenCredential( + std::shared_ptr innerCredential, + Azure::Nullable reportedLifetime = {}) + // Report the name of the credential that this one wraps. The + // connection asks for that name to decide whether it puts a shared + // access signature token or a JWT, so a new name here would break a + // connection string credential. + : Azure::Core::Credentials::TokenCredential(innerCredential->GetCredentialName()), + m_innerCredential(std::move(innerCredential)), m_reportedLifetime(reportedLifetime) + { + } + + int GetTokenCount() const { return m_tokenCount.load(); } + + private: + std::shared_ptr m_innerCredential; + Azure::Nullable m_reportedLifetime; + mutable std::atomic m_tokenCount{0}; + + Azure::Core::Credentials::AccessToken GetToken( + Azure::Core::Credentials::TokenRequestContext const& requestContext, + Azure::Core::Context const& context) const override + { + auto token = m_innerCredential->GetToken(requestContext, context); + if (m_reportedLifetime.HasValue()) + { + token.ExpiresOn = std::chrono::system_clock::now() + m_reportedLifetime.Value(); + } + m_tokenCount++; + return token; + } + }; + + // The connection replaces a token one buffer, seven minutes, before the + // token expires. A token that reports seven minutes and thirty seconds of + // life is due in about thirty seconds. That gap is long enough to see the + // refresh and then to check that the refreshed token is the one in use. + constexpr std::chrono::seconds SlowRefreshLifetime{450}; + + // A token that reports less life than the buffer is due as soon as it + // arrives, so the refresh thread replaces it every twenty seconds, which is + // the smallest interval between two refresh passes. This drives the tests + // that need several refreshes in a short run. + constexpr std::chrono::seconds FastRefreshLifetime{80}; + + Azure::Messaging::EventHubs::Models::EventData MakeTestEvent(std::uint8_t tag) + { + Azure::Messaging::EventHubs::Models::EventData event; + event.Body = {'T', tag}; + return event; + } + + // Return the credential that the refresh tests wrap and count. + // + // When the environment gives a connection string, build a shared access + // signature credential from it. That path needs no Microsoft Entra sign in, + // so a developer can run these tests from a workstation. Otherwise use the + // credential of the test harness, which is what the pipeline uses. + std::shared_ptr MakeInnerCredential( + std::shared_ptr harnessCredential) + { + auto const connectionString + = Azure::Core::_internal::Environment::GetVariable("EVENTHUBS_CONNECTION_STRING"); + if (!connectionString.empty()) + { + return std::make_shared< + Azure::Core::Amqp::_internal::ServiceBusSasConnectionStringCredential>( + connectionString); + } + return harnessCredential; + } + + // Wait until the credential issued at least `count` tokens, or until the + // timeout. Returns the count that was reached. + int WaitForTokenCount( + std::shared_ptr const& credential, + int count, + std::chrono::seconds timeout) + { + auto const deadline = std::chrono::steady_clock::now() + timeout; + while (credential->GetTokenCount() < count && std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + } + return credential->GetTokenCount(); + } + } // namespace + + // L1. The connection must replace the CBS token before that token expires, + // and the replacement must reach the service. + TEST_P(ProducerClientTest, TokenRefreshBeforeExpiry_LIVEONLY_) + { + auto credential = std::make_shared( + MakeInnerCredential(GetTestCredential()), SlowRefreshLifetime); + + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + ASSERT_NO_THROW(client.Send(MakeTestEvent('1'))); + auto const countAfterFirstSend = credential->GetTokenCount(); + ASSERT_GE(countAfterFirstSend, 1); + + // The test makes no client call here, so only the refresh thread can raise + // the count. + auto const countAfterRefresh + = WaitForTokenCount(credential, countAfterFirstSend + 1, std::chrono::seconds(120)); + ASSERT_GT(countAfterRefresh, countAfterFirstSend) + << "The connection did not replace the CBS token before the token expired."; + + // Make sure the replacement reached the service. A refresh that failed makes + // the connection drop the cached token, and then this send has to ask the + // credential for another token. So a send that does not raise the count + // proves that the refreshed token is in the cache and that the service + // accepted it. + ASSERT_NO_THROW(client.Send(MakeTestEvent('2'))); + EXPECT_EQ(credential->GetTokenCount(), countAfterRefresh) + << "The send asked for a new token, so the refresh did not reach the service."; + client.Close(); + } + + // L2. The client must keep sending past two token lifetimes, while the + // refresh thread replaces the token repeatedly. This is the continuity gate + // from issue #7254, on a compressed clock. L1 is the test that proves a + // refresh reaches the service; this one proves that a long run keeps working. + TEST_P(ProducerClientTest, TokenRefreshContinuityPastTwoLifetimes_LIVEONLY_) + { + auto credential = std::make_shared( + MakeInnerCredential(GetTestCredential()), FastRefreshLifetime); + + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + auto const runFor = FastRefreshLifetime * 2 + std::chrono::seconds(10); + auto const start = std::chrono::steady_clock::now(); + int sendCount = 0; + while (std::chrono::steady_clock::now() - start < runFor) + { + ASSERT_NO_THROW(client.Send(MakeTestEvent('3'))) + << "A send failed after " << sendCount << " sends."; + sendCount++; + std::this_thread::sleep_for(std::chrono::seconds(20)); + } + + EXPECT_GT(sendCount, 0); + EXPECT_GE(credential->GetTokenCount(), 3) + << "The connection did not refresh the token while the client was running."; + client.Close(); + } + + // L3. Shutting the client down must stop the refresh thread, and must not + // hang. + TEST_P(ProducerClientTest, TokenRefreshStopsOnClose_LIVEONLY_) + { + auto credential = std::make_shared( + MakeInnerCredential(GetTestCredential()), FastRefreshLifetime); + + { + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + ASSERT_NO_THROW(client.Send(MakeTestEvent('4'))); + EXPECT_GE(WaitForTokenCount(credential, 2, std::chrono::seconds(90)), 2); + + auto const closeStart = std::chrono::steady_clock::now(); + client.Close(); + EXPECT_LT(std::chrono::steady_clock::now() - closeStart, std::chrono::seconds(70)) + << "Close did not return promptly while the refresh thread was running."; + } + + // The client is gone, so the refresh thread is stopped. The count must not + // move after this point. + auto const countAfterShutdown = credential->GetTokenCount(); + std::this_thread::sleep_for(std::chrono::seconds(30)); + EXPECT_EQ(credential->GetTokenCount(), countAfterShutdown) + << "The refresh thread kept running after the client was destroyed."; + } + + // L4. A token with a normal life must be fetched once. This is the control + // case. It fails if the new expiry test makes the cache authenticate again on + // every call. + TEST_P(ProducerClientTest, NormalTokenIsNotRefetched_LIVEONLY_) + { + // No reported lifetime, so the real expiry of the token applies. + auto credential + = std::make_shared(MakeInnerCredential(GetTestCredential())); + + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + ASSERT_NO_THROW(client.Send(MakeTestEvent('5'))); + auto const countAfterFirstSend = credential->GetTokenCount(); + + std::this_thread::sleep_for(std::chrono::seconds(10)); + ASSERT_NO_THROW(client.Send(MakeTestEvent('6'))); + + EXPECT_EQ(credential->GetTokenCount(), countAfterFirstSend) + << "The connection asked for a new token for an audience that already had a good one."; + client.Close(); + } + + // L5. Calls from many threads must keep working while the refresh thread + // replaces the token. The refresh thread and the calling threads take the same + // lock, so this is the test for a deadlock. + TEST_P(ProducerClientTest, TokenRefreshUnderConcurrentCalls_LIVEONLY_) + { + auto credential = std::make_shared( + MakeInnerCredential(GetTestCredential()), FastRefreshLifetime); + + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + std::string const eventHubName{GetEventHubName()}; + std::atomic failed{false}; + std::vector threads; + for (int i = 0; i < 8; i++) + { + threads.emplace_back([&client, &eventHubName, &failed]() { + auto const start = std::chrono::steady_clock::now(); + while (std::chrono::steady_clock::now() - start < std::chrono::seconds(60)) + { + try + { + auto result = client.GetEventHubProperties(); + if (result.Name != eventHubName) + { + failed = true; + } + } + catch (std::exception const&) + { + failed = true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + } + }); + } + for (auto& t : threads) + { + if (t.joinable()) + { + t.join(); + } + } + + EXPECT_FALSE(failed.load()) << "A call failed while the refresh thread replaced the token."; + EXPECT_GE(credential->GetTokenCount(), 2) << "The refresh thread did not run during the test."; + client.Close(); + } +#endif // ENABLE_UAMQP + namespace { static std::string GetSuffix(const testing::TestParamInfo& info) { From b5d63c90b834edaa9a58e8e99407b7bbbad4cb4b Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Mon, 10 Aug 2026 16:50:51 -0400 Subject: [PATCH 2/7] fix(core-amqp): serialize the CBS operation on a connection The refresh thread releases the token mutex for its CBS operation, so a caller and the refresh thread could run two claims based security objects on one connection at the same time. uAMQP names the CBS links after the node, so both objects attach a link called "$cbs-sender" and one called "$cbs-receiver". AMQP 1.0 section 2.6.1 requires a link name to be unique for one direction between two containers, so the broker detaches one of them. A caller that authenticates in that window can get an authentication error for a token that is good. Add a mutex that covers the CBS operation alone, and take it on both paths. A caller takes the token mutex and then this mutex. The refresh thread takes this mutex only while it does not hold the token mutex, so the lock order stays acyclic and a shutdown still cancels the operation. Also wait for the refresh to reach the service in the live test before it checks the token count, because the count rises when the refresh asks for the token, which is before the put. Refs: #7254 --- .../azure-core-amqp/src/amqp/connection.cpp | 43 +++++++++++++++---- .../uamqp/amqp/private/connection_impl.hpp | 12 ++++++ .../test/ut/producer_client_test.cpp | 10 +++++ 3 files changed, 56 insertions(+), 9 deletions(-) diff --git a/sdk/core/azure-core-amqp/src/amqp/connection.cpp b/sdk/core/azure-core-amqp/src/amqp/connection.cpp index 6a95b170e5..ce41cd7e55 100644 --- a/sdk/core/azure-core-amqp/src/amqp/connection.cpp +++ b/sdk/core/azure-core-amqp/src/amqp/connection.cpp @@ -258,13 +258,20 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { requestContext.Scopes = m_options.AuthenticationScopes; auto accessToken{GetCredential()->GetToken(requestContext, context)}; - PutTokenForAudience( - session, - (IsSasCredential() ? CbsTokenType::Sas : CbsTokenType::Jwt), - audienceUrl, - accessToken.Token, - accessToken.ExpiresOn, - context); + { +#if ENABLE_UAMQP + // Only one claims based security object may exist on this connection at + // a time. See m_cbsMutex. + std::lock_guard cbsLock(m_cbsMutex); +#endif + PutTokenForAudience( + session, + (IsSasCredential() ? CbsTokenType::Sas : CbsTokenType::Jwt), + audienceUrl, + accessToken.Token, + accessToken.ExpiresOn, + context); + } if (m_options.EnableTrace) { @@ -295,6 +302,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { #if ENABLE_UAMQP // Start the refresh thread if it is not running yet. The caller holds the // token mutex. + // + // A thread that stopped on an error stays joinable, so this function does not + // start a second one. That is deliberate. The connection then refreshes each + // token when a caller uses it, which is the behavior the error log describes. void ConnectionImpl::StartTokenRefresh() { if (m_tokenRefreshStop) @@ -332,6 +343,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { { // The refresh thread is running this call, which means it released the // last reference to this connection. It cannot join itself. + // + // This path only happens when the connection is destroyed while it is + // still open, and the destructor stops the process on that condition a + // moment after this call returns. So the detached thread does not + // outlive the connection today. If those asserts ever go away, this + // thread comes back to a destroyed mutex, so give it a way to stop + // before you relax them. threadToJoin.detach(); } else @@ -470,8 +488,15 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { requestContext.Scopes = scopes; accessToken = credential->GetToken(requestContext, context); - PutTokenForAudience( - session, tokenType, audienceUrl, accessToken.Token, accessToken.ExpiresOn, context); + { + // Only one claims based security object may exist on this connection at + // a time, so this waits for a caller that authenticates right now. See + // m_cbsMutex. This mutex is released before the token mutex is taken + // again, which keeps the lock order acyclic. + std::lock_guard cbsLock(m_cbsMutex); + PutTokenForAudience( + session, tokenType, audienceUrl, accessToken.Token, accessToken.ExpiresOn, context); + } refreshed = true; } catch (std::exception const& e) diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp index 3a5d51da66..a2aa77a182 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp @@ -173,6 +173,18 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { std::shared_ptr m_credential{}; std::map m_tokenStore; + // Serializes the CBS operation itself. uAMQP names the CBS links after the + // node, so every claims based security object on this connection attaches a + // link called "$cbs-sender" and one called "$cbs-receiver". AMQP 1.0 section + // 2.6.1 requires a link name to be unique for one direction between two + // containers, so only one of these objects may exist at a time. The refresh + // thread does its work without the token mutex, so this mutex is what keeps + // the refresh and a caller apart. + // + // Lock order: a caller takes m_tokenMutex and then this mutex. The refresh + // thread takes this mutex only while it does not hold m_tokenMutex. + std::mutex m_cbsMutex; + // The session that authenticated each audience. The pointer is weak, so the // refresh thread never keeps a session alive. std::map> m_tokenSessions; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp index df8dab65de..dac56b52b3 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp @@ -485,6 +485,16 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test { ASSERT_GT(countAfterRefresh, countAfterFirstSend) << "The connection did not replace the CBS token before the token expired."; + // The count rises when the refresh asks for the token, which is before the + // refresh puts that token on the wire. Wait for the put to finish. A put + // that fails drops the cached token, and the send below then has to ask for + // a new one, which fails this test. A put that takes longer than this wait + // to fail stays invisible to the test. The next refresh is about thirty + // seconds out, so this wait cannot let another refresh raise the count. + std::this_thread::sleep_for(std::chrono::seconds(5)); + ASSERT_EQ(credential->GetTokenCount(), countAfterRefresh) + << "A second refresh ran during the wait, so this test cannot judge the first one."; + // Make sure the replacement reached the service. A refresh that failed makes // the connection drop the cached token, and then this send has to ask the // credential for another token. So a send that does not raise the count From 21a89dcc8002d5d55b120a1c5a6eec3dd7f264ed Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Mon, 10 Aug 2026 17:51:37 -0400 Subject: [PATCH 3/7] style(core-amqp): format the connection tests with clang-format 11 The CI task "Validate Clang Format" runs clang-format-11 across the sdk folder and fails when a file differs. The new tests were formatted with a newer clang-format, which wraps some calls differently, so the task failed on three pipelines. Format the file with clang-format 11. No code changes. --- .../test/ut/connection_tests.cpp | 30 ++++++++----------- 1 file changed, 12 insertions(+), 18 deletions(-) diff --git a/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp b/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp index 0ce236ef14..3a20734016 100644 --- a/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp +++ b/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp @@ -50,38 +50,32 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { TEST_F(TestTokenRefresh, CachedTokenIsUsableWhileItHasLifeLeft) { auto const now = std::chrono::system_clock::now(); - EXPECT_TRUE( - Azure::Core::Amqp::_detail::IsCachedTokenUsable( - TokenExpiringIn(std::chrono::hours(1)), now)); - EXPECT_TRUE( - Azure::Core::Amqp::_detail::IsCachedTokenUsable( - TokenExpiringIn(std::chrono::minutes(2)), now)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::hours(1)), now)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::minutes(2)), now)); } TEST_F(TestTokenRefresh, CachedTokenIsNotUsableNearOrAfterExpiry) { auto const now = std::chrono::system_clock::now(); // Inside the minimum lifetime that a caller may use. - EXPECT_FALSE( - Azure::Core::Amqp::_detail::IsCachedTokenUsable( - TokenExpiringIn(std::chrono::seconds(10)), now)); + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::seconds(10)), now)); // Already expired. - EXPECT_FALSE( - Azure::Core::Amqp::_detail::IsCachedTokenUsable( - TokenExpiringIn(std::chrono::seconds(-30)), now)); + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::seconds(-30)), now)); } TEST_F(TestTokenRefresh, RefreshIsDueOneBufferBeforeExpiry) { auto const now = std::chrono::system_clock::now(); // A normal token has a long life, so no refresh is due yet. - EXPECT_FALSE( - Azure::Core::Amqp::_detail::IsTokenRefreshDue( - TokenExpiringIn(std::chrono::minutes(90)), now)); + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsTokenRefreshDue( + TokenExpiringIn(std::chrono::minutes(90)), now)); // Inside the buffer, so the refresh thread must replace the token. - EXPECT_TRUE( - Azure::Core::Amqp::_detail::IsTokenRefreshDue( - TokenExpiringIn(std::chrono::minutes(6)), now)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsTokenRefreshDue( + TokenExpiringIn(std::chrono::minutes(6)), now)); } // A credential can put any value in ExpiresOn, and the cast from From d4133b3123b496ec7577726a47e5e6c01239b8e7 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 12 Aug 2026 14:45:42 -0400 Subject: [PATCH 4/7] docs(core-amqp): note the non-transient unauthorized-access classification Event Hubs classifies amqp:unauthorized-access as not transient, so a send that meets that condition stops at the first attempt. Say so in the changelog entry, because it shows why the refresh matters. --- sdk/core/azure-core-amqp/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index 124b62e02d..c2cfe52955 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/sdk/core/azure-core-amqp/CHANGELOG.md @@ -11,7 +11,7 @@ ### Bugs Fixed - The connection no longer returns a cached CBS token that is at or near its expiry. It authenticates the audience again instead. [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) -- The connection now replaces each cached CBS token before that token expires, so a client that runs for longer than one token lifetime keeps working. This refresh applies to the uAMQP transport. [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) +- The connection now replaces each cached CBS token before that token expires, so a client that runs for longer than one token lifetime keeps working. This refresh applies to the uAMQP transport. Without this refresh, a send that gets the `amqp:unauthorized-access` condition stops at the first attempt, because the Event Hubs producer treats that condition as not transient. [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) ### Other Changes From bc58d769b65d99aee8af15bf90e0cbfc89607809 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 12 Aug 2026 16:19:31 -0400 Subject: [PATCH 5/7] fix(core-amqp): let the CBS refresh thread outlive the connection The token refresh thread promotes a weak session pointer for the length of one refresh. A session holds its connection, so that promoted pointer can be the last reference to both. When the thread releases it, the session destructor and then the connection destructor run on the refresh thread. That gave two failures. First, the connection destructor calls StopTokenRefresh, which sees that the thread is joining itself and detaches it. The destructor then destroyed the token mutex, the condition variable, the stop flag, and the token maps, because all four were connection members. The detached thread came back to a destroyed mutex on the next line. The destructor asserts do not stop this, because Event Hubs on uAMQP never calls Connection::Open or Connection::Close. Second, the refresh takes an early return when the token for the audience is gone from the cache. On that path the session pointer died under the token mutex, so the connection destructor took a mutex that this thread already held. That mutex is not recursive, so the thread locked itself. The mutex, the condition variable, the stop flag, and the two token maps now live in a TokenRefreshState block that the connection and the thread both own. The thread body is a static function that takes the connection as a raw pointer, so the thread never keeps the connection alive. When a release can have destroyed the connection, RefreshTokenForAudience returns true, and the thread leaves without touching the connection again. The state block stays alive until the thread returns. The promoted session goes in a ReleaseOutsideLock hold. The hold gives up the token mutex for the release and takes it again after, on every path out of the function, and also on the early return and on an exception. So the connection destructor never runs on this thread while this thread holds the token mutex. --- .../azure-core-amqp/src/amqp/connection.cpp | 154 ++++++++++++------ .../src/amqp/private/token_refresh.hpp | 86 ++++++++++ .../amqp/private/connection_impl.hpp | 8 +- .../uamqp/amqp/private/connection_impl.hpp | 35 ++-- 4 files changed, 217 insertions(+), 66 deletions(-) diff --git a/sdk/core/azure-core-amqp/src/amqp/connection.cpp b/sdk/core/azure-core-amqp/src/amqp/connection.cpp index ce41cd7e55..3fd8633197 100644 --- a/sdk/core/azure-core-amqp/src/amqp/connection.cpp +++ b/sdk/core/azure-core-amqp/src/amqp/connection.cpp @@ -215,13 +215,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { } } - std::unique_lock lock(m_tokenMutex); + std::unique_lock lock(m_tokenState->Mutex); // If we have authenticated this audience, we're done and can return success. // A cached token is only good while it has enough life left to use. A token // that is at or near its expiry is discarded here, so the audience is // authenticated again below. - auto token = m_tokenStore.find(audienceUrl); - if (token != m_tokenStore.end()) + auto token = m_tokenState->TokenStore.find(audienceUrl); + if (token != m_tokenState->TokenStore.end()) { if (IsCachedTokenUsable(token->second, std::chrono::system_clock::now())) { @@ -233,7 +233,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // Point the refresh thread at a session that is in use now. The // session that first authenticated this audience can be gone while // another session still uses the token. - m_tokenSessions[audienceUrl] = session; + m_tokenState->TokenSessions[audienceUrl] = session; #endif return token->second; } @@ -242,7 +242,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { Log::Stream(Logger::Level::Verbose) << "Cached token for " << audienceUrl << " is at or near expiry, authenticating again."; } - m_tokenStore.erase(token); + m_tokenState->TokenStore.erase(token); } // We've not authenticated this audience. // Authenticate it with the server @@ -281,12 +281,12 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // Assign, do not emplace. A refreshed token must replace the token that is // already in the cache. - m_tokenStore[audienceUrl] = accessToken; + m_tokenState->TokenStore[audienceUrl] = accessToken; #if ENABLE_UAMQP // Remember the session that authenticated this audience, so the refresh // thread can put a new token on the same session. The pointer is weak, so // the refresh thread never keeps a session alive. - m_tokenSessions[audienceUrl] = session; + m_tokenState->TokenSessions[audienceUrl] = session; StartTokenRefresh(); #endif return accessToken; @@ -308,15 +308,19 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // token when a caller uses it, which is the behavior the error log describes. void ConnectionImpl::StartTokenRefresh() { - if (m_tokenRefreshStop) + if (m_tokenState->Stop) { return; } if (!m_tokenRefreshThread.joinable()) { - m_tokenRefreshThread = std::thread([this]() { TokenRefreshThread(); }); + // The thread co-owns the shared state, so the state outlives this + // connection. The raw `this` pointer is good only while the state says + // the connection is alive. See TokenRefreshState. + m_tokenRefreshThread + = std::thread([this, state = m_tokenState]() { TokenRefreshThread(this, state); }); } - m_tokenRefreshCv.notify_all(); + m_tokenState->Cv.notify_all(); } void ConnectionImpl::StopTokenRefresh() @@ -330,9 +334,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { std::thread threadToJoin; { - std::unique_lock lock(m_tokenMutex); - m_tokenRefreshStop = true; - m_tokenRefreshCv.notify_all(); + std::unique_lock lock(m_tokenState->Mutex); + m_tokenState->Stop = true; + m_tokenState->Cv.notify_all(); threadToJoin = std::move(m_tokenRefreshThread); } @@ -342,14 +346,15 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { if (threadToJoin.get_id() == std::this_thread::get_id()) { // The refresh thread is running this call, which means it released the - // last reference to this connection. It cannot join itself. + // last reference to this connection. It cannot join itself, so detach + // it. // - // This path only happens when the connection is destroyed while it is - // still open, and the destructor stops the process on that condition a - // moment after this call returns. So the detached thread does not - // outlive the connection today. If those asserts ever go away, this - // thread comes back to a destroyed mutex, so give it a way to stop - // before you relax them. + // The detached thread outlives this connection, and that is safe. It + // holds a shared_ptr to the state block, so the mutex, the condition + // variable, the stop flag, and the token maps all stay alive. Stop is + // set above, so the thread comes back from the release, takes the mutex + // in the state block, sees Stop, and leaves without touching this + // connection. threadToJoin.detach(); } else @@ -359,22 +364,32 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { } } - void ConnectionImpl::TokenRefreshThread() + // The body of the refresh thread. + // + // This is a static function, and `connection` is a raw pointer on purpose. + // The thread must not own the connection, because a thread that owns the + // object it refreshes keeps that object alive forever. The `state` block is + // shared instead. The thread uses `connection` only while it knows the + // connection is alive, which means until RefreshTokenForAudience reports that + // the connection can be gone. + void ConnectionImpl::TokenRefreshThread( + ConnectionImpl* connection, + std::shared_ptr state) { // A background thread must not let an exception escape, because that ends // the process. Every failure of one refresh is handled inside // RefreshTokenForAudience, so this catch is for the unexpected. try { - std::unique_lock lock(m_tokenMutex); - while (!m_tokenRefreshStop) + std::unique_lock lock(state->Mutex); + while (!state->Stop) { auto const now = std::chrono::system_clock::now(); // Idle wake time when no token is close to its expiry. auto nextWake = now + IdleTokenRefreshPoll; std::vector dueAudiences; - for (auto const& entry : m_tokenStore) + for (auto const& entry : state->TokenStore) { if (IsTokenRefreshDue(entry.second, now)) { @@ -392,11 +407,17 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { for (auto const& audience : dueAudiences) { - if (m_tokenRefreshStop) + if (state->Stop) { break; } - RefreshTokenForAudience(audience, lock); + if (connection->RefreshTokenForAudience(*state, audience, lock)) + { + // That call released a session reference, which can have destroyed + // this connection. `connection` is not safe to use now, so leave. + // The state block stays alive until this thread returns. + return; + } } if (!dueAudiences.empty()) @@ -407,16 +428,15 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { nextWake = std::chrono::system_clock::now() + MinimumTokenRefreshInterval; } - if (m_tokenStore.empty()) + if (state->TokenStore.empty()) { // Nothing to refresh. Wait until an audience is authenticated, or // until shutdown. - m_tokenRefreshCv.wait( - lock, [this]() { return m_tokenRefreshStop || !m_tokenStore.empty(); }); + state->Cv.wait(lock, [&state]() { return state->Stop || !state->TokenStore.empty(); }); } else { - m_tokenRefreshCv.wait_until(lock, nextWake, [this]() { return m_tokenRefreshStop; }); + state->Cv.wait_until(lock, nextWake, [&state]() { return state->Stop; }); } } } @@ -440,31 +460,46 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // mutex for the credential call and for the CBS operation, because both go to // the network. Holding the mutex there would block every caller that opens a // link, and it would stop a shutdown from cancelling this work. - void ConnectionImpl::RefreshTokenForAudience( + // + // This function holds a strong session pointer while the mutex is free, so it + // can hold the last reference to the session and, through the session, the + // last reference to this connection. ReleaseOutsideLock releases that + // pointer, always with the mutex free. The return value tells the caller + // whether this connection can be gone. + bool ConnectionImpl::RefreshTokenForAudience( + TokenRefreshState& state, std::string const& audienceUrl, std::unique_lock& lock) { - std::shared_ptr session; - auto sessionEntry = m_tokenSessions.find(audienceUrl); - if (sessionEntry != m_tokenSessions.end()) + std::shared_ptr promotedSession; + auto sessionEntry = state.TokenSessions.find(audienceUrl); + if (sessionEntry != state.TokenSessions.end()) { - session = sessionEntry->second.lock(); + promotedSession = sessionEntry->second.lock(); } - if (!session) + if (!promotedSession) { // The session that authenticated this audience is gone. Drop the entry, // and the next link open authenticates the audience again. - m_tokenStore.erase(audienceUrl); - m_tokenSessions.erase(audienceUrl); - return; + state.TokenStore.erase(audienceUrl); + state.TokenSessions.erase(audienceUrl); + return false; } + // The hold owns this reference from here on, on every path out of this + // function. + ReleaseOutsideLock sessionHold{std::move(promotedSession), lock}; + // Remember the token that this refresh replaces, so the result does not // overwrite a newer token that a caller stored while the mutex was free. - auto const previousEntry = m_tokenStore.find(audienceUrl); - if (previousEntry == m_tokenStore.end()) + auto const previousEntry = state.TokenStore.find(audienceUrl); + if (previousEntry == state.TokenStore.end()) { - return; + // A caller dropped this token while this pass was in flight. There is + // nothing to replace. Release the session here, so the flag that this + // function returns is read after the release. + sessionHold.Release(); + return state.Stop; } std::string const previousToken{previousEntry->second.Token}; @@ -495,7 +530,12 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // again, which keeps the lock order acyclic. std::lock_guard cbsLock(m_cbsMutex); PutTokenForAudience( - session, tokenType, audienceUrl, accessToken.Token, accessToken.ExpiresOn, context); + sessionHold.Get(), + tokenType, + audienceUrl, + accessToken.Token, + accessToken.ExpiresOn, + context); } refreshed = true; } @@ -507,17 +547,28 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { { failureMessage = "unknown error"; } - // Release the session before the mutex is taken again. This reference can be - // the last one, and a session destructor takes other locks. - session.reset(); + // This is the one point where this thread can destroy the connection. A + // session destructor releases the session's reference to the connection, so + // this release can run the connection destructor on this thread. That + // destructor calls StopTokenRefresh, which sets Stop in the state block and + // detaches this thread. Every line after this one uses `state`, which this + // thread co-owns, and no member of this connection. + sessionHold.Release(); lock.lock(); - auto currentEntry = m_tokenStore.find(audienceUrl); - if (currentEntry == m_tokenStore.end() || currentEntry->second.Token != previousToken) + if (state.Stop) + { + // Either a shutdown is in progress, or the release above destroyed this + // connection. Do not touch this connection again. + return true; + } + + auto currentEntry = state.TokenStore.find(audienceUrl); + if (currentEntry == state.TokenStore.end() || currentEntry->second.Token != previousToken) { // A caller replaced or dropped this token while the mutex was free. That // token is newer than this one, so keep it. - return; + return false; } if (refreshed) @@ -534,9 +585,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // Drop the cache entry, so the next link open authenticates again. Log::Stream(Logger::Level::Warning) << "Could not refresh the token for audience " << audienceUrl << ": " << failureMessage; - m_tokenStore.erase(currentEntry); - m_tokenSessions.erase(audienceUrl); + state.TokenStore.erase(currentEntry); + state.TokenSessions.erase(audienceUrl); } + return false; } #endif // ENABLE_UAMQP }}}} // namespace Azure::Core::Amqp::_detail diff --git a/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp b/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp index 3ba85857b9..130e9ed18e 100644 --- a/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp +++ b/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp @@ -7,9 +7,95 @@ #include #include +#include +#include +#include +#include +#include namespace Azure { namespace Core { namespace Amqp { namespace _detail { + class SessionImpl; + + // The state that the connection shares with the token refresh thread. + // + // The refresh thread promotes a weak session pointer to a strong one, so it + // can hold the last reference to a session and, through that session, the + // last reference to the connection. When it releases that reference, the + // connection destructor runs on the refresh thread. That destructor sets + // Stop, signals Cv, and detaches the thread, because a thread cannot join + // itself. + // + // The thread holds a shared_ptr to this block, so the block stays alive after + // the connection is gone. The thread comes back from the release, takes + // Mutex, sees Stop, and leaves without touching the connection. + struct TokenRefreshState final + { + // Protects every other member of this block. + std::mutex Mutex; + std::condition_variable Cv; + bool Stop{false}; + // The cached token for each audience. + std::map TokenStore; + // The session that authenticated each audience. The pointer is weak, so the + // refresh thread never keeps a session alive. + std::map> TokenSessions; + }; + + // Holds a shared pointer that a thread must not drop while it holds a lock, + // and releases it with that lock free. + // + // The token refresh thread promotes a weak session pointer for the length of + // one refresh. A session holds the connection, so that promoted pointer can + // be the last reference to both, and the release then runs the connection + // destructor on the refresh thread. That destructor takes the token mutex, so + // a release under the token mutex would lock a mutex that this thread already + // holds, and the mutex is not recursive. The release also runs other + // destructors that take other locks. + // + // Every exit from the scope runs this destructor, so no return path and no + // exception leaves the release under the lock. + template class ReleaseOutsideLock final { + public: + ReleaseOutsideLock(std::shared_ptr held, std::unique_lock& lock) + : m_held{std::move(held)}, m_lock{lock} + { + } + + ~ReleaseOutsideLock() { Release(); } + + ReleaseOutsideLock(ReleaseOutsideLock const&) = delete; + ReleaseOutsideLock& operator=(ReleaseOutsideLock const&) = delete; + ReleaseOutsideLock(ReleaseOutsideLock&&) = delete; + ReleaseOutsideLock& operator=(ReleaseOutsideLock&&) = delete; + + std::shared_ptr const& Get() const { return m_held; } + + // Release the pointer with the lock free, then put the lock back in the + // state it was in. This is safe to call more than once. + void Release() + { + if (!m_held) + { + return; + } + bool const wasLocked = m_lock.owns_lock(); + if (wasLocked) + { + m_lock.unlock(); + } + m_held.reset(); + if (wasLocked) + { + m_lock.lock(); + } + } + + private: + std::shared_ptr m_held; + std::unique_lock& m_lock; + }; + // The .NET client refreshes a CBS token seven minutes before the token // expires. See AmqpConnectionScope.cs in Azure.Messaging.EventHubs. Use the // same buffer here. diff --git a/sdk/core/azure-core-amqp/src/impl/rust_amqp/amqp/private/connection_impl.hpp b/sdk/core/azure-core-amqp/src/impl/rust_amqp/amqp/private/connection_impl.hpp index ba4813a624..025c163f23 100644 --- a/sdk/core/azure-core-amqp/src/impl/rust_amqp/amqp/private/connection_impl.hpp +++ b/sdk/core/azure-core-amqp/src/impl/rust_amqp/amqp/private/connection_impl.hpp @@ -3,6 +3,7 @@ #pragma once +#include "../../../../amqp/private/token_refresh.hpp" #include "azure/core/amqp/internal/common/global_state.hpp" #include "azure/core/amqp/internal/connection.hpp" #include "azure/core/amqp/internal/network/transport.hpp" @@ -153,10 +154,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { bool m_connectionOpened{false}; std::atomic m_openCount{0}; - // mutex protecting the token acquisition process. - std::mutex m_tokenMutex; std::shared_ptr m_credential{}; - std::map m_tokenStore; + + // The token mutex and the token cache. The Rust stack has no refresh + // thread, so nothing else shares this block here. + std::shared_ptr m_tokenState{std::make_shared()}; #if ENABLE_UAMQP ConnectionImpl( diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp index a2aa77a182..3b67926b9f 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp @@ -3,6 +3,7 @@ #pragma once +#include "../../../../amqp/private/token_refresh.hpp" #include "../../../../amqp/private/unique_handle.hpp" #include "azure/core/amqp/internal/common/global_state.hpp" #include "azure/core/amqp/internal/connection.hpp" @@ -168,10 +169,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { bool m_connectionOpened = false; std::atomic m_openCount{0}; - // mutex protecting the token acquisition process. - std::mutex m_tokenMutex; std::shared_ptr m_credential{}; - std::map m_tokenStore; + + // The token mutex, the token cache, and the stop protocol for the refresh + // thread. The connection and the refresh thread both own this block, so the + // thread finds it alive even after the connection is destroyed. See + // TokenRefreshState. + std::shared_ptr m_tokenState{std::make_shared()}; // Serializes the CBS operation itself. uAMQP names the CBS links after the // node, so every claims based security object on this connection attaches a @@ -181,23 +185,30 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // thread does its work without the token mutex, so this mutex is what keeps // the refresh and a caller apart. // - // Lock order: a caller takes m_tokenMutex and then this mutex. The refresh - // thread takes this mutex only while it does not hold m_tokenMutex. + // Lock order: a caller takes the token mutex and then this mutex. The + // refresh thread takes this mutex only while it does not hold the token + // mutex. std::mutex m_cbsMutex; - // The session that authenticated each audience. The pointer is weak, so the - // refresh thread never keeps a session alive. - std::map> m_tokenSessions; // The thread that replaces each cached token before the token expires. std::thread m_tokenRefreshThread; - std::condition_variable m_tokenRefreshCv; - bool m_tokenRefreshStop{false}; // Cancelled on shutdown, to stop a CBS operation that is in flight. Azure::Core::Context m_tokenRefreshContext; void StartTokenRefresh(); - void TokenRefreshThread(); - void RefreshTokenForAudience( + + // The body of the refresh thread. This is a static function on purpose. The + // thread can outlive the connection, so it must be able to finish without a + // live `this`. It uses `connection` only while the shared state says the + // connection is alive. + static void TokenRefreshThread( + ConnectionImpl* connection, + std::shared_ptr state); + + // Replace the token for one audience. Return true when the caller must stop + // at once, because this call can have destroyed the connection. + bool RefreshTokenForAudience( + TokenRefreshState& state, std::string const& audienceUrl, std::unique_lock& lock); From b6f425db98ed2663cc4f9bcccd4d415a3fdb6e8d Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 12 Aug 2026 16:19:39 -0400 Subject: [PATCH 6/7] test(core-amqp): cover the token refresh state and the session hold The refresh thread body needs a service, so no unit test reaches it. The two pieces that the lifetime fix added do not need one. TestReleaseOutsideLock puts an object in the hold that records the state of the lock in its destructor. The tests show that the release drops the pointer with the lock free, that it puts the lock back afterwards, and that the destructor does the same on the early return path and on an exception. Those are the properties that keep the connection destructor off the token mutex. TestTokenRefresh gets two more tests for the shared state block: the block starts empty, and it stays alive and usable after the owner that made it is gone. --- .../test/ut/connection_tests.cpp | 188 ++++++++++++++++++ 1 file changed, 188 insertions(+) diff --git a/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp b/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp index 3a20734016..a77f6a4a52 100644 --- a/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp +++ b/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp @@ -20,7 +20,10 @@ #include #include +#include +#include #include +#include #include @@ -115,6 +118,191 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable(token, now)); } + // The state that the connection shares with the refresh thread. + TEST_F(TestTokenRefresh, TheSharedStateStartsEmptyAndRuns) + { + Azure::Core::Amqp::_detail::TokenRefreshState state; + EXPECT_FALSE(state.Stop); + EXPECT_TRUE(state.TokenStore.empty()); + EXPECT_TRUE(state.TokenSessions.empty()); + } + + // The refresh thread co-owns the state block, so the block stays alive after + // the connection that made it is gone. That is what lets the thread come back + // from a release that destroyed the connection, take the mutex, and read the + // stop flag. + TEST_F(TestTokenRefresh, TheSharedStateOutlivesTheOwnerThatMadeIt) + { + std::weak_ptr observer; + std::shared_ptr threadCopy; + { + auto connectionCopy = std::make_shared(); + observer = connectionCopy; + threadCopy = connectionCopy; + } + ASSERT_FALSE(observer.expired()); + + // The owner is gone, so this stands for the connection destructor setting + // the stop flag before it detaches the thread. + { + std::unique_lock lock(threadCopy->Mutex); + threadCopy->Stop = true; + } + EXPECT_TRUE(threadCopy->Stop); + + threadCopy.reset(); + EXPECT_TRUE(observer.expired()); + } + + // Tests for ReleaseOutsideLock, the hold that the refresh thread puts the + // promoted session in. The session can be the last reference to the + // connection, and the connection destructor takes the token mutex, so the + // hold must always drop the pointer with that mutex free. + class TestReleaseOutsideLock : public testing::Test { + protected: + // Records the state of the lock at the moment it is destroyed. + class LockObserver final { + public: + LockObserver(std::unique_lock& lock, bool& destroyed, bool& lockWasHeld) + : m_lock{lock}, m_destroyed{destroyed}, m_lockWasHeld{lockWasHeld} + { + } + + ~LockObserver() + { + m_destroyed = true; + m_lockWasHeld = m_lock.owns_lock(); + } + + private: + std::unique_lock& m_lock; + bool& m_destroyed; + bool& m_lockWasHeld; + }; + + std::mutex m_mutex; + bool m_destroyed{false}; + bool m_lockWasHeld{true}; + + std::shared_ptr MakeObserver(std::unique_lock& lock) + { + return std::make_shared(lock, m_destroyed, m_lockWasHeld); + } + }; + + TEST_F(TestReleaseOutsideLock, ReleaseDropsThePointerWithTheLockFree) + { + std::unique_lock lock(m_mutex); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + EXPECT_TRUE(lock.owns_lock()); + EXPECT_FALSE(m_destroyed); + hold.Release(); + EXPECT_TRUE(m_destroyed); + } + EXPECT_FALSE(m_lockWasHeld); + // The lock is back in the state the caller left it in. + EXPECT_TRUE(lock.owns_lock()); + } + + // The early return in the refresh path leaves the scope with the lock held + // and the session still in the hold. The destructor must give up the lock for + // that release too, or it locks a mutex this thread already holds. + TEST_F(TestReleaseOutsideLock, TheDestructorAlsoReleasesWithTheLockFree) + { + std::unique_lock lock(m_mutex); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + EXPECT_FALSE(m_destroyed); + } + EXPECT_TRUE(m_destroyed); + EXPECT_FALSE(m_lockWasHeld); + EXPECT_TRUE(lock.owns_lock()); + } + + // An exception on the refresh path must not leave the release under the lock + // either. + TEST_F(TestReleaseOutsideLock, AnExceptionStillReleasesWithTheLockFree) + { + std::unique_lock lock(m_mutex); + bool caught{false}; + try + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + throw std::runtime_error("the refresh failed"); + } + catch (std::runtime_error const&) + { + caught = true; + } + EXPECT_TRUE(caught); + EXPECT_TRUE(m_destroyed); + EXPECT_FALSE(m_lockWasHeld); + EXPECT_TRUE(lock.owns_lock()); + } + + // The refresh path releases on the normal path and then leaves the scope, so + // the pointer is released once and the second call does nothing. + TEST_F(TestReleaseOutsideLock, ASecondReleaseDoesNothing) + { + std::unique_lock lock(m_mutex); + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + hold.Release(); + EXPECT_EQ(nullptr, hold.Get()); + + m_destroyed = false; + hold.Release(); + EXPECT_FALSE(m_destroyed); + EXPECT_TRUE(lock.owns_lock()); + } + + // The refresh path gives up the token mutex for the network work. A release + // that happens then must leave the lock free, not take it. + TEST_F(TestReleaseOutsideLock, AFreeLockStaysFree) + { + std::unique_lock lock(m_mutex); + lock.unlock(); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + hold.Release(); + } + EXPECT_TRUE(m_destroyed); + EXPECT_FALSE(m_lockWasHeld); + EXPECT_FALSE(lock.owns_lock()); + } + + // The hold keeps the pointer usable for the whole refresh, and it does not + // destroy an object that another owner still holds. + TEST_F(TestReleaseOutsideLock, TheHoldKeepsThePointerAndSharesIt) + { + std::unique_lock lock(m_mutex); + auto observer = MakeObserver(lock); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{observer, lock}; + EXPECT_EQ(observer.get(), hold.Get().get()); + hold.Release(); + // This test still owns the object, so the release did not destroy it. + EXPECT_FALSE(m_destroyed); + } + EXPECT_FALSE(m_destroyed); + observer.reset(); + EXPECT_TRUE(m_destroyed); + } + + // An empty hold is what the refresh path never builds, but the class must not + // touch the lock for one. + TEST_F(TestReleaseOutsideLock, AnEmptyHoldLeavesTheLockAlone) + { + std::unique_lock lock(m_mutex); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{nullptr, lock}; + hold.Release(); + EXPECT_TRUE(lock.owns_lock()); + } + EXPECT_TRUE(lock.owns_lock()); + EXPECT_FALSE(m_destroyed); + } + #if !defined(AZ_PLATFORM_MAC) TEST_F(TestConnections, SimpleConnection) { From ffbb94e68ca2d65f7e674d0c4a0cb2cafb47061e Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 13 Aug 2026 13:13:50 -0400 Subject: [PATCH 7/7] test(core-amqp): cover the model value and terminus behaviour The core coverage gate needs 81 percent of lines. The branch reports 80.97 percent, so it fails by three lines. These tests close that gap with a margin. The gate builds the Rust AMQP stack, so every new test targets code that the Rust configuration compiles. The three model files hold the largest blocks of untested behaviour that the public API can reach: - AmqpValue ordering for the non scalar types. AmqpValue is the key type of AmqpMap, so operator< must work for timestamps, UUIDs, described values, and composite values, and not only for the scalar types. - AmqpValue equality for composite and described values. The underlying library cannot compare these two types, so operator== unwraps them first. No test covered that path. - The ostream inserter for AmqpValueType::Invalid and Unknown. No value can hold these two types, so the other tests never reach them. - Copy assignment and move assignment for MessageSource and MessageTarget. Only the copy constructor had a test. - The source default outcome round trip for all four AMQP delivery outcomes, and the two error cases. Only the accepted outcome had a test. Local llvm-cov on the Rust stack shows the newly covered lines: 36 in amqp_value.cpp, 37 in message_source.cpp, and 9 in message_target.cpp. The uAMQP build gives 134 of 134 tests. The Rust build gives 134 of 134 tests. --- .../test/ut/amqp_value_tests.cpp | 88 +++++++++++++++++++ .../test/ut/message_source_target.cpp | 81 +++++++++++++++++ 2 files changed, 169 insertions(+) diff --git a/sdk/core/azure-core-amqp/test/ut/amqp_value_tests.cpp b/sdk/core/azure-core-amqp/test/ut/amqp_value_tests.cpp index cec0ff665b..d767f3360b 100644 --- a/sdk/core/azure-core-amqp/test/ut/amqp_value_tests.cpp +++ b/sdk/core/azure-core-amqp/test/ut/amqp_value_tests.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include @@ -563,6 +564,93 @@ TEST_F(TestValues, TestDescribed) } } +TEST_F(TestValues, ValueTypeOstreamInserter) +{ + // Invalid and Unknown are the two types that no AmqpValue can hold, so the other tests never + // reach them. The inserter still has to name them, because it reports the type of a value that + // the peer sent. + { + std::stringstream ss; + ss << AmqpValueType::Invalid; + EXPECT_EQ("Invalid", ss.str()); + } + { + std::stringstream ss; + ss << AmqpValueType::Unknown; + EXPECT_EQ("Unknown", ss.str()); + } +} + +// AmqpValue is the key type of AmqpMap, so its ordering must work for every value type, and not +// only for the scalar types. +TEST_F(TestValues, ValueOrdering) +{ + // Values of different types order by the numeric order of AmqpValueType. Null comes before Int. + EXPECT_LT(AmqpValue{}, AmqpValue{5}); + EXPECT_FALSE(AmqpValue{5} < AmqpValue{}); + + // Two null values are equal, so neither one is less than the other. + EXPECT_FALSE(AmqpValue{} < AmqpValue{}); + + { + AmqpValue smaller{U'a'}; + AmqpValue larger{U'b'}; + EXPECT_LT(smaller, larger); + EXPECT_FALSE(larger < smaller); + } + { + AmqpValue earlier{AmqpTimestamp{std::chrono::milliseconds{1000}}.AsAmqpValue()}; + AmqpValue later{AmqpTimestamp{std::chrono::milliseconds{2000}}.AsAmqpValue()}; + EXPECT_LT(earlier, later); + EXPECT_FALSE(later < earlier); + } + { + std::array smallerBytes{}; + std::array largerBytes{}; + largerBytes[0] = 1; + AmqpValue smaller{Azure::Core::Uuid::CreateFromArray(smallerBytes)}; + AmqpValue larger{Azure::Core::Uuid::CreateFromArray(largerBytes)}; + EXPECT_LT(smaller, larger); + EXPECT_FALSE(larger < smaller); + } + { + // Described values with the same descriptor order by their value. + AmqpDescribed smaller{AmqpSymbol{"descriptor"}, 1}; + AmqpDescribed larger{AmqpSymbol{"descriptor"}, 2}; + EXPECT_LT(smaller.AsAmqpValue(), larger.AsAmqpValue()); + EXPECT_FALSE(larger.AsAmqpValue() < smaller.AsAmqpValue()); + } + { + // Composite values order by their fields. + AmqpComposite smaller("composite", {1}); + AmqpComposite larger("composite", {2}); + EXPECT_LT(smaller.AsAmqpValue(), larger.AsAmqpValue()); + EXPECT_FALSE(larger.AsAmqpValue() < smaller.AsAmqpValue()); + } +} + +TEST_F(TestValues, CompositeAndDescribedEquality) +{ + // The underlying amqpvalue_are_equal does not handle composite or described values, so + // AmqpValue::operator== unwraps them and compares the unwrapped values. + { + AmqpComposite composite1("composite", {25, 25.0f}); + AmqpComposite composite2("composite", {25, 25.0f}); + AmqpComposite composite3("composite", {26, 25.0f}); + EXPECT_EQ(composite1.AsAmqpValue(), composite2.AsAmqpValue()); + EXPECT_NE(composite1.AsAmqpValue(), composite3.AsAmqpValue()); + } + { + AmqpDescribed described1{AmqpSymbol{"descriptor"}, 25}; + AmqpDescribed described2{AmqpSymbol{"descriptor"}, 25}; + AmqpDescribed described3{AmqpSymbol{"descriptor"}, 26}; + EXPECT_EQ(described1.AsAmqpValue(), described2.AsAmqpValue()); + EXPECT_NE(described1.AsAmqpValue(), described3.AsAmqpValue()); + } + // Values of different types are never equal, even when they read the same. + EXPECT_NE(AmqpValue{5}, AmqpValue{"5"}); +} + class TestValueSerialization : public testing::Test { protected: void SetUp() override {} diff --git a/sdk/core/azure-core-amqp/test/ut/message_source_target.cpp b/sdk/core/azure-core-amqp/test/ut/message_source_target.cpp index 9067b68172..d76fb8a855 100644 --- a/sdk/core/azure-core-amqp/test/ut/message_source_target.cpp +++ b/sdk/core/azure-core-amqp/test/ut/message_source_target.cpp @@ -5,6 +5,8 @@ #include #include +#include + #include class TestSourceTarget : public testing::Test { @@ -271,6 +273,31 @@ TEST_F(TestSourceTarget, TargetThroughValue) EXPECT_EQ(target.GetAddress(), target2.GetAddress()); } +TEST_F(TestSourceTarget, TargetAssignment) +{ + { + MessageTarget target{"address2"}; + { + MessageTarget other{"address1"}; + target = other; + } + // Copy assignment copies the implementation, so the assigned object stays valid after the + // original goes out of scope. + EXPECT_EQ(AmqpValue{"address1"}, target.GetAddress()); + } + { + MessageTarget target{"address1"}; + MessageTarget moved{std::move(target)}; + EXPECT_EQ(AmqpValue{"address1"}, moved.GetAddress()); + } + { + MessageTarget target{"address2"}; + MessageTarget other{"address1"}; + target = std::move(other); + EXPECT_EQ(AmqpValue{"address1"}, target.GetAddress()); + } +} + MessageSource ReturnsSource() { return MessageSource(); } MessageSource ReturnsSource(const char* str) { return MessageSource(str); } MessageSource ReturnsSource(const std::string& str) { return MessageSource(str); } @@ -537,3 +564,57 @@ TEST_F(TestSourceTarget, SourceProperties) EXPECT_EQ(source.GetAddress(), source2.GetAddress()); } } + +TEST_F(TestSourceTarget, SourceAssignment) +{ + { + MessageSource source{"address2"}; + { + MessageSource other{"address1"}; + source = other; + } + // Copy assignment copies the implementation, so the assigned object stays valid after the + // original goes out of scope. + EXPECT_EQ(AmqpValue{"address1"}, source.GetAddress()); + } + { + MessageSource source{"address1"}; + MessageSource moved{std::move(source)}; + EXPECT_EQ(AmqpValue{"address1"}, moved.GetAddress()); + } + { + MessageSource source{"address2"}; + MessageSource other{"address1"}; + source = std::move(other); + EXPECT_EQ(AmqpValue{"address1"}, source.GetAddress()); + } +} + +// uAMQP accepts any value as the default outcome. The Rust stack accepts only the four delivery +// outcomes that the AMQP spec names, so the round trip and the two error cases below are specific +// to that stack. +#if ENABLE_RUST_AMQP +TEST_F(TestSourceTarget, SourceDefaultOutcome) +{ + for (auto const& outcome : + {"amqp:accepted:list", "amqp:rejected:list", "amqp:released:list", "amqp:modified:list"}) + { + MessageSourceOptions options; + options.DefaultOutcome = AmqpSymbol{outcome}; + MessageSource source(options); + EXPECT_EQ(source.GetDefaultOutcome(), AmqpSymbol{outcome}); + } + { + // A default outcome that is a string, and not a symbol, is rejected. + MessageSourceOptions options; + options.DefaultOutcome = AmqpValue{"amqp:accepted:list"}; + EXPECT_ANY_THROW(MessageSource source(options)); + } + { + // A symbol that names no delivery outcome is rejected. + MessageSourceOptions options; + options.DefaultOutcome = AmqpSymbol{"amqp:unknown:list"}; + EXPECT_ANY_THROW(MessageSource source(options)); + } +} +#endif