feat(browser): stream bounded live session updates (#26)

tests/FinalFactory.Rendezvous.Tests/Browser/SessionBrowserTestData.cs

@@ -7,18 +7,25 @@ namespace FinalFactory.Rendezvous.Tests.Browser;
 
 internal sealed class SessionBrowserFixture : IDisposable
 {
-    private readonly EphemeralStateFixture _state = new();
+    private readonly EphemeralStateFixture _state;
 
     public SessionBrowserFixture()
     {
+        Changes = new(new SessionChangeJournalOptions());
+        _state = new(changes: Changes);
         Cursors = new();
-        Browser = new(_state.Store, Cursors, _state.Clock);
+        StreamCursors = new();
+        Browser = new(_state.Store, Cursors, StreamCursors, Changes, _state.Clock);
+        Streams = new(Changes, StreamCursors, _state.Clock);
     }
 
     public InMemoryEphemeralRendezvousStore Store => _state.Store;
     public ManualRendezvousClock Clock => _state.Clock;
     public SessionBrowserCursorCodec Cursors { get; }
+    public SessionStreamCursorCodec StreamCursors { get; }
+    public SessionChangeJournal Changes { get; }
     public SessionBrowserService Browser { get; }
+    public SessionStreamService Streams { get; }
     public TenantScope Scope => _state.Scope;
 
     public StoredListing Add(
@@ -67,5 +74,9 @@ internal sealed class SessionBrowserFixture : IDisposable
         PageSize = pageSize,
     };
 
-    public void Dispose() => Cursors.Dispose();
+    public void Dispose()
+    {
+        Cursors.Dispose();
+        StreamCursors.Dispose();
+    }
 }

tests/FinalFactory.Rendezvous.Tests/Browser/SessionStreamServiceTests.cs

@@ -0,0 +1,325 @@
+using FinalFactory.Rendezvous.Contracts;
+using FinalFactory.Rendezvous.Server.Browser;
+using FinalFactory.Rendezvous.Server.State;
+using FinalFactory.Rendezvous.Tests.State;
+
+namespace FinalFactory.Rendezvous.Tests.Browser;
+
+public sealed class SessionStreamServiceTests
+{
+    [Fact]
+    public void SnapshotPlusUpdateMatchesFreshProjection()
+    {
+        using SessionBrowserFixture fixture = new();
+        StoredListing listing = fixture.Add();
+        BrowseSessionsRequest request = fixture.Request();
+        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
+        using SessionStreamSubscription subscription = AssertSuccess(
+            fixture.Streams.Subscribe(request, snapshot.StreamCursor));
+
+        StoreResult<StoredListing> updated = fixture.Store.UpdateListing(Update(
+            listing,
+            displayName: "Updated host",
+            currentPlayers: 4));
+        Assert.True(updated.Succeeded);
+        SessionStreamEvent delta = Assert.Single(fixture.Streams.Read(subscription).Events);
+        Assert.Equal(SessionStreamEventKind.SessionUpsert, delta.Kind);
+        Assert.Equal("Updated host", delta.Session!.DisplayName);
+        Assert.Equal(4, delta.Session.Capacity.CurrentPlayers);
+
+        BrowseSessionsResponse fresh = AssertSuccess(fixture.Browser.Browse(request));
+        SessionListing expected = Assert.Single(fresh.Items);
+        Assert.Equal(expected.DisplayName, delta.Session.DisplayName);
+        Assert.Equal(expected.Capacity.CurrentPlayers, delta.Session.Capacity.CurrentPlayers);
+        Assert.DoesNotContain("lease", System.Text.Json.JsonSerializer.Serialize(delta, ContractJson.Options), StringComparison.OrdinalIgnoreCase);
+    }
+
+    [Fact]
+    public void PresenceStalenessRecoveryAndRevocationProduceRemoveUpsertRemove()
+    {
+        using SessionBrowserFixture fixture = new();
+        StoredListing listing = fixture.Add();
+        BrowseSessionsRequest request = fixture.Request();
+        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
+        using SessionStreamSubscription subscription = AssertSuccess(
+            fixture.Streams.Subscribe(request, snapshot.StreamCursor));
+
+        fixture.Clock.Advance(TimeSpan.FromSeconds(21));
+        AssertSuccess(fixture.Browser.Browse(request));
+        SessionStreamEvent stale = Assert.Single(fixture.Streams.Read(subscription).Events);
+        Assert.Equal(SessionStreamEventKind.SessionRemove, stale.Kind);
+        Assert.Equal(listing.Definition.ListingId, stale.ListingId);
+
+        StoreResult<StoredListing> rebound = fixture.Store.BindHostPresence(new(
+            listing.Definition.HostPresenceHandle,
+            listing.Definition.HostPresenceFingerprint,
+            new ObservedEndpoint(AddressFamilyKind.Ipv4, "203.0.113.20", 40_020),
+            null));
+        Assert.True(rebound.Succeeded);
+        Assert.Equal(
+            SessionStreamEventKind.SessionUpsert,
+            Assert.Single(fixture.Streams.Read(subscription).Events).Kind);
+
+        Assert.True(fixture.Store.RevokeListing(listing.Definition.ListingId).Succeeded);
+        Assert.Equal(
+            SessionStreamEventKind.SessionRemove,
+            Assert.Single(fixture.Streams.Read(subscription).Events).Kind);
+    }
+
+    [Fact]
+    public void CreationAndLeaseExpiryProduceUpsertThenRemove()
+    {
+        using SessionBrowserFixture fixture = new();
+        BrowseSessionsRequest request = fixture.Request();
+        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
+        using SessionStreamSubscription subscription = AssertSuccess(
+            fixture.Streams.Subscribe(request, snapshot.StreamCursor));
+
+        StoredListing listing = fixture.Add();
+        SessionStreamEvent created = Assert.Single(fixture.Streams.Read(subscription).Events);
+        Assert.Equal(SessionStreamEventKind.SessionUpsert, created.Kind);
+        Assert.Equal(listing.Definition.ListingId, created.Session!.ListingId);
+
+        for (int refresh = 0; refresh < 3; refresh++)
+        {
+            fixture.Clock.Advance(TimeSpan.FromSeconds(19));
+            Assert.True(fixture.Store.BindHostPresence(new(
+                listing.Definition.HostPresenceHandle,
+                listing.Definition.HostPresenceFingerprint,
+                new ObservedEndpoint(AddressFamilyKind.Ipv4, "203.0.113.20", 40_020),
+                null)).Succeeded);
+        }
+        fixture.Clock.Advance(TimeSpan.FromSeconds(4));
+        AssertSuccess(fixture.Browser.Browse(request));
+
+        SessionStreamEvent expired = Assert.Single(fixture.Streams.Read(subscription).Events);
+        Assert.Equal(SessionStreamEventKind.SessionRemove, expired.Kind);
+        Assert.Equal(listing.Definition.ListingId, expired.ListingId);
+        Assert.Equal(
+            StoreResultCode.NotFound,
+            fixture.Store.GetListing(listing.Definition.ListingId, requireFreshPresence: false).Code);
+    }
+
+    [Fact]
+    public void ScopeProtocolRegionAndFullFiltersNeverLeak()
+    {
+        using SessionBrowserFixture fixture = new();
+        BrowseSessionsRequest request = fixture.Request();
+        request.ExcludeFull = true;
+        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
+        using SessionStreamSubscription subscription = AssertSuccess(
+            fixture.Streams.Subscribe(request, snapshot.StreamCursor));
+
+        fixture.Add(scope: new(new("other-game"), fixture.Scope.EnvironmentId));
+        fixture.Add(protocolVersion: 99);
+        fixture.Add(regionId: new("other-region"));
+        fixture.Add(currentPlayers: 8, maximumPlayers: 8);
+        Assert.Empty(fixture.Streams.Read(subscription).Events);
+    }
+
+    [Fact]
+    public void VisibilityCompatibilityAndRegionChangesEnterAndLeaveTheFilter()
+    {
+        using SessionBrowserFixture fixture = new();
+        StoredListing listing = fixture.Add();
+        BrowseSessionsRequest request = fixture.Request();
+        BrowseSessionsResponse snapshot = AssertSuccess(fixture.Browser.Browse(request));
+        using SessionStreamSubscription subscription = AssertSuccess(
+            fixture.Streams.Subscribe(request, snapshot.StreamCursor));
+
+        listing = fixture.Store.UpdateListing(Update(
+            listing,
+            listing.Definition.DisplayName,
+            1,
+            visibility: ListingVisibility.Unlisted)).Value!;
+        Assert.Equal(SessionStreamEventKind.SessionRemove, SingleKind(fixture, subscription));
+        listing = fixture.Store.UpdateListing(Update(
+            listing,
+            listing.Definition.DisplayName,
+            1,
+            visibility: ListingVisibility.Public)).Value!;
+        Assert.Equal(SessionStreamEventKind.SessionUpsert, SingleKind(fixture, subscription));
+        listing = fixture.Store.UpdateListing(Update(
+            listing,
+            listing.Definition.DisplayName,
+            1,
+            protocolVersion: 99)).Value!;
+        Assert.Equal(SessionStreamEventKind.SessionRemove, SingleKind(fixture, subscription));
+        listing = fixture.Store.UpdateListing(Update(
+            listing,
+            listing.Definition.DisplayName,
+            1,
+            protocolVersion: 7)).Value!;
+        Assert.Equal(SessionStreamEventKind.SessionUpsert, SingleKind(fixture, subscription));
+        listing = fixture.Store.UpdateListing(Update(
+            listing,
+            listing.Definition.DisplayName,
+            1,
+            regionId: new RegionId("other-region"))).Value!;
+        Assert.Equal(SessionStreamEventKind.SessionRemove, SingleKind(fixture, subscription));
+    }
+
+    [Fact]
+    public void ReplayGapAndForeignCursorForceResetAndSubscriberLimitFailsClosed()
+    {
+        ManualRendezvousClock clock = new();
+        SessionChangeJournal changes = new(new SessionChangeJournalOptions
+        {
+            ReplayCapacity = 64,
+            MaximumSubscribers = 1,
+            MaximumSubscribersPerTenant = 1,
+        });
+        using SessionStreamCursorCodec cursors = new();
+        SessionStreamService streams = new(changes, cursors, clock);
+        EphemeralStateFixture state = new(changes: changes);
+        StoredListing listing = state.CreateVisibleListing(out _);
+        BrowseSessionsRequest request = new()
+        {
+            GameId = state.Scope.GameId,
+            EnvironmentId = state.Scope.EnvironmentId,
+            ProtocolVersion = listing.Definition.ProtocolVersion,
+        };
+        VisibleListingQuery query = new(state.Scope, listing.Definition.ProtocolVersion, null);
+        string initial = cursors.Encode(query, changes.CurrentRevision, clock.UtcNow);
+        using SessionStreamSubscription subscription = AssertSuccess(streams.Subscribe(request, initial));
+        Assert.Equal(
+            RendezvousErrorCode.CapacityExceeded,
+            streams.Subscribe(request, initial).Error);
+
+        for (int index = 0; index < 65; index++)
+        {
+            listing = state.Store.UpdateListing(Update(
+                listing,
+                displayName: $"Host {index}",
+                currentPlayers: index % 8)).Value!;
+        }
+        Assert.True(streams.Read(subscription).RequiresReset);
+        subscription.Dispose();
+
+        using SessionStreamSubscription foreign = AssertSuccess(streams.Subscribe(
+            request,
+            "not-a-valid-cursor"));
+        Assert.True(streams.Read(foreign).RequiresReset);
+    }
+
+    [Fact]
+    public void BurstIsBoundedAndCoalescedWithoutLosingFinalState()
+    {
+        ManualRendezvousClock clock = new();
+        SessionChangeJournal changes = new(new SessionChangeJournalOptions
+        {
+            ReplayCapacity = 1024,
+            MaximumBatchSize = 128,
+        });
+        using SessionStreamCursorCodec cursors = new();
+        SessionStreamService streams = new(changes, cursors, clock);
+        EphemeralStateFixture state = new(changes: changes);
+        StoredListing listing = state.CreateVisibleListing(out _);
+        BrowseSessionsRequest request = new()
+        {
+            GameId = state.Scope.GameId,
+            EnvironmentId = state.Scope.EnvironmentId,
+            ProtocolVersion = listing.Definition.ProtocolVersion,
+        };
+        VisibleListingQuery query = new(state.Scope, listing.Definition.ProtocolVersion, null);
+        using SessionStreamSubscription subscription = AssertSuccess(streams.Subscribe(
+            request,
+            cursors.Encode(query, changes.CurrentRevision, clock.UtcNow)));
+
+        for (int index = 0; index < 1000; index++)
+        {
+            listing = state.Store.UpdateListing(Update(
+                listing,
+                displayName: $"Host {index}",
+                currentPlayers: index % 8)).Value!;
+        }
+
+        List<SessionStreamEvent> emitted = [];
+        while (subscription.Revision < changes.CurrentRevision)
+        {
+            SessionStreamReadResult read = streams.Read(subscription);
+            Assert.False(read.RequiresReset);
+            emitted.AddRange(read.Events);
+        }
+
+        Assert.Equal(8, emitted.Count);
+        SessionStreamEvent final = emitted[^1];
+        Assert.Equal(SessionStreamEventKind.SessionUpsert, final.Kind);
+        Assert.Equal("Host 999", final.Session!.DisplayName);
+        Assert.Equal(7, final.Session.Capacity.CurrentPlayers);
+    }
+
+    [Fact]
+    public void SubscriberLimitIsEnforcedPerTenantAndReleasedOnDispose()
+    {
+        ManualRendezvousClock clock = new();
+        SessionChangeJournal changes = new(new SessionChangeJournalOptions
+        {
+            MaximumSubscribers = 2,
+            MaximumSubscribersPerTenant = 1,
+        });
+        using SessionStreamCursorCodec cursors = new();
+        SessionStreamService streams = new(changes, cursors, clock);
+        TenantScope firstScope = new(new("first-game"), new("production"));
+        TenantScope secondScope = new(new("second-game"), new("production"));
+        BrowseSessionsRequest firstRequest = Request(firstScope);
+        BrowseSessionsRequest secondRequest = Request(secondScope);
+        string firstCursor = cursors.Encode(
+            new VisibleListingQuery(firstScope, 7, null),
+            changes.CurrentRevision,
+            clock.UtcNow);
+        string secondCursor = cursors.Encode(
+            new VisibleListingQuery(secondScope, 7, null),
+            changes.CurrentRevision,
+            clock.UtcNow);
+
+        SessionStreamSubscription first = AssertSuccess(streams.Subscribe(firstRequest, firstCursor));
+        Assert.Equal(
+            RendezvousErrorCode.CapacityExceeded,
+            streams.Subscribe(firstRequest, firstCursor).Error);
+        using SessionStreamSubscription second = AssertSuccess(
+            streams.Subscribe(secondRequest, secondCursor));
+        first.Dispose();
+        using SessionStreamSubscription replacement = AssertSuccess(
+            streams.Subscribe(firstRequest, firstCursor));
+    }
+
+    private static BrowseSessionsRequest Request(TenantScope scope) => new()
+    {
+        GameId = scope.GameId,
+        EnvironmentId = scope.EnvironmentId,
+        ProtocolVersion = 7,
+    };
+
+    private static UpdateListingCommand Update(
+        StoredListing listing,
+        string displayName,
+        int currentPlayers,
+        RegionId? regionId = null,
+        uint? protocolVersion = null,
+        ListingVisibility? visibility = null) => new(
+        listing.Definition.ListingId,
+        listing.Definition.LeaseId,
+        listing.Definition.LeaseFingerprint,
+        listing.Definition.OwnerSubject,
+        listing.Definition.BuildVersion,
+        displayName,
+        currentPlayers,
+        listing.Definition.MaximumPlayers,
+        listing.Definition.Metadata,
+        listing.Definition.DedicatedFallback,
+        regionId,
+        protocolVersion,
+        visibility);
+
+    private static SessionStreamEventKind SingleKind(
+        SessionBrowserFixture fixture,
+        SessionStreamSubscription subscription) =>
+        Assert.Single(fixture.Streams.Read(subscription).Events).Kind;
+
+    private static T AssertSuccess<T>(BrowserServiceResult<T> result)
+    {
+        Assert.True(result.Succeeded, result.Error.ToString());
+        return Assert.IsType<T>(result.Value);
+    }
+}

tests/FinalFactory.Rendezvous.Tests/Client/RendezvousClientBehaviorTests.cs

@@ -167,6 +167,88 @@ public sealed class RendezvousClientBehaviorTests
         Assert.Contains("gameId=space-game", handler.RequestUris[0].Query, StringComparison.Ordinal);
     }
 
