Sync CSOT spec tests (mongodb#2872)

Co-authored-by: Jamis Buck <[email protected]>

lib/mongo/database/view.rb

@@ -123,7 +123,7 @@ def collection_names(options = {})
       #
       # @since 2.0.5
       def list_collections(options = {})
-        session = client.send(:get_session, options)
+        session = client.get_session(options)
         collections_info(session, ServerSelector.primary, options)
       end
 

lib/mongo/index/view.rb

@@ -100,7 +100,7 @@ class View
       # @since 2.0.0
       def drop_one(name, options = {})
         raise Error::MultiIndexDrop.new if name == Index::ALL
-        drop_by_name(name, comment: options[:comment])
+        drop_by_name(name, options)
       end
 
       # Drop all indexes on the collection.
@@ -117,7 +117,7 @@ def drop_one(name, options = {})
       #
       # @since 2.0.0
       def drop_all(options = {})
-        drop_by_name(Index::ALL, comment: options[:comment])
+        drop_by_name(Index::ALL, options)
       end
 
       # Creates an index on the collection.
@@ -171,7 +171,7 @@ def create_one(keys, options = {})
         if session = @options[:session]
           create_options[:session] = session
         end
-        %i(commit_quorum session comment).each do |key|
+        %i(commit_quorum session comment timeout_ms max_time_ms).each do |key|
           if value = options.delete(key)
             create_options[key] = value
           end
@@ -220,7 +220,7 @@ def create_many(*models)
           options = models.pop
         end
 
-        client.send(:with_session, @options.merge(options)) do |session|
+        client.with_session(@options.merge(options)) do |session|
           server = next_primary(nil, session)
 
           indexes = normalize_models(models, server)
@@ -239,8 +239,12 @@ def create_many(*models)
             write_concern: write_concern,
             comment: options[:comment],
           }
-
-          Operation::CreateIndex.new(spec).execute(server, context: Operation::Context.new(client: client, session: session))
+          context = Operation::Context.new(
+            client: client,
+            session: session,
+            operation_timeouts: operation_timeouts(options)
+          )
+          Operation::CreateIndex.new(spec).execute(server, context: context)
         end
       end
 
@@ -343,7 +347,7 @@ def operation_timeouts(opts = {})
 
       private
 
-      def drop_by_name(name, comment: nil)
+      def drop_by_name(name, opts = {})
         client.send(:with_session, @options) do |session|
           spec = {
             db_name: database.name,
@@ -352,9 +356,14 @@ def drop_by_name(name, comment: nil)
             session: session,
             write_concern: write_concern,
           }
-          spec[:comment] = comment unless comment.nil?
+          spec[:comment] = opts[:comment] unless opts[:comment].nil?
           server = next_primary(nil, session)
-          Operation::DropIndex.new(spec).execute(server, context: Operation::Context.new(client: client, session: session))
+          context = Operation::Context.new(
+            client: client,
+            session: session,
+            operation_timeouts: operation_timeouts(opts)
+          )
+          Operation::DropIndex.new(spec).execute(server, context: context)
         end
       end
 

lib/mongo/operation/shared/sessions_supported.rb

@@ -114,7 +114,7 @@ def apply_read_pref!(selector)
       end
 
       def apply_txn_opts!(selector)
-        session.add_txn_opts!(selector, read_command?(selector))
+        session.add_txn_opts!(selector, read_command?(selector), context)
       end
 
       def suppress_read_write_concern!(selector)

lib/mongo/session.rb

@@ -936,7 +936,7 @@ def add_txn_num!(command)
     #
     # @since 2.6.0
     # @api private
-    def add_txn_opts!(command, read)
+    def add_txn_opts!(command, read, context)
       command.tap do |c|
         # The read concern should be added to any command that starts a transaction.
         if starting_transaction?
@@ -990,6 +990,14 @@ def add_txn_opts!(command, read)
         if c[:writeConcern] && c[:writeConcern][:w] && c[:writeConcern][:w].is_a?(Symbol)
           c[:writeConcern][:w] = c[:writeConcern][:w].to_s
         end
