Fix[MQB]: do not access queue handle flags from cluster thread [179311632] (#764)

Signed-off-by: Evgeny Malygin <[email protected]>

Author: 678098

src/groups/mqb/mqbblp/mqbblp_cluster.cpp

@@ -1045,6 +1045,9 @@ void Cluster::onPutEvent(const mqbi::DispatcherPutEvent& event)
     // it needs to be forwarded to the queue after appropriate checks.  The
     // replica source node is event.clusterNode().  This routine is similar to
     // that of ClientSession.
+    // Note that we don't check if the queue is opened in WRITE mode here, we
+    // just pass the event to the corresponding *QUEUE* dispatcher thread that
+    // can safely access the `QueueHandleParameters` to check this.
 
     mqbnet::ClusterNode*      source = event.clusterNode();
     mqbc::ClusterNodeSession* ns =
@@ -1128,28 +1131,6 @@ void Cluster::onPutEvent(const mqbi::DispatcherPutEvent& event)
                              bmqp::QueueId::k_DEFAULT_SUBQUEUE_ID) !=
                          queueState.d_subQueueInfosMap.end());
 
-        // Ensure that queue is opened in WRITE mode
-        if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(
-                !bmqt::QueueFlagsUtil::isWriter(
-                    queueState.d_handle_p->handleParameters().flags()))) {
-            BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
-
-            BMQU_THROTTLEDACTION_THROTTLE(
-                d_throttledFailedPutMessages,
-                BALL_LOG_WARN << description() << ": PUT message for queue ["
-                              << queueState.d_handle_p->queue()->uri()
-                              << "] not opened in WRITE mode.";);
-
-            sendAck(bmqt::AckResult::e_REFUSED,
-                    bmqp::AckMessage::k_NULL_CORRELATION_ID,
-                    putIt.header().messageGUID(),
-                    queueId.id(),
-                    "putEvent::notWritable",
-                    ns,
-                    true);  // isSelfGenerated
-            continue;       // CONTINUE
-        }
-
         // Ensure that final closeQueue request has not been sent.
 
         if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(

src/groups/mqb/mqbblp/mqbblp_localqueue.cpp

@@ -421,14 +421,18 @@ void LocalQueue::postMessage(const bmqp::PutHeader&              putHeader,
                 << "] from client [" << source->client()->description()
                 << "]. Queue not opened in WRITE mode by the client.";);
 
-        // Note that a NACK is not sent in this case.  This is a case of client
-        // violating the contract, by attempting to post a message after
-        // closing/reconfiguring the queue.  Since this is out of contract, its
-        // ok not to send the NACK.  If it is still desired to send a NACK, it
-        // will need some enqueuing b/w client and queue dispatcher threads to
-        // ensure that despite NACKs being sent, closeQueue response is still
-        // the last event to be sent to the client for the given queue.
+        bmqp::AckMessage ackMessage;
+        ackMessage
+            .setStatus(bmqp::ProtocolUtil::ackResultToCode(
+                bmqt::AckResult::e_REFUSED))
+            .setMessageGUID(putHeader.messageGUID());
+        // CorrelationId & QueueId are left unset as those fields will
+        // be filled downstream.
 
+        source->onAckMessage(ackMessage);
+
+        d_state_p->stats()
+            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_NACK>(1);
         return;  // RETURN
     }
 

src/groups/mqb/mqbblp/mqbblp_remotequeue.cpp

@@ -904,33 +904,6 @@ void RemoteQueue::postMessage(const bmqp::PutHeader&              putHeaderIn,
     translation.applyTo(&putHeader);
 
     // Relay the PUT message via clusterProxy/cluster
-    if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(!bmqt::QueueFlagsUtil::isWriter(
-            source->handleParameters().flags()))) {
-        BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
-
-        // Either queue was not opened in the WRITE mode (which should have
-        // been caught in the SDK) or client is posting a message after closing
-        // or reconfiguring the queue (which may not be caught in the SDK).
-
-        if (d_throttledFailedPutMessages.requestPermission()) {
-            BALL_LOG_WARN
-                << "[THROTTLED] #CLIENT_IMPROPER_BEHAVIOR "
-                << "Failed PUT message for queue [" << d_state_p->uri()
-                << "] from client [" << source->client()->description()
-                << "]. Queue not opened in WRITE mode by the client.";
-        }
-
-        // Note that a NACK is not sent in this case.  This is a case of client
-        // violating the contract, by attempting to post a message after
-        // closing/reconfiguring the queue.  Since this is out of contract, its
-        // ok not to send the NACK.  If it is still desired to send a NACK, it
-        // will need some enqueuing b/w client and queue dispatcher threads to
-        // ensure that despite NACKs being sent, closeQueue response is still
-        // the last event to be sent to the client for the given queue.
-
-        return;  // RETURN
-    }
-
     SubStreamContext& ctx = d_producerState;
 
     if (ctx.d_state == SubStreamContext::e_NONE) {
@@ -944,6 +917,31 @@ void RemoteQueue::postMessage(const bmqp::PutHeader&              putHeaderIn,
 
     if (ctx.d_state == SubStreamContext::e_CLOSED) {
         isInvalid = true;
+
+        if (d_throttledFailedPutMessages.requestPermission()) {
+            BALL_LOG_WARN << "[THROTTLED] Failed PUT message for queue ["
+                          << d_state_p->uri()
+                          << "]. Upstream is already closed.";
+        }
+    }
+    else if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(
+                 !bmqt::QueueFlagsUtil::isWriter(
+                     source->handleParameters().flags()))) {
+        BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
+
+        // Either queue was not opened in the WRITE mode (which should have
+        // been caught in the SDK) or client is posting a message after closing
+        // or reconfiguring the queue (which may not be caught in the SDK).
+
+        isInvalid = true;
+
+        if (d_throttledFailedPutMessages.requestPermission()) {
+            BALL_LOG_WARN
+                << "[THROTTLED] #CLIENT_IMPROPER_BEHAVIOR "
+                << "Failed PUT message for queue [" << d_state_p->uri()
+                << "] from client [" << source->client()->description()
+                << "]. Queue not opened in WRITE mode by the client.";
+        }
     }
     else if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(
                  d_pendingMessages.find(putHeader.messageGUID()) !=