Feat[MQB]: Thread safe Cluster Quorum Manager and ITs

.github/workflows/check-mem-files.yaml

@@ -0,0 +1,47 @@
+name: Check alphabetical order in .mem files
+
+on:
+  pull_request:
+    types:
+      - "opened"
+      - "reopened"
+      - "synchronize"
+      - "labeled"
+      - "unlabeled"
+
+jobs:
+  check-sorting:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Check alphabetical order in .mem files
+        run: |
+          #!/usr/bin/env bash
+          set -euo pipefail
+
+          echo "Checking all .mem files in src/ for sorted order..."
+
+          # Find all .mem files recursively under src/
+          files=$(find src -type f -name "*.mem")
+
+          # Track whether we find any unsorted files
+          unsorted=0
+
+          for f in $files; do
+            # Compare file with its sorted version
+            if ! diff -q <(sort "$f") "$f" > /dev/null; then
+              echo "❌ File not sorted alphabetically: $f"
+              echo "   To fix, run: sort -o $f $f"
+              unsorted=1
+            fi
+          done
+
+          if [ "$unsorted" -eq 1 ]; then
+            echo
+            echo "Some .mem files are not sorted alphabetically."
+            exit 1
+          fi
+
+          echo "✅ All .mem files are sorted."

src/applications/bmqtool/package/bmqtool.mem

@@ -2,8 +2,8 @@ m_bmqtool_application
 m_bmqtool_filelogger
 m_bmqtool_inpututil
 m_bmqtool_interactive
-m_bmqtool_storageinspector
 m_bmqtool_messages
 m_bmqtool_parameters
 m_bmqtool_poster
 m_bmqtool_statutil
+m_bmqtool_storageinspector

src/groups/bmq/bmqp/package/bmqp.mem

@@ -10,8 +10,8 @@ bmqp_ctrlmsg_messages
 bmqp_event
 bmqp_eventutil
 bmqp_heartbeatmonitor
-bmqp_messageproperties
 bmqp_messageguidgenerator
+bmqp_messageproperties
 bmqp_optionsview
 bmqp_optionutil
 bmqp_protocol

src/groups/bmq/group/bmq.mem

@@ -1,5 +1,5 @@
-# [OFFLINE ONLY]
 
+# [OFFLINE ONLY]
 bmqa
 bmqc
 bmqeval

src/groups/mqb/mqbauthn/package/mqbauthn.mem

@@ -1,5 +1,5 @@
+
 mqbauthn_anonauthenticator
+mqbauthn_authenticationcontroller
 mqbauthn_basicauthenticator
 mqbauthn_pluginlibrary
-mqbauthn_authenticationcontroller
-

src/groups/mqb/mqbblp/mqbblp_clusterorchestrator.cpp

