feat: fix NPE on Describe method & invoke onStatus on !Success (#175)

CHANGELOG.md

@@ -1,3 +1,6 @@
+- Check status of the transport or server for an invalidated session
+- Fixed NPE in DescribeTable
+
 ## v0.6.0
 - ADO.NET over query-service
 - Add EndpointPool

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

@@ -175,8 +175,9 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha
             ? new ExecuteQuerySettings { TransportTimeout = TimeSpan.FromSeconds(CommandTimeout) }
             : ExecuteQuerySettings.DefaultInstance;
 
-        var ydbDataReader = await YdbDataReader.CreateYdbDataReader(YdbConnection.Session.ExecuteQuery(_commandText,
-            DbParameterCollection.YdbParameters, execSettings, Transaction?.TransactionControl), Transaction);
+        var ydbDataReader = await YdbDataReader.CreateYdbDataReader(YdbConnection.Session.ExecuteQuery(
+                _commandText, DbParameterCollection.YdbParameters, execSettings, Transaction?.TransactionControl),
+            YdbConnection.Session.OnStatus, Transaction);
 
         YdbConnection.LastReader = ydbDataReader;
         YdbConnection.LastCommand = CommandText;

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

@@ -11,6 +11,7 @@ public sealed class YdbDataReader : DbDataReader, IAsyncEnumerable<YdbDataRecord
     private readonly IAsyncEnumerator<ExecuteQueryResponsePart> _stream;
     private readonly YdbTransaction? _ydbTransaction;
     private readonly RepeatedField<IssueMessage> _issueMessagesInStream = new();
+    private readonly Action<Status> _onStatus;
 
     private int _currentRowIndex = -1;
     private long _resultSetIndex = -1;
@@ -48,17 +49,22 @@ private interface IMetadata
     private Value.ResultSet.Row CurrentRow => CurrentResultSet.Rows[_currentRowIndex];
     private int RowsCount => ReaderMetadata.RowsCount;
 
-    private YdbDataReader(IAsyncEnumerator<ExecuteQueryResponsePart> resultSetStream, YdbTransaction? ydbTransaction)
+    private YdbDataReader(
+        IAsyncEnumerator<ExecuteQueryResponsePart> resultSetStream,
+        Action<Status> onStatus,
+        YdbTransaction? ydbTransaction)
     {
         _stream = resultSetStream;
+        _onStatus = onStatus;
         _ydbTransaction = ydbTransaction;
     }
 
     internal static async Task<YdbDataReader> CreateYdbDataReader(
         IAsyncEnumerator<ExecuteQueryResponsePart> resultSetStream,
+        Action<Status> onStatus,
         YdbTransaction? ydbTransaction = null)
     {
-        var ydbDataReader = new YdbDataReader(resultSetStream, ydbTransaction);
+        var ydbDataReader = new YdbDataReader(resultSetStream, onStatus, ydbTransaction);
         await ydbDataReader.Init();
 
         return ydbDataReader;
@@ -474,7 +480,11 @@ private async Task<State> NextExecPart()
                     _issueMessagesInStream.AddRange(_stream.Current.Issues);
                 }
 
-                throw new YdbException(Status.FromProto(part.Status, _issueMessagesInStream));
+                var status = Status.FromProto(part.Status, _issueMessagesInStream);
+
+                _onStatus(status);
+
+                throw new YdbException(status);
             }
 
             _currentResultSet = part.ResultSet?.FromProto();
@@ -498,6 +508,8 @@ private async Task<State> NextExecPart()
         {
             OnFailReadStream();
 
+            _onStatus(e.Status);
+
             throw new YdbException(e.Status);
         }
     }

src/Ydb.Sdk/src/Pool/SessionPool.cs

