fixes

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

Author: dorjesinpo

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

@@ -1179,24 +1179,28 @@ QueueEngineUtil_AppState::processDeliveryList(bsls::TimeInterval*    delay,
         // Retrieve message from the storage
         reader->reset(*it);
 
-        if (!reader->hasReceipt()) {
-            // TODO: remove extra logging
-            BMQ_LOGTHROTTLE_INFO << "#STORAGE_UNKNOWN_MESSAGE " << "Queue: '"
-                                 << d_queue_p->description() << "', app: '"
-                                 << appId() << "' has unconfirmed GUID: '"
-                                 << *it << "'. Stopping the redelivery now.";
-            break;
-        }
-
         if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(reader->atEnd())) {
             BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
 
-            // The message got gc'ed or purged
+            // The message got gc'ed or purged.  Do not stop the redelivery:
+            // fall through and treat it as sent so it is erased from the list
+            // and the remaining messages still get a chance to be delivered.
             BMQ_LOGTHROTTLE_INFO << "#STORAGE_UNKNOWN_MESSAGE " << "Queue: '"
                                  << d_queue_p->description() << "', app: '"
                                  << appId() << "' could not redeliver GUID: '"
                                  << *it << "' (not in the storage)";
         }
+        else if (!reader->hasReceipt()) {
+            // The message is in the storage but not yet committed/receipted
+            // (e.g. Raft has not replicated it to a majority yet).  Stop here
+            // and retry on the next flush.
+            // TODO: remove extra logging
+            BMQ_LOGTHROTTLE_INFO << "#STORAGE_UNKNOWN_MESSAGE " << "Queue: '"
+                                 << d_queue_p->description() << "', app: '"
+                                 << appId() << "' has unconfirmed GUID: '"
+                                 << *it << "'. Stopping the redelivery now.";
+            break;
+        }
         else if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(
                      !reader->appMessageView(ordinal()).isPending())) {
             BMQ_LOGTHROTTLE_INFO

src/groups/mqb/mqbc/mqbc_clusterutil.cpp

@@ -1119,7 +1119,17 @@ bool ClusterUtil::assignQueue(ClusterState*         clusterState,
                                        status,
                                        allocator);
 
-    if (status->category() == bmqp_ctrlmsg::StatusCategory::E_SUCCESS) {
+    // Only apply to the CSL if 'startQueueAssignment' actually produced an
+    // advisory.  When the queue is already ASSIGNING or ASSIGNED (e.g. a
+    // concurrent request for the same queue is already in flight or
+    // committed), 'startQueueAssignment' returns success without populating
+    // 'queueAdvisory'. Applying that empty advisory (LSN [ 0, 0 ]) would be
+    // rejected by the ledger as stale and surface to the requester as a
+    // spurious "CSL failure". The already-pending advisory will still commit
+    // and resolve all pending contexts for the queue, so replying success here
+    // is correct.
+    if (status->category() == bmqp_ctrlmsg::StatusCategory::E_SUCCESS &&
+        !queueAdvisory.queues().empty()) {
         BSLS_ASSERT_SAFE(result);
 
         // Apply 'queueAssignmentAdvisory' to CSL

src/groups/mqb/mqbnet/mqbnet_channel.cpp

@@ -489,6 +489,7 @@ Channel::writeBufferedItem(bool*                                  isConsumed,
     case bmqp::EventType::e_AUTHENTICATION:
     case bmqp::EventType::e_RAFT_CLUSTER:
     case bmqp::EventType::e_RAFT_PARTITION:
+    case bmqp::EventType::e_RAFT_SNAPSHOT:
     default: {
         ControlArgs x(item);
         rc = writeImmediate(isConsumed,

src/groups/mqb/mqbraft/mqbraft_clusterstateraft.cpp

@@ -242,25 +242,19 @@ void ClusterStateRaft::applyCommittedEntry(const LogEntry& entry)
     if (clusterMessage.choice().selectionId() ==
         bmqp_ctrlmsg::ClusterMessageChoice::
             SELECTION_ID_PARTITION_PRIMARY_ADVISORY) {
-        // Skip 'ClusterUtil::apply': for this message type it routes to
-        // 'applyPartitionPrimary', which is an unconditional no-op in Raft
-        // mode (the data-partition Raft, not the CSL, owns primary/leaseId --
-        // 'isRaftEnabled()' early-return) -- true for both our own artificial
-        // advisory and any stale recovered legacy 'partitionPrimaryAdvisory'.
-        // This is Raft-only code (ClusterStateRaft), so that branch is always
-        // taken; nothing is lost by not calling it.  Persistence is already
-        // guaranteed by the Raft commit itself, independent of this call.
+        // Don't apply to ClusterState: under Raft the data-partition Raft, not
+        // the CSL, owns partition primary/leaseId, so 'ClusterUtil::apply' ->
+        // 'applyPartitionPrimary' is a no-op ('isRaftEnabled()' early-return)
+        // and the Raft commit already guarantees persistence.
         //
-        // Record the CSL-side half of the two independent signals the
-        // orchestrator's readiness check compares (this committed advisory
-        // vs. local data-partition Raft leadership) in a dedicated field
-        // ('advisoryConfirmedLeaseId'), separate from 'primaryLeaseId' (which
-        // reflects only the locally-observed side), so it can be recorded
-        // regardless of whether this node knows the partition's primary yet.
-        // Monotonic: a stale recovered legacy entry must not regress a
-        // leaseId already confirmed by a fresher commit.  Deciding
-        // activation/availability from this is the orchestrator's job (via
-        // the readiness check below), not this method's.
+        // The orchestrator declares a partition ready only once two leaseIds
+        // agree: the one this CSL advisory just committed, and the one
+        // observed locally from data-partition Raft leadership
+        // ('primaryLeaseId').  The two can arrive in either order, so store
+        // the advisory's leaseId in its own field, 'advisoryConfirmedLeaseId',
+        // rather than overwriting 'primaryLeaseId'.  Update it monotonically
+        // so a stale advisory replayed at recovery can't lower a leaseId a
+        // later commit confirmed.
         const bmqp_ctrlmsg::PartitionPrimaryAdvisory& adv =
             clusterMessage.choice().partitionPrimaryAdvisory();
         for (bsl::vector<bmqp_ctrlmsg::PartitionPrimaryInfo>::const_iterator

src/groups/mqb/mqbraft/mqbraft_cslraftlog.t.cpp

@@ -151,9 +151,6 @@ class Tester {
     }
 
     CslRaftLog&                     raftLog() { return d_cslRaftLog; }
-    bsl::shared_ptr<mqbsi::Log>&    log() { return d_log_sp; }
-    bdlbb::PooledBlobBufferFactory& factory()  { return d_bufferFactory; }
-    CslRaftLog::BlobSpPool*         blobPool() { return d_blobSpPool_sp.get(); }
     bslma::Allocator*               allocator() { return d_allocator_p; }
 };
 

src/groups/mqb/mqbraft/mqbraft_partitionraft.cpp

@@ -928,6 +928,7 @@ int PartitionRaft::propose(
     const bsl::shared_ptr<mqbs::FileStore::PendingWrite>& pw)
 {
     // executed by the partition *DISPATCHER* thread
+    enum { rc_UNAVAILABLE = -1 };
 
     // Compute the rollover footprint (DATA and QLIST bytes) from the write.
     // The JOURNAL reserve is always checked by 'rolloverIfNeeded'; only
@@ -946,17 +947,45 @@ int PartitionRaft::propose(
             *pw->d_appIdKeyPairs_p);
     }
 
-    int rc = rolloverIfNeeded(dataBytes, qlistBytes);
-    if (0 != rc) {
-        return rc;  // RETURN
+    if (!d_fileStore_sp->isFileSetAvailable()) {
+        // The journal is read-only: a rollover could not reclaim enough space
+        // (outstanding records exceed the policy threshold).  Mirror legacy
+        // 'FileStore::writeQueueOpRecord': the ONLY write still permitted is a
+        // full-queue PURGE, written into the reserved PURGE area, which frees
+        // outstanding records and lets 'onPurgeComplete' roll over to recover.
+        // A per-appId purge (non-null appKey) writes per-message deletion
+        // records, for which there is no room -- reject it, like any other
+        // write, so the partition stays read-only.
+        const bool isPurge = pw->d_recordType ==
+                                 mqbs::RecordType::e_QUEUE_OP &&
+                             pw->d_queueOpType == mqbs::QueueOpType::e_PURGE &&
+                             pw->d_appKey.isNull();
+        if (!isPurge || !d_fileStore_sp->primaryHasPurgeReserve()) {
+            return rc_UNAVAILABLE;  // RETURN
+        }
+
+        BALL_LOG_WARN << "Partition [" << d_partitionId
+                      << "] Writing PURGE record for queueKey ["
+                      << pw->d_queueKey
+                      << "] into the reserved journal area despite the "
+                      << "partition being read-only (unavailable).";
+        // Fall through: append the PURGE directly (no rollover); the reserved
+        // area guarantees room.
     }
+    else {
+        int rc = rolloverIfNeeded(dataBytes, qlistBytes);
+        if (0 != rc) {
+            return rc;  // RETURN
+        }
 
-    // If a rollover is in flight, buffer the write for replay into the new
-    // file once the rollover commits ('bufferPendingWrite' reserves
-    // 'pw->d_handle').
-    if (d_isRolloverPending) {
-        return d_raftLog_mp->bufferPendingWrite(pw,
-                                                d_raftNode_mp->currentTerm());
+        // If a rollover is in flight, buffer the write for replay into the new
+        // file once the rollover commits ('bufferPendingWrite' reserves
+        // 'pw->d_handle').
+        if (d_isRolloverPending) {
+            return d_raftLog_mp->bufferPendingWrite(
+                pw,
+                d_raftNode_mp->currentTerm());
+        }
     }
 
     // Otherwise enqueue it for 'append()'; the record's sequence number
@@ -967,9 +996,9 @@ int PartitionRaft::propose(
     d_raftLog_mp->setPendingWrite(pw);
 
     RaftNodeOutput output(d_allocator_p);
-    rc = d_raftNode_mp->propose(&output,
-                                bsl::shared_ptr<bdlbb::Blob>(),
-                                pw->d_id);
+    int            rc = d_raftNode_mp->propose(&output,
+                                    bsl::shared_ptr<bdlbb::Blob>(),
+                                    pw->d_id);
     if (rc != 0) {
         return rc;
     }
@@ -1121,23 +1150,37 @@ int PartitionRaft::rolloverIfNeeded(bsls::Types::Uint64 dataBytes,
                                     bsls::Types::Uint64 qlistBytes)
 {
     // executed by the partition *DISPATCHER* thread
-    enum { rc_SUCCESS = 0 };
+    enum { rc_SUCCESS = 0, rc_READONLY = -1 };
 
-    if (!isLeader() ||
-        !d_fileStore_sp->primaryNeedsRollover(dataBytes, qlistBytes)) {
+    if (!isLeader()) {
         return rc_SUCCESS;  // RETURN
     }
 
-    // A rollover is required.  Rather than NACK the triggering write, propose
-    // 'e_ROLLOVER' (unless one is already in flight -- at most one uncommitted
-    // rollover at a time) and return success.  'setPendingWrite()' then
-    // buffers this triggering write, and every subsequent one, until the
-    // rollover commits.
-    if (!d_isRolloverPending) {
-        proposeRollover();
+    switch (d_fileStore_sp->primaryRolloverNeed(dataBytes, qlistBytes)) {
+    case mqbs::FileStore::e_ROLLOVER_NONE: {
+        return rc_SUCCESS;  // RETURN
+    }
+    case mqbs::FileStore::e_ROLLOVER_READONLY: {
+        // A rollover cannot reclaim enough space -- the partition is full.
+        // 'primaryRolloverNeed' has already marked it read-only and panicked;
+        // NACK the triggering write rather than roll over into a same-size
+        // file and overflow it.  The partition recovers only via a full purge
+        // (see 'onPurgeComplete').
+        return rc_READONLY;  // RETURN
+    }
+    case mqbs::FileStore::e_ROLLOVER_NEEDED:
+    default: {
+        // A rollover is required and will reclaim enough space.  Rather than
+        // NACK the triggering write, propose 'e_ROLLOVER' (unless one is
+        // already in flight -- at most one uncommitted rollover at a time).
+        // 'setPendingWrite()' then buffers this triggering write, and every
+        // subsequent one, until the rollover commits.
+        if (!d_isRolloverPending) {
+            proposeRollover();
+        }
+        return rc_SUCCESS;  // RETURN
+    }
     }
-
-    return rc_SUCCESS;
 }
 
 void PartitionRaft::drainPendingWrites()
@@ -1658,9 +1701,18 @@ void PartitionRaft::onPurgeComplete()
     // would drive 'rolloverImpl' directly -- outside the Raft log.  Only the
     // leader proposes, and only if the file set is at capacity;
     // 'rolloverIfNeeded' is a no-op otherwise, so this is safe on replicas too
-    // (they roll over via the committed 'e_ROLLOVER' apply hook).  The legacy
-    // 'd_journalFileAvailable' re-enable is not needed here: its false-setters
-    // (legacy 'rolloverIfNeeded' / 'setActivePrimary') are legacy-only.
+    // (they roll over via the committed 'e_ROLLOVER' apply hook).
+    //
+    // If a full purge just recovered a read-only partition, the journal is
+    // marked unavailable.  Re-enable it (mirrors legacy
+    // 'FileStore::onPurgeComplete') so the reclaiming rollover's sync-point
+    // marker can be written -- 'writeSyncPointRecord' refuses on an
+    // unavailable journal.  'rolloverIfNeeded' re-disables it if the freed
+    // space is still insufficient (a rollover would still overflow).
+    if (isLeader() && !d_fileStore_sp->isFileSetAvailable()) {
+        d_fileStore_sp->setAvailabilityStatus(true);
+    }
+
     rolloverIfNeeded(0, 0);
 }
 

src/groups/mqb/mqbraft/mqbraft_raftnode.t.cpp

@@ -261,14 +261,6 @@ class TestCluster {
         }
     }
 
-    /// Tick only node 'id', deliver messages until quiet.
-    void tickNode(int id)
-    {
-        RaftNodeOutput output(d_allocator_p);
-        d_nodes[id]->tick(&output);
-        runUntilQuiet(&output);
-    }
-
     /// Find the leader node.  Return -1 if none.
     int findLeader() const
     {
@@ -284,18 +276,6 @@ class TestCluster {
         return leader;
     }
 
-    /// Return count of nodes that think 'nodeId' is the leader.
-    int leaderAgreement(int nodeId) const
-    {
-        int count = 0;
-        for (int i = 0; i < d_numNodes; ++i) {
-            if (d_nodes[i]->leaderId() == nodeId) {
-                ++count;
-            }
-        }
-        return count;
-    }
-
     bsl::shared_ptr<bdlbb::Blob> makeBlob(const char* data)
     {
         bsl::shared_ptr<bdlbb::Blob> blob =
@@ -309,7 +289,6 @@ class TestCluster {
     // ACCESSORS
     RaftNode*      node(int id) { return d_nodes[id]; }
     MemoryRaftLog* log(int id) { return d_logs[id]; }
-    int            numNodes() const { return d_numNodes; }
 };
 
 /// Tick the cluster until a leader emerges or maxTicks is reached.