diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index a6101b6625..3e08a88c3c 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/sdk/core/azure-core-amqp/CHANGELOG.md @@ -11,6 +11,8 @@ ### Bugs Fixed +- The uAMQP message sender now encodes delivery annotations, message annotations, and the footer as described sections, so a uAMQP receiver can decode a message that carries them. Before, the sender wrote the bare maps, and the receiving link failed with "Error decoding message" and went to the error state. [[#7376]](https://github.com/Azure/azure-sdk-for-cpp/issues/7376) +- On the uAMQP transport, `MessageSender::Open` and `MessageReceiver::Open` now throw `_detail::CbsPutTokenFailedException` when the service rejects the CBS put-token. The type derives from `std::runtime_error` and carries the original `AuthenticationException`, which `RethrowOriginal()` throws again. The Event Hubs clients use the type to tell a rejected put-token from a credential failure. `ManagementClient::Open` and `ManagementClient::ExecuteOperation` still throw `AuthenticationException`. [[#7376]](https://github.com/Azure/azure-sdk-for-cpp/issues/7376) - uAMQP pollable registration and removal no longer block each other while a poll is in flight. The polling registry now waits on completion notifications, and sender, receiver, and link setup and teardown do not hold connection locks across registry operations. The polling thread now sleeps diff --git a/sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/claims_based_security.hpp b/sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/claims_based_security.hpp index b6e93fe10c..8774c28578 100644 --- a/sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/claims_based_security.hpp +++ b/sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/claims_based_security.hpp @@ -7,8 +7,10 @@ #include +#include #include #include +#include namespace Azure { namespace Core { namespace Amqp { namespace _detail { class ClaimsBasedSecurityImpl; @@ -53,6 +55,23 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { CbsOpenResult Result; }; +#if ENABLE_UAMQP + /** @brief Identifies a failed uAMQP CBS put-token operation. */ + class CbsPutTokenFailedException final : public std::runtime_error { + public: + CbsPutTokenFailedException(std::exception_ptr original, std::string const& what) + : std::runtime_error(what), m_original{std::move(original)} + { + } + + std::exception_ptr GetOriginal() const { return m_original; } + [[noreturn]] void RethrowOriginal() const { std::rethrow_exception(m_original); } + + private: + std::exception_ptr m_original; + }; +#endif + enum class CbsTokenType { Invalid, diff --git a/sdk/core/azure-core-amqp/src/amqp/connection.cpp b/sdk/core/azure-core-amqp/src/amqp/connection.cpp index 4f2f4be85d..12183250f4 100644 --- a/sdk/core/azure-core-amqp/src/amqp/connection.cpp +++ b/sdk/core/azure-core-amqp/src/amqp/connection.cpp @@ -195,9 +195,14 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { = claimsBasedSecurity->PutToken(tokenType, audienceUrl, token, expiresOn, context); if (std::get<0>(result) != CbsOperationResult::Ok) { - throw Azure::Core::Credentials::AuthenticationException( + auto failure = Azure::Core::Credentials::AuthenticationException( "Could not authenticate client. Error Status: " + std::to_string(std::get<1>(result)) + " reason: " + std::get<2>(result)); +#if ENABLE_UAMQP + throw CbsPutTokenFailedException(std::make_exception_ptr(failure), failure.what()); +#else + throw failure; +#endif } Log::Stream(Logger::Level::Verbose) << "Close CBS object"; claimsBasedSecurity->Close(context); diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/management.cpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/management.cpp index 5934703b23..218f0685d1 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/management.cpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/management.cpp @@ -3,6 +3,7 @@ #include "azure/core/amqp/internal/management.hpp" +#include "azure/core/amqp/internal/claims_based_security.hpp" #include "azure/core/amqp/internal/models/messaging_values.hpp" #include "azure/core/amqp/models/amqp_message.hpp" #include "private/connection_impl.hpp" @@ -90,6 +91,22 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { } } + // The put-token marker is for the Event Hubs retry loops. Management callers keep + // the AuthenticationException contract. + Credentials::AccessToken ManagementClientImpl::AuthenticateManagementAudience( + Context const& context) + { + try + { + return m_session->GetConnection()->AuthenticateAudience( + m_session, m_managementEntityPath + "/" + m_options.ManagementNodeName, context); + } + catch (CbsPutTokenFailedException const& failure) + { + failure.RethrowOriginal(); + } + } + _internal::ManagementOpenStatus ManagementClientImpl::Open(Context const& context) { std::unique_lock lock(m_openCloseLock); @@ -107,8 +124,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { */ if (m_options.ManagementNodeName == "$management") { - m_accessToken = m_session->GetConnection()->AuthenticateAudience( - m_session, m_managementEntityPath + "/" + m_options.ManagementNodeName, context); + m_accessToken = AuthenticateManagementAudience(context); } { _internal::MessageSenderOptions messageSenderOptions; @@ -243,8 +259,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // than one thread and that member has no lock. if (!m_accessToken.Token.empty()) { - auto accessToken{m_session->GetConnection()->AuthenticateAudience( - m_session, m_managementEntityPath + "/" + m_options.ManagementNodeName, context)}; + auto accessToken{AuthenticateManagementAudience(context)}; messageToSend.ApplicationProperties["security_token"] = Models::AmqpValue{accessToken.Token}; } diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/management_impl.hpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/management_impl.hpp index aa21a021b2..bba0d1b05c 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/management_impl.hpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/management_impl.hpp @@ -109,6 +109,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { bool m_sendCompleted{false}; void CloseSenderAndReceiverAfterFailedOpen() noexcept; + Azure::Core::Credentials::AccessToken AuthenticateManagementAudience(Context const& context); void SetState(ManagementState newState); // Reflect the error state to the OnError callback and return a delivery rejected status. Models::AmqpValue IndicateError( diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/vendor/azure-uamqp-c/src/message_sender.c b/sdk/core/azure-core-amqp/src/impl/uamqp/vendor/azure-uamqp-c/src/message_sender.c index 00fbf9f0ed..c74a308069 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/vendor/azure-uamqp-c/src/message_sender.c +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/vendor/azure-uamqp-c/src/message_sender.c @@ -241,8 +241,11 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message size_t body_data_count = 0; size_t body_sequence_count = 0; AMQP_VALUE msg_annotations = NULL; + AMQP_VALUE msg_annotations_value = NULL; AMQP_VALUE footer = NULL; + AMQP_VALUE footer_value = NULL; AMQP_VALUE delivery_annotations = NULL; + AMQP_VALUE delivery_annotations_value = NULL; bool is_error = false; // message header @@ -275,14 +278,23 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message (message_get_message_annotations(message, &msg_annotations) == 0) && (msg_annotations != NULL)) { - if (amqpvalue_get_encoded_size(msg_annotations, &encoded_size) != 0) + msg_annotations_value = amqpvalue_create_message_annotations(msg_annotations); + if (msg_annotations_value == NULL) { - LogError("Cannot obtain message annotations encoded size"); + LogError("Cannot create message annotations AMQP value"); is_error = true; } else { - total_encoded_size += encoded_size; + if (amqpvalue_get_encoded_size(msg_annotations_value, &encoded_size) != 0) + { + LogError("Cannot obtain message annotations encoded size"); + is_error = true; + } + else + { + total_encoded_size += encoded_size; + } } } @@ -341,14 +353,23 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message (message_get_footer(message, &footer) == 0) && (footer != NULL)) { - if (amqpvalue_get_encoded_size(footer, &encoded_size) != 0) + footer_value = amqpvalue_create_footer(footer); + if (footer_value == NULL) { - LogError("Cannot obtain footer encoded size"); + LogError("Cannot create footer AMQP value"); is_error = true; } else { - total_encoded_size += encoded_size; + if (amqpvalue_get_encoded_size(footer_value, &encoded_size) != 0) + { + LogError("Cannot obtain footer encoded size"); + is_error = true; + } + else + { + total_encoded_size += encoded_size; + } } } @@ -357,14 +378,23 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message (message_get_delivery_annotations(message, &delivery_annotations) == 0) && (delivery_annotations != NULL)) { - if (amqpvalue_get_encoded_size(delivery_annotations, &encoded_size) != 0) + delivery_annotations_value = amqpvalue_create_delivery_annotations(delivery_annotations); + if (delivery_annotations_value == NULL) { - LogError("Cannot obtain delivery annotations encoded size"); + LogError("Cannot create delivery annotations AMQP value"); is_error = true; } else { - total_encoded_size += encoded_size; + if (amqpvalue_get_encoded_size(delivery_annotations_value, &encoded_size) != 0) + { + LogError("Cannot obtain delivery annotations encoded size"); + is_error = true; + } + else + { + total_encoded_size += encoded_size; + } } } @@ -554,13 +584,13 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message if ((result == SEND_ONE_MESSAGE_OK) && (msg_annotations != NULL)) { - if (amqpvalue_encode(msg_annotations, encode_bytes, &payload) != 0) + if (amqpvalue_encode(msg_annotations_value, encode_bytes, &payload) != 0) { LogError("Cannot encode message annotations value"); result = SEND_ONE_MESSAGE_ERROR; } - log_message_chunk(message_sender, "Message Annotations:", msg_annotations); + log_message_chunk(message_sender, "Message Annotations:", msg_annotations_value); } if ((result == SEND_ONE_MESSAGE_OK) && (properties != NULL)) @@ -587,24 +617,24 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message if ((result == SEND_ONE_MESSAGE_OK) && (footer != NULL)) { - if (amqpvalue_encode(footer, encode_bytes, &payload) != 0) + if (amqpvalue_encode(footer_value, encode_bytes, &payload) != 0) { LogError("Cannot encode footer value"); result = SEND_ONE_MESSAGE_ERROR; } - log_message_chunk(message_sender, "Footer:", footer); + log_message_chunk(message_sender, "Footer:", footer_value); } if ((result == SEND_ONE_MESSAGE_OK) && (delivery_annotations != NULL)) { - if (amqpvalue_encode(delivery_annotations, encode_bytes, &payload) != 0) + if (amqpvalue_encode(delivery_annotations_value, encode_bytes, &payload) != 0) { LogError("Cannot encode delivery annotations value"); result = SEND_ONE_MESSAGE_ERROR; } - log_message_chunk(message_sender, "Delivery annotations:", delivery_annotations); + log_message_chunk(message_sender, "Delivery annotations:", delivery_annotations_value); } if (result == SEND_ONE_MESSAGE_OK) @@ -764,6 +794,11 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message annotations_destroy(msg_annotations); } + if (msg_annotations_value != NULL) + { + amqpvalue_destroy(msg_annotations_value); + } + if (application_properties != NULL) { amqpvalue_destroy(application_properties); @@ -789,10 +824,20 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message annotations_destroy(footer); } + if (footer_value != NULL) + { + amqpvalue_destroy(footer_value); + } + if (delivery_annotations != NULL) { annotations_destroy(delivery_annotations); } + + if (delivery_annotations_value != NULL) + { + amqpvalue_destroy(delivery_annotations_value); + } } return result; diff --git a/sdk/core/azure-core-amqp/test/ut/message_sender_receiver.cpp b/sdk/core/azure-core-amqp/test/ut/message_sender_receiver.cpp index 64535a8bb7..e523dd4055 100644 --- a/sdk/core/azure-core-amqp/test/ut/message_sender_receiver.cpp +++ b/sdk/core/azure-core-amqp/test/ut/message_sender_receiver.cpp @@ -1565,5 +1565,92 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { CloseAmqpConnection(connection); } +#if !defined(USE_NATIVE_BROKER) + // A message that carries these three sections must decode on the receiving link. + TEST_F(TestMessageSendReceive, ReceiverDecodesAnnotationsAndFooter) + { + std::string brokerEndpoint = GetBrokerEndpoint() + "/annotations"; + + class AnnotatingEndpoint : public MessageTests::MockServiceEndpoint { + public: + AnnotatingEndpoint( + std::string const& name, + MessageTests::MockServiceEndpointOptions const& options) + : MockServiceEndpoint(name, options) + { + } + virtual ~AnnotatingEndpoint() = default; + + void SendOnce(Azure::Core::Amqp::Models::AmqpMessage message) + { + m_message = std::move(message); + m_shouldSend = true; + } + + private: + mutable bool m_shouldSend{false}; + Azure::Core::Amqp::Models::AmqpMessage m_message; + + void Poll() const override + { + if (m_shouldSend && HasMessageSender()) + { + m_shouldSend = false; + EXPECT_EQ(MessageSendStatus::Ok, std::get<0>(GetMessageSender().Send(m_message))); + } + } + + void MessageReceived( + std::string const&, + std::shared_ptr const&) override + { + } + }; + auto serviceEndpoint = std::make_shared( + brokerEndpoint, MessageTests::MockServiceEndpointOptions{}); + m_mockServer.AddServiceEndpoint(serviceEndpoint); + + auto connection{CreateAmqpConnection({})}; + auto session{CreateAmqpSession(connection)}; + StartServerListening(); + + MessageReceiverOptions receiverOptions; + receiverOptions.Name = "annotations-receiver"; + receiverOptions.MessageTarget = "egress"; + receiverOptions.SettleMode = Azure::Core::Amqp::_internal::ReceiverSettleMode::First; + receiverOptions.MaxLinkCredit = 10; + MessageReceiver receiver(session.CreateMessageReceiver(brokerEndpoint, receiverOptions)); + receiver.Open(); + + Azure::Core::Amqp::Models::AmqpMessage sent; + sent.DeliveryAnnotations[Models::AmqpSymbol{"x-opt-delivery"}] = Models::AmqpValue{"delivery"}; + sent.MessageAnnotations[Models::AmqpSymbol{"x-opt-offset"}] = Models::AmqpValue{"10"}; + sent.MessageAnnotations[Models::AmqpSymbol{"x-opt-partition-key"}] + = Models::AmqpValue{"partition"}; + sent.Footer[Models::AmqpSymbol{"x-opt-footer"}] = Models::AmqpValue{"footer"}; + sent.SetBody(Models::AmqpValue{"annotated body"}); + serviceEndpoint->SendOnce(sent); + + Azure::Core::Context receiveContext{Azure::DateTime::clock::now() + std::chrono::seconds(10)}; + auto received = receiver.WaitForIncomingMessage(receiveContext); + if (received.first) + { + EXPECT_TRUE(sent.DeliveryAnnotations == received.first->DeliveryAnnotations); + EXPECT_TRUE(sent.MessageAnnotations == received.first->MessageAnnotations); + EXPECT_TRUE(sent.Footer == received.first->Footer); + EXPECT_EQ("annotated body", static_cast(received.first->GetBodyAsAmqpValue())); + } + else + { + ADD_FAILURE() << "The receiver returned no message: " << received.second; + } + + receiver.Close(); + StopServerListening(); + EndAmqpSession(session); + CloseAmqpConnection(connection); + } +#endif + #endif // !defined(AZ_PLATFORM_MAC) }}}} // namespace Azure::Core::Amqp::Tests diff --git a/sdk/core/azure-core-amqp/test/ut/mock_amqp_server.hpp b/sdk/core/azure-core-amqp/test/ut/mock_amqp_server.hpp index 4ac5c693fe..df376e316a 100644 --- a/sdk/core/azure-core-amqp/test/ut/mock_amqp_server.hpp +++ b/sdk/core/azure-core-amqp/test/ut/mock_amqp_server.hpp @@ -16,7 +16,11 @@ #include #include +#include +#include #include +#include +#include #include @@ -62,9 +66,20 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { { } + virtual ~MockServiceEndpoint() = default; + const std::string& GetName() const { return m_name; } - bool OnLinkAttached( + void DetachLink( + Azure::Core::Amqp::_internal::Session const& session, + Azure::Core::Amqp::_internal::LinkEndpoint const& linkEndpoint, + bool closeLink, + Models::_internal::AmqpError const& error) const + { + session.SendDetach(linkEndpoint, closeLink, error); + } + + virtual bool OnLinkAttached( Azure::Core::Amqp::_internal::Session const& session, std::string const& linkName, Azure::Core::Amqp::_internal::LinkEndpoint& linkEndpoint, @@ -81,6 +96,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { if (role == Azure::Core::Amqp::_internal::SessionRole::Receiver) { GTEST_LOG_(INFO) << "Role is receiver, create sender."; + WaitForStaleLink(linkName, m_sender); if (!HasMessageSender(linkName)) { GTEST_LOG_(INFO) << "No sender found, create new sender for " << linkName; @@ -89,6 +105,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { senderOptions.Name = linkName; senderOptions.MessageSource = source; senderOptions.InitialDeliveryCount = 0; + // The server side has no credential. An authentication step would still read the + // target address, and an Event Hubs consumer attaches without one. + senderOptions.AuthenticationRequired = false; m_sender[linkName] = std::make_unique( session.CreateMessageSender(linkEndpoint, target, senderOptions, this)); // NOTE: The linkEndpoint needs to be attached before this function returns in order to @@ -110,6 +129,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { else if (role == Azure::Core::Amqp::_internal::SessionRole::Sender) { GTEST_LOG_(INFO) << "Role is sender, create receiver."; + WaitForStaleLink(linkName, m_receiver); if (!HasMessageReceiver(linkName)) { GTEST_LOG_(INFO) << "No receiver found, create new receiver for " << linkName; @@ -118,6 +138,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { receiverOptions.Name = linkName; receiverOptions.MessageTarget = target; receiverOptions.InitialDeliveryCount = 0; + // The server side has no credential. An authentication step would still read the + // source address, and an Event Hubs producer attaches without one. + receiverOptions.AuthenticationRequired = false; m_receiver[linkName] = std::make_unique( session.CreateMessageReceiver(linkEndpoint, source, receiverOptions, this)); @@ -188,6 +211,35 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { } protected: + template + static std::unique_ptr TakeLink( + std::string const& linkName, + std::map>& links) + { + auto link = links.find(linkName); + if (link == links.end()) + { + return nullptr; + } + std::unique_ptr taken{std::move(link->second)}; + links.erase(link); + return taken; + } + + // A client that reconnects attaches the same link names. The message loop removes the + // old link a moment after the old connection ends, so a new attach waits for that. + template + static void WaitForStaleLink( + std::string const& linkName, + std::map> const& links) + { + auto const deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (links.find(linkName) != links.end() && std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + bool HasMessageSender(std::string const& linkName = {}) const { if (linkName.empty()) @@ -315,10 +367,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { { std::string senderName = std::get<0>(*senderDisconnected); GTEST_LOG_(INFO) << "Sender disconnected: " << senderName; - std::unique_ptr sender{ - m_sender[senderName].release()}; - m_sender.erase(senderName); - sender->Close(m_listenerContext); + auto sender = TakeLink(senderName, m_sender); + if (sender) + { + sender->Close(m_listenerContext); + } } auto receiverDisconnected = m_messageReceiverDisconnectedQueue.TryWaitForResult(); @@ -326,10 +379,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { { std::string receiverName = std::get<0>(*receiverDisconnected); GTEST_LOG_(INFO) << "Receiver disconnected: " << receiverName; - std::unique_ptr receiver{ - m_receiver[receiverName].release()}; - m_receiver.erase(receiverName); - receiver->Close(m_listenerContext); + auto receiver = TakeLink(receiverName, m_receiver); + if (receiver) + { + receiver->Close(m_listenerContext); + } } auto receiverPollingEnable = m_receiverPollingEnableQueue.TryWaitForResult(); @@ -337,28 +391,41 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { { std::string receiverName = std::get<0>(*receiverPollingEnable); GTEST_LOG_(INFO) << "Enable link polling for receiver: " << receiverName; - m_receiver[receiverName]->EnableLinkPolling(); + if (HasMessageReceiver(receiverName)) + { + GetMessageReceiver(receiverName).EnableLinkPolling(); + } } + // A client that recovers closes its connection and attaches the same links on a new + // one. The loop stays alive for that second attach until StopProcessing cancels it. if (m_receiver.empty() && m_sender.empty()) { - GTEST_LOG_(INFO) << "No more links, exiting message loop."; - break; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + else + { + std::this_thread::yield(); } - - std::this_thread::yield(); } } // Inherited via MessageReceiverEvents void OnMessageReceiverStateChanged( - Azure::Core::Amqp::_internal::MessageReceiver const&, + Azure::Core::Amqp::_internal::MessageReceiver const& receiver, Azure::Core::Amqp::_internal::MessageReceiverState newState, Azure::Core::Amqp::_internal::MessageReceiverState oldState) override { GTEST_LOG_(INFO) << "MockServiceEndpoint(" << m_name << "): Message Receiver State changed.Old state : " << oldState << " New state: " << newState; + // A client that drops its connection raises no disconnect event for the link. The + // Idle state is the one signal, so it removes the link too. + if (newState == Azure::Core::Amqp::_internal::MessageReceiverState::Idle + && oldState != Azure::Core::Amqp::_internal::MessageReceiverState::Idle) + { + m_messageReceiverDisconnectedQueue.CompleteOperation(receiver.GetLinkName()); + } } virtual void OnMessageReceiverDisconnected( @@ -371,13 +438,18 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { // Inherited via MessageSenderEvents void OnMessageSenderStateChanged( - Azure::Core::Amqp::_internal::MessageSender const&, + Azure::Core::Amqp::_internal::MessageSender const& sender, Azure::Core::Amqp::_internal::MessageSenderState newState, Azure::Core::Amqp::_internal::MessageSenderState oldState) override { GTEST_LOG_(INFO) << "MockServiceEndpoint(" << m_name << ") Message Sender State changed.Old state : " << oldState << " New state: " << newState; + if (newState == Azure::Core::Amqp::_internal::MessageSenderState::Idle + && oldState != Azure::Core::Amqp::_internal::MessageSenderState::Idle) + { + m_messageSenderDisconnectedQueue.CompleteOperation(sender.GetLinkName()); + } } void OnMessageSenderDisconnected( @@ -544,24 +616,27 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { public: AmqpServerMock( std::string name = testing::UnitTest::GetInstance()->current_test_info()->name()) - : m_connectionId{"Mock Server for " + name}, m_testPort{FindAvailableSocket()} + : AmqpServerMock(FindAvailableSocket(), std::move(name), true) { - // Every server mock has CBS endpoint support - MockServiceEndpointOptions options; - options.EnableTrace = m_enableTrace; - options.ListenerContext = m_listenerContext; - AddServiceEndpoint(std::make_shared(options)); } AmqpServerMock( uint16_t listeningPort, std::string name = testing::UnitTest::GetInstance()->current_test_info()->name()) + : AmqpServerMock(listeningPort, std::move(name), true) + { + } + + AmqpServerMock(uint16_t listeningPort, std::string name, bool addCbsEndpoint) : m_connectionId{"Mock Server for " + name}, m_testPort{listeningPort} { - // Every server mock has CBS endpoint support - MockServiceEndpointOptions options; - options.EnableTrace = m_enableTrace; - options.ListenerContext = m_listenerContext; - AddServiceEndpoint(std::make_shared(options)); + if (addCbsEndpoint) + { + // Every server mock has CBS endpoint support + MockServiceEndpointOptions options; + options.EnableTrace = m_enableTrace; + options.ListenerContext = m_listenerContext; + AddServiceEndpoint(std::make_shared(options)); + } } virtual ~AmqpServerMock() @@ -579,6 +654,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { } uint16_t GetPort() const { return m_testPort; } + std::size_t GetConnectionCount() const { return m_connectionCount.load(); } Azure::Core::Context& GetListenerContext() { return m_listenerContext; } void StartListening() @@ -685,6 +761,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { auto newConnection = std::make_shared( amqpTransport, options, this, this); m_connections.push_back(newConnection); + m_connectionCount.fetch_add(1); newConnection->Listen(); } @@ -696,12 +773,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { { GTEST_LOG_(INFO) << "Connection State changed. Connection: " << m_connectionId << " Old state : " << oldState << " New state: " << newState; - if (newState == Azure::Core::Amqp::_internal::ConnectionState::End - || newState == Azure::Core::Amqp::_internal::ConnectionState::Error) - { - // If the connection is closed, then we should close the connection. - m_listenerContext.Cancel(); - } + // The listener context is shared with every service endpoint, so a cancel here would + // stop the server after the first client connection ends. StopListening cancels it. } virtual bool OnNewEndpoint( Azure::Core::Amqp::_internal::Connection const& connection, @@ -768,6 +841,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { // The set of incoming connections, used when tearing down the mock server. std::list> m_connections; + std::atomic m_connectionCount{0}; // The set of sessions. std::list> m_sessions; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md b/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md index 72f6ffa4c7..fd9e7abe54 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md @@ -11,6 +11,7 @@ ### Bugs Fixed +- [[#7376]](https://github.com/Azure/azure-sdk-for-cpp/issues/7376) The uAMQP backend now retries a CBS PutToken failure or exact unauthorized send or receive error once on a fresh connection, subject to `RetryOptions`, while preserving the final error. A CBS open `Error` now uses the ordinary retry budget in both `CreateBatch` and `Send`; before this change `CreateBatch` made one extra attempt. Each ordinary retry phase gets its own budget after an authentication recovery. - [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) A teardown of the cached sender no longer runs while another thread sends on that sender. `ProducerClient::Send` gives each attempt a copy of the sender, and a failed attempt on one thread closed the object that a second thread was using. On the Rust AMQP backend that close frees the sender, so the race was a use after free. Each partition now has a guard that lets sends run at the same time and makes a teardown wait for the sends in flight. `ProducerClient::Close` uses the same guard, and it now logs a failed close and continues instead of leaving the other objects open. - [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) `ProducerClient::CreateBatch` now builds a new sender when it cannot read the maximum message size. The client caches a sender for each partition, and a cached sender holds a link that the service detaches after 30 idle minutes. The size of a batch comes from the attached link, so this call was the first one to touch the dead link, and it threw. The `Send(EventData)` overloads go through this call, so the whole producer failed after an idle period even though `Send` builds a new sender on each attempt. - [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) Updated producer retries to honor `EventHubsException::IsTransient`, treat empty AMQP error conditions as transient, stop immediately for unknown and known non-transient failures, preserve bounded retries for AMQP runtime failures, and make backoff cancellable through `Azure::Core::Context`. Retry accounting now always performs the initial attempt and treats `MaxRetries` as additional retry attempts. diff --git a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/consumer_client.hpp b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/consumer_client.hpp index ac3e2335c8..f6fac82c3e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/consumer_client.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/consumer_client.hpp @@ -15,10 +15,17 @@ #include #include #include + +#include +#include +#include namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { class EventHubsPropertiesClient; - } +#if defined(_azure_BUILDING_TESTS) + class ConsumerClientTestAccess; +#endif + } // namespace _detail class ConsumerClient; @@ -199,6 +206,13 @@ namespace Azure { namespace Messaging { namespace EventHubs { Core::Context const& context = {}); private: +#if defined(_azure_BUILDING_TESTS) + friend class _detail::ConsumerClientTestAccess; +#endif + std::mutex m_partitionClientStatesLock; + std::vector> m_partitionClientStates; + bool m_partitionClientStatesClosing{false}; + /// The connection string for the Event Hubs namespace std::string m_connectionString; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/partition_client.hpp b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/partition_client.hpp index aa663fe2cc..9f47fef27e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/partition_client.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/partition_client.hpp @@ -11,10 +11,15 @@ #include #include +#include + namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { class PartitionClientFactory; - } + struct PartitionClientState; + } // namespace _detail + + class ConsumerClient; /**brief PartitionClientOptions provides options for the ConsumerClient::CreatePartitionClient * function. */ @@ -66,7 +71,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { /// Assign a PartitionClient to another PartitionClient PartitionClient& operator=(PartitionClient const& other) = delete; /// Move a PartitionClient to another PartitionClient - PartitionClient& operator=(PartitionClient&& other) = default; + PartitionClient& operator=(PartitionClient&& other); /** Destroy this partition client. */ @@ -89,6 +94,10 @@ namespace Azure { namespace Messaging { namespace EventHubs { private: friend class _detail::PartitionClientFactory; + friend class ConsumerClient; + + std::shared_ptr<_detail::PartitionClientState> m_state; + /// The message receiver used to receive events from the partition. Azure::Core::Amqp::_internal::MessageReceiver m_receiver; @@ -116,6 +125,12 @@ namespace Azure { namespace Messaging { namespace EventHubs { */ Azure::Core::Http::Policies::RetryOptions m_retryOptions{}; +#if ENABLE_UAMQP + explicit PartitionClient(std::shared_ptr<_detail::PartitionClientState> state); + std::shared_ptr<_detail::PartitionClientState> GetState() const { return m_state; } +#endif + +#if ENABLE_RUST_AMQP /** Creates a new PartitionClient * * @param messageReceiver Message Receiver for the partition client. @@ -133,10 +148,15 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::string receiverName, PartitionClientOptions options, Core::Http::Policies::RetryOptions retryOptions); +#endif +#if ENABLE_RUST_AMQP || ENABLE_UAMQP /// Closes the faulted receiver and attaches a new one starting after the last offset. void RebuildReceiver(Core::Context const& context); +#endif +#if ENABLE_RUST_AMQP std::string GetStartExpression(Models::StartPosition const& startPosition); +#endif }; }}} // namespace Azure::Messaging::EventHubs diff --git a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/producer_client.hpp b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/producer_client.hpp index a0d185668d..9fd4abb91a 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/producer_client.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/producer_client.hpp @@ -195,6 +195,8 @@ namespace Azure { namespace Messaging { namespace EventHubs { } private: + struct ProducerCallState; + /// The connection string for the Event Hubs namespace std::string m_connectionString; @@ -259,6 +261,16 @@ namespace Azure { namespace Messaging { namespace EventHubs { // Ensure that a message sender for the specified partition has been created. void EnsureSender(std::string const& partitionId, Azure::Core::Context const& context); + EventDataBatch CreateBatch( + EventDataBatchOptions const& options, + Azure::Core::Context const& context, + ProducerCallState& callState); + + void Send( + EventDataBatch const& eventDataBatch, + Core::Context const& context, + ProducerCallState& callState); + // Calls EnsureSender, and discards a failed attach unless the context is cancelled. void EnsureSenderOrInvalidate( std::string const& partitionId, @@ -268,7 +280,8 @@ namespace Azure { namespace Messaging { namespace EventHubs { // security open reported CbsOpenResult::Error. void EstablishSenderWithRetry( std::string const& partitionId, - Azure::Core::Context const& context); + Azure::Core::Context const& context, + ProducerCallState& callState); // Discards the sender, session, and connection for the partition. A null generation, // as Close passes, removes whatever is present regardless of generation. diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/consumer_client.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/consumer_client.cpp index 12122175b4..18b2becafe 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/consumer_client.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/consumer_client.cpp @@ -9,6 +9,7 @@ #include #include +#include #include using namespace Azure::Core::Diagnostics::_internal; @@ -67,6 +68,26 @@ namespace Azure { namespace Messaging { namespace EventHubs { void ConsumerClient::Close(Azure::Core::Context const& context) { Log::Stream(Logger::Level::Verbose) << "Close consumer client."; +#if ENABLE_UAMQP + std::vector> partitionClientStates; + { + std::lock_guard lock(m_partitionClientStatesLock); + if (m_partitionClientStatesClosing) + { + return; + } + m_partitionClientStatesClosing = true; + partitionClientStates.reserve(m_partitionClientStates.size()); + for (auto const& weakState : m_partitionClientStates) + { + if (auto state = weakState.lock()) + { + partitionClientStates.push_back(std::move(state)); + } + } + m_partitionClientStates.clear(); + } +#endif { std::unique_lock lock(m_propertiesClientLock); if (m_propertiesClient) @@ -83,6 +104,12 @@ namespace Azure { namespace Messaging { namespace EventHubs { m_propertiesClient.reset(); } } +#if ENABLE_UAMQP + for (auto const& state : partitionClientStates) + { + _detail::ClosePartitionClientState(state, context); + } +#endif Log::Stream(Logger::Level::Verbose) << "Closing message receivers."; // Tear down the sessions and then the connections, in that order. _detail::ForEachBestEffort( @@ -96,6 +123,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { }); #if ENABLE_RUST_AMQP + static_cast(m_partitionClientStatesClosing); Log::Stream(Logger::Level::Verbose) << "Closing sessions."; _detail::ForEachBestEffort( m_sessions.begin(), @@ -218,6 +246,53 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::string suffix = !partitionId.empty() ? "/Partitions/" + partitionId : ""; std::string hostUrl = m_hostUrl + suffix; +#if ENABLE_UAMQP + { + std::lock_guard lock(m_partitionClientStatesLock); + if (m_partitionClientStatesClosing) + { + throw Azure::Core::OperationCancelledException("Consumer client is closed."); + } + } + auto partition = _detail::PartitionClientFactory::CreatePartitionClient( + m_fullyQualifiedNamespace, + m_credential, + m_targetPort, + m_consumerClientOptions.ApplicationID, + m_consumerClientOptions.CppStandardVersion, + "Consumer for " + m_consumerClientOptions.ApplicationID + " on " + partitionId, + std::move(hostUrl), + m_consumerClientOptions.Name, + options, + m_consumerClientOptions.RetryOptions, + context); + bool closeLatePartition = false; + { + std::lock_guard lock(m_partitionClientStatesLock); + if (m_partitionClientStatesClosing) + { + closeLatePartition = true; + } + else + { + m_partitionClientStates.erase( + std::remove_if( + m_partitionClientStates.begin(), + m_partitionClientStates.end(), + [](std::weak_ptr<_detail::PartitionClientState> const& state) { + return state.expired(); + }), + m_partitionClientStates.end()); + m_partitionClientStates.push_back(partition.GetState()); + } + } + if (closeLatePartition) + { + _detail::ClosePartitionClientState(partition.GetState(), context); + throw Azure::Core::OperationCancelledException("Consumer client is closed."); + } + return partition; +#elif ENABLE_RUST_AMQP EnsureSession(partitionId, context); return _detail::PartitionClientFactory::CreatePartitionClient( @@ -227,6 +302,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { options, m_consumerClientOptions.RetryOptions, context); +#endif } Models::EventHubProperties ConsumerClient::GetEventHubProperties(Core::Context const& context) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/partition_client.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/partition_client.cpp index b8053a61d1..661b960e5d 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/partition_client.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/partition_client.cpp @@ -9,10 +9,16 @@ #include "private/retry_operation.hpp" #include +#include #include #include +#include #include +#include +#include +#include +#include #include using namespace Azure::Core::Diagnostics::_internal; @@ -182,6 +188,328 @@ namespace Azure { namespace Messaging { namespace EventHubs { } // namespace +#if ENABLE_UAMQP + namespace _detail { + enum class PendingFailureKind + { + None, + Ordinary, + Authentication, + Permanent, + }; + + struct ReceiverStack final + { + ReceiverStack( + Azure::Core::Amqp::_internal::Connection connection, + Azure::Core::Amqp::_internal::Session session, + Azure::Core::Amqp::_internal::MessageReceiver receiver) + : Connection{std::move(connection)}, Session{std::move(session)}, Receiver{ + std::move(receiver)} + { + } + + Azure::Core::Amqp::_internal::Connection Connection; + Azure::Core::Amqp::_internal::Session Session; + Azure::Core::Amqp::_internal::MessageReceiver Receiver; + }; + + struct PartitionClientState final + { + PartitionClientState( + std::string fullyQualifiedNamespace, + std::shared_ptr credential, + std::uint16_t targetPort, + std::string applicationId, + long cppStandardVersion, + std::string containerId, + std::string partitionUrl, + std::string receiverName, + PartitionClientOptions options, + Azure::Core::Http::Policies::RetryOptions retryOptions) + : FullyQualifiedNamespace{std::move(fullyQualifiedNamespace)}, + Credential{std::move(credential)}, TargetPort{targetPort}, + ApplicationId{std::move(applicationId)}, CppStandardVersion{cppStandardVersion}, + ContainerId{std::move(containerId)}, PartitionUrl{std::move(partitionUrl)}, + ReceiverName{std::move(receiverName)}, Options{std::move(options)}, + RetryOptions{std::move(retryOptions)} + { + } + + std::mutex Lock; + std::mutex ReceiveLock; + std::condition_variable ReceiveCondition; + std::shared_ptr Stack; + std::uint64_t Generation{0}; + bool Closed{false}; + bool ActiveReceive{false}; + Azure::Core::Context ActiveReceiveContext; + + std::string FullyQualifiedNamespace; + std::shared_ptr Credential; + std::uint16_t TargetPort; + std::string ApplicationId; + long CppStandardVersion; + std::string ContainerId; + std::string PartitionUrl; + std::string ReceiverName; + PartitionClientOptions Options; + Azure::Core::Http::Policies::RetryOptions RetryOptions; + Azure::Nullable LastReceivedOffset; + Azure::Nullable PendingError; + std::exception_ptr PendingFailure; + PendingFailureKind PendingKind{PendingFailureKind::None}; + }; + } // namespace _detail + + namespace { + void CloseReceiverStack( + std::shared_ptr<_detail::ReceiverStack> const& stack, + Azure::Core::Context const& context) + { + if (!stack) + { + return; + } + + try + { + stack->Receiver.Close(context); + } + catch (std::exception const& ex) + { + Log::Stream(Logger::Level::Warning) + << "Exception while closing a message receiver: " << ex.what(); + } + try + { + stack->Session.End(context); + } + catch (std::exception const& ex) + { + Log::Stream(Logger::Level::Warning) + << "Exception while ending a receiver session: " << ex.what(); + } + // The uAMQP connection closes when the final stack object is destroyed. Connection::Close + // is intentionally private for this backend. + } + + std::shared_ptr<_detail::ReceiverStack> CreateReceiverStack( + _detail::PartitionClientState const& state, + PartitionClientOptions const& options, + Azure::Core::Context const& context) + { + Azure::Core::Amqp::_internal::ConnectionOptions connectionOptions; + connectionOptions.ContainerId = state.ContainerId; + connectionOptions.EnableTrace = _detail::EnableAmqpTrace; + connectionOptions.AuthenticationScopes = {"https://eventhubs.azure.net/.default"}; + connectionOptions.Port = state.TargetPort; + _detail::EventHubsUtilities::SetUserAgent( + connectionOptions, state.ApplicationId, state.CppStandardVersion); + + Azure::Core::Amqp::_internal::Connection connection{ + state.FullyQualifiedNamespace, state.Credential, connectionOptions}; + + Azure::Core::Amqp::_internal::SessionOptions sessionOptions; + sessionOptions.InitialIncomingWindowSize + = static_cast((std::numeric_limits::max)()); + Azure::Core::Amqp::_internal::Session session{connection.CreateSession(sessionOptions)}; + auto receiver + = CreateMessageReceiver(session, state.PartitionUrl, state.ReceiverName, options); + auto stack = std::make_shared<_detail::ReceiverStack>( + std::move(connection), std::move(session), std::move(receiver)); + try + { + stack->Receiver.Open(context); + } + catch (...) + { + CloseReceiverStack(stack, context); + throw; + } + return stack; + } + + class ReceiveLease final { + public: + ReceiveLease( + std::shared_ptr<_detail::PartitionClientState> state, + Azure::Core::Context const& parentContext) + : m_state{std::move(state)}, m_receiveLock{m_state->ReceiveLock}, + m_childContext{parentContext.WithDeadline(parentContext.GetDeadline())} + { + std::lock_guard lock(m_state->Lock); + if (m_state->Closed) + { + throw Azure::Core::OperationCancelledException("Partition client is closed."); + } + m_state->ActiveReceive = true; + m_state->ActiveReceiveContext = m_childContext; + m_stack = m_state->Stack; + } + + ~ReceiveLease() + { + // Release the snapshot before notifying close callers that the backend call has ended. + m_stack.reset(); + { + std::lock_guard lock(m_state->Lock); + m_state->ActiveReceive = false; + m_state->ActiveReceiveContext = Azure::Core::Context{}; + } + m_state->ReceiveCondition.notify_all(); + } + + std::shared_ptr<_detail::ReceiverStack> GetStack() const { return m_stack; } + Azure::Core::Context const& GetContext() const { return m_childContext; } + + void RefreshStack() + { + std::lock_guard lock(m_state->Lock); + if (m_state->Closed) + { + throw Azure::Core::OperationCancelledException("Partition client is closed."); + } + m_stack = m_state->Stack; + } + + private: + std::shared_ptr<_detail::PartitionClientState> m_state; + std::unique_lock m_receiveLock; + Azure::Core::Context m_childContext; + std::shared_ptr<_detail::ReceiverStack> m_stack; + }; + } // namespace +#endif + +#if ENABLE_UAMQP && defined(_azure_EVENTHUBS_TEST_HOOKS) + namespace { + std::mutex PartitionClientStateCloseHookLock; + std::function PartitionClientStateCloseHook; + } // namespace +#endif + +#if ENABLE_UAMQP + void _detail::ClosePartitionClientState( + std::shared_ptr<_detail::PartitionClientState> const& state, + Azure::Core::Context const& context) + { + if (!state) + { + return; + } + +#if ENABLE_UAMQP && defined(_azure_EVENTHUBS_TEST_HOOKS) + std::function closeHook; + { + std::lock_guard lock(PartitionClientStateCloseHookLock); + closeHook = std::move(PartitionClientStateCloseHook); + } + if (closeHook) + { + closeHook(); + } +#endif + + std::shared_ptr<_detail::ReceiverStack> stackToClose; + Azure::Core::Context activeReceiveContext; + bool activeReceive = false; + { + std::lock_guard lock(state->Lock); + if (!state->Closed) + { + state->Closed = true; + ++state->Generation; + stackToClose = std::move(state->Stack); + } + activeReceive = state->ActiveReceive; + if (activeReceive) + { + activeReceiveContext = state->ActiveReceiveContext; + } + } + if (activeReceive) + { + activeReceiveContext.Cancel(); + std::unique_lock lock(state->Lock); + state->ReceiveCondition.wait(lock, [&state] { return !state->ActiveReceive; }); + } + CloseReceiverStack(stackToClose, context); + } +#endif + +#if ENABLE_UAMQP && defined(_azure_EVENTHUBS_TEST_HOOKS) + namespace _detail { + void SetPartitionClientStateCloseHook(std::function hook) + { + std::lock_guard lock(PartitionClientStateCloseHookLock); + PartitionClientStateCloseHook = std::move(hook); + } + } // namespace _detail +#endif + +#if ENABLE_UAMQP + PartitionClient _detail::PartitionClientFactory::CreatePartitionClient( + std::string fullyQualifiedNamespace, + std::shared_ptr credential, + std::uint16_t targetPort, + std::string applicationId, + long cppStandardVersion, + std::string containerId, + std::string partitionUrl, + std::string receiverName, + PartitionClientOptions options, + Azure::Core::Http::Policies::RetryOptions retryOptions, + Azure::Core::Context const& context) + { + auto state = std::make_shared<_detail::PartitionClientState>( + std::move(fullyQualifiedNamespace), + std::move(credential), + targetPort, + std::move(applicationId), + cppStandardVersion, + std::move(containerId), + std::move(partitionUrl), + std::move(receiverName), + std::move(options), + std::move(retryOptions)); + + _detail::RetryOperation retryOperation{state->RetryOptions}; + _detail::RetryOperation::AuthenticationRecoveryState authenticationState; + for (;;) + { + std::shared_ptr<_detail::ReceiverStack> candidate; + try + { + if (!retryOperation.Execute( + [&]() -> bool { + candidate = CreateReceiverStack(*state, state->Options, context); + return true; + }, + context)) + { + throw std::runtime_error("Could not create the message receiver."); + } + + { + std::lock_guard lock(state->Lock); + state->Stack = std::move(candidate); + state->Generation++; + } + return PartitionClient{std::move(state)}; + } + catch (Azure::Core::Amqp::_detail::CbsPutTokenFailedException const& failure) + { + std::chrono::milliseconds retryAfter{}; + if (!retryOperation.ShouldRetryAuthentication(authenticationState, retryAfter)) + { + failure.RethrowOriginal(); + } + _detail::RetryOperation::WaitForRetryDelay(retryAfter, context); + } + } + } +#elif ENABLE_RUST_AMQP PartitionClient _detail::PartitionClientFactory::CreatePartitionClient( Azure::Core::Amqp::_internal::Session const& session, std::string const& partitionUrl, @@ -202,7 +530,75 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::move(options), std::move(retryOptions)); } +#endif + +#if ENABLE_UAMQP + PartitionClient::PartitionClient(std::shared_ptr<_detail::PartitionClientState> state) + : m_state{std::move(state)}, + m_receiver{m_state->Stack->Receiver}, m_session{m_state->Stack->Session} + { + } + + void PartitionClient::Close(Core::Context const& context) + { + _detail::ClosePartitionClientState(m_state, context); + } + + void PartitionClient::RebuildReceiver(Core::Context const& context) + { + auto state = m_state; + PartitionClientOptions options; + std::uint64_t expectedGeneration; + std::shared_ptr<_detail::ReceiverStack> oldStack; + { + std::lock_guard lock(state->Lock); + if (state->Closed) + { + throw Azure::Core::OperationCancelledException("Partition client is closed."); + } + oldStack = std::move(state->Stack); + expectedGeneration = ++state->Generation; + options = state->Options; + options.StartPosition + = _detail::ResumeStartPosition(state->Options.StartPosition, state->LastReceivedOffset); + } + + Log::Stream(Logger::Level::Informational) + << "Rebuild the message receiver for " << state->PartitionUrl << "."; + CloseReceiverStack(oldStack, context); + auto candidate = CreateReceiverStack(*state, options, context); + auto candidateReceiver = candidate->Receiver; + auto candidateSession = candidate->Session; + bool installed = false; + { + std::lock_guard lock(state->Lock); + if (state->Closed || state->Generation != expectedGeneration) + { + // Close the candidate after releasing the state lock. + } + else + { + installed = true; + state->Stack = std::move(candidate); + ++state->Generation; + } + } + + if (!installed) + { + CloseReceiverStack(candidate, context); + throw Azure::Core::OperationCancelledException("Partition client was closed."); + } + + // Drop the copies of the old stack, or its connection stays open until the client dies. + m_receiver = std::move(candidateReceiver); + m_session = std::move(candidateSession); + + Log::Stream(Logger::Level::Informational) + << "The message receiver for " << state->PartitionUrl << " is attached again."; + } +#elif ENABLE_RUST_AMQP /** Creates a new PartitionClient * * @param messageReceiver Message Receiver for the partition client. @@ -256,6 +652,29 @@ namespace Azure { namespace Messaging { namespace EventHubs { Log::Stream(Logger::Level::Informational) << "The message receiver for " << m_partitionUrl << " is attached again."; } +#endif + + PartitionClient& PartitionClient::operator=(PartitionClient&& other) + { + if (this == &other) + { + return *this; + } + +#if ENABLE_UAMQP + _detail::ClosePartitionClientState(m_state, {}); +#endif + m_state = std::move(other.m_state); + m_receiver = std::move(other.m_receiver); + m_session = std::move(other.m_session); + m_partitionUrl = std::move(other.m_partitionUrl); + m_receiverName = std::move(other.m_receiverName); + m_lastReceivedOffset = std::move(other.m_lastReceivedOffset); + m_pendingError = std::move(other.m_pendingError); + m_partitionOptions = std::move(other.m_partitionOptions); + m_retryOptions = std::move(other.m_retryOptions); + return *this; + } PartitionClient::~PartitionClient() { @@ -263,7 +682,11 @@ namespace Azure { namespace Messaging { namespace EventHubs { { Log::Stream(Logger::Level::Verbose) << "~PartitionClient() " << "Close Receiver."; +#if ENABLE_UAMQP + _detail::ClosePartitionClientState(m_state, {}); +#elif ENABLE_RUST_AMQP m_receiver.Close(); +#endif } catch (std::exception const& ex) { @@ -272,6 +695,251 @@ namespace Azure { namespace Messaging { namespace EventHubs { } } +#if ENABLE_UAMQP + std::vector> PartitionClient::ReceiveEvents( + uint32_t maxMessages, + Core::Context const& context) + { + std::vector> messages; + auto state = m_state; + ReceiveLease lease{state, context}; + + // RetryOperation::Execute's budget never resets, so this loop keeps its own counter. + Azure::Core::Http::Policies::RetryOptions retryOptions; + { + std::lock_guard lock(state->Lock); + retryOptions = state->RetryOptions; + } + _detail::RetryOperation retryOperation{retryOptions}; + int32_t rebuildAttempt = 0; + _detail::RetryOperation::AuthenticationRecoveryState authenticationState; + + // Keep the event, and record the offset a rebuild must start after. + auto keepMessage + = [&](std::shared_ptr const& message) { + auto eventData = std::make_shared(message); + if (eventData->Offset.HasValue()) + { + std::lock_guard lock(state->Lock); + state->LastReceivedOffset = eventData->Offset.Value(); + } + rebuildAttempt = 0; + messages.push_back(eventData); + }; + + // True: the receiver works again. False: return the events held. Throws if none are held. + auto recover = [&](Azure::Core::Amqp::Models::_internal::AmqpError const* error, + bool authenticationFailure, + std::exception_ptr initialFailure) -> bool { + EventHubsException exception = error + ? _detail::EventHubsExceptionFactory::CreateEventHubsException(*error) + : EventHubsException{"Authentication failure."}; + Azure::Core::Amqp::Models::_internal::AmqpError currentError; + if (error) + { + currentError = *error; + } + std::exception_ptr originalFailure{std::move(initialFailure)}; + bool permanentFailure = false; + + for (;;) + { + std::chrono::milliseconds retryAfter{}; + bool shouldRetry = false; + if (!permanentFailure) + { + shouldRetry = authenticationFailure + ? retryOperation.ShouldRetryAuthentication(authenticationState, retryAfter) + : _detail::ShouldRebuildReceiver(exception) + && retryOperation.ShouldRetry(false, rebuildAttempt, retryAfter); + } + if (!shouldRetry) + { + if (!messages.empty()) + { + // The service will not send these again. The next call gets a new budget. + Log::Stream(Logger::Level::Warning) + << "Cannot rebuild the message receiver now. Return " << messages.size() + << " events and keep the error for the next call: " << exception.what(); + std::lock_guard lock(state->Lock); + if (error || !currentError.Condition.ToString().empty() + || !currentError.Description.empty()) + { + state->PendingError = currentError; + } + else + { + state->PendingError.Reset(); + } + state->PendingFailure + = originalFailure ? originalFailure : std::make_exception_ptr(exception); + state->PendingKind = permanentFailure + ? _detail::PendingFailureKind::Permanent + : (authenticationFailure ? _detail::PendingFailureKind::Authentication + : _detail::PendingFailureKind::Ordinary); + return false; + } + if (originalFailure) + { + std::rethrow_exception(originalFailure); + } + throw exception; + } + + if (!authenticationFailure) + { + rebuildAttempt++; + } + _detail::RetryOperation::WaitForRetryDelay(retryAfter, lease.GetContext()); + + try + { + RebuildReceiver(lease.GetContext()); + lease.RefreshStack(); + return true; + } + catch (Azure::Core::OperationCancelledException const&) + { + throw; + } + catch (EventHubsException const& rebuildFailure) + { + Log::Stream(Logger::Level::Warning) + << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); + exception = rebuildFailure; + originalFailure = nullptr; + currentError.Condition + = Azure::Core::Amqp::Models::_internal::AmqpErrorCondition{exception.ErrorCondition}; + currentError.Description = exception.ErrorDescription; + authenticationFailure = false; + permanentFailure = false; + } + catch (Azure::Core::Amqp::_detail::CbsPutTokenFailedException const& rebuildFailure) + { + Log::Stream(Logger::Level::Warning) + << "Authentication recovery failed: " << rebuildFailure.what(); + exception = EventHubsException{"Authentication failure."}; + currentError = Azure::Core::Amqp::Models::_internal::AmqpError{}; + originalFailure = rebuildFailure.GetOriginal(); + authenticationFailure = true; + permanentFailure = false; + } + catch (Azure::Core::Credentials::AuthenticationException const& rebuildFailure) + { + Log::Stream(Logger::Level::Warning) + << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); + exception = _detail::TranslateAuthenticationFailure(rebuildFailure); + originalFailure = std::current_exception(); + currentError = Azure::Core::Amqp::Models::_internal::AmqpError{}; + permanentFailure = true; + } + catch (std::exception const& rebuildFailure) + { + Log::Stream(Logger::Level::Warning) + << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); + EventHubsException translated{rebuildFailure.what()}; + translated.IsTransient = true; + exception = translated; + originalFailure = nullptr; + currentError = Azure::Core::Amqp::Models::_internal::AmqpError{}; + currentError.Description = translated.ErrorDescription; + authenticationFailure = false; + permanentFailure = false; + } + } + }; + + // No event is held yet, so this recover either works or throws. + Azure::Nullable pendingError; + std::exception_ptr pendingFailure; + _detail::PendingFailureKind pendingKind = _detail::PendingFailureKind::None; + { + std::lock_guard lock(state->Lock); + pendingKind = state->PendingKind; + pendingFailure = state->PendingFailure; + state->PendingKind = _detail::PendingFailureKind::None; + state->PendingFailure = nullptr; + if (state->PendingError.HasValue()) + { + pendingError = state->PendingError.Value(); + state->PendingError.Reset(); + } + } + if (pendingKind == _detail::PendingFailureKind::Permanent) + { + std::rethrow_exception(pendingFailure); + } + if (pendingKind != _detail::PendingFailureKind::None) + { + recover( + pendingError.HasValue() ? &pendingError.Value() : nullptr, + pendingKind == _detail::PendingFailureKind::Authentication, + std::move(pendingFailure)); + } + + while (messages.size() < maxMessages && !lease.GetContext().IsCancelled()) + { + std::pair< + std::shared_ptr, + Azure::Core::Amqp::Models::_internal::AmqpError> + result; + + // TryWaitForIncomingMessage returns two empty values if there is no data available. + auto stack = lease.GetStack(); + if (!stack) + { + std::lock_guard lock(state->Lock); + if (state->Closed) + { + throw Azure::Core::OperationCancelledException("Partition client is closed."); + } + throw std::runtime_error("Partition client has no receiver stack."); + } + result = stack->Receiver.TryWaitForIncomingMessage(); + if (result.first) + { + keepMessage(result.first); + } + else if (result.second) + { + bool const authenticationFailure = result.second.Condition + == Azure::Core::Amqp::Models::_internal::AmqpErrorCondition::UnauthorizedAccess; + if (!recover(&result.second, authenticationFailure, nullptr)) + { + break; + } + } + // If no messages have arrived, wait for one. Otherwise return the messages already held. + else if (!messages.empty()) + { + break; + } + else + { + result = stack->Receiver.WaitForIncomingMessage(lease.GetContext()); + if (result.first) + { + Log::Stream(Logger::Level::Verbose) + << "Received message. Message count now " << messages.size(); + keepMessage(result.first); + } + else if (!recover( + &result.second, + result.second.Condition + == Azure::Core::Amqp::Models::_internal::AmqpErrorCondition:: + UnauthorizedAccess, + nullptr)) + { + break; + } + } + } + Log::Stream(Logger::Level::Verbose) + << "Receive Events. Return " << messages.size() << " messages."; + + return messages; + } +#elif ENABLE_RUST_AMQP /** Receive events from the partition. * * @param maxMessages The maximum number of messages to receive. @@ -436,4 +1104,5 @@ namespace Azure { namespace Messaging { namespace EventHubs { return messages; } +#endif }}} // namespace Azure::Messaging::EventHubs diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp index 78fd0d68ac..1fb87d1dd4 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp @@ -107,6 +107,11 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail return exception.ErrorCondition != "amqp:link:message-size-exceeded"; } + inline bool IsUnauthorizedAccess(EventHubsException const& exception) + { + return exception.ErrorCondition == "amqp:unauthorized-access"; + } + // A rebuild starts after the last delivered offset, so the caller sees no duplicate // event. Before the first delivery there is no offset yet, so keep the original position. inline Models::StartPosition ResumeStartPosition( @@ -132,6 +137,20 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail class PartitionClientFactory final { public: +#if ENABLE_UAMQP + static PartitionClient CreatePartitionClient( + std::string fullyQualifiedNamespace, + std::shared_ptr credential, + std::uint16_t targetPort, + std::string applicationId, + long cppStandardVersion, + std::string containerId, + std::string partitionUrl, + std::string receiverName, + PartitionClientOptions options, + Azure::Core::Http::Policies::RetryOptions retryOptions, + Azure::Core::Context const& context); +#elif ENABLE_RUST_AMQP static PartitionClient CreatePartitionClient( Azure::Core::Amqp::_internal::Session const& session, std::string const& partitionUrl, @@ -139,9 +158,16 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail PartitionClientOptions options, Azure::Core::Http::Policies::RetryOptions retryOptions, Azure::Core::Context const& context); +#endif PartitionClientFactory() = delete; }; +#if ENABLE_UAMQP + void ClosePartitionClientState( + std::shared_ptr const& state, + Azure::Core::Context const& context); +#endif + class EventHubsPropertiesClient { public: EventHubsPropertiesClient( diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/private/retry_operation.hpp b/sdk/eventhubs/azure-messaging-eventhubs/src/private/retry_operation.hpp index 9e8f312653..3879fed249 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/private/retry_operation.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/private/retry_operation.hpp @@ -55,6 +55,11 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail double jitterFactor = -1); public: + struct AuthenticationRecoveryState final + { + bool Used{false}; + }; + // A caller with its own recovery loop uses this only for the backoff math. bool ShouldRetry( bool response, @@ -62,6 +67,15 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail std::chrono::milliseconds& retryAfter, double jitterFactor = -1); + bool ShouldRetryAuthentication( + AuthenticationRecoveryState& state, + std::chrono::milliseconds& retryAfter, + double jitterFactor = -1); + + static void WaitForRetryDelay( + std::chrono::milliseconds retryAfter, + Azure::Core::Context const& context); + explicit RetryOperation(Azure::Core::Http::Policies::RetryOptions const& retryOptions) : m_retryOptions(retryOptions) { diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/producer_client.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/producer_client.cpp index 0fc527e9a3..1e60f7ca50 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/producer_client.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/producer_client.cpp @@ -16,6 +16,7 @@ #include #include +#include #include #include #include @@ -24,10 +25,49 @@ using namespace Azure::Core::Diagnostics::_internal; using namespace Azure::Core::Diagnostics; namespace { const std::string DefaultAuthScope = "https://eventhubs.azure.net/.default"; + +#if ENABLE_UAMQP && defined(_azure_EVENTHUBS_TEST_HOOKS) +std::mutex ProducerSessionSnapshotHookLock; +std::function ProducerSessionSnapshotHook; + +void InvokeProducerSessionSnapshotHook() +{ + std::function hook; + { + std::lock_guard lock(ProducerSessionSnapshotHookLock); + hook = std::move(ProducerSessionSnapshotHook); + } + if (hook) + { + hook(); + } } +#endif +} // namespace namespace Azure { namespace Messaging { namespace EventHubs { + struct ProducerClient::ProducerCallState final + { + explicit ProducerCallState(Azure::Core::Http::Policies::RetryOptions const& retryOptions) + : Ordinary{retryOptions} + { + } + + _detail::RetryOperation Ordinary; + _detail::RetryOperation::AuthenticationRecoveryState Authentication; + }; + +#if ENABLE_UAMQP && defined(_azure_EVENTHUBS_TEST_HOOKS) + namespace _detail { + void SetProducerSessionSnapshotHook(std::function hook) + { + std::lock_guard lock(ProducerSessionSnapshotHookLock); + ProducerSessionSnapshotHook = std::move(hook); + } + } // namespace _detail +#endif + ProducerClient::ProducerClient( std::string const& connectionString, std::string const& eventHub, @@ -106,7 +146,16 @@ namespace Azure { namespace Messaging { namespace EventHubs { EventDataBatchOptions const& options, Core::Context const& context) { - EstablishSenderWithRetry(options.PartitionId, context); + ProducerCallState callState{m_producerClientOptions.RetryOptions}; + return CreateBatch(options, context, callState); + } + + EventDataBatch ProducerClient::CreateBatch( + EventDataBatchOptions const& options, + Core::Context const& context, + ProducerCallState& callState) + { + EstablishSenderWithRetry(options.PartitionId, context, callState); EventDataBatchOptions optionsToUse{options}; if (!options.MaxBytes.HasValue()) @@ -142,7 +191,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { << (options.PartitionId.empty() ? std::string("") : options.PartitionId) << "'. Discard the stack and build it again: " << ex.what() << std::endl; InvalidateSender(options.PartitionId, observedGeneration, context); - EstablishSenderWithRetry(options.PartitionId, context); + EstablishSenderWithRetry(options.PartitionId, context, callState); std::uint64_t rebuiltGeneration = 0; optionsToUse.MaxBytes = readMaxMessageSize(rebuiltGeneration); } @@ -152,68 +201,78 @@ namespace Azure { namespace Messaging { namespace EventHubs { } void ProducerClient::Send(EventDataBatch const& eventDataBatch, Core::Context const& context) + { + ProducerCallState callState{m_producerClientOptions.RetryOptions}; + Send(eventDataBatch, context, callState); + } + + void ProducerClient::Send( + EventDataBatch const& eventDataBatch, + Core::Context const& context, + ProducerCallState& callState) { auto message = eventDataBatch.ToAmqpMessage(); - Azure::Messaging::EventHubs::_detail::RetryOperation retryOp( - m_producerClientOptions.RetryOptions); // Defense in depth: RetryOperation::Execute rethrows the last exception when retries // are exhausted, but if the lambda ever returns false directly the batch must not be // silently dropped. See issue #7130. auto const& partitionId = eventDataBatch.GetPartitionId(); - if (!retryOp.Execute( - [&]() -> bool { - EnsureSenderOrInvalidate(partitionId, context); - std::uint64_t observedGeneration = 0; - auto& guard = GetPartitionGuard(partitionId); - try - { - // Keeps a teardown off the sender copy; sends still run together. - std::shared_lock stackLock(guard.stackLock); - auto sender = GetSender(partitionId); - observedGeneration = guard.generation.load(); - auto result = sender.Send(message, context); + bool transferUnauthorized = false; + auto send = [&]() -> bool { + transferUnauthorized = false; + EnsureSenderOrInvalidate(partitionId, context); + std::uint64_t observedGeneration = 0; + auto& guard = GetPartitionGuard(partitionId); + try + { + // Keeps a teardown off the sender copy; sends still run together. + std::shared_lock stackLock(guard.stackLock); + auto sender = GetSender(partitionId); + observedGeneration = guard.generation.load(); + auto result = sender.Send(message, context); #if ENABLE_UAMQP - auto sendStatus = std::get<0>(result); - if (sendStatus == Azure::Core::Amqp::_internal::MessageSendStatus::Ok) - { - return true; - } - // Throw an exception about the error we just received. - throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: - CreateEventHubsException(std::get<1>(result)); + auto sendStatus = std::get<0>(result); + if (sendStatus == Azure::Core::Amqp::_internal::MessageSendStatus::Ok) + { + return true; + } + // Throw an exception about the error we just received. + auto transferException = Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: + CreateEventHubsException(std::get<1>(result)); + transferUnauthorized = _detail::IsUnauthorizedAccess(transferException); + throw transferException; #elif ENABLE_RUST_AMQP - if (result) - { - throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: - CreateEventHubsException(result); - } - return true; + if (result) + { + throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: + CreateEventHubsException(result); + } + return true; #endif - } - catch (Azure::Core::OperationCancelledException const&) - { - throw; - } - catch (Azure::Messaging::EventHubs::EventHubsException const& ex) - { - if (!context.IsCancelled() && _detail::ShouldInvalidateSender(ex)) - { - InvalidateSender(partitionId, observedGeneration, context); - } - throw; - } - catch (std::exception const&) - { - if (!context.IsCancelled()) - { - InvalidateSender(partitionId, observedGeneration, context); - } - throw; - } - }, - context)) - { + } + catch (Azure::Core::OperationCancelledException const&) + { + throw; + } + catch (Azure::Messaging::EventHubs::EventHubsException const& ex) + { + if (!context.IsCancelled() && _detail::ShouldInvalidateSender(ex)) + { + InvalidateSender(partitionId, observedGeneration, context); + } + throw; + } + catch (std::exception const&) + { + if (!context.IsCancelled()) + { + InvalidateSender(partitionId, observedGeneration, context); + } + throw; + } + }; + + auto throwRetriesExhausted = [&]() { std::string failureDetail = "ProducerClient::Send failed after exhausting " + std::to_string(m_producerClientOptions.RetryOptions.MaxRetries) + " retry attempts (partition='" @@ -224,24 +283,70 @@ namespace Azure { namespace Messaging { namespace EventHubs { ex.ErrorCondition = "eventhubs:client:retries-exhausted"; ex.IsTransient = true; throw ex; + }; + +#if ENABLE_UAMQP + while (true) + { + try + { + if (!callState.Ordinary.Execute(send, context)) + { + throwRetriesExhausted(); + } + return; + } + catch (Azure::Core::Amqp::_detail::CbsPutTokenFailedException const& failure) + { + std::chrono::milliseconds retryAfter{}; + if (!callState.Ordinary.ShouldRetryAuthentication(callState.Authentication, retryAfter)) + { + failure.RethrowOriginal(); + } + Azure::Messaging::EventHubs::_detail::RetryOperation::WaitForRetryDelay( + retryAfter, context); + } + catch (Azure::Messaging::EventHubs::EventHubsException const& ex) + { + if (!transferUnauthorized || !_detail::IsUnauthorizedAccess(ex)) + { + throw; + } + + std::chrono::milliseconds retryAfter{}; + if (!callState.Ordinary.ShouldRetryAuthentication(callState.Authentication, retryAfter)) + { + throw; + } + Azure::Messaging::EventHubs::_detail::RetryOperation::WaitForRetryDelay( + retryAfter, context); + } + } +#else + if (!callState.Ordinary.Execute(send, context)) + { + throwRetriesExhausted(); } +#endif } void ProducerClient::Send(Models::EventData const& eventData, Core::Context const& context) { - auto batch = CreateBatch(EventDataBatchOptions{}, context); + ProducerCallState callState{m_producerClientOptions.RetryOptions}; + auto batch = CreateBatch(EventDataBatchOptions{}, context, callState); if (!batch.TryAdd(eventData)) { throw std::runtime_error("Could not add message to batch."); } - Send(batch, context); + Send(batch, context, callState); } void ProducerClient::Send( std::vector const& eventData, Core::Context const& context) { - auto batch = CreateBatch(EventDataBatchOptions{}, context); + ProducerCallState callState{m_producerClientOptions.RetryOptions}; + auto batch = CreateBatch(EventDataBatchOptions{}, context, callState); for (const auto& data : eventData) { if (!batch.TryAdd(data)) @@ -249,7 +354,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { throw std::runtime_error("Could not add message to batch."); } } - Send(batch, context); + Send(batch, context, callState); } Azure::Core::Amqp::_internal::Connection ProducerClient::CreateConnection( @@ -312,6 +417,93 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::string const& partitionId, Azure::Core::Context const& context) { +#if ENABLE_UAMQP + auto& guard = GetPartitionGuard(partitionId); + std::shared_lock stackLock(guard.stackLock); + auto const observedGeneration = guard.generation.load(); + { + std::lock_guard lock(m_sendersLock); + if (m_senders.find(partitionId) != m_senders.end()) + { + return; + } + } + + EnsureSession(partitionId, context); + auto session = GetSession(partitionId); + stackLock.unlock(); + +#if defined(_azure_EVENTHUBS_TEST_HOOKS) + InvokeProducerSessionSnapshotHook(); +#endif + + std::string targetUrl{m_targetUrl}; + if (!partitionId.empty()) + { + targetUrl += "/Partitions/" + partitionId; + } + + Azure::Core::Amqp::_internal::MessageSenderOptions senderOptions; + senderOptions.Name = m_producerClientOptions.Name; + senderOptions.EnableTrace = _detail::EnableAmqpTrace; + senderOptions.MaxMessageSize = m_producerClientOptions.MaxMessageSize; + + // Copy the session before opening the sender. No client map lock may span network work. + auto sender = session.CreateMessageSender(targetUrl, senderOptions); + auto openResult{sender.Open(context)}; + if (openResult) + { + Azure::Core::Diagnostics::_internal::Log::Stream( + Azure::Core::Diagnostics::Logger::Level::Error) + << "Failed to create message sender: " << openResult; + throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: + CreateEventHubsException(openResult); + } + + bool discardCandidate = false; + bool staleWithoutSender = false; + { + // Keep the partition stack lock before the sender map lock. Invalidation uses the same + // order, so a candidate cannot be installed after its stack was removed. + std::unique_lock stackLock(guard.stackLock); + std::lock_guard sendersLock(m_sendersLock); + if (guard.generation.load() != observedGeneration) + { + discardCandidate = true; + staleWithoutSender = m_senders.find(partitionId) == m_senders.end(); + } + else if (m_senders.find(partitionId) != m_senders.end()) + { + discardCandidate = true; + } + else + { + m_senders.emplace(partitionId, std::move(sender)); + guard.generation.fetch_add(1); + return; + } + } + + if (discardCandidate) + { + try + { + sender.Close(context); + } + catch (std::exception const& ex) + { + Log::Stream(Logger::Level::Warning) + << "Exception while closing a discarded message sender: " << ex.what(); + } + } + if (staleWithoutSender) + { + EventHubsException staleStack{ + "The message sender stack changed while the sender was being established."}; + staleStack.IsTransient = true; + throw staleStack; + } +#else std::unique_lock lock(m_sendersLock); if (m_senders.find(partitionId) == m_senders.end()) { @@ -342,6 +534,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { m_senders.emplace(partitionId, std::move(sender)); GetPartitionGuard(partitionId).generation.fetch_add(1); } +#endif } void ProducerClient::EnsureSenderOrInvalidate( std::string const& partitionId, @@ -369,19 +562,40 @@ namespace Azure { namespace Messaging { namespace EventHubs { } // Establishing the stack resolves the host, opens the socket, negotiates TLS and runs the CBS - // handshake, so it is the step most exposed to a transient transport failure. `Send` runs under - // `RetryOperation`; the batch path did not, so a burst after an idle period lost every event - // whose stack failed to build. - // - // Retry once, only for CbsOpenResult::Error - see CbsOpenFailedException. The bound is one - // attempt because uAMQP logs the transport reason but returns no value carrying it, so `Error` - // cannot separate a transient failure from a permanent one; do not make this a loop. - // `EnsureSenderOrInvalidate` invalidates before it rethrows, so the retry builds a new - // connection rather than reusing a socket a failed open may have left non-closed. + // handshake, so it is the step most exposed to a transient transport failure. Keep ordinary + // transport retries separate from the one-shot authentication recovery. A failed sender stack + // is invalidated before either retry builds a new connection. void ProducerClient::EstablishSenderWithRetry( std::string const& partitionId, - Azure::Core::Context const& context) + Azure::Core::Context const& context, + ProducerCallState& callState) { +#if ENABLE_UAMQP + auto establish = [&]() -> bool { + EnsureSenderOrInvalidate(partitionId, context); + return true; + }; + + while (true) + { + try + { + static_cast(callState.Ordinary.Execute(establish, context)); + return; + } + catch (Azure::Core::Amqp::_detail::CbsPutTokenFailedException const& failure) + { + std::chrono::milliseconds retryAfter{}; + if (!callState.Ordinary.ShouldRetryAuthentication(callState.Authentication, retryAfter)) + { + failure.RethrowOriginal(); + } + Azure::Messaging::EventHubs::_detail::RetryOperation::WaitForRetryDelay( + retryAfter, context); + } + } +#else + (void)callState; try { EnsureSenderOrInvalidate(partitionId, context); @@ -403,6 +617,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { << std::endl; EnsureSenderOrInvalidate(partitionId, context); } +#endif } Azure::Core::Amqp::_internal::MessageSender ProducerClient::GetSender( diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/retry_operation.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/retry_operation.cpp index a45f57d83a..90308ad175 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/retry_operation.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/retry_operation.cpp @@ -4,6 +4,7 @@ #include "azure/messaging/eventhubs/eventhubs_exception.hpp" +#include #include #include @@ -14,7 +15,9 @@ namespace { constexpr std::chrono::milliseconds CancellationCheckInterval{100}; -void WaitForRetryDelay(std::chrono::milliseconds retryAfter, Azure::Core::Context const& context) +void WaitForRetryDelayImpl( + std::chrono::milliseconds retryAfter, + Azure::Core::Context const& context) { auto const deadline = std::chrono::steady_clock::now() + retryAfter; while (true) @@ -76,6 +79,28 @@ bool Azure::Messaging::EventHubs::_detail::RetryOperation::Execute( { throw; } +#if ENABLE_UAMQP + catch (Azure::Core::Amqp::_detail::CbsPutTokenFailedException const&) + { + throw; + } +#endif +#if ENABLE_UAMQP + // Only CbsOpenResult::Error can be transient. uAMQP gives no value that separates a + // transient open failure from a permanent one, so MaxRetries is the only bound. + catch (Azure::Core::Amqp::_detail::CbsOpenFailedException const& e) + { + context.ThrowIfCancelled(); + if (e.Result != Azure::Core::Amqp::_detail::CbsOpenResult::Error) + { + throw; + } + if (!ShouldRetry(false, retryCount, retryAfter)) + { + throw; + } + } +#endif catch (std::runtime_error const& e) { context.ThrowIfCancelled(); @@ -91,10 +116,32 @@ bool Azure::Messaging::EventHubs::_detail::RetryOperation::Execute( } ++retryCount; - WaitForRetryDelay(retryAfter, context); + WaitForRetryDelayImpl(retryAfter, context); } } +bool Azure::Messaging::EventHubs::_detail::RetryOperation::ShouldRetryAuthentication( + AuthenticationRecoveryState& state, + std::chrono::milliseconds& retryAfter, + double jitterFactor) +{ + if (state.Used || m_retryOptions.MaxRetries <= 0) + { + return false; + } + + state.Used = true; + retryAfter = CalculateExponentialDelay(1, jitterFactor); + return true; +} + +void Azure::Messaging::EventHubs::_detail::RetryOperation::WaitForRetryDelay( + std::chrono::milliseconds retryAfter, + Azure::Core::Context const& context) +{ + WaitForRetryDelayImpl(retryAfter, context); +} + bool Azure::Messaging::EventHubs::_detail::RetryOperation::ShouldRetry( bool response, int32_t attempt, diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt index b21bcfe95b..e9814e71b0 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt @@ -15,8 +15,7 @@ include(TestProxyPrep) SetUpTestProxy("sdk/eventhubs") ################## Unit Tests ########################## -add_executable ( - azure-messaging-eventhubs-test +set(EVENTHUBS_TEST_SOURCES azure_messaging_eventhubs_test.cpp checkpoint_store_test.cpp connection_string_test.cpp @@ -36,8 +35,19 @@ add_executable ( test_checkpoint_store.hpp ) +if (NOT USE_RUST_AMQP) + list(APPEND EVENTHUBS_TEST_SOURCES auth_recovery_test.cpp) +endif() + +add_executable (azure-messaging-eventhubs-test ${EVENTHUBS_TEST_SOURCES}) + target_compile_definitions(azure-messaging-eventhubs-test PRIVATE _azure_BUILDING_TESTS) +if (NOT USE_RUST_AMQP) + target_compile_definitions(azure-messaging-eventhubs-test PRIVATE _azure_TESTING_BUILD) + target_compile_definitions(azure-messaging-eventhubs PRIVATE _azure_EVENTHUBS_TEST_HOOKS) +endif() + create_per_service_target_build(eventhubs azure-messaging-eventhubs-test) create_map_file(azure-messaging-eventhubs-test azure-messaging-eventhubs-test.map) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/auth_recovery_test.cpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/auth_recovery_test.cpp new file mode 100644 index 0000000000..0a334b0d6d --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/auth_recovery_test.cpp @@ -0,0 +1,1021 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#include "../../../../core/azure-core-amqp/test/ut/mock_amqp_server.hpp" +#include "eventhubs_test_base.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { + void SetProducerSessionSnapshotHook(std::function hook); + void SetPartitionClientStateCloseHook(std::function hook); + + class ConsumerClientTestAccess final { + public: + static std::size_t PartitionClientStateCount(ConsumerClient& consumer) + { + std::lock_guard lock(consumer.m_partitionClientStatesLock); + return consumer.m_partitionClientStates.size(); + } + + static bool PartitionClientStateExpired(ConsumerClient& consumer, std::size_t index) + { + std::lock_guard lock(consumer.m_partitionClientStatesLock); + return consumer.m_partitionClientStates.at(index).expired(); + } + }; +}}}} // namespace Azure::Messaging::EventHubs::_detail + +#if defined(AZ_PLATFORM_POSIX) +#include + +#include +#include +#elif defined(AZ_PLATFORM_WINDOWS) +#define WIN32_LEAN_AND_MEAN +#define NOMINMAX +#include +#endif + +namespace Azure { namespace Core { namespace Amqp { namespace Tests { + + uint16_t FindAvailableSocket() + { + auto state = Azure::Core::Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance(); + (void)state; + + for (uint32_t port = 45000; port != 46000; ++port) + { +#if defined(AZ_PLATFORM_WINDOWS) + auto socketHandle = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (socketHandle == INVALID_SOCKET) + { + continue; + } +#else + auto socketHandle = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (socketHandle < 0) + { + continue; + } +#endif + + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + address.sin_port = htons(static_cast(port)); + auto const result + = bind(socketHandle, reinterpret_cast(&address), sizeof(address)); +#if defined(AZ_PLATFORM_WINDOWS) + closesocket(socketHandle); +#else + close(socketHandle); +#endif + if (result == 0) + { + return static_cast(port); + } + } + + throw std::runtime_error("Could not find a free test socket."); + } + +}}}} // namespace Azure::Core::Amqp::Tests + +namespace Azure { namespace Messaging { namespace EventHubs { namespace Test { + namespace { + using Azure::Core::Amqp::_internal::Connection; + using Azure::Core::Amqp::_internal::ConnectionOptions; + using Azure::Core::Amqp::_internal::MessageReceiver; + using Azure::Core::Amqp::_internal::MessageSender; + using Azure::Core::Amqp::_internal::Session; + using Azure::Core::Amqp::_internal::SessionRole; + using Azure::Core::Amqp::Models::AmqpMessage; + using Azure::Core::Amqp::Models::AmqpSymbol; + using Azure::Core::Amqp::Models::AmqpValue; + using Azure::Core::Amqp::Models::_internal::AmqpError; + using Azure::Core::Amqp::Models::_internal::AmqpErrorCondition; + using Azure::Core::Amqp::Tests::MessageTests::AmqpServerMock; + using Azure::Core::Amqp::Tests::MessageTests::MockServiceEndpoint; + using Azure::Core::Amqp::Tests::MessageTests::MockServiceEndpointOptions; + + Azure::Core::Http::Policies::RetryOptions FastRetryOptions(int32_t maxRetries = 1) + { + Azure::Core::Http::Policies::RetryOptions options; + options.MaxRetries = maxRetries; + options.RetryDelay = std::chrono::milliseconds(1); + options.MaxRetryDelay = std::chrono::milliseconds(2); + return options; + } + + Azure::Messaging::EventHubs::EventDataBatchOptions BatchOptions() + { + Azure::Messaging::EventHubs::EventDataBatchOptions options; + options.MaxBytes = 1024; + options.PartitionId = "0"; + return options; + } + + class CbsScript final { + public: + std::atomic OpenFailures{0}; + std::atomic PutTokenFailures{0}; + std::atomic OpenAttempts{0}; + std::atomic PutTokenAttempts{0}; + }; + + class EventScript final { + public: + std::atomic TransferFailures{0}; + std::atomic TransferAttempts{0}; + std::atomic AcceptedTransfers{0}; + std::atomic DeliveryLinks{0}; + std::atomic DeliveryNumber{0}; + std::atomic DetachesSent{0}; + bool DeliverEvents{false}; + std::mutex FilterLock; + std::vector ReceiverFilters; + }; + + std::string SelectorFilter(Azure::Core::Amqp::Models::_internal::MessageSource const& source) + { + auto filter = source.GetFilter(); + auto const selector = filter.find(AmqpSymbol{"apache.org:selector-filter:string"}); + if (selector == filter.end()) + { + return {}; + } + return static_cast(selector->second.AsDescribed().GetValue()); + } + + template bool WaitUntil(Predicate predicate) + { + auto const deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (!predicate()) + { + if (std::chrono::steady_clock::now() >= deadline) + { + return false; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + return true; + } + + // Clears a test hook when the scope ends, so a failed assertion cannot leave it armed for + // the next test. + class HookGuard final { + public: + HookGuard(void (*setter)(std::function), std::function hook) + : m_setter{setter} + { + m_setter(std::move(hook)); + } + ~HookGuard() { m_setter({}); } + HookGuard(HookGuard const&) = delete; + HookGuard& operator=(HookGuard const&) = delete; + + private: + void (*m_setter)(std::function); + }; + + bool Consume(std::atomic& count) + { + auto current = count.load(); + while (current > 0 && !count.compare_exchange_weak(current, current - 1)) + { + } + return current > 0; + } + + class ScriptedCbsEndpoint final : public MockServiceEndpoint { + public: + ScriptedCbsEndpoint( + MockServiceEndpointOptions const& options, + std::shared_ptr script) + : MockServiceEndpoint("$cbs", options), m_script{std::move(script)} + { + } + + bool OnLinkAttached( + Session const& session, + std::string const& linkName, + Azure::Core::Amqp::_internal::LinkEndpoint& linkEndpoint, + SessionRole role, + Azure::Core::Amqp::Models::_internal::MessageSource const& source, + Azure::Core::Amqp::Models::_internal::MessageTarget const& target) override + { + if (role == SessionRole::Receiver) + { + ++m_script->OpenAttempts; + if (Consume(m_script->OpenFailures)) + { + AmqpError error; + error.Condition = AmqpErrorCondition::InternalError; + error.Description = "CBS open failed"; + DetachLink(session, linkEndpoint, true, error); + return false; + } + } + return MockServiceEndpoint::OnLinkAttached( + session, linkName, linkEndpoint, role, source, target); + } + + private: + void MessageReceived(std::string const&, std::shared_ptr const& message) override + { + auto const operation + = static_cast(message->ApplicationProperties.at("operation")); + if (operation != "put-token") + { + return; + } + + ++m_script->PutTokenAttempts; + bool const failed = Consume(m_script->PutTokenFailures); + AmqpMessage response; + auto correlationId = message->Properties.CorrelationId; + if (correlationId.IsNull()) + { + correlationId = message->Properties.MessageId; + } + response.Properties.CorrelationId = correlationId; + response.ApplicationProperties["status-code"] = failed ? 401 : 200; + response.ApplicationProperties["status-description"] + = failed ? "CBS PutToken failed" : "OK-put"; + response.SetBody(AmqpValue{}); + + auto const result = GetMessageSender().Send(response, GetListenerContext()); + if (std::get<0>(result) != Azure::Core::Amqp::_internal::MessageSendStatus::Ok) + { + GTEST_LOG_(INFO) << "Failed to send scripted CBS response: " << std::get<1>(result); + } + } + + std::shared_ptr m_script; + }; + + class EventHubEndpoint final : public MockServiceEndpoint { + public: + EventHubEndpoint( + std::string name, + MockServiceEndpointOptions const& options, + std::shared_ptr script) + : MockServiceEndpoint(std::move(name), options), m_script{std::move(script)} + { + } + + ~EventHubEndpoint() override + { + for (auto& worker : m_deliveryWorkers) + { + if (worker.joinable()) + { + worker.join(); + } + } + } + + bool OnLinkAttached( + Session const& session, + std::string const& linkName, + Azure::Core::Amqp::_internal::LinkEndpoint& linkEndpoint, + SessionRole role, + Azure::Core::Amqp::Models::_internal::MessageSource const& source, + Azure::Core::Amqp::Models::_internal::MessageTarget const& target) override + { + // The base call hands the endpoint to the new link, which owns it from then on. A + // worker that detaches later needs the raw handle, because the wrapper is emptied. + auto* const endpointHandle = linkEndpoint.Get(); + if (role == SessionRole::Receiver) + { + std::lock_guard lock(m_script->FilterLock); + m_script->ReceiverFilters.push_back(SelectorFilter(source)); + } + auto const attached = MockServiceEndpoint::OnLinkAttached( + session, linkName, linkEndpoint, role, source, target); + if (attached && role == SessionRole::Receiver && m_script->DeliverEvents) + { + ++m_script->DeliveryLinks; + m_deliveryWorkers.emplace_back([this, session, endpointHandle, linkName]() { + Deliver(session, endpointHandle, linkName); + }); + } + return attached; + } + + protected: + AmqpValue OnMessageReceived(MessageReceiver const&, std::shared_ptr const&) + override + { + ++m_script->TransferAttempts; + if (Consume(m_script->TransferFailures)) + { + return Azure::Core::Amqp::Models::_internal::Messaging::DeliveryRejected( + "amqp:unauthorized-access", "stale transfer", {}); + } + ++m_script->AcceptedTransfers; + return Azure::Core::Amqp::Models::_internal::Messaging::DeliveryAccepted(); + } + + private: + void MessageReceived(std::string const&, std::shared_ptr const&) override {} + + void Deliver( + Session const& session, + LINK_ENDPOINT_INSTANCE_TAG* endpointHandle, + std::string const& linkName) + { + while (!GetListenerContext().IsCancelled() && !HasMessageSender(linkName)) + { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + if (GetListenerContext().IsCancelled()) + { + return; + } + + auto sender = GetMessageSender(linkName); + auto const delivery = ++m_script->DeliveryNumber; + auto const offset = delivery == 1 ? "10" : "11"; + AmqpMessage message; + message.MessageAnnotations[AmqpSymbol{"x-opt-offset"}] = AmqpValue{offset}; + message.SetBody(AmqpValue{"event"}); + auto const sendResult = sender.Send(message, GetListenerContext()); + if (std::get<0>(sendResult) != Azure::Core::Amqp::_internal::MessageSendStatus::Ok) + { + GTEST_LOG_(INFO) << "Failed to send scripted event: " << std::get<1>(sendResult); + return; + } + + if (delivery == 1) + { + AmqpError error; + error.Condition = AmqpErrorCondition::UnauthorizedAccess; + error.Description = "stale receive"; + auto endpoint + = Azure::Core::Amqp::_detail::LinkEndpointFactory::CreateLinkEndpoint(endpointHandle); + DetachLink(session, endpoint, true, error); + ++m_script->DetachesSent; + } + } + + std::shared_ptr m_script; + std::vector m_deliveryWorkers; + }; + + class AuthRecoveryServer final { + public: + AuthRecoveryServer( + int openFailures = 0, + int putTokenFailures = 0, + int transferFailures = 0, + bool deliverEvents = false) + : m_port{Azure::Core::Amqp::Tests::FindAvailableSocket()}, + m_server{m_port, testing::UnitTest::GetInstance()->current_test_info()->name(), false}, + m_cbsScript{std::make_shared()}, m_eventScript{ + std::make_shared()} + { + m_cbsScript->OpenFailures = openFailures; + m_cbsScript->PutTokenFailures = putTokenFailures; + m_eventScript->TransferFailures = transferFailures; + m_eventScript->DeliverEvents = deliverEvents; + + MockServiceEndpointOptions endpointOptions; + endpointOptions.ListenerContext = m_server.GetListenerContext(); + m_server.AddServiceEndpoint( + std::make_shared(endpointOptions, m_cbsScript)); + m_server.AddServiceEndpoint(std::make_shared( + ProducerPartitionEndpoint(), endpointOptions, m_eventScript)); + m_server.AddServiceEndpoint(std::make_shared( + ProducerGatewayEndpoint(), endpointOptions, m_eventScript)); + m_server.AddServiceEndpoint(std::make_shared( + ConsumerPartitionEndpoint(), endpointOptions, m_eventScript)); + } + + ~AuthRecoveryServer() { Stop(); } + + void Start() + { + if (!m_started) + { + m_server.StartListening(); + m_started = true; + } + } + + void Stop() + { + if (m_started) + { + m_server.StopListening(); + m_started = false; + } + } + + uint16_t Port() const { return m_port; } + std::size_t ConnectionCount() const { return m_server.GetConnectionCount(); } + int PutTokenAttempts() const { return m_cbsScript->PutTokenAttempts.load(); } + int CbsOpenAttempts() const { return m_cbsScript->OpenAttempts.load(); } + int TransferAttempts() const { return m_eventScript->TransferAttempts.load(); } + int AcceptedTransfers() const { return m_eventScript->AcceptedTransfers.load(); } + int DeliveryLinks() const { return m_eventScript->DeliveryLinks.load(); } + int DetachesSent() const { return m_eventScript->DetachesSent.load(); } + std::vector ReceiverFilters() const + { + std::lock_guard lock(m_eventScript->FilterLock); + return m_eventScript->ReceiverFilters; + } + + // The mock sends the first event and the unauthorized detach as soon as the receiver + // attaches. A receive that starts after both frames arrived sees the event first and + // then the error in one call, which is the partial delivery the tests need. + bool WaitForFirstDetach() const + { + if (!WaitUntil([this]() { return DetachesSent() >= 1; })) + { + return false; + } + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + return true; + } + + void SetPutTokenFailures(int failures) { m_cbsScript->PutTokenFailures = failures; } + + std::string ConnectionString() const + { + return "Endpoint=sb://127.0.0.1:" + std::to_string(m_port) + + "/;SharedAccessKeyName=TestKey;SharedAccessKey=abcdabcd;EntityPath=eh;" + "UseDevelopmentEmulator=true"; + } + + std::string ProducerPartitionEndpoint() const + { + return "amqp://127.0.0.1:" + std::to_string(m_port) + "/eh/Partitions/0"; + } + + std::string ProducerGatewayEndpoint() const + { + return "amqp://127.0.0.1:" + std::to_string(m_port) + "/eh"; + } + + std::string ConsumerPartitionEndpoint() const + { + // The consumer client builds its partition URL without the port. + return "amqp://127.0.0.1/eh/ConsumerGroups/$Default/Partitions/0"; + } + + private: + uint16_t m_port; + AmqpServerMock m_server; + std::shared_ptr m_cbsScript; + std::shared_ptr m_eventScript; + bool m_started{false}; + }; + + class FailingCredential final : public Azure::Core::Credentials::TokenCredential { + public: + FailingCredential() : TokenCredential("FailingCredential") {} + + Azure::Core::Credentials::AccessToken GetToken( + Azure::Core::Credentials::TokenRequestContext const&, + Azure::Core::Context const&) const override + { + ++m_attempts; + throw Azure::Core::Credentials::AuthenticationException("credential failure"); + } + + int Attempts() const { return m_attempts.load(); } + + private: + mutable std::atomic m_attempts{0}; + }; + + } // anonymous namespace + + class AuthRecoveryTest : public EventHubsTestBase { + protected: + void SetUp() override + { + EventHubsTestBase::SetUp(); +#if defined(AZ_PLATFORM_MAC) + GTEST_SKIP() << "The uAMQP socket client tests are not supported on Apple platforms."; +#endif + } + }; + + TEST_F(AuthRecoveryTest, ProducerCreateBatchRecoversPutTokenAuthenticationWithFreshStack) + { + AuthRecoveryServer server(0, 1); + server.Start(); + + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ProducerClient producer(server.ConnectionString(), "", options); + + EXPECT_NO_THROW(producer.CreateBatch(BatchOptions())); + EXPECT_EQ(2U, server.ConnectionCount()); + EXPECT_EQ(2, server.PutTokenAttempts()); + } + + TEST_F(AuthRecoveryTest, ProducerSendRecoversStaleUnauthorizedWithOneFreshStack) + { + AuthRecoveryServer server(0, 0, 1); + server.Start(); + + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ProducerClient producer(server.ConnectionString(), "", options); + auto batch = producer.CreateBatch(BatchOptions()); + ASSERT_TRUE(batch.TryAdd(Models::EventData{"payload"})); + + EXPECT_NO_THROW(producer.Send(batch)); + EXPECT_EQ(2U, server.ConnectionCount()); + EXPECT_EQ(2, server.TransferAttempts()); + EXPECT_EQ(1, server.AcceptedTransfers()); + } + + TEST_F(AuthRecoveryTest, ProducerCloseAtSessionSnapshotPreservesRetryContract) + { + AuthRecoveryServer server; + server.Start(); + + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ProducerClient producer(server.ConnectionString(), "", options); + + std::atomic hookCalled{false}; + std::exception_ptr closeFailure; + HookGuard snapshotHook{_detail::SetProducerSessionSnapshotHook, [&]() { + hookCalled = true; + std::thread closeThread([&]() { + try + { + producer.Close(); + } + catch (...) + { + closeFailure = std::current_exception(); + } + }); + closeThread.join(); + }}; + + std::exception_ptr operationFailure; + try + { + producer.CreateBatch(BatchOptions()); + } + catch (...) + { + operationFailure = std::current_exception(); + } + + ASSERT_TRUE(hookCalled.load()); + if (closeFailure) + { + std::rethrow_exception(closeFailure); + } + if (operationFailure) + { + try + { + std::rethrow_exception(operationFailure); + } + catch (std::out_of_range const&) + { + ADD_FAILURE() << "Producer session invalidation escaped as std::out_of_range."; + } + catch (EventHubsException const& exception) + { + ADD_FAILURE() << "Producer retry escaped as EventHubsException: " << exception.what(); + } + catch (std::exception const& exception) + { + ADD_FAILURE() << "Producer retry escaped as an unexpected exception: " << exception.what(); + } + } + + EXPECT_FALSE(operationFailure); + EXPECT_EQ(2U, server.ConnectionCount()); + } + + TEST_F(AuthRecoveryTest, ProducerConvenienceSendSharesOneRecoveryBudgetAcrossBatchAndTransfer) + { + AuthRecoveryServer server(0, 1, 1); + server.Start(); + + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ProducerClient producer(server.ConnectionString(), "", options); + EventHubsException failure{"no failure"}; + try + { + producer.Send(Models::EventData{"payload"}); + ADD_FAILURE() << "Expected unauthorized transfer after the authentication budget was used."; + } + catch (EventHubsException const& exception) + { + failure = exception; + } + + EXPECT_EQ(2U, server.ConnectionCount()); + EXPECT_EQ(2, server.PutTokenAttempts()); + EXPECT_EQ(1, server.TransferAttempts()); + EXPECT_EQ("amqp:unauthorized-access", failure.ErrorCondition); + EXPECT_EQ("stale transfer", failure.ErrorDescription); + EXPECT_FALSE(failure.IsTransient); + } + + TEST_F(AuthRecoveryTest, ConsumerCreatePartitionClientRecoversPutToken) + { + AuthRecoveryServer server(0, 1); + server.Start(); + + ConsumerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ConsumerClient consumer(server.ConnectionString(), "", DefaultConsumerGroup, options); + + auto partition = consumer.CreatePartitionClient("0"); + EXPECT_TRUE(partition.ReceiveEvents(0).empty()); + EXPECT_EQ(2U, server.ConnectionCount()); + EXPECT_EQ(2, server.PutTokenAttempts()); + } + + TEST_F(AuthRecoveryTest, ConsumerPartitionRegistryExpiresPrunesAndClosesStates) + { + AuthRecoveryServer server; + server.Start(); + + ConsumerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ConsumerClient consumer(server.ConnectionString(), "", DefaultConsumerGroup, options); + + { + auto context = Azure::Core::Context{Azure::DateTime::clock::now() + std::chrono::seconds(5)}; + auto partition = consumer.CreatePartitionClient("0", {}, context); + EXPECT_EQ(1U, _detail::ConsumerClientTestAccess::PartitionClientStateCount(consumer)); + EXPECT_FALSE(_detail::ConsumerClientTestAccess::PartitionClientStateExpired(consumer, 0)); + } + + EXPECT_EQ(1U, _detail::ConsumerClientTestAccess::PartitionClientStateCount(consumer)); + EXPECT_TRUE(_detail::ConsumerClientTestAccess::PartitionClientStateExpired(consumer, 0)); + + auto context = Azure::Core::Context{Azure::DateTime::clock::now() + std::chrono::seconds(5)}; + auto partition = consumer.CreatePartitionClient("0", {}, context); + EXPECT_EQ(1U, _detail::ConsumerClientTestAccess::PartitionClientStateCount(consumer)); + + std::atomic closeHookCalls{0}; + HookGuard closeHook{_detail::SetPartitionClientStateCloseHook, [&]() { ++closeHookCalls; }}; + consumer.Close(Azure::Core::Context{Azure::DateTime::clock::now() + std::chrono::seconds(5)}); + + EXPECT_EQ(1, closeHookCalls.load()); + EXPECT_EQ(0U, _detail::ConsumerClientTestAccess::PartitionClientStateCount(consumer)); + } + + TEST_F(AuthRecoveryTest, PartitionMoveAssignmentClosesActiveReceive) + { + AuthRecoveryServer server; + server.Start(); + + ConsumerClientOptions destinationOptions; + destinationOptions.Name = "destination"; + destinationOptions.RetryOptions = FastRetryOptions(); + ConsumerClient destinationConsumer( + server.ConnectionString(), "", DefaultConsumerGroup, destinationOptions); + + ConsumerClientOptions sourceOptions; + sourceOptions.Name = "source"; + sourceOptions.RetryOptions = FastRetryOptions(); + ConsumerClient sourceConsumer( + server.ConnectionString(), "", DefaultConsumerGroup, sourceOptions); + + auto destination = destinationConsumer.CreatePartitionClient("0"); + auto source = sourceConsumer.CreatePartitionClient("0"); + + Azure::Core::Context receiveContext{Azure::DateTime::clock::now() + std::chrono::seconds(5)}; + std::atomic receiveStarted{false}; + std::atomic receiveComplete{false}; + std::exception_ptr receiveFailure; + std::thread receiveThread([&]() { + receiveStarted = true; + try + { + destination.ReceiveEvents(1, receiveContext); + } + catch (...) + { + receiveFailure = std::current_exception(); + } + receiveComplete = true; + }); + + auto cleanup = [&]() { + receiveContext.Cancel(); + if (receiveThread.joinable()) + { + receiveThread.join(); + } + }; + + auto const startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(1); + while (!receiveStarted.load() && std::chrono::steady_clock::now() < startDeadline) + { + std::this_thread::yield(); + } + if (!receiveStarted.load()) + { + cleanup(); + ADD_FAILURE() << "The destination receive did not start."; + return; + } + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + if (receiveComplete.load()) + { + cleanup(); + ADD_FAILURE() << "The destination receive did not block."; + return; + } + + std::atomic closeHookCalls{0}; + HookGuard closeHook{_detail::SetPartitionClientStateCloseHook, [&]() { ++closeHookCalls; }}; + auto const moveStart = std::chrono::steady_clock::now(); + std::exception_ptr moveFailure; + try + { + destination = std::move(source); + } + catch (...) + { + moveFailure = std::current_exception(); + } + auto const moveElapsed = std::chrono::steady_clock::now() - moveStart; + auto const receiveDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(1); + while (!receiveComplete.load() && std::chrono::steady_clock::now() < receiveDeadline) + { + std::this_thread::yield(); + } + auto const receiveExited = receiveComplete.load(); + cleanup(); + + ASSERT_FALSE(moveFailure); + EXPECT_EQ(1, closeHookCalls.load()); + EXPECT_TRUE(receiveExited); + EXPECT_LT(moveElapsed, std::chrono::seconds(1)); + if (receiveFailure) + { + try + { + std::rethrow_exception(receiveFailure); + } + catch (Azure::Core::OperationCancelledException const&) + { + } + catch (std::exception const& exception) + { + ADD_FAILURE() << "The destination receive failed unexpectedly: " << exception.what(); + } + } + } + + TEST_F(AuthRecoveryTest, ReceiverReceiveRecoversUnauthorizedAndResumesWithoutDuplicate) + { + AuthRecoveryServer server(0, 0, 0, true); + server.Start(); + + ConsumerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ConsumerClient consumer(server.ConnectionString(), "", DefaultConsumerGroup, options); + PartitionClientOptions partitionOptions; + partitionOptions.StartPosition.Earliest = true; + auto partition = consumer.CreatePartitionClient("0", partitionOptions); + ASSERT_TRUE(server.WaitForFirstDetach()); + + // A receive returns as soon as it holds an event and the queue is empty, so the second + // event can arrive in a later call. + std::vector> events; + Azure::Core::Context receiveContext{Azure::DateTime::clock::now() + std::chrono::seconds(10)}; + while (events.size() < 2) + { + auto batch = partition.ReceiveEvents(2, receiveContext); + events.insert(events.end(), batch.begin(), batch.end()); + } + ASSERT_EQ(2U, events.size()); + ASSERT_TRUE(events[0]->Offset.HasValue()); + ASSERT_TRUE(events[1]->Offset.HasValue()); + EXPECT_EQ("10", events[0]->Offset.Value()); + EXPECT_EQ("11", events[1]->Offset.Value()); + EXPECT_EQ(2U, server.ConnectionCount()); + EXPECT_EQ(2, server.DeliveryLinks()); + + auto const filters = server.ReceiverFilters(); + ASSERT_EQ(2U, filters.size()); + EXPECT_EQ("amqp.annotation.x-opt-offset > '-1'", filters[0]); + EXPECT_EQ("amqp.annotation.x-opt-offset >'10'", filters[1]); + } + + TEST_F(AuthRecoveryTest, ReceiverPartialDeliveryPreservesPendingAuthenticationFailure) + { + AuthRecoveryServer server(0, 0, 0, true); + server.Start(); + + ConsumerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ConsumerClient consumer(server.ConnectionString(), "", DefaultConsumerGroup, options); + PartitionClientOptions partitionOptions; + partitionOptions.StartPosition.Earliest = true; + auto partition = consumer.CreatePartitionClient("0", partitionOptions); + ASSERT_TRUE(server.WaitForFirstDetach()); + + server.SetPutTokenFailures(2); + auto events = partition.ReceiveEvents(2); + + ASSERT_EQ(1U, events.size()); + ASSERT_TRUE(events[0]->Offset.HasValue()); + EXPECT_EQ("10", events[0]->Offset.Value()); + EXPECT_EQ(2U, server.ConnectionCount()); + EXPECT_EQ(2, server.PutTokenAttempts()); + + try + { + partition.ReceiveEvents(1); + ADD_FAILURE() << "Expected the pending authentication failure."; + } + catch (Azure::Core::Credentials::AuthenticationException const& exception) + { + EXPECT_STREQ( + "Could not authenticate client. Error Status: 401 reason: CBS PutToken failed", + exception.what()); + } + EXPECT_EQ(3U, server.ConnectionCount()); + EXPECT_EQ(3, server.PutTokenAttempts()); + } + + TEST_F(AuthRecoveryTest, CbsOpenErrorUsesOrdinaryBudgetAndLeavesAuthBudgetAvailable) + { + AuthRecoveryServer server(1, 1); + server.Start(); + + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ProducerClient producer(server.ConnectionString(), "", options); + + EXPECT_NO_THROW(producer.CreateBatch(BatchOptions())); + EXPECT_EQ(3U, server.ConnectionCount()); + EXPECT_EQ(3, server.CbsOpenAttempts()); + EXPECT_EQ(2, server.PutTokenAttempts()); + } + + TEST_F(AuthRecoveryTest, PositiveMaxRetriesEnablesRecoveryAndZeroDisablesIt) + { + { + AuthRecoveryServer server(0, 1); + server.Start(); + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(1); + ProducerClient producer(server.ConnectionString(), "", options); + EXPECT_NO_THROW(producer.CreateBatch(BatchOptions())); + EXPECT_EQ(2U, server.ConnectionCount()); + } + + { + AuthRecoveryServer server(0, 1); + server.Start(); + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(0); + ProducerClient producer(server.ConnectionString(), "", options); + EXPECT_THROW( + producer.CreateBatch(BatchOptions()), Azure::Core::Credentials::AuthenticationException); + EXPECT_EQ(1U, server.ConnectionCount()); + } + } + + TEST_F(AuthRecoveryTest, SecondAuthenticationFailureStopsWithoutThirdAttemptAndPreservesFailure) + { + { + AuthRecoveryServer server(0, 2); + server.Start(); + + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ProducerClient producer(server.ConnectionString(), "", options); + try + { + producer.CreateBatch(BatchOptions()); + ADD_FAILURE() << "Expected the second CBS PutToken failure."; + } + catch (Azure::Core::Credentials::AuthenticationException const& exception) + { + EXPECT_STREQ( + "Could not authenticate client. Error Status: 401 reason: CBS PutToken failed", + exception.what()); + } + EXPECT_EQ(2U, server.ConnectionCount()); + EXPECT_EQ(2, server.PutTokenAttempts()); + } + { + AuthRecoveryServer server(0, 0, 2); + server.Start(); + + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(); + ProducerClient producer(server.ConnectionString(), "", options); + auto batch = producer.CreateBatch(BatchOptions()); + ASSERT_TRUE(batch.TryAdd(Models::EventData{"payload"})); + + EventHubsException failure{"no failure"}; + try + { + producer.Send(batch); + ADD_FAILURE() << "Expected the second unauthorized transfer failure."; + } + catch (EventHubsException const& exception) + { + failure = exception; + } + EXPECT_EQ(2U, server.ConnectionCount()); + EXPECT_EQ(2, server.TransferAttempts()); + EXPECT_EQ("amqp:unauthorized-access", failure.ErrorCondition); + EXPECT_EQ("stale transfer", failure.ErrorDescription); + EXPECT_FALSE(failure.IsTransient); + } + } + + // The properties path has no authentication recovery, so a rejected put-token must reach + // the caller as the public AuthenticationException on both clients. + TEST_F(AuthRecoveryTest, PropertiesCallSurfacesPutTokenRejectionAsAuthenticationException) + { + { + AuthRecoveryServer server(0, 1); + server.Start(); + ProducerClientOptions options; + options.RetryOptions = FastRetryOptions(0); + ProducerClient producer(server.ConnectionString(), "", options); + EXPECT_THROW( + producer.GetEventHubProperties(), Azure::Core::Credentials::AuthenticationException); + EXPECT_EQ(1U, server.ConnectionCount()); + EXPECT_EQ(1, server.PutTokenAttempts()); + } + { + AuthRecoveryServer server(0, 1); + server.Start(); + ConsumerClientOptions options; + options.RetryOptions = FastRetryOptions(0); + ConsumerClient consumer(server.ConnectionString(), "", DefaultConsumerGroup, options); + EXPECT_THROW( + consumer.GetEventHubProperties(), Azure::Core::Credentials::AuthenticationException); + EXPECT_EQ(1U, server.ConnectionCount()); + EXPECT_EQ(1, server.PutTokenAttempts()); + } + } + + TEST_F(AuthRecoveryTest, CredentialAuthenticationExceptionIsPermanent) + { + AuthRecoveryServer server; + server.Start(); + + auto credential = std::make_shared(); + ConnectionOptions options; + options.Port = server.Port(); + options.AuthenticationScopes = {"https://eventhubs.azure.net/.default"}; + Connection connection("localhost", credential, options); + Session session{connection.CreateSession({})}; + MessageSender sender{session.CreateMessageSender(server.ProducerGatewayEndpoint(), {})}; + + bool threw = false; + try + { + auto const result = sender.Open(); + (void)result; + } + catch (Azure::Core::Credentials::AuthenticationException const&) + { + threw = true; + } + EXPECT_TRUE(threw); + EXPECT_EQ(1, credential->Attempts()); + } + +}}}} // namespace Azure::Messaging::EventHubs::Test