From dd1279e3a05966ce54ecd6a13a1042816aeb3a34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Dang=C3=A5rden?= Date: Sat, 17 Oct 2026 16:52:01 +0200 Subject: [PATCH 1/2] Drive SSE idle-timeout tests with FakeTimeProvider KeepAliveComments_PreventIdleTimeout used real wall-clock margins (1 s timeout, 100 ms keep-alive gaps) and failed in CI when thread-pool load delayed the gaps past the timeout. IdleTimeoutStream now takes an optional TimeProvider (default System), exposed internally on SSEClient. The tests advance a FakeTimeProvider only after the reader has provably consumed each push and re-armed, so scheduling cannot fire the timer. KeepAliveComments also asserts the fake clock does fire the timeout once data stops, and SlowButRegularData_DoesNotTimeOut and IdleTimeoutSet_StalledServer use the same clock. No public API or runtime behaviour change. --- Directory.Packages.props | 1 + .../IdleTimeoutStream.cs | 9 ++-- .../SSEClient.cs | 5 +- .../IdleTimeoutStreamTests.cs | 19 +++++--- .../Linq2GraphQL.Tests.csproj | 1 + test/Linq2GraphQL.Tests/SseClientTests.cs | 47 ++++++++++++++----- .../TestSupport/SseTestServer.cs | 9 ++++ test/Linq2GraphQL.Tests/packages.lock.json | 6 +++ 8 files changed, 73 insertions(+), 24 deletions(-) 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 4f1298c3..097dc2fb 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs @@ -10,10 +10,12 @@ internal sealed class IdleTimeoutStream : Stream { private readonly Stream inner; private readonly TimeSpan idleTimeout; - private readonly CancellationTokenSource idleCts = new(); + private readonly CancellationTokenSource idleCts; 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,7 +24,8 @@ internal IdleTimeoutStream(Stream inner, TimeSpan idleTimeout) this.inner = inner ?? throw new ArgumentNullException(nameof(inner)); this.idleTimeout = idleTimeout; - idleCts.CancelAfter(idleTimeout); + // A CTS bound to a TimeProvider uses that provider's timers for CancelAfter as well. + idleCts = new CancellationTokenSource(idleTimeout, timeProvider ?? TimeProvider.System); } public override bool CanRead => inner.CanRead; 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 5ae78827..ea678832 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]; @@ -206,7 +208,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; @@ -218,7 +220,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..d7feb8a1 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(1), }; + 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(1)); 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(1), }; + // 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 900 ms is far beyond the 1 s timeout: only re-arming on comments keeps it alive. + time.Advance(TimeSpan.FromMilliseconds(900)); } - 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(1)); + 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, )", From 51ce6e1b91e528edcbced14aa967eb03eb8de347 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Dang=C3=A5rden?= Date: Sun, 18 Oct 2026 08:33:29 +0200 Subject: [PATCH 2/2] Drop nullable annotation on IdleTimeoutStream timeProvider (CS8632) The Subscriptions project is not in a nullable context, so the '?' raised a warning that main does not have. --- src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs b/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs index 097dc2fb..405a85ed 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs @@ -15,7 +15,7 @@ internal sealed class IdleTimeoutStream : Stream /// 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) + internal IdleTimeoutStream(Stream inner, TimeSpan idleTimeout, TimeProvider timeProvider = null) { if (idleTimeout <= TimeSpan.Zero) {