feat: simplify timeout settings (#175)

NguyenHoangSon96 · web-flow · commit 3c8accd46097 · 2025-09-15T20:56:12.000+07:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ### Features
 
 1. [#174](https://github.com/InfluxCommunity/influxdb3-csharp/pull/174): Support passing HttpClient to InfluxDBClient.
+1. [#175](https://github.com/InfluxCommunity/influxdb3-csharp/pull/175): Add QueryTimeout and WriteTimeout to ClientConfig.
 
 ### CI
 
diff --git a/Client.Test.Integration/QueryWriteTest.cs b/Client.Test.Integration/QueryWriteTest.cs
@@ -169,22 +169,92 @@ public async Task MaxReceiveMessageSize()
     }
 
     [Test]
-    public void GrpcDeadline()
+    public async Task TimeoutExceededByDeadline()
     {
         using var client = new InfluxDBClient(new ClientConfig
         {
             Host = Host,
             Token = Token,
             Database = Database,
+            WriteTimeout = TimeSpan.FromSeconds(11),
+            QueryTimeout = TimeSpan.FromSeconds(11),
             QueryOptions = new QueryOptions()
             {
-                Deadline = DateTime.UtcNow.AddMicroseconds(1)
+                Deadline = DateTime.UtcNow.AddTicks(1) // Deadline will have a higher priority than QueryTimeout
             }
         });
+        await client.WriteRecordAsync("mem,tag=a field=1");
+        TestQuery(client);
+        TestQueryBatches(client);
+        TestQueryPoints(client);
+    }
+
+    [Test]
+    public async Task TimeoutExceededByQueryTimeout()
+    {
+        using var client = new InfluxDBClient(new ClientConfig
+        {
+            Host = Host,
+            Token = Token,
+            Database = Database,
+            WriteTimeout = TimeSpan.FromSeconds(11),
+            QueryTimeout = TimeSpan.FromTicks(1)
+        });
+        await client.WriteRecordAsync("mem,tag=a field=1");
+        TestQuery(client);
+        TestQueryBatches(client);
+        TestQueryPoints(client);
+    }
+
+    [Test]
+    public async Task TimeoutExceeded()
+    {
+        using var client = new InfluxDBClient(new ClientConfig
+        {
+            Host = Host,
+            Token = Token,
+            Database = Database,
+            WriteTimeout = TimeSpan.FromSeconds(11),
+            QueryTimeout = TimeSpan.FromSeconds(11),
+            QueryOptions =
+            {
+                Deadline = DateTime.UtcNow.AddSeconds(11),
+            }
+        });
+        await client.WriteRecordAsync("mem,tag=a field=1");
+        var timeout = TimeSpan.FromTicks(1);
+        TestQuery(client, timeout);
+        TestQueryBatches(client, timeout);
+        TestQueryPoints(client, timeout);
+    }
+
+    private static void TestQuery(InfluxDBClient client, TimeSpan? timeout = null)
+    {
+        var ex = Assert.ThrowsAsync<RpcException>(async () =>
+        {
+            await foreach (var _ in client.Query("SELECT * FROM mem", timeout: timeout))
+            {
+            }
+        });
+        Assert.That(ex.StatusCode, Is.EqualTo(StatusCode.DeadlineExceeded));
+    }
 
+    private static void TestQueryBatches(InfluxDBClient client, TimeSpan? timeout = null)
+    {
+        var ex = Assert.ThrowsAsync<RpcException>(async () =>
+        {
+            await foreach (var _ in client.QueryBatches("SELECT * FROM mem", timeout: timeout))
+            {
+            }
+        });
+        Assert.That(ex.StatusCode, Is.EqualTo(StatusCode.DeadlineExceeded));
+    }
+
+    private static void TestQueryPoints(InfluxDBClient client, TimeSpan? timeout = null)
+    {
         var ex = Assert.ThrowsAsync<RpcException>(async () =>
         {
-            await foreach (var _ in client.Query("SELECT value FROM stat"))
+            await foreach (var _ in client.QueryPoints("SELECT * FROM mem", timeout: timeout))
             {
             }
         });
diff --git a/Client.Test.Integration/WriteTest.cs b/Client.Test.Integration/WriteTest.cs
@@ -33,6 +33,7 @@ public async Task WriteWithError()
                     Assert.That(iaex.Message,
                         Does.Contain("Found trailing content")
                             .Or.Contain("partial write of line protocol occurred")
+                            .Or.Contain("write buffer error: parsing for line protocol failed")
                     );
                     Assert.That(iaex.StatusCode.ToString(), Is.EqualTo("BadRequest"));
                     Assert.That(iaex.StatusCode, Is.EqualTo(HttpStatusCode.BadRequest));
diff --git a/Client.Test/InfluxDBClientQueryTest.cs b/Client.Test/InfluxDBClientQueryTest.cs
@@ -54,7 +54,7 @@ public async Task PassNamedParametersToFlightClient()
         var mockFlightSqlClient = new Mock<IFlightSqlClient>();
         mockFlightSqlClient
             .Setup(m => m.Execute(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<QueryType>(),
-                It.IsAny<Dictionary<string, object>>(), It.IsAny<Dictionary<string, string>>()))
+                It.IsAny<Dictionary<string, object>>(), It.IsAny<Dictionary<string, string>>(), It.IsAny<TimeSpan>()))
             .Returns(new List<RecordBatch>().ToAsyncEnumerable());
 
         //
@@ -74,9 +74,9 @@ public async Task PassNamedParametersToFlightClient()
             { "max-frequency", 3.5 }
         };
 
