feat: Added KeepAlivePingDelay and KeepAlivePingTimeout (#290)

Author: KirillKurdyukov

.github/scripts/publish.sh

@@ -37,7 +37,7 @@ else
   VERSION="$MAJOR.$MINOR.$PATCH";
 fi;
 TAG="v$VERSION"
-echo "## v$VERSION" >> $CHANGELOG_FILE.tmp
+echo "## v$VERSION\n" >> $CHANGELOG_FILE.tmp
 cat $CHANGELOG_FILE >> $CHANGELOG_FILE.tmp
 mv $CHANGELOG_FILE.tmp $CHANGELOG_FILE;
 git add $CHANGELOG_FILE;

CHANGELOG.md

@@ -1,16 +1,24 @@
+- Added `KeepAlivePingTimeout`, with a default value of 10 seconds.
+- Added `KeepAlivePingDelay`, with a default value of 10 seconds.
+
 ## v0.15.3
+
 - Added SeqNo to `Ydb.Sdk.Services.Topic.Reader.Message`.
 
 ## v0.15.2
+
 - Added SeqNo to `WriteResult`.
 - Changed signature of the `TopicClient.DropTopic` method.
 
 ## v0.15.1
-- Fixed Writer: possible creation of a session after `DisposeAsync()`, which this could happen when there are canceled tasks in `InFlightMessages`.
+
+- Fixed Writer: possible creation of a session after `DisposeAsync()`, which this could happen when there are canceled
+  tasks in `InFlightMessages`.
 - Dev: `Writer.MoveNext()` changed exception on cancelToken from `WriterException` to `TaskCanceledException`.
 - Dev: changed log level from `Warning` to `Information` in `(Reader / Writer).Initialize()` when it is disposed.
 
 ## v0.15.0
+
 - Dev: added `ValueTask<string?> GetAuthInfoAsync()` in ICredentialProvider.
 - Feat: `Writer.DisposeAsync()` waits for all in-flight messages to complete.
 - Feat: `Reader.DisposeAsync()` waits for all pending commits to be completed.

src/Ydb.Sdk/src/Ado/YdbConnectionStringBuilder.cs

@@ -27,6 +27,8 @@ private void InitDefaultValues()
         _database = "/local";
         _maxSessionPool = 100;
         _useTls = false;
+        _keepAlivePingDelay = SocketHttpHandlerDefaults.DefaultKeepAlivePingSeconds;
+        _keepAlivePingTimeout = SocketHttpHandlerDefaults.DefaultKeepAlivePingTimeoutSeconds;
     }
 
     public string Host
@@ -137,6 +139,41 @@ public string? RootCertificate
 
     private string? _rootCertificate;
 
+    public int KeepAlivePingDelay
+    {
+        get => _keepAlivePingDelay;
+        set
+        {
+            if (value < 0)
+            {
+                throw new ArgumentOutOfRangeException(nameof(value), value, "Invalid keep alive ping delay: " + value);
+            }
+
+            _keepAlivePingDelay = value;
+            SaveValue(nameof(KeepAlivePingDelay), value);
+        }
+    }
+
+    private int _keepAlivePingDelay;
+
+    public int KeepAlivePingTimeout
+    {
+        get => _keepAlivePingTimeout;
+        set
+        {
+            if (value < 0)
+            {
+                throw new ArgumentOutOfRangeException(nameof(value), value,
+                    "Invalid keep alive ping timeout: " + value);
+            }
+
+            _keepAlivePingTimeout = value;
+            SaveValue(nameof(KeepAlivePingTimeout), value);
+        }
+    }
+
+    private int _keepAlivePingTimeout;
+
     public ILoggerFactory? LoggerFactory { get; init; }
 
     public ICredentialsProvider? CredentialsProvider { get; init; }
@@ -193,7 +230,15 @@ internal Task<Driver> BuildDriver()
             credentials: credentialsProvider,
             customServerCertificate: cert,
             customServerCertificates: ServerCertificates
