SERVER-54874: Ensure reading consistent config.collections and config.chunks when refreshing the CatalogCache

(cherry picked from commit 538cb2ff982aab2a60c4f3ab53eb1ba6d1acfeb9)

1 files changed, 31 insertions, 0 deletions

diff --git a/src/mongo/db/rs_local_client.cpp b/src/mongo/db/rs_local_client.cpp
index b1d8894d57a..975be198be0 100644
--- a/src/mongo/db/rs_local_client.cpp
+++ b/src/mongo/db/rs_local_client.cpp
@@ -156,4 +156,35 @@ StatusWith<Shard::QueryResponse> RSLocalClient::queryOnce(
     }
 }
 
+Status RSLocalClient::runAggregation(
+    OperationContext* opCtx,
+    const AggregateCommand& aggRequest,
+    std::function<bool(const std::vector<BSONObj>& batch)> callback) {
+    DBDirectClient client(opCtx);
+    auto cursor = uassertStatusOKWithContext(
+        DBClientCursor::fromAggregationRequest(
+            &client, aggRequest, true /* secondaryOk */, true /* useExhaust */),
+        "Failed to establish a cursor for aggregation");
+
+    while (cursor->more()) {
+        std::vector<BSONObj> batchDocs;
+        batchDocs.reserve(cursor->objsLeftInBatch());
+        while (cursor->moreInCurrentBatch()) {
+            batchDocs.emplace_back(cursor->nextSafe().getOwned());
+        }
+
+        try {
+            if (!callback(batchDocs)) {
+                break;
+            }
+        } catch (const DBException& ex) {
+            return ex
+                .toStatus(str::stream()
+                          << "Exception while running aggregation retrieval of results callback");
+        }
+    }
+
+    return Status::OK();
+}
+
 }  // namespace mongo