- ADO.NET: session is now deactivated when cancelled.

KirillKurdyukov · KirillKurdyukov · commit bbdef73d5fb5 · 2025-06-24T14:58:50.000+03:00
- Fixed bug ADO.NET: throws an `InvalidOperationException` if the connection is broken during the next invocation.
diff --git a/.github/workflows/slo.yml b/.github/workflows/slo.yml
@@ -31,7 +31,7 @@ jobs:
       matrix:
         workload:
           - AdoNet
-#          - EF
+          - EF
     
     concurrency:
       group: slo-${{ github.ref }}-${{ matrix.workload }}
diff --git a/src/Ydb.Sdk/src/Ado/YdbCommand.cs b/src/Ydb.Sdk/src/Ado/YdbCommand.cs
@@ -174,7 +174,7 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha
             throw new YdbOperationInProgressException(YdbConnection);
         }
 
-        YdbConnection.EnsureConnectionOpen();
+        YdbConnection.ThrowIfConnectionClosed();
 
         var ydbParameters = DbParameterCollection.YdbParameters;
         var (sql, paramNames) = SqlParser.Parse(
diff --git a/src/Ydb.Sdk/src/Ado/YdbConnection.cs b/src/Ydb.Sdk/src/Ado/YdbConnection.cs
@@ -28,7 +28,7 @@ internal Session Session
     {
         get
         {
-            EnsureConnectionOpen();
+            ThrowIfConnectionClosed();
 
             return _session;
         }
@@ -53,7 +53,7 @@ public YdbConnection(YdbConnectionStringBuilder connectionStringBuilder)
 
     protected override YdbTransaction BeginDbTransaction(IsolationLevel isolationLevel)
     {
-        EnsureConnectionOpen();
+        ThrowIfConnectionClosed();
 
         return BeginTransaction(isolationLevel switch
         {
@@ -66,7 +66,7 @@ protected override YdbTransaction BeginDbTransaction(IsolationLevel isolationLev
 
     public YdbTransaction BeginTransaction(TxMode txMode = TxMode.SerializableRw)
     {
-        EnsureConnectionOpen();
+        ThrowIfConnectionClosed();
 
         if (CurrentTransaction is { Completed: false })
         {
@@ -90,7 +90,7 @@ public override void ChangeDatabase(string databaseName)
 
     public override async Task OpenAsync(CancellationToken cancellationToken)
     {
-        EnsureConnectionClosed();
+        ThrowIfConnectionOpen();
 
         try
         {
@@ -152,7 +152,7 @@ public override string ConnectionString
         set
 #pragma warning restore CS8765 // Nullability of type of parameter doesn't match overridden member (possibly because of nullability attributes).
         {
-            EnsureConnectionClosed();
+            ThrowIfConnectionOpen();
 
             // ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract
             _connectionStringBuilder = value != null ? new YdbConnectionStringBuilder(value) : null;
@@ -186,7 +186,7 @@ public override string ServerVersion
     {
         get
         {
-            EnsureConnectionOpen();
+            ThrowIfConnectionClosed();
 
             return string.Empty; // TODO ServerVersion
         }
@@ -217,17 +217,17 @@ public override Task<DataTable> GetSchemaAsync(
         CancellationToken cancellationToken = default
     ) => YdbSchema.GetSchemaAsync(this, collectionName, restrictionValues, cancellationToken);
 
-    internal void EnsureConnectionOpen()
+    internal void ThrowIfConnectionClosed()
     {
-        if (ConnectionState == ConnectionState.Closed)
+        if (ConnectionState is ConnectionState.Closed or ConnectionState.Broken)
         {
             throw new InvalidOperationException("Connection is closed");
         }
     }
 
-    private void EnsureConnectionClosed()
+    private void ThrowIfConnectionOpen()
     {
-        if (ConnectionState != ConnectionState.Closed)
+        if (ConnectionState == ConnectionState.Open)
         {
             throw new InvalidOperationException("Connection already open");
         }
diff --git a/src/Ydb.Sdk/src/Pool/SessionPool.cs b/src/Ydb.Sdk/src/Pool/SessionPool.cs
@@ -225,6 +225,7 @@ internal void OnStatus(Status status)
     {
         // ReSharper disable once InvertIf
         if (status.StatusCode is
+            StatusCode.Cancelled or
             StatusCode.BadSession or
             StatusCode.SessionBusy or
             StatusCode.InternalError or
diff --git a/src/Ydb.Sdk/tests/Ado/Specification/YdbConnectionTests.cs b/src/Ydb.Sdk/tests/Ado/Specification/YdbConnectionTests.cs
@@ -1,5 +1,6 @@
 using AdoNet.Specification.Tests;
 using Xunit;
+using Ydb.Sdk.Ado;
 
 namespace Ydb.Sdk.Tests.Ado.Specification;
 
@@ -33,11 +34,12 @@ public override void ServerVersion_returns_value()
         base.ServerVersion_returns_value();
     }
 
-#pragma warning disable xUnit1004
-    [Fact(Skip = "TODO Supported cancel OpenAsync.")]
-#pragma warning restore xUnit1004
-    public override Task OpenAsync_is_canceled()
+    public override async Task OpenAsync_is_canceled()
     {
-        return base.OpenAsync_is_canceled();
+        await using var connection = CreateConnection();
+        connection.ConnectionString = ConnectionString;
+        var task = connection.OpenAsync(CanceledToken);
+        await Assert.ThrowsAnyAsync<YdbException>(() => task);
+        Assert.True(task.IsFaulted);
     }
 }
diff --git a/src/Ydb.Sdk/tests/Ado/YdbConnectionTests.cs b/src/Ydb.Sdk/tests/Ado/YdbConnectionTests.cs
@@ -222,10 +222,11 @@ public async Task OpenAsync_WhenCancelTokenIsCanceled_ThrowYdbException()
         Assert.Equal("The connection pool has been exhausted, either raise 'MaxSessionPool' (currently 10) " +
                      "or 'CreateSessionTimeout' (currently 5 seconds) in your connection string.",
             (await Assert.ThrowsAsync<YdbException>(async () => await connection.OpenAsync(cts.Token))).Message);
+        Assert.Equal(ConnectionState.Closed, connection.State);
     }
 
     [Fact]
-    public async Task YdbCommand_WhenCancelTokenIsCanceled_ThrowYdbException()
+    public async Task YdbDataReader_WhenCancelTokenIsCanceled_ThrowYdbException()
     {
         await using var connection = await CreateOpenConnectionAsync();
         var command = new YdbCommand(connection) { CommandText = "SELECT 1; SELECT 1; SELECT 1;" };
@@ -236,16 +237,13 @@ public async Task YdbCommand_WhenCancelTokenIsCanceled_ThrowYdbException()
         await ydbDataReader.ReadAsync(cts.Token); // first part in memory
         Assert.False(ydbDataReader.IsClosed);
         Assert.Equal(1, ydbDataReader.GetValue(0));
-
+        Assert.Equal(ConnectionState.Open, connection.State);
         Assert.Equal(StatusCode.Cancelled, (await Assert.ThrowsAsync<YdbException>(
             async () => await ydbDataReader.NextResultAsync(cts.Token))).Code);
         Assert.True(ydbDataReader.IsClosed);
-        Assert.Equal(StatusCode.Cancelled, (await Assert.ThrowsAsync<YdbException>(
-            async () => await command.ExecuteReaderAsync(cts.Token))).Code);
-        Assert.Equal(StatusCode.Cancelled, (await Assert.ThrowsAsync<YdbException>(
-            async () => await command.ExecuteScalarAsync(cts.Token))).Code);
-        Assert.Equal(StatusCode.Cancelled, (await Assert.ThrowsAsync<YdbException>(
-            async () => await command.ExecuteNonQueryAsync(cts.Token))).Code);
+        Assert.Equal(ConnectionState.Broken, connection.State);
+        // ReSharper disable once MethodSupportsCancellation
+        await connection.OpenAsync();
 
         // ReSharper disable once MethodSupportsCancellation
         ydbDataReader = await command.ExecuteReaderAsync();
@@ -259,16 +257,40 @@ public async Task YdbCommand_WhenCancelTokenIsCanceled_ThrowYdbException()
         Assert.Equal(StatusCode.Cancelled, (await Assert.ThrowsAsync<YdbException>(
             async () => await ydbDataReader.NextResultAsync(cts.Token))).Code);
         Assert.True(ydbDataReader.IsClosed);
+        Assert.Equal(ConnectionState.Broken, connection.State);
     }
 
     [Fact]
-    public async Task ExecuteMethods_WhenExecutedYdbDataReaderThenCancelTokenIsCanceled_ReturnValues()
+    public async Task ExecuteMethods_WhenCancelTokenIsCanceled_ConnectionIsBroken()
     {
         await using var connection = await CreateOpenConnectionAsync();
-        var ydbCommand = new YdbCommand(connection) { CommandText = "SELECT 1; SELECT 1; "};
+        var command = new YdbCommand(connection) { CommandText = "SELECT 1; SELECT 1; SELECT 1;" };
+        using var cts = new CancellationTokenSource();
+        cts.Cancel();
+
+        Assert.Equal(StatusCode.Cancelled, (await Assert.ThrowsAsync<YdbException>(
+            async () => await command.ExecuteReaderAsync(cts.Token))).Code);
+        Assert.Equal(ConnectionState.Broken, connection.State);
+        // ReSharper disable once MethodSupportsCancellation
+        await connection.OpenAsync();
+        Assert.Equal(StatusCode.Cancelled, (await Assert.ThrowsAsync<YdbException>(
+            async () => await command.ExecuteScalarAsync(cts.Token))).Code);
+        Assert.Equal(ConnectionState.Broken, connection.State);
+        // ReSharper disable once MethodSupportsCancellation
+        await connection.OpenAsync();
+        Assert.Equal(StatusCode.Cancelled, (await Assert.ThrowsAsync<YdbException>(
+            async () => await command.ExecuteNonQueryAsync(cts.Token))).Code);
+        Assert.Equal(ConnectionState.Broken, connection.State);
+    }
+
+    [Fact]
+    public async Task ExecuteReaderAsync_WhenExecutedYdbDataReaderThenCancelTokenIsCanceled_ReturnValues()
+    {
+        await using var connection = await CreateOpenConnectionAsync();
+        var ydbCommand = new YdbCommand(connection) { CommandText = "SELECT 1; SELECT 1; " };
         var cts = new CancellationTokenSource();
         var ydbDataReader = await ydbCommand.ExecuteReaderAsync(cts.Token);
-        
+
         await ydbDataReader.ReadAsync(cts.Token);
         Assert.Equal(1, ydbDataReader.GetValue(0));
         Assert.True(await ydbDataReader.NextResultAsync(cts.Token));

Original file line number	Diff line number	Diff line change
`@@ -174,7 +174,7 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha`
`174`	`174`	`throw new YdbOperationInProgressException(YdbConnection);`
`175`	`175`	`}`
`176`	`176`
`177`		`- YdbConnection.EnsureConnectionOpen();`
	`177`	`+ YdbConnection.ThrowIfConnectionClosed();`
`178`	`178`
`179`	`179`	`var ydbParameters = DbParameterCollection.YdbParameters;`
`180`	`180`	`var (sql, paramNames) = SqlParser.Parse(`
Original file line number	Diff line number	Diff line change
`@@ -28,7 +28,7 @@ internal Session Session`
`28`	`28`	`{`
`29`	`29`	`get`
`30`	`30`	`{`
`31`		`- EnsureConnectionOpen();`
	`31`	`+ ThrowIfConnectionClosed();`
`32`	`32`
`33`	`33`	`return _session;`
`34`	`34`	`}`
`@@ -53,7 +53,7 @@ public YdbConnection(YdbConnectionStringBuilder connectionStringBuilder)`
`53`	`53`
`54`	`54`	`protected override YdbTransaction BeginDbTransaction(IsolationLevel isolationLevel)`
`55`	`55`	`{`
`56`		`- EnsureConnectionOpen();`
	`56`	`+ ThrowIfConnectionClosed();`
`57`	`57`
`58`	`58`	`return BeginTransaction(isolationLevel switch`
`59`	`59`	`{`
`@@ -66,7 +66,7 @@ protected override YdbTransaction BeginDbTransaction(IsolationLevel isolationLev`
`66`	`66`
`67`	`67`	`public YdbTransaction BeginTransaction(TxMode txMode = TxMode.SerializableRw)`
`68`	`68`	`{`
`69`		`- EnsureConnectionOpen();`
	`69`	`+ ThrowIfConnectionClosed();`
`70`	`70`
`71`	`71`	`if (CurrentTransaction is { Completed: false })`
`72`	`72`	`{`
`@@ -90,7 +90,7 @@ public override void ChangeDatabase(string databaseName)`
`90`	`90`
`91`	`91`	`public override async Task OpenAsync(CancellationToken cancellationToken)`
`92`	`92`	`{`
`93`		`- EnsureConnectionClosed();`
	`93`	`+ ThrowIfConnectionOpen();`
`94`	`94`
`95`	`95`	`try`
`96`	`96`	`{`
`@@ -152,7 +152,7 @@ public override string ConnectionString`
`152`	`152`	`set`
`153`	`153`	`#pragma warning restore CS8765 // Nullability of type of parameter doesn't match overridden member (possibly because of nullability attributes).`
`154`	`154`	`{`
`155`		`- EnsureConnectionClosed();`
	`155`	`+ ThrowIfConnectionOpen();`
`156`	`156`
`157`	`157`	`// ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract`
`158`	`158`	`_connectionStringBuilder = value != null ? new YdbConnectionStringBuilder(value) : null;`
`@@ -186,7 +186,7 @@ public override string ServerVersion`
`186`	`186`	`{`
`187`	`187`	`get`
`188`	`188`	`{`
`189`		`- EnsureConnectionOpen();`
	`189`	`+ ThrowIfConnectionClosed();`
`190`	`190`
`191`	`191`	`return string.Empty; // TODO ServerVersion`
`192`	`192`	`}`
`@@ -217,17 +217,17 @@ public override Task<DataTable> GetSchemaAsync(`
`217`	`217`	`CancellationToken cancellationToken = default`
`218`	`218`	`) => YdbSchema.GetSchemaAsync(this, collectionName, restrictionValues, cancellationToken);`
`219`	`219`
`220`		`- internal void EnsureConnectionOpen()`
	`220`	`+ internal void ThrowIfConnectionClosed()`
`221`	`221`	`{`
`222`		`- if (ConnectionState == ConnectionState.Closed)`
	`222`	`+ if (ConnectionState is ConnectionState.Closed or ConnectionState.Broken)`
`223`	`223`	`{`
`224`	`224`	`throw new InvalidOperationException("Connection is closed");`
`225`	`225`	`}`
`226`	`226`	`}`
`227`	`227`
`228`		`- private void EnsureConnectionClosed()`
	`228`	`+ private void ThrowIfConnectionOpen()`
`229`	`229`	`{`
`230`		`- if (ConnectionState != ConnectionState.Closed)`
	`230`	`+ if (ConnectionState == ConnectionState.Open)`
`231`	`231`	`{`
`232`	`232`	`throw new InvalidOperationException("Connection already open");`
`233`	`233`	`}`
Original file line number	Diff line number	Diff line change
`@@ -225,6 +225,7 @@ internal void OnStatus(Status status)`
`225`	`225`	`{`
`226`	`226`	`// ReSharper disable once InvertIf`
`227`	`227`	`if (status.StatusCode is`
	`228`	`+ StatusCode.Cancelled or`
`228`	`229`	`StatusCode.BadSession or`
`229`	`230`	`StatusCode.SessionBusy or`
`230`	`231`	`StatusCode.InternalError or`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
`1`	`1`	`using AdoNet.Specification.Tests;`
`2`	`2`	`using Xunit;`
	`3`	`+using Ydb.Sdk.Ado;`
`3`	`4`
`4`	`5`	`namespace Ydb.Sdk.Tests.Ado.Specification;`
`5`	`6`
`@@ -33,11 +34,12 @@ public override void ServerVersion_returns_value()`
`33`	`34`	`base.ServerVersion_returns_value();`
`34`	`35`	`}`
`35`	`36`
`36`		`-#pragma warning disable xUnit1004`
`37`		`- [Fact(Skip = "TODO Supported cancel OpenAsync.")]`
`38`		`-#pragma warning restore xUnit1004`
`39`		`- public override Task OpenAsync_is_canceled()`
	`37`	`+ public override async Task OpenAsync_is_canceled()`
`40`	`38`	`{`
`41`		`- return base.OpenAsync_is_canceled();`
	`39`	`+ await using var connection = CreateConnection();`
	`40`	`+ connection.ConnectionString = ConnectionString;`
	`41`	`+ var task = connection.OpenAsync(CanceledToken);`
	`42`	`+ await Assert.ThrowsAnyAsync<YdbException>(() => task);`
	`43`	`+ Assert.True(task.IsFaulted);`
`42`	`44`	`}`
`43`	`45`	`}`