SERVER-87433 Handle failover behavior for ReshardingRecipientService during abort (#32278)

GitOrigin-RevId: 8304e51dbf0bfae3710b8932cbb00231499724fd

Author: krutishah139

jstests/sharding/resharding_failover_during_abort.js

@@ -0,0 +1,70 @@
+/**
+ * Simulates a failover prior to removing the recipient doc while resharding is aborting from an
+ * unrecoverable error on the donor. Resharding should abort successfully after stepUp.
+ *
+ * See BF-32038 for more details.
+ * @tags: [
+ *  requires_fcv_70,
+ * ]
+ */
+(function() {
+"use strict";
+
+load("jstests/libs/discover_topology.js");
+load("jstests/libs/fail_point_util.js");
+load("jstests/sharding/libs/resharding_test_fixture.js");
+
+const reshardingTest = new ReshardingTest({numDonors: 2, numRecipients: 2});
+reshardingTest.setup();
+
+const donorShardNames = reshardingTest.donorShardNames;
+const recipientShardNames = reshardingTest.recipientShardNames;
+
+const sourceCollection = reshardingTest.createShardedCollection({
+    ns: "reshardingDb.coll",
+    shardKeyPattern: {oldKey: 1},
+    chunks: [
+        {min: {oldKey: MinKey}, max: {oldKey: 10}, shard: donorShardNames[0]},
+        {min: {oldKey: 10}, max: {oldKey: MaxKey}, shard: donorShardNames[1]},
+    ],
+});
+const mongos = sourceCollection.getMongo();
+const topology = DiscoverTopology.findConnectedNodes(mongos);
+const donor = new Mongo(topology.shards[donorShardNames[0]].primary);
+const recipient = new Mongo(topology.shards[recipientShardNames[0]].primary);
+
+const reshardingDonorFailsBeforeObtainingTimestampFp =
+    configureFailPoint(donor, "reshardingDonorFailsBeforeObtainingTimestamp");
+const hangBeforeRemovingRecipientDocFp =
+    configureFailPoint(recipient, "removeRecipientDocFailpoint");
+
+reshardingTest.withReshardingInBackground(
+    {
+        newShardKeyPattern: {newKey: 1},
+        newChunks: [
+            {min: {newKey: MinKey}, max: {newKey: 10}, shard: recipientShardNames[0]},
+            {min: {newKey: 10}, max: {newKey: MaxKey}, shard: recipientShardNames[1]},
+        ],
+    },
+    () => {
+        hangBeforeRemovingRecipientDocFp.wait();
+
+        const recipientDoc =
+            recipient.getCollection('config.localReshardingOperations.recipient').findOne({
+                ns: "reshardingDb.coll"
+            });
+        assert(recipientDoc != null);
+        assert(recipientDoc.mutableState.state === "done");
+        assert(recipientDoc.mutableState.abortReason != null);
+        assert(recipientDoc.mutableState.abortReason.code === ErrorCodes.ReshardCollectionAborted);
+
+        reshardingTest.stepUpNewPrimaryOnShard(recipientShardNames[0]);
+        const recipientRS = reshardingTest.getReplSetForShard(recipientShardNames[0]);
+        recipientRS.awaitSecondaryNodes();
+        recipientRS.awaitReplication();
+        reshardingTest.retryOnceOnNetworkError(hangBeforeRemovingRecipientDocFp.off);
+    },
+    {expectedErrorCode: ErrorCodes.InternalError});
+
+reshardingTest.teardown();
+})();

src/mongo/db/s/resharding/resharding_coordinator_observer_test.cpp

@@ -252,5 +252,46 @@ TEST_F(ReshardingCoordinatorObserverTest, onDonorsReportedMinFetchTimestamp) {
 
     reshardingObserver->interrupt(Status{ErrorCodes::Interrupted, "interrupted"});
 }
+
+TEST_F(ReshardingCoordinatorObserverTest, onAllParticipantsReportingAbortReason) {
+    auto reshardingObserver = std::make_shared<ReshardingCoordinatorObserver>();
+    auto futRecipient = reshardingObserver->awaitAllRecipientsDone();
+    auto futDonor = reshardingObserver->awaitAllDonorsDone();
+    ASSERT_FALSE(futRecipient.isReady());
+    ASSERT_FALSE(futDonor.isReady());
+
+    auto abortReason = Status{ErrorCodes::InternalError, "We gotta abort"};
+    auto abortReasonFromCoordinator =
+        Status{ErrorCodes::ReshardCollectionAborted, "Recieved abort from coordinator"};
+
+    auto recipientShards0 =
+        makeMockRecipientsInState(RecipientStateEnum::kAwaitingFetchTimestamp, Timestamp(1, 1));
+    std::vector<DonorShardEntry> donorShards0{
+        {resharding::makeDonorShard(
+            ShardId{"s1"}, DonorStateEnum::kError, Timestamp(1, 1), abortReason)},
+        {resharding::makeDonorShard(
+            ShardId{"s2"}, DonorStateEnum::kPreparingToDonate, Timestamp(1, 1))}};
+
+    auto coordinatorDoc0 =
+        makeCoordinatorDocWithRecipientsAndDonors(recipientShards0, donorShards0, abortReason);
+    reshardingObserver->onReshardingParticipantTransition(coordinatorDoc0);
+    ASSERT_FALSE(futRecipient.isReady());
+    ASSERT_FALSE(futDonor.isReady());
+
+    auto recipientShards1 = makeMockRecipientsInState(
+        RecipientStateEnum::kDone, Timestamp(1, 1), abortReasonFromCoordinator);
+    std::vector<DonorShardEntry> donorShards1{
+        {resharding::makeDonorShard(
+            ShardId{"s1"}, DonorStateEnum::kDone, Timestamp(1, 1), abortReason)},
+        {resharding::makeDonorShard(
+            ShardId{"s2"}, DonorStateEnum::kDone, Timestamp(1, 1), abortReasonFromCoordinator)}};
+
+    auto coordinatorDoc1 =
+        makeCoordinatorDocWithRecipientsAndDonors(recipientShards1, donorShards1, abortReason);
+    reshardingObserver->onReshardingParticipantTransition(coordinatorDoc1);
+    ASSERT_TRUE(futRecipient.isReady());
+    ASSERT_TRUE(futDonor.isReady());
+}
+
 }  // namespace
 }  // namespace mongo

