SERVER-21382 fixing sharding migration to transfer only document deletions relevant to the chunk being migrated, not every deletion

(cherry picked from commit 3663e004dfc2f73b82b3d88b5fa1ac6b7dcd1d33)

9 files changed, 71 insertions, 20 deletions

diff --git a/src/mongo/db/catalog/collection.cpp b/src/mongo/db/catalog/collection.cpp
index 3be9467c79e..daa74399995 100644
--- a/src/mongo/db/catalog/collection.cpp
+++ b/src/mongo/db/catalog/collection.cpp
@@ -488,11 +488,8 @@ void Collection::deleteDocument(OperationContext* txn,
 
     Snapshotted<BSONObj> doc = docFor(txn, loc);
 
-    BSONElement e = doc.value()["_id"];
-    BSONObj id;
-    if (e.type()) {
-        id = e.wrap();
-    }
+    auto opObserver = getGlobalServiceContext()->getOpObserver();
+    OpObserver::DeleteState deleteState = opObserver->aboutToDelete(txn, ns(), doc.value());
 
     /* check if any cursors point to us.  if so, advance them. */
     _cursorManager.invalidateDocument(txn, loc, INVALIDATION_DELETION);
@@ -501,9 +498,7 @@ void Collection::deleteDocument(OperationContext* txn,
 
     _recordStore->deleteRecord(txn, loc);
 
-    if (!id.isEmpty()) {
-        getGlobalServiceContext()->getOpObserver()->onDelete(txn, ns().ns(), id);
-    }
+    opObserver->onDelete(txn, ns(), std::move(deleteState));
 }
 
 Counter64 moveCounter;
diff --git a/src/mongo/db/op_observer.cpp b/src/mongo/db/op_observer.cpp
index fc2f4485b53..b4dacd091c5 100644
--- a/src/mongo/db/op_observer.cpp
+++ b/src/mongo/db/op_observer.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/s/d_state.h"
 #include "mongo/scripting/engine.h"
+#include "mongo/db/operation_context.h"
 
 namespace mongo {
 
@@ -91,16 +92,37 @@ void OpObserver::onUpdate(OperationContext* txn, oplogUpdateEntryArgs args) {
     }
 }
 
+OpObserver::DeleteState OpObserver::aboutToDelete(OperationContext* txn,
+                                                  const NamespaceString& ns,
+                                                  const BSONObj& doc) {
+    OpObserver::DeleteState deleteState;
+    BSONElement idElement = doc["_id"];
+    if (!idElement.eoo()) {
+        deleteState.idDoc = idElement.wrap();
+    }
+    deleteState.isMigrating = isInMigratingChunk(txn, ns, doc);
+    return deleteState;
+}
+
 void OpObserver::onDelete(OperationContext* txn,
-                          const std::string& ns,
-                          const BSONObj& idDoc,
+                          const NamespaceString& ns,
+                          OpObserver::DeleteState deleteState,
                           bool fromMigrate) {
-    repl::logOp(txn, "d", ns.c_str(), idDoc, nullptr, fromMigrate);
+    if (deleteState.idDoc.isEmpty())
+        return;
 
-    getGlobalAuthorizationManager()->logOp(txn, "d", ns.c_str(), idDoc, nullptr);
-    logOpForSharding(txn, "d", ns.c_str(), idDoc, nullptr, fromMigrate);
-    logOpForDbHash(txn, ns.c_str());
-    if (strstr(ns.c_str(), ".system.js")) {
+    repl::logOp(txn, "d", ns.ns().c_str(), deleteState.idDoc, nullptr, fromMigrate);
+
+    AuthorizationManager::get(txn->getServiceContext())
+        ->logOp(txn, "d", ns.ns().c_str(), deleteState.idDoc, nullptr);
+    logOpForSharding(txn,
+                     "d",
+                     ns.ns().c_str(),
+                     deleteState.idDoc,
+                     nullptr,
+                     fromMigrate || !deleteState.isMigrating);
+    logOpForDbHash(txn, ns.ns().c_str());
+    if (ns.coll() == "system.js") {
         Scope::storedFuncMod(txn);
     }
 }
diff --git a/src/mongo/db/op_observer.h b/src/mongo/db/op_observer.h
index 36dafda30c6..e9c18f4bbe7 100644
--- a/src/mongo/db/op_observer.h
+++ b/src/mongo/db/op_observer.h
@@ -51,6 +51,12 @@ class OpObserver {
 public:
     OpObserver() {}
     ~OpObserver() {}
+
+    struct DeleteState {
+        BSONObj idDoc;
+        bool isMigrating = false;
+    };
+
     void onCreateIndex(OperationContext* txn,
                        const std::string& ns,
                        BSONObj indexDoc,
@@ -61,9 +67,10 @@ public:
                    std::vector<BSONObj>::const_iterator end,
                    bool fromMigrate = false);
     void onUpdate(OperationContext* txn, oplogUpdateEntryArgs args);
+    DeleteState aboutToDelete(OperationContext* txn, const NamespaceString& ns, const BSONObj& doc);
     void onDelete(OperationContext* txn,
-                  const std::string& ns,
-                  const BSONObj& idDoc,
+                  const NamespaceString& ns,
+                  DeleteState deleteState,
                   bool fromMigrate = false);
     void onOpMessage(OperationContext* txn, const BSONObj& msgObj);
     void onCreateCollection(OperationContext* txn,
diff --git a/src/mongo/db/operation_context.h b/src/mongo/db/operation_context.h
index 619d937d249..3a7e3165dda 100644
--- a/src/mongo/db/operation_context.h
+++ b/src/mongo/db/operation_context.h
@@ -48,7 +48,7 @@ class WriteUnitOfWork;
 
 /**
  * This class encompasses the state required by an operation and lives from the time a network
- * peration is dispatched until its execution is finished. Note that each "getmore" on a cursor
+ * operation is dispatched until its execution is finished. Note that each "getmore" on a cursor
  * is a separate operation. On construction, an OperationContext associates itself with the
  * current client, and only on destruction it deassociates itself. At any time a client can be
  * associated with at most one OperationContext. Each OperationContext has a RecoveryUnit
diff --git a/src/mongo/db/s/migration_destination_manager.cpp b/src/mongo/db/s/migration_destination_manager.cpp
index bf7a483d993..27e767cd6a9 100644
--- a/src/mongo/db/s/migration_destination_manager.cpp
+++ b/src/mongo/db/s/migration_destination_manager.cpp
@@ -161,6 +161,8 @@ bool opReplicatedEnough(OperationContext* txn,
     return majorityStatus.isOK() && userStatus.isOK();
 }
 
+MONGO_FP_DECLARE(failMigrationReceivedOutOfRangeDelete);
+
 }  // namespace
 
 // Enabling / disabling these fail points pauses / resumes MigrateStatus::_go(), the thread which
@@ -825,6 +827,9 @@ bool MigrationDestinationManager::_applyMigrateOp(OperationContext* txn,
             BSONObj fullObj;
             if (Helpers::findById(txn, ctx.db(), ns.c_str(), id, fullObj)) {
                 if (!isInRange(fullObj, min, max, shardKeyPattern)) {
+                    if (MONGO_FAIL_POINT(failMigrationReceivedOutOfRangeDelete)) {
+                        invariant(0);
+                    }
                     continue;
                 }
             }
diff --git a/src/mongo/db/s/migration_source_manager.cpp b/src/mongo/db/s/migration_source_manager.cpp
index f497c3f4d3d..59eb96222fb 100644
--- a/src/mongo/db/s/migration_source_manager.cpp
+++ b/src/mongo/db/s/migration_source_manager.cpp
@@ -234,6 +234,8 @@ void MigrationSourceManager::logOp(OperationContext* txn,
         // won't be transferred to the recipient shard. Also ignore ops from
         // _migrateClone and _transferMods since it is impossible to move a chunk
         // to self.
+        // Also ignore out of range deletes when migrating (notInActiveChunk is set in
+        // OpObserver::onDelete)
         return;
     }
 
@@ -283,11 +285,17 @@ void MigrationSourceManager::logOp(OperationContext* txn,
         }
     }
 
-    // Note: can't check if delete is in active chunk since the document is gone!
-
     txn->recoveryUnit()->registerChange(new LogOpForShardingHandler(this, idObj, op));
 }
 
+bool MigrationSourceManager::isInMigratingChunk(const NamespaceString& ns, const BSONObj& doc) {
+    if (!_active)
+        return false;
+    if (ns != _nss)
+        return false;
+    return isInRange(doc, _min, _max, _shardKeyPattern);
+}
+
 bool MigrationSourceManager::transferMods(OperationContext* txn,
                                           string& errmsg,
                                           BSONObjBuilder& b) {
diff --git a/src/mongo/db/s/migration_source_manager.h b/src/mongo/db/s/migration_source_manager.h
index 4921610332d..8c6251fe49b 100644
--- a/src/mongo/db/s/migration_source_manager.h
+++ b/src/mongo/db/s/migration_source_manager.h
@@ -71,6 +71,8 @@ public:
                BSONObj* patt,
                bool notInActiveChunk);
 
+    bool isInMigratingChunk(const NamespaceString& ns, const BSONObj& doc);
+
     /**
      * Called from the source of a migration process, this method transfers the accummulated local
      * mods from source to destination.
diff --git a/src/mongo/s/d_migrate.cpp b/src/mongo/s/d_migrate.cpp
index 8d1d2e783f6..6a7abe7f723 100644
--- a/src/mongo/s/d_migrate.cpp
+++ b/src/mongo/s/d_migrate.cpp
@@ -441,4 +441,8 @@ void logOpForSharding(OperationContext* txn,
         shardingState->migrationSourceManager()->logOp(txn, opstr, ns, obj, patt, notInActiveChunk);
 }
 
+bool isInMigratingChunk(OperationContext* txn, const NamespaceString& ns, const BSONObj& doc) {
+    return ShardingState::get(txn)->migrationSourceManager()->isInMigratingChunk(ns, doc);
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/d_state.h b/src/mongo/s/d_state.h
index 8029ef92aef..ce6136ea17c 100644
--- a/src/mongo/s/d_state.h
+++ b/src/mongo/s/d_state.h
@@ -37,6 +37,7 @@ class BSONObj;
 class Client;
 class OperationContext;
 class ShardedConnectionInfo;
+class NamespaceString;
 
 // -----------------
 // --- core ---
@@ -69,4 +70,11 @@ void logOpForSharding(OperationContext* txn,
                       const BSONObj& obj,
                       BSONObj* patt,
                       bool forMigrateCleanup);
+
+/**
+ * Checks if 'doc' in 'ns' belongs to a currently migrating chunk.
+ *
+ * Note: Must be holding global IX lock when calling this method.
+ */
+bool isInMigratingChunk(OperationContext* txn, const NamespaceString& ns, const BSONObj& doc);
 }