v2.5.7: CosmoKvConnection allows concurrent reads, AsyncLocal active-access

Before: every QueryAsync and ExecuteAsync went through a single
SemaphoreSlim(1, 1) at the CosmoKvConnection layer, so even
embarrassingly-parallel SELECTs ran strictly one-at-a-time on a
shared connection. The cosmokvd UDS rollout finding ("UDS savings
invisible at marivil stress") traced back here: the per-call
transport cost was dwarfed by queueing behind this single lock.

Three changes:

1. New AsyncReaderWriterLock — counting semaphore with 256 slots;
   readers take 1, writers drain all. Mirrors the cosmokvd
   TxnRegistry pattern. Fairness caveat: under sustained read load
   a pending writer can wait indefinitely, but the mail-server
   workload (read-heavy, short bursts of writes) doesn't hit this
   in practice. Replace with a queue-fair lock if it ever bites.

2. Executor._kv → Executor._activeKv (AsyncLocal<IKvAccess?>). The
   single mutable field that SetActiveAccess was writing depended
   on the connection-level lock for race-freedom; AsyncLocal threads
   each statement's value through its own async flow so concurrent
   QueryAsync calls don't trample each other's routing. Shared
   _defaultExprCache moves to ConcurrentDictionary for the same
   concurrency reason.

3. CosmoKvConnection.QueryAsync now takes the READ lock; ExecuteAsync
   / BeginTransactionAsync / CommitAsync / RollbackAsync take the
   WRITE lock. Reads run in parallel up to the rwlock's reader
   capacity; writes still mutually exclude reads and other writers.

3 new xUnit tests in Phase29ConcurrentReadsTests verify:
  - 16 concurrent SELECTs measurably faster than serial (proves
    reads aren't serialised)
  - 32 concurrent SELECTs filtered by bucket each return their
    correct row count (proves AsyncLocal doesn't bleed across reads)
  - Reader observing COUNT(*) during concurrent inserts never sees
    a count-going-backwards (proves snapshot stability)

All 269 CosmoKv driver tests pass (266 prior + 3 new). Companion
deployment work (cosmokvd + mail-server) will follow once this
package indexes on NuGet.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: vkuttyp

Directory.Build.props

@@ -7,6 +7,6 @@
     <Authors>vkuttyp</Authors>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <RepositoryUrl>https://github.com/vkuttyp/CosmoSQLClient-Dotnet</RepositoryUrl>
-    <Version>2.5.6</Version>
+    <Version>2.5.7</Version>
   </PropertyGroup>
 </Project>

src/CosmoSQLClient.CosmoKv/AsyncReaderWriterLock.cs

@@ -0,0 +1,64 @@
+namespace CosmoSQLClient.CosmoKv;
+
+/// <summary>
+/// Minimal async reader-writer lock. Multiple readers can hold the
+/// lock concurrently; a writer acquires exclusively against all
+/// readers AND other writers. Implemented as a counting semaphore
+/// where readers take 1 slot and writers drain all slots — the same
+/// pattern <c>CosmoKvD.Services.TxnRegistry</c> uses for its
+/// auto-commit vs explicit-BEGIN coordination.
+///
+/// Fairness: under sustained read load a pending writer can wait
+/// indefinitely if readers keep arriving. The current mail-server
+/// workload is read-heavy with infrequent bursts of writes, and
+/// writes here are short (single SQL statement), so starvation
+/// hasn't been observed. If it becomes an issue, replace with a
+/// queue-fair RW lock at that point.
+/// </summary>
+internal sealed class AsyncReaderWriterLock
+{
+    private readonly SemaphoreSlim _slots;
+    private readonly int _capacity;
+
+    public AsyncReaderWriterLock(int readerCapacity = 256)
+    {
+        _capacity = readerCapacity;
+        _slots    = new SemaphoreSlim(readerCapacity, readerCapacity);
+    }
+
+    /// <summary>Acquire one reader slot. Returns immediately if capacity is available.</summary>
+    public Task AcquireReadAsync(CancellationToken ct) => _slots.WaitAsync(ct);
+
+    /// <summary>Release one reader slot previously acquired by <see cref="AcquireReadAsync"/>.</summary>
+    public void ReleaseRead() => _slots.Release();
+
+    /// <summary>
+    /// Acquire exclusive write access — drains every reader slot.
+    /// Blocks until all currently-running readers have finished.
+    /// </summary>
+    public async Task AcquireWriteAsync(CancellationToken ct)
+    {
+        int acquired = 0;
+        try
+        {
+            for (int i = 0; i < _capacity; i++)
+            {
+                await _slots.WaitAsync(ct).ConfigureAwait(false);
+                acquired++;
+            }
+        }
+        catch
+        {
+            // Release any slots we managed to grab before cancellation —
+            // we'd otherwise wedge the lock for everyone else.
+            for (int i = 0; i < acquired; i++) _slots.Release();
+            throw;
+        }
+    }
+
+    /// <summary>Release exclusive write access — restores all reader slots.</summary>
+    public void ReleaseWrite()
+    {
+        _slots.Release(_capacity);
+    }
+}

src/CosmoSQLClient.CosmoKv/CosmoKvConnection.cs

@@ -35,7 +35,16 @@ public sealed class CosmoKvConnection : ISqlDatabase
     /// <summary>The active explicit transaction, or <c>null</c> for auto-commit.</summary>
     private CosmoKvTxn? _txn;
 
-    private readonly SemaphoreSlim _lock = new(1, 1);
+    /// <summary>
+    /// Reader-writer lock — multiple QueryAsync calls run concurrently
+    /// (CosmoKv's MVCC already gives them snapshot isolation; the
+    /// Executor's per-call state is now AsyncLocal so concurrent reads
+    /// don't trample each other). ExecuteAsync and Begin/Commit/Rollback
+    /// take the write lock for exclusive access — they mutate <see cref="_txn"/>
+    /// and the underlying writer task. Replaces the prior single
+    /// SemaphoreSlim(1,1) that serialised everything.
+    /// </summary>
+    private readonly AsyncReaderWriterLock _rwLock = new();
     private bool _disposed;
 
     /// <summary>
@@ -173,17 +182,19 @@ public static Task<CosmoKvConnection> OpenAsync(
     public async Task<IReadOnlyList<SqlRow>> QueryAsync(
         string sql, IReadOnlyList<SqlParameter>? parameters = null, CancellationToken ct = default)
     {
-        await _lock.WaitAsync(ct);
+        await _rwLock.AcquireReadAsync(ct).ConfigureAwait(false);
         try
         {
             EnsureOpen();
             var stmt = TSqlParser.Parse(sql);
-            // Reads piggyback on the explicit txn for snapshot consistency,
-            // or hit the Db directly otherwise (single-statement reads
-            // don't need their own implicit txn — CosmoKv's iterator gives
-            // sufficient consistency for an isolated read).
-            if (_txn is not null)
-                _executor!.SetActiveAccess(new TxnKvAccess(_txn));
+            // Snapshot _txn under the read lock so a concurrent BEGIN
+            // (which holds the write lock and excludes us) can't change
+            // our view mid-execution. Once snapshotted, the per-call
+            // access lives in the Executor's AsyncLocal slot, so
+            // sibling readers don't see our routing.
+            var snapshotTxn = _txn;
+            if (snapshotTxn is not null)
+                _executor!.SetActiveAccess(new TxnKvAccess(snapshotTxn));
             else
                 _executor!.SetActiveAccess(null);
             var (cols, rows) = await _executor!.QueryAsync(stmt, parameters, ct);
@@ -192,7 +203,7 @@ public async Task<IReadOnlyList<SqlRow>> QueryAsync(
         finally
         {
             _executor?.SetActiveAccess(null);
-            _lock.Release();
+            _rwLock.ReleaseRead();
         }
     }
 
@@ -220,7 +231,7 @@ public async IAsyncEnumerable<SqlRow> QueryStreamAsync(
     public async Task<int> ExecuteAsync(
         string sql, IReadOnlyList<SqlParameter>? parameters = null, CancellationToken ct = default)
     {
-        await _lock.WaitAsync(ct);
+        await _rwLock.AcquireWriteAsync(ct).ConfigureAwait(false);
         try
         {
             EnsureOpen();
@@ -272,7 +283,7 @@ public async Task<int> ExecuteAsync(
         finally
         {
             _executor?.SetActiveAccess(null);
-            _lock.Release();
+            _rwLock.ReleaseWrite();
         }
     }
 
@@ -322,23 +333,23 @@ private static IReadOnlyList<SqlRow> MaterializeRows(
 
     public async Task BeginTransactionAsync(CancellationToken ct = default)
     {
-        await _lock.WaitAsync(ct);
+        await _rwLock.AcquireWriteAsync(ct).ConfigureAwait(false);
         try { await BeginTransactionAsyncLocked(ct); }
-        finally { _lock.Release(); }
+        finally { _rwLock.ReleaseWrite(); }
     }
 
     public async Task CommitAsync(CancellationToken ct = default)
     {
-        await _lock.WaitAsync(ct);
+        await _rwLock.AcquireWriteAsync(ct).ConfigureAwait(false);
         try { await CommitAsyncLocked(ct); }
-        finally { _lock.Release(); }
+        finally { _rwLock.ReleaseWrite(); }
     }
 
     public async Task RollbackAsync(CancellationToken ct = default)
     {
-        await _lock.WaitAsync(ct);
+        await _rwLock.AcquireWriteAsync(ct).ConfigureAwait(false);
         try { await RollbackAsyncLocked(ct); }
-        finally { _lock.Release(); }
+        finally { _rwLock.ReleaseWrite(); }
     }
 
     // ── Transaction internals (lock already held by caller) ─────────────────
@@ -407,7 +418,9 @@ public async ValueTask DisposeAsync()
         if (_disposed) return;
         _disposed = true;
         await CloseAsync();
-        _lock.Dispose();
+        // _rwLock holds a SemaphoreSlim; .NET will reclaim it on GC.
+        // We don't expose a hard-disposal hook because outstanding async
+        // operations might still be holding slots when DisposeAsync runs.
     }
 
     private void EnsureOpen()

src/CosmoSQLClient.CosmoKv/Execution/Executor.cs

@@ -20,28 +20,48 @@ internal sealed class Executor
     private readonly Catalog   _catalog;
 
     /// <summary>
-    /// Active storage handle for the current statement. The connection
-    /// swaps this to <see cref="TxnKvAccess"/> when a transaction (explicit
-    /// or implicit-per-statement) is active, otherwise it routes through
-    /// <see cref="DbKvAccess"/>. The connection's <c>SemaphoreSlim</c>
-    /// serializes calls so this single-field handoff is race-free.
+    /// Per-call active storage handle. Stored in <see cref="AsyncLocal{T}"/>
+    /// so concurrent statement executions on the same Executor (which is
+    /// shared across an entire CosmoKvConnection) don't trample each
+    /// other's access routing. The connection sets this to a
+    /// <see cref="TxnKvAccess"/> when running inside an explicit (or
+    /// implicit-per-statement) transaction, and null otherwise; the
+    /// getter <see cref="Kv"/> falls back to a shared
+    /// <see cref="DbKvAccess"/> for auto-commit reads.
+    ///
+    /// Pre-refactor this was a single mutable field, race-free only
+    /// because CosmoKvConnection's <c>SemaphoreSlim</c> serialized every
+    /// QueryAsync/ExecuteAsync call. Moving to AsyncLocal unblocks
+    /// concurrent reads at the connection layer (the actual bottleneck
+    /// per the UDS finding).
     /// </summary>
-    private IKvAccess _kv;
+    private readonly AsyncLocal<IKvAccess?> _activeKv = new();
+
+    /// <summary>
+    /// Shared fresh-snapshot auto-commit access. Returned by <see cref="Kv"/>
+    /// whenever no per-call <see cref="_activeKv"/> override is set.
+    /// </summary>
+    private readonly IKvAccess _defaultKv;
+
+    /// <summary>Effective storage handle for the in-flight statement.</summary>
+    private IKvAccess Kv => _activeKv.Value ?? _defaultKv;
 
     internal Executor(CosmoKvDb db, Catalog catalog)
     {
-        _db      = db;
-        _catalog = catalog;
-        _kv      = new DbKvAccess(db);
+        _db        = db;
+        _catalog   = catalog;
+        _defaultKv = new DbKvAccess(db);
     }
 
     /// <summary>
-    /// Swap the active <see cref="IKvAccess"/>. Pass <c>null</c> to revert
-    /// to a fresh auto-committing <see cref="DbKvAccess"/>. Called by
-    /// <c>CosmoKvConnection</c> before and after each statement.
+    /// Swap the active <see cref="IKvAccess"/> for the current async flow.
+    /// Pass <c>null</c> to fall back to the shared auto-commit
+    /// <see cref="DbKvAccess"/>. Each async flow has its own AsyncLocal
+    /// slot, so concurrent statements on the same Executor see independent
+    /// values.
     /// </summary>
     public void SetActiveAccess(IKvAccess? kv)
-        => _kv = kv ?? new DbKvAccess(_db);
+        => _activeKv.Value = kv;
 
     /// <summary>
     /// When <c>true</c>, INSERTs may carry explicit values for IDENTITY
@@ -53,8 +73,9 @@ public void SetActiveAccess(IKvAccess? kv)
 
     // Cache of parsed DEFAULT expressions keyed by source SQL — the same
     // string fires once per row inserted, parser is cheap but constant
-    // overhead adds up under bulk loads.
-    private readonly Dictionary<string, Ast.Expression> _defaultExprCache = new();
+    // overhead adds up under bulk loads. ConcurrentDictionary so the
+    // cache is safe under the AsyncLocal-based concurrent-read model.
+    private readonly System.Collections.Concurrent.ConcurrentDictionary<string, Ast.Expression> _defaultExprCache = new();
 
     /// <summary>
     /// Evaluate a column's <c>DefaultExpressionSql</c> (v2.4 non-literal
@@ -177,7 +198,7 @@ private async Task<int> ExecuteCreateIndex(CreateIndexStatement s, CancellationT
         {
             var enc = IndexKeyCodec.Encode(ProjectIndexedValues(row, colIdxs));
             var key = KvKeys.IndexEntry(table.Name, index.Name, enc, rowId);
-            await _kv.SetAsync(key, Array.Empty<byte>(), ct);
+            await Kv.SetAsync(key, Array.Empty<byte>(), ct);
         }
         return 0;
     }
@@ -343,7 +364,7 @@ SqlTypeFamily.TinyInt or SqlTypeFamily.SmallInt or SqlTypeFamily.Int
 
             var key   = KvKeys.Row(table.Name, rowId);
             var bytes = RowCodec.Encode(record);
-            await _kv.SetAsync(key, bytes, ct);
+            await Kv.SetAsync(key, bytes, ct);
             await WriteIndexEntriesAsync(table, record, rowId, ct);
             rowsInserted++;
 
@@ -429,7 +450,7 @@ private async Task WriteIndexEntriesAsync(
             }
             var enc = IndexKeyCodec.Encode(vals);
             var key = KvKeys.IndexEntry(table.Name, idx.Name, enc, rowId);
-            await _kv.SetAsync(key, Array.Empty<byte>(), ct);
+            await Kv.SetAsync(key, Array.Empty<byte>(), ct);
         }
     }
 
@@ -450,7 +471,7 @@ private async Task EnforceUniqueAsync(
         Buffer.BlockCopy(prefix,      0, fullPrefix, 0,             prefix.Length);
         Buffer.BlockCopy(encWithTerm, 0, fullPrefix, prefix.Length, encWithTerm.Length);
 
-        await foreach (var item in _kv.IterateAsync(
+        await foreach (var item in Kv.IterateAsync(
             new global::CosmoKv.IteratorOptions { Prefix = fullPrefix }, ct))
         {
             long otherRowId = KvKeys.ExtractRowId(item.Key);
@@ -471,7 +492,7 @@ private async Task DeleteIndexEntriesAsync(
             var colIdxs = ResolveIndexColumns(table, idx);
             var enc = IndexKeyCodec.Encode(ProjectIndexedValues(row, colIdxs));
             var key = KvKeys.IndexEntry(table.Name, idx.Name, enc, rowId);
-            await _kv.DeleteAsync(key, ct);
+            await Kv.DeleteAsync(key, ct);
         }
     }
 
@@ -498,15 +519,15 @@ private static int[] ResolveColumnMapping(InsertStatement s, TableSchema t)
     private async Task<long> NextIdentityAsync(string tableName, long seed, long increment)
     {
         var key = KvKeys.SeqCounter(tableName);
-        var existing = await _kv.GetAsync(key);
+        var existing = await Kv.GetAsync(key);
         long next;
         if (existing is null || existing.Length == 0)
             next = seed;
         else
             next = BinaryPrimitives.ReadInt64BigEndian(existing) + increment;
         Span<byte> buf = stackalloc byte[8];
         BinaryPrimitives.WriteInt64BigEndian(buf, next);
-        await _kv.SetAsync(key, buf.ToArray());
+        await Kv.SetAsync(key, buf.ToArray());
         return next;
     }
 
@@ -519,14 +540,14 @@ private async Task<long> NextIdentityAsync(string tableName, long seed, long inc
     private async Task BumpIdentityCounterAsync(string tableName, long supplied)
     {
         var key = KvKeys.SeqCounter(tableName);
-        var existing = await _kv.GetAsync(key);
+        var existing = await Kv.GetAsync(key);
         long current = (existing is null || existing.Length == 0)
             ? 0
             : BinaryPrimitives.ReadInt64BigEndian(existing);
         if (supplied <= current) return;
         Span<byte> buf = stackalloc byte[8];
         BinaryPrimitives.WriteInt64BigEndian(buf, supplied);
-        await _kv.SetAsync(key, buf.ToArray());
+        await Kv.SetAsync(key, buf.ToArray());
     }
 
     // ── SELECT (with Phase 2 WHERE/ORDER BY/TOP/OFFSET) ─────────────────────
@@ -1213,7 +1234,7 @@ private async IAsyncEnumerable<SqlValue[]> ScanRowsAsync(
     {
         var prefix = Encoding.UTF8.GetBytes(
             $"{KvKeys.DataPrefix}{table.Name.ToLowerInvariant()}:");
-        await foreach (var item in _kv.IterateAsync(
+        await foreach (var item in Kv.IterateAsync(
             new global::CosmoKv.IteratorOptions { Prefix = prefix }, ct))
         {
             byte[] bytes = await item.ReadValueAsync(ct);
@@ -1243,7 +1264,7 @@ private async IAsyncEnumerable<SqlValue[]> ScanRowsAsync(
         // Prefix, so we walk the prefix and skip entries outside [lower,upper).
         // Phase 4-bench may push this into the iterator if the entry count
         // becomes the dominant cost.
-        await foreach (var item in _kv.IterateAsync(iterOpts, ct))
+        await foreach (var item in Kv.IterateAsync(iterOpts, ct))
         {
             var keyAfterPrefix = ((ReadOnlySpan<byte>)item.Key)[prefix.Length..];
             // The keyAfterPrefix is `<encoded-value><rowid-be>`.
@@ -1258,7 +1279,7 @@ private async IAsyncEnumerable<SqlValue[]> ScanRowsAsync(
                 continue;
             }
             long rowId = KvKeys.ExtractRowId(item.Key);
-            var rowBytes = await _kv.GetAsync(KvKeys.Row(plan.Table.Name, rowId), ct);
+            var rowBytes = await Kv.GetAsync(KvKeys.Row(plan.Table.Name, rowId), ct);
             if (rowBytes is null || rowBytes.Length == 0) continue;
             yield return (RowCodec.Decode(rowBytes), rowId);
         }
@@ -1349,7 +1370,7 @@ private async Task<UpdateResult> ExecuteUpdateCore(
         foreach (var (rowId, oldRow, newRow) in pending)
         {
             await DeleteIndexEntriesAsync(table, oldRow, rowId, ct);
-            await _kv.SetAsync(KvKeys.Row(table.Name, rowId), RowCodec.Encode(newRow), ct);
+            await Kv.SetAsync(KvKeys.Row(table.Name, rowId), RowCodec.Encode(newRow), ct);
             await WriteIndexEntriesAsync(table, newRow, rowId, ct);
 
             if (outputPlan is not null)
@@ -1427,7 +1448,7 @@ private async Task<DeleteResult> ExecuteDeleteCore(
         foreach (var (rowId, row) in victims)
         {
             await DeleteIndexEntriesAsync(table, row, rowId, ct);
-            await _kv.DeleteAsync(KvKeys.Row(table.Name, rowId), ct);
+            await Kv.DeleteAsync(KvKeys.Row(table.Name, rowId), ct);
 
             if (outputPlan is not null)
             {