+
+        # Ignore wtimeout if csot
+        if context&.csot?
+          c[:writeConcern]&.delete(:wtimeout)
+        end
+
+        # We must not send an empty (server default) write concern.
+        c.delete(:writeConcern) if c[:writeConcern]&.empty?
       end
     end
 

spec/lite_spec_helper.rb

@@ -168,16 +168,6 @@ def require_atlas
     end
   end
 
-  if SpecConfig.instance.ci? && !%w(1 true yes).include?(ENV['INTERACTIVE']&.downcase)
-    # Tests should take under 10 seconds ideally but it seems
-    # we have some that run for more than 10 seconds in CI.
-    config.around(:each) do |example|
-      TimeoutInterrupt.timeout(example_timeout_seconds, ExampleTimeout) do
-        example.run
-      end
-    end
-  end
-
   if SpecConfig.instance.ci?
     if defined?(Rfc::Rif)
       unless BSON::Environment.jruby?

spec/runners/unified/crud_operations.rb

@@ -23,10 +23,10 @@ def get_find_view(op)
           'allowDiskUse', 'returnKey', 'projection',
           'skip', 'hint', 'maxTimeMS', 'timeoutMS',
           'collation', 'noCursorTimeout', 'oplogReplay', 'allowPartialResults',
-          'timeoutMode',
+          'timeoutMode', 'maxAwaitTimeMS', 'cursorType', 'timeoutMode',
           { 'showRecordId' => :show_disk_loc, 'max' => :max_value, 'min' => :min_value },
           allow_extra: true)
-        symbolize_options!(opts, :timeout_mode)
+        symbolize_options!(opts, :timeout_mode, :cursor_type)
 
         opts[:session] = entities.get(:session, session) if session
 
@@ -50,49 +50,31 @@ def get_find_view(op)
     def count(op)
       collection = entities.get(:collection, op.use!('object'))
       use_arguments(op) do |args|
-        opts = {}
+        opts = extract_options(args, 'comment', 'timeoutMS', 'maxTimeMS', allow_extra: true)
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
         end
-        if comment = args.use('comment')
-          opts[:comment] = comment
-        end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
-        end
         collection.count(args.use!('filter'), **opts)
       end
     end
 
     def count_documents(op)
       collection = entities.get(:collection, op.use!('object'))
       use_arguments(op) do |args|
-        opts = {}
+        opts = extract_options(args, 'comment', 'timeoutMS', 'maxTimeMS', allow_extra: true)
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
         end
-        if comment = args.use('comment')
-          opts[:comment] = comment
-        end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
-        end
         collection.find(args.use!('filter')).count_documents(**opts)
       end
     end
 
     def estimated_document_count(op)
       collection = entities.get(:collection, op.use!('object'))
       use_arguments(op) do |args|
-        opts = {}
-        if max_time_ms = args.use('maxTimeMS')
-          opts[:max_time_ms] = max_time_ms
-        end
-        if comment = args.use('comment')
-          opts[:comment] = comment
-        end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
+        opts = extract_options(args, 'comment', 'timeoutMS', 'maxTimeMS', allow_extra: true)
+        if session = args.use('session')
+          opts[:session] = entities.get(:session, session)
         end
         collection.estimated_document_count(**opts)
       end
@@ -101,16 +83,10 @@ def estimated_document_count(op)
     def distinct(op)
       collection = entities.get(:collection, op.use!('object'))
       use_arguments(op) do |args|
-        opts = {}
+        opts = extract_options(args, 'comment', 'timeoutMS', 'maxTimeMS', allow_extra: true)
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
         end
-        if comment = args.use('comment')
-          opts[:comment] = comment
-        end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
-        end
         req = collection.find(args.use!('filter'), **opts).distinct(args.use!('fieldName'), **opts)
         result = req.to_a
       end
@@ -126,16 +102,15 @@ def find_one_and_update(op)
           comment: args.use('comment'),
           hint: args.use('hint'),
           upsert: args.use('upsert'),
+          timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         if return_document = args.use('returnDocument')
           opts[:return_document] = return_document.downcase.to_sym
         end
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
         end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
-        end
         collection.find_one_and_update(filter, update, **opts)
       end
     end
@@ -149,13 +124,12 @@ def find_one_and_replace(op)
           let: args.use('let'),
           comment: args.use('comment'),
           hint: args.use('hint'),
