Establish CultMesh reliability failure harness

This commit is contained in:
Meta Vens 2026-07-12 19:56:20 +01:00
commit fc65e25fd5
6 changed files with 607 additions and 13 deletions

View file

@ -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
{
/// <summary>
/// Clock used by CultMesh reliability owners.
/// </summary>
public interface ICultMeshClock
{
DateTimeOffset UtcNow { get; }
Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken = default);
}
/// <summary>
/// Wall clock used outside deterministic tests.
/// </summary>
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
}
/// <summary>
/// One bounded operational observation emitted by a CultMesh reliability owner.
/// </summary>
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)
{
}
}
/// <summary>
/// Bounded diagnostic projection suitable for inspection and tests.
/// </summary>
public sealed class CultMeshDiagnosticBuffer : ICultMeshDiagnosticSink
{
private readonly object _gate = new object();
private readonly Queue<CultMeshDiagnosticEvent> _events;
public CultMeshDiagnosticBuffer(int capacity = 256)
{
if (capacity <= 0) throw new ArgumentOutOfRangeException(nameof(capacity));
Capacity = capacity;
_events = new Queue<CultMeshDiagnosticEvent>(capacity);
}
public int Capacity { get; }
public IReadOnlyList<CultMeshDiagnosticEvent> 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<AssemblyInformationalVersionAttribute>()
.FirstOrDefault()?.InformationalVersion;
return string.IsNullOrWhiteSpace(informational)
? assembly.GetName().Version?.ToString() ?? "unknown"
: informational;
}
}
}

View file

@ -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.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
/// <summary>Gets or sets the clock that owns discovery deadlines.</summary>
public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance;
/// <summary>Gets or sets the bounded operational diagnostic sink.</summary>
public ICultMeshDiagnosticSink Diagnostics { get; set; } = CultMeshNullDiagnosticSink.Instance;
/// <summary>Gets or sets the identity of this bootstrap lookup source.</summary>
public string SourceId { get; set; } = "configured-bootstrap";
}
/// <summary>
@ -169,6 +179,7 @@ namespace GameCult.Mesh
public sealed class CultMeshVerseDiscoveryClient
{
private readonly CultMeshVerseDiscoveryClientOptions _options;
private long _diagnosticSequence;
/// <summary>
/// Creates a Verse discovery client.
@ -202,17 +213,38 @@ namespace GameCult.Mesh
client.OnCultNet<CultNetErrorMessage>(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;
}
}
/// <summary>
@ -251,7 +283,7 @@ namespace GameCult.Mesh
string endpoint,
Task<Exception>? 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<Exception>? 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";
}
}
}

View file

@ -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:

View file

@ -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);
/// <summary>
/// Transport-neutral hostile-network harness. Delivery is driven only by the
/// injected clock and schedule, so a failure timeline is exactly replayable.
/// </summary>
internal sealed class CultMeshDeterministicNetwork
{
private readonly Func<DateTimeOffset> _utcNow;
private readonly Dictionary<long, CultMeshFault[]> _faults;
private readonly List<CultMeshTestPacket> _pending = new();
private long _sendOrdinal;
private long _deliveryOrdinal;
private bool _partitioned;
private string _endpoint;
public CultMeshDeterministicNetwork(
Func<DateTimeOffset> utcNow,
string endpoint,
IEnumerable<CultMeshFault>? faults = null)
{
_utcNow = utcNow ?? throw new ArgumentNullException(nameof(utcNow));
_endpoint = endpoint ?? throw new ArgumentNullException(nameof(endpoint));
_faults = (faults ?? Array.Empty<CultMeshFault>())
.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<CultMeshFault>()))
{
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<CultMeshTestPacket> 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;
}
}

View file

@ -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;
}
}
}

View file

@ -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<TimeoutException>(() =>
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>(T message) where T : ICultNetSchemaMessage
{
throw new InvalidOperationException("A disconnected client cannot send.");
}
public void OnCultNet<T>(Action<T> 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
{