diff --git a/tests/FinalFactory.Rendezvous.Tests/Client/RendezvousClientIntegrationTests.cs b/tests/FinalFactory.Rendezvous.Tests/Client/RendezvousClientIntegrationTests.cs index 61c7e6c..1317922 100644 --- a/tests/FinalFactory.Rendezvous.Tests/Client/RendezvousClientIntegrationTests.cs +++ b/tests/FinalFactory.Rendezvous.Tests/Client/RendezvousClientIntegrationTests.cs @@ -1,6 +1,7 @@ using System.Diagnostics; using System.Net; using System.Net.Sockets; +using System.Runtime.CompilerServices; using FinalFactory.Rendezvous.Client; using FinalFactory.Rendezvous.Contracts; using FinalFactory.Rendezvous.Server.Abuse; @@ -146,13 +147,19 @@ public sealed class RendezvousClientIntegrationTests public async Task BrowserStreamResetsInvalidCursorReplaysReconnectAndReleasesConnections() { await using ClientTestHost host = await ClientTestHost.StartAsync(); - RendezvousPublisherClient publisher = new(host.HttpClient); - RendezvousSessionBrowserClient browser = new( - host.HttpClient, - new RendezvousClientOptions - { - RequestTimeout = TimeSpan.FromSeconds(15), - }); + // Budgets are sized for the release pipeline, not for a developer machine. That job + // runs this suite inside a builder container on a busy act_runner host, alongside + // parallel image builds, so the in-process Kestrel host and the HttpClient driving it + // share a thread pool that is routinely starved. A loopback round trip that costs + // microseconds locally can then cost seconds - and the first stream request in the + // process additionally pays the one-time JIT and serializer warm-up of the SSE path. + // 30s is the ceiling RendezvousClientOptions.Validate permits for RequestTimeout. + RendezvousClientOptions loadedRunner = new() + { + RequestTimeout = TimeSpan.FromSeconds(30), + }; + RendezvousPublisherClient publisher = new(host.HttpClient, loadedRunner); + RendezvousSessionBrowserClient browser = new(host.HttpClient, loadedRunner); PublishedSession session = AssertSuccess(await publisher.RegisterAsync( CreateRegistration(200), host.PublisherCredential)); @@ -161,18 +168,22 @@ public sealed class RendezvousClientIntegrationTests BrowseSessionsResponse snapshot = AssertSuccess(await browser.BrowseAsync(request)); Assert.False(string.IsNullOrWhiteSpace(snapshot.StreamCursor)); - using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(30)); - await using (IAsyncEnumerator> invalid = browser - .StreamAsync(request, CorruptCursor(snapshot.StreamCursor), timeout.Token) + using CancellationTokenSource timeout = new(TimeSpan.FromMinutes(2)); + await using (IAsyncEnumerator> invalid = + StreamWithOpenRetry( + browser, + request, + CorruptCursor(snapshot.StreamCursor), + timeout.Token) .GetAsyncEnumerator(timeout.Token)) { Assert.True(await invalid.MoveNextAsync()); Assert.Equal(SessionStreamEventKind.Reset, AssertSuccess(invalid.Current).Kind); Assert.False(await invalid.MoveNextAsync()); } - await using IAsyncEnumerator> events = browser - .StreamAsync(request, snapshot.StreamCursor, timeout.Token) - .GetAsyncEnumerator(timeout.Token); + await using IAsyncEnumerator> events = + StreamWithOpenRetry(browser, request, snapshot.StreamCursor, timeout.Token) + .GetAsyncEnumerator(timeout.Token); Task upsertPending = events.MoveNextAsync().AsTask(); Assert.True((await publisher.UpdateAsync( session, @@ -190,8 +201,8 @@ public sealed class RendezvousClientIntegrationTests Assert.Equal(SessionStreamEventKind.SessionUpsert, upsert.Kind); Assert.Equal("Live update", upsert.Session!.DisplayName); - await using (IAsyncEnumerator> replay = browser - .StreamAsync(request, snapshot.StreamCursor, timeout.Token) + await using (IAsyncEnumerator> replay = + StreamWithOpenRetry(browser, request, snapshot.StreamCursor, timeout.Token) .GetAsyncEnumerator(timeout.Token)) { Assert.True(await replay.MoveNextAsync()); @@ -212,6 +223,68 @@ public sealed class RendezvousClientIntegrationTests Assert.Equal(session.ListingId, remove.ListingId); } + // Opening an event stream is a single, unretried request in the SDK, so a handshake that + // loses its wall-clock budget on a saturated runner surfaces as a typed ServiceUnavailable + // first element instead of the expected event. Reopening is semantically free: every call + // site passes a replayable snapshot cursor, so a reopened stream observes exactly the + // events the first attempt would have delivered. + // + // The retry cannot hide a product regression. StreamSessions never answers with a + // ServiceUnavailable envelope - its only rejections are InvalidRequest, + // UnsupportedContractVersion, CapacityExceeded and RateLimited - so this code path is + // reachable only from the transport catch in RendezvousHttpTransport.OpenStreamAsync, + // i.e. a timed-out or dropped handshake. Attempts are bounded, and the final failure is + // yielded verbatim, so a stream that is genuinely unopenable still fails the test with the + // original message. + private const int StreamOpenAttempts = 3; + + private static async IAsyncEnumerable> StreamWithOpenRetry( + RendezvousSessionBrowserClient browser, + BrowseSessionsRequest request, + string streamCursor, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + for (int attempt = 1; ; attempt++) + { + bool reopen = false; + await using (IAsyncEnumerator> source = browser + .StreamAsync(request, streamCursor, cancellationToken) + .GetAsyncEnumerator(cancellationToken)) + { + bool opening = true; + while (await source.MoveNextAsync()) + { + RendezvousClientResult current = source.Current; + if (opening + && !current.IsSuccess + && current.Error == RendezvousErrorCode.ServiceUnavailable + && attempt < StreamOpenAttempts) + { + reopen = true; + break; + } + + opening = false; + // Keepalives are protocol filler with no session semantics. The server emits + // one whenever a subscription idles for its keepalive interval, which a slow + // runner reaches between the assertions below; dropping them keeps the + // reset/upsert/remove expectations exact. + if (current.IsSuccess && current.Value!.Kind == SessionStreamEventKind.Keepalive) + { + continue; + } + + yield return current; + } + } + + if (!reopen) + { + yield break; + } + } + } + private static string CorruptCursor(string cursor) { char replacement = cursor[^1] == 'a' ? 'b' : 'a';