diff --git a/src/GameCult.Networking/Server.cs b/src/GameCult.Networking/Server.cs index f158b1d2..d9b763b9 100644 --- a/src/GameCult.Networking/Server.cs +++ b/src/GameCult.Networking/Server.cs @@ -23,7 +23,7 @@ namespace GameCult.Networking private const string EmailPattern = @"^([0-9a-zA-Z]([\+\-_\.][0-9a-zA-Z]+)*)+@(([0-9a-zA-Z][-\w]*[0-9a-zA-Z]*\.)+[a-zA-Z0-9]{2,17})$"; private const string UsernamePattern = @"^[A-Za-z0-9]+(?:[ _-][A-Za-z0-9]+)*$"; - private const int ServerPort = 3075; + private const int DefaultPort = 3075; private const float SessionTimeoutSeconds = 1800; // 30 minutes private const float SessionRefreshThresholdSeconds = 300; // 5 minutes private const int MaxConnectionAttemptsPerMinute = 30; @@ -63,6 +63,12 @@ namespace GameCult.Networking /// public bool LogSensitivePayloads { get; set; } + /// + /// Gets or sets the UDP port binds (default 3075; 0 asks the OS for a free one). After + /// it holds the port actually bound. + /// + public int Port { get; set; } = DefaultPort; + private float Time => (float)(_timer?.Elapsed.TotalSeconds ?? 0d); /// @@ -74,7 +80,7 @@ namespace GameCult.Networking { TransportId = "litenetlib-server", Host = "0.0.0.0", - Port = ServerPort + Port = Port }); /// @@ -357,7 +363,8 @@ namespace GameCult.Networking UnsyncedEvents = true, NatPunchEnabled = true }; - _netManager.Start(ServerPort); + _netManager.Start(Port); + Port = _netManager.LocalPort; listener.NetworkErrorEvent += (point, code) => Logger.LogInfo($"{point.Address}: Error {code}"); @@ -406,25 +413,26 @@ namespace GameCult.Networking var frame = transport.Receive(bytes); if (string.Equals(frame.ChannelId, "schema", StringComparison.Ordinal)) { - ICultNetSchemaMessage cultNetMessage; - try - { - cultNetMessage = LiteNetLibTransportConnection.DecodeSchema(frame); - } - catch (Exception ex) when (transport.WriteReceipts != null) - { - // A receipt-capable peer is told which frame was unreadable (by its id, if the frame still holds one). - Logger.LogWarning($"CultNet schema frame unreadable: {ex.Message}"); - transport.AnswerUnreadableSchemaFrame(frame.Payload); - return; - } - + var cultNetMessage = LiteNetLibTransportConnection.DecodeSchema(frame); Logger.LogDebug($"Received CultNet schema message {cultNetMessage.SchemaVersion}"); await HandleCultNetSchemaMessageAsync(peer, user, cultNetMessage).ConfigureAwait(false); return; } - var message = LiteNetLibTransportConnection.DecodeLegacy(frame); + Message message; + try + { + message = LiteNetLibTransportConnection.DecodeLegacy(frame); + } + catch (Exception ex) when (transport.WriteReceipts != null) + { + // A frame that decodes on neither channel is classed legacy, so this is where an unreadable frame lands. + // A receipt-capable peer is told which frame was unreadable (by its id, if the frame still holds one). + Logger.LogWarning($"CultNet frame unreadable: {ex.Message}"); + transport.AnswerUnreadableSchemaFrame(frame.Payload); + return; + } + if (LogSensitivePayloads) { Logger.LogDebug($"Received message: {MessagePackSerializer.ConvertToJson(new ReadOnlyMemory(frame.Payload))}"); @@ -519,7 +527,7 @@ namespace GameCult.Networking } }); - Logger.LogInfo($"Server started on port {ServerPort}."); + Logger.LogInfo($"Server started on port {Port}."); } private async Task HandleCultNetSchemaMessageAsync(NetPeer peer, User user, ICultNetSchemaMessage message) diff --git a/tests/GameCult.Networking.Tests/CultNetWriteReceiptLoopbackTests.cs b/tests/GameCult.Networking.Tests/CultNetWriteReceiptLoopbackTests.cs new file mode 100644 index 00000000..8a1f0b46 --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetWriteReceiptLoopbackTests.cs @@ -0,0 +1,188 @@ +#nullable enable +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Caching; +using LiteNetLib; +using NUnit.Framework; +using static GameCult.Networking.Tests.NetworkingTests; + +namespace GameCult.Networking.Tests +{ + // The receipt server on a real socket: Server.Start on an ephemeral loopback port and real LiteNetLib clients. The + // handler-level tests (CultNetWriteReceiptServerTests) hand frames to the handlers; only a running server decides what + // a frame that does not decode is answered with. + public sealed class CultNetWriteReceiptLoopbackTests + { + private static readonly TimeSpan Cap = TimeSpan.FromSeconds(10); + private static readonly CultRecordKey KeyA = new("tests:loopback:a"); + private static readonly CultRecordKey KeyB = new("tests:loopback:b"); + private const string ShardId = "loopback-shard"; + + private sealed class Client : IDisposable + { + private readonly NetManager _manager; + private readonly NetPeer _peer; + private readonly BlockingCollection _received = new(); + public int Closed; + + public Client(int port, string key) + { + var listener = new EventBasedNetListener(); + var connected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var decoder = new LiteNetLibTransportConnection(); + listener.PeerConnectedEvent += _ => connected.TrySetResult(true); + listener.PeerDisconnectedEvent += (_, _) => Interlocked.Increment(ref Closed); + listener.NetworkReceiveEvent += (_, reader, _, _) => + _received.Add(LiteNetLibTransportConnection.DecodeSchema(decoder.Receive(reader.GetRemainingBytes()))); + _manager = new NetManager(listener) { UnsyncedEvents = true }; + _manager.Start(0); + _peer = _manager.Connect("127.0.0.1", port, key); + if (!connected.Task.Wait(Cap)) Assert.Fail("the client did not connect"); + } + + public void Send(T message) where T : ICultNetSchemaMessage => + _peer.Send(CultNetSchemaMessageSerialization.Serialize(message), DeliveryMethod.ReliableOrdered); + + public void SendRaw(byte[] bytes) => _peer.Send(bytes, DeliveryMethod.ReliableOrdered); + + public T Next() where T : ICultNetSchemaMessage + { + if (!_received.TryTake(out var message, Cap)) Assert.Fail($"no {typeof(T).Name} arrived within {Cap}"); + Assert.That(message, Is.InstanceOf()); + return (T)message!; + } + + public void ExpectNothingMore(string because) + { + // A reliable ordered channel: anything the server sent before the probe's answer has already arrived. + Assert.That(_received.Count, Is.Zero, because); + } + + public void Dispose() + { + _manager.Stop(); + _received.Dispose(); + } + } + + private sealed class Authority : ICultNetRemoteWriteAuthority + { + public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request) => + request.RecordKey.Equals(KeyB) ? CultNetRemoteWriteDecision.Refuse("denied") : CultNetRemoteWriteDecision.Grant(); + } + + private sealed class Running : IDisposable + { + public Server Server = null!; + public CultCache Cache = null!; + public CultNetDatabase Database = null!; + public string SchemaId = string.Empty; + private CultNetDatabaseServer _databaseServer = null!; + + public static Running Start() + { + var cache = new CultCache(); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + Shards = [new CultNetShardDescriptor(ShardId, "primary", epoch: 1, isPrimary: true, schemaIds: [schemaId])] + }); + var server = new Server(cache, ServerSecurityOptions.Development()) { Port = 0 }; + var databaseServer = new CultNetDatabaseServer(server, database, new CultNetDatabaseServerOptions + { + RemoteWriteAuthority = new Authority() + }); + server.Start(); + return new Running { Server = server, Cache = cache, Database = database, SchemaId = schemaId, _databaseServer = databaseServer }; + } + + public Client Connect() => new(Server.Port, ServerSecurityOptions.Development().ConnectionKey); + + public CultNetDocumentPutRawMessage PutOf(string id, CultRecordKey key, string text) + { + var message = Database.Documents.CreateRawDocumentPutMessage(id, new CultRecordHandle(key), + new NetworkSchemaNote { Schema = "tests.networking_note.v1", Text = text }); + message.ShardId = ShardId; + message.ShardEpoch = 1; + return message; + } + + public void Dispose() + { + _databaseServer.Dispose(); + Server.Dispose(); + } + } + + private static CultNetHelloMessage HelloOf(params string[] versions) => + new() { RuntimeId = "loopback-client", RuntimeKind = "test", SupportedMessageVersions = versions }; + + private static readonly string[] BothFrames = [CultNetSchemaVersions.WriteAdmitted, CultNetSchemaVersions.ErrorV1]; + + [Test] + public void StartBindsTheRequestedPortAndReportsTheOneBound() + { + using var running = Running.Start(); + Assert.That(running.Server.Port, Is.Not.Zero.And.Not.EqualTo(3075)); + } + + // contracts/cultnet/interop/write-receipt-vectors.json, error_v1_unreadable_frame_has_null_message_id: code + // message_invalid, details.field "message", messageId null. + [TestCase(new byte[] { 0xC1 }, TestName = "AnUnreadableFrameFromAReceiptPeerIsAnsweredWithMessageInvalidAndNoId_NeverUsedByte")] + [TestCase(new byte[] { 0x92, 0x01, 0x02 }, TestName = "AnUnreadableFrameFromAReceiptPeerIsAnsweredWithMessageInvalidAndNoId_NotAMap")] + public void AnUnreadableFrameFromAReceiptPeerIsAnsweredWithMessageInvalidAndNoId(byte[] garbage) + { + using var running = Running.Start(); + using var client = running.Connect(); + client.Send(HelloOf(BothFrames)); + client.Next(); + + client.SendRaw(garbage); + + var error = client.Next(); + Assert.That(error.Code, Is.EqualTo("message_invalid")); + Assert.That(error.Details!.Field, Is.EqualTo("message")); + Assert.That(error.MessageId, Is.Null); + } + + [Test] + public async Task AReceiptPeerIsStillServedAfterAnUnreadableFrameAndALegacyPeerIsAnsweredAsBefore() + { + using var running = Running.Start(); + using var receipt = running.Connect(); + receipt.Send(HelloOf(BothFrames)); + receipt.Next(); + receipt.SendRaw(new byte[] { 0xC1 }); + receipt.Next(); + + receipt.Send(running.PutOf("r1", KeyA, "one")); + Assert.That(receipt.Next().MessageId, Is.EqualTo("r1")); + + using var legacy = running.Connect(); + legacy.Send(HelloOf()); // a hello that lists neither frame is answered with nothing + legacy.SendRaw(new byte[] { 0xC1 }); // unanswered: were it answered, error.v1 would arrive ahead of the refusal + legacy.Send(running.PutOf("l1", KeyB, "two")); + var refusal = legacy.Next(); + Assert.That(refusal, Is.TypeOf()); + Assert.That(refusal.Code, Is.EqualTo("write_not_authorized")); + legacy.Send(running.PutOf("l2", KeyA, "three")); + await WaitFor(() => running.Cache.Get(KeyA)?.Text == "three"); + legacy.ExpectNothingMore("a legacy peer is never sent a receipt or error.v1"); + Assert.That(running.Cache.Get(KeyB), Is.Null); + } + + private static async Task WaitFor(Func condition) + { + var deadline = DateTime.UtcNow + Cap; + while (!condition()) + { + if (DateTime.UtcNow > deadline) Assert.Fail("the condition did not hold within the cap"); + await Task.Delay(5); + } + } + } +}