test(client): survive a starved stream handshake (#1)

Opening an SSE stream is a single unretried request bounded by
RendezvousClientOptions.RequestTimeout. On the release runner the suite
shares a builder container with parallel image builds, so the loopback
handshake can lose its whole wall-clock budget to thread-pool starvation;
the client then reports a typed ServiceUnavailable that the test asserted
against as if it were the reset event.

Retry the open from the same replayable cursor, raise the per-request
budget to the contract ceiling, and drop protocol keepalives so a slow
interval between assertions cannot be mistaken for a session event. The
retry only fires on the transport failure the stream endpoint can never
produce as an envelope, is bounded, and yields the final failure verbatim,
so a genuinely broken stream still fails.

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);
+        // Budgets are sized for the release pipeline, not for a developer machine. That job
-        RendezvousSessionBrowserClient browser = new(
+        // runs this suite inside a builder container on a busy act_runner host, alongside
-            host.HttpClient,
+        // parallel image builds, so the in-process Kestrel host and the HttpClient driving it
-            new RendezvousClientOptions
+        // 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(15),
+            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,17 +168,21 @@ public sealed class RendezvousClientIntegrationTests
         BrowseSessionsResponse snapshot = AssertSuccess(await browser.BrowseAsync(request));
         Assert.False(string.IsNullOrWhiteSpace(snapshot.StreamCursor));
 
-        using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(30));
+        using CancellationTokenSource timeout = new(TimeSpan.FromMinutes(2));
-        await using (IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> invalid = browser
+        await using (IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> invalid =
-                         .StreamAsync(request, CorruptCursor(snapshot.StreamCursor), timeout.Token)
+                     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<RendezvousClientResult<SessionStreamEvent>> events = browser
+        await using IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> events =
-            .StreamAsync(request, snapshot.StreamCursor, timeout.Token)
+            StreamWithOpenRetry(browser, request, snapshot.StreamCursor, timeout.Token)
                 .GetAsyncEnumerator(timeout.Token);
         Task<bool> upsertPending = events.MoveNextAsync().AsTask();
         Assert.True((await publisher.UpdateAsync(
@@ -190,8 +201,8 @@ public sealed class RendezvousClientIntegrationTests
         Assert.Equal(SessionStreamEventKind.SessionUpsert, upsert.Kind);
         Assert.Equal("Live update", upsert.Session!.DisplayName);
 
-        await using (IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> replay = browser
+        await using (IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> replay =
-                         .StreamAsync(request, snapshot.StreamCursor, timeout.Token)
+                     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<RendezvousClientResult<SessionStreamEvent>> StreamWithOpenRetry(
+        RendezvousSessionBrowserClient browser,
+        BrowseSessionsRequest request,
+        string streamCursor,
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        for (int attempt = 1; ; attempt++)
+        {
+            bool reopen = false;
+            await using (IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> source = browser
+                             .StreamAsync(request, streamCursor, cancellationToken)
+                             .GetAsyncEnumerator(cancellationToken))
+            {
+                bool opening = true;
+                while (await source.MoveNextAsync())
+                {
+                    RendezvousClientResult<SessionStreamEvent> 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';