diff --git a/Directory.Packages.props b/Directory.Packages.props index 83225ea4..72542c81 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -35,6 +35,7 @@ + diff --git a/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs b/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs index 0d435042..243963ed 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs @@ -11,9 +11,12 @@ internal sealed class IdleTimeoutStream : Stream { private readonly Stream inner; private readonly TimeSpan idleTimeout; + private readonly TimeProvider timeProvider; private bool disposed; - internal IdleTimeoutStream(Stream inner, TimeSpan idleTimeout) + /// Clock driving the idle timer; when null. + /// Tests pass a fake provider so the timer only advances when they say so. + internal IdleTimeoutStream(Stream inner, TimeSpan idleTimeout, TimeProvider timeProvider = null) { if (idleTimeout <= TimeSpan.Zero) { @@ -22,6 +25,7 @@ internal IdleTimeoutStream(Stream inner, TimeSpan idleTimeout) this.inner = inner ?? throw new ArgumentNullException(nameof(inner)); this.idleTimeout = idleTimeout; + this.timeProvider = timeProvider ?? TimeProvider.System; } public override bool CanRead => inner.CanRead; @@ -37,20 +41,21 @@ public override async ValueTask ReadAsync(Memory buffer, Cancellation // A fresh timer per read, disposed when the read completes: it runs only while waiting // for data, and a timer that fired during a read which still returned bytes cannot leak // into the next read (a CancellationTokenSource cannot be reset once cancelled). - // CreateLinkedTokenSource with a non-cancelable token is equivalent to a plain source. - using var readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - readCts.CancelAfter(idleTimeout); + // The timer source is bound to the TimeProvider (a linked source cannot be), then linked + // with the caller's token for the read itself. + using var timerCts = new CancellationTokenSource(idleTimeout, timeProvider); + using var readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timerCts.Token); try { return await inner.ReadAsync(buffer, readCts.Token).ConfigureAwait(false); } catch (OperationCanceledException) when ( - readCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) + timerCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) { - // Only the idle timer or the caller can cancel readCts; it fired and the caller did - // not cancel, so the timer fired with no bytes received: the connection is dead. An - // OperationCanceledException the inner stream raised on its own propagates unchanged. + // The idle timer fired and the caller did not cancel, so no bytes arrived within the + // window: the connection is dead. An OperationCanceledException the inner stream + // raised on its own propagates unchanged. throw new IOException($"SSE idle timeout: no data received within {FormatTimeout(idleTimeout)}."); } } diff --git a/src/Linq2GraphQL.Client.Subscriptions/SSEClient.cs b/src/Linq2GraphQL.Client.Subscriptions/SSEClient.cs index 7bd2f1b8..c7fa7d7b 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/SSEClient.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/SSEClient.cs @@ -32,6 +32,9 @@ public SSEClient(GraphClient graphClient, GraphQLRequest payload) this.payload = payload; } + /// Clock for the idle timeout; tests substitute a fake provider. + internal TimeProvider TimeProvider { get; init; } = TimeProvider.System; + /// /// UTF-8 payloads of "next" events decoded to strings — deferred per subscriber: /// the decode only happens for listeners actually attached here. @@ -154,7 +157,7 @@ internal async Task Start(CancellationToken cancellationToken) var idleTimeout = graphClient.ClientOptions?.SubscriptionIdleTimeout; if (idleTimeout is { } timeout) { - source = new IdleTimeoutStream(rawStream, timeout); + source = new IdleTimeoutStream(rawStream, timeout, TimeProvider); } lock (resourceLock) diff --git a/test/Linq2GraphQL.Tests/IdleTimeoutStreamTests.cs b/test/Linq2GraphQL.Tests/IdleTimeoutStreamTests.cs index 8a879248..31fbef6d 100644 --- a/test/Linq2GraphQL.Tests/IdleTimeoutStreamTests.cs +++ b/test/Linq2GraphQL.Tests/IdleTimeoutStreamTests.cs @@ -1,5 +1,6 @@ using System.Text; using Linq2GraphQL.Client.Subscriptions; +using Microsoft.Extensions.Time.Testing; namespace Linq2GraphQL.Tests; @@ -43,11 +44,12 @@ public async Task DataArrivingWithinWindow_ReadsSucceed() [Fact] public async Task SlowButRegularData_DoesNotTimeOut() { - // One byte every 50ms for ~1s with a 750ms idle timeout: the whole read outlasts the - // timeout, so it only passes if the timer re-arms per read. The 15x per-gap margin keeps - // it stable when the full suite loads the thread pool. - using var stream = new IdleTimeoutStream(new DrippingStream(TimeSpan.FromMilliseconds(50), 20), - TimeSpan.FromMilliseconds(750)); + // One byte every 500ms (fake clock) for 10s with a 750ms idle timeout: the whole read + // outlasts the timeout, so it only passes if the timer re-arms per read. The clock only + // moves inside each read, so scheduling delays cannot fire the timer. + var time = new FakeTimeProvider(); + using var stream = new IdleTimeoutStream(new DrippingStream(time, TimeSpan.FromMilliseconds(500), 20), + TimeSpan.FromMilliseconds(750), time); using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); var buffer = new byte[32]; @@ -346,7 +348,7 @@ public override async ValueTask ReadAsync(Memory buffer, Cancellation public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); } - private sealed class DrippingStream(TimeSpan delay, int count) : Stream + private sealed class DrippingStream(FakeTimeProvider time, TimeSpan delay, int count) : Stream { private int remaining = count; @@ -358,7 +360,10 @@ private sealed class DrippingStream(TimeSpan delay, int count) : Stream public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) { - await Task.Delay(delay, cancellationToken); + await Task.Yield(); + // The byte "arrives" after the delay; a fired idle timer cancels this read. + time.Advance(delay); + cancellationToken.ThrowIfCancellationRequested(); if (remaining-- <= 0) { return 0; diff --git a/test/Linq2GraphQL.Tests/Linq2GraphQL.Tests.csproj b/test/Linq2GraphQL.Tests/Linq2GraphQL.Tests.csproj index dc40db00..09926f77 100644 --- a/test/Linq2GraphQL.Tests/Linq2GraphQL.Tests.csproj +++ b/test/Linq2GraphQL.Tests/Linq2GraphQL.Tests.csproj @@ -12,6 +12,7 @@ + diff --git a/test/Linq2GraphQL.Tests/SseClientTests.cs b/test/Linq2GraphQL.Tests/SseClientTests.cs index 6aee2cd8..4e3678b8 100644 --- a/test/Linq2GraphQL.Tests/SseClientTests.cs +++ b/test/Linq2GraphQL.Tests/SseClientTests.cs @@ -3,6 +3,7 @@ using Linq2GraphQL.Client; using System.Reactive.Linq; using Linq2GraphQL.Client.Subscriptions; +using Microsoft.Extensions.Time.Testing; namespace Linq2GraphQL.Tests; @@ -13,11 +14,20 @@ namespace Linq2GraphQL.Tests; /// public class SseClientTests { - private static SSEClient CreateClient(FakeSseHandler handler, GraphClientOptions? options = null) + private static SSEClient CreateClient(FakeSseHandler handler, GraphClientOptions? options = null, + TimeProvider? timeProvider = null) { var graphClient = TestGraphClientFactory.Create(handler, options); var request = new GraphQLRequest { Query = "subscription { customerAdded { customerId } }" }; - return new SSEClient(graphClient, request); + return new SSEClient(graphClient, request) { TimeProvider = timeProvider ?? TimeProvider.System }; + } + + /// Pushes a chunk and waits until the reader has consumed it and parked on its next read. + private static async Task PushAndAwaitNextReadAsync(ScriptableSseStream stream, string text) + { + var parked = stream.ParkedReads; + stream.Push(text); + await WaitUntilAsync(() => stream.ParkedReads > parked); } private static (List payloads, List errors, List failures) @@ -275,14 +285,17 @@ public async Task IdleTimeoutSet_StalledServer_FailsTheStream() var options = new GraphClientOptions { SubscriptionProtocol = SubscriptionProtocol.ServerSentEvents, - SubscriptionIdleTimeout = TimeSpan.FromMilliseconds(100), + SubscriptionIdleTimeout = TimeSpan.FromSeconds(30), }; + var time = new FakeTimeProvider(); var handler = FakeSseHandler.Streaming(out var stream); - using var client = CreateClient(handler, options); + using var client = CreateClient(handler, options, time); var failures = new List(); client.Failures.Subscribe(failures.Add); var started = client.Start(); + await WaitUntilAsync(() => stream.ParkedReads > 0); + time.Advance(TimeSpan.FromSeconds(30)); await started.WaitAsync(TimeSpan.FromSeconds(10)); var failure = Assert.Single(failures); @@ -297,27 +310,35 @@ public async Task KeepAliveComments_PreventIdleTimeout() var options = new GraphClientOptions { SubscriptionProtocol = SubscriptionProtocol.ServerSentEvents, - SubscriptionIdleTimeout = TimeSpan.FromMilliseconds(1000), + SubscriptionIdleTimeout = TimeSpan.FromSeconds(30), }; + // Fake clock: the idle timer only moves when the test advances it, after the reader has + // provably re-armed, so thread-pool load cannot make it fire early. + var time = new FakeTimeProvider(); var handler = FakeSseHandler.Streaming(out var stream); - using var client = CreateClient(handler, options); + using var client = CreateClient(handler, options, time); var (payloads, _, failures) = Subscribe(client); - _ = client.Start(); + var started = client.Start(); await client.Connected.WaitAsync(TimeSpan.FromSeconds(10)); for (var i = 0; i < 15; i++) { - stream.Push(": ka\n\n"); - // ~1.5 s total > idle timeout, so only re-arming on comments keeps it alive; the 10x - // per-gap margin keeps it stable when the full suite loads the thread pool. - await Task.Delay(100); + await PushAndAwaitNextReadAsync(stream, ": ka\n\n"); + // 15 x 29 s is far beyond the 30 s timeout: only re-arming on comments keeps it alive. + time.Advance(TimeSpan.FromSeconds(29)); } - stream.Push("event: next\ndata: \"alive\"\n\n"); + await PushAndAwaitNextReadAsync(stream, "event: next\ndata: \"alive\"\n\n"); await WaitUntilAsync(() => payloads.Count == 1); + Assert.Empty(failures); + + // Control: the fake clock really drives the timer — silence past the timeout fails it. + time.Advance(TimeSpan.FromSeconds(30)); + await started.WaitAsync(TimeSpan.FromSeconds(10)); Assert.Single(payloads); - Assert.Empty(failures); + var failure = Assert.Single(failures); + Assert.Contains("idle", failure.Message, StringComparison.OrdinalIgnoreCase); } [Fact] diff --git a/test/Linq2GraphQL.Tests/TestSupport/SseTestServer.cs b/test/Linq2GraphQL.Tests/TestSupport/SseTestServer.cs index 1cc1f1ff..8c422b63 100644 --- a/test/Linq2GraphQL.Tests/TestSupport/SseTestServer.cs +++ b/test/Linq2GraphQL.Tests/TestSupport/SseTestServer.cs @@ -19,6 +19,14 @@ internal sealed class ScriptableSseStream : Stream private byte[]? current; private int currentOffset; private bool ended; + private int parkedReads; + + /// + /// How many times a read has parked on an empty queue. Once it moves past a value + /// sampled before a push, the reader has consumed that push and issued its next read + /// (so a wrapping IdleTimeoutStream has re-armed its timer). + /// + public int ParkedReads => Volatile.Read(ref parkedReads); private static TaskCompletionSource NewSignal() { @@ -91,6 +99,7 @@ public override async Task ReadAsync(byte[] buffer, int offset, int count, } wait = available; + Interlocked.Increment(ref parkedReads); } await wait.Task.WaitAsync(cancellationToken).ConfigureAwait(false); diff --git a/test/Linq2GraphQL.Tests/packages.lock.json b/test/Linq2GraphQL.Tests/packages.lock.json index b5ead633..79a71836 100644 --- a/test/Linq2GraphQL.Tests/packages.lock.json +++ b/test/Linq2GraphQL.Tests/packages.lock.json @@ -32,6 +32,12 @@ "System.Numerics.Tensors": "10.0.12" } }, + "Microsoft.Extensions.TimeProvider.Testing": { + "type": "Direct", + "requested": "[10.10.0, )", + "resolved": "10.10.0", + "contentHash": "8duZdxuO3Oo8tlyQXVq1drXee6GJC9DjUU97Pq3YaIqf92jrAHsWxbbRKwvjQI48Mer3IANwUKF0elqvHeY4Rw==" + }, "Microsoft.NET.Test.Sdk": { "type": "Direct", "requested": "[17.14.1, )",