Add scaffolding for pool throttling

src/Microsoft.Data.SqlClient/netcore/src/Microsoft.Data.SqlClient.csproj

@@ -954,6 +954,11 @@
     <Compile Include="Microsoft\Data\SqlClientX\IO\TdsWriter.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\IO\TdsWriteStream.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\PoolingDataSource.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\AsyncFlagFunc.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\BlockingPeriodRateLimiter.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\ConcurrencyRateLimiter.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\RateLimiterBase.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\PassthroughRateLimiter.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\SqlConnectionX.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\SqlConnector.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\SqlDataSource.cs" />
@@ -982,7 +987,7 @@
   </ItemGroup>
   <ItemGroup>
     <PackageReference Include="Microsoft.Data.SqlClient.SNI.runtime" Version="$(MicrosoftDataSqlClientSNIRuntimeVersion)" />
-    <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="$(MicrosoftExtensionsCachingMemoryVersion)"  />
+    <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="$(MicrosoftExtensionsCachingMemoryVersion)" />
     <!-- Enable the project reference for debugging purposes. -->
     <!-- <ProjectReference Include="$(SqlServerSourceCode)\Microsoft.SqlServer.Server.csproj" /> -->
     <PackageReference Include="Microsoft.SqlServer.Server" Version="$(MicrosoftSqlServerServerVersion)" />

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/PoolingDataSource.cs

@@ -13,6 +13,7 @@
 using System.Threading.Tasks;
 using Microsoft.Data.ProviderBase;
 using Microsoft.Data.SqlClient;