-        _ = await _client.QueryPoints(query, database: "my-db", queryType: queryType, namedParameters: namedParameters)
+        _ = await _client.QueryPoints(query, database: "my-db", queryType: queryType, namedParameters: namedParameters, timeout: TimeSpan.MaxValue)
             .ToListAsync();
-        mockFlightSqlClient.Verify(m => m.Execute(query, "my-db", queryType, namedParameters, new Dictionary<string, string>()), Times.Exactly(1));
+        mockFlightSqlClient.Verify(m => m.Execute(query, "my-db", queryType, namedParameters, new Dictionary<string, string>(), TimeSpan.MaxValue), Times.Exactly(1));
     }
 
     [Test]
@@ -109,7 +109,7 @@ public async Task PassHeadersToFlightClient()
         var mockFlightSqlClient = new Mock<IFlightSqlClient>();
         mockFlightSqlClient
             .Setup(m => m.Execute(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<QueryType>(),
-                It.IsAny<Dictionary<string, object>>(), It.IsAny<Dictionary<string, string>>()))
+                It.IsAny<Dictionary<string, object>>(), It.IsAny<Dictionary<string, string>>(), It.IsAny<TimeSpan?>()))
             .Returns(new List<RecordBatch>().ToAsyncEnumerable());
 
         //
@@ -128,6 +128,6 @@ public async Task PassHeadersToFlightClient()
         }};
         _ = await _client.QueryPoints(query, database: "my-db", queryType: queryType, headers: headers)
             .ToListAsync();
-        mockFlightSqlClient.Verify(m => m.Execute(query, "my-db", queryType, new Dictionary<string, object>(), headers), Times.Exactly(1));
+        mockFlightSqlClient.Verify(m => m.Execute(query, "my-db", queryType, new Dictionary<string, object>(), headers, null), Times.Exactly(1));
     }
 }
diff --git a/Client.Test/InfluxDBClientWriteTest.cs b/Client.Test/InfluxDBClientWriteTest.cs
@@ -2,13 +2,14 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Net;
-using System.Net.Http;
+using System.Threading;
 using System.Threading.Tasks;
 using InfluxDB3.Client.Config;
 using InfluxDB3.Client.Write;
 using WireMock.Matchers;
 using WireMock.RequestBuilders;
 using WireMock.ResponseBuilders;
+using WriteOptions = InfluxDB3.Client.Config.WriteOptions;
 
 namespace InfluxDB3.Client.Test;
 
@@ -493,55 +494,113 @@ public void WriteNoSyncTrueNotSupported()
     }
 
     [Test]
