RUBY-3820 Send afterClusterTime on writes in causally-consistent sessions

lib/mongo/index/view.rb

@@ -350,7 +350,9 @@ def operation_timeouts(opts = {})
       private
 
       def drop_by_name(name, opts = {})
-        client.send(:with_session, @options) do |session|
+        session_options = @options.dup
+        session_options[:session] = opts[:session] if opts[:session]
+        client.send(:with_session, session_options) do |session|
           spec = {
             db_name: database.name,
             coll_name: collection.name,

lib/mongo/operation/create/op_msg.rb

@@ -24,6 +24,7 @@ class Create
       # @since 2.5.2
       class OpMsg < OpMsgBase
         include ExecutableTransactionLabel
+        include CausalConsistencySupported
 
         private
 

lib/mongo/operation/create_index/op_msg.rb

@@ -24,6 +24,7 @@ class CreateIndex
       # @since 2.5.2
       class OpMsg < OpMsgBase
         include ExecutableTransactionLabel
+        include CausalConsistencySupported
 
         private
 

lib/mongo/operation/delete/op_msg.rb

@@ -27,6 +27,7 @@ class OpMsg < OpMsgBase
         include ExecutableNoValidate
         include ExecutableTransactionLabel
         include PolymorphicResult
+        include CausalConsistencySupported
 
         private
 

lib/mongo/operation/drop/op_msg.rb

@@ -24,6 +24,7 @@ class Drop
       # @since 2.5.2
       class OpMsg < OpMsgBase
         include ExecutableTransactionLabel
+        include CausalConsistencySupported
       end
     end
   end

lib/mongo/operation/drop_database/op_msg.rb

@@ -24,6 +24,7 @@ class DropDatabase
       # @since 2.5.2
       class OpMsg < OpMsgBase
         include ExecutableTransactionLabel
+        include CausalConsistencySupported
       end
     end
   end

lib/mongo/operation/drop_index/op_msg.rb

@@ -24,6 +24,7 @@ class DropIndex
       # @since 2.5.2
       class OpMsg < OpMsgBase
         include ExecutableTransactionLabel
+        include CausalConsistencySupported
 
         private
 

lib/mongo/operation/insert/op_msg.rb

@@ -28,6 +28,7 @@ class OpMsg < OpMsgBase
         include ExecutableNoValidate
         include ExecutableTransactionLabel
         include PolymorphicResult
+        include CausalConsistencySupported
 
         private
 

lib/mongo/operation/shared/causal_consistency_supported.rb

@@ -28,8 +28,10 @@ module CausalConsistencySupported
       #
       # This method overrides the causal consistency addition logic of
       # SessionsSupported and is meant to be used with operations classified
-      # as "read operations accepting a read concern", as these are defined
-      # in the causal consistency spec.
+      # as "read and write operations accepting a read concern", as these are
+      # defined in the causal consistency spec. For write operations this
+      # attaches only afterClusterTime (no read concern level) so that causal
+      # ordering is preserved across writes in a causally consistent session.
       #
       # In order for the override to work correctly the
       # CausalConsistencySupported module must be included after

lib/mongo/operation/update/op_msg.rb

@@ -27,6 +27,7 @@ class OpMsg < OpMsgBase
         include ExecutableNoValidate
         include ExecutableTransactionLabel
         include PolymorphicResult
+        include CausalConsistencySupported
 
         private
 

lib/mongo/operation/write_command/op_msg.rb

@@ -21,6 +21,8 @@ class WriteCommand
       #
       # @api private
       class OpMsg < OpMsgBase
+        include CausalConsistencySupported
+
         private
 
         def selector(connection)

spec/runners/unified/crud_operations.rb

@@ -216,23 +216,32 @@ def update_many(op)
           timeout_ms: args.use('timeoutMS'),
           max_time_ms: args.use('maxTimeMS')
         }
+        if (session = args.use('session'))
+          opts[:session] = entities.get(:session, session)
+        end
         collection.update_many(args.use!('filter'), args.use!('update'), **opts)
       end
     end
 
     def replace_one(op)
       collection = entities.get(:collection, op.use!('object'))
       use_arguments(op) do |args|
-        collection.replace_one(
-          args.use!('filter'),
-          args.use!('replacement'),
+        opts = {
           comment: args.use('comment'),
           upsert: args.use('upsert'),
           let: args.use('let'),
           hint: args.use('hint'),
           timeout_ms: args.use('timeoutMS'),
           max_time_ms: args.use('maxTimeMS'),
           sort: args.use('sort')
+        }
+        if (session = args.use('session'))
+          opts[:session] = entities.get(:session, session)
+        end
+        collection.replace_one(
+          args.use!('filter'),
+          args.use!('replacement'),
+          **opts
         )
       end
     end
@@ -264,6 +273,9 @@ def delete_many(op)
           timeout_ms: args.use('timeoutMS'),
           max_time_ms: args.use('maxTimeMS')
         }
