Skip to content

Fix[MQBC]: skip FSM events if primary goes down #1193

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
678098 merged 1 commit into from
Mar 25, 2026
Merged

Fix[MQBC]: skip FSM events if primary goes down #1193

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/groups/mqb/mqbc/mqbc_clusterstatemanager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,7 +285,8 @@ void ClusterStateManager::do_updatePrimaryInPFSMs(

// PRECONDITIONS
BSLS_ASSERT_SAFE(d_cluster_p->inDispatcherThread());
BSLS_ASSERT_SAFE(d_clusterFSM.isSelfHealed());
BSLS_ASSERT_SAFE(d_clusterFSM.isSelfHealed() ||
event.first == ClusterStateTableEvent::e_RST_PRIMARY);

const ClusterStateTableEvent::Enum eventType = event.first;
const bsl::vector<int>& modifiedPartitions =
Expand Down
1 change: 1 addition & 0 deletions src/groups/mqb/mqbc/mqbc_clusterstatetable.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -461,6 +461,7 @@ class ClusterStateTable
logUnexpectedCSLCommit_and_abort,
UNKNOWN);
CST_CFG(UNKNOWN, STOP_NODE, stopPFSMs, STOPPED);
CST_CFG(UNKNOWN, RST_PRIMARY, updatePrimaryInPFSMs, UNKNOWN);
Comment thread
kaikulimu marked this conversation as resolved.
CST_CFG(FOL_HEALING,
SLCT_LDR,
stopWatchDog_cancelRequests_reapplyEvent,
Expand Down
27 changes: 20 additions & 7 deletions src/groups/mqb/mqbc/mqbc_clusterutil.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -654,15 +654,28 @@ void ClusterUtil::onPartitionPrimaryAssignment(

ns->addPartitionRaw(partitionId);

// Notify the storage about (potentially same) mapping. This must be
// done before calling
// 'ClusterQueueHelper::afterPartitionPrimaryAssignment' (via
// d_afterPartitionPrimaryAssignmentCb), because ClusterQueueHelper
// assumes that storage is aware of the mapping.
storageManager->setPrimaryForPartition(partitionId, primary, leaseId);
if (!clusterData->clusterConfig()
.clusterAttributes()
.isFSMWorkflow()) {
// In legacy mode, set primary directly. In FSM workflow,
// setting is done through the Partition FSM's 'do_setPrimary'
// action upon processing 'e_DETECT_SELF_REPLICA' or
// 'e_DETECT_SELF_PRIMARY', keeping d_partitionInfoVec in sync
// with the PFSM state.
storageManager->setPrimaryForPartition(partitionId,
primary,
leaseId);
}
}
else {
storageManager->clearPrimaryForPartition(partitionId, oldPrimary);
if (!clusterData->clusterConfig()
.clusterAttributes()
.isFSMWorkflow()) {
// In legacy mode, clear primary directly. In FSM workflow,
// clearing is done through the Partition FSM's
// 'do_cleanupMetadata' action upon processing 'e_RST_UNKNOWN'.
storageManager->clearPrimaryForPartition(partitionId, oldPrimary);
}
}
}

