SERVER-42500 Trigger best-effort shard refreshes after refineCollectionShardKey

Author: JamesHeppenstall

jstests/sharding/refine_collection_shard_key_basic.js

@@ -86,7 +86,8 @@ function setupCRUDBeforeRefine() {
 }
 
 function validateCRUDAfterRefine() {
-    // Force a refresh on each shard to simulate the asynchronous 'setShardVersion' completing.
+    // Force a refresh on the mongos and each shard because refineCollectionShardKey only triggers
+    // best-effort shard refreshes.
     flushRoutersAndRefreshShardMetadata(st, {ns: kNsName});
 
     const session = mongos.startSession({retryWrites: true});
@@ -592,5 +593,30 @@ assert.eq(3, oldTagsArr.length);
 assert.commandWorked(mongos.adminCommand({refineCollectionShardKey: kNsName, key: newKeyDoc}));
 validateUnrelatedCollAfterRefine(oldCollArr, oldChunkArr, oldTagsArr);
 
+// Verify that all shards containing chunks in the namespace 'db.foo' eventually refresh (i.e. the
+// secondary shard will not refresh because it does not contain any chunks in 'db.foo'). NOTE: This
+// will only succeed in a linear jstest without failovers.
+dropAndReshardColl(oldKeyDoc);
+assert.commandWorked(mongos.getCollection(kNsName).createIndex(newKeyDoc));
+
+assert.commandWorked(
+    mongos.adminCommand({moveChunk: kNsName, find: {a: 0, b: 0}, to: secondaryShard}));
+assert.commandWorked(
+    mongos.adminCommand({moveChunk: kNsName, find: {a: 0, b: 0}, to: primaryShard}));
+
+const oldPrimaryEpoch = st.shard0.adminCommand({getShardVersion: kNsName, fullMetadata: true})
+                            .metadata.shardVersionEpoch.toString();
+const oldSecondaryEpoch = st.shard1.adminCommand({getShardVersion: kNsName, fullMetadata: true})
+                              .metadata.shardVersionEpoch.toString();
+
+assert.commandWorked(mongos.adminCommand({refineCollectionShardKey: kNsName, key: newKeyDoc}));
+
+assert.soon(() => oldPrimaryEpoch !==
+                st.shard0.adminCommand({getShardVersion: kNsName, fullMetadata: true})
+                    .metadata.shardVersionEpoch.toString());
+assert.soon(() => oldSecondaryEpoch ===
+                st.shard1.adminCommand({getShardVersion: kNsName, fullMetadata: true})
+                    .metadata.shardVersionEpoch.toString());
+
 st.stop();
 })();

jstests/sharding/refine_collection_shard_key_drops_chunks.js

@@ -49,17 +49,14 @@ assert.eq({a: 5, b: 5}, chunkArr[1].max);
 assert.eq({a: 5, b: 5}, chunkArr[2]._id);
 assert.eq({a: MaxKey, b: MaxKey}, chunkArr[2].max);
 
-assert.commandWorked(mongos.adminCommand({refineCollectionShardKey: kNsName, key: newKeyDoc}));
-
 // Enable failpoint 'hangPersistCollectionAndChangedChunksAfterDropChunks' and flush the routing
 // table cache.
 assert.commandWorked(shard.adminCommand({
     configureFailPoint: 'hangPersistCollectionAndChangedChunksAfterDropChunks',
     mode: 'alwaysOn'
 }));
-const awaitShellToFlushRoutingTableCacheUpdates = startParallelShell(() => {
-    assert.commandWorked(db.adminCommand({_flushRoutingTableCacheUpdates: 'db.foo'}));
-}, st.rs0.getPrimary().port);
+
+assert.commandWorked(mongos.adminCommand({refineCollectionShardKey: kNsName, key: newKeyDoc}));
 
 // Verify that all chunks belonging to 'db.foo' have been deleted.
 waitForFailpoint('Hit hangPersistCollectionAndChangedChunksAfterDropChunks', 1);
@@ -70,11 +67,13 @@ assert.eq(0, chunkArr.length);
 // flushing the routing table cache.
 assert.commandWorked(shard.adminCommand(
     {configureFailPoint: 'hangPersistCollectionAndChangedChunksAfterDropChunks', mode: 'off'}));
-awaitShellToFlushRoutingTableCacheUpdates();
 
-// Verify that 'config.cache.chunks.db.foo' is as expected after refineCollectionShardKey.
-chunkArr = shard.getCollection(kConfigCacheChunks).find({}).sort({min: 1}).toArray();
-assert.eq(3, chunkArr.length);
+// Verify that 'config.cache.chunks.db.foo' is as expected after refineCollectionShardKey. NOTE: We
+// use assert.soon here because refineCollectionShardKey doesn't block for each shard to refresh.
+assert.soon(() => {
+    chunkArr = shard.getCollection(kConfigCacheChunks).find({}).sort({min: 1}).toArray();
+    return (3 === chunkArr.length);
+});
 assert.eq({a: MinKey, b: MinKey, c: MinKey, d: MinKey}, chunkArr[0]._id);
 assert.eq({a: 0, b: 0, c: MinKey, d: MinKey}, chunkArr[0].max);
 assert.eq({a: 0, b: 0, c: MinKey, d: MinKey}, chunkArr[1]._id);

