336 lines
15 KiB
C#
336 lines
15 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics;
|
|
using System.IO;
|
|
using System.Linq;
|
|
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using FluentAssertions;
|
|
using GameCult.Caching;
|
|
using GameCult.Networking;
|
|
using NUnit.Framework;
|
|
|
|
namespace GameCult.Mesh.Tests;
|
|
|
|
[TestFixture]
|
|
public sealed class CultMeshContentSessionTests
|
|
{
|
|
private const string RouteGeneration = "content-test-route-1";
|
|
private string _directory = null!;
|
|
|
|
[SetUp]
|
|
public void SetUp() =>
|
|
_directory = Path.Combine(Path.GetTempPath(), "cultmesh-content-session-tests", Guid.NewGuid().ToString("N"));
|
|
|
|
[TearDown]
|
|
public void TearDown()
|
|
{
|
|
if (Directory.Exists(_directory)) Directory.Delete(_directory, true);
|
|
}
|
|
|
|
[Test]
|
|
public async Task TcpContentSessionStreamsLargeBodyAndReusesCommittedBody()
|
|
{
|
|
var payload = Enumerable.Range(0, 700_000).Select(value => (byte)(value % 251)).ToArray();
|
|
var artifact = CultMesh.PackCdnArtifact("aetheria/world/windows", payload);
|
|
var providerChunks = artifact.Chunks.ToDictionary(
|
|
chunk => (chunk.ChunkHash.StartsWith("sha256:", StringComparison.OrdinalIgnoreCase)
|
|
? chunk.ChunkHash.Substring("sha256:".Length)
|
|
: chunk.ChunkHash).Trim().ToLowerInvariant(),
|
|
StringComparer.Ordinal);
|
|
using var contentServer = new CultMeshTcpContentServer(
|
|
new TcpListener(IPAddress.Loopback, 0),
|
|
hash => providerChunks.TryGetValue(hash, out var chunk) ? chunk : null,
|
|
new CultMeshTcpContentServerOptions
|
|
{
|
|
VerseId = "aetheria",
|
|
AuthorityRuntimeId = "aetheria.daemon",
|
|
RouteGeneration = RouteGeneration
|
|
});
|
|
var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}";
|
|
var connector = new CultMeshTcpContentTransportConnector();
|
|
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) });
|
|
using var sessions = new CultMeshSessionManager(
|
|
discovery,
|
|
Array.Empty<ICultMeshTransportConnector>(),
|
|
new ICultMeshContentTransportConnector[] { connector }, CultMeshTestTrust.LocalSessions);
|
|
var provider = new CultMeshSessionContentProvider(
|
|
"aetheria.daemon",
|
|
sessions,
|
|
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
|
|
using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
|
|
typeof(CultMeshContentTransferStateDocument)));
|
|
var transfer = new CultMeshContentTransferService(
|
|
transferState,
|
|
new[] { provider },
|
|
new CultMeshContentTransferOptions(_directory));
|
|
|
|
var first = await transfer.FetchAsync(artifact.Manifest);
|
|
var requestsAfterCold = contentServer.ChunkRequestsServed;
|
|
var second = await transfer.FetchAsync(artifact.Manifest);
|
|
|
|
File.ReadAllBytes(first).Should().Equal(payload);
|
|
second.Should().Be(first);
|
|
requestsAfterCold.Should().Be(artifact.Manifest.Chunks.Length);
|
|
contentServer.ChunkRequestsServed.Should().Be(requestsAfterCold,
|
|
"a committed verified body is the warm-cache authority");
|
|
connector.ConnectCount.Should().Be(1,
|
|
"all chunk requests borrow one identity-first content session");
|
|
}
|
|
|
|
[Test]
|
|
public async Task TcpContentSessionRejectsEndpointServingAnotherAuthority()
|
|
{
|
|
using var content = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
|
|
typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk)));
|
|
using var contentServer = new CultMeshTcpContentServer(
|
|
new TcpListener(IPAddress.Loopback, 0),
|
|
content,
|
|
new CultMeshTcpContentServerOptions
|
|
{
|
|
VerseId = "aetheria",
|
|
AuthorityRuntimeId = "intruder.daemon",
|
|
RouteGeneration = RouteGeneration
|
|
});
|
|
var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}";
|
|
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) });
|
|
using var sessions = new CultMeshSessionManager(
|
|
discovery,
|
|
Array.Empty<ICultMeshTransportConnector>(),
|
|
new ICultMeshContentTransportConnector[] { new CultMeshTcpContentTransportConnector() }, CultMeshTestTrust.LocalSessions);
|
|
|
|
Func<Task> connect = async () => await sessions.ConnectContentAsync(
|
|
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
|
|
|
|
var error = await connect.Should().ThrowAsync<CultMeshSessionException>();
|
|
error.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority);
|
|
}
|
|
|
|
[Test]
|
|
public async Task RudpContentSessionCarriesConcurrentChunkWindowWithoutPerFragmentSchedulerDelay()
|
|
{
|
|
var payload = Enumerable.Range(0, 1024 * 1024).Select(value => (byte)(value % 251)).ToArray();
|
|
var artifact = CultMesh.PackCdnArtifact("aetheria/world/rudp", payload);
|
|
using var providerCache = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
|
|
typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk)));
|
|
await CultMeshCdn.PublishAsync(providerCache, artifact);
|
|
using var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
|
|
socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
|
using var wireServer = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
|
|
{
|
|
RuntimeId = "cultmesh-content-rudp-test",
|
|
Socket = socket,
|
|
MaxFragmentBytes = 1024,
|
|
MaxPendingReliablePackets = 8192
|
|
});
|
|
using var contentServer = new CultMeshLegacyRudpContentServer(wireServer, providerCache);
|
|
using var identityServer = new CultMeshSessionIdentityServer(
|
|
wireServer,
|
|
"aetheria.daemon",
|
|
new[] { "aetheria" },
|
|
new[] { CultMeshProtocols.Content.Value },
|
|
new[] { RouteGeneration });
|
|
using var pumpCancellation = new CancellationTokenSource();
|
|
var pump = Task.Run(async () =>
|
|
{
|
|
while (!pumpCancellation.IsCancellationRequested)
|
|
{
|
|
var progress = await wireServer.PollAvailableAsync(256);
|
|
wireServer.PollResends();
|
|
if (progress.TransportItemsConsumed == 0)
|
|
await Task.Delay(1, pumpCancellation.Token);
|
|
}
|
|
});
|
|
|
|
var endpoint = $"rudp://127.0.0.1:{wireServer.LocalEndPoint.Port}";
|
|
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) });
|
|
var connector = new CultMeshLegacyRudpContentTransportConnector(
|
|
new CultMeshLegacyRudpContentTransportOptions { ResponseTimeout = TimeSpan.FromSeconds(20) });
|
|
using var sessions = new CultMeshSessionManager(
|
|
discovery,
|
|
Array.Empty<ICultMeshTransportConnector>(),
|
|
new ICultMeshContentTransportConnector[] { connector }, CultMeshTestTrust.LocalSessions);
|
|
var provider = new CultMeshSessionContentProvider(
|
|
"aetheria.daemon",
|
|
sessions,
|
|
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
|
|
using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
|
|
typeof(CultMeshContentTransferStateDocument)));
|
|
var transfer = new CultMeshContentTransferService(
|
|
transferState,
|
|
new[] { provider },
|
|
new CultMeshContentTransferOptions(_directory));
|
|
|
|
var elapsed = Stopwatch.StartNew();
|
|
string path;
|
|
try
|
|
{
|
|
path = await transfer.FetchAsync(artifact.Manifest);
|
|
}
|
|
catch (Exception error)
|
|
{
|
|
var stats = wireServer.Stats;
|
|
Assert.Fail(
|
|
$"RUDP content transfer failed after {elapsed.Elapsed}; " +
|
|
$"server sent {stats.BytesSent} bytes and received {stats.BytesReceived} bytes; " +
|
|
$"pump status is {pump.Status}: {error}");
|
|
return;
|
|
}
|
|
finally
|
|
{
|
|
pumpCancellation.Cancel();
|
|
try { await pump; }
|
|
catch (OperationCanceledException) { }
|
|
}
|
|
elapsed.Stop();
|
|
|
|
File.ReadAllBytes(path).Should().Equal(payload);
|
|
elapsed.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(10),
|
|
"a bounded content request window must not pay one scheduler quantum per RUDP fragment");
|
|
}
|
|
|
|
[Test]
|
|
public void ContentWireMessagesRoundTripThroughTheSchemaDispatcher()
|
|
{
|
|
var request = new CultMeshContentChunkRequestMessage
|
|
{
|
|
MessageId = "request-1",
|
|
ChunkHash = new string('a', 64),
|
|
RecordKey = "mesh:cdn:chunk:" + new string('a', 64),
|
|
ExpectedSizeBytes = 42
|
|
};
|
|
var response = new CultMeshContentChunkResponseMessage
|
|
{
|
|
MessageId = "request-1",
|
|
Found = true,
|
|
ChunkHash = new string('a', 64),
|
|
SizeBytes = 3,
|
|
Payload = new byte[] { 1, 2, 3 }
|
|
};
|
|
|
|
CultNetSchemaMessageSerialization.Deserialize(CultNetSchemaMessageSerialization.Serialize(request))
|
|
.Should().BeEquivalentTo(request);
|
|
CultNetSchemaMessageSerialization.Deserialize(CultNetSchemaMessageSerialization.Serialize(response))
|
|
.Should().BeEquivalentTo(response);
|
|
}
|
|
|
|
private sealed class RouteSource : ICultMeshLookupSource
|
|
{
|
|
private readonly string _endpoint;
|
|
|
|
public RouteSource(string endpoint = "rudp://127.0.0.14:3076") => _endpoint = endpoint;
|
|
|
|
public string SourceId => "test-rendezvous";
|
|
|
|
public Task<IReadOnlyList<CultMeshDiscoveryObservation>> LookupAsync(
|
|
CultMeshDiscoveryQuery query,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
return Task.FromResult<IReadOnlyList<CultMeshDiscoveryObservation>>(new[]
|
|
{
|
|
new CultMeshDiscoveryObservation(
|
|
new CultMeshVerseDescriptor(
|
|
"aetheria",
|
|
"Aetheria",
|
|
CultMeshVerseAuthorityModel.OperatorCluster,
|
|
new CultMeshVerseCompatibility("cultmesh.v1", "rules"),
|
|
authorityRoutes: new[]
|
|
{
|
|
new CultMeshAuthorityRoute(
|
|
"aetheria.daemon",
|
|
_endpoint,
|
|
new[] { CultMeshProtocols.Content.Value },
|
|
generation: RouteGeneration)
|
|
}),
|
|
SourceId,
|
|
now,
|
|
now.AddMinutes(1),
|
|
CultMeshDiscoveryTrust.Signed)
|
|
});
|
|
}
|
|
}
|
|
|
|
private sealed class LoopbackConnector : ICultMeshTransportConnector
|
|
{
|
|
private readonly ICultNetSchemaClient _client;
|
|
private int _connectCount;
|
|
public LoopbackConnector(ICultNetSchemaClient client) => _client = client;
|
|
public string ConnectorId => "loopback";
|
|
public int Priority => 0;
|
|
public int ConnectCount => Volatile.Read(ref _connectCount);
|
|
public bool CanConnect(CultMeshTransportCandidate candidate) => true;
|
|
public Task<ICultNetSchemaClient> ConnectAsync(
|
|
CultMeshTransportCandidate candidate,
|
|
CultMeshProtocolId protocol,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
Interlocked.Increment(ref _connectCount);
|
|
protocol.Should().Be(CultMeshProtocols.Content);
|
|
return Task.FromResult(_client);
|
|
}
|
|
}
|
|
|
|
private sealed class LoopbackSchemaServer : ICultNetSchemaServer
|
|
{
|
|
private readonly Dictionary<Type, Delegate> _handlers = new();
|
|
|
|
public void OnCultNet<TMessage>(Func<TMessage, ICultNetSchemaServerPeer, Task> callback)
|
|
where TMessage : ICultNetSchemaMessage => _handlers[typeof(TMessage)] = callback;
|
|
|
|
public void RemoveCultNetMessageListener<TMessage>(Delegate callback)
|
|
where TMessage : ICultNetSchemaMessage => _handlers.Remove(typeof(TMessage));
|
|
|
|
public Task DispatchAsync<TMessage>(TMessage message, ICultNetSchemaServerPeer peer)
|
|
where TMessage : ICultNetSchemaMessage =>
|
|
((Func<TMessage, ICultNetSchemaServerPeer, Task>)_handlers[typeof(TMessage)])(message, peer);
|
|
}
|
|
|
|
private sealed class LoopbackSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient
|
|
{
|
|
private readonly LoopbackSchemaServer _server;
|
|
private readonly Dictionary<Type, List<Delegate>> _handlers = new();
|
|
public LoopbackSchemaClient(LoopbackSchemaServer server) => _server = server;
|
|
public bool Connected => true;
|
|
public int ContentRequestCount { get; private set; }
|
|
public int SnapshotRequestCount { get; private set; }
|
|
public void Connect(string host, int port) { }
|
|
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage
|
|
{
|
|
if (message is CultMeshContentChunkRequestMessage content)
|
|
{
|
|
ContentRequestCount++;
|
|
_server.DispatchAsync(content, new LoopbackPeer(this)).GetAwaiter().GetResult();
|
|
}
|
|
else if (message is CultNetSnapshotRequestMessage)
|
|
{
|
|
SnapshotRequestCount++;
|
|
throw new InvalidOperationException("Content session attempted to use a snapshot request.");
|
|
}
|
|
}
|
|
public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
|
|
{
|
|
if (!_handlers.TryGetValue(typeof(T), out var handlers))
|
|
_handlers[typeof(T)] = handlers = new List<Delegate>();
|
|
handlers.Add(callback);
|
|
}
|
|
public void Dispose() { }
|
|
public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) =>
|
|
verseId == "aetheria" && authorityRuntimeId == "aetheria.daemon";
|
|
public void Emit<T>(T message) where T : ICultNetSchemaMessage
|
|
{
|
|
if (!_handlers.TryGetValue(typeof(T), out var handlers)) return;
|
|
foreach (var handler in handlers.ToArray()) ((Action<T>)handler)(message);
|
|
}
|
|
}
|
|
|
|
private sealed class LoopbackPeer : ICultNetSchemaServerPeer
|
|
{
|
|
private readonly LoopbackSchemaClient _client;
|
|
public LoopbackPeer(LoopbackSchemaClient client) => _client = client;
|
|
public void SendCultNet<TMessage>(TMessage message) where TMessage : ICultNetSchemaMessage =>
|
|
_client.Emit(message);
|
|
}
|
|
}
|