diff --git a/ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs b/ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs index f7ab9c837..c4b6b6442 100644 --- a/ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs +++ b/ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs @@ -73,13 +73,23 @@ public void Validate_EmptyDatabase_ThrowsArgumentException() } [Test] - public void Validate_NonPositiveReadTimeout_ThrowsArgumentOutOfRangeException() + public void Validate_NegativeReadTimeout_ThrowsArgumentOutOfRangeException() { - var options = new ClickHouseTcpClientOptions { ReadTimeout = TimeSpan.Zero }; + var options = new ClickHouseTcpClientOptions { ReadTimeout = TimeSpan.FromSeconds(-1) }; Assert.Throws(() => options.Validate()); } + [Test] + public void Validate_ZeroReadTimeout_IsAccepted() + { + // The opt-out, as it is for the pool's limits: a caller reading a stream that is legitimately silent for + // arbitrarily long has to be able to say so. + var options = new ClickHouseTcpClientOptions { ReadTimeout = TimeSpan.Zero }; + + Assert.DoesNotThrow(() => options.Validate()); + } + [TestCase(0)] [TestCase(-1)] [TestCase(65536)] diff --git a/ClickHouse.Driver.Tcp.Tests/Integration/ClickHouseTcpCancellationIntegrationTests.cs b/ClickHouse.Driver.Tcp.Tests/Integration/ClickHouseTcpCancellationIntegrationTests.cs new file mode 100644 index 000000000..9ce6bc6c7 --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Integration/ClickHouseTcpCancellationIntegrationTests.cs @@ -0,0 +1,125 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Tcp.Client; +using ClickHouse.Driver.Tcp.Format; +using ClickHouse.Driver.Tcp.Tests.Utilities; +using ClickHouse.Driver.Tcp.Types; + +namespace ClickHouse.Driver.Tcp.Tests.Integration; + +// Giving up on a result has to reach the server, not just the client: without the Cancel packet the server keeps +// running the query and writing into a socket nobody reads. Error code 735, QUERY_WAS_CANCELLED_BY_CLIENT, is +// raised only where the server reads that packet, so a query logged with it ended because the client asked rather +// than because the connection went away. +[TestFixture] +[Category("Integration")] +public class ClickHouseTcpCancellationIntegrationTests +{ + private const int QueryWasCancelledByClient = 735; + + private static readonly CancellationToken None = CancellationToken.None; + + [Test] + public async Task StreamAsync_CancelledMidResult_StopsTheQueryOnTheServer() + { + await using ClickHouseTcpClient client = TcpServerFixture.CreateClient(); + string queryId = Guid.NewGuid().ToString(); + + using var cts = new CancellationTokenSource(); + Assert.CatchAsync(async () => + { + await foreach (Block block in Unbounded(client, queryId, cts.Token)) + { + _ = block; + await cts.CancelAsync(); + } + }); + + Assert.That(await CancelledByClientAsync(client, queryId), Is.True); + } + + [Test] + public async Task StreamAsync_EnumerationAbandonedEarly_StopsTheQueryOnTheServer() + { + await using ClickHouseTcpClient client = TcpServerFixture.CreateClient(); + string queryId = Guid.NewGuid().ToString(); + + await foreach (Block block in Unbounded(client, queryId, None)) + { + _ = block; + break; + } + + Assert.That(await CancelledByClientAsync(client, queryId), Is.True); + } + + [Test] + public async Task StreamAsync_CancelledMidResult_ReturnsThePoolSlotForTheNextOperation() + { + // MaxPoolSize 1, so a lost permit or a connection put back broken leaves nothing for the next query to + // run on: the assertion below could not pass by chance on a fresh connection. + ClickHouseTcpClientOptions options = TcpServerFixture.Options() with { MaxPoolSize = 1 }; + await using var client = new ClickHouseTcpClient(options); + + using var cts = new CancellationTokenSource(); + Assert.CatchAsync(async () => + { + await foreach (Block block in Unbounded(client, Guid.NewGuid().ToString(), cts.Token)) + { + _ = block; + await cts.CancelAsync(); + } + }); + + var answer = 0; + await foreach (Block block in client.StreamAsync("SELECT 42", cancellationToken: None)) + { + answer = ((IColumn)block[0]).Values[0]; + } + + Assert.That(answer, Is.EqualTo(42)); + } + + [Test] + public async Task StreamAsync_QueryLongerThanReadTimeout_SurvivesBecauseTheDeadlineMeasuresSilence() + { + // Roughly two seconds of work delivered in 200ms blocks, under a one-second deadline. The deadline bounds + // the gap between packets, not the response, so only the blocks have to fit inside it. The rows are + // selected rather than counted: an aggregate the planner can answer without them prunes the sleep away + // and the query returns instantly, proving nothing. + ClickHouseTcpClientOptions options = TcpServerFixture.Options() with { ReadTimeout = TimeSpan.FromSeconds(1) }; + await using var client = new ClickHouseTcpClient(options); + + var rows = 0; + await foreach (Block block in client.StreamAsync( + "SELECT number, sleepEachRow(0.02) FROM system.numbers LIMIT 100 SETTINGS max_block_size = 10", + cancellationToken: None)) + { + rows += block.RowCount; + } + + Assert.That(rows, Is.EqualTo(100)); + } + + // A result with no end, so the server is certainly still producing it when the client stops reading. Slow and + // small on purpose: a result that saturates the socket blocks the server in a write, where it reads nothing. + private static IAsyncEnumerable Unbounded(ClickHouseTcpClient client, string queryId, CancellationToken cancellationToken) + => client.StreamAsync( + "SELECT number, sleepEachRow(0.02) FROM system.numbers SETTINGS max_block_size = 10", + new ClickHouseTcpQueryOptions { QueryId = queryId }, + cancellationToken); + + // Waits for the query's own log record, which the server writes once the query has actually stopped, and + // reports whether it ended because the client cancelled it. + private static async Task CancelledByClientAsync(ClickHouseTcpClient client, string queryId) + { + object code = await QueryLog.ScalarAsync( + client, + $"SELECT exception_code FROM system.query_log WHERE query_id = '{queryId}' AND type != 'QueryStart' ORDER BY event_time_microseconds DESC LIMIT 1"); + + return Convert.ToInt32(code, CultureInfo.InvariantCulture) == QueryWasCancelledByClient; + } +} diff --git a/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionInsertTests.cs b/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionInsertTests.cs index 6a4863a23..9dc2a565c 100644 --- a/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionInsertTests.cs +++ b/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionInsertTests.cs @@ -192,6 +192,101 @@ public async Task InsertAsync_TokenAlreadyCancelled_ThrowsWithoutClaimingConnect }); } + [Test] + public async Task InsertAsync_CancelledWhileAwaitingTheSchemaBlock_SendsCancelBeforeTerminating() + { + // The Query and the end-of-input block are written, then the read for the schema blocks. The server has + // consumed everything the client sent, so a Cancel lands on a packet boundary. + var transport = new ScriptedDuplexStream(await ServerHelloBytesAsync(54476), blockWhenExhausted: true); + using var connection = new ClickHouseTcpConnection(transport, socket: null); + await connection.HandshakeAsync(Handshake, None); + + using var cts = new CancellationTokenSource(); + Task insert = connection.InsertAsync("INSERT INTO t VALUES", Columns(UInt64Column(1)), cancellationToken: cts.Token).AsTask(); + await cts.CancelAsync(); + + Assert.CatchAsync(async () => await insert); + Assert.Multiple(() => + { + Assert.That(transport.Written[^1], Is.EqualTo((byte)ClientPacketType.Cancel)); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + }); + } + + [Test] + public async Task InsertAsync_ServerGoesSilentAwaitingTheSchemaBlock_ThrowsTimeoutAndSendsCancel() + { + // The schema read has to carry the deadline's token. With the caller's it is unbounded, and an insert + // against a server that stopped answering waits for TCP rather than for ReadTimeout. + var transport = new ScriptedDuplexStream(await ServerHelloBytesAsync(54476), blockWhenExhausted: true); + using var connection = new ClickHouseTcpConnection(transport, socket: null, readTimeout: TimeSpan.FromMilliseconds(200)); + await connection.HandshakeAsync(Handshake, None); + + var thrown = Assert.CatchAsync( + async () => await connection.InsertAsync("INSERT INTO t VALUES", Columns(UInt64Column(1)), cancellationToken: None)); + + Assert.Multiple(() => + { + Assert.That(thrown.Message, Does.Contain("ReadTimeout")); + Assert.That(transport.Written[^1], Is.EqualTo((byte)ClientPacketType.Cancel)); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + }); + } + + [Test] + public async Task InsertAsync_ServerGoesSilentDrainingTheAcknowledgement_ThrowsTimeout() + { + // The other read the insert makes: the drain to end-of-stream, after every row has gone out. A regression + // that bounds only the schema read leaves this one waiting for TCP. + byte[] script = Concat( + await ServerHelloBytesAsync(54476), + await SchemaBlockAsync(("x", "UInt64"))); + var transport = new ScriptedDuplexStream(script, blockWhenExhausted: true); + using var connection = new ClickHouseTcpConnection(transport, socket: null, readTimeout: TimeSpan.FromMilliseconds(200)); + await connection.HandshakeAsync(Handshake, None); + + var thrown = Assert.CatchAsync( + async () => await connection.InsertAsync("INSERT INTO t VALUES", Columns(UInt64Column(1)), cancellationToken: None)); + + Assert.Multiple(() => + { + Assert.That(thrown.Message, Does.Contain("ReadTimeout")); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + }); + } + + [Test] + public async Task InsertAsync_CancelledWhileStreamingRows_LeavesTheTruncatedBlockWithoutAppendingCancel() + { + // Cancelling from the column factory takes the insert down inside the row stream, where a block is + // part-written. A Cancel appended there is read as more block bytes rather than as a packet, so none is + // sent and the connection is simply closed. + byte[] script = Concat( + await ServerHelloBytesAsync(54476), + await SchemaBlockAsync(("x", "UInt64")), + EndOfStreamPacket()); + var transport = new ScriptedDuplexStream(script); + using var connection = new ClickHouseTcpConnection(transport, socket: null); + await connection.HandshakeAsync(Handshake, None); + + using var cts = new CancellationTokenSource(); + Assert.CatchAsync(async () => await connection.InsertAsync( + "INSERT INTO t VALUES", + rowCount: 1, + buildColumns: _ => + { + cts.Cancel(); + return new StubInsertColumnSource(UInt64Column(1)); + }, + cancellationToken: cts.Token)); + + Assert.Multiple(() => + { + Assert.That(transport.Written[^1], Is.Not.EqualTo((byte)ClientPacketType.Cancel)); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + }); + } + [Test] public async Task InsertAsync_ColumnCountDisagreesWithSchema_ThrowsArgumentButStaysReady() { diff --git a/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionQueryTests.cs b/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionQueryTests.cs index 29498bf11..7ad45f740 100644 --- a/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionQueryTests.cs +++ b/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionQueryTests.cs @@ -3,6 +3,7 @@ using System.IO; using System.Threading; using System.Threading.Tasks; +using ClickHouse.Driver.Compression; using ClickHouse.Driver.Tcp.Format; using ClickHouse.Driver.Tcp.Protocol; using ClickHouse.Driver.Tcp.Tests.Utilities; @@ -155,6 +156,211 @@ await DataPacketAsync(new ulong[] { 2 }), Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); } + [Test] + public async Task QueryAsync_EnumerationAbandonedBeforeEndOfStream_SendsCancelBeforeTerminating() + { + byte[] script = Concat( + await ServerHelloBytesAsync(54476), + await DataPacketAsync(new ulong[] { 1 }), + await DataPacketAsync(new ulong[] { 2 }), + EndOfStreamPacket()); + var transport = new ScriptedDuplexStream(script); + using var connection = new ClickHouseTcpConnection(transport, socket: null); + await connection.HandshakeAsync(Handshake, None); + + var writtenBeforeAbandoning = 0; + await foreach (Block block in connection.QueryAsync("SELECT 1", cancellationToken: None)) + { + _ = block; + writtenBeforeAbandoning = transport.Written.Length; + break; + } + + // The server is still producing a result nobody will read, so it is told to stop before the socket goes. + // Cancel carries no body, so exactly one byte follows the request. + Assert.Multiple(() => + { + Assert.That(transport.Written, Has.Length.EqualTo(writtenBeforeAbandoning + 1)); + AssertCancelSent(transport); + }); + } + + [Test] + public async Task QueryAsync_CancelledWhileReadingResponse_SendsCancelBeforeTerminating() + { + // The Query is written, then the read for the first packet blocks (as against a server still working). + var transport = new ScriptedDuplexStream(await ServerHelloBytesAsync(54476), blockWhenExhausted: true); + using var connection = new ClickHouseTcpConnection(transport, socket: null); + await connection.HandshakeAsync(Handshake, None); + + using var cts = new CancellationTokenSource(); + Task drain = DrainAsync(connection, cts.Token); + await cts.CancelAsync(); + + Assert.CatchAsync(async () => await drain); + Assert.Multiple(() => + { + AssertCancelSent(transport); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + }); + } + + [Test] + public async Task QueryAsync_ServerGoesSilentPastReadTimeout_ThrowsTimeoutAndSendsCancel() + { + var transport = new ScriptedDuplexStream(await ServerHelloBytesAsync(54476), blockWhenExhausted: true); + using var connection = new ClickHouseTcpConnection(transport, socket: null, readTimeout: TimeSpan.FromMilliseconds(200)); + await connection.HandshakeAsync(Handshake, None); + + var thrown = Assert.CatchAsync(async () => await DrainAsync(connection)); + + Assert.Multiple(() => + { + Assert.That(thrown.Message, Does.Contain("ReadTimeout")); + AssertCancelSent(transport); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + }); + } + + [Test] + public async Task QueryAsync_CompressedAndTheServerStopsInsideABlock_ThrowsTimeoutNamingReadTimeout() + { + // A compressed block body is decoded through the frame reader, whose own buffer carries no deadline: it + // is served from bytes the transport already handed over. A stall part-way through a frame therefore has + // to be caught by the transport buffer underneath it, and the timeout has to come back out through the + // decoder rather than be mistaken for a malformed frame. + byte[] script = Concat( + await ServerHelloBytesAsync(54476), + await BytesAsync(w => + { + w.WriteVarUInt((ulong)ServerPacketType.Data); + w.WriteString(string.Empty); // The envelope is never framed; the frames begin after the table name. + })); + var transport = new ScriptedDuplexStream(script, blockWhenExhausted: true); + using var connection = new ClickHouseTcpConnection(transport, socket: null, Lz4Compressor.Default, readTimeout: TimeSpan.FromMilliseconds(200)); + await connection.HandshakeAsync(Handshake, None); + + var thrown = Assert.CatchAsync(async () => await DrainAsync(connection)); + + Assert.Multiple(() => + { + Assert.That(thrown.Message, Does.Contain("ReadTimeout")); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + }); + } + + [Test] + public async Task QueryAsync_CallerCancelsWhileTheDeadlineIsArmed_ReportsCancellationNotTimeout() + { + // Both are live and both cancel the same linked token, so the two have to be told apart by which one + // fired. A generous deadline, so only the caller can have caused this. + var transport = new ScriptedDuplexStream(await ServerHelloBytesAsync(54476), blockWhenExhausted: true); + using var connection = new ClickHouseTcpConnection(transport, socket: null, readTimeout: TimeSpan.FromSeconds(30)); + await connection.HandshakeAsync(Handshake, None); + + using var cts = new CancellationTokenSource(); + Task drain = DrainAsync(connection, cts.Token); + await cts.CancelAsync(); + + Assert.CatchAsync(async () => await drain); + } + + [Test] + public async Task HandshakeAsync_SlowerThanReadTimeout_CompletesBecauseTheDeadlineCoversResponsesOnly() + { + // Connecting is bounded by DialTimeout, which covers the handshake. Arming ReadTimeout here as well + // would stack two deadlines on the one exchange and fail a connect the caller's own bound allowed. + var transport = new ScriptedDuplexStream(await ServerHelloBytesAsync(54476), maxChunk: 2, readDelay: TimeSpan.FromMilliseconds(20)); + using var connection = new ClickHouseTcpConnection(transport, socket: null, readTimeout: TimeSpan.FromMilliseconds(50)); + + await connection.HandshakeAsync(Handshake, None); + + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Ready)); + } + + [Test] + public async Task QueryAsync_SecondQueryOnTheSameConnection_RearmsTheDeadlineForItsOwnReads() + { + // A pooled connection carries many operations, so the deadline is opened and closed repeatedly. A source + // left over from the first query would either be disposed under the second or already cancelled by it. + byte[] script = Concat( + await ServerHelloBytesAsync(54476), + await DataPacketAsync(new ulong[] { 1 }), + EndOfStreamPacket(), + await DataPacketAsync(new ulong[] { 2 }), + EndOfStreamPacket()); + using var connection = new ClickHouseTcpConnection(new ScriptedDuplexStream(script, maxChunk: 1), socket: null, readTimeout: TimeSpan.FromMilliseconds(200)); + await connection.HandshakeAsync(Handshake, None); + + List first = await MaterializeAsync(connection); + await Task.Delay(TimeSpan.FromMilliseconds(300)); + List second = await MaterializeAsync(connection); + + Assert.Multiple(() => + { + CollectionAssert.AreEqual(new ulong[] { 1 }, first[0]); + CollectionAssert.AreEqual(new ulong[] { 2 }, second[0]); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Ready)); + }); + } + + [Test] + public async Task QueryAsync_ResponseSlowerOverallThanReadTimeout_CompletesBecauseTheDeadlineMeasuresSilence() + { + // Two bytes every 20ms. The query's own share of that is about two dozen reads, so it runs well past the + // 250ms deadline in total while no single gap comes near it, and the deadline must be rearmed by each + // arriving chunk rather than left to run for the response's duration. Only the reads after the handshake + // count towards this: the handshake is deliberately outside the deadline. + byte[] script = Concat( + await ServerHelloBytesAsync(54476), + await DataPacketAsync(new ulong[] { 1, 2, 3 }), + EndOfStreamPacket()); + var transport = new ScriptedDuplexStream(script, maxChunk: 2, readDelay: TimeSpan.FromMilliseconds(20)); + using var connection = new ClickHouseTcpConnection(transport, socket: null, readTimeout: TimeSpan.FromMilliseconds(250)); + await connection.HandshakeAsync(Handshake, None); + + var rows = await MaterializeAsync(connection); + + Assert.Multiple(() => + { + Assert.That(rows, Has.Count.EqualTo(1)); + CollectionAssert.AreEqual(new ulong[] { 1, 2, 3 }, rows[0]); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Ready)); + }); + } + + [Test] + public async Task QueryAsync_ConsumerHoldsABlockPastReadTimeout_IsNotTreatedAsASilentServer() + { + // The deadline must not run while the iterator is parked at its yield, or a slow consumer reads as a + // server that stopped answering. + // + // One byte per read, or the handshake's first fill swallows the whole script — the buffer is 16 KiB and + // this is under a hundred bytes — and the query then reads from memory, arming nothing and proving + // nothing. + byte[] script = Concat( + await ServerHelloBytesAsync(54476), + await DataPacketAsync(new ulong[] { 1 }), + await DataPacketAsync(new ulong[] { 2 }), + EndOfStreamPacket()); + using var connection = new ClickHouseTcpConnection(new ScriptedDuplexStream(script, maxChunk: 1), socket: null, readTimeout: TimeSpan.FromMilliseconds(150)); + await connection.HandshakeAsync(Handshake, None); + + var blocks = 0; + await foreach (Block block in connection.QueryAsync("SELECT 1", cancellationToken: None)) + { + _ = block; + blocks++; + await Task.Delay(TimeSpan.FromMilliseconds(300)); + } + + Assert.Multiple(() => + { + Assert.That(blocks, Is.EqualTo(2)); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Ready)); + }); + } + [Test] public async Task QueryAsync_TokenAlreadyCancelled_ThrowsWithoutClaimingConnection() { @@ -422,6 +628,10 @@ public async Task QueryAsync_AfterTerminate_ThrowsObjectDisposed() Assert.ThrowsAsync(async () => await DrainAsync(connection)); } + // Cancel is a bare uvarint with no body, so a sent one is the single trailing byte of the client's writes. + private static void AssertCancelSent(ScriptedDuplexStream transport) + => Assert.That(transport.Written[^1], Is.EqualTo((byte)ClientPacketType.Cancel), "the Cancel packet should be the last thing written"); + private static async Task ConnectedAsync(byte[] script) { var connection = new ClickHouseTcpConnection(new ScriptedDuplexStream(script), socket: null); @@ -443,9 +653,9 @@ private static async Task> MaterializeAsync(ClickHouseTcpConnectio } // Enumerates the response without reading block contents (for tests that assert an exception or state). - private static async Task DrainAsync(ClickHouseTcpConnection connection) + private static async Task DrainAsync(ClickHouseTcpConnection connection, CancellationToken cancellationToken = default) { - await foreach (Block block in connection.QueryAsync("SELECT 1", cancellationToken: None)) + await foreach (Block block in connection.QueryAsync("SELECT 1", cancellationToken: cancellationToken)) { _ = block; } diff --git a/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionTests.cs b/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionTests.cs index 544889ce4..4314a1110 100644 --- a/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionTests.cs +++ b/ClickHouse.Driver.Tcp.Tests/Protocol/ClickHouseTcpConnectionTests.cs @@ -173,6 +173,25 @@ public async Task PingAsync_CancelledWhileAwaitingPong_TerminatesConnection() Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); } + [Test] + public async Task PingAsync_ServerNeverAnswersWithinReadTimeout_ThrowsTimeoutAndTerminates() + { + // The reply read has to carry the deadline's token, not the caller's; with the caller's it is unbounded + // and a ping to a wedged server hangs until TCP gives up. + byte[] script = await ServerHelloBytesAsync(54476); + using var connection = new ClickHouseTcpConnection( + new ScriptedDuplexStream(script, blockWhenExhausted: true), socket: null, readTimeout: TimeSpan.FromMilliseconds(200)); + await connection.HandshakeAsync(Handshake, None); + + var thrown = Assert.CatchAsync(async () => await connection.PingAsync(None)); + + Assert.Multiple(() => + { + Assert.That(thrown.Message, Does.Contain("ReadTimeout")); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + }); + } + [Test] public async Task PingAsync_AfterTerminate_ThrowsObjectDisposed() { diff --git a/ClickHouse.Driver.Tcp.Tests/Utilities/QueryLog.cs b/ClickHouse.Driver.Tcp.Tests/Utilities/QueryLog.cs new file mode 100644 index 000000000..d0f1c772c --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Utilities/QueryLog.cs @@ -0,0 +1,81 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Tcp.Client; + +namespace ClickHouse.Driver.Tcp.Tests.Utilities; + +/// +/// Reads system.query_log from tests without racing the server. The native-transport counterpart of +/// ClickHouse.Driver.Tests.QueryLog, which takes the HTTP client and cannot be shared. +/// +/// +/// SYSTEM FLUSH LOGS only flushes what the server has already queued, and a query's record is queued +/// independently of its response reaching the client — so a flush issued right after the query can miss it. +/// Every method here retries the flush-and-read, so assert on what they return rather than on a raw read taken +/// after a single flush. +/// +/// The retry budget is larger than the HTTP helper's, because the queries this one waits on are cancelled ones: +/// the record is written when the server actually stops, which is after the client has already given up on it. +/// +/// +/// The flush names query_log rather than flushing everything: the framework suites share one server. +/// +/// +internal static class QueryLog +{ + /// Number of flush-and-read attempts before giving up. + internal const int MaxAttempts = 40; + + private static readonly TimeSpan RetryDelay = TimeSpan.FromMilliseconds(100); + + /// + /// Returns the first column of the first row of , retrying while it matches no rows, + /// and failing the test with a distinct message if it never does. + /// + /// + /// A missing row and a NULL value are indistinguishable here, so select an expression that is never NULL for + /// a row that exists. + /// + /// Client to run the flush and the lookup on. + /// Lookup returning one row with the value under test in its first column. + /// The value read once the row became visible. + internal static async Task ScalarAsync(ClickHouseTcpClient client, string sql) + { + for (var attempt = 1; attempt <= MaxAttempts; attempt++) + { + await client.ExecuteAsync("SYSTEM FLUSH LOGS query_log", cancellationToken: CancellationToken.None); + + object value = await ReadFirstAsync(client, sql); + if (value is not null) + { + return value; + } + + if (attempt < MaxAttempts) + { + await Task.Delay(RetryDelay); + } + } + + string message = $"No system.query_log row appeared after {MaxAttempts} flush attempts, so the value under test could not be determined. Query: {sql}"; + Assert.Fail(message); + + // Only reached inside Assert.Multiple, where Assert.Fail records the failure and continues: throw rather + // than hand the caller a null it would misread as a value. + throw new InvalidOperationException(message); + } + + // Drains the result rather than returning from inside the enumeration: abandoning it cancels the query and + // drops the connection, so every retry would pay a reconnect. + private static async Task ReadFirstAsync(ClickHouseTcpClient client, string sql) + { + object first = null; + await foreach (object[] row in client.QueryAsync(sql, cancellationToken: CancellationToken.None)) + { + first ??= row[0]; + } + + return first; + } +} diff --git a/ClickHouse.Driver.Tcp.Tests/Utilities/ScriptedDuplexStream.cs b/ClickHouse.Driver.Tcp.Tests/Utilities/ScriptedDuplexStream.cs index 2210830ee..88e7b8223 100644 --- a/ClickHouse.Driver.Tcp.Tests/Utilities/ScriptedDuplexStream.cs +++ b/ClickHouse.Driver.Tcp.Tests/Utilities/ScriptedDuplexStream.cs @@ -16,14 +16,16 @@ internal sealed class ScriptedDuplexStream : Stream private readonly byte[] script; private readonly int maxChunk; private readonly bool blockWhenExhausted; + private readonly TimeSpan readDelay; private readonly MemoryStream sink = new(); private int position; - public ScriptedDuplexStream(byte[] script, int maxChunk = int.MaxValue, bool blockWhenExhausted = false) + public ScriptedDuplexStream(byte[] script, int maxChunk = int.MaxValue, bool blockWhenExhausted = false, TimeSpan readDelay = default) { this.script = script; this.maxChunk = maxChunk < 1 ? 1 : maxChunk; this.blockWhenExhausted = blockWhenExhausted; + this.readDelay = readDelay; } /// The bytes the connection has written (the client → server side of the exchange). @@ -70,6 +72,13 @@ public override async ValueTask ReadAsync(Memory buffer, Cancellation await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false); } + // A server that answers, but slowly. Paired with maxChunk it makes a response whose total time exceeds an + // idle deadline while no single gap in it does. + if (readDelay > TimeSpan.Zero) + { + await Task.Delay(readDelay, cancellationToken).ConfigureAwait(false); + } + return Read(buffer.Span); } diff --git a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs index 31d34ab16..92fc4dd0d 100644 --- a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs +++ b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs @@ -175,10 +175,24 @@ public sealed record ClickHouseTcpClientOptions public TimeSpan DialTimeout { get; init; } = DefaultDialTimeout; /// - /// The idle deadline for reading a response — reset each time a packet arrives — so a long streaming query - /// is not killed for taking a long time overall. Defaults to 300s. Stored but not yet enforced; the - /// idle-deadline read loop lands in a later change. + /// How long the server may stay silent while a response is being read, after which the operation fails with a + /// and the connection is discarded. Defaults to 300s; + /// leaves the caller's own as the + /// only bound. /// + /// + /// + /// This measures silence, not duration. The clock runs only while the client is waiting on the transport and + /// is reset by every byte that arrives, so a query that streams for an hour never trips it, and neither does a + /// consumer that holds a block for longer than the deadline before asking for the next one. What it catches is + /// a server, or a path to one, that has stopped answering — including a connection dropped without a FIN, + /// which no client-side probe can see and which TCP alone takes about fifteen minutes to give up on. + /// + /// + /// It bounds the response only. Opening a connection is bounded separately by , which + /// also covers the handshake, so the two never apply to the same exchange. + /// + /// public TimeSpan ReadTimeout { get; init; } = DefaultReadTimeout; /// @@ -447,7 +461,17 @@ internal void Validate() RequireUsableTimeout(DialTimeout, nameof(DialTimeout)); - RequireUsableTimeout(ReadTimeout, nameof(ReadTimeout)); + // Zero is the opt-out, as it is for the pool's limits: a caller reading a stream that is legitimately + // silent for arbitrarily long has to be able to say so. + if (ReadTimeout < TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException(nameof(ReadTimeout), ReadTimeout, "ReadTimeout must not be negative; use TimeSpan.Zero to disable the deadline."); + } + + if (ReadTimeout.TotalMilliseconds > int.MaxValue) + { + throw new ArgumentOutOfRangeException(nameof(ReadTimeout), ReadTimeout, $"ReadTimeout must not exceed {TimeSpan.FromMilliseconds(int.MaxValue)} (about 24.8 days)."); + } if (MaxSendBufferBytes <= 0) { diff --git a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs index 577229963..0acec0c74 100644 --- a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs +++ b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs @@ -171,7 +171,7 @@ public TimeSpan DialTimeout set => this["DialTimeout"] = value.TotalSeconds; } - /// The idle read deadline, in seconds. Defaults to 300. + /// How long the server may stay silent mid-response, in seconds. Defaults to 300; 0 disables it. public TimeSpan ReadTimeout { get => GetTimeSpanSecondsOrDefault("ReadTimeout", ClickHouseTcpClientOptions.DefaultReadTimeout); diff --git a/ClickHouse.Driver.Tcp/Client/IConnectionFactory.cs b/ClickHouse.Driver.Tcp/Client/IConnectionFactory.cs index 835bcb6e3..c10605f0e 100644 --- a/ClickHouse.Driver.Tcp/Client/IConnectionFactory.cs +++ b/ClickHouse.Driver.Tcp/Client/IConnectionFactory.cs @@ -68,7 +68,7 @@ public async ValueTask CreateAsync(CancellationToken ca try { ClickHouseTcpConnection connection = await ClickHouseTcpConnection.ConnectAsync( - options.Host, options.ResolvedPort, options.ToHandshakeParameters(), tls, linked.Token, options.Compressor).ConfigureAwait(false); + options.Host, options.ResolvedPort, options.ToHandshakeParameters(), tls, linked.Token, options.Compressor, options.ReadTimeout).ConfigureAwait(false); activity?.SetSuccess(); if (logger is not null) diff --git a/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs b/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs index ed01eeed0..73d477e9e 100644 --- a/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs +++ b/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs @@ -58,11 +58,19 @@ internal sealed class ClickHouseTcpConnection : IDisposable, IAsyncDisposable // for timezone-less DateTime/DateTime64 result columns. private const string SessionTimezoneSetting = "session_timezone"; + // How long to spend delivering the Cancel packet before giving up on it. Short and not configurable: the + // connection is closed either way, so all this bounds is how long a cancelling caller waits on a server that + // has stopped reading. + private static readonly TimeSpan CancelSendTimeout = TimeSpan.FromSeconds(2); + private readonly Socket socket; private readonly Stream stream; private readonly ClickHouseBinaryReader reader; private readonly ClickHouseBinaryWriter writer; + // Bounds how long the server may stay silent mid-response. Null when the caller's token is the only bound. + private readonly IdleReadDeadline readDeadline; + // Null means every query on this connection is uncompressed. Compression is per-query on the wire, but the // codec is a client-level option today, so it is fixed for a connection's life; a per-query override would // move this to the operation entry points. @@ -85,12 +93,20 @@ internal sealed class ClickHouseTcpConnection : IDisposable, IAsyncDisposable /// The duplex transport stream (a network stream in production). /// The underlying socket, closed on termination; null when the stream owns teardown. /// Frame codec for this connection's queries, or null to run them uncompressed. - internal ClickHouseTcpConnection(Stream stream, Socket socket, IClickHouseCompressor compressor = null) + /// + /// How long the server may stay silent mid-response before the operation fails. + /// leaves the caller's token as the only bound, which is the default for the scripted-stream seam. + /// + internal ClickHouseTcpConnection(Stream stream, Socket socket, IClickHouseCompressor compressor = null, TimeSpan readTimeout = default) { this.stream = stream; this.socket = socket; this.compressor = compressor; - reader = new ClickHouseBinaryReader(stream); + readDeadline = readTimeout == TimeSpan.Zero ? null : new IdleReadDeadline(readTimeout); + + // The deadline belongs to this buffer alone: it is the one that reads the socket. The frame decoder's + // buffer is served from bytes that already arrived through here, so it can never stall on the network. + reader = new ClickHouseBinaryReader(new ReadBuffer(stream, deadline: readDeadline), ownsBuffer: true); writer = new ClickHouseBinaryWriter(stream); state = TcpConnectionState.Handshaking; } @@ -122,9 +138,8 @@ internal ClickHouseTcpConnection(Stream stream, Socket socket, IClickHouseCompre /// up, which on Linux takes about fifteen minutes. That is inherent to a client-side check, so the pool does not /// rely on this alone: it also refuses a connection that has sat idle past IdleTimeout, which covers the /// common case of an intermediary dropping a connection nobody was using. Neither catches a drop that strikes a - /// connection in active use. The answer to that is an idle read deadline rather than a stricter probe, and - /// that deadline does not exist yet: ReadTimeout is parsed and stored but nothing enforces it, so a - /// caller's own is currently the only bound on such a stall. + /// connection in active use; the answer to that is ReadTimeout, the idle deadline every read of an + /// operation runs under, rather than a stricter probe here. /// /// internal bool IsReusable @@ -217,7 +232,8 @@ public static async ValueTask ConnectAsync( ClientHandshakeParameters handshake, TlsParameters tls, CancellationToken cancellationToken, - IClickHouseCompressor compressor = null) + IClickHouseCompressor compressor = null, + TimeSpan readTimeout = default) { ArgumentNullException.ThrowIfNull(host); ArgumentNullException.ThrowIfNull(handshake); @@ -261,8 +277,9 @@ public static async ValueTask ConnectAsync( } // HandshakeAsync terminates the connection (closing this socket) on any failure, so a throw here needs - // no extra cleanup. - var connection = new ClickHouseTcpConnection(transport, socket, compressor); + // no extra cleanup. The handshake itself runs under the caller's connect deadline rather than + // readTimeout, so the two never stack on the one exchange. + var connection = new ClickHouseTcpConnection(transport, socket, compressor, readTimeout); await connection.HandshakeAsync(handshake, cancellationToken).ConfigureAwait(false); return connection; } @@ -278,54 +295,63 @@ public static async ValueTask ConnectAsync( /// The server replied with an Exception. /// The server replied with something other than Pong or Exception. /// The connection failed while the ping was in flight. + /// The server stayed silent for longer than the connection's ReadTimeout. public async ValueTask PingAsync(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); BeginOperation(); - ServerPacketType reply; + CancellationToken io = BeginRead(cancellationToken); try { - writer.WriteClientPacketType(ClientPacketType.Ping); - await writer.FlushAsync(cancellationToken).ConfigureAwait(false); + ServerPacketType reply; + try + { + writer.WriteClientPacketType(ClientPacketType.Ping); + await writer.FlushAsync(cancellationToken).ConfigureAwait(false); - // A Ping is only ever sent on an idle connection, never mid-query, so no Progress or other - // interleaved packet can precede the reply — unlike a query response, which the read loop drains. - // A single read therefore suffices; anything but Pong or a (complete) Exception is a violation. - reply = await reader.ReadServerPacketTypeAsync(cancellationToken).ConfigureAwait(false); - } - catch - { - // The failed/cancelled I/O has unwound, but the stream position is unknown; discard the connection. - Terminate(); - throw; - } + // A Ping is only ever sent on an idle connection, never mid-query, so no Progress or other + // interleaved packet can precede the reply — unlike a query response, which the read loop drains. + // A single read therefore suffices; anything but Pong or a (complete) Exception is a violation. + reply = await reader.ReadServerPacketTypeAsync(io).ConfigureAwait(false); + } + catch + { + // The failed/cancelled I/O has unwound, but the stream position is unknown; discard the connection. + Terminate(); + throw; + } - switch (reply) - { - case ServerPacketType.Pong: - state = TcpConnectionState.Ready; - return; + switch (reply) + { + case ServerPacketType.Pong: + state = TcpConnectionState.Ready; + return; - case ServerPacketType.Exception: - ClickHouseTcpServerException exception; - try - { - exception = await ClickHouseTcpServerException.ReadAsync(reader, cancellationToken).ConfigureAwait(false); - } - catch - { - Terminate(); - throw; - } + case ServerPacketType.Exception: + ClickHouseTcpServerException exception; + try + { + exception = await ClickHouseTcpServerException.ReadAsync(reader, io).ConfigureAwait(false); + } + catch + { + Terminate(); + throw; + } - Terminate(); - throw exception; + Terminate(); + throw exception; - default: - Terminate(); - throw new ClickHouseTcpProtocolException( - $"Unexpected packet type {reply} ({(ulong)reply}) in response to Ping; expected Pong or Exception."); + default: + Terminate(); + throw new ClickHouseTcpProtocolException( + $"Unexpected packet type {reply} ({(ulong)reply}) in response to Ping; expected Pong or Exception."); + } + } + finally + { + EndRead(); } } @@ -357,6 +383,11 @@ public async ValueTask PingAsync(CancellationToken cancellationToken) /// } /// /// + /// + /// Dispose the enumerator. Everything this method owns is released from one finally, so a consumer that stops + /// advancing without disposing keeps the connection, the last block's buffers, and the read deadline's + /// registration on the caller's token for as long as the caller's token source lives. + /// /// /// The SQL text. /// Per-query settings as textual values, or null for none. @@ -371,6 +402,7 @@ public async ValueTask PingAsync(CancellationToken cancellationToken) /// The server reported an error while executing the query. /// The server sent an unexpected packet. /// The connection failed while the response was being read. + /// The server stayed silent for longer than the connection's ReadTimeout. /// was cancelled. internal async IAsyncEnumerable QueryAsync( string sql, @@ -391,7 +423,9 @@ internal async IAsyncEnumerable QueryAsync( NegotiatedProtocol negotiated = server.Negotiated; ClickHouseTcpServerException pending = null; Block current = null; - bool completed = false; + bool responseCompleted = false; + bool reusable = false; + bool flushedWholePackets = false; // Encode the Query packet into the write buffer before any of it reaches the socket. A failure here is a // client-side error (e.g. parameters on a protocol revision that predates them): nothing has been sent, @@ -407,6 +441,7 @@ internal async IAsyncEnumerable QueryAsync( throw; } + CancellationToken io = BeginRead(cancellationToken); try { // The end-of-input marker is written here rather than above, because framing it is not buffer-only @@ -416,6 +451,10 @@ internal async IAsyncEnumerable QueryAsync( await WriteEndOfInputBlockAsync(cancellationToken).ConfigureAwait(false); await writer.FlushAsync(cancellationToken).ConfigureAwait(false); + // The whole request is on the wire and the client writes nothing more, so from here every exit + // short of end-of-stream leaves a query the server is still running and can still be cancelled. + flushedWholePackets = true; + while (true) { // Resuming here means the consumer has advanced past the previously yielded block, so its @@ -426,23 +465,25 @@ internal async IAsyncEnumerable QueryAsync( current = null; } - ServerPacketType packet = await reader.ReadServerPacketTypeAsync(cancellationToken).ConfigureAwait(false); + ServerPacketType packet = await reader.ReadServerPacketTypeAsync(io).ConfigureAwait(false); if (packet == ServerPacketType.EndOfStream) { - completed = true; + responseCompleted = true; + reusable = true; break; } if (packet == ServerPacketType.Exception) { - pending = await ClickHouseTcpServerException.ReadAsync(reader, cancellationToken).ConfigureAwait(false); + pending = await ClickHouseTcpServerException.ReadAsync(reader, io).ConfigureAwait(false); + responseCompleted = true; break; } if (packet == ServerPacketType.Data) { - Block block = await ReadBlockAsync(ServerPacketType.Data, negotiated, readContext, cancellationToken).ConfigureAwait(false); + Block block = await ReadBlockAsync(ServerPacketType.Data, negotiated, readContext, io).ConfigureAwait(false); if (block.RowCount != 0) { // Held as the current block so it is released when the consumer advances or stops. @@ -458,21 +499,31 @@ internal async IAsyncEnumerable QueryAsync( { // Everything else is interleaved metadata: consumed to stay stream-aligned, surfaced to the // callbacks when set. An unexpected packet throws from here. - await ConsumeMetadataAsync(packet, negotiated, readContext, telemetry, callbacks, cancellationToken).ConfigureAwait(false); + await ConsumeMetadataAsync(packet, negotiated, readContext, telemetry, callbacks, io).ConfigureAwait(false); } } } finally { + // First, so that nothing below can throw past it and leave the deadline holding a registration on the + // caller's token. Nothing below reads from the transport, so none of it needs the deadline. + EndRead(); + // Release the last yielded block (still current) on end-of-stream, early disposal, or error. current?.Dispose(); - if (completed) + if (reusable) { state = TcpConnectionState.Ready; } else { + // A response that has not reached a terminal packet may still be running on the server. + if (!responseCompleted) + { + await TrySendCancelAsync(flushedWholePackets).ConfigureAwait(false); + } + Terminate(); } } @@ -536,6 +587,7 @@ internal async IAsyncEnumerable QueryAsync( /// The server reported an error while executing the insert. /// The connection failed while the blocks were being sent or the response read. /// The server sent an unexpected packet, or no schema block. + /// The server stayed silent for longer than the connection's ReadTimeout. /// was cancelled. internal ValueTask InsertAsync( string sql, @@ -614,8 +666,11 @@ private async ValueTask InsertCoreAsync( Exception buildFailure = null; IReadOnlyList values = null; IInsertColumnSource source = null; - bool completed = false; + bool responseCompleted = false; + bool reusable = false; + bool flushedWholePackets = false; string mismatchError = null; + CancellationToken io = BeginRead(cancellationToken); try { // The empty end-of-input block must follow the Query: the server waits for it before sending the @@ -623,11 +678,13 @@ private async ValueTask InsertCoreAsync( Query.Write(writer, negotiated, clientMetadata, queryId, sql, settings, parameters, compressor is not null); await WriteEndOfInputBlockAsync(cancellationToken).ConfigureAwait(false); await writer.FlushAsync(cancellationToken).ConfigureAwait(false); + flushedWholePackets = true; // Drain metadata until the schema block (the first Data packet) or a terminal packet. - (Block schema, ClickHouseTcpServerException error) = await ReadToNextDataBlockAsync(negotiated, readContext, telemetry, callbacks, cancellationToken).ConfigureAwait(false); + (Block schema, ClickHouseTcpServerException error) = await ReadToNextDataBlockAsync(negotiated, readContext, telemetry, callbacks, io).ConfigureAwait(false); if (schema is null) { + responseCompleted = true; if (error is null) { // Clean end-of-stream with no schema: the server never opened the row-stream phase (e.g. @@ -637,7 +694,7 @@ private async ValueTask InsertCoreAsync( } // The Exception packet does not say whether the server accepted the query and returned to its - // request loop, so leave completed false and retire the connection in the finally below. + // request loop, so leave reusable false and retire the connection in the finally below. pending = error; } else @@ -675,29 +732,46 @@ private async ValueTask InsertCoreAsync( } } - // Always run, even after a factory failure: the row stream has to be closed for the server to - // finish the insert. A gather failure is deferred the same way, so the stream still closes cleanly. + // A row stream flushes each block as it is built, so an exit part-way through leaves a truncated + // Data packet on the wire. A Cancel appended to that is read as more block bytes, not as a + // packet, so the row phase is the one stretch where there is nothing useful to send. + // + // The call always runs, even after a factory failure: the row stream has to be closed for the + // server to finish the insert. A gather failure is deferred the same way, so it still closes + // cleanly and this stays a whole-packet boundary. + flushedWholePackets = false; Exception gatherFailure = await StreamInsertRowsAsync(plan, source, rowCount, maxRowsPerBlock, maxSendBufferBytes, negotiated, cancellationToken).ConfigureAwait(false); + flushedWholePackets = true; buildFailure ??= gatherFailure; // A clean acknowledgement leaves the connection reusable. A server Exception is parked for the // caller but retires the connection, because its packet does not prove the server will accept // another request. - pending = await DrainToEndOfStreamAsync(negotiated, readContext, telemetry, callbacks, cancellationToken).ConfigureAwait(false); - completed = pending is null; + pending = await DrainToEndOfStreamAsync(negotiated, readContext, telemetry, callbacks, io).ConfigureAwait(false); + responseCompleted = true; + reusable = pending is null; } } finally { + // First, so that nothing below can throw past it and leave the deadline holding a registration on the + // caller's token. Nothing below reads from the transport, so none of it needs the deadline. + EndRead(); + // Only the factory's source is ours to release; a caller's own columns outlive the insert. source?.Dispose(); - if (completed) + if (reusable) { state = TcpConnectionState.Ready; } else { + if (!responseCompleted) + { + await TrySendCancelAsync(flushedWholePackets).ConfigureAwait(false); + } + Terminate(); } } @@ -1384,4 +1458,67 @@ private void BeginOperation() $"The connection is busy ({state}); a single connection carries one in-flight operation at a time."); } } + + /// + /// Opens the idle read deadline over an operation's token, and returns the token the operation must pass to + /// every read it makes. Pair with in a finally. + /// + /// + /// Writes keep the caller's token, deliberately. A deadline that elapses just as a read completes cannot be + /// recalled — disarming does not stop a timer callback already running — so giving this token to a write + /// would let it fail as cancelled for a token the caller never cancelled. + /// + /// The caller's token for this operation. + /// The token to use for the operation's reads. + private CancellationToken BeginRead(CancellationToken cancellationToken) + => readDeadline?.Begin(cancellationToken) ?? cancellationToken; + + /// Closes the idle read deadline opened by . + private void EndRead() => readDeadline?.End(); + + /// + /// Tells the server to stop the query this connection is running, so it does not keep working and writing + /// into a socket nobody will read. Best effort: the connection is closed next whatever happens here, so a + /// failure to deliver the packet must not replace the failure that brought us here. + /// + /// + /// The server reads this between the blocks it sends, so a result large enough to fill the socket blocks it in + /// a write where it reads nothing, and the close that follows stops the query instead. Verified against a real + /// server: a slow result ends with QUERY_WAS_CANCELLED_BY_CLIENT, a saturating one with a broken pipe. Both + /// stop it, so this is what turns a silent abandonment into an explicit one rather than the only way out. + /// + /// + /// Whether everything the client has flushed ended a packet. False leaves the Cancel unsent, because a + /// server part-way through reading a packet takes the next byte as more of that packet, not as a new one. + /// Only a caller that has completed a flush may pass true: a flush that fails part-way through leaves bytes + /// on the wire that writer.Reset() cannot take back. + /// + /// A task that completes once the packet has been sent, or given up on. + private async ValueTask TrySendCancelAsync(bool flushedWholePackets) + { + // Terminated means the socket is already gone, from a concurrent AbortTransport. Checked rather than + // left to the flush to discover, which only fails safely because a disposed writer holds an empty array. + if (!flushedWholePackets || state == TcpConnectionState.Terminated) + { + return; + } + + try + { + // Discard anything the interrupted operation left buffered, so Cancel is the whole of what goes out. + writer.Reset(); + writer.WriteClientPacketType(ClientPacketType.Cancel); + + // Not the operation's token: it is usually the cancelled one that brought us here, and flushing on it + // would send nothing. The separate deadline stops a server that has also stopped reading from holding + // the caller here — one byte, so it only elapses against a peer whose receive window is shut. It runs + // before the pool lease is given back, so it is also how long the next caller can wait for the slot. + using var deadline = new CancellationTokenSource(CancelSendTimeout); + await writer.FlushAsync(deadline.Token).ConfigureAwait(false); + } + catch (Exception e) when (e is not (OutOfMemoryException or StackOverflowException)) + { + // A connection too broken to carry one byte needs no cancelling. + } + } } diff --git a/ClickHouse.Driver.Tcp/Protocol/IdleReadDeadline.cs b/ClickHouse.Driver.Tcp/Protocol/IdleReadDeadline.cs new file mode 100644 index 000000000..6673ca204 --- /dev/null +++ b/ClickHouse.Driver.Tcp/Protocol/IdleReadDeadline.cs @@ -0,0 +1,86 @@ +using System; +using System.Threading; + +namespace ClickHouse.Driver.Tcp.Protocol; + +/// +/// Bounds how long the transport may go without delivering a byte, so an operation against a server that has +/// stopped answering fails instead of waiting for TCP to give up. +/// +/// +/// +/// The deadline is armed immediately before each read from the transport and disarmed as soon as that read +/// returns, so it measures silence rather than elapsed time: a result that streams for an hour never trips it, +/// and neither does a consumer that holds a block for longer than the deadline before asking for the next one. +/// +/// +/// Not thread-safe, and scoped to one operation: opens it over the caller's token and +/// returns the token every read of that operation must observe, closes it. +/// +/// +internal sealed class IdleReadDeadline +{ + private readonly TimeSpan timeout; + private CancellationTokenSource source; + private CancellationToken callerToken; + + /// Initializes a deadline of . + /// + /// How long the transport may stay silent before the read fails. Must be positive and within what a timer + /// can hold, which checks; a connection given + /// builds no deadline at all rather than one of zero length. + /// + internal IdleReadDeadline(TimeSpan timeout) => this.timeout = timeout; + + /// Whether the deadline elapsed, as opposed to the caller cancelling. + internal bool Elapsed => source is { IsCancellationRequested: true } && !callerToken.IsCancellationRequested; + + /// + /// Opens the deadline over an operation's token. The returned token fires when either the caller cancels or + /// the transport stays silent too long. + /// + /// + /// For the operation's reads only. Writes keep the caller's token: disarming cannot recall a timer + /// callback that has already begun, so a deadline elapsing just as a read completes would otherwise cancel + /// the write that follows, for a token the caller never cancelled. + /// + /// The caller's token for this operation. + /// The token every read of this operation must observe. + internal CancellationToken Begin(CancellationToken operationToken) + { + callerToken = operationToken; + source = CancellationTokenSource.CreateLinkedTokenSource(operationToken); + return source.Token; + } + + /// Closes the deadline. Safe to call without a matching . + internal void End() + { + source?.Dispose(); + source = null; + callerToken = default; + } + + /// Starts the clock, immediately before a read from the transport. + internal void Arm() => Reschedule(timeout); + + /// Stops the clock, as soon as a read from the transport returns. + internal void Disarm() => Reschedule(Timeout.InfiniteTimeSpan); + + /// The failure to report when a read gave up because the transport stayed silent. + /// A timeout naming the option that set the deadline. + internal TimeoutException ToException() + => new($"The server sent nothing for {timeout.TotalSeconds:0.###}s while a response was being read (ReadTimeout)."); + + private void Reschedule(TimeSpan delay) + { + // Already fired, or never opened: CancelAfter would throw on the disposed source, and there is nothing + // left to bound either way — the read is about to unwind. + if (source is null || source.IsCancellationRequested) + { + return; + } + + source.CancelAfter(delay); + } +} diff --git a/ClickHouse.Driver.Tcp/Protocol/ReadBuffer.cs b/ClickHouse.Driver.Tcp/Protocol/ReadBuffer.cs index b02d81631..959b09e17 100644 --- a/ClickHouse.Driver.Tcp/Protocol/ReadBuffer.cs +++ b/ClickHouse.Driver.Tcp/Protocol/ReadBuffer.cs @@ -27,6 +27,7 @@ internal sealed class ReadBuffer : IDisposable private readonly Stream stream; private readonly bool readsFromTransport; + private readonly IdleReadDeadline deadline; private byte[] buffer; private int capacity; private int head; // index of the first valid byte @@ -42,9 +43,14 @@ internal sealed class ReadBuffer : IDisposable /// Whether is the connection itself, so that a failed read is the transport failing. /// False for an adapter stream, whose own layer decides what its failures mean. /// + /// + /// Bounds how long may stay silent, or null to wait as long as the caller's token + /// allows. An adapter stream's reads are served from bytes the transport already delivered, so they cannot + /// stall on the network and are given none. + /// /// is null. /// is below . - public ReadBuffer(Stream stream, int capacity = 16384, bool readsFromTransport = true) + public ReadBuffer(Stream stream, int capacity = 16384, bool readsFromTransport = true, IdleReadDeadline deadline = null) { this.stream = stream ?? throw new ArgumentNullException(nameof(stream)); if (capacity < MaxContiguous) @@ -53,6 +59,7 @@ public ReadBuffer(Stream stream, int capacity = 16384, bool readsFromTransport = } this.readsFromTransport = readsFromTransport; + this.deadline = deadline; buffer = ArrayPool.Shared.Rent(capacity); this.capacity = buffer.Length; // Rent may return a larger array; use all of it. } @@ -67,6 +74,7 @@ public ReadBuffer(Stream stream, int capacity = 16384, bool readsFromTransport = /// The number of contiguous bytes that must be available; must not exceed . /// A token to observe for cancellation. /// exceeds the buffer capacity. + /// The stream stayed silent past the idle deadline. /// The stream ended before enough bytes arrived, or the read failed. public async ValueTask EnsureAsync(int needed, CancellationToken cancellationToken) { @@ -137,6 +145,7 @@ public ReadOnlySpan ReadSpan(int count) /// /// The region to fill completely with consumed bytes. /// A token to observe for cancellation. + /// The stream stayed silent past the idle deadline. /// The stream ended before the destination was filled, or the read failed. public async ValueTask ReadIntoAsync(Memory destination, CancellationToken cancellationToken) { @@ -150,21 +159,7 @@ public async ValueTask ReadIntoAsync(Memory destination, CancellationToken while (!destination.IsEmpty) { - int read; - try - { - read = await stream.ReadAsync(destination, cancellationToken).ConfigureAwait(false); - } - catch (Exception e) when (readsFromTransport && TransportFailure.IsTransportFailure(e)) - { - throw TransportFailure.Read(e); - } - - if (read == 0) - { - throw TransportFailure.EndOfStream(); - } - + int read = await ReadFromStreamAsync(destination, cancellationToken).ConfigureAwait(false); destination = destination.Slice(read); } } @@ -204,21 +199,50 @@ private void Advance(int n) private async ValueTask FillOnceAsync(CancellationToken cancellationToken) { int writeStart = head + buffered; + int read = await ReadFromStreamAsync(buffer.AsMemory(writeStart, capacity - writeStart), cancellationToken).ConfigureAwait(false); + buffered += read; + } + + /// + /// Reads once from the stream, holding the idle deadline open only for as long as that read is pending. The + /// single point where a byte arrives, so it is also where a silent transport, a closed one and a failed one + /// are each turned into the exception that describes them. + /// + /// Where to put the bytes; filled in part or in whole. + /// A token to observe for cancellation. + /// The number of bytes read, always at least one. + /// The stream delivered nothing within the deadline. + /// The stream ended, or the read failed. + private async ValueTask ReadFromStreamAsync(Memory destination, CancellationToken cancellationToken) + { int read; + deadline?.Arm(); try { - read = await stream.ReadAsync(buffer.AsMemory(writeStart, capacity - writeStart), cancellationToken).ConfigureAwait(false); + read = await stream.ReadAsync(destination, cancellationToken).ConfigureAwait(false); + } + catch (Exception e) when (deadline is { Elapsed: true } && (e is OperationCanceledException || TransportFailure.IsTransportFailure(e))) + { + // The deadline's token cancelled the read, not the caller's. Report the silence rather than a + // cancellation the caller never asked for. A transport failure counts too: SslStream turns a + // cancelled read into an IOException unless the token matches the one it was given, and a timeout + // reported as a broken connection names neither the option that caused it nor the fix. + throw deadline.ToException(); } catch (Exception e) when (readsFromTransport && TransportFailure.IsTransportFailure(e)) { throw TransportFailure.Read(e); } + finally + { + deadline?.Disarm(); + } if (read == 0) { throw TransportFailure.EndOfStream(); } - buffered += read; + return read; } }