RUBY-3372 CSOT Cursors (#2863)

* cursors CSOT

* need to add context as parameter

* wups, maxTimeMS, not timeoutMS

* there is no msg variable here

* context#timeout_ms is not actually used anywhere

Author: jamis

lib/mongo/client.rb

@@ -1104,7 +1104,7 @@ def watch(pipeline = [], options = {})
       return use(Database::ADMIN).watch(pipeline, options) unless database.name == Database::ADMIN
 
       view_options = options.dup
-      view_options[:await_data] = true if options[:max_await_time_ms]
+      view_options[:cursor_type] = :tailable_await if options[:max_await_time_ms]
 
       Mongo::Collection::View::ChangeStream.new(
         Mongo::Collection::View.new(self["#{Database::COMMAND}.aggregate"], {}, view_options),

lib/mongo/collection.rb

@@ -165,6 +165,7 @@ def initialize(database, name, options = {})
       @database = database
       @name = name.to_s.freeze
       @options = options.dup
+      @timeout_ms = options.delete(:timeout_ms)
 =begin WriteConcern object support
       if @options[:write_concern].is_a?(WriteConcern::Base)
         # Cache the instance so that we do not needlessly reconstruct it.
@@ -504,6 +505,9 @@ def drop(opts = {})
     # @option options [ Integer ] :skip The number of docs to skip before returning results.
     # @option options [ Hash ] :sort The key and direction pairs by which the result set
     #   will be sorted.
+    # @option options [ :cursor_lifetime | :iteration ] :timeout_mode How to interpret
+    #   :timeout_ms (whether it applies to the lifetime of the cursor, or per
+    #   iteration).
     # @option options [ Integer ] :timeout_ms The per-operation timeout in milliseconds.
     #    Must a positive integer. The default value is unset which means infinite.
     # @option options [ Hash ] :let Mapping of variables to use in the command.
@@ -538,10 +542,6 @@ def find(filter = nil, options = {})
     #   See the server documentation for details.
     # @option options [ Integer ] :max_time_ms The maximum amount of time in
     #   milliseconds to allow the aggregation to run.
-    # @option options [ true | false ] :use_cursor Indicates whether the command
-    #   will request that the server provide results using a cursor. Note that
-    #   as of server version 3.6, aggregations always provide results using a
-    #   cursor and this option is therefore not valid.
     # @option options [ Session ] :session The session to use.
     # @option options [ Integer ] :timeout_ms The per-operation timeout in milliseconds.
     #   Must a positive integer. The default value is unset which means infinite.
@@ -612,6 +612,11 @@ def aggregate(pipeline, options = {})
     #   events included with this flag set are: createIndexes, dropIndexes,
     #   modify, create, shardCollection, reshardCollection,
     #   refineCollectionShardKey.
+    # @option options [ :cursor_lifetime | :iteration ] :timeout_mode How to interpret
+    #   :timeout_ms (whether it applies to the lifetime of the cursor, or per
+    #   iteration).
+    # @option options [ Integer ] :timeout_ms The maximum amount of time to
+    #   allow the query to run, in milliseconds.
     #
     # @note A change stream only allows 'majority' read concern.
     # @note This helper method is preferable to running a raw aggregation with
@@ -622,7 +627,7 @@ def aggregate(pipeline, options = {})
     # @since 2.5.0
     def watch(pipeline = [], options = {})
       view_options = options.dup
-      view_options[:await_data] = true if options[:max_await_time_ms]
+      view_options[:cursor_type] = :tailable_await if options[:max_await_time_ms]
       View::ChangeStream.new(View.new(self, {}, view_options), pipeline, nil, options)
     end
 
@@ -966,12 +971,17 @@ def delete_many(filter = nil, options = {})
     # @option options [ Integer ] :max_time_ms The maximum amount of time to allow the command
     #   to run in milliseconds.
     # @option options [ Session ] :session The session to use.
+    # @option options [ :cursor_lifetime | :iteration ] :timeout_mode How to interpret
+    #   :timeout_ms (whether it applies to the lifetime of the cursor, or per
+    #   iteration).
+    # @option options [ Integer ] :timeout_ms The maximum amount of time to
+    #   allow the query to run, in milliseconds.
     #
     # @return [ Array<Cursor> ] An array of cursors.
     #
     # @since 2.1
     def parallel_scan(cursor_count, options = {})
-      find({}, options).send(:parallel_scan, cursor_count, options)
+      find({}, options).parallel_scan(cursor_count, options)
     end
 
     # Replaces a single document in the collection with the new document.
@@ -1205,14 +1215,14 @@ def system_collection?
     #
     # @api private
     def timeout_ms
-      options[:timeout_ms] || database.timeout_ms
+      @timeout_ms || database.timeout_ms
     end
 
     # @return [ Hash ] timeout_ms value set on the operation level (if any),
     #   and/or timeout_ms that is set on collection/database/client level (if any).
     #
     # @api private
-    def operation_timeouts(opts)
+    def operation_timeouts(opts = {})
       # TODO: We should re-evaluate if we need two timeouts separately.
       {}.tap do |result|
         if opts[:timeout_ms].nil?

lib/mongo/collection/view.rb

@@ -66,7 +66,6 @@ class View
                      :nro_write_with_retry,
                      :read_with_retry,
                      :read_with_retry_cursor,
-                     :timeout_ms,
                      :write_with_retry,
                      :write_concern_with_session
 
@@ -75,6 +74,12 @@ class View
 
       alias :selector :filter
 
+      # @return [ Integer | nil | The timeout_ms value that was passed as an
+      #   option to the view.
+      #
+      # @api private
+      attr_reader :operation_timeout_ms
+
       # Compare two +View+ objects.
       #
       # @example Compare the view with another object.
@@ -152,17 +157,24 @@ def hash
       #   document more than once. Deprecated as of MongoDB server version 4.0.
       # @option options [ Hash ] :sort The key and direction pairs used to sort
       #   the results.
+      # @option options [ :cursor_lifetime | :iteration ] :timeout_mode How to interpret
+      #   :timeout_ms (whether it applies to the lifetime of the cursor, or per
+      #   iteration).
       # @option options [ Integer ] :timeout_ms The per-operation timeout in milliseconds.
       #   Must a positive integer. The default value is unset which means infinite.
       #
       # @since 2.0.0
       def initialize(collection, filter = {}, options = {})
         validate_doc!(filter)
-        @collection = collection
 
         filter = BSON::Document.new(filter)
         options = BSON::Document.new(options)
 
+        @collection = collection
+        @operation_timeout_ms = options.delete(:timeout_ms)
+
+        validate_timeout_mode!(options)
+
         # This is when users pass $query in filter and other modifiers
         # alongside?
         query = filter.delete(:$query)
@@ -174,6 +186,14 @@ def initialize(collection, filter = {}, options = {})
         @options = Operation::Find::Builder::Modifiers.map_driver_options(modifiers).merge!(options).freeze
       end
 
+      # The timeout_ms value to use for this operation; either specified as an
+      # option to the view, or inherited from the collection.
+      #
+      # @return [ Integer | nil ] the timeout_ms for this operation
+      def timeout_ms
+        operation_timeout_ms || collection.timeout_ms
+      end
+
       # Get a human-readable string representation of +View+.
       #
       # @example Get the inspection.
@@ -199,6 +219,20 @@ def write_concern
         WriteConcern.get(options[:write_concern] || options[:write] || collection.write_concern)
       end
 
+      # @return [ Hash ] timeout_ms value set on the operation level (if any),
+      #   and/or timeout_ms that is set on collection/database/client level (if any).
+      #
+      # @api private
+      def operation_timeouts(opts = {})
+        {}.tap do |result|
+          if opts[:timeout_ms] || operation_timeout_ms
+            result[:operation_timeout_ms] = opts[:timeout_ms] || operation_timeout_ms
+          else
+            result[:inherited_timeout_ms] = collection.timeout_ms
+          end
+        end
+      end
+
       private
 
       def initialize_copy(other)
@@ -208,27 +242,14 @@ def initialize_copy(other)
       end
 
       def new(options)
+        options = options.merge(timeout_ms: operation_timeout_ms) if operation_timeout_ms
         View.new(collection, filter, options)
       end
 
       def view; self; end
 
       def with_session(opts = {}, &block)
-        client.send(:with_session, @options.merge(opts), &block)
-      end
-
-      # @return [ Hash ] timeout_ms value set on the operation level (if any),
-      #   and/or timeout_ms that is set on collection/database/client level (if any).
-      #
-      # @api private
-      def operation_timeouts(opts)
-        {}.tap do |result|
-          if opts[:timeout_ms].nil?
-            result[:inherited_timeout_ms] = collection.timeout_ms
-          else
-            result[:operation_timeout_ms] = opts[:timeout_ms]
-          end
-        end
+        client.with_session(@options.merge(opts), &block)
       end
     end
   end

lib/mongo/collection/view/aggregation.rb

@@ -37,7 +37,7 @@ class Aggregation
         attr_reader :pipeline
 
         # Delegate necessary operations to the view.
-        def_delegators :view, :collection, :read, :cluster
+        def_delegators :view, :collection, :read, :cluster, :cursor_type
 
         # Delegate necessary operations to the collection.
         def_delegators :collection, :database, :client
@@ -87,22 +87,34 @@ def allow_disk_use(value = nil)
         #   See the server documentation for details.
         # @option options [ Integer ] :max_time_ms The maximum amount of time in
         #   milliseconds to allow the aggregation to run.
-        # @option options [ true, false ] :use_cursor Indicates whether the command
-        #   will request that the server provide results using a cursor. Note that
-        #   as of server version 3.6, aggregations always provide results using a
-        #   cursor and this option is therefore not valid.
         # @option options [ Session ] :session The session to use.
+        # @option options [ :cursor_lifetime | :iteration ] :timeout_mode How to interpret
+        #   :timeout_ms (whether it applies to the lifetime of the cursor, or per
+        #   iteration).
         # @option options [ Integer ] :timeout_ms The per-operation timeout in milliseconds.
         #   Must a positive integer. The default value is unset which means infinite.
         #
         # @since 2.0.0
         def initialize(view, pipeline, options = {})
           @view = view
           @pipeline = pipeline.dup
+
+          @timeout_ms = options.delete(:timeout_ms)
+          @options = BSON::Document.new(options).freeze
+
+          validate_timeout_mode!(options)
+
           unless Mongo.broken_view_aggregate || view.filter.empty?
             @pipeline.unshift(:$match => view.filter)
           end
-          @options = BSON::Document.new(options).freeze
+        end
+
+        # @return [ Integer | nil ] the timeout_ms value that was passed as
+        #   an option to this object, or which was inherited from the view.
+        #
+        # @api private
+        def timeout_ms
+          @timeout_ms || view.timeout_ms
         end
 
         # Get the explain plan for the aggregation.
@@ -182,15 +194,10 @@ def effective_read_preference(connection)
 
         end
 
-        def send_initial_query(server, session)
-          context = Operation::Context.new(
-            client: client,
-            session: session,
-            operation_timeouts: operation_timeouts(options)
-          )
+        def send_initial_query(server, context)
           server.with_connection do |connection|
             initial_query_op(
-              session,
+              context.session,
               effective_read_preference(connection)
             ).execute_with_connection(
               connection,
@@ -214,16 +221,16 @@ def cache_options
           }
         end
 
-      # @return [ Hash ] timeout_ms value set on the operation level (if any),
-      #   and/or timeout_ms that is set on collection/database/client level (if any).
-      #
-      # @api private
-        def operation_timeouts(opts)
+        # @return [ Hash ] timeout_ms value set on the operation level (if any),
+        #   and/or timeout_ms that is set on collection/database/client level (if any).
+        #
+        # @api private
+        def operation_timeouts(opts = {})
           {}.tap do |result|
-            if opts[:timeout_ms].nil?
-              result[:inherited_timeout_ms] = collection.timeout_ms
+            if opts[:timeout_ms] || @timeout_ms
+              result[:operation_timeout_ms] = opts.delete(:timeout_ms) || @timeout_ms
             else
-              result[:operation_timeout_ms] = opts.delete(:timeout_ms)
+              result[:inherited_timeout_ms] = view.timeout_ms
             end
           end
         end

lib/mongo/collection/view/builder/aggregation.rb

@@ -113,17 +113,11 @@ def aggregation_command
               command[:readConcern] = Options::Mapper.transform_values_to_strings(
                 read_concern)
             end
-            command[:cursor] = cursor if cursor
+            command[:cursor] = batch_size_doc
             command.merge!(Options::Mapper.transform_documents(options, MAPPINGS))
             command
           end
 
-          def cursor
-            if options[:use_cursor] == true || options[:use_cursor].nil?
-              batch_size_doc
-            end
-          end
-
           def batch_size_doc
             value = options[:batch_size] || view.batch_size
             if value == 0 && write?

lib/mongo/collection/view/change_stream.rb

@@ -60,6 +60,10 @@ class ChangeStream < Aggregation
         # @since 2.5.0
         attr_reader :options
 
+        # @return [ Cursor ] the underlying cursor for this operation
+        # @api private
+        attr_reader :cursor
+
         # Initialize the change stream for the provided collection view, pipeline
         # and options.
         #
@@ -231,13 +235,16 @@ def try_next
         #   This method ignores any errors that occur when closing the
         #   server-side cursor.
         #
+        # @params [ Hash ] opts Options to be passed to the cursor close
+        #   command.
+        #
         # @return [ nil ] Always nil.
         #
         # @since 2.5.0
-        def close
+        def close(opts = {})
           unless closed?
             begin
-              @cursor.close
+              @cursor.close(opts)
             rescue Error::OperationFailure::Family, Error::SocketError, Error::SocketTimeoutError, Error::MissingConnection
               # ignore
             end
@@ -303,14 +310,16 @@ def create_cursor!
           # (rolling upgrades)
           @start_at_operation_time_supported = nil
 
-          session = client.send(:get_session, @options)
+          session = client.get_session(@options)
+          context = Operation::Context.new(client: client, session: session, operation_timeouts: operation_timeouts)
+
           start_at_operation_time = nil
           start_at_operation_time_supported = nil
-          @cursor = read_with_retry_cursor(session, server_selector, view) do |server|
+          @cursor = read_with_retry_cursor(session, server_selector, view, context: context) do |server|
             server.with_connection do |connection|
               start_at_operation_time_supported = connection.description.server_version_gte?('4.0')
 
-              result = send_initial_query(connection, session)
+              result = send_initial_query(connection, context)
               if doc = result.replies.first && result.replies.first.documents.first
                 start_at_operation_time = doc['operationTime']
               else
@@ -390,11 +399,11 @@ def change_doc
           end
         end
 
-        def send_initial_query(connection, session)
-          initial_query_op(session, view.read_preference)
+        def send_initial_query(connection, context)
+          initial_query_op(context.session, view.read_preference)
             .execute_with_connection(
               connection,
-              context: Operation::Context.new(client: client, session: session),
+              context: context,
             )
         end