+    [Fact]
+    public async Task StreamRejectsMalformedAndOversizedEventEnvelopes()
+    {
+        string[] bodies =
+        [
+            "event: session_upsert\nid: valid-cursor\ndata: {}\n\n",
+            "data: " + new string('x', ContractLimits.SessionStreamEventMaxBytes + 1) + "\n\n",
+        ];
+        foreach (string body in bodies)
+        {
+            StringContent content = new(body, Encoding.UTF8, "text/event-stream");
+            ScriptedHandler handler = new(Response(HttpStatusCode.OK, content));
+            using HttpClient httpClient = new(handler)
+            {
+                BaseAddress = new("http://rendezvous.test/"),
+            };
+            RendezvousSessionBrowserClient browser = new(httpClient);
+            await using IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> events = browser
+                .StreamAsync(new BrowseSessionsRequest
+                {
+                    GameId = new("space-game"),
+                    EnvironmentId = new("production"),
+                    ProtocolVersion = 7,
+                }, "valid-stream-cursor")
+                .GetAsyncEnumerator();
+
+            Assert.True(await events.MoveNextAsync());
+            Assert.False(events.Current.IsSuccess);
+            Assert.Equal(RendezvousErrorCode.InternalError, events.Current.Error);
+        }
+    }
+
+    [Fact]
+    public async Task StreamRequiresANonEmptySnapshotCursor()
+    {
+        using HttpClient httpClient = new(new ScriptedHandler())
+        {
+            BaseAddress = new("http://rendezvous.test/"),
+        };
+        RendezvousSessionBrowserClient browser = new(httpClient);
+        await Assert.ThrowsAsync<ArgumentException>(async () =>
+        {
+            await foreach (RendezvousClientResult<SessionStreamEvent> _ in browser.StreamAsync(
+                               new BrowseSessionsRequest
+                               {
+                                   GameId = new("space-game"),
+                                   EnvironmentId = new("production"),
+                                   ProtocolVersion = 7,
+                               },
+                               string.Empty))
+            {
+            }
+        });
+    }
+
+    [Fact]
+    public async Task StreamOpeningIsBoundedByTheConfiguredRequestTimeout()
+    {
+        using HttpClient httpClient = new(new SilentHandler())
+        {
+            BaseAddress = new("http://rendezvous.test/"),
+        };
+        RendezvousSessionBrowserClient browser = new(
+            httpClient,
+            new RendezvousClientOptions
+            {
+                MaximumSafeRetries = 0,
+                RequestTimeout = TimeSpan.FromMilliseconds(20),
+            });
+        await using IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> events = browser
+            .StreamAsync(new BrowseSessionsRequest
+            {
+                GameId = new("space-game"),
+                EnvironmentId = new("production"),
+                ProtocolVersion = 7,
+            }, "valid-stream-cursor")
+            .GetAsyncEnumerator();
+
+        Assert.True(await events.MoveNextAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(2)));
+        Assert.Equal(RendezvousErrorCode.ServiceUnavailable, events.Current.Error);
+    }
+
     [Fact]
     public async Task LeaseMaintainerReportsLeaseLoss()
     {

tests/FinalFactory.Rendezvous.Tests/Client/RendezvousClientIntegrationTests.cs

@@ -142,6 +142,77 @@ public sealed class RendezvousClientIntegrationTests
         }
     }
 
