[BUG] core-amqp: ManagementClientImpl::ExecuteOperation leaks its response queue on cancellation and reads m_messageQueues without its lock

[sagar0207]

Library name

azure-core-amqp

Please describe the feature.

Two defects in ManagementClientImpl::ExecuteOperation (sdk/core/azure-core-amqp/src/impl/uamqp/amqp/management.cpp), both on the uAMQP backend.

1. The per-request queue leaks when an operation is cancelled

ExecuteOperation creates a response queue keyed by a fresh request id:

auto requestId = Azure::Core::Uuid::CreateUuid().ToString();
{
  std::unique_lock<std::recursive_mutex> lock(m_messageQueuesLock);
  m_messageQueues.emplace(requestId, std::make_unique<ManagementOperationQueue>());
  m_sendCompleted = false;
}

The function has three exits. Two remove that queue and one does not:

Exit	Removes the queue
send failed	yes
result received	yes
WaitForResult returned null → throw OperationCancelledException	no
rethrow from the catch (...) handler at the end	no

So every cancelled management operation leaves one entry in m_messageQueues for the lifetime of the client. A management client lives as long as the connection that owns it, and the $cbs client performs a put-token on every authentication, so a long running client that sees cancellations grows this map without bound.

The request id is a Uuid, so a leaked entry cannot be mis-matched to a later request. The cost is memory and a growing map on the receive path.

2. m_messageQueues is read without its lock

auto result = m_messageQueues.at(requestId)->WaitForResult(context);

Every other access to m_messageQueues takes m_messageQueuesLock, including OnMessageReceived, which runs on the polling thread and inserts and erases entries. This at() reads the map with no lock, so it races with those modifications.

The lock cannot simply be held across the call: OnMessageReceived takes the same lock to find the queue it must complete, so holding it through WaitForResult would deadlock. The queue pointer has to be taken under the lock and the lock released before the wait.

Observed

While investigating claims based security failures at scale (450 pods, 200 connections each, ~22.6 hours, ~2M connection rebuilds driven by a 30 minute idle link timeout), the receive path reported:

AMQP Error processing ClaimsBasedSecurity: Error {Condition =amqp:internal-error,
  Description=Message Delivery Rejected: Received message correlation ID does not match request ID., Info={}}

That message comes from OnMessageReceived when a response arrives whose request id is no longer in m_messageQueues. It occurred twice, so it is not a significant failure source on its own, and it is not necessarily caused by either defect above; a send that reports a non-Ok status can still have been delivered, and the response then arrives after that path has already removed the queue.

The two defects above are reported from reading the code and stand on their own regardless of that observation.

Suggested fix

Remove the queue from a scope guard so every exit cleans up, and take the queue pointer under the lock before waiting on it without the lock.