Skip to content

Make Phase0 service tests deterministic under parallel load #107

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Xueyang-Song merged 3 commits into from
Oct 28, 2026
Merged

Make Phase0 service tests deterministic under parallel load #107

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f3a287b
Make Phase0 service tests deterministic under parallel load
Xueyang-Song Oct 28, 2026
be4bfa5
Merge branch 'main' into claude/serene-napier-f42b74
Xueyang-Song Oct 28, 2026
f2da032
Merge branch 'main' into claude/serene-napier-f42b74
Xueyang-Song Oct 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 15 additions & 24 deletions JitHub.WinUI.Tests/Services/Phase0QueryServiceTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
using System.Threading.Tasks;
using JitHub.Services;
using JitHub.WinUI.Tests.TestDoubles;
using JitHub.WinUI.Tests.TestSupport;
using Xunit;

namespace JitHub.WinUI.Tests.Services;
Expand Down Expand Up @@ -63,7 +64,7 @@ public async Task GetAsync_StaleCacheReturnsCachedValueBeforeBackgroundRefresh()
Assert.True(result.IsRefreshInProgress);

release.SetResult();
await WaitForAsync(() => transport.CallCount > 0);
await AsyncTestWait.UntilAsync(() => transport.CallCount > 0);
}

[Fact]
Expand Down Expand Up @@ -143,7 +144,7 @@ public async Task GetAsync_StalePrefetchPreservesPrefetchLaneDuringBackgroundRef
GitHubQueryService service = CreateService(store, transport);

CachedResult<Phase0TestPayload> result = await service.GetAsync(query, QueryFetchPolicy.StaleFirst);
await WaitForAsync(() => transport.LastRequest is not null);
await AsyncTestWait.UntilAsync(() => transport.LastRequest is not null);

Assert.Equal("cached", result.Value!.Name);
Assert.Equal(GitHubRequestPriority.Prefetch, transport.LastRequest!.Priority);
Expand Down Expand Up @@ -171,12 +172,12 @@ public async Task GetAsync_NetworkOnlyPromotesMatchingQueuedStaleRefreshToForegr
GitHubQueryService service = new(store, transport, queue, new NoopTelemetryService());
CachedResult<Phase0TestPayload> stale = await service.GetAsync(query, QueryFetchPolicy.StaleFirst);
Assert.Equal("cached", stale.Value!.Name);
await WaitForAsync(() => queue.InFlightCount == 2);
await AsyncTestWait.UntilAsync(() => queue.InFlightCount == 2);

GitHubQuery<Phase0TestPayload> foregroundQuery = query with { Priority = GitHubRequestPriority.UserInitiated };
CachedResult<Phase0TestPayload> refreshed = await service
.GetAsync(foregroundQuery, QueryFetchPolicy.NetworkOnly)
.WaitAsync(TimeSpan.FromSeconds(2));
.WaitAsync(AsyncTestWait.HangGuard);

Assert.Equal("network", refreshed.Value!.Name);
Assert.Equal(1, transport.CallCount);
Expand All @@ -199,7 +200,7 @@ public async Task GetAsync_CancelledForegroundJoinDoesNotCancelStaleBackgroundRe

CachedResult<Phase0TestPayload> stale = await service.GetAsync(query, QueryFetchPolicy.StaleFirst);
Assert.Equal("cached", stale.Value!.Name);
await WaitForAsync(() => transport.CallCount == 1);
await AsyncTestWait.UntilAsync(() => transport.CallCount == 1);

using CancellationTokenSource callerCancellation = new();
GitHubQuery<Phase0TestPayload> foregroundQuery = query with { Priority = GitHubRequestPriority.UserInitiated };
Expand All @@ -213,7 +214,7 @@ public async Task GetAsync_CancelledForegroundJoinDoesNotCancelStaleBackgroundRe
Assert.False(transport.LastCancellationToken.IsCancellationRequested);

