Fix flaky Task.WhenAny + Assert.Same timing races

tests/B3.EntryPoint.Client.Tests/ColdStartResumeTests.cs

@@ -2,6 +2,7 @@
 using B3.EntryPoint.Client.Fixp;
 using B3.EntryPoint.Client.State;
 using B3.EntryPoint.Client.TestPeer;
+using B3.EntryPoint.Client.Tests.TestSupport;
 
 namespace B3.EntryPoint.Client.Tests;
 
@@ -263,8 +264,8 @@ public async Task TerminateOnDispose_DefaultTrue_SendsTerminate()
 
         await client.DisposeAsync();
 
-        var completed = await Task.WhenAny(terminateSeen.Task, Task.Delay(2000, cts.Token));
-        Assert.Same(terminateSeen.Task, completed);
+        await AsyncAssert.CompletesWithinAsync(
+            terminateSeen.Task, TimeSpan.FromSeconds(3), "expected a Terminate frame on dispose", cts.Token);
     }
 
     [Fact]

tests/B3.EntryPoint.Client.Tests/Fixp/KeepAliveSchedulerTests.cs

@@ -1,4 +1,5 @@
 using B3.EntryPoint.Client.Fixp;
+using B3.EntryPoint.Client.Tests.TestSupport;
 
 namespace B3.EntryPoint.Client.Tests.Fixp;
 
@@ -94,10 +95,17 @@ Task<ulong> SendAsync(CancellationToken ct)
             TimeSpan.FromMilliseconds(40),
             SendAsync);
         scheduler.Start();
-        var completed = await Task.WhenAny(twoTicks.Task, Task.Delay(TimeSpan.FromSeconds(5)));
-        scheduler.Stop();
-        scheduler.Dispose();
-        Assert.Same(twoTicks.Task, completed);
+        try
+        {
+            // Widened from 5s to 10s (see #245): under CI-runner contention,
+            // even a 40ms tick interval can starve past a tight timeout.
+            await AsyncAssert.CompletesWithinAsync(twoTicks.Task, TimeSpan.FromSeconds(10), "expected two keep-alive ticks");
+        }
+        finally
+        {
+            scheduler.Stop();
+            scheduler.Dispose();
+        }
         int finalCount;
         lock (ticks) finalCount = ticks.Count;
         Assert.True(finalCount >= 2, $"expected >=2 ticks, got {finalCount}");

tests/B3.EntryPoint.Client.Tests/TestSupport/AsyncAssert.cs

@@ -0,0 +1,35 @@
+using System.Diagnostics;
+using Xunit.Sdk;
+
+namespace B3.EntryPoint.Client.Tests.TestSupport;
+
+/// <summary>
+/// Replaces the flaky <c>Assert.Same(task, await Task.WhenAny(task,
+/// Task.Delay(timeout)))</c> pattern (see #245) with a helper that reports
+/// elapsed time on timeout, making CI flakes easier to diagnose, and that
+/// re-awaits the original task so any exception it faulted with propagates
+/// instead of being swallowed.
+/// </summary>
+public static class AsyncAssert
+{
+    public static async Task CompletesWithinAsync(Task task, TimeSpan timeout, string? because = null, CancellationToken cancellationToken = default)
+    {
+        var stopwatch = Stopwatch.StartNew();
+        var delay = Task.Delay(timeout, cancellationToken);
+        var completed = await Task.WhenAny(task, delay).ConfigureAwait(false);
+        if (!ReferenceEquals(completed, task))
+        {
+            var reason = because is null ? string.Empty : $" ({because})";
+            throw new XunitException(
+                $"Expected the awaited task to complete within {timeout}{reason}, but it did not. Elapsed: {stopwatch.Elapsed}.");
+        }
+
+        await task.ConfigureAwait(false);
+    }
+
+    public static async Task<T> CompletesWithinAsync<T>(Task<T> task, TimeSpan timeout, string? because = null, CancellationToken cancellationToken = default)
+    {
+        await CompletesWithinAsync((Task)task, timeout, because, cancellationToken).ConfigureAwait(false);
+        return await task.ConfigureAwait(false);
+    }
+}

tests/B3.EntryPoint.Conformance/Infrastructure/AsyncAssert.cs

