diff --git a/lib/servus.akka b/lib/servus.akka index 8ae97f82..9e79e604 160000 --- a/lib/servus.akka +++ b/lib/servus.akka @@ -1 +1 @@ -Subproject commit 8ae97f8223e1f6cd323616aeac4bd5f3083bd3e0 +Subproject commit 9e79e604016b2eb31a6c56e65f05373ad4584915 diff --git a/src/GaudiHTTP.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt b/src/GaudiHTTP.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt index 28b4a943..d1ee6cc0 100644 --- a/src/GaudiHTTP.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt +++ b/src/GaudiHTTP.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt @@ -66,6 +66,7 @@ namespace GaudiHTTP.Client public System.Net.Security.RemoteCertificateValidationCallback? ServerCertificateValidationCallback { get; set; } public int? SocketReceiveBufferSize { get; set; } public int? SocketSendBufferSize { get; set; } + public double StreamRetryBackoffJitter { get; set; } public double StreamRetryBackoffMultiplier { get; set; } public System.TimeSpan StreamRetryInitialBackoff { get; set; } public System.TimeSpan StreamRetryMaxBackoff { get; set; } @@ -146,6 +147,10 @@ namespace GaudiHTTP.Client public int MaxResponseHeaderCount { get; set; } public int MaxResponseHeaderLineLength { get; set; } public int MaxResponseHeadersLength { get; set; } + public double ReconnectBackoffJitter { get; set; } + public double ReconnectBackoffMultiplier { get; set; } + public System.TimeSpan ReconnectInitialBackoff { get; set; } + public System.TimeSpan ReconnectMaxBackoff { get; set; } public int? RequestBodyChunkSize { get; set; } } public sealed class Http2ClientOptions @@ -167,6 +172,10 @@ namespace GaudiHTTP.Client public int MaxReconnectBufferSize { get; set; } public int MaxResponseHeaderListSize { get; set; } public int MaxStreamWindowSize { get; set; } + public double ReconnectBackoffJitter { get; set; } + public double ReconnectBackoffMultiplier { get; set; } + public System.TimeSpan ReconnectInitialBackoff { get; set; } + public System.TimeSpan ReconnectMaxBackoff { get; set; } public int? RequestBodyChunkSize { get; set; } public double WindowScaleThresholdMultiplier { get; set; } } @@ -184,6 +193,10 @@ namespace GaudiHTTP.Client public int MaxReconnectBufferSize { get; set; } public int QpackBlockedStreams { get; set; } public int QpackMaxTableCapacity { get; set; } + public double ReconnectBackoffJitter { get; set; } + public double ReconnectBackoffMultiplier { get; set; } + public System.TimeSpan ReconnectInitialBackoff { get; set; } + public System.TimeSpan ReconnectMaxBackoff { get; set; } public int? RequestBodyChunkSize { get; set; } } public interface IGaudiHttpClient : System.IDisposable diff --git a/src/GaudiHTTP.AcceptanceTests/H10/ResilienceSpec.cs b/src/GaudiHTTP.AcceptanceTests/H10/ResilienceSpec.cs index c49bea95..894f2f23 100644 --- a/src/GaudiHTTP.AcceptanceTests/H10/ResilienceSpec.cs +++ b/src/GaudiHTTP.AcceptanceTests/H10/ResilienceSpec.cs @@ -96,7 +96,8 @@ public async Task Resilience_should_fail_gracefully_on_corrupt_gzip() var response = await SendDecompressingAsync(request, (_, _) => responseBytes); Assert.Equal(HttpStatusCode.OK, response.StatusCode); - await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + await Assert.ThrowsAsync(async () => + await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken)); } [Fact(Timeout = 5000)] @@ -122,7 +123,8 @@ public async Task Resilience_should_fail_gracefully_on_corrupt_brotli() var response = await SendDecompressingAsync(request, (_, _) => responseBytes); Assert.Equal(HttpStatusCode.OK, response.StatusCode); - await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + await Assert.ThrowsAsync(async () => + await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken)); } [Fact(Timeout = 10000)] diff --git a/src/GaudiHTTP.AcceptanceTests/H11/ResilienceSpec.cs b/src/GaudiHTTP.AcceptanceTests/H11/ResilienceSpec.cs index b2e737c4..b15b2c76 100644 --- a/src/GaudiHTTP.AcceptanceTests/H11/ResilienceSpec.cs +++ b/src/GaudiHTTP.AcceptanceTests/H11/ResilienceSpec.cs @@ -101,7 +101,8 @@ public async Task Resilience_should_fail_gracefully_on_corrupt_gzip() var response = await SendDecompressingAsync(request, (_, _) => responseBytes); Assert.Equal(HttpStatusCode.OK, response.StatusCode); - await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + await Assert.ThrowsAsync(async () => + await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken)); } [Fact(Timeout = 5000)] @@ -127,7 +128,8 @@ public async Task Resilience_should_fail_gracefully_on_corrupt_brotli() var response = await SendDecompressingAsync(request, (_, _) => responseBytes); Assert.Equal(HttpStatusCode.OK, response.StatusCode); - await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + await Assert.ThrowsAsync(async () => + await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken)); } [Fact(Timeout = 5000)] diff --git a/src/GaudiHTTP.AcceptanceTests/ModuleInit.cs b/src/GaudiHTTP.AcceptanceTests/ModuleInit.cs deleted file mode 100644 index e340fa05..00000000 --- a/src/GaudiHTTP.AcceptanceTests/ModuleInit.cs +++ /dev/null @@ -1,13 +0,0 @@ -using System.Runtime.CompilerServices; -using Servus.Akka.Transport; - -namespace GaudiHTTP.AcceptanceTests; - -public static class ModuleInit -{ - [ModuleInitializer] - public static void Init() - { - WireBuffer.ConfigureWrapperPool(0); - } -} diff --git a/src/GaudiHTTP.AcceptanceTests/TLS/ResilienceSpec.cs b/src/GaudiHTTP.AcceptanceTests/TLS/ResilienceSpec.cs index 75cd94c1..e18d96fd 100644 --- a/src/GaudiHTTP.AcceptanceTests/TLS/ResilienceSpec.cs +++ b/src/GaudiHTTP.AcceptanceTests/TLS/ResilienceSpec.cs @@ -99,7 +99,8 @@ public async Task Resilience_should_fail_gracefully_on_corrupt_gzip_over_https() var response = await SendDecompressingAsync(request, (_, _) => responseBytes); Assert.Equal(HttpStatusCode.OK, response.StatusCode); - await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + await Assert.ThrowsAsync(async () => + await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken)); } [Fact(Timeout = 5000)] @@ -125,7 +126,8 @@ public async Task Resilience_should_fail_gracefully_on_corrupt_brotli_over_https var response = await SendDecompressingAsync(request, (_, _) => responseBytes); Assert.Equal(HttpStatusCode.OK, response.StatusCode); - await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken); + await Assert.ThrowsAsync(async () => + await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken)); } [Fact(Timeout = 5000)] diff --git a/src/GaudiHTTP.Tests.Shared/FakeClientOps.cs b/src/GaudiHTTP.Tests.Shared/FakeClientOps.cs index 3a5586f8..ac2e790f 100644 --- a/src/GaudiHTTP.Tests.Shared/FakeClientOps.cs +++ b/src/GaudiHTTP.Tests.Shared/FakeClientOps.cs @@ -9,6 +9,8 @@ internal sealed class FakeClientOps : IClientStageOperations public List Responses { get; } = []; public List Outbound { get; } = []; public List BodyMessages { get; } = []; + public List<(string Name, TimeSpan Duration)> ScheduledTimers { get; } = []; + public List CancelledTimers { get; } = []; public FakeClientOps() { @@ -20,10 +22,12 @@ public FakeClientOps() public void OnScheduleTimer(string name, TimeSpan duration) { + ScheduledTimers.Add((name, duration)); } public void OnCancelTimer(string name) { + CancelledTimers.Add(name); } public IActorRef StageActor { get; } diff --git a/src/GaudiHTTP.Tests/ModuleInit.cs b/src/GaudiHTTP.Tests/ModuleInit.cs deleted file mode 100644 index 9e0224f7..00000000 --- a/src/GaudiHTTP.Tests/ModuleInit.cs +++ /dev/null @@ -1,13 +0,0 @@ -using System.Runtime.CompilerServices; -using Servus.Akka.Transport; - -namespace GaudiHTTP.Tests; - -public static class ModuleInit -{ - [ModuleInitializer] - public static void Init() - { - WireBuffer.ConfigureWrapperPool(0); - } -} diff --git a/src/GaudiHTTP.Tests/Protocol/Semantics/Encoding/DecompressingContentEdgeCasesSpec.cs b/src/GaudiHTTP.Tests/Protocol/Semantics/Encoding/DecompressingContentEdgeCasesSpec.cs index 925bc663..dfed8da1 100644 --- a/src/GaudiHTTP.Tests/Protocol/Semantics/Encoding/DecompressingContentEdgeCasesSpec.cs +++ b/src/GaudiHTTP.Tests/Protocol/Semantics/Encoding/DecompressingContentEdgeCasesSpec.cs @@ -1,4 +1,5 @@ using System.IO.Compression; +using System.Net.Http; using GaudiHTTP.Internal; namespace GaudiHTTP.Tests.Protocol.Semantics.Encoding; @@ -36,29 +37,31 @@ public void SerializeToStream_should_decompress_deflate() } [Fact(Timeout = 5000)] - public async Task SerializeToStreamAsync_should_silently_handle_corrupt_gzip() + public async Task SerializeToStreamAsync_should_throw_on_corrupt_gzip() { var corrupt = new byte[] { 0x00, 0x01, 0x02, 0x03, 0xFF }; var inner = new ByteArrayContent(corrupt); using var content = new DecompressingContent(inner, "gzip"); using var ms = new MemoryStream(); - await content.CopyToAsync(ms, TestContext.Current.CancellationToken); - Assert.Equal(0, ms.Length); + // A corrupt/truncated compressed body must surface as an error, not silently truncate. + var ex = await Assert.ThrowsAsync(() => + content.CopyToAsync(ms, TestContext.Current.CancellationToken)); + Assert.NotNull(ex.InnerException); } [Fact(Timeout = 5000)] - public void SerializeToStream_should_silently_handle_corrupt_gzip() + public void SerializeToStream_should_throw_on_corrupt_gzip() { var corrupt = new byte[] { 0x00, 0x01, 0x02, 0x03, 0xFF }; var inner = new ByteArrayContent(corrupt); using var content = new DecompressingContent(inner, "gzip"); using var ms = new MemoryStream(); - content.CopyTo(ms, null, CancellationToken.None); - Assert.Equal(0, ms.Length); + var ex = Assert.Throws(() => content.CopyTo(ms, null, CancellationToken.None)); + Assert.NotNull(ex.InnerException); } [Fact(Timeout = 5000)] diff --git a/src/GaudiHTTP.Tests/Protocol/Semantics/ReconnectBackoffSpec.cs b/src/GaudiHTTP.Tests/Protocol/Semantics/ReconnectBackoffSpec.cs new file mode 100644 index 00000000..4a50f8b4 --- /dev/null +++ b/src/GaudiHTTP.Tests/Protocol/Semantics/ReconnectBackoffSpec.cs @@ -0,0 +1,53 @@ +using GaudiHTTP.Protocol.Semantics; + +namespace GaudiHTTP.Tests.Protocol.Semantics; + +public sealed class ReconnectBackoffSpec +{ + [Fact(Timeout = 5000)] + public void Compute_with_zero_jitter_should_grow_geometrically() + { + var initial = TimeSpan.FromMilliseconds(100); + var max = TimeSpan.FromSeconds(30); + + var r1 = ReconnectBackoff.Compute(1, initial, max, 2.0, 0.0, Random.Shared); + var r2 = ReconnectBackoff.Compute(2, initial, max, 2.0, 0.0, Random.Shared); + var r3 = ReconnectBackoff.Compute(3, initial, max, 2.0, 0.0, Random.Shared); + + Assert.Equal(100, r1.TotalMilliseconds, 3); + Assert.Equal(200, r2.TotalMilliseconds, 3); + Assert.Equal(400, r3.TotalMilliseconds, 3); + } + + [Fact(Timeout = 5000)] + public void Compute_should_cap_at_max() + { + var max = TimeSpan.FromSeconds(5); + + var delay = ReconnectBackoff.Compute(20, TimeSpan.FromMilliseconds(100), max, 2.0, 0.0, Random.Shared); + + Assert.Equal(max.TotalMilliseconds, delay.TotalMilliseconds, 3); + } + + [Fact(Timeout = 5000)] + public void Compute_should_apply_symmetric_jitter_within_bounds() + { + var initial = TimeSpan.FromMilliseconds(1000); + var max = TimeSpan.FromSeconds(30); + + for (var i = 0; i < 200; i++) + { + var delay = ReconnectBackoff.Compute(1, initial, max, 2.0, 0.2, Random.Shared).TotalMilliseconds; + Assert.InRange(delay, 800, 1200); + } + } + + [Fact(Timeout = 5000)] + public void Compute_should_never_return_below_one_millisecond() + { + var delay = ReconnectBackoff.Compute(1, TimeSpan.FromMilliseconds(0.5), + TimeSpan.FromSeconds(1), 2.0, 0.2, Random.Shared); + + Assert.True(delay >= TimeSpan.FromMilliseconds(1), $"expected >= 1ms, was {delay.TotalMilliseconds}ms"); + } +} diff --git a/src/GaudiHTTP.Tests/Protocol/Semantics/ReconnectPolicySpec.cs b/src/GaudiHTTP.Tests/Protocol/Semantics/ReconnectPolicySpec.cs index 87a0a1d0..d98e0527 100644 --- a/src/GaudiHTTP.Tests/Protocol/Semantics/ReconnectPolicySpec.cs +++ b/src/GaudiHTTP.Tests/Protocol/Semantics/ReconnectPolicySpec.cs @@ -71,6 +71,67 @@ public void OnAttemptFailed_should_exhaust_and_disconnect_once_max_attempts_is_r Assert.IsType(ops.Outbound[^1]); } + [Fact(Timeout = 5000)] + public void OnAttemptFailed_should_defer_the_retry_behind_a_backoff_timer_when_backoff_is_configured() + { + var ops = new FakeClientOps(); + var policy = new ReconnectPolicy(ops, maxAttempts: 3, + initialBackoff: TimeSpan.FromMilliseconds(100), maxBackoff: TimeSpan.FromSeconds(5)); + policy.Start("buffered-request", MakeTransportOptions()); + ops.Outbound.Clear(); + + var exhausted = policy.OnAttemptFailed(MakeTransportOptions(), out _); + + Assert.False(exhausted); + // The retry must NOT connect immediately — that is the zero-delay busy-loop we are fixing. + Assert.DoesNotContain(ops.Outbound, o => o is ConnectTransport); + var timer = Assert.Single(ops.ScheduledTimers); + Assert.Equal(ReconnectPolicy.BackoffTimerName, timer.Name); + Assert.True(timer.Duration > TimeSpan.Zero); + Assert.True(timer.Duration <= TimeSpan.FromSeconds(5)); + } + + [Fact(Timeout = 5000)] + public void OnReconnectTimerFired_should_emit_the_connect_transport_for_its_own_timer() + { + var ops = new FakeClientOps(); + var policy = new ReconnectPolicy(ops, maxAttempts: 3, + initialBackoff: TimeSpan.FromMilliseconds(100), maxBackoff: TimeSpan.FromSeconds(5)); + policy.Start("buffered-request", MakeTransportOptions()); + policy.OnAttemptFailed(MakeTransportOptions(), out _); + ops.Outbound.Clear(); + + var handled = policy.OnReconnectTimerFired(ReconnectPolicy.BackoffTimerName); + + Assert.True(handled); + Assert.Single(ops.Outbound, o => o is ConnectTransport); + } + + [Fact(Timeout = 5000)] + public void OnReconnectTimerFired_should_ignore_unrelated_timer_names() + { + var ops = new FakeClientOps(); + var policy = new ReconnectPolicy(ops, maxAttempts: 3, + initialBackoff: TimeSpan.FromMilliseconds(100), maxBackoff: TimeSpan.FromSeconds(5)); + + Assert.False(policy.OnReconnectTimerFired("keep-alive-timeout")); + Assert.Empty(ops.Outbound); + } + + [Fact(Timeout = 5000)] + public void OnAttemptFailed_without_backoff_should_connect_immediately() + { + var ops = new FakeClientOps(); + var policy = new ReconnectPolicy(ops, maxAttempts: 3); + policy.Start("buffered-request", MakeTransportOptions()); + ops.Outbound.Clear(); + + policy.OnAttemptFailed(MakeTransportOptions(), out _); + + Assert.Single(ops.Outbound, o => o is ConnectTransport); + Assert.Empty(ops.ScheduledTimers); + } + [Fact(Timeout = 5000)] public void CanReconnect_should_reflect_whether_max_attempts_is_positive() { diff --git a/src/GaudiHTTP.Tests/Protocol/Syntax/Http11/Client/Http11StateMachineReconnectSpec.cs b/src/GaudiHTTP.Tests/Protocol/Syntax/Http11/Client/Http11StateMachineReconnectSpec.cs index eb3a4d89..2df7ff0a 100644 --- a/src/GaudiHTTP.Tests/Protocol/Syntax/Http11/Client/Http11StateMachineReconnectSpec.cs +++ b/src/GaudiHTTP.Tests/Protocol/Syntax/Http11/Client/Http11StateMachineReconnectSpec.cs @@ -28,6 +28,20 @@ private static (HttpRequestMessage Request, PendingRequest Pending) MakeTrackedR return (request, pending); } + private static (HttpRequestMessage Request, PendingRequest Pending) MakeTrackedRequest( + HttpMethod method, string path = "/") + { + var pending = PendingRequest.Rent(); + var version = pending.Version; + var request = new HttpRequestMessage(method, $"http://example.com{path}") + { + Version = new Version(1, 1) + }; + request.Options.Set(OptionsKey.Key, pending); + request.Options.Set(OptionsKey.VersionKey, version); + return (request, pending); + } + private static readonly ConnectionInfo DummyConnectionInfo = new( new IPEndPoint(IPAddress.Loopback, 5000), new IPEndPoint(IPAddress.Loopback, 80), @@ -101,9 +115,82 @@ public void DecodeServerData_should_fail_requests_when_max_reconnect_attempts_ex Assert.Contains(ops.Outbound, o => o is DisconnectTransport); } + [Fact(Timeout = 5000)] + public void Cleanup_should_fail_inflight_requests_instead_of_dropping_them() + { + var ops = new FakeClientOps(); + var sm = new Http11ClientStateMachine(TestClientOptions.Create(maxPipelineDepth: 4), ops); + var (req, pending) = MakeTrackedRequest(HttpMethod.Get, "/a"); + sm.OnRequest(req); + + // Stage teardown (KillSwitch abort → PostStop → Cleanup) must FAIL in-flight requests, not + // silently drop them — otherwise the caller hangs until its client-side timeout. + sm.Cleanup(); + + Assert.True(pending.GetValueTask().IsFaulted); + } + + [Fact(Timeout = 5000)] + public void Cleanup_during_reconnect_should_fail_buffered_replay_requests() + { + var ops = new FakeClientOps(); + var sm = new Http11ClientStateMachine(TestClientOptions.Create(maxPipelineDepth: 4, http1MaxReconnectAttempts: 3), ops); + var (req, pending) = MakeTrackedRequest(HttpMethod.Get, "/a"); + sm.OnRequest(req); + + // Move the request into the reconnect replay buffer, then tear the stage down. + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + sm.Cleanup(); + + Assert.True(pending.GetValueTask().IsFaulted); + } + + [Fact(Timeout = 5000)] + [Trait("RFC", "RFC9110-9.2.2")] + public void DecodeServerData_should_not_replay_non_idempotent_request_on_reconnect() + { + var ops = new FakeClientOps(); + var sm = new Http11ClientStateMachine(TestClientOptions.Create(maxPipelineDepth: 4, http1MaxReconnectAttempts: 3), ops); + var (post, postPending) = MakeTrackedRequest(HttpMethod.Post, "/submit"); + sm.OnRequest(post); + ops.Outbound.Clear(); + + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + + // The server may already have processed the POST; replaying it risks a duplicate. + Assert.True(postPending.GetValueTask().IsFaulted); + + // On restore there is nothing safe to replay — no request bytes go back out. + sm.DecodeServerData(new TransportConnected(DummyConnectionInfo)); + Assert.Empty(ops.Outbound.OfType()); + } + + [Fact(Timeout = 5000)] + [Trait("RFC", "RFC9110-9.2.2")] + public void DecodeServerData_should_replay_idempotent_but_fail_non_idempotent_on_reconnect() + { + var ops = new FakeClientOps(); + var sm = new Http11ClientStateMachine(TestClientOptions.Create(maxPipelineDepth: 4, http1MaxReconnectAttempts: 3), ops); + var (get, getPending) = MakeTrackedRequest(HttpMethod.Get, "/a"); + var (post, postPending) = MakeTrackedRequest(HttpMethod.Post, "/b"); + sm.OnRequest(get); + sm.OnRequest(post); + ops.Outbound.Clear(); + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + + Assert.True(postPending.GetValueTask().IsFaulted); + Assert.False(getPending.GetValueTask().IsFaulted); + + ops.Outbound.Clear(); + sm.DecodeServerData(new TransportConnected(DummyConnectionInfo)); + + // Only the idempotent GET is replayed. + Assert.Single(ops.Outbound.OfType()); + } + [Fact(Timeout = 5000)] [Trait("RFC", "RFC9112-9.3")] - public void DecodeServerData_should_emit_new_connect_when_reconnect_attempt_under_limit() + public void DecodeServerData_should_defer_retry_behind_backoff_then_connect_when_timer_fires() { var ops = new FakeClientOps(); var sm = new Http11ClientStateMachine(TestClientOptions.Create(maxPipelineDepth: 4, http1MaxReconnectAttempts: 3), ops); @@ -112,9 +199,34 @@ public void DecodeServerData_should_emit_new_connect_when_reconnect_attempt_unde sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); var countAfterFirst = ops.Outbound.OfType().Count(); + // Second failure schedules a backoff timer instead of reconnecting immediately. + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + + Assert.True(sm.IsReconnecting); + Assert.Equal(countAfterFirst, ops.Outbound.OfType().Count()); + Assert.Contains(ops.ScheduledTimers, t => t.Name == "reconnect-backoff"); + + // Firing the timer performs the actual reconnect. + sm.OnTimerFired("reconnect-backoff"); + Assert.Equal(countAfterFirst + 1, ops.Outbound.OfType().Count()); + } + + [Fact(Timeout = 5000)] + [Trait("RFC", "RFC9112-9.3")] + public void DecodeServerData_with_zero_backoff_should_connect_immediately_on_retry() + { + var ops = new FakeClientOps(); + var sm = new Http11ClientStateMachine( + TestClientOptions.Create(maxPipelineDepth: 4, http1MaxReconnectAttempts: 3, http1ReconnectInitialBackoff: TimeSpan.Zero), ops); + sm.OnRequest(MakeRequest()); + + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + var countAfterFirst = ops.Outbound.OfType().Count(); + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); Assert.True(sm.IsReconnecting); Assert.Equal(countAfterFirst + 1, ops.Outbound.OfType().Count()); + Assert.DoesNotContain(ops.ScheduledTimers, t => t.Name == "reconnect-backoff"); } } \ No newline at end of file diff --git a/src/GaudiHTTP.Tests/Protocol/Syntax/Http2/Client/StateMachine/Http2StateMachineReconnectSpec.cs b/src/GaudiHTTP.Tests/Protocol/Syntax/Http2/Client/StateMachine/Http2StateMachineReconnectSpec.cs index 37ef1942..83420d87 100644 --- a/src/GaudiHTTP.Tests/Protocol/Syntax/Http2/Client/StateMachine/Http2StateMachineReconnectSpec.cs +++ b/src/GaudiHTTP.Tests/Protocol/Syntax/Http2/Client/StateMachine/Http2StateMachineReconnectSpec.cs @@ -52,6 +52,36 @@ private static (HttpRequestMessage Request, PendingRequest Pending) MakeTrackedP new IPEndPoint(IPAddress.Loopback, 443), TransportProtocol.Tcp); + [Fact(Timeout = 5000)] + public void Cleanup_should_fail_inflight_requests_instead_of_dropping_them() + { + var ops = new FakeClientOps(); + var sm = new Http2ClientStateMachine(TestClientOptions.Create(), ops); + sm.PreStart(); + var (req, pending) = MakeTrackedGet("/a"); + sm.OnRequest(req); + + // Stage teardown must FAIL a stream still awaiting its response, not silently drop it. + sm.Cleanup(); + + Assert.True(pending.GetValueTask().IsFaulted); + } + + [Fact(Timeout = 5000)] + public void Cleanup_during_reconnect_should_fail_buffered_replay_requests() + { + var ops = new FakeClientOps(); + var sm = new Http2ClientStateMachine(TestClientOptions.Create(), ops); + sm.PreStart(); + var (req, pending) = MakeTrackedGet("/a"); + sm.OnRequest(req); + + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); // buffers GET for replay + sm.Cleanup(); + + Assert.True(pending.GetValueTask().IsFaulted); + } + [Fact(Timeout = 5000)] [Trait("RFC", "RFC9113-6.8")] public void DecodeServerData_should_start_reconnect_on_disconnect_with_inflight() @@ -142,7 +172,7 @@ public void DecodeServerData_should_fail_when_max_reconnect_exceeded() [Fact(Timeout = 5000)] [Trait("RFC", "RFC9113-6.8")] - public void DecodeServerData_should_emit_new_connect_when_reconnect_under_limit() + public void DecodeServerData_should_defer_retry_behind_backoff_then_connect_when_timer_fires() { var ops = new FakeClientOps(); var sm = new Http2ClientStateMachine(TestClientOptions.Create(http2MaxReconnectAttempts: 3), ops); @@ -152,10 +182,35 @@ public void DecodeServerData_should_emit_new_connect_when_reconnect_under_limit( sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); var countAfterFirst = ops.Outbound.OfType().Count(); + // Second failure schedules a backoff timer instead of reconnecting immediately. + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + + Assert.True(sm.IsReconnecting); + Assert.Equal(countAfterFirst, ops.Outbound.OfType().Count()); + Assert.Contains(ops.ScheduledTimers, t => t.Name == "reconnect-backoff"); + + sm.OnTimerFired("reconnect-backoff"); + Assert.Equal(countAfterFirst + 1, ops.Outbound.OfType().Count()); + } + + [Fact(Timeout = 5000)] + [Trait("RFC", "RFC9113-6.8")] + public void DecodeServerData_with_zero_backoff_should_connect_immediately_on_retry() + { + var ops = new FakeClientOps(); + var sm = new Http2ClientStateMachine( + TestClientOptions.Create(http2MaxReconnectAttempts: 3, http2ReconnectInitialBackoff: TimeSpan.Zero), ops); + sm.PreStart(); + sm.OnRequest(MakeGet()); + + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + var countAfterFirst = ops.Outbound.OfType().Count(); + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); Assert.True(sm.IsReconnecting); Assert.Equal(countAfterFirst + 1, ops.Outbound.OfType().Count()); + Assert.DoesNotContain(ops.ScheduledTimers, t => t.Name == "reconnect-backoff"); } [Fact(Timeout = 5000)] diff --git a/src/GaudiHTTP.Tests/Protocol/Syntax/Http3/Client/StateMachine/Http3ClientConnectionErrorSpec.cs b/src/GaudiHTTP.Tests/Protocol/Syntax/Http3/Client/StateMachine/Http3ClientConnectionErrorSpec.cs index aca5335e..00ba5ac3 100644 --- a/src/GaudiHTTP.Tests/Protocol/Syntax/Http3/Client/StateMachine/Http3ClientConnectionErrorSpec.cs +++ b/src/GaudiHTTP.Tests/Protocol/Syntax/Http3/Client/StateMachine/Http3ClientConnectionErrorSpec.cs @@ -1,5 +1,6 @@ using Servus.Akka.Transport; using GaudiHTTP.Client; +using GaudiHTTP.Internal; using GaudiHTTP.Protocol.Syntax.Http3; using GaudiHTTP.Protocol.Syntax.Http3.Client; using GaudiHTTP.Tests.Shared; @@ -10,6 +11,15 @@ public sealed class Http3ClientConnectionErrorSpec { private readonly FakeClientOps _clientOps = new(); + private static (HttpRequestMessage Request, PendingRequest Pending) MakeTrackedGet(string path = "/") + { + var pending = PendingRequest.Rent(); + var req = new HttpRequestMessage(HttpMethod.Get, $"https://example.com{path}") { Version = new Version(3, 0) }; + req.Options.Set(OptionsKey.Key, pending); + req.Options.Set(OptionsKey.VersionKey, pending.Version); + return (req, pending); + } + private Http3ClientStateMachine CreateMachine() { var sm = new Http3ClientStateMachine(new GaudiClientOptions(), _clientOps); @@ -67,6 +77,74 @@ public void Connection_failure_as_stream_errors_then_disconnect_should_reconnect Assert.Single(_clientOps.Outbound, o => o is ConnectTransport); } + [Fact(Timeout = 5000)] + [Trait("RFC", "RFC9114-5.2")] + public void Reconnect_retry_should_defer_behind_backoff_then_connect_when_timer_fires() + { + var sm = CreateMachine(); + sm.OnRequest(new HttpRequestMessage(HttpMethod.Get, "https://example.com/") { Version = new Version(3, 0) }); + _clientOps.Outbound.Clear(); + + // First disconnect starts the reconnect (immediate first attempt). + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + var countAfterFirst = _clientOps.Outbound.OfType().Count(); + + // Second disconnect fails the attempt → deferred behind a backoff timer, not connected instantly. + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + + Assert.True(sm.IsReconnecting); + Assert.Equal(countAfterFirst, _clientOps.Outbound.OfType().Count()); + Assert.Contains(_clientOps.ScheduledTimers, t => t.Name == "reconnect-backoff"); + + sm.OnTimerFired("reconnect-backoff"); + Assert.Equal(countAfterFirst + 1, _clientOps.Outbound.OfType().Count()); + } + + [Fact(Timeout = 5000)] + [Trait("RFC", "RFC9114-5.2")] + public void Reconnect_retry_with_zero_backoff_should_connect_immediately() + { + var ops = new FakeClientOps(); + var sm = new Http3ClientStateMachine( + new GaudiClientOptions { Http3 = { ReconnectInitialBackoff = TimeSpan.Zero } }, ops); + sm.PreStart(); + sm.DecodeServerData(new TransportConnected(null!)); + sm.OnRequest(new HttpRequestMessage(HttpMethod.Get, "https://example.com/") { Version = new Version(3, 0) }); + ops.Outbound.Clear(); + + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + var countAfterFirst = ops.Outbound.OfType().Count(); + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); + + Assert.Equal(countAfterFirst + 1, ops.Outbound.OfType().Count()); + Assert.DoesNotContain(ops.ScheduledTimers, t => t.Name == "reconnect-backoff"); + } + + [Fact(Timeout = 5000)] + public void Cleanup_should_fail_inflight_requests_instead_of_dropping_them() + { + var sm = CreateMachine(); + var (req, pending) = MakeTrackedGet("/a"); + sm.OnRequest(req); + + sm.Cleanup(); + + Assert.True(pending.GetValueTask().IsFaulted); + } + + [Fact(Timeout = 5000)] + public void Cleanup_during_reconnect_should_fail_buffered_replay_requests() + { + var sm = CreateMachine(); + var (req, pending) = MakeTrackedGet("/a"); + sm.OnRequest(req); + + sm.DecodeServerData(new TransportDisconnected(DisconnectReason.Error)); // buffers GET for replay + sm.Cleanup(); + + Assert.True(pending.GetValueTask().IsFaulted); + } + [Fact(Timeout = 5000)] [Trait("RFC", "RFC9114-7.2.4")] public void Second_settings_frame_on_control_stream_should_disconnect() diff --git a/src/GaudiHTTP.Tests/Streams/Stages/Server/ApplicationBridgeStageSpec.cs b/src/GaudiHTTP.Tests/Streams/Stages/Server/ApplicationBridgeStageSpec.cs index 803e6a23..96c423f0 100644 --- a/src/GaudiHTTP.Tests/Streams/Stages/Server/ApplicationBridgeStageSpec.cs +++ b/src/GaudiHTTP.Tests/Streams/Stages/Server/ApplicationBridgeStageSpec.cs @@ -14,12 +14,13 @@ public sealed class ApplicationBridgeStageSpec : StreamTestBase private sealed class FakeApplication(Func handler) : IHttpApplication { + public Action? OnDispose { get; init; } + public IFeatureCollection CreateContext(IFeatureCollection contextFeatures) => contextFeatures; public Task ProcessRequestAsync(IFeatureCollection context) => handler(context); public void DisposeContext(IFeatureCollection context, Exception? exception) - { - } + => OnDispose?.Invoke(context, exception); } private static IFeatureCollection Request(string protocol = "HTTP/2") @@ -135,6 +136,31 @@ public void ApplicationBridgeStage_should_handle_handler_exceptions() Assert.Equal(500, result.Get()?.StatusCode); } + [Fact(Timeout = 5000)] + public void ApplicationBridgeStage_should_surface_handler_exception_to_context_disposal() + { + Exception? captured = null; + var app = new FakeApplication(_ => throw new InvalidOperationException("handler boom")) + { + OnDispose = (_, ex) => captured = ex + }; + var stage = CreateStage(app); + + var (upstream, downstream) = this.SourceProbe() + .Via(stage) + .ToMaterialized(this.SinkProbe(), Keep.Both) + .Run(Materializer); + + downstream.Request(1); + upstream.SendNext(Request(), TestContext.Current.CancellationToken); + + var result = downstream.ExpectNext(TestContext.Current.CancellationToken); + Assert.Equal(500, result.Get()?.StatusCode); + + // The handler crash must not vanish — it reaches context disposal so operators can observe it. + Assert.IsType(captured); + } + [Fact(Timeout = 5000)] public void ApplicationBridgeStage_should_complete_upstream_finished_no_pending() { diff --git a/src/GaudiHTTP.Tests/TestSupport/TestClientOptions.cs b/src/GaudiHTTP.Tests/TestSupport/TestClientOptions.cs index ae0a1069..24286484 100644 --- a/src/GaudiHTTP.Tests/TestSupport/TestClientOptions.cs +++ b/src/GaudiHTTP.Tests/TestSupport/TestClientOptions.cs @@ -7,8 +7,10 @@ internal static class TestClientOptions internal static GaudiClientOptions Create( int? maxPipelineDepth = null, int? http1MaxReconnectAttempts = null, + TimeSpan? http1ReconnectInitialBackoff = null, int? maxConcurrentStreams = null, int? http2MaxReconnectAttempts = null, + TimeSpan? http2ReconnectInitialBackoff = null, int? initialStreamWindowSize = null, int? maxFrameSize = null, TimeSpan? keepAlivePingDelay = null, @@ -26,6 +28,11 @@ internal static GaudiClientOptions Create( options.Http1.MaxReconnectAttempts = http1MaxReconnectAttempts.Value; } + if (http1ReconnectInitialBackoff.HasValue) + { + options.Http1.ReconnectInitialBackoff = http1ReconnectInitialBackoff.Value; + } + if (maxConcurrentStreams.HasValue) { options.Http2.MaxConcurrentStreams = maxConcurrentStreams.Value; @@ -36,6 +43,11 @@ internal static GaudiClientOptions Create( options.Http2.MaxReconnectAttempts = http2MaxReconnectAttempts.Value; } + if (http2ReconnectInitialBackoff.HasValue) + { + options.Http2.ReconnectInitialBackoff = http2ReconnectInitialBackoff.Value; + } + if (initialStreamWindowSize.HasValue) { options.Http2.InitialStreamWindowSize = initialStreamWindowSize.Value; diff --git a/src/GaudiHTTP/Client/GaudiClientOptions.cs b/src/GaudiHTTP/Client/GaudiClientOptions.cs index 7bdaaf60..c5ed9af4 100644 --- a/src/GaudiHTTP/Client/GaudiClientOptions.cs +++ b/src/GaudiHTTP/Client/GaudiClientOptions.cs @@ -135,6 +135,12 @@ public sealed class GaudiClientOptions /// public double StreamRetryBackoffMultiplier { get; set; } = 2.0; + /// + /// Fractional jitter (0..1) applied symmetrically to each stream-materialization retry delay to + /// avoid a thundering herd of correlated re-materializations across clients. Default is 0.2 (±20%). + /// + public double StreamRetryBackoffJitter { get; set; } = 0.2; + /// /// Maximum number of stream materialization retry attempts before the client gives up and /// reports a failure. Default is 10. diff --git a/src/GaudiHTTP/Client/Http1ClientOptions.cs b/src/GaudiHTTP/Client/Http1ClientOptions.cs index 42e1e9fc..e6913664 100644 --- a/src/GaudiHTTP/Client/Http1ClientOptions.cs +++ b/src/GaudiHTTP/Client/Http1ClientOptions.cs @@ -71,6 +71,30 @@ public sealed class Http1ClientOptions /// public int MaxReconnectAttempts { get; set; } = 3; + /// + /// Delay before the first reconnect retry after a failed attempt. Successive retries grow by + /// up to , with + /// applied. Spacing retries avoids a tight reconnect + /// busy-loop against a peer that refuses connections instantly. Default is 100 ms. + /// + public TimeSpan ReconnectInitialBackoff { get; set; } = TimeSpan.FromMilliseconds(100); + + /// + /// Upper bound on the exponential reconnect backoff delay. Default is 5 seconds. + /// + public TimeSpan ReconnectMaxBackoff { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Growth factor applied to the reconnect backoff on each successive retry. Default is 2.0. + /// + public double ReconnectBackoffMultiplier { get; set; } = 2.0; + + /// + /// Fractional jitter (0..1) applied symmetrically to each reconnect backoff delay to avoid a + /// thundering herd of correlated reconnects. Default is 0.2 (±20%). + /// + public double ReconnectBackoffJitter { get; set; } = 0.2; + /// /// Maximum number of header fields accepted in an HTTP/1.x response. /// Guards against malicious servers flooding the client with header lines. Default is 100 diff --git a/src/GaudiHTTP/Client/Http2ClientOptions.cs b/src/GaudiHTTP/Client/Http2ClientOptions.cs index 4b4312f4..996296b8 100644 --- a/src/GaudiHTTP/Client/Http2ClientOptions.cs +++ b/src/GaudiHTTP/Client/Http2ClientOptions.cs @@ -119,6 +119,30 @@ public sealed class Http2ClientOptions /// public int MaxReconnectBufferSize { get; set; } = 64; + /// + /// Delay before the first reconnect retry after a failed attempt. Successive retries grow by + /// up to , with + /// applied. Spacing retries avoids a tight reconnect + /// busy-loop against a peer that refuses connections instantly. Default is 100 ms. + /// + public TimeSpan ReconnectInitialBackoff { get; set; } = TimeSpan.FromMilliseconds(100); + + /// + /// Upper bound on the exponential reconnect backoff delay. Default is 5 seconds. + /// + public TimeSpan ReconnectMaxBackoff { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Growth factor applied to the reconnect backoff on each successive retry. Default is 2.0. + /// + public double ReconnectBackoffMultiplier { get; set; } = 2.0; + + /// + /// Fractional jitter (0..1) applied symmetrically to each reconnect backoff delay to avoid a + /// thundering herd of correlated reconnects. Default is 0.2 (±20%). + /// + public double ReconnectBackoffJitter { get; set; } = 0.2; + /// /// Delay before sending a keep-alive PING frame when no frames have been received. /// Set to to disable keep-alive pings (default). diff --git a/src/GaudiHTTP/Client/Http3ClientOptions.cs b/src/GaudiHTTP/Client/Http3ClientOptions.cs index eb22a64a..69d2d77b 100644 --- a/src/GaudiHTTP/Client/Http3ClientOptions.cs +++ b/src/GaudiHTTP/Client/Http3ClientOptions.cs @@ -92,4 +92,28 @@ public sealed class Http3ClientOptions /// Default is 64. /// public int MaxReconnectBufferSize { get; set; } = 64; + + /// + /// Delay before the first reconnect retry after a failed attempt. Successive retries grow by + /// up to , with + /// applied. Spacing retries avoids a tight reconnect + /// busy-loop against a peer that refuses connections instantly. Default is 100 ms. + /// + public TimeSpan ReconnectInitialBackoff { get; set; } = TimeSpan.FromMilliseconds(100); + + /// + /// Upper bound on the exponential reconnect backoff delay. Default is 5 seconds. + /// + public TimeSpan ReconnectMaxBackoff { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Growth factor applied to the reconnect backoff on each successive retry. Default is 2.0. + /// + public double ReconnectBackoffMultiplier { get; set; } = 2.0; + + /// + /// Fractional jitter (0..1) applied symmetrically to each reconnect backoff delay to avoid a + /// thundering herd of correlated reconnects. Default is 0.2 (±20%). + /// + public double ReconnectBackoffJitter { get; set; } = 0.2; } \ No newline at end of file diff --git a/src/GaudiHTTP/Internal/DecompressingContent.cs b/src/GaudiHTTP/Internal/DecompressingContent.cs index 3f7731f5..cb15da0f 100644 --- a/src/GaudiHTTP/Internal/DecompressingContent.cs +++ b/src/GaudiHTTP/Internal/DecompressingContent.cs @@ -18,6 +18,7 @@ protected override void SerializeToStream(Stream stream, TransportContext? conte } catch (Exception ex) when (ex is InvalidDataException or InvalidOperationException or Protocol.HttpProtocolException) { + throw DecodeFailure(ex); } } @@ -32,6 +33,7 @@ protected override async Task SerializeToStreamAsync(Stream stream, TransportCon } catch (Exception ex) when (ex is InvalidDataException or InvalidOperationException or Protocol.HttpProtocolException) { + throw DecodeFailure(ex); } } @@ -46,9 +48,17 @@ protected override async Task SerializeToStreamAsync(Stream stream, TransportCon } catch (Exception ex) when (ex is InvalidDataException or InvalidOperationException or Protocol.HttpProtocolException) { + throw DecodeFailure(ex); } } + // HttpRequestException (not IOException/HttpIOException) so HttpContent's stream-copy wrapping + // passes it through unwrapped — the caller sees a single, clear content-decode failure. + private HttpRequestException DecodeFailure(Exception inner) + => new( + $"Failed to decode the response body (Content-Encoding: {encoding}); the compressed data is corrupt or truncated.", + inner); + protected override bool TryComputeLength(out long length) { length = 0; diff --git a/src/GaudiHTTP/Protocol/Multiplexed/ReconnectionManager.cs b/src/GaudiHTTP/Protocol/Multiplexed/ReconnectionManager.cs index ea70bc0a..67ac660c 100644 --- a/src/GaudiHTTP/Protocol/Multiplexed/ReconnectionManager.cs +++ b/src/GaudiHTTP/Protocol/Multiplexed/ReconnectionManager.cs @@ -8,6 +8,9 @@ internal sealed class ReconnectionManager(int maxAttempts, int maxBufferSize = i public bool IsReconnecting { get; private set; } public int BufferedCount => _buffer.Count; + /// Number of reconnect attempts so far in the current reconnect sequence (1-based). + public int Attempts => _attempts; + public void OnConnectionLost(IReadOnlyList replayableRequests) { IsReconnecting = true; diff --git a/src/GaudiHTTP/Protocol/Semantics/ReconnectBackoff.cs b/src/GaudiHTTP/Protocol/Semantics/ReconnectBackoff.cs new file mode 100644 index 00000000..b676e929 --- /dev/null +++ b/src/GaudiHTTP/Protocol/Semantics/ReconnectBackoff.cs @@ -0,0 +1,41 @@ +namespace GaudiHTTP.Protocol.Semantics; + +/// +/// Shared exponential-backoff-with-jitter calculation for client reconnect retries, used by the +/// HTTP/1.x and the HTTP/2 & HTTP/3 state machines so all +/// protocols space their reconnect attempts identically. Spacing retries (rather than reconnecting +/// with zero delay) avoids a tight busy-loop against a peer that refuses connections instantly, and +/// the jitter avoids a thundering herd of correlated reconnects across connections. +/// +internal static class ReconnectBackoff +{ + /// Timer name under which a deferred reconnect attempt is scheduled. + public const string TimerName = "reconnect-backoff"; + + /// + /// Computes the delay before the -th reconnect retry (1-based: + /// 1 = the first retry after the initial immediate attempt). The delay grows geometrically by + /// from , is capped at + /// , then has symmetric applied. Never returns + /// less than 1 ms so progress is always made. + /// + public static TimeSpan Compute( + int retryNumber, + TimeSpan initial, + TimeSpan max, + double multiplier, + double jitter, + Random rng) + { + var cap = max > TimeSpan.Zero ? max : initial; + var exponent = Math.Max(0, retryNumber - 1); + var growth = multiplier <= 0 ? 1.0 : multiplier; + var scaled = initial.TotalMilliseconds * Math.Pow(growth, exponent); + var capped = Math.Min(scaled, cap.TotalMilliseconds); + + var clampedJitter = Math.Clamp(jitter, 0.0, 1.0); + var factor = 1.0 + clampedJitter * (2.0 * rng.NextDouble() - 1.0); + var jittered = Math.Min(capped * factor, cap.TotalMilliseconds); + return TimeSpan.FromMilliseconds(Math.Max(1.0, jittered)); + } +} diff --git a/src/GaudiHTTP/Protocol/Semantics/ReconnectPolicy.cs b/src/GaudiHTTP/Protocol/Semantics/ReconnectPolicy.cs index 96ee7bd9..3d4c145d 100644 --- a/src/GaudiHTTP/Protocol/Semantics/ReconnectPolicy.cs +++ b/src/GaudiHTTP/Protocol/Semantics/ReconnectPolicy.cs @@ -12,15 +12,44 @@ namespace GaudiHTTP.Protocol.Semantics; /// internal sealed class ReconnectPolicy { + /// Timer name used to defer a reconnect attempt behind an exponential backoff delay. + public const string BackoffTimerName = ReconnectBackoff.TimerName; + private readonly IClientStageOperations _ops; private readonly int _maxAttempts; + private readonly TimeSpan _initialBackoff; + private readonly TimeSpan _maxBackoff; + private readonly double _multiplier; + private readonly double _jitter; + private readonly Random _rng; private int _attempts; private TBuffered? _buffered; + private TransportOptions? _transportOptions; - public ReconnectPolicy(IClientStageOperations ops, int maxAttempts) + /// + /// Delay before the first retry after a failed attempt. When less than or equal to + /// (the default) reconnects fire immediately, preserving the legacy + /// zero-delay behaviour; a positive value defers each retry behind . + /// + /// Upper bound on the (pre-jitter) exponential backoff delay. + /// Growth factor applied per successive retry. + /// Fractional jitter (0..1) applied symmetrically to each computed delay. + public ReconnectPolicy( + IClientStageOperations ops, + int maxAttempts, + TimeSpan initialBackoff = default, + TimeSpan maxBackoff = default, + double multiplier = 2.0, + double jitter = 0.2, + Random? rng = null) { _ops = ops ?? throw new ArgumentNullException(nameof(ops)); _maxAttempts = maxAttempts; + _initialBackoff = initialBackoff; + _maxBackoff = maxBackoff > TimeSpan.Zero ? maxBackoff : initialBackoff; + _multiplier = multiplier <= 0 ? 1.0 : multiplier; + _jitter = Math.Clamp(jitter, 0.0, 1.0); + _rng = rng ?? Random.Shared; } public bool CanReconnect => _maxAttempts > 0; @@ -32,11 +61,14 @@ public ReconnectPolicy(IClientStageOperations ops, int maxAttempts) /// /// Begins reconnecting: stashes for later replay/failure, resets - /// the attempt counter to 1, and emits the first . + /// the attempt counter to 1, and emits the first . The first + /// attempt fires immediately (a real disconnect just occurred); only subsequent retries via + /// are spaced by the backoff. /// public void Start(TBuffered buffered, TransportOptions transportOptions) { _buffered = buffered; + _transportOptions = transportOptions; _attempts = 1; _ops.OnOutbound(new ConnectTransport(transportOptions)); } @@ -52,6 +84,12 @@ public void Start(TBuffered buffered, TransportOptions transportOptions) var buffered = _buffered; _buffered = default; _attempts = 0; + _transportOptions = default; + if (_initialBackoff > TimeSpan.Zero) + { + _ops.OnCancelTimer(BackoffTimerName); + } + return buffered; } @@ -74,7 +112,43 @@ public bool OnAttemptFailed(TransportOptions transportOptions, out TBuffered? bu buffered = default; _attempts++; - _ops.OnOutbound(new ConnectTransport(transportOptions)); + _transportOptions = transportOptions; + + if (_initialBackoff > TimeSpan.Zero) + { + // Defer the retry behind a backoff timer instead of reconnecting immediately, so a + // connection-refused peer is not hammered in a tight loop (see OnReconnectTimerFired). + _ops.OnScheduleTimer(BackoffTimerName, ComputeBackoff(_attempts)); + } + else + { + _ops.OnOutbound(new ConnectTransport(transportOptions)); + } + return false; } + + /// + /// Emits the deferred when the backoff timer named + /// fires. Returns for any other timer so + /// the caller can continue dispatching, and is a no-op if the reconnect was already abandoned. + /// + public bool OnReconnectTimerFired(string name) + { + if (name != BackoffTimerName) + { + return false; + } + + if (_transportOptions is not null) + { + _ops.OnOutbound(new ConnectTransport(_transportOptions)); + } + + return true; + } + + // _attempts is post-increment: 2 for the first retry, 3 for the second, ... so retryNumber = attempts - 1. + private TimeSpan ComputeBackoff(int attempt) + => ReconnectBackoff.Compute(attempt - 1, _initialBackoff, _maxBackoff, _multiplier, _jitter, _rng); } diff --git a/src/GaudiHTTP/Protocol/Syntax/Http10/Client/Http10ClientStateMachine.cs b/src/GaudiHTTP/Protocol/Syntax/Http10/Client/Http10ClientStateMachine.cs index 554cad19..38ff2fa8 100644 --- a/src/GaudiHTTP/Protocol/Syntax/Http10/Client/Http10ClientStateMachine.cs +++ b/src/GaudiHTTP/Protocol/Syntax/Http10/Client/Http10ClientStateMachine.cs @@ -56,7 +56,13 @@ public Http10ClientStateMachine(GaudiClientOptions options, IClientStageOperatio { _ops = ops; _options = options; - _reconnectPolicy = new ReconnectPolicy(ops, options.Http1.MaxReconnectAttempts); + _reconnectPolicy = new ReconnectPolicy( + ops, + options.Http1.MaxReconnectAttempts, + options.Http1.ReconnectInitialBackoff, + options.Http1.ReconnectMaxBackoff, + options.Http1.ReconnectBackoffMultiplier, + options.Http1.ReconnectBackoffJitter); var decoderOpts = options.ToHttp10DecoderOptions(); @@ -195,6 +201,7 @@ public void OnUpstreamFinished() public void OnTimerFired(string name) { + _reconnectPolicy.OnReconnectTimerFired(name); } public void OnBodyMessage(object msg) diff --git a/src/GaudiHTTP/Protocol/Syntax/Http11/Client/Http11ClientStateMachine.cs b/src/GaudiHTTP/Protocol/Syntax/Http11/Client/Http11ClientStateMachine.cs index 435b5f95..b59386bd 100644 --- a/src/GaudiHTTP/Protocol/Syntax/Http11/Client/Http11ClientStateMachine.cs +++ b/src/GaudiHTTP/Protocol/Syntax/Http11/Client/Http11ClientStateMachine.cs @@ -107,7 +107,13 @@ public Http11ClientStateMachine( { _ops = ops; _options = options; - _reconnectPolicy = new ReconnectPolicy>(ops, options.Http1.MaxReconnectAttempts); + _reconnectPolicy = new ReconnectPolicy>( + ops, + options.Http1.MaxReconnectAttempts, + options.Http1.ReconnectInitialBackoff, + options.Http1.ReconnectMaxBackoff, + options.Http1.ReconnectBackoffMultiplier, + options.Http1.ReconnectBackoffJitter); var decoderOpts = options.ToHttp11DecoderOptions(); var encoderOpts = options.ToHttp11EncoderOptions(); @@ -359,6 +365,7 @@ public void OnUpstreamFinished() public void OnTimerFired(string name) { + _reconnectPolicy.OnReconnectTimerFired(name); } public void OnBodyMessage(object msg) @@ -395,7 +402,21 @@ public void OnOutboundFlushed() public void Cleanup() { - _inFlightQueue.Clear(); + // Fail (don't silently drop) requests still in flight or buffered for reconnect replay, so + // callers fault promptly on stage teardown instead of hanging until their client-side timeout. + if (_inFlightQueue.Count > 0) + { + RequestFault.FailAll(_inFlightQueue, + new HttpRequestException("HTTP/1.1 connection was torn down before the request completed.")); + _inFlightQueue.Clear(); + } + + if (_reconnectPolicy.TakeBuffered() is { Count: > 0 } bufferedForReplay) + { + RequestFault.FailAll(bufferedForReplay, + new HttpRequestException("HTTP/1.1 connection was torn down before the buffered request could be replayed.")); + } + _pendingBodyResponse?.Dispose(); _pendingBodyResponse = null; _outboundBodyPending = false; @@ -673,8 +694,28 @@ private void FailOrphanedRequests() private void StartReconnect() { - var buffered = new Queue(_inFlightQueue); - _inFlightQueue.Clear(); + // Only idempotent in-flight requests are safe to replay: a non-idempotent request + // (e.g. POST) may already have been received and processed by the server, so replaying + // it after connection loss risks a duplicate side effect (RFC 9110 §9.2.2). Fail those + // instead of buffering them. Mirrors the HTTP/2 IsStreamSafeToReplay gate. + var buffered = new Queue(); + while (_inFlightQueue.Count > 0) + { + var request = _inFlightQueue.Dequeue(); + if (MethodProperties.IsIdempotent(request.Method)) + { + buffered.Enqueue(request); + } + else + { + Tracing.For("Protocol").Info(this, + "HTTP/1.1: not replaying non-idempotent request {0} {1} on reconnect (server may have already processed it)", + request.Method, request.RequestUri); + request.Fail(new HttpRequestException( + "Non-idempotent HTTP/1.1 request was not replayed after connection loss; the server may have already processed it.")); + } + } + _connectionState = ConnectionState.Reconnecting; _reconnectPolicy.Start(buffered, _transportOptions!); } diff --git a/src/GaudiHTTP/Protocol/Syntax/Http2/Client/Http2ClientStateMachine.cs b/src/GaudiHTTP/Protocol/Syntax/Http2/Client/Http2ClientStateMachine.cs index ced26583..46ea9731 100644 --- a/src/GaudiHTTP/Protocol/Syntax/Http2/Client/Http2ClientStateMachine.cs +++ b/src/GaudiHTTP/Protocol/Syntax/Http2/Client/Http2ClientStateMachine.cs @@ -2,6 +2,7 @@ using GaudiHTTP.Client; using GaudiHTTP.Internal; using GaudiHTTP.Protocol.Multiplexed; +using GaudiHTTP.Protocol.Semantics; using GaudiHTTP.Streams.Stages.Client; using static Servus.Senf; @@ -156,6 +157,15 @@ public void OnTimerFired(string name) } } + break; + } + case ReconnectBackoff.TimerName: + { + if (_reconnect.IsReconnecting && _transportOptions is not null) + { + ops.OnOutbound(new ConnectTransport(_transportOptions)); + } + break; } } @@ -178,7 +188,24 @@ public void OnRequestCancelled(HttpRequestMessage request) public void OnBodyMessage(object msg) => _clientSession.OnBodyMessage(msg); - public void Cleanup() => _clientSession.Cleanup(); + public void Cleanup() + { + // Fail (don't silently drop) requests still in flight or buffered for reconnect replay, so + // callers fault promptly on stage teardown instead of hanging until their client-side timeout. + // request.Fail is idempotent, so streams that already delivered a response are unaffected. + if (_reconnect.IsReconnecting) + { + _reconnect.FailAllBuffered(new HttpRequestException("HTTP/2 connection torn down during reconnect.")); + _reconnect.Reset(); + } + + foreach (var (_, request) in _clientSession.GetCorrelationMap()) + { + request.Fail(new HttpRequestException("HTTP/2 connection was torn down before the request completed.")); + } + + _clientSession.Cleanup(); + } private void OnConnectionLost(int lastStreamId) { @@ -262,7 +289,22 @@ private void OnReconnectAttemptFailed() return; } - ops.OnOutbound(new ConnectTransport(_transportOptions!)); + // Defer the retry behind a backoff timer instead of reconnecting immediately, so a + // connection-refused peer is not hammered in a tight loop (see OnTimerFired). + if (options.Http2.ReconnectInitialBackoff > TimeSpan.Zero) + { + ops.OnScheduleTimer(ReconnectBackoff.TimerName, ReconnectBackoff.Compute( + _reconnect.Attempts - 1, + options.Http2.ReconnectInitialBackoff, + options.Http2.ReconnectMaxBackoff, + options.Http2.ReconnectBackoffMultiplier, + options.Http2.ReconnectBackoffJitter, + Random.Shared)); + } + else + { + ops.OnOutbound(new ConnectTransport(_transportOptions!)); + } } private void ScheduleKeepAlivePing() diff --git a/src/GaudiHTTP/Protocol/Syntax/Http3/Client/Http3ClientStateMachine.cs b/src/GaudiHTTP/Protocol/Syntax/Http3/Client/Http3ClientStateMachine.cs index 39cf4396..a4fe3222 100644 --- a/src/GaudiHTTP/Protocol/Syntax/Http3/Client/Http3ClientStateMachine.cs +++ b/src/GaudiHTTP/Protocol/Syntax/Http3/Client/Http3ClientStateMachine.cs @@ -2,6 +2,7 @@ using GaudiHTTP.Client; using GaudiHTTP.Internal; using GaudiHTTP.Protocol.Multiplexed; +using GaudiHTTP.Protocol.Semantics; using GaudiHTTP.Protocol.Syntax.Http3.Qpack; using GaudiHTTP.Streams.Stages.Client; using static Servus.Senf; @@ -203,6 +204,16 @@ public void OnUpstreamFinished() public void OnTimerFired(string name) { + if (name == ReconnectBackoff.TimerName) + { + if (_reconnect.IsReconnecting && _transportOptions is not null) + { + _ops.OnOutbound(new ConnectTransport(_transportOptions)); + } + + return; + } + if (name != IdleTimeoutCheckTimer) { return; @@ -249,6 +260,20 @@ public void OnOutboundFlushed() public void Cleanup() { + // Fail (don't silently drop) requests still in flight or buffered for reconnect replay, so + // callers fault promptly on stage teardown instead of hanging until their client-side timeout. + // request.Fail is idempotent, so streams that already delivered a response are unaffected. + if (_reconnect.IsReconnecting) + { + _reconnect.FailAllBuffered(new HttpRequestException("HTTP/3 connection torn down during reconnect.")); + _reconnect.Reset(); + } + + foreach (var (_, request) in _clientSession.GetCorrelationMap()) + { + request.Fail(new HttpRequestException("HTTP/3 connection was torn down before the request completed.")); + } + _clientSession.Cleanup(); } @@ -346,7 +371,22 @@ private void OnReconnectAttemptFailed() return; } - _ops.OnOutbound(new ConnectTransport(_transportOptions!)); + // Defer the retry behind a backoff timer instead of reconnecting immediately, so a + // connection-refused peer is not hammered in a tight loop (see OnTimerFired). + if (_options.Http3.ReconnectInitialBackoff > TimeSpan.Zero) + { + _ops.OnScheduleTimer(ReconnectBackoff.TimerName, ReconnectBackoff.Compute( + _reconnect.Attempts - 1, + _options.Http3.ReconnectInitialBackoff, + _options.Http3.ReconnectMaxBackoff, + _options.Http3.ReconnectBackoffMultiplier, + _options.Http3.ReconnectBackoffJitter, + Random.Shared)); + } + else + { + _ops.OnOutbound(new ConnectTransport(_transportOptions!)); + } } private void ScheduleIdleCheck() diff --git a/src/GaudiHTTP/Streams/Lifecycle/ClientStreamOwner.cs b/src/GaudiHTTP/Streams/Lifecycle/ClientStreamOwner.cs index c754d1e0..3b9fe239 100644 --- a/src/GaudiHTTP/Streams/Lifecycle/ClientStreamOwner.cs +++ b/src/GaudiHTTP/Streams/Lifecycle/ClientStreamOwner.cs @@ -8,6 +8,7 @@ using Akka.Streams.Dsl; using Servus.Akka.Transport; using GaudiHTTP.Internal; +using GaudiHTTP.Protocol.Semantics; using GaudiHTTP.Streams.Pooling; using static Servus.Senf; @@ -23,10 +24,15 @@ internal sealed record RegisterConsumer( ChannelWriter FallbackResponseWriter); internal sealed record UnregisterConsumer(Guid ConsumerId); + // Reconnect layering contract: socket loss is absorbed inband by the transport state machine and + // the protocol layer (ReconnectPolicy / ReconnectionManager), which buffer and replay requests + // WITHOUT ever terminating the materialized graph — so the owner is never involved in socket-loss + // reconnect. The owner re-materializes ONLY on faults the protocol layer cannot recover from: a + // graph build/materialization exception or an engine-flow/decoder desync that FailStages the stage. + // All three layers share the same jittered backoff via ReconnectBackoff.Compute. private TimeSpan CalculateBackoff(int attempt) => - TimeSpan.FromMilliseconds( - Math.Min(_initialBackoff.TotalMilliseconds * Math.Pow(_backoffMultiplier, attempt), - _maxBackoff.TotalMilliseconds)); + ReconnectBackoff.Compute(attempt + 1, _initialBackoff, _maxBackoff, _backoffMultiplier, + _backoffJitter, Random.Shared); private static readonly TimeSpan ShutdownTimeout = TimeSpan.FromSeconds(5); @@ -41,6 +47,7 @@ private TimeSpan CalculateBackoff(int attempt) => private readonly TimeSpan _initialBackoff; private readonly TimeSpan _maxBackoff; private readonly double _backoffMultiplier; + private readonly double _backoffJitter; private readonly int _maxRetryAttempts; private int _retryAttempts; @@ -73,6 +80,7 @@ public ClientStreamOwner(GaudiClientOptions clientOptions, PipelineDescriptor pi _initialBackoff = initialBackoffOverride ?? clientOptions.StreamRetryInitialBackoff; _maxBackoff = clientOptions.StreamRetryMaxBackoff; _backoffMultiplier = clientOptions.StreamRetryBackoffMultiplier; + _backoffJitter = clientOptions.StreamRetryBackoffJitter; _maxRetryAttempts = clientOptions.MaxStreamRetryAttempts; Initializing(); diff --git a/src/GaudiHTTP/Streams/Stages/Server/ApplicationBridgeStage.cs b/src/GaudiHTTP/Streams/Stages/Server/ApplicationBridgeStage.cs index 9ef1ee3f..022cb98c 100644 --- a/src/GaudiHTTP/Streams/Stages/Server/ApplicationBridgeStage.cs +++ b/src/GaudiHTTP/Streams/Stages/Server/ApplicationBridgeStage.cs @@ -204,10 +204,11 @@ private void OnPush() { DispatchAsync(features, seq); } - catch (Exception) + catch (Exception ex) { - FinishRequest(seq, features, error: null, failStatus: 500, emit: true, - cleanupSlot: false, resetBackpressure: false); + Tracing.For("Handler").Error(this, "handler dispatch threw for seq {0}: {1}", seq, ex); + FinishRequest(seq, features, error: ex, failStatus: 500, emit: true, + cleanupSlot: true, resetBackpressure: false); } TryPullNext(); @@ -220,9 +221,12 @@ private void DispatchAsync(IFeatureCollection features, int seq) { appContext = _stage._application.CreateContext(ContainerFor(features)); } - catch (Exception) + catch (Exception ex) { - FinishRequest(seq, features, error: null, failStatus: 500, emit: true, + // Context creation failed before a slot exists, so there is no AppContext to dispose + // the error onto; tracing is the only place operators can see this crash. + Tracing.For("Handler").Error(this, "context creation threw for seq {0}: {1}", seq, ex); + FinishRequest(seq, features, error: ex, failStatus: 500, emit: true, cleanupSlot: false, trackMetrics: false); return; }