diff --git a/src/GameCult.Mesh/CultMeshReliabilityDiagnostics.cs b/src/GameCult.Mesh/CultMeshReliabilityDiagnostics.cs new file mode 100644 index 00000000..5391c77e --- /dev/null +++ b/src/GameCult.Mesh/CultMeshReliabilityDiagnostics.cs @@ -0,0 +1,183 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; + +namespace GameCult.Mesh +{ + /// + /// Clock used by CultMesh reliability owners. + /// + public interface ICultMeshClock + { + DateTimeOffset UtcNow { get; } + + Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken = default); + } + + /// + /// Wall clock used outside deterministic tests. + /// + public sealed class CultMeshSystemClock : ICultMeshClock + { + public static CultMeshSystemClock Instance { get; } = new CultMeshSystemClock(); + + private CultMeshSystemClock() + { + } + + public DateTimeOffset UtcNow => DateTimeOffset.UtcNow; + + public Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken = default) + { + return Task.Delay(delay, cancellationToken); + } + } + + public enum CultMeshReliabilityOrgan + { + Discovery, + Session, + Authority, + ContentTransfer, + Stream + } + + public enum CultMeshDiagnosticKind + { + DiscoveryObservation, + CandidateRejected, + ConnectionAttempt, + PathChanged, + RetryDecision, + AuthorityDecision, + VerifiedRange, + StreamGap + } + + /// + /// One bounded operational observation emitted by a CultMesh reliability owner. + /// + public sealed class CultMeshDiagnosticEvent + { + public CultMeshDiagnosticEvent( + long sequence, + DateTimeOffset observedAtUtc, + CultMeshReliabilityOrgan organ, + CultMeshDiagnosticKind kind, + string operationId, + string subjectId, + string state, + string reasonCode = "", + string sourceId = "", + string endpoint = "", + string schemaVersion = "", + string libraryVersion = "") + { + if (sequence < 0) throw new ArgumentOutOfRangeException(nameof(sequence)); + Sequence = sequence; + ObservedAtUtc = observedAtUtc; + Organ = organ; + Kind = kind; + OperationId = operationId ?? ""; + SubjectId = subjectId ?? ""; + State = state ?? ""; + ReasonCode = reasonCode ?? ""; + SourceId = sourceId ?? ""; + Endpoint = endpoint ?? ""; + SchemaVersion = schemaVersion ?? ""; + LibraryVersion = string.IsNullOrWhiteSpace(libraryVersion) + ? CultMeshLibraryVersion.Current + : libraryVersion; + } + + public long Sequence { get; } + public DateTimeOffset ObservedAtUtc { get; } + public CultMeshReliabilityOrgan Organ { get; } + public CultMeshDiagnosticKind Kind { get; } + public string OperationId { get; } + public string SubjectId { get; } + public string State { get; } + public string ReasonCode { get; } + public string SourceId { get; } + public string Endpoint { get; } + public string SchemaVersion { get; } + public string LibraryVersion { get; } + } + + public interface ICultMeshDiagnosticSink + { + void Emit(CultMeshDiagnosticEvent diagnostic); + } + + public sealed class CultMeshNullDiagnosticSink : ICultMeshDiagnosticSink + { + public static CultMeshNullDiagnosticSink Instance { get; } = new CultMeshNullDiagnosticSink(); + + private CultMeshNullDiagnosticSink() + { + } + + public void Emit(CultMeshDiagnosticEvent diagnostic) + { + } + } + + /// + /// Bounded diagnostic projection suitable for inspection and tests. + /// + public sealed class CultMeshDiagnosticBuffer : ICultMeshDiagnosticSink + { + private readonly object _gate = new object(); + private readonly Queue _events; + + public CultMeshDiagnosticBuffer(int capacity = 256) + { + if (capacity <= 0) throw new ArgumentOutOfRangeException(nameof(capacity)); + Capacity = capacity; + _events = new Queue(capacity); + } + + public int Capacity { get; } + + public IReadOnlyList Snapshot() + { + lock (_gate) + { + return _events.ToArray(); + } + } + + public void Emit(CultMeshDiagnosticEvent diagnostic) + { + if (diagnostic == null) throw new ArgumentNullException(nameof(diagnostic)); + lock (_gate) + { + while (_events.Count >= Capacity) + { + _events.Dequeue(); + } + _events.Enqueue(diagnostic); + } + } + } + + public static class CultMeshLibraryVersion + { + public static string Current { get; } = Resolve(); + + private static string Resolve() + { + var assembly = typeof(CultMeshLibraryVersion).Assembly; + var informational = assembly + .GetCustomAttributes(typeof(AssemblyInformationalVersionAttribute), false) + .OfType() + .FirstOrDefault()?.InformationalVersion; + return string.IsNullOrWhiteSpace(informational) + ? assembly.GetName().Version?.ToString() ?? "unknown" + : informational; + } + } +} diff --git a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs index c6b9b4b6..0955a99b 100644 --- a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs +++ b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using GameCult.Networking; @@ -161,6 +162,15 @@ namespace GameCult.Mesh /// Defaults to endpoint selection: rudp:// uses RUDP and cultnet:// uses LiteNetLib. /// public Func? CreateClient { get; set; } + + /// Gets or sets the clock that owns discovery deadlines. + public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance; + + /// Gets or sets the bounded operational diagnostic sink. + public ICultMeshDiagnosticSink Diagnostics { get; set; } = CultMeshNullDiagnosticSink.Instance; + + /// Gets or sets the identity of this bootstrap lookup source. + public string SourceId { get; set; } = "configured-bootstrap"; } /// @@ -169,6 +179,7 @@ namespace GameCult.Mesh public sealed class CultMeshVerseDiscoveryClient { private readonly CultMeshVerseDiscoveryClientOptions _options; + private long _diagnosticSequence; /// /// Creates a Verse discovery client. @@ -202,17 +213,38 @@ namespace GameCult.Mesh client.OnCultNet(error => completion.TrySetException(new InvalidOperationException(error.Error))); - client.Connect(host, port); - var backgroundFailure = (client as ICultNetSchemaClientHealth)?.BackgroundFailure; - await WaitForConnectionAsync(client, endpoint, backgroundFailure).ConfigureAwait(false); - client.SendCultNet(new CultMeshVerseCatalogRequestMessage + Emit(CultMeshDiagnosticKind.ConnectionAttempt, messageId, endpoint, "connecting"); + try { - MessageId = messageId, - TransportVersion = request?.TransportVersion, - VerseIds = request?.VerseIds - }); + client.Connect(host, port); + var backgroundFailure = (client as ICultNetSchemaClientHealth)?.BackgroundFailure; + await WaitForConnectionAsync(client, endpoint, backgroundFailure).ConfigureAwait(false); + client.SendCultNet(new CultMeshVerseCatalogRequestMessage + { + MessageId = messageId, + TransportVersion = request?.TransportVersion, + VerseIds = request?.VerseIds + }); - return await WaitForResponseAsync(completion.Task, endpoint, backgroundFailure).ConfigureAwait(false); + var response = await WaitForResponseAsync(completion.Task, endpoint, backgroundFailure).ConfigureAwait(false); + Emit( + CultMeshDiagnosticKind.DiscoveryObservation, + messageId, + endpoint, + "fresh", + schemaVersion: response.SchemaVersion); + return response; + } + catch (Exception error) when (!(error is OperationCanceledException)) + { + Emit( + CultMeshDiagnosticKind.CandidateRejected, + messageId, + endpoint, + "unavailable", + ReasonCode(error)); + throw; + } } /// @@ -251,7 +283,7 @@ namespace GameCult.Mesh string endpoint, Task? backgroundFailure) { - var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; + var deadline = _options.Clock.UtcNow + _options.ConnectTimeout; while (!client.Connected) { if (backgroundFailure?.IsCompleted == true) @@ -261,12 +293,12 @@ namespace GameCult.Mesh $"Verse discovery client failed while connecting to {endpoint}.", error); } - if (DateTimeOffset.UtcNow >= deadline) + if (_options.Clock.UtcNow >= deadline) { throw new TimeoutException($"Timed out connecting to Verse discovery endpoint {endpoint}."); } - await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false); + await _options.Clock.DelayAsync(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false); } } @@ -275,7 +307,7 @@ namespace GameCult.Mesh string endpoint, Task? backgroundFailure) { - var timeoutTask = Task.Delay(_options.ResponseTimeout); + var timeoutTask = _options.Clock.DelayAsync(_options.ResponseTimeout); var completed = backgroundFailure == null ? await Task.WhenAny(responseTask, timeoutTask).ConfigureAwait(false) : await Task.WhenAny(responseTask, timeoutTask, backgroundFailure).ConfigureAwait(false); @@ -293,5 +325,34 @@ namespace GameCult.Mesh return await responseTask.ConfigureAwait(false); } + + private void Emit( + CultMeshDiagnosticKind kind, + string operationId, + string endpoint, + string state, + string reasonCode = "", + string schemaVersion = "") + { + _options.Diagnostics.Emit(new CultMeshDiagnosticEvent( + Interlocked.Increment(ref _diagnosticSequence), + _options.Clock.UtcNow, + CultMeshReliabilityOrgan.Discovery, + kind, + operationId, + endpoint, + state, + reasonCode, + _options.SourceId, + endpoint, + schemaVersion)); + } + + private static string ReasonCode(Exception error) + { + if (error is TimeoutException) return "timeout"; + if (error is InvalidOperationException) return "transport_failure"; + return "lookup_failure"; + } } } diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index 3e4e14ec..1ec8067d 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -202,6 +202,36 @@ failure. counters. - Diagnostics have bounded cost and can be disabled or sampled on hot paths. +### Phase 0 Body Map + +**Owner:** reliability organs own their clocks and emit typed observations; +`CultMeshDiagnosticBuffer` owns only a bounded inspection projection. + +**Inputs:** injected `ICultMeshClock`, organ state transitions, endpoint and +source identity, reason code, and served schema/library version. + +**Outputs:** ordered `CultMeshDiagnosticEvent` observations. The deterministic +test network emits scheduled packet deliveries under loss, duplication, +reordering, latency, partition, endpoint rotation, corruption, and restart. + +**Derived state:** diagnostic sequence and the bounded recent-event window. +Neither confers reachability, authority, or retry policy. + +**Forbidden writers:** diagnostic sinks and fault harnesses cannot choose a +route, mutate discovery candidates, authorize a peer, or repair a session. + +**Shared paths:** the existing Verse discovery client now uses the injected +clock for connection and response deadlines and emits the same typed timeline +for success, timeout, and transport failure. + +**Cut line:** wall-clock reads and unobservable timeout delays have been removed +from Verse discovery. Caller-owned catalog mutation and endpoint sequencing +remain compatibility authority until Phase 1 replaces and demotes them. + +**Verification layer:** tests observe the emitted discovery timeline and served +version, while the deterministic network harness proves exact replay of every +required hostile-network action. + ## Phase Closure Contract Every implementation phase must close with all of the following: diff --git a/tests/GameCult.Mesh.Tests/CultMeshDeterministicNetwork.cs b/tests/GameCult.Mesh.Tests/CultMeshDeterministicNetwork.cs new file mode 100644 index 00000000..5573204a --- /dev/null +++ b/tests/GameCult.Mesh.Tests/CultMeshDeterministicNetwork.cs @@ -0,0 +1,131 @@ +using System; +using System.Collections.Generic; +using System.Linq; + +#nullable enable + +namespace GameCult.Mesh.Tests; + +internal enum CultMeshFaultKind +{ + Drop, + Duplicate, + Delay, + Reorder, + Partition, + Corrupt, + RotateEndpoint, + Restart +} + +internal sealed record CultMeshFault( + long SendOrdinal, + CultMeshFaultKind Kind, + TimeSpan Delay = default, + string Endpoint = ""); + +internal sealed record CultMeshTestPacket( + long SendOrdinal, + string Endpoint, + byte[] Payload, + DateTimeOffset DeliverAtUtc, + long DeliveryOrdinal); + +/// +/// Transport-neutral hostile-network harness. Delivery is driven only by the +/// injected clock and schedule, so a failure timeline is exactly replayable. +/// +internal sealed class CultMeshDeterministicNetwork +{ + private readonly Func _utcNow; + private readonly Dictionary _faults; + private readonly List _pending = new(); + private long _sendOrdinal; + private long _deliveryOrdinal; + private bool _partitioned; + private string _endpoint; + + public CultMeshDeterministicNetwork( + Func utcNow, + string endpoint, + IEnumerable? faults = null) + { + _utcNow = utcNow ?? throw new ArgumentNullException(nameof(utcNow)); + _endpoint = endpoint ?? throw new ArgumentNullException(nameof(endpoint)); + _faults = (faults ?? Array.Empty()) + .GroupBy(fault => fault.SendOrdinal) + .ToDictionary(group => group.Key, group => group.ToArray()); + } + + public int RestartCount { get; private set; } + + public string Endpoint => _endpoint; + + public void Send(byte[] payload) + { + if (payload == null) throw new ArgumentNullException(nameof(payload)); + var ordinal = ++_sendOrdinal; + var copies = 1; + var delay = TimeSpan.Zero; + var reorder = false; + var dropped = false; + var bytes = payload.ToArray(); + + foreach (var fault in _faults.GetValueOrDefault(ordinal, Array.Empty())) + { + switch (fault.Kind) + { + case CultMeshFaultKind.Drop: + dropped = true; + break; + case CultMeshFaultKind.Duplicate: + copies++; + break; + case CultMeshFaultKind.Delay: + delay += fault.Delay; + break; + case CultMeshFaultKind.Reorder: + reorder = true; + break; + case CultMeshFaultKind.Partition: + _partitioned = !_partitioned; + break; + case CultMeshFaultKind.Corrupt: + if (bytes.Length > 0) bytes[0] ^= 0xff; + break; + case CultMeshFaultKind.RotateEndpoint: + _endpoint = fault.Endpoint; + break; + case CultMeshFaultKind.Restart: + _pending.Clear(); + _partitioned = false; + RestartCount++; + break; + } + } + + if (dropped || _partitioned) return; + for (var copy = 0; copy < copies; copy++) + { + var deliveryOrdinal = ++_deliveryOrdinal; + _pending.Add(new CultMeshTestPacket( + ordinal, + _endpoint, + bytes.ToArray(), + _utcNow() + delay, + reorder ? -deliveryOrdinal : deliveryOrdinal)); + } + } + + public IReadOnlyList ReceiveAvailable() + { + var now = _utcNow(); + var available = _pending + .Where(packet => packet.DeliverAtUtc <= now) + .OrderBy(packet => packet.DeliverAtUtc) + .ThenBy(packet => packet.DeliveryOrdinal) + .ToArray(); + foreach (var packet in available) _pending.Remove(packet); + return available; + } +} diff --git a/tests/GameCult.Mesh.Tests/CultMeshReliabilityDiagnosticsTests.cs b/tests/GameCult.Mesh.Tests/CultMeshReliabilityDiagnosticsTests.cs new file mode 100644 index 00000000..d40ee4e0 --- /dev/null +++ b/tests/GameCult.Mesh.Tests/CultMeshReliabilityDiagnosticsTests.cs @@ -0,0 +1,118 @@ +using System; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using NUnit.Framework; + +namespace GameCult.Mesh.Tests; + +public sealed class CultMeshReliabilityDiagnosticsTests +{ + [Test] + public void DiagnosticBuffer_IsBoundedAndCarriesServedVersion() + { + var clock = new TestClock(new DateTimeOffset(2026, 7, 12, 18, 0, 0, TimeSpan.Zero)); + var diagnostics = new CultMeshDiagnosticBuffer(capacity: 2); + + for (var sequence = 1; sequence <= 3; sequence++) + { + diagnostics.Emit(new CultMeshDiagnosticEvent( + sequence, + clock.UtcNow, + CultMeshReliabilityOrgan.Discovery, + CultMeshDiagnosticKind.DiscoveryObservation, + "lookup:aetheria", + "aetheria", + sequence == 3 ? "degraded" : "fresh", + schemaVersion: "gamecult.mesh.discovery_diagnostic.v1")); + clock.Advance(TimeSpan.FromSeconds(1)); + } + + var snapshot = diagnostics.Snapshot(); + snapshot.Select(item => item.Sequence).Should().Equal(2, 3); + snapshot[^1].State.Should().Be("degraded"); + snapshot[^1].LibraryVersion.Should().Be(CultMeshLibraryVersion.Current); + snapshot[^1].LibraryVersion.Should().NotBeNullOrWhiteSpace(); + snapshot[^1].SchemaVersion.Should().Be("gamecult.mesh.discovery_diagnostic.v1"); + } + + [Test] + public void DeterministicNetwork_ReplaysLossDuplicationReorderingLatencyAndCorruption() + { + var clock = new TestClock(new DateTimeOffset(2026, 7, 12, 18, 0, 0, TimeSpan.Zero)); + var network = new CultMeshDeterministicNetwork( + () => clock.UtcNow, + "rudp://first:3076", + new[] + { + new CultMeshFault(1, CultMeshFaultKind.Drop), + new CultMeshFault(2, CultMeshFaultKind.Duplicate), + new CultMeshFault(2, CultMeshFaultKind.Delay, TimeSpan.FromSeconds(2)), + new CultMeshFault(3, CultMeshFaultKind.Reorder), + new CultMeshFault(3, CultMeshFaultKind.Corrupt) + }); + + network.Send(Encoding.UTF8.GetBytes("lost")); + network.Send(Encoding.UTF8.GetBytes("twice")); + network.Send(new byte[] { 0x01, 0x02 }); + + var immediate = network.ReceiveAvailable(); + immediate.Should().ContainSingle(); + immediate[0].SendOrdinal.Should().Be(3); + immediate[0].Payload.Should().Equal(0xfe, 0x02); + + clock.Advance(TimeSpan.FromSeconds(2)); + var delayed = network.ReceiveAvailable(); + delayed.Should().HaveCount(2); + delayed.Select(packet => Encoding.UTF8.GetString(packet.Payload)).Should().OnlyContain(value => value == "twice"); + } + + [Test] + public void DeterministicNetwork_ReplaysPartitionEndpointRotationAndRestart() + { + var clock = new TestClock(new DateTimeOffset(2026, 7, 12, 18, 0, 0, TimeSpan.Zero)); + var network = new CultMeshDeterministicNetwork( + () => clock.UtcNow, + "rudp://first:3076", + new[] + { + new CultMeshFault(1, CultMeshFaultKind.Partition), + new CultMeshFault(2, CultMeshFaultKind.Partition), + new CultMeshFault(2, CultMeshFaultKind.RotateEndpoint, Endpoint: "rudp://second:3076"), + new CultMeshFault(3, CultMeshFaultKind.Restart) + }); + + network.Send(new byte[] { 1 }); + network.Send(new byte[] { 2 }); + network.ReceiveAvailable().Should().ContainSingle().Which.Endpoint.Should().Be("rudp://second:3076"); + + network.Send(new byte[] { 3 }); + network.RestartCount.Should().Be(1); + network.Endpoint.Should().Be("rudp://second:3076"); + network.ReceiveAvailable().Should().ContainSingle().Which.Payload.Should().Equal(3); + } + + private sealed class TestClock : ICultMeshClock + { + public TestClock(DateTimeOffset utcNow) + { + UtcNow = utcNow; + } + + public DateTimeOffset UtcNow { get; private set; } + + public Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + Advance(delay); + return Task.CompletedTask; + } + + public void Advance(TimeSpan duration) + { + UtcNow += duration; + } + } +} diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 42707754..a08dc846 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -2789,6 +2789,38 @@ namespace GameCult.Networking.Tests Assert.That(response.Verses.Single().VerseId, Is.EqualTo("local")); } + [Test] + public async Task CultMeshVerseDiscoveryClient_ReportsDeterministicConnectionFailureTimeline() + { + var clock = new ManualCultMeshClock(new DateTimeOffset(2026, 7, 12, 18, 0, 0, TimeSpan.Zero)); + var diagnostics = new CultMeshDiagnosticBuffer(); + var client = new CultMeshVerseDiscoveryClient(new CultMeshVerseDiscoveryClientOptions + { + CreateClient = () => new NeverConnectedSchemaClient(), + Clock = clock, + Diagnostics = diagnostics, + SourceId = "test-bootstrap", + ConnectTimeout = TimeSpan.FromMilliseconds(50) + }); + + var error = Assert.ThrowsAsync(() => + client.FetchAsync("rudp://unavailable.example.test:4075")); + Assert.That(error!.Message, Does.Contain("unavailable.example.test")); + + var timeline = diagnostics.Snapshot(); + Assert.That(timeline.Select(item => item.Kind), Is.EqualTo(new[] + { + CultMeshDiagnosticKind.ConnectionAttempt, + CultMeshDiagnosticKind.CandidateRejected + })); + Assert.That(timeline[0].ObservedAtUtc, Is.EqualTo(new DateTimeOffset(2026, 7, 12, 18, 0, 0, TimeSpan.Zero))); + Assert.That(timeline[1].ObservedAtUtc, Is.EqualTo(new DateTimeOffset(2026, 7, 12, 18, 0, 0, 50, TimeSpan.Zero))); + Assert.That(timeline[1].State, Is.EqualTo("unavailable")); + Assert.That(timeline[1].ReasonCode, Is.EqualTo("timeout")); + Assert.That(timeline[1].SourceId, Is.EqualTo("test-bootstrap")); + Assert.That(timeline[1].LibraryVersion, Is.Not.Empty); + } + [Test] public async Task CultMeshPeerExchangeClient_Uses_SchemaClientPort() { @@ -4912,6 +4944,45 @@ namespace GameCult.Networking.Tests } } + private sealed class NeverConnectedSchemaClient : ICultNetSchemaClient + { + public bool Connected => false; + + public void Connect(string host, int port) + { + } + + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + throw new InvalidOperationException("A disconnected client cannot send."); + } + + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + } + + public void Dispose() + { + } + } + + private sealed class ManualCultMeshClock : ICultMeshClock + { + public ManualCultMeshClock(DateTimeOffset utcNow) + { + UtcNow = utcNow; + } + + public DateTimeOffset UtcNow { get; private set; } + + public Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + UtcNow += delay; + return Task.CompletedTask; + } + } + [MessagePack.MessagePackObject] public sealed class PlayerDataPayload {