Expand Down
15 changes: 9 additions & 6 deletions src/groups/mqb/mqbc/mqbc_partitionfsm.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,9 @@ class PartitionFSMEventData {
bmqp_ctrlmsg::PartitionSequenceNumber>
PartitionSeqNumDataRange;

/// The placeholder value for invalid or unspecified leaseId.
static const unsigned int k_INVALID_LEASE_ID = 0;

private:
// DATA

Expand Down Expand Up @@ -131,8 +134,8 @@ class PartitionFSMEventData {
int requestId,
int partitionId,
int incrementCount,
mqbnet::ClusterNode* primary = 0,
unsigned int primaryLeaseId = 0,
mqbnet::ClusterNode* primary = 0,
unsigned int primaryLeaseId = k_INVALID_LEASE_ID,
const bmqp_ctrlmsg::PartitionSequenceNumber& seqNum =
bmqp_ctrlmsg::PartitionSequenceNumber(),
const bmqp_ctrlmsg::PartitionSequenceNumber&
Expand Down Expand Up @@ -327,7 +330,7 @@ inline PartitionFSMEventData::PartitionFSMEventData()
, d_partitionId(mqbi::Storage::k_INVALID_PARTITION_ID)
, d_incrementCount(1)
, d_primary_p(0)
, d_primaryLeaseId(0) // Invalid placeholder LeaseId
, d_primaryLeaseId(k_INVALID_LEASE_ID)
, d_partitionSequenceNumber()
, d_firstSyncPointAfterRolloverSequenceNumber()
, d_partitionMaxFileSizes()
Expand Down Expand Up @@ -409,7 +412,7 @@ inline PartitionFSMEventData::PartitionFSMEventData(
, d_partitionId(partitionId)
, d_incrementCount(incrementCount)
, d_primary_p(0)
, d_primaryLeaseId(0) // Invalid placeholder primaryLeaseId
, d_primaryLeaseId(k_INVALID_LEASE_ID)
, d_partitionSequenceNumber(seqNum)
, d_firstSyncPointAfterRolloverSequenceNumber(
firstSyncPointAfterRollloverSeqNum)
Expand All @@ -432,7 +435,7 @@ inline PartitionFSMEventData::PartitionFSMEventData(
, d_partitionId(partitionId)
, d_incrementCount(incrementCount)
, d_primary_p(0)
, d_primaryLeaseId(0) // Invalid placeholder primaryLeaseId
, d_primaryLeaseId(k_INVALID_LEASE_ID)
, d_partitionSequenceNumber()
, d_firstSyncPointAfterRolloverSequenceNumber()
, d_partitionMaxFileSizes()
Expand All @@ -453,7 +456,7 @@ inline PartitionFSMEventData::PartitionFSMEventData(
, d_partitionId(partitionId)
, d_incrementCount(incrementCount)
, d_primary_p(0)
, d_primaryLeaseId(0) // Invalid placeholder primaryLeaseId
, d_primaryLeaseId(k_INVALID_LEASE_ID)
, d_partitionSequenceNumber()
, d_firstSyncPointAfterRolloverSequenceNumber()
, d_partitionMaxFileSizes()
Expand Down
17 changes: 11 additions & 6 deletions src/groups/mqb/mqbc/mqbc_partitionstatetable.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,6 +332,8 @@ class PartitionStateTableActions {

virtual void do_processLiveData(const ARGS& args) = 0;

virtual void do_setPrimary(const ARGS& args) = 0;

virtual void do_cleanupMetadata(const ARGS& args) = 0;

virtual void do_startSendDataChunks(const ARGS& args) = 0;
Expand Down Expand Up @@ -380,10 +382,11 @@ class PartitionStateTableActions {
const ARGS& args);

void
do_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq(
do_setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq(
const ARGS& args);

void do_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest(
void
do_setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest(
const ARGS& args);

void do_storeReplicaSeq_primaryStateResponse_checkQuorumMaxFileSizesAndSeq(
Expand Down Expand Up @@ -520,12 +523,12 @@ class PartitionStateTable
PST_CFG(
UNKNOWN,
DETECT_SELF_PRIMARY,
startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq,
setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq,
PRIMARY_HEALING_STG1);
PST_CFG(
UNKNOWN,
DETECT_SELF_REPLICA,
startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest,
setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest,
REPLICA_WAITING);
PST_CFG(UNKNOWN,
PRIMARY_STATE_RQST,
Expand Down Expand Up @@ -859,9 +862,10 @@ void PartitionStateTableActions<ARGS>::

template <typename ARGS>
void PartitionStateTableActions<ARGS>::
do_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq(
do_setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_replicaStateRequest_checkQuorumMaxFileSizesAndSeq(
const ARGS& args)
{
do_setPrimary(args);
do_startWatchDog(args);
do_openRecoveryFileSet(args);
do_storeSelfSeq(args);
Expand All @@ -871,9 +875,10 @@ void PartitionStateTableActions<ARGS>::

template <typename ARGS>
void PartitionStateTableActions<ARGS>::
do_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest(
do_setPrimary_startWatchDog_openRecoveryFileSet_storeSelfSeq_primaryStateRequest(
const ARGS& args)
{
do_setPrimary(args);
do_startWatchDog(args);
do_openRecoveryFileSet(args);
do_storeSelfSeq(args);
Expand Down
134 changes: 122 additions & 12 deletions src/groups/mqb/mqbc/mqbc_storagemanager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,11 +336,13 @@ void StorageManager::dispatchEventToPartition(PartitionFSM::Event::Enum event,
// PRECONDITIONS
BSLS_ASSERT_SAFE(eventDataVec.size() >= 1);

// NOTE: it is assumed that all elements in 'eventDataVec' have the same
// 'partitionId'.
const int partitionId = eventDataVec[0].partitionId();
BSLS_ASSERT_SAFE(0 <= partitionId &&
partitionId < static_cast<int>(d_fileStores.size()));
// Verify other events with 0-indexed event's partitionId
for (size_t i = 1; i < eventDataVec.size(); ++i) {
BSLS_ASSERT_SAFE(partitionId == eventDataVec[i].partitionId());
}

if (d_cluster_p->isStopping()) {
BALL_LOG_WARN << d_clusterData_p->identity().description()
Expand All @@ -353,17 +355,88 @@ void StorageManager::dispatchEventToPartition(PartitionFSM::Event::Enum event,
mqbs::FileStore* fs = d_fileStores[partitionId].get();
BSLS_ASSERT_SAFE(fs);
if (fs->inDispatcherThread()) {
d_partitionFSMVec[partitionId]->enqueueEvent(
PartitionFSM::EventWithData(event, eventDataVec));
executeEventInPartitionThread(event, eventDataVec);
}
else {
fs->execute(bdlf::BindUtil::bind(
&PartitionFSM::enqueueEvent,
d_partitionFSMVec[partitionId].get(),
PartitionFSM::EventWithData(event, eventDataVec)));
&StorageManager::executeEventInPartitionThread,
this,
event,
eventDataVec));
}
}

void StorageManager::executeEventInPartitionThread(
PartitionFSM::Event::Enum event,
const EventData& eventDataVec)
{
// Thread: QUEUE dispatcher

// PRECONDITIONS
BSLS_ASSERT_SAFE(eventDataVec.size() >= 1);

const int partitionId = eventDataVec[0].partitionId();
BSLS_ASSERT_SAFE(0 <= partitionId &&
partitionId < static_cast<int>(d_fileStores.size()));
BSLS_ASSERT_SAFE(d_fileStores[partitionId]->inDispatcherThread());

// Verify events
if (eventDataVec.size() == 1) {
const PartitionFSMEventData& evt = eventDataVec[0];

// Do not perform extra checks if primaryLeaseId is unspecified.
if (PartitionFSMEventData::k_INVALID_LEASE_ID !=
evt.primaryLeaseId()) {
const PartitionInfo& pinfo = d_partitionInfoVec[partitionId];

const bool isLeaseIdOutdated = evt.primaryLeaseId() <
pinfo.primaryLeaseId();
const bool isPrimaryMismatch = evt.primaryLeaseId() ==
pinfo.primaryLeaseId() &&
evt.primary() && pinfo.primary() &&
evt.primary()->nodeId() !=
pinfo.primary()->nodeId();

if (isLeaseIdOutdated || isPrimaryMismatch) {
BALL_LOG_WARN
<< d_clusterData_p->identity().description()
<< " Partition [" << partitionId
<< "]: dropping stale event: source "
<< evt.source()->nodeDescription()
<< ", event primaryLeaseId [" << evt.primaryLeaseId()
<< "], current primaryLeaseId [" << pinfo.primaryLeaseId()
<< "], event primary ["
<< (evt.primary() ? evt.primary()->nodeDescription()
: "null")
<< "], current primary ["
<< (pinfo.primary() ? pinfo.primary()->nodeDescription()
: "null")
<< "]";

if (0 <= evt.requestId()) {
bmqp_ctrlmsg::ControlMessage controlMsg;
controlMsg.rId() = evt.requestId();
bmqp_ctrlmsg::Status& response =
controlMsg.choice().makeStatus();

response.category() =
bmqp_ctrlmsg::StatusCategory::E_REFUSED;
response.code() = mqbi::ClusterErrorCode::e_UNKNOWN;
response.message() = "Primary mismatch";

d_clusterData_p->messageTransmitter().sendMessageSafe(
controlMsg,
evt.source());
}
return;
}
}
}

d_partitionFSMVec[partitionId]->enqueueEvent(
PartitionFSM::EventWithData(event, eventDataVec));
}

void StorageManager::setPrimaryStatusForPartitionDispatched(
int partitionId,
bmqp_ctrlmsg::PrimaryStatus::Value value)
Expand Down Expand Up @@ -1580,10 +1653,13 @@ void StorageManager::do_replicaStateResponse(const EventWithData& event)
BSLS_ASSERT_SAFE(eventDataVec.size() == 1);

const PartitionFSMEventData& eventData = eventDataVec[0];
int partitionId = eventData.partitionId();
const int partitionId = eventData.partitionId();

BSLS_ASSERT_SAFE(0 <= partitionId &&
partitionId < static_cast<int>(d_fileStores.size()));
BSLS_ASSERT_SAFE(eventData.source());
BSLS_ASSERT_SAFE(eventData.source()->nodeId() ==
d_partitionInfoVec[partitionId].primary()->nodeId());

bmqp_ctrlmsg::ControlMessage controlMsg;
controlMsg.rId() = eventData.requestId();
Expand All @@ -1609,10 +1685,6 @@ void StorageManager::do_replicaStateResponse(const EventWithData& event)
response.partitionMaxFileSizes() = getSelfPartitionMaxFileSizes(
partitionId);

BSLS_ASSERT_SAFE(eventData.source());
BSLS_ASSERT_SAFE(eventData.source()->nodeId() ==
d_partitionInfoVec[partitionId].primary()->nodeId());

fileStore(partitionId).sendMessage(controlMsg, eventData.source());

BALL_LOG_INFO << d_clusterData_p->identity().description()
Expand Down Expand Up @@ -2795,6 +2867,36 @@ void StorageManager::do_processLiveData(const EventWithData& event)
source);
}

void StorageManager::do_setPrimary(const EventWithData& event)
{
// executed by the *QUEUE DISPATCHER* thread associated with the
// paritionId contained in 'event'

const EventData& eventDataVec = event.second;
BSLS_ASSERT_SAFE(eventDataVec.size() == 1);

const PartitionFSMEventData& eventData = eventDataVec[0];
const int partitionId = eventData.partitionId();
mqbnet::ClusterNode* primary = eventData.primary();
const unsigned int leaseId = eventData.primaryLeaseId();

// PRECONDITIONS
BSLS_ASSERT_SAFE(0 <= partitionId &&
partitionId < static_cast<int>(d_fileStores.size()));
BSLS_ASSERT_SAFE(d_fileStores[partitionId]->inDispatcherThread());
BSLS_ASSERT_SAFE(primary);

PartitionInfo& pinfo = d_partitionInfoVec[partitionId];
if (pinfo.primary() && (pinfo.primary()->nodeId() == primary->nodeId())) {
pinfo.setPrimaryLeaseId(leaseId);
return; // RETURN
}

pinfo.setPrimary(primary);
pinfo.setPrimaryLeaseId(leaseId);
pinfo.setPrimaryStatus(bmqp_ctrlmsg::PrimaryStatus::E_PASSIVE);
}

void StorageManager::do_cleanupMetadata(const EventWithData& event)
{
// executed by the *QUEUE DISPATCHER* thread associated with the
Expand All @@ -2815,6 +2917,12 @@ void StorageManager::do_cleanupMetadata(const EventWithData& event)
d_partitionFSMVec[partitionId]->state() ==
PartitionFSM::State::e_STOPPED);

StorageUtil::clearPrimaryForPartition(
d_fileStores[partitionId].get(),
&d_partitionInfoVec[partitionId],
d_clusterData_p->identity().description(),
partitionId);

d_nodeToContextMapVec[partitionId].clear();
d_numReplicaDataResponsesReceivedVec[partitionId] = 0;
d_recoveryManager_mp->resetReceiveDataCtx(partitionId);
Expand Down Expand Up @@ -4905,6 +5013,8 @@ void StorageManager::processReplicaStateRequest(
message.rId().isNull() ? -1 : message.rId().value(),
partitionId,
1,
source,
replicaStateRequest.latestSequenceNumber().primaryLeaseId(),
replicaStateRequest.latestSequenceNumber(),
replicaStateRequest.firstSyncPointAfterRolloverSequenceNumber(),
replicaStateRequest.partitionMaxFileSizes());
Expand Down
10 changes: 9 additions & 1 deletion src/groups/mqb/mqbc/mqbc_storagemanager.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -458,12 +458,18 @@ class StorageManager BSLS_KEYWORD_FINAL
void onPartitionRecovery(int partitionId);

/// Dispatch the event to *QUEUE DISPATCHER* thread associated with
/// the partitionId as per the specified `eventDataVec` with the
/// the `partitionId` as per the specified `eventDataVec` with the
/// specified `event`. If we are already in *QUEUE DISPATCHER* thread,
/// then execute the event in place.
void dispatchEventToPartition(PartitionFSM::Event::Enum event,
const EventData& eventDataVec);

/// Verify and execute the event in *QUEUE DISPATCHER* thread associated
/// with the `partitionId` as per the specified `eventDataVec` with the
/// specified `event`.
void executeEventInPartitionThread(PartitionFSM::Event::Enum event,
const EventData& eventDataVec);

/// Set the primary status of the specified `partitionId` to the specified
/// `value`.
///
Expand Down Expand Up @@ -668,6 +674,8 @@ class StorageManager BSLS_KEYWORD_FINAL

void do_processLiveData(const EventWithData& event) BSLS_KEYWORD_OVERRIDE;

void do_setPrimary(const EventWithData& event) BSLS_KEYWORD_OVERRIDE;

void do_cleanupMetadata(const EventWithData& event) BSLS_KEYWORD_OVERRIDE;

void
Expand Down
Loading
Loading