+          timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
         end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
-        end
         collection.find_one_and_replace(filter, update, **opts)
       end
     end
@@ -168,13 +142,12 @@ def find_one_and_delete(op)
           let: args.use('let'),
           comment: args.use('comment'),
           hint: args.use('hint'),
+          timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
         end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
-        end
         collection.find_one_and_delete(filter, **opts)
       end
     end
@@ -184,7 +157,8 @@ def insert_one(op)
       use_arguments(op) do |args|
         opts = {
           comment: args.use('comment'),
-          timeout_ms: args.use('timeoutMS')
+          timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
@@ -197,17 +171,16 @@ def insert_many(op)
       collection = entities.get(:collection, op.use!('object'))
       use_arguments(op) do |args|
         opts = {
-          comment: args.use('comment')
+          comment: args.use('comment'),
+          timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         unless (ordered = args.use('ordered')).nil?
           opts[:ordered] = ordered
         end
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
         end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
-        end
         collection.insert_many(args.use!('documents'), **opts)
       end
     end
@@ -221,6 +194,7 @@ def update_one(op)
           hint: args.use('hint'),
           upsert: args.use('upsert'),
           timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
@@ -237,6 +211,7 @@ def update_many(op)
           comment: args.use('comment'),
           hint: args.use('hint'),
           timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         collection.update_many(args.use!('filter'), args.use!('update'), **opts)
       end
@@ -252,7 +227,8 @@ def replace_one(op)
           upsert: args.use('upsert'),
           let: args.use('let'),
           hint: args.use('hint'),
-          timeout_ms: args.use('timeoutMS')
+          timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         )
       end
     end
@@ -264,13 +240,12 @@ def delete_one(op)
           let: args.use('let'),
           comment: args.use('comment'),
           hint: args.use('hint'),
+          timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         if session = args.use('session')
           opts[:session] = entities.get(:session, session)
         end
-        if timeout_ms = args.use('timeoutMS')
-          opts[:timeout_ms] = timeout_ms
-        end
         collection.delete_one(args.use!('filter'), **opts)
       end
     end
@@ -282,6 +257,8 @@ def delete_many(op)
           let: args.use('let'),
           comment: args.use('comment'),
           hint: args.use('hint'),
+          timeout_ms: args.use('timeoutMS'),
+          max_time_ms: args.use('maxTimeMS')
         }
         collection.delete_many(args.use!('filter'), **opts)
       end
@@ -306,6 +283,9 @@ def bulk_write(op)
         if timeout_ms = args.use('timeoutMS')
           opts[:timeout_ms] = timeout_ms
         end
+        if max_time_ms = args.use('maxTimeMS')
+          opts[:max_time_ms] = max_time_ms
+        end
         collection.bulk_write(requests, **opts)
       end
     end
@@ -315,18 +295,14 @@ def aggregate(op)
       args = op.use!('arguments')
       pipeline = args.use!('pipeline')
 
-      opts = extract_options(args, 'let', 'comment', 'batchSize',
-        'allowDiskUse', 'timeoutMode', 'timeoutMS', allow_extra: true)
+      opts = extract_options(args, 'let', 'comment', 'batchSize', 'maxTimeMS',
+        'allowDiskUse', 'timeoutMode', 'timeoutMS', 'maxTimeMS', allow_extra: true)
       symbolize_options!(opts, :timeout_mode)
 
       if session = args.use('session')
         opts[:session] = entities.get(:session, session)
       end
 
-      if timeout_ms = args.use('timeoutMS')
-        opts[:timeout_ms] = timeout_ms
-      end
-
       unless args.empty?
         raise NotImplementedError, "Unhandled spec keys: #{args} in #{test_spec}"
       end
@@ -339,7 +315,8 @@ def create_find_cursor(op)
       args = op.use!('arguments')
 
       filter = args.use('filter')
-      opts = extract_options(args, 'batchSize', 'timeoutMS')
+      opts = extract_options(args, 'batchSize', 'timeoutMS', 'cursorType', 'maxAwaitTimeMS')
+      symbolize_options!(opts, :cursor_type)
 
       view = obj.find(filter, opts)
       view.each # to initialize the cursor