-    public async Task TestSetHttpClient()
+    public void TimeoutExceededByTimeout()
     {
         MockServer
             .Given(Request.Create().WithPath("/api/v2/write").UsingPost())
-            .RespondWith(Response.Create().WithStatusCode(HttpStatusCode.OK));
-
-        var httpClient = new HttpClient();
-        httpClient.DefaultRequestHeaders.UserAgent.ParseAdd("my-user-agent");
-        httpClient.DefaultRequestHeaders.Add("X-Client-ID", "123");
+            .RespondWith(Response.Create().WithStatusCode(204).WithDelay(TimeSpan.FromSeconds(2)));
 
         _client = new InfluxDBClient(new ClientConfig
         {
             Host = MockServerUrl,
             Token = "my-token",
             Database = "my-database",
-            HttpClient = httpClient
+            Timeout = TimeSpan.FromTicks(1)
         });
+        TestWriteRecordAsync(_client);
+        TestWriteRecordsAsync(_client);
+        TestWritePointAsync(_client);
+        TestWritePointsAsync(_client);
+    }
 
-        await _client.WriteRecordAsync("mem,tag=a field=1");
-        var requests = MockServer.LogEntries.ToList();
-        using (Assert.EnterMultipleScope())
+    [Test]
+    public void TimeoutExceededByWriteTimeout()
+    {
+        MockServer
+            .Given(Request.Create().WithPath("/api/v2/write").UsingPost())
+            .RespondWith(Response.Create().WithStatusCode(204).WithDelay(TimeSpan.FromSeconds(2)));
+
+        _client = new InfluxDBClient(new ClientConfig
         {
-            Assert.That(requests[0].RequestMessage.Headers?["User-Agent"].First(), Is.EqualTo("my-user-agent"));
-            Assert.That(requests[0].RequestMessage.Headers["X-Client-ID"].First(), Is.EqualTo("123"));
-        }
-        Assert.Pass();
+            Host = MockServerUrl,
+            Token = "my-token",
+            Database = "my-database",
+            QueryTimeout = TimeSpan.FromSeconds(11),
+            Timeout = TimeSpan.FromSeconds(11),
+            WriteTimeout = TimeSpan.FromTicks(1) // WriteTimeout has a higher priority than Timeout
+        });
+        TestWriteRecordAsync(_client);
+        TestWriteRecordsAsync(_client);
+        TestWritePointAsync(_client);
+        TestWritePointsAsync(_client);
+
     }
 
     [Test]
-    public void TestCheckHttpClientStillOpen()
+    public void TimeoutExceededByToken()
     {
         MockServer
-            .Given(Request.Create().WithPath("/test").UsingGet())
-            .RespondWith(
-                Response.Create()
-                    .WithStatusCode(HttpStatusCode.OK)
-                    .WithBody("Still ok"));
+            .Given(Request.Create().WithPath("/api/v2/write").UsingPost())
+            .RespondWith(Response.Create().WithStatusCode(204).WithDelay(TimeSpan.FromSeconds(2)));
 
-        var httpClient = new HttpClient(new HttpClientHandler());
         _client = new InfluxDBClient(new ClientConfig
         {
             Host = MockServerUrl,
             Token = "my-token",
             Database = "my-database",
-            HttpClient = httpClient
+            QueryTimeout = TimeSpan.FromSeconds(11),
+            Timeout = TimeSpan.FromSeconds(11),
+            WriteTimeout = TimeSpan.FromSeconds(11)
         });
-        _client.Dispose();
+        var cancellationToken = new CancellationTokenSource(TimeSpan.FromTicks(1)).Token;
+        TestWriteRecordAsync(_client, cancellationToken);
+        TestWriteRecordsAsync(_client, cancellationToken);
+        TestWritePointAsync(_client, cancellationToken);
+        TestWritePointsAsync(_client, cancellationToken);
+    }
 
