feat: update ydb transaction semantics

Author: KirillKurdyukov

src/Ydb.Sdk/src/Ado/YdbCommand.cs

@@ -185,13 +185,14 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha
             ? new ExecuteQuerySettings { TransportTimeout = TimeSpan.FromSeconds(CommandTimeout) }
             : new ExecuteQuerySettings();
 
+        var transaction = YdbConnection.CurrentTransaction;
+
         var ydbDataReader = await YdbDataReader.CreateYdbDataReader(YdbConnection.Session.ExecuteQuery(
-                preparedSql.ToString(), ydbParameters, execSettings, Transaction?.TransactionControl),
-            YdbConnection.Session.OnStatus, Transaction);
+                preparedSql.ToString(), ydbParameters, execSettings, transaction?.TransactionControl),
+            YdbConnection.Session.OnStatus, transaction);
 
         YdbConnection.LastReader = ydbDataReader;
         YdbConnection.LastCommand = CommandText;
-        YdbConnection.LastTransaction = Transaction;
 
         return ydbDataReader;
     }

src/Ydb.Sdk/src/Ado/YdbConnection.cs

@@ -67,7 +67,16 @@ public YdbTransaction BeginTransaction(TxMode txMode = TxMode.SerializableRw)
     {
         EnsureConnectionOpen();
 
-        return new YdbTransaction(this, txMode);
+        if (CurrentTransaction is { Completed: false })
+        {
+            throw new InvalidOperationException(
+                "A transaction is already in progress; nested/concurrent transactions aren't supported."
+            );
+        }
+
+        CurrentTransaction = new YdbTransaction(this, txMode);
+
+        return CurrentTransaction;
     }
 
     public override void ChangeDatabase(string databaseName)
@@ -121,9 +130,9 @@ public override async Task CloseAsync()
                 await LastReader.CloseAsync();
             }
 
-            if (LastTransaction is { Completed: false })
+            if (CurrentTransaction is { Completed: false })
             {
-                await LastTransaction.RollbackAsync();
+                await CurrentTransaction.RollbackAsync();
             }
 
             OnStateChange(OpenToClosedEventArgs);
@@ -160,8 +169,8 @@ public override string ConnectionString
 
     internal YdbDataReader? LastReader { get; set; }
     internal string LastCommand { get; set; } = string.Empty;
-    internal YdbTransaction? LastTransaction { get; set; }
     internal bool IsBusy => LastReader is { IsClosed: false };
+    internal YdbTransaction? CurrentTransaction { get; private set; }
 
     public override string DataSource => string.Empty; // TODO
 

src/Ydb.Sdk/src/Ado/YdbDataReader.cs

@@ -497,7 +497,7 @@ private async Task<State> NextExecPart()
             _currentResultSet = part.ResultSet?.FromProto();
             ReaderMetadata = _currentResultSet != null ? new Metadata(_currentResultSet) : EmptyMetadata.Instance;
 
-            if (_ydbTransaction != null)
+            if (_ydbTransaction != null && part.TxMeta != null)
             {
                 _ydbTransaction.TxId ??= part.TxMeta.Id;
             }

src/Ydb.Sdk/src/Ado/YdbTransaction.cs

@@ -10,6 +10,8 @@ public sealed class YdbTransaction : DbTransaction
     private readonly TxMode _txMode;
 
     private bool _failed;
+    private YdbConnection? _ydbConnection;
+    private bool _isDisposed;
 
     internal string? TxId { get; set; }
     internal bool Completed { get; private set; }
@@ -32,7 +34,7 @@ internal bool Failed
 
     internal YdbTransaction(YdbConnection ydbConnection, TxMode txMode)
     {
-        DbConnection = ydbConnection;
+        _ydbConnection = ydbConnection;
         _txMode = txMode;
     }
 
@@ -44,7 +46,7 @@ public override void Commit()
     // TODO propagate cancellation token
     public override async Task CommitAsync(CancellationToken cancellationToken = new())
     {
-        await FinishTransaction(txId => DbConnection.Session.CommitTransaction(txId));
+        await FinishTransaction(txId => DbConnection!.Session.CommitTransaction(txId));
     }
 
     public override void Rollback()
@@ -62,36 +64,43 @@ public override void Rollback()
             return;
         }
 
-        await FinishTransaction(txId => DbConnection.Session.RollbackTransaction(txId));
+        await FinishTransaction(txId => DbConnection!.Session.RollbackTransaction(txId));
     }
 
-    protected override YdbConnection DbConnection { get; }
+    protected override YdbConnection? DbConnection
+    {
+        get
+        {
+            CheckDisposed();
+            return _ydbConnection;
+        }
+    }
 
     public override IsolationLevel IsolationLevel => _txMode == TxMode.SerializableRw
         ? IsolationLevel.Serializable
         : IsolationLevel.Unspecified;
 
     private async Task FinishTransaction(Func<string, Task<Status>> finishMethod)
     {
-        if (Completed || DbConnection.State == ConnectionState.Closed)
+        if (DbConnection?.State == ConnectionState.Closed || Completed)
         {
             throw new InvalidOperationException("This YdbTransaction has completed; it is no longer usable");
         }
 
-        if (DbConnection.IsBusy)
+        if (DbConnection!.IsBusy)
         {
             throw new YdbOperationInProgressException(DbConnection);
         }
 
-        Completed = true;
-
-        if (TxId == null)
-        {
-            return; // transaction isn't started
-        }
-
         try
         {
+            Completed = true;
+
+            if (TxId == null)
+            {
+                return; // transaction isn't started
+            }
+
             var status = await finishMethod(TxId); // Commit or Rollback
 
             if (status.IsNotSuccess)
@@ -111,5 +120,43 @@ private async Task FinishTransaction(Func<string, Task<Status>> finishMethod)
 
             throw new YdbException(e.Status);
         }
+        finally
+        {
+            _ydbConnection = null;
+        }
+    }
+
+    protected override void Dispose(bool disposing)
+    {
+        if (_isDisposed || !disposing)
+            return;
+
+        if (!Completed)
+        {
+            Rollback();
+        }
+
+        _isDisposed = true;
+    }
+
+    public override async ValueTask DisposeAsync()
+    {
+        if (_isDisposed)
+            return;
+
+        if (!Completed)
+        {
+            await RollbackAsync();
+        }
+
+        _isDisposed = true;
+    }
+
+    private void CheckDisposed()
+    {
+        if (_isDisposed)
+        {
+            throw new ObjectDisposedException(nameof(YdbTransaction));
+        }
     }
 }

src/Ydb.Sdk/tests/Ado/Specification/YdbTransactionTests.cs

@@ -0,0 +1,10 @@
+using AdoNet.Specification.Tests;
+
+namespace Ydb.Sdk.Tests.Ado.Specification;
+
+public class YdbTransactionTests : TransactionTestBase<YdbFactoryFixture>
+{
+    public YdbTransactionTests(YdbFactoryFixture fixture) : base(fixture)
+    {
+    }
+}