Azure · Sagar Patel (sagar0207) · Aug 28, 2026
@@ -36,6 +36,7 @@
 - A claims based security open that fails now throws `CbsOpenFailedException`, which carries the `CbsOpenResult`. The three failures need different handling: `Error` reached the transport and may be retried, while `Cancelled` is the caller's own cancellation or deadline and `Invalid` is a state error. The result was previously readable only by matching the message text, so a reword would have changed caller behavior with no compiler error. The type derives from `std::runtime_error` and carries the same message, so existing handlers keep working. The Rust backend reports every open failure by throwing rather than by returning a result, so those throws are classified at the shared call site and carry the same type.
 - The uAMQP management client now closes the message sender when the message receiver fails to open. Two handlers returned a status without that close, and a message sender that stays open stops the process in its own destructor.
 - The uAMQP management client now names the management node and the open status in the lines that it writes when an open fails, and it keeps the text of the exception that ended the open. The message sender open failure moved from the Error level to the Warning level, because that call reports the failure to its caller.
+- The uAMQP management client no longer leaves the response queue of a cancelled operation behind. `ExecuteOperation` creates one queue for each request and removed it on the two exits that return, but not on the exit that throws for a wait that ended without a result, and not on the rethrow that follows. A management client lives as long as the connection that owns it, and the `$cbs` client runs a put-token for every authentication, so a long running client grew that map by one entry for every cancelled operation. The removal now runs from a scope guard, so every exit clears it. The request identifier is a UUID, so a leaked entry could never be matched to a later request; the cost was memory. `ExecuteOperation` also read the queue out of the map with `at()` while holding no lock. That function runs on more than one thread and it is where entries are inserted and removed, so the read raced with another caller's insert or erase. It now takes the queue under the lock and waits without it, because the receive path needs the same lock to complete the queue. [[#7386]](https://github.com/Azure/azure-sdk-for-cpp/issues/7386)
 
 ### Other Changes
 

@@ -208,6 +208,15 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
         Models::AmqpMessage messageToSend,
         Context const& context = {});
 
+#if _azure_TESTING_BUILD && ENABLE_UAMQP
+    /** @brief The number of requests still waiting for a response.
+     *
+     * `ExecuteOperation` creates one queue for each request and removes it on the way out. A test
+     * uses this to assert that an operation that ended without a response left nothing behind.
+     */
+    std::size_t GetPendingOperationCount() const;
+#endif
+
   private:
     friend class Azure::Core::Amqp::_detail::ManagementClientFactory;
     ManagementClient(std::shared_ptr<_detail::ManagementClientImpl> impl) : m_impl{impl} {}

@@ -47,4 +47,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     return m_impl->ExecuteOperation(
         operationToPerform, typeOfOperation, locales, messageToSend, context);
   }
+
+#if _azure_TESTING_BUILD && ENABLE_UAMQP
+  std::size_t ManagementClient::GetPendingOperationCount() const
+  {
+    return m_impl->GetPendingOperationCount();
+  }
+#endif
 }}}} // namespace Azure::Core::Amqp::_internal
@@ -269,6 +269,26 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
         m_sendCompleted = false;
       }
 
+      // Every exit from this function has to remove the queue for this request. The two returns
+      // below did that inline, but the cancellation throw did not, and neither did the rethrow in
+      // the handler at the end, so a cancelled operation left its queue in the map for the life of
+      // the client. A client that lives as long as the connection therefore grew one entry for
+      // every cancelled operation. The guard below removes the queue on every path.
+      //
+      // The identifier is held by reference. Copying it would allocate, and an allocation that
+      // throws here would leave the entry that was just inserted with no guard to remove it.
+      // `requestId` is declared above the guard, so it outlives it.
+      struct QueueRemover final
+      {
+        ManagementClientImpl* Client;
+        std::string const& RequestId;
+        ~QueueRemover()
+        {
+          std::unique_lock<std::recursive_mutex> lock(Client->m_messageQueuesLock);
+          Client->m_messageQueues.erase(RequestId);
+        }
+      } const queueRemover{this, requestId};
+
       auto sendResult = m_messageSender->Send(messageToSend, context);
       if (std::get<0>(sendResult) != _internal::MessageSendStatus::Ok)
       {
@@ -298,28 +318,29 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
         rv.StatusCode = 500;
         rv.Error = sendError;
         rv.Message = nullptr;
-        {
-          std::unique_lock<std::recursive_mutex> lock(m_messageQueuesLock);
-          // Remove the queue from the map, we don't need it anymore.
-          m_messageQueues.erase(requestId);
-        }
         return rv;
       }
 