+        if (session = args.use('session'))
+          opts[:session] = entities.get(:session, session)
+        end
         collection.delete_many(args.use!('filter'), **opts)
       end
     end
@@ -276,6 +288,9 @@ def bulk_write(op)
         end
         opts = {}
         opts[:ordered] = args.use!('ordered') if args.key?('ordered')
+        if (session = args.use('session'))
+          opts[:session] = entities.get(:session, session)
+        end
         if (comment = args.use('comment'))
           opts[:comment] = comment
         end

spec/runners/unified/ddl_operations.rb

@@ -97,8 +97,23 @@ def list_colls(op, name_only: false)
     def drop_collection(op)
       database = entities.get(:database, op.use!('object'))
       use_arguments(op) do |args|
+        opts = {}
+        if (session = args.use('session'))
+          opts[:session] = entities.get(:session, session)
+        end
         collection = database[args.use!('collection')]
-        collection.drop
+        collection.drop(**opts)
+      end
+    end
+
+    def drop_database(op)
+      client = entities.get(:client, op.use!('object'))
+      use_arguments(op) do |args|
+        opts = {}
+        if (session = args.use('session'))
+          opts[:session] = entities.get(:session, session)
+        end
+        client.use(args.use!('database')).database.drop(**opts)
       end
     end
 
@@ -158,6 +173,9 @@ def drop_indexes(op)
       collection = entities.get(:collection, op.use!('object'))
       use_arguments(op) do |args|
         opts = extract_options(args, 'maxTimeMS', 'timeoutMS', allow_extra: true)
+        if (session = args.use('session'))
+          opts[:session] = entities.get(:session, session)
+        end
         collection.indexes.drop_all(**opts)
       end
     end

spec/spec_tests/causal_consistency_unified_spec.rb

@@ -0,0 +1,12 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+require 'runners/unified'
+
+base = "#{CURRENT_PATH}/spec_tests/data/causal_consistency_unified"
+CAUSAL_CONSISTENCY_UNIFIED_TESTS = Dir.glob("#{base}/**/*.yml").sort
+
+describe 'Causal consistency unified spec tests' do
+  define_unified_spec_tests(base, CAUSAL_CONSISTENCY_UNIFIED_TESTS)
+end

spec/spec_tests/data/causal_consistency_unified/causal-consistency-clientBulkWrite.yml

@@ -0,0 +1,75 @@
+description: "causal consistency bulkWrite include afterClusterTime"
+
+schemaVersion: "1.3"
+
+runOnRequirements:
+  - minServerVersion: "8.0"
+    topologies: [replicaset, sharded, load-balanced]
+
+createEntities:
+  - client:
+      id: &client0 client0
+      useMultipleMongoses: false
+      uriOptions:
+        retryWrites: false
+      observeEvents: [commandStartedEvent]
+  - database:
+      id: &database0 database0
+      client: *client0
+      databaseName: &databaseName causal-consistency-tests
+  - collection:
+      id: &collection0 collection0
+      database: *database0
+      collectionName: &collectionName test
+  - session:
+      id: &session0 session0
+      client: *client0
+      sessionOptions:
+        causalConsistency: true
+
+initialData:
+  - collectionName: *collectionName
+    databaseName: *databaseName
+    documents:
+      - { _id: 1, x: 11 }
+      - { _id: 2, x: 22 }
+      - { _id: 3, x: 33 }
+
+# In a causally consistent session, once an operationTime has been established by a prior
+# operation, subsequent write commands MUST include readConcern.afterClusterTime so the
+# server can apply the write causally after the previously-observed data.
+
+tests:
+  - description: "clientBulkWrite includes afterClusterTime in causally consistent session"
+    operations:
+      - name: find
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+        expectResult: [{ _id: 1, x: 11 }]
+      - name: clientBulkWrite
+        object: *client0
+        arguments:
+          session: *session0
+          models:
+          - insertOne:
+              namespace: causal-consistency-tests.test
+              document: { _id: 4 }
+    expectEvents:
+      - client: *client0
+        events:
+          - commandStartedEvent:
+              commandName: find
+              command:
+                find: *collectionName
+                readConcern: { $$exists: false }
+                lsid: { $$sessionLsid: *session0 }
+          - commandStartedEvent:
+              commandName: bulkWrite
+              command:
+                bulkWrite: 1
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }