v2.5.10: add CosmoKvPipesConnectionPool — fix single-socket bottleneck

vkuttyp · claude · vkuttyp · commit 5b5c2d09f8b4 · 2026-05-21T01:05:48.000+03:00
v2.5.9's bare CosmoKvPipesConnection serialises every Query/Execute
behind its SemaphoreSlim(1,1) because the underlying socket is one
bidirectional stream. Under marivil's binary-protocol stress run
2026-05-21, that surfaced as IMAP throughput collapsing 87% and 493
`sp_Users_GetByEmail` 30s-timeouts — every SQL call on the
mail-server process was queueing behind one socket while the HTTP
path got connection pooling for free via HttpClient.

CosmoKvPipesConnectionPool : ISqlDatabase wraps N pooled sockets,
shape mirrors MsSqlConnectionPool:
  - Channel&lt;T&gt; idle queue + Interlocked _opened counter for lazy growth
  - AcquireAsync: try idle, then grow up to MaxPoolSize, then wait
  - ReleaseAsync: return to idle, or discard if dropped
  - QueryStreamAsync holds the lease for the FULL enumeration

New `MaxPoolSize` field on CosmoKvPipesConfiguration (default 16),
parseable via `MaxPoolSize=N` in the connection string.

4 new xUnit tests (10/10 in the project now): basic round-trip,
32 concurrent SELECTs complete under 5s, MaxPoolSize=1 still
correct (just slow — sanity check), Dispose closes all sockets.

This is the API CosmoMailDb.Create should wire — the bare
CosmoKvPipesConnection is now treated as a building block (still
public for advanced callers who want to manage their own pool).

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/Directory.Build.props b/Directory.Build.props
@@ -7,6 +7,6 @@
     <Authors>vkuttyp</Authors>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <RepositoryUrl>https://github.com/vkuttyp/CosmoSQLClient-Dotnet</RepositoryUrl>
-    <Version>2.5.9</Version>
+    <Version>2.5.10</Version>
   </PropertyGroup>
 </Project>
diff --git a/src/CosmoSQLClient.CosmoKvPipes/CosmoKvPipesConfiguration.cs b/src/CosmoSQLClient.CosmoKvPipes/CosmoKvPipesConfiguration.cs
@@ -19,6 +19,14 @@ public sealed record CosmoKvPipesConfiguration
     /// <summary>Per-request timeout. Default 30 s.</summary>
     public TimeSpan Timeout { get; init; } = TimeSpan.FromSeconds(30);
 
+    /// <summary>
+    /// Max concurrent sockets in the connection pool. Default 16 — the
+    /// mail-server's typical SMTP/IMAP/EWS/EAS bursts fit comfortably,
+    /// matching cosmokvd's default 32 auto-commit slots without over-
+    /// committing. Set higher under heavy parallel SELECT load.
+    /// </summary>
+    public int MaxPoolSize { get; init; } = 16;
+
     /// <summary>Absolute socket-file path parsed out of <see cref="Endpoint"/>.</summary>
     public string SocketPath
     {
@@ -33,7 +41,7 @@ public string SocketPath
     }
 
     public string ConnectionString =>
-        $"Endpoint={Endpoint};AuthToken={AuthToken};Timeout={(int)Timeout.TotalSeconds}";
+        $"Endpoint={Endpoint};AuthToken={AuthToken};Timeout={(int)Timeout.TotalSeconds};MaxPoolSize={MaxPoolSize}";
 
     public static CosmoKvPipesConfiguration Parse(string connectionString)
     {
@@ -64,11 +72,16 @@ public static CosmoKvPipesConfiguration Parse(string connectionString)
         if (Get("Timeout") is string t && int.TryParse(t, out var sec) && sec > 0)
             timeout = TimeSpan.FromSeconds(sec);
 
+        int maxPool = 16;
+        if (Get("MaxPoolSize", "Pooling") is string p && int.TryParse(p, out var pp) && pp > 0)
+            maxPool = pp;
+
         return new CosmoKvPipesConfiguration
         {
-            Endpoint  = endpoint,
-            AuthToken = token,
-            Timeout   = timeout,
+            Endpoint    = endpoint,
+            AuthToken   = token,
+            Timeout     = timeout,
+            MaxPoolSize = maxPool,
         };
     }
 }
diff --git a/src/CosmoSQLClient.CosmoKvPipes/CosmoKvPipesConnectionPool.cs b/src/CosmoSQLClient.CosmoKvPipes/CosmoKvPipesConnectionPool.cs
@@ -0,0 +1,195 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using CosmoSQLClient.Core;
+
+namespace CosmoSQLClient.CosmoKvPipes;
+
+/// <summary>
+/// Connection pool for <see cref="CosmoKvPipesConnection"/>. Each pooled
+/// socket is independent — N concurrent SQL calls run over N sockets
+/// rather than queueing behind one in-process lock. This is the same
+/// shape <c>MsSqlConnectionPool</c> uses for its underlying connections.
+/// <para>
+/// Why: a single <c>CosmoKvPipesConnection</c> serialises every Query /
+/// Execute behind its <c>SemaphoreSlim(1, 1)</c> because the underlying
+/// socket is one bidirectional stream — interleaved frames from
+/// concurrent callers would corrupt the protocol. Without a pool, ALL
+/// SQL traffic on a mail-server process queues behind one connection,
+/// which marivil's binary-protocol stress run on 2026-05-21 surfaced
+/// as 493 <c>sp_Users_GetByEmail</c> 30 s-timeouts and IMAP throughput
+/// collapsing 87% vs the HTTP-over-UDS baseline.
+/// </para>
+/// <para>
+/// Implementation: <see cref="Channel{T}"/>-based idle queue + a count
+/// of "opened so far," same lazy-grow strategy <c>MsSqlConnectionPool</c>
+/// uses. Connections are created only when needed up to
+/// <see cref="MaxConnections"/>, then any caller arriving when the pool
+/// is saturated waits on the idle channel for a release.
+/// </para>
+/// </summary>
+public sealed class CosmoKvPipesConnectionPool : ISqlDatabase
+{
+    private readonly CosmoKvPipesConfiguration _config;
+    private readonly Channel<CosmoKvPipesConnection> _idle;
+    private int _opened;
+    private bool _disposed;
+
+    public int MaxConnections { get; }
+
+    public bool IsOpen => !_disposed;
+
+    public CosmoKvPipesConnectionPool(CosmoKvPipesConfiguration config, int maxConnections = 16)
+    {
+        _config         = config;
+        MaxConnections  = Math.Max(1, maxConnections);
+        _idle           = Channel.CreateUnbounded<CosmoKvPipesConnection>(
+                            new UnboundedChannelOptions { SingleReader = false, SingleWriter = false });
+    }
+
+    public CosmoKvPipesConnectionPool(string connectionString, int maxConnections = 16)
+        : this(CosmoKvPipesConfiguration.Parse(connectionString), maxConnections) { }
+
+    // ── ISqlDatabase delegation ─────────────────────────────────────────────
+
+    public Task<IReadOnlyList<SqlRow>> QueryAsync(
+        string sql, IReadOnlyList<SqlParameter>? parameters = null, CancellationToken ct = default)
+        => WithConnAsync(c => c.QueryAsync(sql, parameters, ct), ct);
+
+    public Task<IReadOnlyList<T>> QueryAsync<T>(
+        string sql, IReadOnlyList<SqlParameter>? parameters = null, CancellationToken ct = default)
+        where T : new()
+        => WithConnAsync(c => c.QueryAsync<T>(sql, parameters, ct), ct);
+
+    public Task<int> ExecuteAsync(
+        string sql, IReadOnlyList<SqlParameter>? parameters = null, CancellationToken ct = default)
+        => WithConnAsync(c => c.ExecuteAsync(sql, parameters, ct), ct);
+
+    public Task<SqlDataTable> QueryTableAsync(
+        string sql, IReadOnlyList<SqlParameter>? parameters = null, CancellationToken ct = default)
+        => WithConnAsync(c => c.QueryTableAsync(sql, parameters, ct), ct);
+
+    /// <summary>
+    /// Streams query results. The lease is held for the FULL enumeration
+    /// duration — closing the enumerator returns the connection. Don't
+    /// hold the enumerator past your immediate consumption loop or you'll
+    /// starve other callers of a slot.
+    /// </summary>
+    public async IAsyncEnumerable<SqlRow> QueryStreamAsync(
+        string sql, IReadOnlyList<SqlParameter>? parameters = null,
+        [EnumeratorCancellation] CancellationToken ct = default)
+    {
+        var conn = await AcquireAsync(ct).ConfigureAwait(false);
+        try
+        {
+            await foreach (var row in conn.QueryStreamAsync(sql, parameters, ct).ConfigureAwait(false))
+                yield return row;
+        }
+        finally
+        {
+            await ReleaseAsync(conn).ConfigureAwait(false);
+        }
+    }
+
+    public async IAsyncEnumerable<System.Text.Json.JsonElement> QueryJsonStreamAsync(
+        string sql, IReadOnlyList<SqlParameter>? parameters = null, int jsonColumnIndex = 0,
+        [EnumeratorCancellation] CancellationToken ct = default)
+    {
+        var conn = await AcquireAsync(ct).ConfigureAwait(false);
+        try
+        {
+            await foreach (var elem in conn.QueryJsonStreamAsync(sql, parameters, jsonColumnIndex, ct).ConfigureAwait(false))
+                yield return elem;
+        }
+        finally
+        {
+            await ReleaseAsync(conn).ConfigureAwait(false);
+        }
+    }
+
+    public Task BeginTransactionAsync(CancellationToken ct = default)
+        => throw new NotSupportedException(
+            "Explicit transactions are not supported on the CosmoKvPipes binary protocol. " +
+            "Use CosmoSQLClient.CosmoKvHttp for BEGIN/COMMIT/ROLLBACK semantics.");
+
+    public Task CommitAsync(CancellationToken ct = default)   => BeginTransactionAsync(ct);
+    public Task RollbackAsync(CancellationToken ct = default) => BeginTransactionAsync(ct);
+
+    public async Task CloseAsync()
+    {
+        if (_disposed) return;
+        _idle.Writer.TryComplete();
+        await foreach (var c in _idle.Reader.ReadAllAsync().ConfigureAwait(false))
+        {
+            try { await c.DisposeAsync().ConfigureAwait(false); } catch { /* best-effort */ }
+        }
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        if (_disposed) return;
+        _disposed = true;
+        await CloseAsync().ConfigureAwait(false);
+    }
+
+    // ── Pool internals ──────────────────────────────────────────────────────
+
+    private async Task<T> WithConnAsync<T>(Func<CosmoKvPipesConnection, Task<T>> op, CancellationToken ct)
+    {
+        var conn = await AcquireAsync(ct).ConfigureAwait(false);
+        try { return await op(conn).ConfigureAwait(false); }
+        finally { await ReleaseAsync(conn).ConfigureAwait(false); }
+    }
+
+    private async Task<CosmoKvPipesConnection> AcquireAsync(CancellationToken ct)
+    {
+        if (_disposed) throw new ObjectDisposedException(nameof(CosmoKvPipesConnectionPool));
+
+        // Fast path: pick up an idle connection if there is one.
+        if (_idle.Reader.TryRead(out var idle) && idle.IsOpen)
+            return idle;
+
+        // No idle. If we haven't grown to MaxConnections yet, lazily open
+        // a new one. Increment-and-check-then-decrement-on-fail is the
+        // racing-safe form (multiple acquirers may briefly observe the
+        // count past max; the decrement on failure keeps the bookkeeping
+        // consistent).
+        int n = Interlocked.Increment(ref _opened);
+        if (n <= MaxConnections)
+        {
+            try { return await CosmoKvPipesConnection.OpenAsync(_config, ct).ConfigureAwait(false); }
+            catch { Interlocked.Decrement(ref _opened); throw; }
+        }
+        Interlocked.Decrement(ref _opened);
+
+        // Pool is saturated — wait for a release on the idle channel.
+        // Loop because a released-but-dead connection should be discarded
+        // and the wait resumed.
+        while (true)
+        {
+            var pooled = await _idle.Reader.ReadAsync(ct).ConfigureAwait(false);
+            if (pooled.IsOpen) return pooled;
+            await DiscardAsync(pooled).ConfigureAwait(false);
+        }
+    }
+
+    private async ValueTask ReleaseAsync(CosmoKvPipesConnection conn)
+    {
+        if (!conn.IsOpen || _disposed)
+        {
+            await DiscardAsync(conn).ConfigureAwait(false);
+            return;
+        }
+        if (!_idle.Writer.TryWrite(conn))
+        {
+            // Channel was completed mid-release (Dispose racing with the
+            // last call). Discard the connection to avoid leaking it.
+            await DiscardAsync(conn).ConfigureAwait(false);
+        }
+    }
+
+    private async ValueTask DiscardAsync(CosmoKvPipesConnection conn)
+    {
+        Interlocked.Decrement(ref _opened);
+        try { await conn.DisposeAsync().ConfigureAwait(false); } catch { /* best-effort */ }
+    }
+}
diff --git a/tests/CosmoSQLClient.CosmoKvPipes.Tests/CosmoKvPipesConnectionPoolTests.cs b/tests/CosmoSQLClient.CosmoKvPipes.Tests/CosmoKvPipesConnectionPoolTests.cs
