RUBY-3046 send no_cursor_timeout to server again (#2557)

* Mongo::Collection::View#no_cursor_timeout regression

* adjust description

* add a unit test

* pass no cursor timeout option to server

* fix test with auth

* restrict to standalones

Co-authored-by: Oleg Pudeyev <code@olegp.name>
Co-authored-by: Dmitry Rybakov <dmitry.rybakov@mongodb.com>

Author: 3 people

lib/mongo/collection/view/iterable.rb

@@ -170,6 +170,7 @@ def initial_query_op(session)
             max_time_ms: options[:max_time_ms],
             max_value: options[:max_value],
             min_value: options[:min_value],
+            no_cursor_timeout: options[:no_cursor_timeout],
             return_key: options[:return_key],
             show_disk_loc: options[:show_disk_loc],
             comment: options[:comment],

spec/mongo/collection/view/readable_spec.rb

@@ -1790,6 +1790,62 @@
     it 'returns a new View' do
       expect(new_view).not_to be(view)
     end
+
+    context 'when sending to server' do
+      let(:subscriber) { Mrss::EventSubscriber.new }
+
+      before do
+        authorized_collection.client.subscribe(Mongo::Monitoring::COMMAND, subscriber)
+      end
+
+      let(:event) do
+        subscriber.single_command_started_event('find')
+      end
+
+      it 'is sent to server' do
+        new_view.to_a
+        event.command.slice('noCursorTimeout').should == {'noCursorTimeout' => true}
+      end
+    end
+
+    context 'integration test' do
+      require_topology :single
+
+      # The number of open cursors with the option set to prevent timeout.
+      def current_no_timeout_count
+        root_authorized_client
+          .command(serverStatus: 1)
+          .documents
+          .first
+          .fetch('metrics')
+          .fetch('cursor')
+          .fetch('open')
+          .fetch('noTimeout')
+      end
+
+      it 'is applied on the server' do
+        # Initialize collection with two documents.
+        new_view.collection.insert_many([{}, {}])
+
+        expect(new_view.count).to be == 2
+
+        # Initial "noTimeout" count should be zero.
+        states = [current_no_timeout_count]
+
+        # The "noTimeout" count should be one while iterating.
+        new_view.batch_size(1).each { states << current_no_timeout_count }
+
+        # Final "noTimeout" count should be back to zero.
+        states << current_no_timeout_count
+
+        # This succeeds on:
+        #   commit aab776ebdfb15ddb9765039f7300e15796de0c5c
+        #
+        # This starts failing with [0, 0, 0, 0] from:
+        #   commit 2d9f0217ec904a1952a1ada2136502eefbca562e
+        expect(states).to be == [0, 1, 1, 0]
+      end
+    end
   end
 
   describe '#projection' do