releaseTransport.SetResult();
await WaitForAsync(() => queue.InFlightCount == 0);
await AsyncTestWait.UntilAsync(() => queue.InFlightCount == 0);
Assert.False(transport.LastCancellationToken.IsCancellationRequested);
Assert.Equal(1, transport.CallCount);
}
Expand Down Expand Up @@ -357,12 +358,17 @@ public async Task RefreshAsync_AccountCleanupCancelsDelayedCachePutBeforePartiti
GitHubQueryService service = new(store, transport, queue, new NoopTelemetryService());
GitHubQuery<Phase0TestPayload> query = CreateQuery("test/cleanup-race", TimeSpan.FromMinutes(5));

// The final assertion relies on the partition starting empty. Checking that here
// also creates the SQLite schema up front, so slow first-use disk I/O on a loaded
// runner is not part of the refresh being raced below.
Assert.Null(await innerStore.TryGetAsync(query));

Task<CachedResult<Phase0TestPayload>> refresh = service.RefreshAsync(query);
await store.PutStarted.Task.WaitAsync(TimeSpan.FromSeconds(2));
await store.PutStarted.Task.WaitAsync(AsyncTestWait.HangGuard);

Task quiesce = accountWork.QuiesceAsync("u1");
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => refresh);
await quiesce.WaitAsync(TimeSpan.FromSeconds(2));
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => refresh.WaitAsync(AsyncTestWait.HangGuard));
await quiesce.WaitAsync(AsyncTestWait.HangGuard);
await store.ClearPartitionAsync("u1");
store.ReleasePut.TrySetResult();

Expand Down Expand Up @@ -420,21 +426,6 @@ private static GitHubRestResponse<Phase0TestPayload> CreateResponse(
RetryAfter: null,
FetchedAt: DateTimeOffset.UtcNow);

private static async Task WaitForAsync(Func<bool> predicate)
{
for (int i = 0; i < 50; i++)
{
if (predicate())
{
return;
}

await Task.Delay(20);
}

Assert.True(predicate());
}

private sealed class FakeTransport : IGitHubRestTransport
{
private readonly Phase0TestPayload? _payload;
Expand Down
67 changes: 38 additions & 29 deletions JitHub.WinUI.Tests/Services/Phase0RequestTransportTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
using System.Threading;
using System.Threading.Tasks;
using JitHub.Services;
using JitHub.WinUI.Tests.TestSupport;
using Xunit;

namespace JitHub.WinUI.Tests.Services;
Expand Down Expand Up @@ -74,7 +75,7 @@ public async Task RequestQueue_DoesNotDedupeMutationsWithIdenticalKeys()
return 84;
});

await bothStarted.Task.WaitAsync(TimeSpan.FromSeconds(2));
await bothStarted.Task.WaitAsync(AsyncTestWait.HangGuard);
release.SetResult();

int[] results = await Task.WhenAll(first, second);
Expand All @@ -99,9 +100,9 @@ public async Task RequestQueue_ForegroundLaneIsNeverBlockedByQueuedPrefetchWork(
Task<int> queuedBackground = queue.EnqueueAsync("prefetch-queued", GitHubRequestPriority.BackgroundRefresh, _ => Task.FromResult(2));
Task<int> foreground = queue.EnqueueAsync("foreground", GitHubRequestPriority.UserInitiated, _ => Task.FromResult(3));

Task completed = await Task.WhenAny(foreground, Task.Delay(TimeSpan.FromSeconds(2)));
Assert.Same(foreground, completed);
Assert.Equal(3, await foreground);
// The background lane stays held until releaseBackground is set below, so a
// foreground read stuck behind background work would never finish here.
Assert.Equal(3, await foreground.WaitAsync(AsyncTestWait.HangGuard));
Assert.False(queuedBackground.IsCompleted);

releaseBackground.SetResult();
Expand Down Expand Up @@ -130,16 +131,16 @@ public async Task RequestQueue_ForegroundJoinPromotesQueuedBackgroundDedupe()
Interlocked.Increment(ref calls);
return Task.FromResult(42);
});
await WaitUntilAsync(() => queue.InFlightCount == 2);
await AsyncTestWait.UntilAsync(() => queue.InFlightCount == 2);

