Fix[mqb]: handle same qId + different uri openQueue

Signed-off-by: Evgeny Malygin <emalygin@bloomberg.net>

Author: 678098

src/groups/mqb/mqba/mqba_clientsession.t.cpp

@@ -553,9 +553,13 @@ class MyMockDomain : public mqbmock::Domain {
     mqbmock::Dispatcher*               d_mockDispatcher;
     bmqp_ctrlmsg::RoutingConfiguration d_routingConfiguration;
     bsl::shared_ptr<MyMockQueueHandle> d_queueHandle;
-    MyQueueEngine                      d_mockQueueEngine;
-    const bool                         d_atMostOnce;
-    bslma::Allocator*                  d_allocator_p;
+    // Previously created handles, kept alive so that their addresses remain
+    // distinct across successive `openQueue` calls (tests relying on handle
+    // pointer identity, e.g. duplicate-queueId, depend on this).
+    bsl::vector<bsl::shared_ptr<MyMockQueueHandle> > d_retiredHandles;
+    MyQueueEngine                                    d_mockQueueEngine;
+    const bool                                       d_atMostOnce;
+    bslma::Allocator*                                d_allocator_p;
 
     // CREATORS
 
@@ -567,6 +571,7 @@ class MyMockDomain : public mqbmock::Domain {
     : mqbmock::Domain(cluster, allocator)
     , d_mockDispatcher(dispatcher)
     , d_queueHandle()
+    , d_retiredHandles(allocator)
     , d_mockQueueEngine(allocator)
     , d_atMostOnce(atMostOnce)
     , d_allocator_p(allocator)
@@ -595,6 +600,12 @@ class MyMockDomain : public mqbmock::Domain {
         queue->_setAtMostOnce(d_atMostOnce);
         queue->_setDispatcher(d_mockDispatcher);
 
+        // Retain the previously created handle so its address is not reused by
+        // the handle created below.
+        if (d_queueHandle) {
+            d_retiredHandles.push_back(d_queueHandle);
+        }
+
         d_queueHandle.createInplace(d_allocator_p,
                                     queue,
                                     clientContext,
@@ -2262,6 +2273,79 @@ static void test11_initiateShutdown()
     }
 }
 
+static void test12_openQueueDuplicateQueueId()
+// ------------------------------------------------------------------------
+// TESTS OPEN QUEUE WITH A DUPLICATE QUEUE ID
+//
+// Concerns:
+//   - A misbehaving client may send a second openQueue request reusing a
+//     queueId that is already associated with a different queue.  The
+//     broker must reject that request gracefully (returning an error to
+//     the client).
+//
+// Plan:
+//   Instantiate a testbench, open a first queue with a given queueId, then
+//   send a second openQueue for a *different* URI reusing the same
+//   queueId.  Verify the first response is a successful OpenQueueResponse
+//   and the second is an E_INVALID_ARGUMENT Status.
+//
+// Testing:
+//   That reusing a queueId for a different queue is rejected.
+// ------------------------------------------------------------------------
+{
+    bmqtst::TestHelper::printTestName(
+        "TESTS OPEN QUEUE WITH A DUPLICATE QUEUE ID");
+
+    const bsl::string uri1("bmq://my.domain/queue-first",
+                           bmqtst::TestHelperUtil::allocator());
+    const bsl::string uri2("bmq://my.domain/queue-second",
+                           bmqtst::TestHelperUtil::allocator());
+    const int         queueId = 1;
+
+    TestBench tb(client(e_FirstHop),
+                 false,  // atMostOnce
+                 bmqtst::TestHelperUtil::allocator());
+
+    // Open the first queue: expect a successful OpenQueueResponse.
+    tb.openQueue(uri1, queueId);
+    tb.d_cs.flush();
+
+    {
+        bmqio::TestChannel::WriteCall writeCall;
+        BMQTST_ASSERT(tb.d_channel->getWriteCall(&writeCall, 0));
+
+        bmqp::Event event(&writeCall.d_blob,
+                          bmqtst::TestHelperUtil::allocator());
+        BMQTST_ASSERT(event.isControlEvent());
+
+        bmqp_ctrlmsg::ControlMessage response(
+            bmqtst::TestHelperUtil::allocator());
+        BMQTST_ASSERT_EQ(0, event.loadControlEvent(&response));
+        BMQTST_ASSERT(response.choice().isOpenQueueResponseValue());
+    }
+
+    // Open a *different* queue reusing the same queueId: expect a failure
+    // Status rather than a crash or a silent overwrite.
+    tb.openQueue(uri2, queueId);
+    tb.d_cs.flush();
+
+    {
+        bmqio::TestChannel::WriteCall writeCall;
+        BMQTST_ASSERT(tb.d_channel->getWriteCall(&writeCall, 1));
+
+        bmqp::Event event(&writeCall.d_blob,
+                          bmqtst::TestHelperUtil::allocator());
+        BMQTST_ASSERT(event.isControlEvent());
+
+        bmqp_ctrlmsg::ControlMessage response(
+            bmqtst::TestHelperUtil::allocator());
+        BMQTST_ASSERT_EQ(0, event.loadControlEvent(&response));
+        BMQTST_ASSERT(response.choice().isStatusValue());
+        BMQTST_ASSERT_EQ(response.choice().status().category(),
+                         bmqp_ctrlmsg::StatusCategory::E_INVALID_ARGUMENT);
+    }
+}
+
 static void testN1_ackConfiguration()
 // ------------------------------------------------------------------------
 // TESTS ACK CONFIGURATION FOR CLIENT SESSION
@@ -2512,6 +2596,7 @@ int main(int argc, char* argv[])
 
         switch (_testCase) {
         case 0:
+        case 12: test12_openQueueDuplicateQueueId(); break;
         case 11: test11_initiateShutdown(); break;
         case 10: test10_newStyleCompressedPush(); break;
         case 9: test9_newStylePush(); break;

src/groups/mqb/mqbblp/mqbblp_queuesessionmanager.cpp

@@ -282,28 +282,50 @@ void QueueSessionManager::onQueueOpenCbDispatched(
         BSLS_ASSERT_SAFE(confirmationCookie->d_handle);
         // in case of success, the cookie must be a valid shared_ptr
 
-        // Update the cookie to point to a null queue handle, which indicates
-        // that requester (this client session) has successfully received and
-        // processed the open-queue response.
-        confirmationCookie->d_handle = 0;
-
-        // Success, configure the handle and the session
+        // Configure the handle and the session
         bsl::pair<QueueStateMap::iterator, bool> ins = d_queues.emplace(
             bsl::make_pair(queueId.id(), QueueState()));
         QueueState& qs = ins.first->second;
 
         if (ins.second) {
-            // First time we use this queue
+            // First time we use this queueId
             qs.d_handle_p = queueHandle;
         }
-        else {
-            // We've used this queue before.  Search for the subId in the
-            // associated map of substream information.  If it is found, do
-            // nothing; otherwise, insert it into the map of substream
-            // information.
-            BSLS_ASSERT_SAFE(queueHandle == ins.first->second.d_handle_p);
+        else if (qs.d_handle_p != queueHandle) {
+            // The client reused an existing queueId for a different queue
+            // handle (i.e. a queue/URI other than the one already associated
+            // with this queueId).  This is a client protocol violation: reject
+            // the request and roll back the handle that was just opened
+            // upstream.
+            BALL_LOG_ERROR
+                << "#CLIENT_IMPROPER_BEHAVIOR "
+                << d_dispatcherClient_p->description()
+                << ": Rejecting openQueue: queueId " << queueId
+                << " is already in use for a different queue [uri: '"
+                << handleParams.uri() << "'].";
+
+            bmqp_ctrlmsg::Status failure;
+            failure.category() =
+                bmqp_ctrlmsg::StatusCategory::E_INVALID_ARGUMENT;
+            failure.code()    = -1;
+            failure.message() = "queueId is already in use for a different "
+                                "queue";
+
+            responseCallback(failure, queueHandle, openQueueResponse);
+            return;  // RETURN
         }
 
+        // At this point the queueId maps to 'queueHandle': either it was just
+        // recorded above (first use of this queueId), or it was already
+        // associated with the same handle (e.g. a new fan-out substream on an
+        // already-opened queue).  Confirm the response and register the
+        // substream.
+
+        // Update the cookie to point to a null queue handle, which indicates
+        // that requester (this client session) has successfully received and
+        // processed the open-queue response.
+        confirmationCookie->d_handle = 0;
+
         QueueState::StreamsMap::iterator subQueueInfo =
             qs.d_subQueueInfosMap.insert(apppId, queueId.subId(), queueId);