@@ -80,20 +80,26 @@ internal async Task<T> ExecOnSession<T>(Func<TSession, Task<T>> onSession, Retry
             }
             catch (Exception e)
             {
-                if (attempt == retrySettings.MaxAttempts - 1)
-                {
-                    throw;
-                }
-
                 var statusErr = e switch
                 {
                     Driver.TransportException transportException => transportException.Status,
                     StatusUnsuccessfulException unsuccessfulException => unsuccessfulException.Status,
                     _ => null
                 };
 
+                if (attempt == retrySettings.MaxAttempts - 1)
+                {
+                    if (statusErr != null)
+                    {
+                        session?.OnStatus(statusErr);
+                    }
+
+                    throw;
+                }
+
                 if (statusErr != null)
                 {
+                    session?.OnStatus(statusErr);
                     var retryRule = retrySettings.GetRetryRule(statusErr.StatusCode);
 
                     if (retryRule.Policy == RetryPolicy.None ||
@@ -103,8 +109,10 @@ internal async Task<T> ExecOnSession<T>(Func<TSession, Task<T>> onSession, Retry
                     }
 
                     Logger.LogTrace(
-                        "Retry: attempt {attempt}, Session ${session?.SessionId}, idempotent error {status} retrying",
+                        "Retry: attempt {attempt}, Session ${session.SessionId}, idempotent error {status} retrying",
                         attempt, session?.SessionId, statusErr);
+
+
                     await Task.Delay(retryRule.BackoffSettings.CalcBackoff(attempt));
                 }
                 else

src/Ydb.Sdk/src/Services/Table/DescribeTable.cs

@@ -65,7 +65,7 @@ public class TableStats
     public ulong StoreSize { get; }
     public ulong Partitions { get; }
     public DateTime CreationTime { get; }
-    public DateTime ModificationTime { get; }
+    public DateTime? ModificationTime { get; }
 
     internal TableStats(Ydb.Table.TableStats? proto)
     {
@@ -79,7 +79,7 @@ internal TableStats(Ydb.Table.TableStats? proto)
         StoreSize = proto.StoreSize;
         Partitions = proto.Partitions;
         CreationTime = proto.CreationTime.ToDateTime();
-        ModificationTime = proto.ModificationTime.ToDateTime();
+        ModificationTime = proto.ModificationTime?.ToDateTime();
     }
 }
 

src/Ydb.Sdk/tests/Ado/YdbDataReaderTests.cs

@@ -11,7 +11,8 @@ public class YdbDataReaderTests
     [Fact]
     public async Task BasedIteration_WhenNotCallMethodRead_ThrowException()
     {
-        var reader = await YdbDataReader.CreateYdbDataReader(EnumeratorSuccess());
+        var statuses = new List<Status>();
+        var reader = await YdbDataReader.CreateYdbDataReader(EnumeratorSuccess(), statuses.Add);
 
         // Read first metadata
         Assert.True(reader.HasRows);
@@ -36,19 +37,25 @@ public async Task BasedIteration_WhenNotCallMethodRead_ThrowException()
 
         Assert.Equal("The reader is closed",
             Assert.Throws<InvalidOperationException>(() => reader.GetValue(0)).Message);
+        Assert.Empty(statuses);
     }
 
     [Fact]
     public void CreateYdbDataReader_WhenAbortedStatus_ThrowException()
     {
+        var statuses = new List<Status>();
         Assert.Equal("Status: Aborted", Assert.Throws<YdbException>(
-            () => YdbDataReader.CreateYdbDataReader(SingleEnumeratorFailed).GetAwaiter().GetResult()).Message);
+                () => YdbDataReader.CreateYdbDataReader(SingleEnumeratorFailed, statuses.Add).GetAwaiter().GetResult())
+            .Message);
+        Assert.Single(statuses);
+        Assert.Equal(StatusCode.Aborted, statuses[0].StatusCode);
     }
 
     [Fact]
     public async Task NextResult_WhenNextResultSkipResultSet_ReturnNextResultSet()
     {
-        var reader = await YdbDataReader.CreateYdbDataReader(EnumeratorSuccess(2));
+        var statuses = new List<Status>();
+        var reader = await YdbDataReader.CreateYdbDataReader(EnumeratorSuccess(2), statuses.Add);
 
         Assert.True(reader.NextResult());
         Assert.True(reader.NextResult());
@@ -57,12 +64,14 @@ public async Task NextResult_WhenNextResultSkipResultSet_ReturnNextResultSet()
 
         Assert.False(reader.Read());
         Assert.False(reader.NextResult());
+        Assert.Empty(statuses);
     }
 
     [Fact]
     public async Task NextResult_WhenFirstRead_ReturnResultSet()
     {
-        var reader = await YdbDataReader.CreateYdbDataReader(EnumeratorSuccess(2));
+        var statuses = new List<Status>();
+        var reader = await YdbDataReader.CreateYdbDataReader(EnumeratorSuccess(2), statuses.Add);
 
         Assert.True(reader.Read());
         Assert.True((bool)reader.GetValue(0));
@@ -74,12 +83,14 @@ public async Task NextResult_WhenFirstRead_ReturnResultSet()
 
         Assert.False(reader.NextResult());
         Assert.False(reader.Read());
+        Assert.Empty(statuses);
     }
 
     [Fact]
     public async Task NextResult_WhenLongResultSet_ReturnResultSet()
     {
-        var reader = await YdbDataReader.CreateYdbDataReader(EnumeratorSuccess(2, true));
+        var statuses = new List<Status>();
+        var reader = await YdbDataReader.CreateYdbDataReader(EnumeratorSuccess(2, true), statuses.Add);
 
         Assert.True(reader.Read());
         Assert.True((bool)reader.GetValue(0));
@@ -93,11 +104,13 @@ public async Task NextResult_WhenLongResultSet_ReturnResultSet()
 
         Assert.False(reader.NextResult());
         Assert.False(reader.Read());
+        Assert.Empty(statuses);
     }
 
     [Fact]
     public async Task Read_WhenReadAsyncThrowException_AggregateIssuesBeforeErrorAndAfter()
     {
+        var statuses = new List<Status>();
         var result = ResultSet.Parser.ParseJson(
             "{ \"columns\": [ { \"name\": \"column0\", " +
             "\"type\": { \"typeId\": \"BOOL\" } } ], " +
@@ -116,7 +129,7 @@ public async Task Read_WhenReadAsyncThrowException_AggregateIssuesBeforeErrorAnd
         nextFailPart.Issues.Add(new IssueMessage { Message = "Some message 3" });
 
         var reader = await YdbDataReader.CreateYdbDataReader(new MockAsyncEnumerator<ExecuteQueryResponsePart>(
-            new List<ExecuteQueryResponsePart> { successPart, failPart, nextFailPart }));
+            new List<ExecuteQueryResponsePart> { successPart, failPart, nextFailPart }), statuses.Add);
 
         Assert.True(reader.Read());
         Assert.Equal(@"Status: Aborted, Issues:
@@ -125,6 +138,8 @@ public async Task Read_WhenReadAsyncThrowException_AggregateIssuesBeforeErrorAnd
 [0] Fatal: Some message 2
 [0] Fatal: Some message 3
 ", Assert.Throws<YdbException>(() => reader.Read()).Message);
+        Assert.Single(statuses);
+        Assert.Equal(StatusCode.Aborted, statuses[0].StatusCode);
     }
 
     private static MockAsyncEnumerator<ExecuteQueryResponsePart> EnumeratorSuccess(int size = 1,

src/Ydb.Sdk/tests/Table/DescribeTableTests.cs → src/Ydb.Sdk/tests/DescribeTableTests.cs

@@ -1,7 +1,8 @@
 using Xunit;
+using Ydb.Sdk.Services.Table;
 using Ydb.Sdk.Tests.Fixture;
 
-namespace Ydb.Sdk.Tests.Table;
+namespace Ydb.Sdk.Tests;
 
 [Trait("Category", "Integration")]
 public class DescribeTableTests : IClassFixture<TableClientFixture>
@@ -29,7 +30,10 @@ public async Task CreateAndDescribe()
 
         await Utils.CreateSimpleTable(_tableClientFixture.TableClient, tablePath, columnName);
 
-        var describeResponse = await _tableClientFixture.TableClient.DescribeTable(tablePath);
+        var describeResponse = await _tableClientFixture.TableClient.DescribeTable(tablePath, new DescribeTableSettings
+        {
+            OperationTimeout = TimeSpan.FromSeconds(5)
+        }.WithTableStats());
         describeResponse.Status.EnsureSuccess();
         Assert.True(describeResponse.Result.PrimaryKey.SequenceEqual(new[] { columnName }));