src/mongo/db/s/resharding/resharding_donor_service.cpp

@@ -75,6 +75,7 @@ MONGO_FAIL_POINT_DEFINE(reshardingPauseDonorBeforeCatalogCacheRefresh);
 MONGO_FAIL_POINT_DEFINE(reshardingPauseDonorAfterBlockingReads);
 MONGO_FAIL_POINT_DEFINE(reshardingDonorFailsAfterTransitionToDonatingOplogEntries);
 MONGO_FAIL_POINT_DEFINE(removeDonorDocFailpoint);
+MONGO_FAIL_POINT_DEFINE(reshardingDonorFailsBeforeObtainingTimestamp);
 
 using namespace fmt::literals;
 
@@ -119,18 +120,6 @@ Timestamp generateMinFetchTimestamp(OperationContext* opCtx, const NamespaceStri
     return generatedOpTime.getTimestamp();
 }
 
-/**
- * Returns whether it is possible for the donor to be in 'state' when resharding will indefinitely
- * abort.
- */
-bool inPotentialAbortScenario(const DonorStateEnum& state) {
-    // Regardless of whether resharding will abort or commit, the donor will eventually reach state
-    // kDone.
-    // Additionally, if the donor is in state kError, it is guaranteed that the coordinator will
-    // eventually begin the abort process.
-    return state == DonorStateEnum::kError || state == DonorStateEnum::kDone;
-}
-
 /**
  * Fulfills the promise if it is not already. Otherwise, does nothing.
  */
@@ -146,6 +135,17 @@ void ensureFulfilledPromise(WithLock lk, SharedPromise<void>& sp, Status error)
     }
 }
 
+/**
+ * Returns whether it is possible for the donor to be in 'state' when resharding will indefinitely
+ * abort.
+ */
+bool inPotentialAbortScenario(const DonorStateEnum& state) {
+    // Regardless of whether resharding will abort or commit, the donor will eventually reach state
+    // kDone. Additionally, if the donor is in state kError, it is guaranteed that the coordinator
+    // will eventually begin the abort process.
+    return state == DonorStateEnum::kError || state == DonorStateEnum::kDone;
+}
+
 class ExternalStateImpl : public ReshardingDonorService::DonorStateMachineExternalState {
 public:
     ShardId myShardId(ServiceContext* serviceContext) const override {
@@ -378,7 +378,7 @@ ExecutorFuture<void> ReshardingDonorService::DonorStateMachine::_finishReshardin
                    }
 
                    // If aborted, the donor must be allowed to transition to done from any state.
-                   _transitionState(DonorStateEnum::kDone);
+                   _transitionToDone(aborted);
                }
 
                {
@@ -611,6 +611,12 @@ void ReshardingDonorService::DonorStateMachine::
         _externalState->waitForCollectionFlush(opCtx.get(), _metadata.getTempReshardingNss());
     }
 