+using Microsoft.Data.SqlClientX.RateLimiters;
 
 namespace Microsoft.Data.SqlClientX
 {
@@ -22,6 +23,7 @@ namespace Microsoft.Data.SqlClientX
     internal sealed class PoolingDataSource : SqlDataSource
     {
         private DbConnectionPoolGroupOptions _connectionPoolGroupOptions;
+        private RateLimiterBase _connectionRateLimiter;
 
         internal int MinPoolSize => _connectionPoolGroupOptions.MinPoolSize;
         internal int MaxPoolSize => _connectionPoolGroupOptions.MaxPoolSize;
@@ -30,13 +32,13 @@ internal sealed class PoolingDataSource : SqlDataSource
         /// Initializes a new PoolingDataSource.
         /// </summary>
         //TODO: support auth contexts and provider info
-        PoolingDataSource(SqlConnectionStringBuilder connectionStringBuilder,
+        internal PoolingDataSource(SqlConnectionStringBuilder connectionStringBuilder,
             SqlCredential credential,
-            RemoteCertificateValidationCallback userCertificateValidationCallback,
-            Action<X509CertificateCollection> clientCertificatesCallback,
-            DbConnectionPoolGroupOptions options) : base(connectionStringBuilder, credential, userCertificateValidationCallback, clientCertificatesCallback)
+            DbConnectionPoolGroupOptions options,
+            RateLimiterBase connectionRateLimiter) : base(connectionStringBuilder, credential)
         {
             _connectionPoolGroupOptions = options;
+            _connectionRateLimiter = connectionRateLimiter;
             //TODO: other construction
         }
 
@@ -46,10 +48,32 @@ internal override ValueTask<SqlConnector> GetInternalConnection(SqlConnectionX o
             throw new NotImplementedException();
         }
 
+        internal struct OpenInternalConnectionState
+        {
+            readonly SqlConnectionX _owningConnection;
+            readonly TimeSpan _timeout;
+
+            internal OpenInternalConnectionState(SqlConnectionX owningConnection, TimeSpan timeout)
+            {
+                _owningConnection = owningConnection;
+                _timeout = timeout;
+            }
+        }
+
         /// <inheritdoc/>
         internal override ValueTask<SqlConnector> OpenNewInternalConnection(SqlConnectionX owningConnection, TimeSpan timeout, bool async, CancellationToken cancellationToken)
         {
-            throw new NotImplementedException();
+            return _connectionRateLimiter.Execute<OpenInternalConnectionState, SqlConnector>(
+                RateLimitedOpen,
+                new OpenInternalConnectionState(owningConnection, timeout),
+                async,
+                cancellationToken
+            );
+
+            ValueTask<SqlConnector> RateLimitedOpen(OpenInternalConnectionState state, bool async, CancellationToken cancellationToken)
+            {
+                throw new NotImplementedException();
+            }
         }
 
         /// <inheritdoc/>

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/AsyncFlagFunc.cs

@@ -0,0 +1,20 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Threading;
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// A function that operates asynchronously based on a flag. If isAsync is true, the function operates asynchronously.
+    /// If isAsync is false, the function operates synchronously.
+    /// </summary>
+    /// <typeparam name="State">The type accepted by the callback as input.</typeparam>
+    /// <typeparam name="TResult">The returned by the callback.</typeparam>
+    /// <param name="state">An instance of State to be passed to the callback.</param>
+    /// <param name="isAsync">Indicates whether the function should operate asynchronously.</param>
+    /// <param name="cancellationToken">Allows cancellation of the operation.</param>
+    /// <returns>Returns the result of the callback.</returns>
+    internal delegate TResult AsyncFlagFunc<in State, out TResult>(State state, bool isAsync, CancellationToken cancellationToken);
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/BlockingPeriodRateLimiter.cs

@@ -0,0 +1,26 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+#nullable enable
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// A rate limiter that enforces a backoff (blocking) period upon error. 
+    /// Each subsequent error increases the blocking duration, up to a maximum, until a success occurs.
+    /// </summary>
+    internal sealed class BlockingPeriodRateLimiter : RateLimiterBase
+    {
+
+        /// <inheritdoc/>
+        internal override ValueTask<TResult> Execute<State, TResult>(AsyncFlagFunc<State, ValueTask<TResult>> callback, State state, bool async, CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
+    }
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/ConcurrencyRateLimiter.cs

@@ -0,0 +1,61 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// A rate limiter that enforces a concurrency limit. 
+    /// When the limit is reached, new requests must wait until a spot is freed upon completion of an existing request.
+    /// </summary>
+    internal class ConcurrencyRateLimiter : RateLimiterBase
+    {
+        private SemaphoreSlim _concurrencyLimitSemaphore;
+
+        /// <summary>
+        /// Initializes a new ConcurrencyRateLimiter with the specified concurrency limit.
+        /// </summary>
+        /// <param name="concurrencyLimit">The maximum number of concurrent requests.</param>
+        internal ConcurrencyRateLimiter(int concurrencyLimit)
+        {
+            _concurrencyLimitSemaphore = new SemaphoreSlim(concurrencyLimit);
+        }
+
+        /// <inheritdoc/>
+        internal sealed override async ValueTask<TResult> Execute<State, TResult>(AsyncFlagFunc<State, ValueTask<TResult>> callback, State state, bool async, CancellationToken cancellationToken = default)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            //TODO: in the future, we can enforce order
+            if (async)
+            {
+                await _concurrencyLimitSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+            }
+            else
+            {
+                _concurrencyLimitSemaphore.Wait(cancellationToken);
+            }
+
+            try
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                if (Next != null)
+                {
+                    return await Next.Execute<State, TResult>(callback, state, async, cancellationToken).ConfigureAwait(false);
+                }
+                else
+                {
+                    return await callback(state, async, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            finally
+            {
+                _concurrencyLimitSemaphore.Release();
+            }
+        }
+    }
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/PassthroughRateLimiter.cs

@@ -0,0 +1,31 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// A no-op rate limiter that simply executes the callback or passes through to the next rate limiter.
+    /// </summary>
+    internal sealed class PassthroughRateLimiter : RateLimiterBase
+    {
+        //TODO: no state, add static instance
+
+        /// <inheritdoc/>
+        internal override ValueTask<TResult> Execute<State, TResult>(AsyncFlagFunc<State, ValueTask<TResult>> callback, State state, bool async, CancellationToken cancellationToken = default)
+        {
+            if (Next != null)
+            {
+                return Next.Execute<State, TResult>(callback, state, async, cancellationToken);
+            }
+            else
+            {
+                return callback(state, async, cancellationToken);
+            }
+        }
+    }
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/RateLimiterBase.cs

@@ -0,0 +1,39 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// An interface for rate limiters that execute arbitraty code. Intended to be small and self contained and chained together to achieve more complex behavior.
+    /// </summary>
+    internal abstract class RateLimiterBase
+    {
+        private RateLimiterBase _next;
+
+        /// <summary>
+        /// The next rate limiter that should be executed within the context of this rate limiter.
+        /// </summary>
+        internal virtual RateLimiterBase Next
+        {
+            get => _next;
+            set => _next = value;
+        }
+
+        /// <summary>
+        /// Execute the provided callback within the context of the rate limit, or pass the responsibility along to the next rate limiter.
+        /// </summary>
+        /// <typeparam name="State">The type accepted by the callback as input.</typeparam>
+        /// <typeparam name="TResult">The type of the result returned by the callback.</typeparam>
+        /// <param name="callback">The callback function to execute.</param>
+        /// <param name="state">An instance of State to be passed to the callback.</param>
+        /// <param name="async">Whether this method should run asynchronously.</param>
+        /// <param name="cancellationToken">Cancels outstanding requests.</param>
+        /// <returns>Returns the result of the callback or the next rate limiter.</returns>
+        internal abstract ValueTask<TResult> Execute<State, TResult>(AsyncFlagFunc<State, ValueTask<TResult>> callback, State state, bool async, CancellationToken cancellationToken = default);
+    }
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/SqlDataSource.cs

@@ -22,10 +22,6 @@ internal abstract class SqlDataSource : DbDataSource
     {
         private readonly SqlConnectionStringBuilder _connectionStringBuilder;
 
-        private readonly RemoteCertificateValidationCallback _userCertificateValidationCallback;
-
-        private readonly Action<X509CertificateCollection> _clientCertificatesCallback;
-
         internal SqlCredential Credential { get; }
 
         //TODO: return SqlConnection after it is updated to wrap SqlConnectionX 
@@ -40,14 +36,10 @@ protected override SqlConnectionX CreateDbConnection()
 
         internal SqlDataSource(
             SqlConnectionStringBuilder connectionStringBuilder,
-            SqlCredential credential,
-            RemoteCertificateValidationCallback userCertificateValidationCallback,
-            Action<X509CertificateCollection> clientCertificatesCallback)
+            SqlCredential credential)
         {
             _connectionStringBuilder = connectionStringBuilder;
             Credential = credential;
-            _userCertificateValidationCallback = userCertificateValidationCallback;
-            _clientCertificatesCallback = clientCertificatesCallback;
         }
 
         /// <inheritdoc />

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/SqlDataSourceBuilder.cs

@@ -5,9 +5,13 @@
 #if NET8_0_OR_GREATER
 
 using System;
+using System.Diagnostics;
 using System.Net.Security;
 using System.Security.Cryptography.X509Certificates;
+using Microsoft.Data.Common;
+using Microsoft.Data.ProviderBase;
 using Microsoft.Data.SqlClient;
+using Microsoft.Data.SqlClientX.RateLimiters;
 
 namespace Microsoft.Data.SqlClientX
 {
@@ -42,11 +46,67 @@ public SqlDataSourceBuilder(string connectionString = null, SqlCredential creden
         /// </summary>
         public SqlDataSource Build()
         {
-            return new UnpooledDataSource(
-                ConnectionStringBuilder,
-                Credential,
-                UserCertificateValidationCallback,
-                ClientCertificatesCallback);
+            if (ConnectionStringBuilder.Pooling)
+            {
+                //TODO: pool group layer
+
+                DbConnectionPoolGroupOptions poolGroupOptions = new DbConnectionPoolGroupOptions(
+                    ConnectionStringBuilder.IntegratedSecurity,
+                    ConnectionStringBuilder.MinPoolSize,
+                    ConnectionStringBuilder.MaxPoolSize,
+                    //TODO: carry over connect timeout conversion logic from SqlConnectionFactory? if not, don't need an extra allocation for this object, just use connection string builder
+                    ConnectionStringBuilder.ConnectTimeout,
+                    ConnectionStringBuilder.LoadBalanceTimeout,
+                    ConnectionStringBuilder.Enlist);
+
+                //TODO: evaluate app context switch for concurrency limit
+                RateLimiterBase rateLimiter = IsBlockingPeriodEnabled() ? new BlockingPeriodRateLimiter() : new PassthroughRateLimiter();
+
+                return new PoolingDataSource(ConnectionStringBuilder,
+                    Credential,
+                    poolGroupOptions,
+                    rateLimiter);
+            }
+            else
+            {
+                return new UnpooledDataSource(
+                    ConnectionStringBuilder,
+                    Credential);
+            }
+        }
+
+        private bool IsBlockingPeriodEnabled()
+        {
+            var policy = ConnectionStringBuilder.PoolBlockingPeriod;
+
+            switch (policy)
+            {
+                case PoolBlockingPeriod.Auto:
+                    {
+                        if (ADP.IsAzureSqlServerEndpoint(ConnectionStringBuilder.DataSource))
+                        {
+                            return false; // in Azure it will be Disabled
+                        }
+                        else
+                        {
+                            return true; // in Non Azure, it will be Enabled
+                        }
+                    }
+                case PoolBlockingPeriod.AlwaysBlock:
+                    {
+                        return true; //Enabled
+                    }
+                case PoolBlockingPeriod.NeverBlock:
+                    {
+                        return false; //Disabled
+                    }
+                default:
+                    {
+                        //we should never get into this path.
+                        Debug.Fail("Unknown PoolBlockingPeriod. Please specify explicit results in above switch case statement.");
+                        return true;
+                    }
+            }
         }
     }
 }