-      auto result = m_messageQueues.at(requestId)->WaitForResult(context);
+      // Take the queue under the lock, then wait on it without the lock. `ExecuteOperation` runs
+      // on more than one thread and it is where entries are inserted and removed, so reading the
+      // map here without the lock races with another caller's insert or erase. The lock cannot be
+      // held across the wait, because the receive path takes it to find the queue it has to
+      // complete. Only this call removes this request's queue, and it does so after the wait, so
+      // the pointer stays valid for the wait.
+      ManagementOperationQueue* operationQueue = nullptr;
+      {
+        std::unique_lock<std::recursive_mutex> lock(m_messageQueuesLock);
+        operationQueue = m_messageQueues.at(requestId).get();
+      }
+
+      auto result = operationQueue->WaitForResult(context);
       if (result)
       {
         _internal::ManagementOperationResult rv;
         rv.Status = std::get<0>(*result);
         rv.StatusCode = std::get<1>(*result);
         rv.Error = std::get<2>(*result);
         rv.Message = std::get<3>(*result);
-
-        {
-          std::unique_lock<std::recursive_mutex> lock(m_messageQueuesLock);
-          // Remove the queue from the map, we don't need it anymore.
-          m_messageQueues.erase(requestId);
-        }
         return rv;
       }
       else

@@ -86,6 +86,19 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
         Models::AmqpMessage messageToSend,
         Context const& context);
 
+#if _azure_TESTING_BUILD
+    /** @brief The number of requests still waiting for a response.
+     *
+     * `ExecuteOperation` creates one queue for each request and removes it on the way out. A test
+     * uses this to assert that an operation that failed or was cancelled left nothing behind.
+     */
+    std::size_t GetPendingOperationCount()
+    {
+      std::unique_lock<std::recursive_mutex> lock(m_messageQueuesLock);
+      return m_messageQueues.size();
+    }
+#endif
+
   private:
     enum class ManagementState
     {

@@ -196,6 +196,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
       {
         m_expectedStatusDescriptionName = expectedStatusDescriptionName;
       }
+      // Accept the request and never answer it. The send then succeeds and the caller waits for a
+      // response that never arrives, which is the only way to reach the exit that throws for a
+      // wait that ended without a result.
+      void SetSwallowRequests(bool swallowRequests) { m_swallowRequests = swallowRequests; }
       ManagementServiceEndpoint(MessageTests::MockServiceEndpointOptions const& options)
           : MockServiceEndpoint("$management", options)
       {
@@ -232,6 +236,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
       void MessageReceived(std::string const&, std::shared_ptr<AmqpMessage> const& incomingMessage)
           override
       {
+        if (m_swallowRequests)
+        {
+          GTEST_LOG_(INFO) << "Swallowing request; no response will be sent.";
+          return;
+        }
         if (incomingMessage->ApplicationProperties.at("operation") == "Test")
         {
           AmqpMessage responseMessage;
@@ -265,6 +274,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
       AmqpValue m_expectedStatusDescription{"Successful"};
       std::string m_expectedStatusCodeName = "statusCode";
       std::string m_expectedStatusDescriptionName = "statusDescription";
+      bool m_swallowRequests{false};
     };
   } // namespace
 #endif
@@ -573,6 +583,66 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
 #endif
   }
 
+  // `ExecuteOperation` creates one response queue per request and removes it on the way out. Two
+  // of its exits did that inline; the exit that throws for a wait that ended without a result did
+  // not, and neither did the rethrow that follows. A management client lives as long as its
+  // connection, so a cancelled operation left an entry behind for good.
+  //
+  // Reaching that exit needs the send to succeed and the response never to arrive, so the endpoint
+  // swallows the request and a deadline ends the wait. Cancelling before the call would not do it:
+  // the send fails first and takes a different exit, one that already removed the queue.
+  TEST_F(TestManagement, ManagementExecuteOperationLeavesNoPendingQueueWhenTheWaitEnds)
+  {
+#if ENABLE_UAMQP
+    auto managementEndpoint
+        = std::make_shared<ManagementServiceEndpoint>(MessageTests::MockServiceEndpointOptions{});
+    m_mockServer.AddServiceEndpoint(managementEndpoint);
+
+    Connection connection{CreateAmqpConnection()};
+    Session session{CreateAmqpSession(connection)};
+    ManagementClientOptions options;
+    options.EnableTrace = true;
+    ManagementClient management(session.CreateManagementClient("Test", options));
+
+    StartServerListening();
+
+    auto openResult = management.Open();
+    ASSERT_EQ(openResult, ManagementOpenStatus::Ok);
+
+    ASSERT_EQ(std::size_t{0}, management.GetPendingOperationCount());
+
+    AmqpMessage messageToSend;
+    messageToSend.SetBody(AmqpValue("Test"));
+
+    managementEndpoint->SetSwallowRequests(true);
+    EXPECT_THROW(
+        {
+          auto unusedResponse = management.ExecuteOperation(
+              "Test",
+              "Test",
+              "Test",
+              messageToSend,
+              Azure::Core::Context{}.WithDeadline(
+                  std::chrono::system_clock::now() + std::chrono::seconds(3)));
+          (void)unusedResponse;
+        },
+        Azure::Core::OperationCancelledException);
+
+    // The queue for that request must be gone. Before the fix it stayed in the map.
+    EXPECT_EQ(std::size_t{0}, management.GetPendingOperationCount());
+
+    managementEndpoint->SetSwallowRequests(false);
+
+    management.Close();
+
+    StopServerListening();
+
+    EndAmqpSession(session);
+    CloseAmqpConnection(connection);
+#else
+#endif
+  }
+
   TEST_F(TestManagement, ManagementRequestResponseExpect500)
   {
 #if ENABLE_UAMQP