CSHARP-2861: Having incorrect credentials along with minPoolSize > 0 makes the driver to endlessly create new connections.

Author: DmitryLukyanov

src/MongoDB.Driver.Core/Core/ConnectionPools/ExclusiveConnectionPool.cs

@@ -258,13 +258,12 @@ private async Task MaintainSizeAsync()
                 {
                     await PrunePoolAsync(maintenanceCancellationToken).ConfigureAwait(false);
                     await EnsureMinSizeAsync(maintenanceCancellationToken).ConfigureAwait(false);
-                    await Task.Delay(_settings.MaintenanceInterval, maintenanceCancellationToken).ConfigureAwait(false);
                 }
                 catch
                 {
-                    // do nothing, this is called in the background and, quite frankly, should never
-                    // result in an error
+                    // ignore exceptions
                 }
+                await Task.Delay(_settings.MaintenanceInterval, maintenanceCancellationToken).ConfigureAwait(false);
             }
         }
 
@@ -322,7 +321,15 @@ private async Task EnsureMinSizeAsync(CancellationToken cancellationToken)
                     // when adding in a connection, we need to open it because 
                     // the whole point of having a min pool size is to have
                     // them available and ready...
-                    await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
+                    try
+                    {
+                        await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
+                    }
+                    catch (Exception)
+                    {
+                        connection.Dispose();
+                        throw;
+                    }
                     _connectionHolder.Return(connection);
                     stopwatch.Stop();
 

tests/MongoDB.Driver.Core.Tests/Core/ConnectionPools/ExclusiveConnectionPoolTests.cs

@@ -17,7 +17,9 @@
 using System.Collections.Generic;
 using System.Net;
 using System.Threading;
+using System.Threading.Tasks;
 using FluentAssertions;
+using MongoDB.Bson.TestHelpers;
 using MongoDB.Bson.TestHelpers.XunitExtensions;
 using MongoDB.Driver.Core.Clusters;
 using MongoDB.Driver.Core.Configuration;
@@ -62,12 +64,7 @@ public ExclusiveConnectionPoolTests()
                 waitQueueSize: 1,
                 waitQueueTimeout: TimeSpan.FromSeconds(2));
 
-            _subject = new ExclusiveConnectionPool(
-                _serverId,
-                _endPoint,
-                _settings,
-                _mockConnectionFactory.Object,
-                _capturedEvents);
+            _subject = CreateSubject();
         }
 
         [Fact]
@@ -463,6 +460,66 @@ public void Initialize_should_scale_up_the_number_of_connections_to_min_size()
             }
         }
 
+        [Fact]
+        public void MaintainSizeAsync_should_call_connection_dispose_when_connection_authentication_fail()
+        {
+            var authenticationFailedConnection = new Mock<IConnection>();
+            authenticationFailedConnection
+                .Setup(c => c.OpenAsync(It.IsAny<CancellationToken>())) // an authentication exception is thrown from _connectionInitializer.InitializeConnection
+                                                                        // that in turn is called from OpenAsync
+                .Throws(new MongoAuthenticationException(new ConnectionId(_serverId), "test message"));
+
+            using (var subject = CreateSubject())
+            {
+                _mockConnectionFactory
+                    .Setup(f => f.CreateConnection(_serverId, _endPoint))
+                    .Returns(() =>
+                    {
+                        subject._maintenanceCancellationTokenSource().Cancel(); // Task.Delay will be canceled 
+                        return authenticationFailedConnection.Object;
+                    });
+
+                var _ = Record.Exception(() => subject.MaintainSizeAsync().GetAwaiter().GetResult());
+                authenticationFailedConnection.Verify(conn => conn.Dispose(), Times.Once);
+            }
+        }
+
+        [Fact]
+        public void MaintainSizeAsync_should_not_try_new_attempt_after_failing_without_delay()
+        {
+            var settings =_settings.With(maintenanceInterval: TimeSpan.FromSeconds(10));
+
+            using (var subject = CreateSubject(settings))
+            {
+                _mockConnectionFactory
+                    .SetupSequence(f => f.CreateConnection(_serverId, _endPoint))
+                    .Throws<Exception>()    // failed attempt
+                    .Returns(() =>          // successful attempt which should be delayed
+                    {
+                        // break the loop. With this line the MaintainSizeAsync will contain only 2 iterations
+                        subject._maintenanceCancellationTokenSource().Cancel();
+                        return new MockConnection(_serverId);
+                    });
+
+                var testResult = Task.WaitAny(
+                    subject.MaintainSizeAsync(),            // if this task is completed first, it will mean that there was no delay (10 sec) 
+                    Task.Delay(TimeSpan.FromSeconds(1)));   // time to be sure that delay is happening,
+                                                            // if the method is running more than 1 second, then delay is happening
+                testResult.Should().Be(1);
+            }
+        }
+
+        // private methods
+        private ExclusiveConnectionPool CreateSubject(ConnectionPoolSettings connectionPoolSettings = null)
+        {
+            return new ExclusiveConnectionPool(
+                _serverId,
+                _endPoint,
+                connectionPoolSettings ?? _settings,
+                _mockConnectionFactory.Object,
+                _capturedEvents);
+        }
+
         private void InitializeAndWait()
         {
             _subject.Initialize();
@@ -481,4 +538,17 @@ private void InitializeAndWait()
             _subject.UsedCount.Should().Be(0);
         }
     }
