RUBY-3035 Reuse client in auto encrypter if unlimited pool (#2552)

* RUBY-3035 Reuse client in auto encrypter if unlimited pool

* 3035

* 3035

* 3035

* 3035

* 3035

* close newly made clients in auto encrypter if its initialization fails

* add note

* add normal cleanup for newly created clients

* fix pool sizes, add tests

* use mrss helper for event retrieval with earlier assertions on existence of the needed events

* fix command monitoring tests

* collapse back into #with

* fix stress test

* update mrss

* remove max pool size setting

* redo the tests

Co-authored-by: Oleg Pudeyev <code@olegp.name>

Author: comandeo-mongo

lib/mongo/client.rb

@@ -767,7 +767,7 @@ def with(new_options = nil)
         # We can't use the same cluster if some options that would affect it
         # have changed.
         if cluster_modifying?(opts)
-          Cluster.create(client)
+          Cluster.create(client, monitoring: opts[:monitoring])
         end
       end
     end

lib/mongo/cluster.rb

@@ -280,20 +280,22 @@ def initialize(seeds, monitoring, options = Options::Redacted.new)
     # Create a cluster for the provided client, for use when we don't want the
     # client's original cluster instance to be the same.
     #
-    # @api private
-    #
     # @example Create a cluster for the client.
     #   Cluster.create(client)
     #
     # @param [ Client ] client The client to create on.
+    # @param [ Monitoring | nil ] monitoring. The monitoring instance to use
+    #   with the new cluster. If nil, a new instance of Monitoring will be
+    #   created.
     #
     # @return [ Cluster ] The cluster.
     #
     # @since 2.0.0
-    def self.create(client)
+    # @api private
+    def self.create(client, monitoring: nil)
       cluster = Cluster.new(
         client.cluster.addresses.map(&:to_s),
-        Monitoring.new,
+        monitoring || Monitoring.new,
         client.cluster_options,
       )
       client.instance_variable_set(:@cluster, cluster)

lib/mongo/crypt/auto_encrypter.rb

@@ -85,6 +85,8 @@ class AutoEncrypter
       # @raise [ ArgumentError ] If required options are missing or incorrectly
       #   formatted.
       def initialize(options)
+        # Note that this call may eventually, via other method invocations,
+        # create additional clients which have to be cleaned up.
         @options = set_default_options(options).freeze
 
         @crypt_handle = Crypt::Handle.new(
@@ -123,6 +125,30 @@ def initialize(options)
           end
           raise
         end
+      rescue
+        if @key_vault_client && @key_vault_client != options[:client] &&
+          @key_vault_client.cluster != options[:client].cluster
+        then
+          begin
+            @key_vault_client.close
+          rescue => e
+            log_warn("Error closing key vault client in auto encrypter's constructor: #{e.class}: #{e}")
+            # Drop this exception so that the original exception is raised
+          end
+        end
+
+        if @metadata_client && @metadata_client != options[:client] &&
+          @metadata_client.cluster != options[:client].cluster
+        then
+          begin
+            @metadata_client.close
+          rescue => e
+            log_warn("Error closing metadata client in auto encrypter's constructor: #{e.class}: #{e}")
+            # Drop this exception so that the original exception is raised
+          end
+        end
+
+        raise
       end
 
       # Whether this encrypter should perform encryption (returns false if
@@ -168,6 +194,18 @@ def decrypt(command)
       def close
         @mongocryptd_client.close if @mongocryptd_client
 
+        if @key_vault_client && @key_vault_client != options[:client] &&
+          @key_vault_client.cluster != options[:client].cluster
+        then
+          @key_vault_client.close
+        end
+
+        if @metadata_client && @metadata_client != options[:client] &&
+          @metadata_client.cluster != options[:client].cluster
+        then
+          @metadata_client.close
+        end
+
         true
       end
 
@@ -210,26 +248,16 @@ def set_or_create_clients(options)
         client = options[:client]
         @key_vault_client = if options[:key_vault_client]
           options[:key_vault_client]
-        # https://jira.mongodb.org/browse/RUBY-3010
-        # https://jira.mongodb.org/browse/RUBY-3011
-        # Specification requires to use existing client when connection pool
-        # size is unlimited (0). Ruby driver does not support unlimited pool
-        # size.
-        # elsif client.options[:max_pool_size] == 0
-        #   client
+        elsif client.options[:max_pool_size] == 0
+          client
         else
           internal_client(client)
         end
 
         @metadata_client = if options[:bypass_auto_encryption]
           nil
-        # Specification requires to use existing client when connection pool
-        # size is unlimited (0). Ruby driver does not support unlimited pool
-        # size.
-        # https://jira.mongodb.org/browse/RUBY-3010
-        # https://jira.mongodb.org/browse/RUBY-3011
-        # elsif client.options[:max_pool_size] == 0
-        #   client
+        elsif client.options[:max_pool_size] == 0
+          client
         else
           internal_client(client)
         end
@@ -246,12 +274,8 @@ def set_or_create_clients(options)
       def internal_client(client)
         @internal_client ||= client.with(
           auto_encryption_options: nil,
-          # Specification requires that the internal client's connection pool
-          # size is unlimited (0). Ruby driver does not support unlimited pool
-          # size.
-          # https://jira.mongodb.org/browse/RUBY-3010
-          # https://jira.mongodb.org/browse/RUBY-3011
-          # max_pool_size: 0
+          min_pool_size: 0,
+          monitoring: client.send(:monitoring),
         )
       end
     end

spec/integration/client_construction_spec.rb

@@ -191,16 +191,82 @@
     context 'with default key vault client' do
       let(:key_vault_client) { nil }
 
-      it 'creates a working key vault client' do
-        key_vault_client = client.encrypter.key_vault_client
+      shared_examples 'creates a working key vault client' do
+        it 'creates a working key vault client' do
+          key_vault_client = client.encrypter.key_vault_client
 
-        result = key_vault_client[:test].insert_one(test: 1)
-        expect(result).to be_ok
+          result = key_vault_client[:test].insert_one(test: 1)
+          expect(result).to be_ok
+        end
       end
 
-      it 'creates a key vault client with the same cluster as the existing client' do
-        key_vault_client = client.encrypter.key_vault_client
-        expect(key_vault_client.cluster).to eq(client.cluster)
+      context 'when top-level max pool size is not 0' do
+        include_examples 'creates a working key vault client'
+
+        shared_examples 'limited connection pool' do
+          it 'creates a key vault client with a different cluster than the existing client' do
+            key_vault_client = client.encrypter.key_vault_client
+            expect(key_vault_client.cluster).not_to eq(client.cluster)
+          end
+
+          # min pool size for the key vault client can be greater than 0
+          # when the key vault client is the same as the top-level client.
+          # This is OK because we aren't making any more connections for FLE,
+          # the minimum was requested by application for its own needs.
+          it 'uses min pool size 0 for key vault client' do
+            key_vault_client = client.encrypter.key_vault_client
+            key_vault_client.options[:min_pool_size].should be 0
+          end
+        end
+
+        context 'when top-level max pool size is not specified' do
+          before do
+            client.options[:max_pool_size].should be nil
+          end
+
+          include_examples 'limited connection pool'
+
+          it 'uses unspecified max pool size for key vault client' do
+            key_vault_client = client.encrypter.key_vault_client
+            key_vault_client.options[:max_pool_size].should be nil
+          end
+        end
+
+        context 'when top-level max pool size is specified' do
+          let(:options) do
+            {
+              auto_encryption_options: auto_encryption_options,
+              max_pool_size: 42,
+            }
+          end
+
+          include_examples 'limited connection pool'
+
+          it 'uses the same max pool size for key vault client' do
+            key_vault_client = client.encrypter.key_vault_client
+            key_vault_client.options[:max_pool_size].should be 42
+          end
+        end
+      end
+
+      context 'when top-level max pool size is 0' do
+        let(:options) do
+          {
+            auto_encryption_options: auto_encryption_options,
+            max_pool_size: 0,
+          }
+        end
+
+        before do
+          client.options[:max_pool_size].should be 0
+        end
+
+        include_examples 'creates a working key vault client'
+
+        it 'creates a key vault client with the same cluster as the existing client' do
+          key_vault_client = client.encrypter.key_vault_client
+          expect(key_vault_client.cluster).to eq(client.cluster)
+        end
       end
     end
   end