@@ -0,0 +1,35 @@
+using System.Diagnostics;
+using Xunit.Sdk;
+
+namespace B3.EntryPoint.Conformance.Infrastructure;
+
+/// <summary>
+/// Replaces the flaky <c>Assert.Same(task, await Task.WhenAny(task,
+/// Task.Delay(timeout)))</c> pattern (see #245) with a helper that reports
+/// elapsed time on timeout, making CI flakes easier to diagnose, and that
+/// re-awaits the original task so any exception it faulted with propagates
+/// instead of being swallowed.
+/// </summary>
+public static class AsyncAssert
+{
+    public static async Task CompletesWithinAsync(Task task, TimeSpan timeout, string? because = null, CancellationToken cancellationToken = default)
+    {
+        var stopwatch = Stopwatch.StartNew();
+        var delay = Task.Delay(timeout, cancellationToken);
+        var completed = await Task.WhenAny(task, delay).ConfigureAwait(false);
+        if (!ReferenceEquals(completed, task))
+        {
+            var reason = because is null ? string.Empty : $" ({because})";
+            throw new XunitException(
+                $"Expected the awaited task to complete within {timeout}{reason}, but it did not. Elapsed: {stopwatch.Elapsed}.");
+        }
+
+        await task.ConfigureAwait(false);
+    }
+
+    public static async Task<T> CompletesWithinAsync<T>(Task<T> task, TimeSpan timeout, string? because = null, CancellationToken cancellationToken = default)
+    {
+        await CompletesWithinAsync((Task)task, timeout, because, cancellationToken).ConfigureAwait(false);
+        return await task.ConfigureAwait(false);
+    }
+}

tests/B3.EntryPoint.Conformance/Spec_4_6_Sequence/SequenceHeartbeatTests.cs

@@ -42,11 +42,11 @@ public async Task KeepAlive_Sequence_Frames_Are_Exchanged()
             if (Interlocked.Increment(ref receivedCount) >= 1) receivedTcs.TrySetResult();
         };
 
-        // Cap the wait at 5×interval so a stalled scheduler fails fast.
-        var timeout = TimeSpan.FromMilliseconds(250 * 5);
+        // Cap the wait at 10×interval so a stalled scheduler still fails fast
+        // while leaving headroom for CI-runner contention (see #245).
+        var timeout = TimeSpan.FromMilliseconds(250 * 10);
         var both = Task.WhenAll(sentTcs.Task, receivedTcs.Task);
-        var completed = await Task.WhenAny(both, Task.Delay(timeout));
-        Assert.Same(both, completed);
+        await AsyncAssert.CompletesWithinAsync(both, timeout, "expected Sequence frames to be sent and received");
         Assert.True(sentCount >= 1, $"Expected at least one Sequence frame sent, got {sentCount}");
         Assert.True(receivedCount >= 1, $"Expected at least one Sequence frame received, got {receivedCount}");
     }

tests/B3.EntryPoint.Conformance/Spec_4_7_Retransmit/NotAppliedTests.cs

@@ -28,9 +28,8 @@ public async Task NotApplied_From_Peer_Surfaces_Event_With_Range()
         var sent = await fx.Peer.InjectNotAppliedAsync(fromSeqNo: 7u, count: 3u);
         Assert.True(sent >= 1, "Expected the NotApplied frame to be written to at least one connection");
 
-        var completed = await Task.WhenAny(na.Task, Task.Delay(TimeSpan.FromSeconds(3)));
-        Assert.Same(na.Task, completed);
-        var evt = await na.Task;
+        var evt = await AsyncAssert.CompletesWithinAsync(
+            na.Task, TimeSpan.FromSeconds(5), "expected a NotAppliedReceived event");
         Assert.Equal(7UL, evt.FromSeqNo);
         Assert.Equal(3u, evt.Count);
     }

tests/B3.EntryPoint.Conformance/Spec_4_7_Retransmit/ReconnectRetransmitTests.cs