+
+    internal static class ExclusiveConnectionPoolReflector
+    {
+        public static CancellationTokenSource _maintenanceCancellationTokenSource(this ExclusiveConnectionPool obj)
+        {
+            return (CancellationTokenSource)Reflector.GetFieldValue(obj, nameof(_maintenanceCancellationTokenSource));
+        }
+
+        public static Task MaintainSizeAsync(this ExclusiveConnectionPool obj)
+        {
+            return (Task)Reflector.Invoke(obj, nameof(MaintainSizeAsync));
+        }
+    }
 }

tests/MongoDB.Driver.Core.Tests/Core/Connections/TcpStreamFactoryTests.cs

@@ -18,15 +18,11 @@
 using System.Net.Sockets;
 using System.Threading;
 using FluentAssertions;
-using MongoDB.Driver.Core.Clusters;
 using MongoDB.Driver.Core.Configuration;
-using MongoDB.Driver.Core.Connections;
-using MongoDB.Driver.Core.Servers;
 using Xunit;
 using System.Threading.Tasks;
 using System.Reflection;
 using System.IO;
-using MongoDB.Driver.Core.TestHelpers;
 using MongoDB.Bson.TestHelpers.XunitExtensions;
 using MongoDB.Driver.Core.TestHelpers.XunitExtensions;
 using MongoDB.Bson.TestHelpers;
@@ -35,6 +31,37 @@ namespace MongoDB.Driver.Core.Connections
 {
     public class TcpStreamFactoryTests
     {
+        [Theory]
+        [ParameterAttributeData]
+        public void Connect_should_dispose_socket_if_socket_fails([Values(false, true)] bool async)
+        {
+            RequireServer.Check();
+
+            var subject = new TcpStreamFactory();
+            var endpoint = new DnsEndPoint("test", 80); // not existed endpoint which will fail when we call socket.Connect
+
+            using (var testSocket = new TestSocket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
+            {
+                Exception exception;
+                if (async)
+                {
+                    exception = Record.Exception(
+                        () =>
+                            subject
+                                .ConnectAsync(testSocket, endpoint, CancellationToken.None)
+                                .GetAwaiter()
+                                .GetResult());
+                }
+                else
+                {
+                    exception = Record.Exception(() => subject.Connect(testSocket, endpoint, CancellationToken.None));
+                }
+
+                exception.Should().NotBeNull();
+                testSocket.DisposeAttempts.Should().Be(1);
+            }
+        }
+
         [Fact]
         public void Constructor_should_throw_an_ArgumentNullException_when_tcpStreamSettings_is_null()
         {
@@ -125,7 +152,7 @@ public void CreateStream_should_call_the_socketConfigurator(
 
             if (async)
             {
-                 subject.CreateStreamAsync(endPoint, CancellationToken.None).GetAwaiter().GetResult();
+                subject.CreateStreamAsync(endPoint, CancellationToken.None).GetAwaiter().GetResult();
             }
             else
             {
@@ -185,10 +212,44 @@ public void SocketConfigurator_can_be_used_to_set_keepAlive(
             var keepAlive = (int)socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive);
             keepAlive.Should().NotBe(0); // .NET returns 1 but Mono returns 8
         }
+
+        // nested types
+        private class TestSocket : Socket
+        {
+            public int DisposeAttempts { get; set; } = 0;
+
+            public TestSocket(AddressFamily addressFamily, SocketType socketType, ProtocolType protocolType) : base(addressFamily, socketType, protocolType)
+            {
+            }
+
+            public TestSocket(SocketType socketType, ProtocolType protocolType) : base(socketType, protocolType)
+            {
+            }
+
+            public TestSocket(SocketInformation socketInformation) : base(socketInformation)
+            {
+            }
+
+            protected override void Dispose(bool disposing)
+            {
+                base.Dispose(disposing);
+                DisposeAttempts++;
+            }
+        }
     }
 
-    public static class TcpStreamFactoryReflector
+    internal static class TcpStreamFactoryReflector
     {
         internal static TcpStreamSettings _settings(this TcpStreamFactory obj) => (TcpStreamSettings)Reflector.GetFieldValue(obj, nameof(_settings));
+
+        internal static void Connect(this TcpStreamFactory obj, Socket socket, EndPoint endPoint, CancellationToken cancellationToken)
+        {
+            Reflector.Invoke(obj, nameof(Connect), socket, endPoint, cancellationToken);
+        }
+
+        internal static Task ConnectAsync(this TcpStreamFactory obj, Socket socket, EndPoint endPoint, CancellationToken cancellationToken)
+        {
+            return (Task)Reflector.Invoke(obj, nameof(ConnectAsync), socket, endPoint, cancellationToken);
+        }
     }
 }