-        var httpResponseMessage = httpClient.Send(new HttpRequestMessage(HttpMethod.Get, "test"));
-        Assert.That(httpResponseMessage.Content.ReadAsStringAsync().Result, Is.EqualTo("Still ok"));
+    private static void TestWriteRecordAsync(InfluxDBClient client, CancellationToken? cancellationToken = null)
+    {
+        Assert.ThrowsAsync<TaskCanceledException>(async () =>
+        {
+            await client.WriteRecordAsync("mem,tag=a field=1", cancellationToken: cancellationToken ?? CancellationToken.None);
+        });
+    }
+
+    private static void TestWriteRecordsAsync(InfluxDBClient client, CancellationToken? cancellationToken = null)
+    {
+        Assert.ThrowsAsync<TaskCanceledException>(async () =>
+        {
+            await client.WriteRecordsAsync(
+                records: new[] { "stat,unit=temperature value=24.5", "stat,unit=temperature value=25.5" },
+                cancellationToken: cancellationToken ?? CancellationToken.None
+            );
+        });
+    }
+
+    private static void TestWritePointAsync(InfluxDBClient client, CancellationToken? cancellationToken = null)
+    {
+        Assert.ThrowsAsync<TaskCanceledException>(async () =>
+        {
+            await client.WritePointAsync(
+                PointData.Measurement("h2o").SetTag("location", "europe").SetField("level", 2),
+                cancellationToken: cancellationToken ?? CancellationToken.None
+            );
+        });
+    }
+
+    private static void TestWritePointsAsync(InfluxDBClient client, CancellationToken? cancellationToken = null)
+    {
+        Assert.ThrowsAsync<TaskCanceledException>(async () =>
+        {
+            await client.WritePointsAsync(
+                points: new[]
+                {
+                    PointData.Measurement("h2o").SetTag("location", "europe").SetField("level", 2),
+                    PointData.Measurement("h2o").SetTag("location", "us-west").SetField("level", 4),
+                },
+                cancellationToken: cancellationToken ?? CancellationToken.None
+            );
+        });
     }
 }
diff --git a/Client.Test/Internal/RestClientTest.cs b/Client.Test/Internal/RestClientTest.cs
@@ -279,19 +279,6 @@ public void AllowHttpRedirects()
         Assert.That(_client, Is.Not.Null);
     }
 
-    [Test]
-    public void Timeout()
-    {
-        CreateAndConfigureRestClient(new ClientConfig
-        {
-            Host = MockServerUrl,
-            Timeout = TimeSpan.FromSeconds(45)
-        });
-
-        var httpClient = GetDeclaredField<HttpClient>(_client.GetType(), _client, "_httpClient");
-        Assert.That(httpClient.Timeout, Is.EqualTo(TimeSpan.FromSeconds(45)));
-    }
-
     private void CreateAndConfigureRestClient(ClientConfig config)
     {
         _httpClient = InfluxDBClient.CreateOrGetHttpClient(config);
diff --git a/Client/Config/ClientConfig.cs b/Client/Config/ClientConfig.cs
@@ -19,7 +19,9 @@ namespace InfluxDB3.Client.Config;
 /// <item>- Organization: The organization to be used for operations.</item>
 /// <item>- Database: The database to be used for InfluxDB operations.</item>
 /// <item>- Headers: The set of HTTP headers to be included in requests.</item>
-/// <item>- Timeout: Timeout to wait before the HTTP request times out. Default to '10 seconds'.</item>
+/// <item>- [Deprecated] Timeout: Timeout to wait before the HTTP request times out. Default to '10 seconds'.</item>
+/// <item>- QueryTimeout: The maximum duration to wait for a query to complete before timing out.</item>
+/// <item>- WriteTimeout: The duration to wait before timing out a write operation to the InfluxDB server.</item>
 /// <item>- AllowHttpRedirects: Automatically following HTTP 3xx redirects. Default to 'false'.</item>
 /// <item>- DisableServerCertificateValidation: Disable server SSL certificate validation. Default to 'false'.</item>
 /// <item>- DisableCertificateRevocationListCheck: Disable SSL certificate revocation list (CRL) checking. Default to 'false'.</item>
@@ -165,8 +167,19 @@ public string Host
     /// <summary>
     /// Timeout to wait before the HTTP request times out. Default to '10 seconds'.
     /// </summary>
+    [Obsolete("Please use more informative properties like WriteTimeout or QueryTimeout")]
     public TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(10);
 
+    /// <summary>
+    /// The maximum duration to wait for a query to complete before timing out.
+    /// </summary>
+    public TimeSpan? QueryTimeout { get; set; }
+
+    /// <summary>
+    /// The duration to wait before timing out a write operation to the InfluxDB server.
+    /// </summary>
+    public TimeSpan? WriteTimeout { get; set; }
+
     /// <summary>
     /// Automatically following HTTP 3xx redirects. Default to 'false'.
     /// </summary>
diff --git a/Client/InfluxDBClient.cs b/Client/InfluxDBClient.cs
diff --git a/Client/Internal/FlightSqlClient.cs b/Client/Internal/FlightSqlClient.cs