Task<int> foreground = queue.EnqueueAsync("shared-read", GitHubRequestPriority.UserInitiated, _ =>
{
Interlocked.Increment(ref calls);
return Task.FromResult(84);
});

Assert.Equal(42, await foreground.WaitAsync(TimeSpan.FromSeconds(2)));
Assert.Equal(42, await background.WaitAsync(TimeSpan.FromSeconds(2)));
Assert.Equal(42, await foreground.WaitAsync(AsyncTestWait.HangGuard));
Assert.Equal(42, await background.WaitAsync(AsyncTestWait.HangGuard));
Assert.Equal(1, calls);
Assert.False(blocker.IsCompleted);

Expand Down Expand Up @@ -186,26 +187,43 @@ public async Task RequestQueue_LastPrefetchSubscriberCancellationStopsActiveWork
{
GitHubRequestQueue queue = new(foregroundReadConcurrency: 1, backgroundReadConcurrency: 1, mutationConcurrency: 1);
TaskCompletionSource workStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
TaskCompletionSource transportCancelled = new(TaskCreationOptions.RunContinuationsAsynchronously);
TaskCompletionSource workStoppedByCancellation = new(TaskCreationOptions.RunContinuationsAsynchronously);
CancellationToken workToken = default;
using CancellationTokenSource callerCancellation = new();
Task<int> prefetch = queue.EnqueueAsync(
"cancel-active-prefetch",
GitHubRequestPriority.Prefetch,
async token =>
{
using CancellationTokenRegistration registration = token.Register(transportCancelled.SetResult);
workToken = token;
workStarted.SetResult();
await Task.Delay(Timeout.InfiniteTimeSpan, token);
try
{
await Task.Delay(Timeout.InfiniteTimeSpan, token);
}
catch (OperationCanceledException) when (token.IsCancellationRequested)
{
// Signal from the work's own cancellation path. A token.Register
// callback scoped to this method can be unregistered by the resumed
// work before the token gets to run it, which loses the signal.
workStoppedByCancellation.SetResult();
throw;
}

return 1;
},
callerCancellation.Token);
await workStarted.Task.WaitAsync(TimeSpan.FromSeconds(10));
await workStarted.Task.WaitAsync(AsyncTestWait.HangGuard);
Assert.False(workToken.IsCancellationRequested);

callerCancellation.Cancel();

await Assert.ThrowsAnyAsync<OperationCanceledException>(() => prefetch);
await transportCancelled.Task.WaitAsync(TimeSpan.FromSeconds(10));
await WaitUntilAsync(() => queue.InFlightCount == 0);
// The caller only observes cancellation after the queue has dropped its last
// subscriber, and dropping the last subscriber cancels the shared work token.
Assert.True(workToken.IsCancellationRequested);
await workStoppedByCancellation.Task.WaitAsync(AsyncTestWait.HangGuard);
await AsyncTestWait.UntilAsync(() => queue.InFlightCount == 0, "the cancelled prefetch to leave the in-flight table");
}

[Fact]
Expand Down Expand Up @@ -233,11 +251,11 @@ public async Task RequestQueue_CancelledQueuedPrefetchIsRemovedBeforeTransportSt
return Task.FromResult(2);
},
callerCancellation.Token);
await WaitUntilAsync(() => queue.InFlightCount == 2);
await AsyncTestWait.UntilAsync(() => queue.InFlightCount == 2);

callerCancellation.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => queued);
await WaitUntilAsync(() => queue.InFlightCount == 1);
await AsyncTestWait.UntilAsync(() => queue.InFlightCount == 1);
releaseBlocker.SetResult();