+    reshardingDonorFailsBeforeObtainingTimestamp.execute([&](const BSONObj& data) {
+        auto errmsgElem = data["errmsg"];
+        StringData errmsg = errmsgElem ? errmsgElem.checkAndGetStringData() : "Failing for test"_sd;
+        uasserted(ErrorCodes::InternalError, errmsg);
+    });
+
     Timestamp minFetchTimestamp = [this] {
         auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
         return generateMinFetchTimestamp(opCtx.get(), _metadata.getSourceNss());
@@ -847,12 +853,12 @@ void ReshardingDonorService::DonorStateMachine::_dropOriginalCollectionThenTrans
             opCtx.get(), _metadata.getSourceNss(), _metadata.getSourceUUID());
     }
 
-    _transitionState(DonorStateEnum::kDone);
+    _transitionToDone(false /* aborted */);
 }
 
 void ReshardingDonorService::DonorStateMachine::_transitionState(DonorStateEnum newState) {
     invariant(newState != DonorStateEnum::kDonatingInitialData &&
-              newState != DonorStateEnum::kError);
+              newState != DonorStateEnum::kError && newState != DonorStateEnum::kDone);
 
     auto newDonorCtx = _donorCtx;
     newDonorCtx.setState(newState);
@@ -864,6 +870,10 @@ void ReshardingDonorService::DonorStateMachine::_transitionState(DonorShardConte
     auto oldState = _donorCtx.getState();
     auto newState = newDonorCtx.getState();
 
+    // The donor state machine enters the kError state on unrecoverable errors and so we don't
+    // expect it to ever transition from kError except to kDone.
+    invariant(oldState != DonorStateEnum::kError || newState == DonorStateEnum::kDone);
+
     _updateDonorDocument(std::move(newDonorCtx));
 
     _metrics->onStateTransition(oldState, newState);
@@ -894,6 +904,16 @@ void ReshardingDonorService::DonorStateMachine::_transitionToError(Status abortR
     _transitionState(std::move(newDonorCtx));
 }
 
+void ReshardingDonorService::DonorStateMachine::_transitionToDone(bool aborted) {
+    auto newDonorCtx = _donorCtx;
+    newDonorCtx.setState(DonorStateEnum::kDone);
+    if (aborted) {
+        resharding::emplaceTruncatedAbortReasonIfExists(newDonorCtx,
+                                                        resharding::coordinatorAbortedError());
+    }
+    _transitionState(std::move(newDonorCtx));
+}
+
 /**
  * Returns a query filter of the form
  * {
@@ -1069,6 +1089,13 @@ CancellationToken ReshardingDonorService::DonorStateMachine::_initAbortSource(
         _abortSource = CancellationSource(stepdownToken);
     }
 
+    if (_donorCtx.getState() == DonorStateEnum::kDone && _donorCtx.getAbortReason()) {
+        // A donor in state kDone with an abortReason is indication that the coordinator
+        // has persisted the decision and called abort on all participants. Canceling the
+        // _abortSource to avoid repeating the future chain.
+        _abortSource->cancel();
+    }
+
     if (auto future = _coordinatorHasDecisionPersisted.getFuture(); future.isReady()) {
         if (auto status = future.getNoThrow(); !status.isOK()) {
             // onReshardingFieldsChanges() missed canceling _abortSource because _initAbortSource()

src/mongo/db/s/resharding/resharding_donor_service.h

@@ -204,6 +204,9 @@ class ReshardingDonorService::DonorStateMachine final
     // Transitions the on-disk and in-memory state to DonorStateEnum::kError.
     void _transitionToError(Status abortReason);
 
+    // Transitions the on-disk and in-memory state to DonorStateEnum::kDone.
+    void _transitionToDone(bool aborted);
+
     BSONObj _makeQueryForCoordinatorUpdate(const ShardId& shardId, DonorStateEnum newState);
 
     ExecutorFuture<void> _updateCoordinator(