-        ), LoggerFactory);
+        )
+        {
+            KeepAlivePingDelay = KeepAlivePingDelay == 0
+                ? Timeout.InfiniteTimeSpan
+                : TimeSpan.FromSeconds(KeepAlivePingDelay),
+            KeepAlivePingTimeout = KeepAlivePingTimeout == 0
+                ? Timeout.InfiniteTimeSpan
+                : TimeSpan.FromSeconds(KeepAlivePingTimeout)
+        }, LoggerFactory);
     }
 
     public override void Clear()
@@ -265,6 +310,12 @@ static YdbConnectionOption()
                 new YdbConnectionOption<string>(StringExtractor,
                     (builder, rootCertificate) => builder.RootCertificate = rootCertificate),
                 "RootCertificate", "Root Certificate");
+            AddOption(new YdbConnectionOption<int>(IntExtractor,
+                    (builder, keepAlivePingDelay) => builder.KeepAlivePingDelay = keepAlivePingDelay),
+                "KeepAlivePingDelay", "Keep Alive Ping Delay");
+            AddOption(new YdbConnectionOption<int>(IntExtractor,
+                    (builder, keepAlivePingTimeout) => builder.KeepAlivePingTimeout = keepAlivePingTimeout),
+                "KeepAlivePingTimeout", "Keep Alive Ping Timeout");
         }
 
         private static void AddOption(YdbConnectionOption option, params string[] keys)

src/Ydb.Sdk/src/DriverConfig.cs

@@ -9,6 +9,12 @@ public class DriverConfig
     public string Database { get; }
     public ICredentialsProvider Credentials { get; }
 
+    public TimeSpan KeepAlivePingDelay { get; set; } =
+        TimeSpan.FromSeconds(SocketHttpHandlerDefaults.DefaultKeepAlivePingSeconds);
+
+    public TimeSpan KeepAlivePingTimeout { get; set; } =
+        TimeSpan.FromSeconds(SocketHttpHandlerDefaults.DefaultKeepAlivePingTimeoutSeconds);
+
     internal X509Certificate2Collection CustomServerCertificates { get; } = new();
     internal TimeSpan EndpointDiscoveryInterval = TimeSpan.FromMinutes(1);
     internal TimeSpan EndpointDiscoveryTimeout = TimeSpan.FromSeconds(10);

src/Ydb.Sdk/src/Pool/ChannelPool.cs