Assert.Equal(1, await blocker);
Expand Down Expand Up @@ -280,13 +298,13 @@ public async Task RequestQueue_CancelledPrefetchDoesNotCancelPromotedForegroundS
"promoted-shared-read",
GitHubRequestPriority.UserInitiated,
_ => Task.FromResult(84));
await sharedStarted.Task.WaitAsync(TimeSpan.FromSeconds(2));
await sharedStarted.Task.WaitAsync(AsyncTestWait.HangGuard);
prefetchCancellation.Cancel();

await Assert.ThrowsAnyAsync<OperationCanceledException>(() => prefetch);
Assert.False(sharedTokenCancelled);
releaseShared.SetResult();
Assert.Equal(42, await foreground.WaitAsync(TimeSpan.FromSeconds(2)));
Assert.Equal(42, await foreground.WaitAsync(AsyncTestWait.HangGuard));
Assert.Equal(1, Volatile.Read(ref transportCalls));
Assert.False(sharedTokenCancelled);
Assert.False(blocker.IsCompleted);
Expand Down Expand Up @@ -330,15 +348,15 @@ public async Task LatestWinsHoverStorm_DoesNotAccumulateQueuedNetworkRequests()
});
}

await finalStarted.Task.WaitAsync(TimeSpan.FromSeconds(2));
Task<int> finalRequest = await finalRequestAssigned.Task.WaitAsync(TimeSpan.FromSeconds(2));
await finalStarted.Task.WaitAsync(AsyncTestWait.HangGuard);
Task<int> finalRequest = await finalRequestAssigned.Task.WaitAsync(AsyncTestWait.HangGuard);
Assert.Equal(1, Volatile.Read(ref transportCalls));
Assert.Equal(99, Volatile.Read(ref startedItem));
Assert.Equal(1, queue.InFlightCount);

scheduler.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => finalRequest);
await WaitUntilAsync(() => queue.InFlightCount == 0);
await AsyncTestWait.UntilAsync(() => queue.InFlightCount == 0);
}

[Fact]
Expand Down Expand Up @@ -532,15 +550,6 @@ protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage reques
Task.FromResult(_handler(request));
}

private static async Task WaitUntilAsync(Func<bool> predicate)
{
using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(2));
while (!predicate())
{
await Task.Delay(10, timeout.Token);
}
}

private sealed class CancellationHandle(CancellationTokenSource cancellation) : IDisposable
{
private CancellationTokenSource? _cancellation = cancellation;
Expand Down
25 changes: 13 additions & 12 deletions JitHub.WinUI.Tests/Services/Phase0TelemetryTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
using JitHub.Services;
using JitHub.Services.CodeViewer;
using JitHub.WinUI.Tests.TestDoubles;
using JitHub.WinUI.Tests.TestSupport;
using Xunit;

namespace JitHub.WinUI.Tests.Services;
Expand Down Expand Up @@ -700,7 +701,7 @@ public async Task StoreTelemetrySink_PacesNativeLoggerCallsOffTheCallerThread()
sink.TrackEvent("shell.route.opened");
sink.TrackEvent("dashboard.opened");

Assert.True(await sink.WaitForIdleAsync(TimeSpan.FromSeconds(2)));
Assert.True(await sink.WaitForIdleAsync(AsyncTestWait.HangGuard));
Assert.Equal(3, dispatchTimes.Count);
Assert.True(dispatchTimes[1] - dispatchTimes[0] >= 40);
Assert.True(dispatchTimes[2] - dispatchTimes[1] >= 40);
Expand All @@ -724,14 +725,14 @@ public async Task StoreTelemetrySink_CoalescesPendingNamesAndBoundsStartupBursts
if (name == "app.started")
{
firstDispatchStarted.Set();
Assert.True(releaseFirstDispatch.Wait(TimeSpan.FromSeconds(2)));
Assert.True(releaseFirstDispatch.Wait(AsyncTestWait.HangGuard));
}
},
TimeSpan.Zero,
queueCapacity: 2);