+    [Fact]
+    public async Task BrowserStreamResetsInvalidCursorReplaysReconnectAndReleasesConnections()
+    {
+        await using ClientTestHost host = await ClientTestHost.StartAsync();
+        RendezvousPublisherClient publisher = new(host.HttpClient);
+        RendezvousSessionBrowserClient browser = new(host.HttpClient);
+        PublishedSession session = AssertSuccess(await publisher.RegisterAsync(
+            CreateRegistration(200),
+            host.PublisherCredential));
+        BindPresence(host, session, 41_200);
+        BrowseSessionsRequest request = BrowseRequest();
+        BrowseSessionsResponse snapshot = AssertSuccess(await browser.BrowseAsync(request));
+        Assert.False(string.IsNullOrWhiteSpace(snapshot.StreamCursor));
+
+        using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(10));
+        await using (IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> invalid = browser
+                         .StreamAsync(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
+            .StreamAsync(request, snapshot.StreamCursor, timeout.Token)
+            .GetAsyncEnumerator(timeout.Token);
+        Task<bool> upsertPending = events.MoveNextAsync().AsTask();
+        Assert.True((await publisher.UpdateAsync(
+            session,
+            new UpdateSessionRequest
+            {
+                BuildVersion = "2.0.0",
+                DisplayName = "Live update",
+                Capacity = new() { CurrentPlayers = 3, MaximumPlayers = 8 },
+                Metadata = new() { ["mode"] = "online-coop" },
+            },
+            host.PublisherCredential,
+            timeout.Token)).IsSuccess);
+        Assert.True(await upsertPending);
+        SessionStreamEvent upsert = AssertSuccess(events.Current);
+        Assert.Equal(SessionStreamEventKind.SessionUpsert, upsert.Kind);
+        Assert.Equal("Live update", upsert.Session!.DisplayName);
+
+        await using (IAsyncEnumerator<RendezvousClientResult<SessionStreamEvent>> replay = browser
+                         .StreamAsync(request, snapshot.StreamCursor, timeout.Token)
+                         .GetAsyncEnumerator(timeout.Token))
+        {
+            Assert.True(await replay.MoveNextAsync());
+            SessionStreamEvent replayed = AssertSuccess(replay.Current);
+            Assert.Equal(SessionStreamEventKind.SessionUpsert, replayed.Kind);
+            Assert.Equal(upsert.Cursor, replayed.Cursor);
+            Assert.Equal("Live update", replayed.Session!.DisplayName);
+        }
+
+        Task<bool> removePending = events.MoveNextAsync().AsTask();
+        Assert.True((await publisher.DeregisterAsync(
+            session,
+            host.PublisherCredential,
+            timeout.Token)).IsSuccess);
+        Assert.True(await removePending);
+        SessionStreamEvent remove = AssertSuccess(events.Current);
+        Assert.Equal(SessionStreamEventKind.SessionRemove, remove.Kind);
+        Assert.Equal(session.ListingId, remove.ListingId);
+    }
+
+    private static string CorruptCursor(string cursor)
+    {
+        char replacement = cursor[^1] == 'a' ? 'b' : 'a';
+        return cursor[..^1] + replacement;
+    }
+
     private static T AssertSuccess<T>(RendezvousClientResult<T> result)
     {
         Assert.True(result.IsSuccess, result.Message);
@@ -210,7 +281,8 @@ public sealed class RendezvousClientIntegrationTests
         {
             ManualRendezvousClock clock = new(ProvisioningTestData.Now);
             EphemeralStoreOptions stateOptions = new();
-            InMemoryEphemeralRendezvousStore store = new(stateOptions, clock, clock);
+            SessionChangeJournal changes = new(new SessionChangeJournalOptions());
+            InMemoryEphemeralRendezvousStore store = new(stateOptions, clock, clock, changes);
             EphemeralCapabilityIssuer capabilities = new();
             ProvisioningRuntime provisioning = ProvisioningRuntime.Create(
                 ProvisioningTestData.CreateOptions(),
@@ -239,7 +311,10 @@ public sealed class RendezvousClientIntegrationTests
             builder.Services.AddSingleton(SessionLeaseTiming.From(stateOptions));
             builder.Services.AddSingleton<SessionLeaseService>();
             builder.Services.AddSingleton<SessionBrowserCursorCodec>();
+            builder.Services.AddSingleton<SessionStreamCursorCodec>();
+            builder.Services.AddSingleton(changes);
             builder.Services.AddSingleton<SessionBrowserService>();
+            builder.Services.AddSingleton<SessionStreamService>();
 
             WebApplication app = builder.Build();
             app.UseExceptionHandler();

tests/FinalFactory.Rendezvous.Tests/Contracts/OpenApiCompatibilityTests.cs

@@ -17,6 +17,7 @@ public sealed class OpenApiCompatibilityTests
         "/v1/operator/principals/revoke",
         "/v1/operator/status",
         "/v1/sessions",
+        "/v1/sessions/stream",
         "/v1/sessions/{listingId}",
         "/v1/sessions/{listingId}/join-attempts",
         "/v1/sessions/{listingId}/renew",
@@ -68,6 +69,25 @@ public sealed class OpenApiCompatibilityTests
         Assert.DoesNotContain(listingProperties, static property =>
             property.Contains("token", StringComparison.OrdinalIgnoreCase)
             || property.Contains("playerId", StringComparison.OrdinalIgnoreCase));
+        JsonElement streamProperties = schemas.GetProperty("SessionStreamEvent")
+            .GetProperty("properties");
+        Assert.True(streamProperties.TryGetProperty("contractVersion", out _));
+        Assert.True(streamProperties.TryGetProperty("kind", out _));
+        Assert.True(streamProperties.TryGetProperty("cursor", out _));
+        Assert.True(streamProperties.TryGetProperty("session", out _));
+        Assert.True(streamProperties.TryGetProperty("listingId", out _));
+        Assert.DoesNotContain(streamProperties.EnumerateObject(), static property =>
+            property.Name.Contains("token", StringComparison.OrdinalIgnoreCase)
+            || property.Name.Contains("capability", StringComparison.OrdinalIgnoreCase)
+            || property.Name.Contains("ticket", StringComparison.OrdinalIgnoreCase)
+            || property.Name.Contains("endpoint", StringComparison.OrdinalIgnoreCase));
+        Assert.True(root.GetProperty("paths")
+            .GetProperty("/v1/sessions/stream")
+            .GetProperty("get")
+            .GetProperty("responses")
+            .GetProperty("200")
+            .GetProperty("content")
+            .TryGetProperty("text/event-stream", out _));
         JsonElement dedicatedFallback = schemas.GetProperty("SessionListing")
             .GetProperty("properties")
             .GetProperty("dedicatedFallback");
@@ -205,7 +225,7 @@ public sealed class OpenApiCompatibilityTests
             }
         }
 
-        Assert.Equal(17, overloadContracts);
+        Assert.Equal(18, overloadContracts);
         (string Path, string Method)[] bodyOperations =
         [
             ("/v1/sessions", "post"),

tests/FinalFactory.Rendezvous.Tests/JoinAttempts/JoinAttemptHttpEndpointTests.cs

@@ -288,7 +288,8 @@ public sealed class JoinAttemptHttpEndpointTests
         {
             ManualRendezvousClock clock = new(ProvisioningTestData.Now);
             EphemeralStoreOptions stateOptions = new();
-            InMemoryEphemeralRendezvousStore store = new(stateOptions, clock, clock);
+            SessionChangeJournal changes = new(new SessionChangeJournalOptions());
+            InMemoryEphemeralRendezvousStore store = new(stateOptions, clock, clock, changes);
             EphemeralCapabilityIssuer capabilities = new();
             ProvisioningRuntime provisioning = ProvisioningRuntime.Create(
                 ProvisioningTestData.CreateOptions(),
@@ -318,7 +319,10 @@ public sealed class JoinAttemptHttpEndpointTests
             builder.Services.AddSingleton(SessionLeaseTiming.From(stateOptions));
             builder.Services.AddSingleton<SessionLeaseService>();
             builder.Services.AddSingleton<SessionBrowserCursorCodec>();
+            builder.Services.AddSingleton<SessionStreamCursorCodec>();
+            builder.Services.AddSingleton(changes);
             builder.Services.AddSingleton<SessionBrowserService>();
+            builder.Services.AddSingleton<SessionStreamService>();
             builder.Services.AddSingleton<JoinAttemptCursorCodec>();
             builder.Services.AddSingleton<JoinAttemptService>();
             ConnectionOutcomeMetrics outcomeMetrics = new();

tests/FinalFactory.Rendezvous.Tests/Sessions/SessionHttpEndpointTests.cs

@@ -28,7 +28,8 @@ public sealed class SessionHttpEndpointTests
     {
         ManualRendezvousClock clock = new(ProvisioningTestData.Now);
         EphemeralStoreOptions stateOptions = new();
-        InMemoryEphemeralRendezvousStore store = new(stateOptions, clock, clock);
+        SessionChangeJournal changes = new(new SessionChangeJournalOptions());
+        InMemoryEphemeralRendezvousStore store = new(stateOptions, clock, clock, changes);
         EphemeralCapabilityIssuer capabilities = new();
         ProvisioningRuntime provisioning = ProvisioningRuntime.Create(
             ProvisioningTestData.CreateOptions(),
@@ -57,7 +58,10 @@ public sealed class SessionHttpEndpointTests
         builder.Services.AddSingleton(SessionLeaseTiming.From(stateOptions));
         builder.Services.AddSingleton<SessionLeaseService>();
         builder.Services.AddSingleton<SessionBrowserCursorCodec>();
+        builder.Services.AddSingleton<SessionStreamCursorCodec>();
+        builder.Services.AddSingleton(changes);
         builder.Services.AddSingleton<SessionBrowserService>();
+        builder.Services.AddSingleton<SessionStreamService>();
         await using WebApplication app = builder.Build();
         app.UseExceptionHandler();
         app.UseMiddleware<HttpAbuseProtectionMiddleware>();

tests/FinalFactory.Rendezvous.Tests/State/EphemeralStateTestData.cs

@@ -1,4 +1,5 @@
 using FinalFactory.Rendezvous.Contracts;
+using FinalFactory.Rendezvous.Server.Browser;
 using FinalFactory.Rendezvous.Server.State;
 
 namespace FinalFactory.Rendezvous.Tests.State;
@@ -24,10 +25,12 @@ internal sealed class EphemeralStateFixture
 {
     private int _sequence;
 
-    public EphemeralStateFixture(EphemeralStoreOptions? options = null)
+    public EphemeralStateFixture(
+        EphemeralStoreOptions? options = null,
+        SessionChangeJournal? changes = null)
     {
         Clock = new();
-        Store = new(options ?? new EphemeralStoreOptions(), Clock, Clock);
+        Store = new(options ?? new EphemeralStoreOptions(), Clock, Clock, changes);
     }
 
     public ManualRendezvousClock Clock { get; }

tests/FinalFactory.Rendezvous.Tests/TestClient/TestClientCommandTests.cs

@@ -59,6 +59,29 @@ public sealed class TestClientCommandTests
         Assert.Equal(130, (int)TestClientExitCode.Cancelled);
     }
 
+    [Fact]
+    public void WatchModeSupportsBoundedRuntimeAndDeliberateRecoveryExercisesOnly()
+    {
+        TestClientParseResult reset = TestClientOptionParser.Parse(
+            ["watch", "--run-seconds", "30", "--exercise-reset", "--script", "--json"]);
+        Assert.True(reset.Succeeded, reset.Error);
+        TestClientOptions resetOptions = Assert.IsType<TestClientOptions>(reset.Options);
+        Assert.Equal(TestClientMode.Watch, resetOptions.Mode);
+        Assert.Equal(TimeSpan.FromSeconds(30), resetOptions.RunDuration);
+        Assert.True(resetOptions.ExerciseReset);
+
+        TestClientParseResult reconnect = TestClientOptionParser.Parse(
+            ["watch", "--exercise-reconnect", "--script"]);
+        Assert.True(reconnect.Succeeded, reconnect.Error);
+        Assert.True(Assert.IsType<TestClientOptions>(reconnect.Options).ExerciseReconnect);
+
+        Assert.False(TestClientOptionParser.Parse(["browse", "--exercise-reconnect"]).Succeeded);
+        Assert.False(TestClientOptionParser.Parse(["browse", "--exercise-reset"]).Succeeded);
+        Assert.False(TestClientOptionParser.Parse(
+            ["watch", "--exercise-reconnect", "--exercise-reset"]).Succeeded);
+        Assert.False(TestClientOptionParser.Parse(["join", "--run-seconds", "30"]).Succeeded);
+    }
+
     [Fact]
     public void HostFailureBudgetStopsAuthorityLossAndBoundsTransientRetries()
     {

tests/FinalFactory.Rendezvous.Tests/TestData/Contracts/v1/browse-sessions.json

@@ -1 +1 @@
-{"contractVersion":1,"items":[{"contractVersion":1,"listingId":"00112233-4455-6677-8899-aabbccddeeff","gameId":"space-game","environmentId":"production","regionId":"eu-central","protocolVersion":7,"buildVersion":"1.4.2","displayName":"Europa Relay","visibility":"public","publisherTrustMode":"managedDedicated","capacity":{"currentPlayers":2,"maximumPlayers":8},"metadata":{"mode":"co-op","map":"europa"}}],"nextCursor":"cursor-002"}
+{"contractVersion":1,"items":[{"contractVersion":1,"listingId":"00112233-4455-6677-8899-aabbccddeeff","gameId":"space-game","environmentId":"production","regionId":"eu-central","protocolVersion":7,"buildVersion":"1.4.2","displayName":"Europa Relay","visibility":"public","publisherTrustMode":"managedDedicated","capacity":{"currentPlayers":2,"maximumPlayers":8},"metadata":{"mode":"co-op","map":"europa"}}],"nextCursor":"cursor-002","streamCursor":"stream-cursor-002"}

tests/FinalFactory.Rendezvous.Tests/TestData/Contracts/v1/client-public-api.txt

@@ -40,6 +40,7 @@ TYPE FinalFactory.Rendezvous.Client.IRendezvousSessionBrowserClient
   METHOD System.Threading.Tasks.Task<FinalFactory.Rendezvous.Client.RendezvousClientResult<System.Collections.Generic.IReadOnlyList<FinalFactory.Rendezvous.Contracts.SessionListing>>> BrowseAllAsync(FinalFactory.Rendezvous.Contracts.BrowseSessionsRequest request, System.Int32 maximumPages, System.Threading.CancellationToken cancellationToken)
   METHOD System.Threading.Tasks.Task<FinalFactory.Rendezvous.Client.RendezvousClientResult<FinalFactory.Rendezvous.Contracts.BrowseSessionsResponse>> BrowseAsync(FinalFactory.Rendezvous.Contracts.BrowseSessionsRequest request, System.Threading.CancellationToken cancellationToken)
   METHOD System.Threading.Tasks.Task<FinalFactory.Rendezvous.Client.RendezvousClientResult<FinalFactory.Rendezvous.Contracts.GetSessionResponse>> GetAsync(FinalFactory.Rendezvous.Contracts.SessionListingId listingId, FinalFactory.Rendezvous.Contracts.GameId gameId, FinalFactory.Rendezvous.Contracts.EnvironmentId environmentId, System.UInt32 protocolVersion, System.Threading.CancellationToken cancellationToken)
+  METHOD System.Collections.Generic.IAsyncEnumerable<FinalFactory.Rendezvous.Client.RendezvousClientResult<FinalFactory.Rendezvous.Contracts.SessionStreamEvent>> StreamAsync(FinalFactory.Rendezvous.Contracts.BrowseSessionsRequest request, System.String streamCursor, System.Threading.CancellationToken cancellationToken)
 TYPE FinalFactory.Rendezvous.Client.LeaseMaintenanceResult
   PROP FinalFactory.Rendezvous.Contracts.RendezvousErrorCode Error {get;}
   PROP FinalFactory.Rendezvous.Client.LeaseMaintenanceStopReason Reason {get;}
@@ -212,6 +213,7 @@ TYPE FinalFactory.Rendezvous.Client.RendezvousSessionBrowserClient
   METHOD System.Threading.Tasks.Task<FinalFactory.Rendezvous.Client.RendezvousClientResult<System.Collections.Generic.IReadOnlyList<FinalFactory.Rendezvous.Contracts.SessionListing>>> BrowseAllAsync(FinalFactory.Rendezvous.Contracts.BrowseSessionsRequest request, System.Int32 maximumPages, System.Threading.CancellationToken cancellationToken)
   METHOD System.Threading.Tasks.Task<FinalFactory.Rendezvous.Client.RendezvousClientResult<FinalFactory.Rendezvous.Contracts.BrowseSessionsResponse>> BrowseAsync(FinalFactory.Rendezvous.Contracts.BrowseSessionsRequest request, System.Threading.CancellationToken cancellationToken)
   METHOD System.Threading.Tasks.Task<FinalFactory.Rendezvous.Client.RendezvousClientResult<FinalFactory.Rendezvous.Contracts.GetSessionResponse>> GetAsync(FinalFactory.Rendezvous.Contracts.SessionListingId listingId, FinalFactory.Rendezvous.Contracts.GameId gameId, FinalFactory.Rendezvous.Contracts.EnvironmentId environmentId, System.UInt32 protocolVersion, System.Threading.CancellationToken cancellationToken)
+  METHOD System.Collections.Generic.IAsyncEnumerable<FinalFactory.Rendezvous.Client.RendezvousClientResult<FinalFactory.Rendezvous.Contracts.SessionStreamEvent>> StreamAsync(FinalFactory.Rendezvous.Contracts.BrowseSessionsRequest request, System.String streamCursor, System.Threading.CancellationToken cancellationToken)
 TYPE FinalFactory.Rendezvous.Client.SessionLeaseMaintainer
   EVENT System.EventHandler LeaseLost
   METHOD System.Threading.Tasks.ValueTask DisposeAsync()

tests/FinalFactory.Rendezvous.Tests/TestData/Contracts/v1/contracts-public-api.txt

@@ -28,6 +28,7 @@ TYPE FinalFactory.Rendezvous.Contracts.BrowseSessionsResponse
   PROP System.Int32 ContractVersion {get;set;}
   PROP System.Collections.Generic.List<FinalFactory.Rendezvous.Contracts.SessionListing> Items {get;set;}
   PROP System.String NextCursor {get;set;}
+  PROP System.String StreamCursor {get;set;}
 TYPE FinalFactory.Rendezvous.Contracts.ConnectionElapsedBucket
   ENUM UnderOneSecond=1
   ENUM OneToFiveSeconds=2
@@ -84,6 +85,7 @@ TYPE FinalFactory.Rendezvous.Contracts.ContractLimits
   FIELD System.Int32 OpaqueHttpCredentialMaxCharacters=1024
   FIELD System.Int32 RegionIdMaxCharacters=32
   FIELD System.Int32 SessionCapacityMaxPlayers=10000
+  FIELD System.Int32 SessionStreamEventMaxBytes=32768
   FIELD System.Int32 UdpCapabilityMaxCharacters=192
   FIELD System.Int32 UdpDatagramMaxBytes=1200
 TYPE FinalFactory.Rendezvous.Contracts.ContractValidation
@@ -345,6 +347,18 @@ TYPE FinalFactory.Rendezvous.Contracts.SessionListingId
   METHOD System.Boolean TryParse(System.String value, FinalFactory.Rendezvous.Contracts.SessionListingId& id)
   METHOD System.Boolean op_Equality(FinalFactory.Rendezvous.Contracts.SessionListingId left, FinalFactory.Rendezvous.Contracts.SessionListingId right)
   METHOD System.Boolean op_Inequality(FinalFactory.Rendezvous.Contracts.SessionListingId left, FinalFactory.Rendezvous.Contracts.SessionListingId right)
+TYPE FinalFactory.Rendezvous.Contracts.SessionStreamEvent
+  CTOR ()
+  PROP System.Int32 ContractVersion {get;set;}
+  PROP System.String Cursor {get;set;}
+  PROP FinalFactory.Rendezvous.Contracts.SessionStreamEventKind Kind {get;set;}
+  PROP System.Nullable<FinalFactory.Rendezvous.Contracts.SessionListingId> ListingId {get;set;}
+  PROP FinalFactory.Rendezvous.Contracts.SessionListing Session {get;set;}
+TYPE FinalFactory.Rendezvous.Contracts.SessionStreamEventKind
+  ENUM SessionUpsert=1
+  ENUM SessionRemove=2
+  ENUM Reset=3
+  ENUM Keepalive=4
 TYPE FinalFactory.Rendezvous.Contracts.UdpDecodeError
   ENUM None=0
   ENUM DatagramTooLarge=1
@@ -371,3 +385,6 @@ TYPE FinalFactory.Rendezvous.Contracts.UpdateSessionRequest
   PROP System.String DisplayName {get;set;}
   PROP System.String LeaseToken {get;set;}
   PROP System.Collections.Generic.Dictionary<System.String,System.String> Metadata {get;set;}
+  PROP System.Nullable<System.UInt32> ProtocolVersion {get;set;}
+  PROP System.Nullable<FinalFactory.Rendezvous.Contracts.RegionId> RegionId {get;set;}
+  PROP System.Nullable<FinalFactory.Rendezvous.Contracts.ListingVisibility> Visibility {get;set;}