@@ -132,42 +132,40 @@ await client.SubmitAsync(new NewOrderRequest
 
         // 7. Submit one more order; assert the peer sees it on the wire with
         //    MsgSeqNum = 6, proving the outbound counter resumed contiguously.
-        await client.SubmitAsync(new NewOrderRequest
-        {
-            ClOrdID = (ClOrdID)6,
-            SecurityId = 4321UL,
-            Side = Side.Buy,
-            OrderType = OrderType.Limit,
-            Price = 10.0m,
-            OrderQty = 100UL,
-        });
-
-        // Wait briefly for the peer to read the frame off the wire (TCS via
-        // a one-shot signal on the MessageReceived event).
+        //
+        // Subscribe the probe BEFORE submitting (previously it was
+        // subscribed only after SubmitAsync returned, with a TryPeek
+        // fallback for anything already enqueued — but the peer could
+        // process the frame and raise MessageReceived in the window
+        // between that TryPeek check and the += subscription, which is
+        // lost forever and made the test hang for the full timeout; see
+        // #245 CI failure). Subscribing first closes that race.
         var seenSix = new TaskCompletionSource<uint>(TaskCreationOptions.RunContinuationsAsynchronously);
         EventHandler<TestPeerMessageEventArgs> probe = (_, args) =>
         {
             if (args.TemplateId != NewOrderSingleData.MESSAGE_ID) return;
             if (!NewOrderSingleData.TryParse(args.Payload.Span, out var reader)) return;
             seenSix.TrySetResult(reader.Data.BusinessHeader.MsgSeqNum.Value);
         };
-        // Drain anything already enqueued from the resubmitted order.
-        if (peerInboundNosSeqs.TryPeek(out var first))
+        fx.Peer.MessageReceived += probe;
+        try
         {
-            seenSix.TrySetResult(first);
+            await client.SubmitAsync(new NewOrderRequest
+            {
+                ClOrdID = (ClOrdID)6,
+                SecurityId = 4321UL,
+                Side = Side.Buy,
+                OrderType = OrderType.Limit,
+                Price = 10.0m,
+                OrderQty = 100UL,
+            });
+
+            await AsyncAssert.CompletesWithinAsync(
+                seenSix.Task, TimeSpan.FromSeconds(5), "expected the resubmitted order to reach the peer");
         }
-        else
+        finally
         {
-            fx.Peer.MessageReceived += probe;
-            try
-            {
-                var completed = await Task.WhenAny(seenSix.Task, Task.Delay(TimeSpan.FromSeconds(3)));
-                Assert.Same(seenSix.Task, completed);
-            }
-            finally
-            {
-                fx.Peer.MessageReceived -= probe;
-            }
+            fx.Peer.MessageReceived -= probe;
         }
         Assert.Equal(6u, await seenSix.Task);
 

tests/B3.EntryPoint.Conformance/Spec_4_7_Retransmit/RetransmitRejectTests.cs

@@ -31,9 +31,8 @@ public async Task Retransmit_Reject_Is_Surfaced_With_Code()
 
         await client.Retransmit.RequestRetransmitAsync(fromSeqNo: 1UL, count: 5U);
 
-        var completed = await Task.WhenAny(rejected.Task, Task.Delay(TimeSpan.FromSeconds(3)));
-        Assert.Same(rejected.Task, completed);
-        var evt = await rejected.Task;
+        var evt = await AsyncAssert.CompletesWithinAsync(
+            rejected.Task, TimeSpan.FromSeconds(5), "expected a RetransmitRejected event");
         Assert.Equal(RetransmitRejectCode.OutOfRange, evt.Code);
     }
 }

tests/B3.EntryPoint.Conformance/Spec_4_7_Retransmit/RetransmitTests.cs

@@ -34,9 +34,8 @@ public async Task Retransmit_Recent_Range_Is_Honoured()
 
         await client.Retransmit.RequestRetransmitAsync(fromSeqNo: 1UL, count: 5U);
 
-        var completed = await Task.WhenAny(received.Task, Task.Delay(TimeSpan.FromSeconds(3)));
-        Assert.Same(received.Task, completed);
-        var evt = await received.Task;
+        var evt = await AsyncAssert.CompletesWithinAsync(
+            received.Task, TimeSpan.FromSeconds(5), "expected a RetransmissionReceived event");
         Assert.Equal(1UL, evt.NextSeqNo);
         Assert.Equal(0u, evt.Count);
     }