diff --git a/.github/workflows/CodeQuality.yml b/.github/workflows/CodeQuality.yml index 1442597..2c75de2 100644 --- a/.github/workflows/CodeQuality.yml +++ b/.github/workflows/CodeQuality.yml @@ -45,4 +45,25 @@ jobs: - name: Sonarqube end env: SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} - run: dotnet sonarscanner end /d:sonar.token="$SONAR_TOKEN" \ No newline at end of file + run: dotnet sonarscanner end /d:sonar.token="$SONAR_TOKEN" + + # Named pipes behave differently on Windows (unbuffered pipe instances instead of Unix domain sockets), and some + # tests only run there ([Platform("Win")]). Analysis and coverage stay in the Linux job above. + test-windows: + name: Test (Windows) + runs-on: windows-latest + steps: + - uses: actions/checkout@v6.0.2 + - name: Setup dotnet + uses: actions/setup-dotnet@v5.1.0 + with: + dotnet-version: '10.0.x' + + - name: Restore dependencies + run: dotnet restore ArgusTransfer.sln + + - name: Build + run: dotnet build ArgusTransfer.sln --no-restore + + - name: Run Tests + run: dotnet test ArgusTransfer.sln --no-restore --no-build --verbosity normal \ No newline at end of file diff --git a/ArgusTransfer.Tests/Serialization/ArgusChunkedWriteStreamTestFixture.cs b/ArgusTransfer.Tests/Serialization/ArgusChunkedWriteStreamTestFixture.cs index 4c12559..3b29de5 100644 --- a/ArgusTransfer.Tests/Serialization/ArgusChunkedWriteStreamTestFixture.cs +++ b/ArgusTransfer.Tests/Serialization/ArgusChunkedWriteStreamTestFixture.cs @@ -21,8 +21,10 @@ namespace ArgusTransfer.Tests.Serialization { using System; + using System.Collections.Generic; using System.IO; using System.Text; + using System.Threading; using System.Threading.Tasks; using ArgusTransfer.Serialization; @@ -75,5 +77,91 @@ public void Verify_that_stream_is_write_only() Assert.That(() => chunked.SetLength(0), Throws.TypeOf()); Assert.That(() => chunked.Flush(), Throws.Nothing); } + + [Test] + public async Task Verify_that_write_without_a_token_uses_the_token_passed_to_the_constructor() + { + using var cts = new CancellationTokenSource(); + using var destination = new TokenRecordingStream(); + await using var chunked = new ArgusChunkedWriteStream(destination, cts.Token); + + await chunked.WriteAsync(Encoding.ASCII.GetBytes("abc").AsMemory()); + + Assert.That(destination.Tokens, Has.Count.EqualTo(3)); + Assert.That(destination.Tokens, Has.All.EqualTo(cts.Token)); + } + + [Test] + public async Task Verify_that_write_with_a_token_uses_that_token() + { + using var constructorCts = new CancellationTokenSource(); + using var writeCts = new CancellationTokenSource(); + using var destination = new TokenRecordingStream(); + await using var chunked = new ArgusChunkedWriteStream(destination, constructorCts.Token); + + await chunked.WriteAsync(Encoding.ASCII.GetBytes("abc").AsMemory(), writeCts.Token); + + Assert.That(destination.Tokens, Has.Count.EqualTo(3)); + Assert.That(destination.Tokens, Has.All.EqualTo(writeCts.Token)); + } + + [Test] + public async Task Verify_that_cancelling_the_constructor_token_cancels_a_write_without_a_token() + { + using var cts = new CancellationTokenSource(); + using var destination = new MemoryStream(); + await using var chunked = new ArgusChunkedWriteStream(destination, cts.Token); + + await cts.CancelAsync(); + + Assert.That( + async () => await chunked.WriteAsync(Encoding.ASCII.GetBytes("abc").AsMemory()), + Throws.InstanceOf()); + Assert.That(destination.Length, Is.Zero); + } + + [Test] + public async Task Verify_that_compressed_chunked_writing_passes_the_token_to_every_write() + { + using var cts = new CancellationTokenSource(); + using var destination = new TokenRecordingStream(); + + // The compression stream writes its last block when it is disposed, without a token of its own; on an + // unbuffered pipe that write must still be cancellable + await ArgusChunkedEncoding.WriteChunkedAsync(new MemoryStream(new byte[200]), destination, new GZipArgusContentEncoding(), cts.Token); + + Assert.That(destination.Tokens, Is.Not.Empty); + Assert.That(destination.Tokens, Has.All.EqualTo(cts.Token)); + } + + /// + /// A that records the of every asynchronous write + /// + private sealed class TokenRecordingStream : MemoryStream + { + /// + /// Gets the tokens of the asynchronous writes, in the order of the writes + /// + public List Tokens { get; } = new List(); + + /// + /// Records the token and writes the bytes + /// + /// + /// The data to write + /// + /// + /// The to record + /// + /// + /// A representing the asynchronous operation + /// + public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + this.Tokens.Add(cancellationToken); + + return base.WriteAsync(buffer, cancellationToken); + } + } } } diff --git a/ArgusTransfer.Tests/Server/ArgusPipeHostBackgroundServiceTestFixture.cs b/ArgusTransfer.Tests/Server/ArgusPipeHostBackgroundServiceTestFixture.cs index 8099f7d..6fd4042 100644 --- a/ArgusTransfer.Tests/Server/ArgusPipeHostBackgroundServiceTestFixture.cs +++ b/ArgusTransfer.Tests/Server/ArgusPipeHostBackgroundServiceTestFixture.cs @@ -1451,6 +1451,573 @@ public async Task Verify_that_MaxConcurrentConnections_makes_further_clients_wai cts.Dispose(); } + [Test] + public void Verify_that_ResponseWriteTimeout_defaults_to_30_seconds() + { + Assert.That(new ArgusPipeHostOptions().ResponseWriteTimeout, Is.EqualTo(TimeSpan.FromSeconds(30))); + } + + /// + /// The size of a response body that no pipe or socket buffer can hold, so writing it blocks on every platform until + /// the client reads + /// + private const int UnbufferableBodySize = 8 * 1024 * 1024; + + private const string ResponseNotSentWarning = "The response was not sent within"; + + private static ArgusRouter CreateLargeResponseRouter() + { + var router = new ArgusRouter(); + + router.MapGet("/large", context => + { + context.Response = new ArgusResponse { StatusCode = ArgusStatusCode.Ok, Body = new string('a', UnbufferableBodySize) }; + return Task.CompletedTask; + }); + + router.MapGet("/small", context => + { + context.Response = new ArgusResponse { StatusCode = ArgusStatusCode.Ok, Body = "ok" }; + return Task.CompletedTask; + }); + + return router; + } + + private static async Task SendRequestWithoutReadingAsync(string pipeName, string route, string accept = null) + { + var pipe = await ConnectRawAsync(pipeName); + var request = new ArgusRequest { Verb = ArgusVerb.GET, Route = route }; + + if (accept != null) + { + request.Accept = accept; + } + + await new ArgusRequestSerializer().WriteAsync(pipe, request); + + return pipe; + } + + private int CountWarnings(string text) + { + return this.mockLogger.Invocations.Count(invocation => + invocation.Arguments.Count > 2 + && invocation.Arguments[0] is LogLevel.Warning + && invocation.Arguments[2].ToString().Contains(text)); + } + + private async Task WaitForWarningAsync(string text) + { + var stopwatch = Stopwatch.StartNew(); + + while (this.CountWarnings(text) == 0 && stopwatch.Elapsed < TimeSpan.FromSeconds(5)) + { + await Task.Delay(25); + } + + return this.CountWarnings(text) > 0; + } + + [Test] + public async Task Verify_that_client_that_never_reads_the_response_is_disconnected_after_ResponseWriteTimeout() + { + var pipeName = $"argus-unread-response-test-{Guid.NewGuid():N}"; + var (hostService, cts) = await this.StartHostAsync( + CreateLargeResponseRouter(), + new ArgusPipeHostOptions { PipeName = pipeName, MaxConcurrentConnections = 1, MaxConcurrentRequests = 1, ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }); + + // The client sends a valid request and then never reads the response, holding the only connection slot and + // the only request slot + await using var silent = await SendRequestWithoutReadingAsync(pipeName, "/large"); + + // The host must give up on the response and free both slots, so that another client is served: it would + // otherwise wait for the connection slot, or get a 503 for the request slot + using var client = new ArgusClient(pipeName); + var response = await client.GetAsync("/small", timeout: TimeSpan.FromSeconds(10)); + + Assert.That(response.StatusCode, Is.EqualTo(ArgusStatusCode.Ok)); + Assert.That(response.Body, Is.EqualTo("ok")); + Assert.That(this.CountWarnings(ResponseNotSentWarning), Is.EqualTo(1)); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + public async Task Verify_that_client_that_stops_reading_the_response_sees_the_connection_closed() + { + var pipeName = $"argus-partial-response-test-{Guid.NewGuid():N}"; + var (hostService, cts) = await this.StartHostAsync( + CreateLargeResponseRouter(), + new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }); + + await using var pipe = await SendRequestWithoutReadingAsync(pipeName, "/large"); + + // Read the start of the response, then stop reading until the timeout has expired + var buffer = new byte[1024]; + var total = await pipe.ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(5)); + + Assert.That(total, Is.GreaterThan(0)); + Assert.That(await this.WaitForWarningAsync(ResponseNotSentWarning), Is.True); + + // What the host had already sent can still be read; then the stream ends before the complete body arrived + var read = total; + + try + { + while (read > 0) + { + read = await pipe.ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(5)); + total += read; + } + } + catch (System.IO.IOException) + { + // a closed connection may also surface as a broken pipe + } + + Assert.That(total, Is.LessThan(UnbufferableBodySize)); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + public async Task Verify_that_slow_reader_receives_the_complete_response_when_ResponseWriteTimeout_is_infinite() + { + var pipeName = $"argus-slow-reader-test-{Guid.NewGuid():N}"; + var (hostService, cts) = await this.StartHostAsync( + CreateLargeResponseRouter(), + new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = Timeout.InfiniteTimeSpan }); + + await using var pipe = await SendRequestWithoutReadingAsync(pipeName, "/large"); + + await Task.Delay(500); + + var response = await new ArgusResponseSerializer().ReadAsync(pipe, CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.That(response.StatusCode, Is.EqualTo(ArgusStatusCode.Ok)); + Assert.That(response.Body, Has.Length.EqualTo(UnbufferableBodySize)); + Assert.That(this.CountWarnings(ResponseNotSentWarning), Is.Zero); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + public async Task Verify_that_response_read_within_ResponseWriteTimeout_is_not_cut_off() + { + var pipeName = $"argus-timely-reader-test-{Guid.NewGuid():N}"; + var (hostService, cts) = await this.StartHostAsync( + CreateLargeResponseRouter(), + new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = TimeSpan.FromSeconds(10) }); + + await using var pipe = await SendRequestWithoutReadingAsync(pipeName, "/large"); + + await Task.Delay(300); + + var response = await new ArgusResponseSerializer().ReadAsync(pipe, CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.That(response.Body, Has.Length.EqualTo(UnbufferableBodySize)); + Assert.That(this.CountWarnings(ResponseNotSentWarning), Is.Zero); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + // The tests below cover the host's small built-in responses. They run on Windows only: there the pipe has no + // buffer, so even a small response blocks until the client reads. On Linux and macOS the socket buffers a small + // response and the write completes without a reader, so there is nothing to time out. + + [Test] + [Platform("Win")] + public async Task Verify_that_unread_BadRequest_response_times_out_after_ResponseWriteTimeout() + { + var pipeName = $"argus-unread-400-test-{Guid.NewGuid():N}"; + var (hostService, cts) = await this.StartHostAsync( + new ArgusRouter(), + new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }); + + await using var pipe = await ConnectRawAsync(pipeName); + await pipe.WriteAsync(System.Text.Encoding.ASCII.GetBytes("GET /items HTTP/1.1\r\n\r\n")); + + Assert.That(await this.WaitForWarningAsync(ResponseNotSentWarning), Is.True); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + [Platform("Win")] + public async Task Verify_that_unread_concurrency_limit_response_times_out_after_ResponseWriteTimeout() + { + var pipeName = $"argus-unread-503-test-{Guid.NewGuid():N}"; + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var router = new ArgusRouter(); + router.MapGet("/slow", async context => + { + await release.Task; + context.Response = new ArgusResponse { StatusCode = ArgusStatusCode.Ok }; + }); + + var (hostService, cts) = await this.StartHostAsync( + router, + new ArgusPipeHostOptions { PipeName = pipeName, MaxConcurrentRequests = 1, ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }); + + // The first request occupies the only processing slot + using var client = new ArgusClient(pipeName); + var firstRequest = client.GetAsync("/slow", timeout: TimeSpan.FromSeconds(10)); + var stopwatch = Stopwatch.StartNew(); + + while (hostService.CurrentRequestCount == 0 && stopwatch.Elapsed < TimeSpan.FromSeconds(5)) + { + await Task.Delay(25); + } + + // The second request is rejected with 503, which its client never reads + await using var silent = await SendRequestWithoutReadingAsync(pipeName, "/slow"); + + Assert.That(await this.WaitForWarningAsync(ResponseNotSentWarning), Is.True); + Assert.That(hostService.RejectedRequestCount, Is.EqualTo(1)); + + release.SetResult(); + var firstResponse = await firstRequest; + + Assert.That(firstResponse.StatusCode, Is.EqualTo(ArgusStatusCode.Ok)); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + [Platform("Win")] + public async Task Verify_that_unread_request_timeout_response_times_out_after_ResponseWriteTimeout() + { + var pipeName = $"argus-unread-timeout-test-{Guid.NewGuid():N}"; + + var router = new ArgusRouter(); + router.MapGet("/slow", async context => + { + await Task.Delay(Timeout.Infinite, context.RequestAborted); + }); + + var (hostService, cts) = await this.StartHostAsync( + router, + new ArgusPipeHostOptions { PipeName = pipeName, RequestTimeout = TimeSpan.FromMilliseconds(100), ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }); + + await using var silent = await SendRequestWithoutReadingAsync(pipeName, "/slow"); + + Assert.That(await this.WaitForWarningAsync(ResponseNotSentWarning), Is.True); + Assert.That(this.CountWarnings("timed out after"), Is.EqualTo(1)); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + [Platform("Win")] + public async Task Verify_that_unread_NotAcceptable_response_times_out_after_ResponseWriteTimeout() + { + var pipeName = $"argus-unread-406-test-{Guid.NewGuid():N}"; + + var hostService = new ArgusPipeHostBackgroundService( + this.mockLogger.Object, + new ArgusRouter(), + Options.Create(new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }), + new ArgusBodySerializerRegistry(new[] { new PlainTextArgusBodySerializer() })); + + using var cts = new CancellationTokenSource(); + await hostService.StartAsync(cts.Token); + + await using var silent = await SendRequestWithoutReadingAsync(pipeName, "/items", accept: "application/xml"); + + Assert.That(await this.WaitForWarningAsync(ResponseNotSentWarning), Is.True); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + } + + [Test] + [Platform("Win")] + public async Task Verify_that_compressed_streamed_response_that_is_read_partially_times_out_after_ResponseWriteTimeout() + { + var pipeName = $"argus-unread-gzip-test-{Guid.NewGuid():N}"; + + var router = new ArgusRouter(); + router.MapGet("/stream", context => + { + context.Response = new ArgusResponse { StatusCode = ArgusStatusCode.Ok, BodyStream = new System.IO.MemoryStream(new byte[200]) }; + return Task.CompletedTask; + }); + + var options = new ArgusPipeHostOptions { PipeName = pipeName, MaxConcurrentConnections = 1, ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }; + options.Compression.Enabled = true; + options.Compression.MinimumBodySize = 0; + + var (hostService, cts) = await this.StartHostAsync(router, options); + + await using var pipe = await ConnectRawAsync(pipeName); + var request = new ArgusRequest { Verb = ArgusVerb.GET, Route = "/stream" }; + request.Headers[ArgusHeaderNames.AcceptEncoding] = "gzip"; + await new ArgusRequestSerializer().WriteAsync(pipe, request); + + // Read the header block and the first chunk, then stop. What remains is the last compressed block, which the + // compression stream writes when it is disposed, without a cancellation token of its own + var head = await ReadUntilAsync(pipe, "\r\n\r\n"); + var sizeLine = await ReadUntilAsync(pipe, "\r\n"); + var chunkSize = int.Parse(sizeLine.Trim(), System.Globalization.NumberStyles.HexNumber, System.Globalization.CultureInfo.InvariantCulture); + await pipe.ReadExactlyAsync(new byte[chunkSize + 2]).AsTask().WaitAsync(TimeSpan.FromSeconds(5)); + + Assert.That(head, Does.Contain("Content-Encoding: gzip")); + Assert.That(await this.WaitForWarningAsync(ResponseNotSentWarning), Is.True); + + // The connection slot is free again + using var client = new ArgusClient(pipeName); + var response = await client.GetAsync("/stream", timeout: TimeSpan.FromSeconds(10)); + + Assert.That(response.StatusCode, Is.EqualTo(ArgusStatusCode.Ok)); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + public async Task Verify_that_streamed_response_that_is_not_read_times_out_after_ResponseWriteTimeout() + { + var pipeName = $"argus-unread-stream-test-{Guid.NewGuid():N}"; + + var router = CreateLargeResponseRouter(); + router.MapGet("/stream", context => + { + context.Response = new ArgusResponse { StatusCode = ArgusStatusCode.Ok, BodyStream = new System.IO.MemoryStream(new byte[UnbufferableBodySize]) }; + return Task.CompletedTask; + }); + + var (hostService, cts) = await this.StartHostAsync( + router, + new ArgusPipeHostOptions { PipeName = pipeName, MaxConcurrentConnections = 1, ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }); + + // A streamed body is sent as many chunk writes instead of one write; the timeout must reach those as well + await using var silent = await SendRequestWithoutReadingAsync(pipeName, "/stream"); + + using var client = new ArgusClient(pipeName); + var response = await client.GetAsync("/small", timeout: TimeSpan.FromSeconds(10)); + + Assert.That(response.StatusCode, Is.EqualTo(ArgusStatusCode.Ok)); + Assert.That(this.CountWarnings(ResponseNotSentWarning), Is.EqualTo(1)); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + public async Task Verify_that_client_disconnecting_during_the_response_is_not_logged_as_a_write_timeout() + { + var pipeName = $"argus-disconnect-during-write-test-{Guid.NewGuid():N}"; + var (hostService, cts) = await this.StartHostAsync( + CreateLargeResponseRouter(), + new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = TimeSpan.FromSeconds(10) }); + + var pipe = await SendRequestWithoutReadingAsync(pipeName, "/large"); + + // Let the host start writing the response, which blocks because nothing is read + await Task.Delay(200); + + Assert.That(hostService.GetActiveRequestTasks(), Has.Count.EqualTo(1)); + + await pipe.DisposeAsync(); + + var stopwatch = Stopwatch.StartNew(); + + while (hostService.GetActiveRequestTasks().Count > 0 && stopwatch.Elapsed < TimeSpan.FromSeconds(5)) + { + await Task.Delay(25); + } + + // The write failed because the client went away, long before the timeout: that is not a write timeout + Assert.That(hostService.GetActiveRequestTasks(), Is.Empty); + Assert.That(this.CountWarnings(ResponseNotSentWarning), Is.Zero); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + public async Task Verify_that_handler_running_longer_than_ResponseWriteTimeout_still_gets_its_response_written() + { + var pipeName = $"argus-slow-handler-test-{Guid.NewGuid():N}"; + + var router = new ArgusRouter(); + router.MapGet("/slow", async context => + { + await Task.Delay(600, context.RequestAborted); + context.Response = new ArgusResponse { StatusCode = ArgusStatusCode.Ok, Body = "done" }; + }); + + var (hostService, cts) = await this.StartHostAsync( + router, + new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = TimeSpan.FromMilliseconds(200) }); + + // The timeout starts when the response is written; the time the handler takes is bounded by RequestTimeout + using var client = new ArgusClient(pipeName); + var response = await client.GetAsync("/slow", timeout: TimeSpan.FromSeconds(10)); + + Assert.That(response.StatusCode, Is.EqualTo(ArgusStatusCode.Ok)); + Assert.That(response.Body, Is.EqualTo("done")); + Assert.That(this.CountWarnings(ResponseNotSentWarning), Is.Zero); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + [Test] + public async Task Verify_that_StopAsync_cancels_a_response_write_that_the_client_never_reads() + { + var pipeName = $"argus-unread-shutdown-test-{Guid.NewGuid():N}"; + var handlerCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var router = new ArgusRouter(); + router.MapGet("/large", context => + { + context.Response = new ArgusResponse { StatusCode = ArgusStatusCode.Ok, Body = new string('a', UnbufferableBodySize) }; + handlerCompleted.SetResult(); + return Task.CompletedTask; + }); + + // Without a write timeout, only the shutdown drain can end the blocked write + var (hostService, cts) = await this.StartHostAsync( + router, + new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = Timeout.InfiniteTimeSpan, ShutdownDrainTimeout = TimeSpan.FromMilliseconds(100) }); + + hostService.CancellationGracePeriod = TimeSpan.FromSeconds(5); + + await using var silent = await SendRequestWithoutReadingAsync(pipeName, "/large"); + await handlerCompleted.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await Task.Delay(100); + + var requestTasks = hostService.GetActiveRequestTasks(); + + Assert.That(requestTasks, Has.Count.EqualTo(1)); + Assert.That(requestTasks.Single().IsCompleted, Is.False, "the response write must be blocked on the client that does not read"); + + var stopwatch = Stopwatch.StartNew(); + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + + Assert.That(stopwatch.Elapsed, Is.LessThan(TimeSpan.FromSeconds(3)), "the drain must cancel the write well within the grace period"); + Assert.That(requestTasks.Single().IsCompletedSuccessfully, Is.True); + Assert.That(this.CountWarnings(ResponseNotSentWarning), Is.Zero, "a write cancelled by shutdown is not a write timeout"); + Assert.That(this.CountWarnings("did not finish after cancellation"), Is.Zero); + + cts.Dispose(); + } + + [Test] + public async Task Verify_that_response_streamed_from_a_stalled_source_is_cut_off_after_ResponseWriteTimeout() + { + var pipeName = $"argus-stalled-source-test-{Guid.NewGuid():N}"; + + var router = new ArgusRouter(); + router.MapGet("/stream", context => + { + context.Response = new ArgusResponse { StatusCode = ArgusStatusCode.Ok, BodyStream = new StalledStream() }; + return Task.CompletedTask; + }); + + var (hostService, cts) = await this.StartHostAsync( + router, + new ArgusPipeHostOptions { PipeName = pipeName, ResponseWriteTimeout = TimeSpan.FromMilliseconds(300) }); + + // The timeout is a budget for the whole response: the client reads promptly, but the handler's stream stops + // producing data, so the response is cut off and the client sees the connection closed + using var client = new ArgusClient(pipeName); + + Assert.That( + async () => await client.GetAsync("/stream", timeout: TimeSpan.FromSeconds(10)), + Throws.InstanceOf()); + Assert.That(await this.WaitForWarningAsync(ResponseNotSentWarning), Is.True); + + await cts.CancelAsync(); + await hostService.StopAsync(CancellationToken.None); + cts.Dispose(); + } + + /// + /// A read-only that produces one byte and then no more data until the read is cancelled + /// + private sealed class StalledStream : System.IO.Stream + { + private bool firstByteRead; + + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => false; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + if (this.firstByteRead) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + } + + this.firstByteRead = true; + buffer.Span[0] = (byte)'a'; + + return 1; + } + + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + public override void Flush() + { + // nothing to flush + } + + public override long Seek(long offset, System.IO.SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } + + private static async Task ReadUntilAsync(System.IO.Stream stream, string terminator) + { + // One byte at a time, so that nothing beyond the terminator is taken from the pipe + var text = new System.Text.StringBuilder(); + var buffer = new byte[1]; + + while (!text.ToString().EndsWith(terminator, StringComparison.Ordinal)) + { + await stream.ReadExactlyAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(5)); + text.Append((char)buffer[0]); + } + + return text.ToString(); + } + [Test] public void Verify_that_MaxRequestHeaderSize_defaults_to_32_KB() { diff --git a/ArgusTransfer/Serialization/ArgusChunkedEncoding.cs b/ArgusTransfer/Serialization/ArgusChunkedEncoding.cs index d1692ff..7acf567 100644 --- a/ArgusTransfer/Serialization/ArgusChunkedEncoding.cs +++ b/ArgusTransfer/Serialization/ArgusChunkedEncoding.cs @@ -77,7 +77,7 @@ public static class ArgusChunkedEncoding /// internal static async Task WriteChunkedAsync(Stream source, Stream destination, IArgusContentEncoding encoding = null, CancellationToken cancellationToken = default) { - await using var chunked = new ArgusChunkedWriteStream(destination); + await using var chunked = new ArgusChunkedWriteStream(destination, cancellationToken); if (encoding == null) { diff --git a/ArgusTransfer/Serialization/ArgusChunkedWriteStream.cs b/ArgusTransfer/Serialization/ArgusChunkedWriteStream.cs index e404c1d..14f69c1 100644 --- a/ArgusTransfer/Serialization/ArgusChunkedWriteStream.cs +++ b/ArgusTransfer/Serialization/ArgusChunkedWriteStream.cs @@ -43,15 +43,27 @@ internal sealed class ArgusChunkedWriteStream : Stream /// private readonly Stream destination; + /// + /// The of the operation the stream is used in, applied to asynchronous writes whose + /// caller passes no token + /// + private readonly CancellationToken cancellationToken; + /// /// Initializes a new instance of the class /// /// /// The that receives the chunks /// - public ArgusChunkedWriteStream(Stream destination) + /// + /// The of the operation the stream is used in. It is applied to asynchronous + /// writes whose caller passes no token of its own, such as the final block a compression stream writes when it + /// is disposed, so that those writes can be cancelled as well + /// + public ArgusChunkedWriteStream(Stream destination, CancellationToken cancellationToken = default) { this.destination = destination; + this.cancellationToken = cancellationToken; } /// @@ -154,7 +166,8 @@ public override Task WriteAsync(byte[] buffer, int offset, int count, Cancellati /// The data to write /// /// - /// The used to signal cancellation + /// The used to signal cancellation. When it cannot be cancelled, the token passed + /// to the constructor is used instead /// /// /// A representing the asynchronous operation @@ -166,9 +179,12 @@ public override async ValueTask WriteAsync(ReadOnlyMemory buffer, Cancella return; } - await this.destination.WriteAsync(SizeLine(buffer.Length).AsMemory(), cancellationToken); - await this.destination.WriteAsync(buffer, cancellationToken); - await this.destination.WriteAsync(CrLf.AsMemory(), cancellationToken); + // A compression stream writes its last block when it is disposed and passes no token then + var token = cancellationToken.CanBeCanceled ? cancellationToken : this.cancellationToken; + + await this.destination.WriteAsync(SizeLine(buffer.Length).AsMemory(), token); + await this.destination.WriteAsync(buffer, token); + await this.destination.WriteAsync(CrLf.AsMemory(), token); } /// diff --git a/ArgusTransfer/Server/ArgusPipeHostBackgroundService.cs b/ArgusTransfer/Server/ArgusPipeHostBackgroundService.cs index e1524fb..3b3fd32 100644 --- a/ArgusTransfer/Server/ArgusPipeHostBackgroundService.cs +++ b/ArgusTransfer/Server/ArgusPipeHostBackgroundService.cs @@ -293,7 +293,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) var rejectResponse = CreateProblemResponse(ArgusStatusCode.ServiceUnavailable, ConcurrencyLimitDetail, request); - await this.responseSerializer.WriteAsync(serverStream, rejectResponse, requestToken); + await this.WriteResponseAsync(token => this.responseSerializer.WriteAsync(serverStream, rejectResponse, token), requestToken); return; } @@ -320,7 +320,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) $"None of the media types in the Accept header '{request.Accept}' can be produced. Supported media types: {supported}.", request); - await this.responseSerializer.WriteAsync(serverStream, notAcceptable, requestToken); + await this.WriteResponseAsync(token => this.responseSerializer.WriteAsync(serverStream, notAcceptable, token), requestToken); return; } } @@ -346,7 +346,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) var timeoutResponse = CreateProblemResponse(ArgusStatusCode.ServiceUnavailable, RequestTimeoutDetail, request); - await this.responseSerializer.WriteAsync(serverStream, timeoutResponse, requestToken); + await this.WriteResponseAsync(token => this.responseSerializer.WriteAsync(serverStream, timeoutResponse, token), requestToken); return; } catch (OperationCanceledException) @@ -380,7 +380,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) ArgusCompression.ApplyToResponse(request, context.Response, this.compression); - await this.responseSerializer.WriteAsync(serverStream, context.Response, negotiatedContentType, requestToken); + await this.WriteResponseAsync(token => this.responseSerializer.WriteAsync(serverStream, context.Response, negotiatedContentType, token), requestToken); } catch (Exception ex) when (ex is ArgusProtocolException or FormatException) { @@ -393,7 +393,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) // The request could not be read, so there is no correlation token or verb to use var badRequest = CreateProblemResponse(ArgusStatusCode.BadRequest, ex.Message, null); - await this.responseSerializer.WriteAsync(serverStream, badRequest, requestToken); + await this.WriteResponseAsync(token => this.responseSerializer.WriteAsync(serverStream, badRequest, token), requestToken); } catch { @@ -423,6 +423,44 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) } } + /// + /// Writes a response to the client within . The host's pipe is + /// unbuffered, so a write only completes when the client reads it; without the timeout a client that never reads + /// its response would hold the connection and its slots until the host stops. The timeout is a budget for the whole + /// response, so for a streamed body it includes the time the handler's stream takes to produce the data. When the + /// timeout expires, a warning is logged and the method returns, after which the caller closes the connection. + /// + /// + /// Writes the response, observing the it is given + /// + /// + /// The that is cancelled when the shutdown drain expires + /// + /// + /// An awaitable + /// + private async Task WriteResponseAsync(Func write, CancellationToken requestToken) + { + using var writeTimeoutCts = CancellationTokenSource.CreateLinkedTokenSource(requestToken); + + if (this.options.ResponseWriteTimeout != Timeout.InfiniteTimeSpan) + { + writeTimeoutCts.CancelAfter(this.options.ResponseWriteTimeout); + } + + try + { + await write(writeTimeoutCts.Token); + } + catch (Exception ex) when ((ex is OperationCanceledException || ex is IOException) && writeTimeoutCts.IsCancellationRequested && !requestToken.IsCancellationRequested) + { + this.logger.LogWarning( + ex, + "The response was not sent within {ResponseWriteTimeout}; closing the connection.", + this.options.ResponseWriteTimeout); + } + } + /// /// Gets or sets the delay before the first retry after accepting a connection failed. The delay doubles with every /// consecutive failure up to . Settable for tests. diff --git a/ArgusTransfer/Server/ArgusPipeHostOptions.cs b/ArgusTransfer/Server/ArgusPipeHostOptions.cs index fb84016..130e070 100644 --- a/ArgusTransfer/Server/ArgusPipeHostOptions.cs +++ b/ArgusTransfer/Server/ArgusPipeHostOptions.cs @@ -85,6 +85,17 @@ public class ArgusPipeHostOptions /// public TimeSpan RequestReadTimeout { get; set; } = TimeSpan.FromSeconds(30); + /// + /// Gets or sets the maximum time the host may take to send a complete response -- status line, headers and + /// body -- to the client. Defaults to 30 seconds. The pipe is unbuffered, so the host can only send as fast as the + /// client reads; when the timeout expires, the host logs a warning and closes the connection, so a client that + /// does not read its response, or reads it too slowly, cannot hold a connection indefinitely. The timeout is a + /// budget for the whole response and not an idle timeout: for a streamed body it also counts the time the + /// handler's stream takes to produce the data. Increase it, or disable it, on a host that streams large or + /// long-lived responses. Set to to disable the timeout. + /// + public TimeSpan ResponseWriteTimeout { get; set; } = TimeSpan.FromSeconds(30); + /// /// Gets or sets the maximum number of simultaneous client connections, including connections whose request is still /// being received. Defaults to 100. When the limit is reached, the host does not read new connections until one is diff --git a/CLAUDE.md b/CLAUDE.md index c981bc5..ad4becf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -101,7 +101,7 @@ Optional retry (`ArgusClient.RetryPolicy`, `null` = disabled) via `ArgusRetryPol Exception handling is layered: if routing throws (anything except `OperationCanceledException`, which maps to 503 on timeout), the host logs it and returns a generic 500 `ArgusProblemDetails` response (no exception details) instead of dropping the connection; dropping it would make clients with a retry policy re-execute the request. `ArgusExceptionHandlerMiddleware` customizes this inside the pipeline (`IncludeExceptionDetails` for development) and rethrows cancellation of `ArgusContext.RequestAborted` so timeouts still reach the host. Both build the response via the internal `ArgusExceptionHandlerMiddleware.CreateErrorResponse`. A pipeline that completes without setting `ArgusContext.Response` is logged as an error and also answered with the generic 500 (never a closed connection). -Connection limits: `ArgusPipeHostOptions.MaxConcurrentConnections` (default 100) is enforced by a connection semaphore acquired *after* the next pipe instance is created and *before* `WaitForConnectionAsync`, so at the limit new connections are not read (no 503 before the request is read, which would deadlock on the unbuffered pipe). The next instance must exist while waiting: on Linux/macOS (Unix domain sockets) the listening socket is shared by all instances of a pipe name and closes when the last one is disposed, which would reset clients queued in the backlog. Clients at the limit either wait in `ConnectAsync` or connect immediately (backlog on Unix; the ready instance on Windows) and wait to be read -- tests must not assume a connect timeout; `MaxConcurrentRequests` still limits processing. `RequestReadTimeout` (default 30 s) bounds receiving the complete request; on expiry the host logs a warning and closes the connection. `StopAsync` drains requests for `ShutdownDrainTimeout`, then cancels them and waits `CancellationGracePeriod` (internal, default 5 s); the drain token source and both semaphores are disposed only when every request task has finished -- otherwise disposal is skipped with a warning, so a handler that ignores cancellation can still release them. The accept loop (`AcceptConnectionAsync`) survives pipe errors: any exception other than cancellation on shutdown is logged and retried with exponential backoff (`AcceptRetryInitialDelay`, internal, default 1 s, doubling up to 30 s, reset after a successful accept), so a transient error never stops the host. +Connection limits: `ArgusPipeHostOptions.MaxConcurrentConnections` (default 100) is enforced by a connection semaphore acquired *after* the next pipe instance is created and *before* `WaitForConnectionAsync`, so at the limit new connections are not read (no 503 before the request is read, which would deadlock on the unbuffered pipe). The next instance must exist while waiting: on Linux/macOS (Unix domain sockets) the listening socket is shared by all instances of a pipe name and closes when the last one is disposed, which would reset clients queued in the backlog. Clients at the limit either wait in `ConnectAsync` or connect immediately (backlog on Unix; the ready instance on Windows) and wait to be read -- tests must not assume a connect timeout; `MaxConcurrentRequests` still limits processing. `RequestReadTimeout` (default 30 s) bounds receiving the complete request; on expiry the host logs a warning and closes the connection. `ResponseWriteTimeout` (default 30 s, `Timeout.InfiniteTimeSpan` disables) bounds sending the complete response (a total budget like `RequestReadTimeout`, not an idle timeout; for a streamed body it includes reading the handler's `BodyStream`): every response write in the connection task goes through `WriteResponseAsync`, which links a timeout to the drain token and on expiry logs a warning and returns so the connection is closed -- the pipe is unbuffered, so a client that never reads would otherwise hold the connection, a `MaxConcurrentRequests` slot and a `MaxConcurrentConnections` slot until shutdown. Never call `responseSerializer.WriteAsync` on the pipe directly. Tests for small responses that are not read are Windows-only (`[Platform("Win")]`): a Unix socket buffers them, so the write completes without a reader. The `Test (Windows)` job in `CodeQuality.yml` runs them in CI; Sonar analysis and coverage stay in the Linux job. `StopAsync` drains requests for `ShutdownDrainTimeout`, then cancels them and waits `CancellationGracePeriod` (internal, default 5 s); the drain token source and both semaphores are disposed only when every request task has finished -- otherwise disposal is skipped with a warning, so a handler that ignores cancellation can still release them. The accept loop (`AcceptConnectionAsync`) survives pipe errors: any exception other than cancellation on shutdown is logged and retried with exponential backoff (`AcceptRetryInitialDelay`, internal, default 1 s, doubling up to 30 s, reset after a successful accept), so a transient error never stops the host. Header limits: the serializers' `MaxHeaderSize` (default 32 KB; host sets it from `ArgusPipeHostOptions.MaxRequestHeaderSize`) bounds the request/status line plus all header lines via the internal `ArgusHeaderBudget`; `ArgusWireReader.ReadLineAsync(maxLength, …)` stops buffering as soon as a line exceeds its limit (`ArgusLineTooLongException`, an internal `ArgusProtocolException`). Chunk-size lines on the byte path are capped at 1 KB. The string reader is not limited.