Make SSE idle-timeout tests deterministic with FakeTimeProvider

Directory.Packages.props

@@ -35,6 +35,7 @@
     <PackageVersion Include="Microsoft.Extensions.Http" Version="10.0.11" />
     <PackageVersion Include="Microsoft.Extensions.Logging" Version="10.0.11" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="10.0.11" />
+    <PackageVersion Include="Microsoft.Extensions.TimeProvider.Testing" Version="10.10.0" />
     <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
     <PackageVersion Include="Moq" Version="4.20.72" />
     <PackageVersion Include="Shouldly" Version="4.3.0" />

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)
+    /// <param name="timeProvider">Clock driving the idle timer; <see cref="TimeProvider.System" /> when null.
+    /// Tests pass a fake provider so the timer only advances when they say so.</param>
+    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<int> ReadAsync(Memory<byte> 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)}.");
         }
     }

src/Linq2GraphQL.Client.Subscriptions/SSEClient.cs

@@ -32,6 +32,9 @@ public SSEClient(GraphClient graphClient, GraphQLRequest payload)
         this.payload = payload;
     }
 
+    /// <summary>Clock for the idle timeout; tests substitute a fake provider.</summary>
+    internal TimeProvider TimeProvider { get; init; } = TimeProvider.System;
+
     /// <summary>
     ///     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)

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<int> ReadAsync(Memory<byte> 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<int> ReadAsync(Memory<byte> 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;

test/Linq2GraphQL.Tests/Linq2GraphQL.Tests.csproj

@@ -12,6 +12,7 @@
     <ItemGroup>
         <PackageReference Include="Microsoft.AspNetCore.Mvc.Testing"/>
         <PackageReference Include="Microsoft.Extensions.AI"/>
+        <PackageReference Include="Microsoft.Extensions.TimeProvider.Testing"/>
         <PackageReference Include="Microsoft.NET.Test.Sdk"/>
         <PackageReference Include="Moq"/>
         <PackageReference Include="Shouldly"/>

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;
 /// </summary>
 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 };
+    }
+
+    /// <summary>Pushes a chunk and waits until the reader has consumed it and parked on its next read.</summary>
+    private static async Task PushAndAwaitNextReadAsync(ScriptableSseStream stream, string text)
+    {
+        var parked = stream.ParkedReads;
+        stream.Push(text);
+        await WaitUntilAsync(() => stream.ParkedReads > parked);
     }
 
     private static (List<byte[]> payloads, List<GraphQueryExecutionException> errors, List<Exception> 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<Exception>();
         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]

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;
+
+    /// <summary>
+    ///     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).
+    /// </summary>
+    public int ParkedReads => Volatile.Read(ref parkedReads);
 
     private static TaskCompletionSource<bool> NewSignal()
     {
@@ -91,6 +99,7 @@ public override async Task<int> ReadAsync(byte[] buffer, int offset, int count,
                 }
 
                 wait = available;
+                Interlocked.Increment(ref parkedReads);
             }
 
             await wait.Task.WaitAsync(cancellationToken).ConfigureAwait(false);

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, )",