@@ -485,11 +485,11 @@ void ClusterOrchestrator::onNodeUnavailable(mqbnet::ClusterNode* node)
 
     BALL_LOG_INFO_BLOCK
     {
-        BALL_LOG_OUTPUT_STREAM
-            << d_clusterData_p->identity().description() << ": "
-            << node->nodeDescription() << " has gone down. "
-            << "Node was primary for " << ns->primaryPartitions().size()
-            << " partition(s): [";
+        BALL_LOG_OUTPUT_STREAM << d_clusterData_p->identity().description()
+                               << ": " << node->nodeDescription()
+                               << " has gone down. " << "Node was primary for "
+                               << ns->primaryPartitions().size()
+                               << " partition(s): [";
         for (unsigned int i = 0; i < ns->primaryPartitions().size(); ++i) {
             BALL_LOG_OUTPUT_STREAM << ns->primaryPartitions()[i];
             if (i + 1 < ns->primaryPartitions().size()) {
@@ -582,7 +582,6 @@ ClusterOrchestrator::ClusterOrchestrator(
                             mqbc::IncoreClusterStateLedger>(
                             d_allocators.get("ClusterStateLedger"),
                             clusterConfig,
-                            mqbc::ClusterStateLedgerConsistency::e_STRONG,
                             d_clusterData_p,
                             clusterState,
                             &d_clusterData_p->blobSpPool())),
@@ -599,7 +598,6 @@ ClusterOrchestrator::ClusterOrchestrator(
                             mqbc::IncoreClusterStateLedger>(
                             d_allocators.get("ClusterStateLedger"),
                             clusterConfig,
-                            mqbc::ClusterStateLedgerConsistency::e_STRONG,
                             d_clusterData_p,
                             clusterState,
                             &d_clusterData_p->blobSpPool())),
@@ -685,6 +683,7 @@ int ClusterOrchestrator::start(bsl::ostream& errorDescription)
         new (*d_allocator_p) mqbnet::Elector(
             d_clusterConfig.elector(),
             &d_clusterData_p->cluster(),
+            &d_clusterData_p->quorumManager(),
             bdlf::BindUtil::bind(&ClusterOrchestrator::onElectorStateChange,
                                  this,
                                  _1,   // ElectorState
@@ -1897,6 +1896,7 @@ int ClusterOrchestrator::processCommand(
         mqbcmd::ElectorResult electorResult;
         int                   rc = d_elector_mp->processCommand(&electorResult,
                                               command.elector());
+
         if (electorResult.isErrorValue()) {
             result->makeError(electorResult.error());
         }

src/groups/mqb/mqbc/mqbc_clusterdata.cpp

@@ -105,6 +105,8 @@ ClusterData::ClusterData(
 , d_resources(resources)
 , d_dispatcherClientData()
 , d_clusterConfig(clusterConfig)
+, d_quorumManager(clusterConfig.elector().quorum(),
+                  static_cast<unsigned int>(clusterConfig.nodes().size()))
 , d_clusterProxyConfig(clusterProxyConfig)
 , d_electorInfo(cluster)
 , d_membership(netCluster, allocator)

src/groups/mqb/mqbc/mqbc_clusterdata.h

@@ -28,6 +28,7 @@
 #include <mqbc_clustermembership.h>
 #include <mqbc_controlmessagetransmitter.h>
 #include <mqbc_electorinfo.h>
+#include <mqbcfg_clusterquorummanager.h>
 #include <mqbcfg_messages.h>
 #include <mqbi_cluster.h>
 #include <mqbi_dispatcher.h>
@@ -171,6 +172,9 @@ class ClusterData {
     /// Elector information.
     ElectorInfo d_electorInfo;
 
+    /// Quorum manager.
+    mqbcfg::ClusterQuorumManager d_quorumManager;
+
     /// The membership information of the cluster.
     ClusterMembership d_membership;
 
@@ -262,6 +266,9 @@ class ClusterData {
     /// Get a modifiable reference to this object's cluster.
     mqbi::Cluster& cluster();
 
+    /// Get a modifiable reference to this object's quorum manager.
+    mqbcfg::ClusterQuorumManager& quorumManager();
+
     /// Get a modifiable reference to this object's messageTransmitter.
     ControlMessageTransmitter& messageTransmitter();
 
@@ -391,6 +398,11 @@ inline mqbi::Cluster& ClusterData::cluster()
     return *d_cluster_p;
 }
 
+inline mqbcfg::ClusterQuorumManager& ClusterData::quorumManager()
+{
+    return d_quorumManager;
+}
+
 inline ControlMessageTransmitter& ClusterData::messageTransmitter()
 {
     return d_messageTransmitter;

src/groups/mqb/mqbc/mqbc_clusterstatemanager.cpp

@@ -64,7 +64,6 @@ ClusterStateManager::ClusterStateManager(
 , d_state_p(clusterState)
 , d_clusterFSM(*this)
 , d_nodeToLedgerLSNMap(allocator)
-, d_lsnQuorum((clusterConfig.nodes().size() / 2) + 1)
 // TODO Add cluster config to determine Eventual vs Strong
 , d_clusterStateLedger_mp(clusterStateLedger)
 , d_storageManager_p(0)
@@ -596,7 +595,7 @@ void ClusterStateManager::do_checkLSNQuorum(const ClusterFSMArgsSp& args)
     BSLS_ASSERT_SAFE(d_clusterData_p->electorInfo().isSelfLeader() &&
                      d_clusterFSM.isSelfLeader());
 
-    if (d_nodeToLedgerLSNMap.size() >= d_lsnQuorum) {
+    if (d_nodeToLedgerLSNMap.size() >= getLsnQuorum()) {
         // If we have a quorum of LSNs (including self LSN)
 
         BALL_LOG_INFO << d_clusterData_p->identity().description()

src/groups/mqb/mqbc/mqbc_clusterstatemanager.h

@@ -156,10 +156,6 @@ class ClusterStateManager BSLS_KEYWORD_FINAL
     /// ledger leader sequence number.
     NodeToLSNMap d_nodeToLedgerLSNMap;
 
-    /// Number of LSN views gathered from cluster nodes (including self)
-    /// required to achieve a quorum.
-    unsigned int d_lsnQuorum;
-
     /// Underlying cluster state ledger.
     ClusterStateLedgerMp d_clusterStateLedger_mp;
 
@@ -629,6 +625,9 @@ class ClusterStateManager BSLS_KEYWORD_FINAL
     ///
     /// @note Used for testing purposes only.
     const NodeToLSNMap& nodeToLSNMap() const;
+
+    /// Get the quorum value.
+    unsigned int getLsnQuorum() const;
 };
 
 // ============================================================================
@@ -676,6 +675,11 @@ ClusterStateManager::nodeToLSNMap() const
     return d_nodeToLedgerLSNMap;
 }
 
+inline unsigned int ClusterStateManager::getLsnQuorum() const
+{
+    return d_clusterData_p->quorumManager().quorum();
+}
+
 }  // close package namespace
 }  // close enterprise namespace
 

src/groups/mqb/mqbc/mqbc_incoreclusterstateledger.cpp

@@ -533,12 +533,6 @@ int IncoreClusterStateLedger::applyRecordInternalImpl(
                           << "' to all cluster nodes";
         }
 
-        // A follower does not reply Ack under eventual consistency
-        if (!isSelfLeader() &&
-            d_consistencyLevel == ClusterStateLedgerConsistency::e_EVENTUAL) {
-            return rc_SUCCESS;  // RETURN
-        }
-
         bmqp_ctrlmsg::ClusterMessage ackMessage;
         ackMessage.choice().makeLeaderAdvisoryAck().sequenceNumberAcked() =
             sequenceNumber;
@@ -688,7 +682,10 @@ int IncoreClusterStateLedger::applyRecordInternalImpl(
         }
 
         iter->second.d_ackCount += 1;
-        if (iter->second.d_ackCount == d_ackQuorum) {
+
+        const unsigned int ackQuorum = getAckQuorum();
+
+        if (iter->second.d_ackCount == ackQuorum) {
             // Consistency level reached. Apply a commit message for the
             // advisory, broadcast it, and invoke the 'CommitCb'.
             bmqp_ctrlmsg::ClusterMessage        commitMessage;
@@ -701,7 +698,7 @@ int IncoreClusterStateLedger::applyRecordInternalImpl(
             BSLS_ASSERT_SAFE(commitAdvisory.sequenceNumber() >
                              commitAdvisory.sequenceNumberCommitted());
 
-            BALL_LOG_INFO << description() << "Quorum of " << d_ackQuorum
+            BALL_LOG_INFO << description() << "Quorum of " << ackQuorum
                           << " acks is achieved for advisory of seqNum "
                           << ack.sequenceNumberAcked()
                           << ", creating and applying commit advisory: "
@@ -1157,23 +1154,18 @@ int IncoreClusterStateLedger::applyImpl(const bdlbb::Blob&   event,
 
 // CREATORS
 IncoreClusterStateLedger::IncoreClusterStateLedger(
-    const mqbcfg::ClusterDefinition&    clusterDefinition,
-    ClusterStateLedgerConsistency::Enum consistencyLevel,
-    ClusterData*                        clusterData,
-    ClusterState*                       clusterState,
-    BlobSpPool*                         blobSpPool_p,
-    bslma::Allocator*                   allocator)
+    const mqbcfg::ClusterDefinition& clusterDefinition,
+    ClusterData*                     clusterData,
+    ClusterState*                    clusterState,
+    BlobSpPool*                      blobSpPool_p,
+    bslma::Allocator*                allocator)
 : d_allocator_p(allocator)
 , d_isOpen(false)
 , d_blobSpPool_p(blobSpPool_p)
 , d_description(allocator)
 , d_commitCb()
 , d_clusterData_p(clusterData)
 , d_clusterState_p(clusterState)
-, d_consistencyLevel(consistencyLevel)
-, d_ackQuorum(consistencyLevel == ClusterStateLedgerConsistency::e_STRONG
-                  ? (clusterDefinition.nodes().size() / 2) + 1
-                  : 1)
 , d_ledgerConfig(allocator)
 , d_ledger_mp(0)
 , d_uncommittedAdvisories(allocator)

src/groups/mqb/mqbc/mqbc_incoreclusterstateledger.h

@@ -95,7 +95,7 @@ struct IncoreClusterStateLedger_ClusterMessageInfo {
     bsls::Types::Uint64 d_timestampNs;
 
     /// Number of ACKs received for this `ClusterMessage`.
-    int d_ackCount;
+    unsigned int d_ackCount;
 
     // TRAITS
     BSLMF_NESTED_TRAIT_DECLARATION(IncoreClusterStateLedger_ClusterMessageInfo,
@@ -184,13 +184,6 @@ class IncoreClusterStateLedger BSLS_KEYWORD_FINAL : public ClusterStateLedger {
     /// Cluster's state.
     const ClusterState* d_clusterState_p;
 
-    /// Desired consistency level (eventual vs. strong), configured by the
-    /// user.
-    ClusterStateLedgerConsistency::Enum d_consistencyLevel;
-
-    /// Number of nodes required to achieve consistency level.
-    int d_ackQuorum;
-
     /// Ledger configuration.
     mqbsi::LedgerConfig d_ledgerConfig;
 
@@ -282,6 +275,9 @@ class IncoreClusterStateLedger BSLS_KEYWORD_FINAL : public ClusterStateLedger {
     ///         dispatcher thread.
     bool isSelfLeader() const;
 
+    // Return the acknowledgment quorum required for this ledger.
+    unsigned int getAckQuorum() const;
+
   public:
     // TRAITS
     BSLMF_NESTED_TRAIT_DECLARATION(IncoreClusterStateLedger,
@@ -290,16 +286,15 @@ class IncoreClusterStateLedger BSLS_KEYWORD_FINAL : public ClusterStateLedger {
     // CREATORS
 
     /// Create a new @bbref{mqbc::IncoreClusterStateLedger} with the specified
-    /// `clusterDefinition`, `consistencyLevel`, `clusterData` and
+    /// `clusterDefinition`, `clusterData` and
     /// `clusterState`, and using the specified `bufferFactory` and `allocator`
     /// to supply memory.
     IncoreClusterStateLedger(
-        const mqbcfg::ClusterDefinition&    clusterDefinition,
-        ClusterStateLedgerConsistency::Enum consistencyLevel,
-        ClusterData*                        clusterData,
-        ClusterState*                       clusterState,
-        BlobSpPool*                         blobSpPool_p,
-        bslma::Allocator*                   allocator);
+        const mqbcfg::ClusterDefinition& clusterDefinition,
+        ClusterData*                     clusterData,
+        ClusterState*                    clusterState,
+        BlobSpPool*                      blobSpPool_p,
+        bslma::Allocator*                allocator);
 
     /// Destructor.
     ~IncoreClusterStateLedger() BSLS_KEYWORD_OVERRIDE;
@@ -456,6 +451,11 @@ inline bool IncoreClusterStateLedger::isSelfLeader() const
     return d_clusterData_p->electorInfo().isSelfLeader();
 }
 
+inline unsigned int IncoreClusterStateLedger::getAckQuorum() const
+{
+    return d_clusterData_p->quorumManager().quorum();
+}
+
 // MANIPULATORS
 //   (virtual mqbc::ClusterStateLedger)
 inline void IncoreClusterStateLedger::setCommitCb(const CommitCb& value)

src/groups/mqb/mqbc/mqbc_incoreclusterstateledger.t.cpp

@@ -285,7 +285,6 @@ struct Tester {
             new (*bmqtst::TestHelperUtil::allocator())
                 mqbc::IncoreClusterStateLedger(
                     d_cluster_mp->_clusterDefinition(),
-                    mqbc::ClusterStateLedgerConsistency::e_STRONG,
                     d_cluster_mp->_clusterData(),
                     d_cluster_mp->_state(),
                     d_cluster_mp->_blobSpPool(),
@@ -1752,8 +1751,8 @@ static void test11_persistanceAcrossRolloverLeader()
         qadvisory.queues().push_back(qinfoI);
     }
 
-    size_t i = 0;
-    bool   hasUncommittedBeforeRollover =
+    int  i = 0;
+    bool hasUncommittedBeforeRollover =
         true;  // Either the qadvisory or its commit can trigger rollover.  If
                // the commit triggers rollover, that means we have an
                // uncomitted advisory before rollover.