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 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 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 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(BrowserServiceResult result) { Assert.True(result.Succeeded, result.Error.ToString()); return Assert.IsType(result.Value); } }