src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp

@@ -63,6 +63,7 @@
 #include "mongo/s/client/shard.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
+#include "mongo/s/request_types/flush_routing_table_cache_updates_gen.h"
 #include "mongo/s/request_types/set_shard_version_request.h"
 #include "mongo/s/shard_key_pattern.h"
 #include "mongo/s/shard_util.h"
@@ -149,6 +150,41 @@ void writeFirstChunksForCollection(OperationContext* opCtx,
     }
 }
 
+void triggerFireAndForgetShardRefreshes(OperationContext* opCtx, const NamespaceString& nss) {
+    const auto shardRegistry = Grid::get(opCtx)->shardRegistry();
+    const auto allShards = uassertStatusOK(Grid::get(opCtx)->catalogClient()->getAllShards(
+                                               opCtx, repl::ReadConcernLevel::kLocalReadConcern))
+                               .value;
+
+    for (const auto& shardEntry : allShards) {
+        const auto chunk = uassertStatusOK(shardRegistry->getConfigShard()->exhaustiveFindOnConfig(
+                                               opCtx,
+                                               ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+                                               repl::ReadConcernLevel::kLocalReadConcern,
+                                               ChunkType::ConfigNS,
+                                               BSON(ChunkType::ns(nss.ns())
+                                                    << ChunkType::shard(shardEntry.getName())),
+                                               BSONObj(),
+                                               1LL))
+                               .docs;
+
+        invariant(chunk.size() == 0 || chunk.size() == 1);
+
+        if (chunk.size() == 1) {
+            const auto shard =
+                uassertStatusOK(shardRegistry->getShard(opCtx, shardEntry.getName()));
+
+            // This is a best-effort attempt to refresh the shard 'shardEntry'. Fire and forget an
+            // asynchronous '_flushRoutingTableCacheUpdates' request.
+            shard->runFireAndForgetCommand(
+                opCtx,
+                ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+                NamespaceString::kAdminDb.toString(),
+                BSON(_flushRoutingTableCacheUpdates::kCommandName << nss.ns()));
+        }
+    }
+}
+
 }  // namespace
 
 void checkForExistingChunks(OperationContext* opCtx, const NamespaceString& nss) {
@@ -760,7 +796,8 @@ void ShardingCatalogManager::refineCollectionShardKey(OperationContext* opCtx,
 
     // Update all config.tags entries for the given namespace by setting their bounds for each new
     // field in the refined key to MinKey (except for the global max tag where the max bounds are
-    // set to MaxKey).
+    // set to MaxKey). NOTE: The last update has majority write concern to ensure that all updates
+    // are majority committed before refreshing each shard.
     uassertStatusOK(
         catalogClient->updateConfigDocuments(opCtx,
                                              TagsType::ConfigNS,
@@ -769,12 +806,13 @@ void ShardingCatalogManager::refineCollectionShardKey(OperationContext* opCtx,
                                              false,  // upsert
                                              ShardingCatalogClient::kLocalWriteConcern));
 
-    uassertStatusOK(catalogClient->updateConfigDocument(opCtx,
-                                                        TagsType::ConfigNS,
-                                                        isGlobalMaxQuery,
-                                                        BSON("$max" << globalMaxBounds),
-                                                        false,  // upsert
-                                                        ShardingCatalogClient::kLocalWriteConcern));
+    uassertStatusOK(
+        catalogClient->updateConfigDocument(opCtx,
+                                            TagsType::ConfigNS,
+                                            isGlobalMaxQuery,
+                                            BSON("$max" << globalMaxBounds),
+                                            false,  // upsert
+                                            ShardingCatalogClient::kMajorityWriteConcern));
 
     log() << "refineCollectionShardKey: updated zone entries for '" << nss.ns() << "': took "
           << executionTimer.millis() << " ms. Total time taken: " << totalTimer.millis() << " ms.";
@@ -784,6 +822,17 @@ void ShardingCatalogManager::refineCollectionShardKey(OperationContext* opCtx,
                                            nss.ns(),
                                            BSONObj(),
                                            ShardingCatalogClient::kLocalWriteConcern);
+
+    // Trigger refreshes on each shard containing chunks in the namespace 'nss'. Since this isn't
+    // necessary for correctness, all refreshes are best-effort.
+    try {
+        triggerFireAndForgetShardRefreshes(opCtx, nss);
+    } catch (const DBException& ex) {
+        log() << ex.toStatus().withContext(str::stream()
+                                           << "refineCollectionShardKey: failed to best-effort "
+                                              "refresh all shards containing chunks in '"
+                                           << nss.ns() << "'");
+    }
 }
 
 }  // namespace mongo