From 6beca4a80abf0a1a00e18466975340fd4c8cdcc2 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 19 Jul 2026 18:50:22 +0100 Subject: [PATCH] Preserve peer locality across schema transports --- .../CultNetDatabaseSubscriptionServer.cs | 4 +-- .../CultNetRudpSchemaServer.cs | 2 +- .../CultNetSchemaServerPort.cs | 8 +++++ .../CultNetTcpSchemaTransport.cs | 5 +++- src/GameCult.Networking/Server.cs | 6 +++- .../NetworkingTests.cs | 30 +++++++++++++++++++ 6 files changed, 50 insertions(+), 5 deletions(-) diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index e55895e9..c5fe47bd 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -171,9 +171,9 @@ namespace GameCult.Networking private static bool IsSameMachine(ICultNetSchemaServerPeer peer) { - if (peer is RudpCultNetSchemaServerPeer rudp && rudp.RemoteEndPoint is IPEndPoint ip) + if (peer is ICultNetSchemaServerPeerLocation located && located.RemoteEndPoint is IPEndPoint ip) return IPAddress.IsLoopback(ip.Address); - return peer is CultNetServerPeer liteNet && IPAddress.IsLoopback(liteNet.Peer.Address); + return false; } private IDisposable Watch( diff --git a/src/GameCult.Networking/CultNetRudpSchemaServer.cs b/src/GameCult.Networking/CultNetRudpSchemaServer.cs index bd8d96cd..1abce308 100644 --- a/src/GameCult.Networking/CultNetRudpSchemaServer.cs +++ b/src/GameCult.Networking/CultNetRudpSchemaServer.cs @@ -70,7 +70,7 @@ namespace GameCult.Networking /// /// Peer context for a schema-v0 RUDP server. /// - public sealed class RudpCultNetSchemaServerPeer : ICultNetSchemaServerPeer + public sealed class RudpCultNetSchemaServerPeer : ICultNetSchemaServerPeer, ICultNetSchemaServerPeerLocation { private readonly RudpCultNetSchemaServer _server; diff --git a/src/GameCult.Networking/CultNetSchemaServerPort.cs b/src/GameCult.Networking/CultNetSchemaServerPort.cs index 4d660071..e3670ba1 100644 --- a/src/GameCult.Networking/CultNetSchemaServerPort.cs +++ b/src/GameCult.Networking/CultNetSchemaServerPort.cs @@ -1,4 +1,5 @@ using System; +using System.Net; using System.Threading.Tasks; namespace GameCult.Networking @@ -33,6 +34,13 @@ namespace GameCult.Networking where TMessage : ICultNetSchemaMessage; } + /// Optional physical peer location used for locality-sensitive transport negotiation. + public interface ICultNetSchemaServerPeerLocation + { + /// Gets the remote physical endpoint when the transport exposes one. + EndPoint? RemoteEndPoint { get; } + } + /// /// Optional transport-neutral peer lifetime surface for stateful service bridges. /// diff --git a/src/GameCult.Networking/CultNetTcpSchemaTransport.cs b/src/GameCult.Networking/CultNetTcpSchemaTransport.cs index 4cdfe06c..168ce924 100644 --- a/src/GameCult.Networking/CultNetTcpSchemaTransport.cs +++ b/src/GameCult.Networking/CultNetTcpSchemaTransport.cs @@ -189,20 +189,23 @@ namespace GameCult.Networking } } - private sealed class TcpPeer : ICultNetSchemaServerPeer, IDisposable + private sealed class TcpPeer : ICultNetSchemaServerPeer, ICultNetSchemaServerPeerLocation, IDisposable { private readonly TcpClient _client; private readonly object _sendGate = new(); + private readonly EndPoint? _remoteEndPoint; public TcpPeer(TcpClient client) { _client = client; + _remoteEndPoint = client.Client.RemoteEndPoint; Transport = new TcpFramedTransportConnection( client.GetStream(), CultNetTransportProfiles.CreateTcpFramed("csharp-tcp-schema-server")); } public bool Connected => _client.Connected; + public EndPoint? RemoteEndPoint => _remoteEndPoint; public TcpFramedTransportConnection Transport { get; } public void SendCultNet(TMessage message) where TMessage : ICultNetSchemaMessage diff --git a/src/GameCult.Networking/Server.cs b/src/GameCult.Networking/Server.cs index 4d468f7e..616f9ccd 100644 --- a/src/GameCult.Networking/Server.cs +++ b/src/GameCult.Networking/Server.cs @@ -3,6 +3,7 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Linq; +using System.Net; using System.Text.RegularExpressions; using System.Threading.Tasks; using GameCult.Caching; @@ -820,7 +821,7 @@ namespace GameCult.Networking /// /// Transport-aware server peer context for built-in CultNet service bodies. /// - public sealed class CultNetServerPeer : ICultNetSchemaServerPeer + public sealed class CultNetServerPeer : ICultNetSchemaServerPeer, ICultNetSchemaServerPeerLocation { internal CultNetServerPeer(NetPeer peer, LiteNetLibTransportConnection transport) { @@ -833,6 +834,9 @@ namespace GameCult.Networking /// public NetPeer Peer { get; } + /// + public EndPoint RemoteEndPoint => new IPEndPoint(Peer.Address, Peer.Port); + /// /// Gets the channel-aware LiteNetLib transport adapter for this peer. /// diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index e7f4a8b4..9b1af662 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1526,6 +1526,36 @@ namespace GameCult.Networking.Tests Assert.That(removed.Active, Is.False); } + [Test] + public async Task DatabaseSubscriptionServer_ClassifiesLoopbackTcpDemandAsSameMachine() + { + var database = new CultNetDatabase(new CultCache()); + using var server = new TcpFramedCultNetSchemaServer(new TcpListener(IPAddress.Loopback, 0)); + using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database); + var active = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + subscriptions.DemandChanged += demand => + { + if (demand.Active) active.TrySetResult(demand); + }; + using var client = new TcpFramedCultNetSchemaClient(); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + client.SendCultNet(new CultNetDatabaseSubscribeMessage + { + MessageId = "subscribe-tcp-body", + SubscriptionId = "world-body", + ConsumerRuntimeId = "eve-unity", + BodyIds = ["world"], + SupportedBodyTransports = ["SharedMemory", "Network"], + IncludeSnapshot = false + }); + + var observed = await AwaitWithTimeout(active.Task, TimeSpan.FromSeconds(2)); + + Assert.That(observed.SameMachine, Is.True); + Assert.That(observed.BodyIds, Is.EqualTo(new[] { "world" })); + } + [Test] public async Task DatabaseSubscriptionServer_ProjectsExactReactiveStateDemandWithoutBodyDemand() {