@@ -77,13 +77,15 @@ internal class GrpcChannelFactory : IChannelFactory<GrpcChannel>
 {
     private readonly ILoggerFactory _loggerFactory;
     private readonly ILogger<GrpcChannelFactory> _logger;
-    private readonly X509Certificate2Collection _x509Certificate2Collection;
+    private readonly DriverConfig _config;
+
+    private X509Certificate2Collection ServerCertificates => _config.CustomServerCertificates;
 
     internal GrpcChannelFactory(ILoggerFactory loggerFactory, DriverConfig config)
     {
         _loggerFactory = loggerFactory;
         _logger = loggerFactory.CreateLogger<GrpcChannelFactory>();
-        _x509Certificate2Collection = config.CustomServerCertificates;
+        _config = config;
     }
 
     public GrpcChannel CreateChannel(string endpoint)
@@ -92,18 +94,24 @@ public GrpcChannel CreateChannel(string endpoint)
 
         var channelOptions = new GrpcChannelOptions
         {
-            LoggerFactory = _loggerFactory
+            LoggerFactory = _loggerFactory,
+            DisposeHttpClient = true
         };
 
-        var httpHandler = new SocketsHttpHandler();
+        var httpHandler = new SocketsHttpHandler
+        {
+            // https://github.com/grpc/proposal/blob/master/A8-client-side-keepalive.md
+            KeepAlivePingDelay = _config.KeepAlivePingDelay,
+            KeepAlivePingTimeout = _config.KeepAlivePingTimeout,
+            KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always
+        };
 
         // https://github.com/grpc/grpc-dotnet/issues/2312#issuecomment-1790661801
         httpHandler.Properties["__GrpcLoadBalancingDisabled"] = true;
 
         channelOptions.HttpHandler = httpHandler;
-        channelOptions.DisposeHttpClient = true;
 
-        if (_x509Certificate2Collection.Count == 0)
+        if (ServerCertificates.Count == 0)
         {
             return GrpcChannel.ForAddress(endpoint, channelOptions);
         }
@@ -124,11 +132,11 @@ public GrpcChannel CreateChannel(string endpoint)
                 try
                 {
                     chain.ChainPolicy.VerificationFlags = X509VerificationFlags.AllowUnknownCertificateAuthority;
-                    chain.ChainPolicy.ExtraStore.AddRange(_x509Certificate2Collection);
+                    chain.ChainPolicy.ExtraStore.AddRange(ServerCertificates);
 
-                    return chain.Build(new X509Certificate2(certificate)) && chain.ChainElements.Any(chainElement =>
-                        _x509Certificate2Collection.Any(trustedCert =>
-                            chainElement.Certificate.Thumbprint == trustedCert.Thumbprint));
+                    return chain.Build(new X509Certificate2(certificate)) &&
+                           chain.ChainElements.Any(chainElement => ServerCertificates.Any(trustedCert =>
+                               chainElement.Certificate.Thumbprint == trustedCert.Thumbprint));
                 }
                 catch (Exception e)
                 {

src/Ydb.Sdk/src/SocketHttpHandlerDefaultSettings.cs

@@ -0,0 +1,14 @@
+namespace Ydb.Sdk;
+
+internal static class SocketHttpHandlerDefaults
+{
+    /// <summary>
+    /// Default interval (in seconds) for sending keep-alive ping messages.
+    /// </summary>
+    internal const int DefaultKeepAlivePingSeconds = 10;
+
+    /// <summary>
+    /// Default timeout (in seconds) for receiving a response to a keep-alive ping.
+    /// </summary>
+    internal const int DefaultKeepAlivePingTimeoutSeconds = 10;
+}

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

@@ -16,6 +16,8 @@ public void InitDefaultValues_WhenEmptyConstructorInvoke_ReturnDefaultConnection
         Assert.Equal(100, connectionString.MaxSessionPool);
         Assert.Null(connectionString.User);
         Assert.Null(connectionString.Password);
+        Assert.Equal(10, connectionString.KeepAlivePingDelay);
+        Assert.Equal(10, connectionString.KeepAlivePingTimeout);
         Assert.Equal("", connectionString.ConnectionString);
     }
 
@@ -33,16 +35,19 @@ public void InitConnectionStringBuilder_WhenUnexpectedKey_ThrowException()
     public void InitConnectionStringBuilder_WhenExpectedKeys_ReturnUpdatedConnectionString()
     {
         var connectionString =
-            new YdbConnectionStringBuilder("Host=server;Port=2135;Database=/my/path;User=Kirill;UseTls=true");
+            new YdbConnectionStringBuilder("Host=server;Port=2135;Database=/my/path;User=Kirill;UseTls=true;" +
+                                           "KeepAlivePingDelay=30;KeepAlivePingTimeout=60");
 
         Assert.Equal(2135, connectionString.Port);
         Assert.Equal("server", connectionString.Host);
         Assert.Equal("/my/path", connectionString.Database);
         Assert.Equal(100, connectionString.MaxSessionPool);
         Assert.Equal("Kirill", connectionString.User);
+        Assert.Equal(30, connectionString.KeepAlivePingDelay);
+        Assert.Equal(60, connectionString.KeepAlivePingTimeout);
         Assert.Null(connectionString.Password);
-        Assert.Equal("Host=server;Port=2135;Database=/my/path;User=Kirill;UseTls=True",
-            connectionString.ConnectionString);
+        Assert.Equal("Host=server;Port=2135;Database=/my/path;User=Kirill;UseTls=True;" +
+                     "KeepAlivePingDelay=30;KeepAlivePingTimeout=60", connectionString.ConnectionString);
     }
 
     [Fact]