sink.TrackEvent("app.started");
Assert.True(firstDispatchStarted.Wait(TimeSpan.FromSeconds(2)));
Assert.True(firstDispatchStarted.Wait(AsyncTestWait.HangGuard));
sink.TrackEvent("shell.route.opened");
sink.TrackEvent("shell.route.opened");
sink.TrackEvent("dashboard.opened");
Expand All @@ -742,7 +743,7 @@ public async Task StoreTelemetrySink_CoalescesPendingNamesAndBoundsStartupBursts
Assert.Equal(3, sink.PendingEventCount);

releaseFirstDispatch.Set();
Assert.True(await sink.WaitForIdleAsync(TimeSpan.FromSeconds(2)));
Assert.True(await sink.WaitForIdleAsync(AsyncTestWait.HangGuard));
Assert.Equal(
["app.started", "shell.route.opened", "dashboard.opened"],
dispatched);
Expand Down Expand Up @@ -784,8 +785,8 @@ public async Task LocalDiagnosticsStore_PreservesQueuedAppendOrder()
.Select(index => store.AppendAsync(CreateDiagnostic($"ordered-{index:D2}")))
.ToArray();

await Task.WhenAll(appends).WaitAsync(TimeSpan.FromSeconds(5));
await store.FlushAsync().WaitAsync(TimeSpan.FromSeconds(5));
await Task.WhenAll(appends).WaitAsync(AsyncTestWait.HangGuard);
await store.FlushAsync().WaitAsync(AsyncTestWait.HangGuard);

IReadOnlyList<LocalDiagnosticEvent> entries = await store.ReadAsync();
Assert.Equal(
Expand Down Expand Up @@ -823,7 +824,7 @@ await store.AppendAsync(new LocalDiagnosticEvent(
})
.ToArray();

await Task.WhenAll(producers).WaitAsync(TimeSpan.FromSeconds(10));
await Task.WhenAll(producers).WaitAsync(AsyncTestWait.HangGuard);
IReadOnlyList<LocalDiagnosticEvent> entries = await store.ReadAsync();

Assert.Equal(producerCount * eventsPerProducer, entries.Count);
Expand Down Expand Up @@ -852,7 +853,7 @@ public async Task LocalDiagnosticsStore_FlushIsAnOrderedBarrierForAcceptedNonBlo
.ToArray();

Assert.All(accepted, Assert.True);
await store.FlushAsync().WaitAsync(TimeSpan.FromSeconds(5));
await store.FlushAsync().WaitAsync(AsyncTestWait.HangGuard);

string[] lines = await File.ReadAllLinesAsync(path);
Assert.Equal(96, lines.Length);
Expand Down Expand Up @@ -882,7 +883,7 @@ public async Task LocalDiagnosticsStore_SaturationCountsDropsAndWritesCoalescedO
}
}

await store.FlushAsync().WaitAsync(TimeSpan.FromSeconds(10));
await store.FlushAsync().WaitAsync(AsyncTestWait.HangGuard);
IReadOnlyList<LocalDiagnosticEvent> entries = await store.ReadAsync();
LocalDiagnosticEvent[] overflowSignals = entries
.Where(static entry => entry.Name == "diagnostics.queue.overflow")
Expand Down Expand Up @@ -910,9 +911,9 @@ public async Task LocalDiagnosticsStore_ShutdownDrainsAcceptedOperationsAndRejec
.Select(index => store.AppendAsync(CreateDiagnostic($"shutdown-{index:D3}")))
.ToArray();

await store.ShutdownAsync().WaitAsync(TimeSpan.FromSeconds(10));
await Task.WhenAll(appends).WaitAsync(TimeSpan.FromSeconds(2));
await store.FlushAsync().WaitAsync(TimeSpan.FromSeconds(2));
await store.ShutdownAsync().WaitAsync(AsyncTestWait.HangGuard);
await Task.WhenAll(appends).WaitAsync(AsyncTestWait.HangGuard);
await store.FlushAsync().WaitAsync(AsyncTestWait.HangGuard);

string[] lines = await File.ReadAllLinesAsync(path);
Assert.Equal(100, lines.Length);
Expand Down
Loading
Loading