fix

Author: KirillKurdyukov

src/Ydb.Sdk/src/Ado/Session/PoolingSessionSource.cs

@@ -75,7 +75,7 @@ private bool TryGetIdleSession([NotNullWhen(true)] out T? session)
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
     private bool CheckIdleSession([NotNullWhen(true)] T? session)
     {
-        if (session == null || session.State == PoolingSessionState.Clean)
+        if (session == null)
         {
             return false;
         }
@@ -312,12 +312,10 @@ protected PoolingSessionBase(PoolingSessionSource<T> source)
     }
 
     internal bool CompareAndSet(PoolingSessionState expected, PoolingSessionState actual) =>
-        Interlocked.CompareExchange(ref _state, (int)expected, (int)actual) == (int)expected;
+        Interlocked.CompareExchange(ref _state, (int)actual, (int)expected) == (int)expected;
 
     internal void Set(PoolingSessionState state) => Interlocked.Exchange(ref _state, (int)state);
 
-    internal PoolingSessionState State => (PoolingSessionState)Volatile.Read(ref _state);
-
     internal DateTime IdleStartTime { get; set; }
 
     public abstract IDriver Driver { get; }

src/Ydb.Sdk/test/Ydb.Sdk.Ado.Tests/Session/PoolingSessionSourceMockTests.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using Xunit;
 using Ydb.Query;
 using Ydb.Sdk.Ado.Session;
@@ -256,9 +257,22 @@ public async Task StressTest_HighContention_OpenClose()
         const int maxSessionSize = 50;
         const int minSessionSize = 10;
         const int highContentionTasks = maxSessionSize * 5;
+        var sessionIdIsBroken = new ConcurrentDictionary<int, bool>();
 
         var mockFactory = new MockPoolingSessionFactory(maxSessionSize)
-            { IsBroken = () => Random.Shared.NextDouble() < 0.05 };
+        {
+            IsBroken = sessionNum =>
+            {
+                var isBroken = Random.Shared.NextDouble() < 0.05;
+                sessionIdIsBroken[sessionNum] = isBroken;
+                return isBroken;
+            },
+            Open = sessionNum =>
+            {
+                sessionIdIsBroken[sessionNum] = false;
+                return Task.CompletedTask;
+            }
+        };
         var settings = new YdbConnectionStringBuilder
             { MaxSessionPool = maxSessionSize, MinSessionPool = minSessionSize };
         var sessionSource = new PoolingSessionSource<MockPoolingSession>(mockFactory, settings);
@@ -273,6 +287,7 @@ public async Task StressTest_HighContention_OpenClose()
                     while (!cts.IsCancellationRequested)
                     {
                         var session = await sessionSource.OpenSession(cts.Token);
+                        Assert.False(sessionIdIsBroken[session.SessionId()]);
                         session.Close();
                         await Task.Delay(Random.Shared.Next(maxSessionSize), cts.Token);
                     }
@@ -312,7 +327,7 @@ public async Task Get_Session_From_Exhausted_Pool()
     public async Task Return_IsBroken_Session()
     {
         const int maxSessionSize = 10;
-        var mockFactory = new MockPoolingSessionFactory(maxSessionSize) { IsBroken = () => true };
+        var mockFactory = new MockPoolingSessionFactory(maxSessionSize) { IsBroken = _ => true };
         var settings = new YdbConnectionStringBuilder
         {
             MaxSessionPool = maxSessionSize,
@@ -336,7 +351,7 @@ public async Task CheckIdleSession_WhenIsBrokenInStack_CreateNewSession()
         var isBroken = false;
         const int maxSessionSize = 10;
         // ReSharper disable once AccessToModifiedClosure
-        var mockFactory = new MockPoolingSessionFactory(maxSessionSize) { IsBroken = () => isBroken };
+        var mockFactory = new MockPoolingSessionFactory(maxSessionSize) { IsBroken = _ => isBroken };
         var settings = new YdbConnectionStringBuilder
         {
             MaxSessionPool = maxSessionSize,
@@ -384,7 +399,7 @@ internal class MockPoolingSessionFactory(int maxSessionSize) : IPoolingSessionFa
     internal int NumSession => Volatile.Read(ref _numSession);
 
     internal Func<int, Task> Open { private get; init; } = _ => Task.CompletedTask;
-    internal Func<bool> IsBroken { private get; init; } = () => false;
+    internal Func<int, bool> IsBroken { private get; init; } = _ => false;
     internal Func<ValueTask> Dispose { private get; init; } = () => ValueTask.CompletedTask;
 
     public MockPoolingSession NewSession(PoolingSessionSource<MockPoolingSession> source) =>
@@ -403,7 +418,7 @@ public MockPoolingSession NewSession(PoolingSessionSource<MockPoolingSession> so
 
                 return Task.CompletedTask;
             },
-            IsBroken,
+            () => IsBroken(_numSession),
             Interlocked.Increment(ref _sessionOpened)
         );
 

src/Ydb.Sdk/test/Ydb.Sdk.Ado.Tests/YdbConnectionTests.cs

@@ -19,19 +19,23 @@ public sealed class YdbConnectionTests : TestBase
     [Fact]
     public async Task ClearPool_WhenHasActiveConnection_CloseActiveConnectionOnClose()
     {
-        var connectionString = ConnectionString + ";MaxSessionPool=100";
-
-        var tasks = GenerateTasks(connectionString);
-        tasks.Add(YdbConnection.ClearPool(new YdbConnection(connectionString)));
-        tasks.AddRange(GenerateTasks(connectionString));
-        await Task.WhenAll(tasks);
-        Assert.Equal(999000, _counter);
-
-        tasks = GenerateTasks(connectionString);
-        tasks.Add(YdbConnection.ClearPool(new YdbConnection(connectionString)));
-        await Task.WhenAll(tasks);
-        Assert.Equal(1498500, _counter);
-        await YdbConnection.ClearPool(new YdbConnection(connectionString));
+        for (var i = 0; i < 10; i++)
+        {
+            var connectionString = ConnectionString + ";MaxSessionPool=100";
+
+            var tasks = GenerateTasks(connectionString);
+            tasks.Add(YdbConnection.ClearPool(new YdbConnection(connectionString)));
+            tasks.AddRange(GenerateTasks(connectionString));
+            await Task.WhenAll(tasks);
+            Assert.Equal(999000, _counter);
+
+            tasks = GenerateTasks(connectionString);
+            tasks.Add(YdbConnection.ClearPool(new YdbConnection(connectionString)));
+            await Task.WhenAll(tasks);
+            Assert.Equal(1498500, _counter);
+            await YdbConnection.ClearPool(new YdbConnection(connectionString));
+            _counter = 0;
+        }
     }
 
     // docker cp ydb-local:/ydb_certs/ca.pem ~/