SERVER-60086 Add read concern check for aggregate commands.

5 files changed, 32 insertions, 6 deletions

diff --git a/jstests/change_streams/ban_from_lookup.js b/jstests/change_streams/ban_from_lookup.js
index 9e2f6ee8c1b..454dd3e9e4e 100644
--- a/jstests/change_streams/ban_from_lookup.js
+++ b/jstests/change_streams/ban_from_lookup.js
@@ -1,5 +1,9 @@
 /**
  * Test that the $changeStream stage cannot be used in a $lookup pipeline or sub-pipeline.
+ *
+ * @tags: [
+ *   change_stream_does_not_expect_txns,
+ * ]
  */
 (function() {
 "use strict";
diff --git a/jstests/noPassthrough/set_window_fields_read_concern_snapshot.js b/jstests/noPassthrough/set_window_fields_read_concern_snapshot.js
index 1668dc899d7..8f19b2c9864 100644
--- a/jstests/noPassthrough/set_window_fields_read_concern_snapshot.js
+++ b/jstests/noPassthrough/set_window_fields_read_concern_snapshot.js
@@ -63,8 +63,7 @@ aggregationCommand = {
     allowDiskUse: true,
     cursor: {},
 };
-assert.commandFailedWithCode(sessionColl.runCommand(aggregationCommand),
-                             ErrorCodes.OperationNotSupportedInTransaction);
+assert.commandFailedWithCode(sessionColl.runCommand(aggregationCommand), ErrorCodes.InvalidOptions);
 // Transaction state is now unusual, abort it and start a new one.
 session.abortTransaction();
 session.startTransaction({readConcern: {level: "snapshot"}});
@@ -75,7 +74,6 @@ aggregationCommand = {
     allowDiskUse: true,
     cursor: {}
 };
-assert.commandFailedWithCode(sessionColl.runCommand(aggregationCommand),
-                             ErrorCodes.OperationNotSupportedInTransaction);
+assert.commandFailedWithCode(sessionColl.runCommand(aggregationCommand), ErrorCodes.InvalidOptions);
 rst.stopSet();
 })();
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 23099af5e82..2167b3234d9 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -577,10 +577,11 @@ Status runAggregate(OperationContext* opCtx,
     boost::intrusive_ptr<ExpressionContext> expCtx;
     auto curOp = CurOp::get(opCtx);
     {
-        // If we are in a transaction, check whether the parsed pipeline supports
-        // being in a transaction.
+        // If we are in a transaction, check whether the parsed pipeline supports being in
+        // a transaction and if the transaction's read concern is supported.
         if (opCtx->inMultiDocumentTransaction()) {
             liteParsedPipeline.assertSupportsMultiDocumentTransaction(request.getExplain());
+            liteParsedPipeline.assertSupportsReadConcern(opCtx, request.getExplain());
         }
 
         const auto& pipelineInvolvedNamespaces = liteParsedPipeline.getInvolvedNamespaces();
diff --git a/src/mongo/db/pipeline/lite_parsed_pipeline.cpp b/src/mongo/db/pipeline/lite_parsed_pipeline.cpp
index 0d68337f3df..36b1d78d9b7 100644
--- a/src/mongo/db/pipeline/lite_parsed_pipeline.cpp
+++ b/src/mongo/db/pipeline/lite_parsed_pipeline.cpp
@@ -106,6 +106,21 @@ void LiteParsedPipeline::assertSupportsMultiDocumentTransaction(
     }
 }
 
+void LiteParsedPipeline::assertSupportsReadConcern(
+    OperationContext* opCtx, boost::optional<ExplainOptions::Verbosity> explain) const {
+    const auto& readConcernArgs = repl::ReadConcernArgs::get(opCtx);
+    auto readConcernSupport = supportsReadConcern(readConcernArgs.getLevel(),
+                                                  readConcernArgs.isImplicitDefault(),
+                                                  explain,
+                                                  serverGlobalParams.enableMajorityReadConcern);
+    if (readConcernArgs.hasLevel()) {
+        if (!readConcernSupport.readConcernSupport.isOK()) {
+            uassertStatusOK(readConcernSupport.readConcernSupport.withContext(
+                "Operation does not support this transaction's read concern"));
+        }
+    }
+}
+
 void LiteParsedPipeline::verifyIsSupported(
     OperationContext* opCtx,
     const std::function<bool(OperationContext*, const NamespaceString&)> isSharded,
@@ -115,6 +130,7 @@ void LiteParsedPipeline::verifyIsSupported(
     const bool inMultiDocumentTransaction = opCtx->inMultiDocumentTransaction();
     if (inMultiDocumentTransaction) {
         assertSupportsMultiDocumentTransaction(explain);
+        assertSupportsReadConcern(opCtx, explain);
     }
     // Verify that no involved namespace is sharded unless allowed by the pipeline.
     for (const auto& nss : getInvolvedNamespaces()) {
diff --git a/src/mongo/db/pipeline/lite_parsed_pipeline.h b/src/mongo/db/pipeline/lite_parsed_pipeline.h
index eecaaa937e1..696c31be0ba 100644
--- a/src/mongo/db/pipeline/lite_parsed_pipeline.h
+++ b/src/mongo/db/pipeline/lite_parsed_pipeline.h
@@ -151,6 +151,13 @@ public:
         boost::optional<ExplainOptions::Verbosity> explain) const;
 
     /**
+     * Verifies that this pipeline is allowed to run with the read concern from the provided opCtx.
+     * Used only when asserting is the desired behavior, otherwise use supportsReadConcern instead.
+     */
+    void assertSupportsReadConcern(OperationContext* opCtx,
+                                   boost::optional<ExplainOptions::Verbosity> explain) const;
+
+    /**
      * Perform checks that verify that the LitePipe is valid. Note that this function must be called
      * before forwarding an aggregation command on an unsharded collection, in order to verify that
      * the involved namespaces are allowed to be sharded.