Move hot bodies off snapshot state
This commit is contained in:
Meta Vens
parent
0ef59b50c2
commit
091200c09d
12 changed files with 643 additions and 364 deletions
|
|
@ -14,7 +14,7 @@
|
|||
<PackageLicenseExpression>MIT</PackageLicenseExpression>
|
||||
<Copyright>© Meta Vens and contributors</Copyright>
|
||||
<PublishRepositoryUrl>true</PublishRepositoryUrl>
|
||||
<Version>1.0.30</Version>
|
||||
<Version>1.0.31</Version>
|
||||
|
||||
<!-- Symbols/debugging -->
|
||||
<EmbedUntrackedSources>true</EmbedUntrackedSources>
|
||||
|
|
|
|||
|
|
@ -572,10 +572,19 @@ namespace GameCult.Mesh
|
|||
public sealed class CultMeshNetworkBodyAdapter : ICultMeshBodyTransportAdapter
|
||||
{
|
||||
private readonly Func<CultMeshBodyDescriptor, byte[]> _fetch;
|
||||
private readonly bool _fetchIsVerified;
|
||||
|
||||
public CultMeshNetworkBodyAdapter(Func<CultMeshBodyDescriptor, byte[]> fetch) =>
|
||||
_fetch = fetch ?? throw new ArgumentNullException(nameof(fetch));
|
||||
|
||||
/// <summary>Uses a session provider that already verifies the response against the descriptor.</summary>
|
||||
public CultMeshNetworkBodyAdapter(CultMeshSessionBodyProvider provider)
|
||||
{
|
||||
if (provider == null) throw new ArgumentNullException(nameof(provider));
|
||||
_fetch = provider.CreateFetchDelegate();
|
||||
_fetchIsVerified = true;
|
||||
}
|
||||
|
||||
public CultMeshBodyTransportKind TransportKind => CultMeshBodyTransportKind.Network;
|
||||
public bool CanOpen(CultMeshBodyDescriptor descriptor) =>
|
||||
descriptor != null && descriptor.TransportKind == TransportKind;
|
||||
|
|
@ -590,9 +599,12 @@ namespace GameCult.Mesh
|
|||
throw new InvalidDataException("CultMesh network body size does not match its descriptor.");
|
||||
if (string.IsNullOrWhiteSpace(descriptor.SemanticHash))
|
||||
throw new InvalidDataException("CultMesh network body descriptor has no semantic digest.");
|
||||
var actualHash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(bytes);
|
||||
if (!string.Equals(actualHash, descriptor.SemanticHash, StringComparison.Ordinal))
|
||||
throw new InvalidDataException("CultMesh network body semantic digest does not match its descriptor.");
|
||||
if (!_fetchIsVerified)
|
||||
{
|
||||
var actualHash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(bytes);
|
||||
if (!string.Equals(actualHash, descriptor.SemanticHash, StringComparison.Ordinal))
|
||||
throw new InvalidDataException("CultMesh network body semantic digest does not match its descriptor.");
|
||||
}
|
||||
return new CultMeshBufferedBodyLease(descriptor, bytes);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ using System.IO;
|
|||
using System.Linq;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using GameCult.Caching;
|
||||
using GameCult.Networking;
|
||||
using MessagePack;
|
||||
|
|
@ -15,7 +14,6 @@ namespace GameCult.Mesh
|
|||
public static class CultMeshBodyPublicationSchemaVersions
|
||||
{
|
||||
public const string Publication = "gamecult.mesh.body_publication.v2";
|
||||
public const string NetworkBody = "gamecult.mesh.network_body.v1";
|
||||
}
|
||||
|
||||
public sealed class CultMeshBodyGeneration
|
||||
|
|
@ -31,203 +29,6 @@ namespace GameCult.Mesh
|
|||
public long LeaseExpiresAtUnixMs { get; set; }
|
||||
}
|
||||
|
||||
[MessagePackObject]
|
||||
[CultDocument("gamecult.mesh.network_body", CultMeshBodyPublicationSchemaVersions.NetworkBody)]
|
||||
public sealed class CultMeshNetworkBodyDocument
|
||||
{
|
||||
[Key(0), CultName] public string CapabilityToken { get; set; } = string.Empty;
|
||||
[Key(1), CultIndex] public string BodyId { get; set; } = string.Empty;
|
||||
[Key(2), CultIndex] public string ProducerId { get; set; } = string.Empty;
|
||||
[Key(3)] public string SchemaId { get; set; } = string.Empty;
|
||||
[Key(4)] public int LayoutVersion { get; set; }
|
||||
[Key(5)] public long ByteSize { get; set; }
|
||||
[Key(6)] public int Capacity { get; set; }
|
||||
[Key(7)] public long ProducerEpoch { get; set; }
|
||||
[Key(8)] public long Sequence { get; set; }
|
||||
[Key(9)] public CultMeshBodySynchronization Synchronization { get; set; }
|
||||
[Key(10)] public long LeaseExpiresAtUnixMs { get; set; }
|
||||
[Key(11)] public string SemanticHash { get; set; } = string.Empty;
|
||||
[Key(12)] public string ManifestRecordKey { get; set; } = string.Empty;
|
||||
|
||||
public static CultRecordKey CreateRecordKey(string capabilityToken)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(capabilityToken))
|
||||
throw new ArgumentException("Network body capability token is required.", nameof(capabilityToken));
|
||||
return new CultRecordKey("mesh:network-body:" + capabilityToken);
|
||||
}
|
||||
|
||||
[IgnoreMember]
|
||||
public CultRecordKey RecordKey => CreateRecordKey(CapabilityToken);
|
||||
}
|
||||
|
||||
public sealed class CultMeshNetworkBodyPublisher
|
||||
{
|
||||
private readonly CultCache _cache;
|
||||
private readonly Func<CultMeshBodyGeneration, bool> _authorizePublication;
|
||||
private readonly int _chunkSizeBytes;
|
||||
|
||||
public CultMeshNetworkBodyPublisher(
|
||||
CultCache cache,
|
||||
Func<CultMeshBodyGeneration, bool> authorizePublication,
|
||||
int chunkSizeBytes = CultMeshCdnPackOptions.DefaultChunkSizeBytes)
|
||||
{
|
||||
_cache = cache ?? throw new ArgumentNullException(nameof(cache));
|
||||
_authorizePublication = authorizePublication ?? throw new ArgumentNullException(nameof(authorizePublication));
|
||||
if (chunkSizeBytes <= 0 || chunkSizeBytes > CultMeshCdnPackOptions.DefaultChunkSizeBytes)
|
||||
throw new ArgumentOutOfRangeException(nameof(chunkSizeBytes), "Network body chunks must be positive and no larger than the CultMesh CDN default bound.");
|
||||
_chunkSizeBytes = chunkSizeBytes;
|
||||
}
|
||||
|
||||
public async Task<CultMeshBodyDescriptor> PublishAsync(CultMeshBodyGeneration generation, byte[] body)
|
||||
{
|
||||
ValidateGeneration(generation);
|
||||
if (body == null) throw new ArgumentNullException(nameof(body));
|
||||
if (!_authorizePublication(generation))
|
||||
throw new UnauthorizedAccessException("CultMesh body producer is not authorized to publish this logical body.");
|
||||
|
||||
var semanticHash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(body);
|
||||
var capabilityToken = CreateCapabilityToken(generation);
|
||||
var recordKey = CultMeshNetworkBodyDocument.CreateRecordKey(capabilityToken);
|
||||
var existing = _cache.Get<CultMeshNetworkBodyDocument>(recordKey);
|
||||
if (existing != null)
|
||||
{
|
||||
ValidateImmutableGeneration(existing, generation, body.LongLength, semanticHash);
|
||||
return CreateDescriptor(existing);
|
||||
}
|
||||
|
||||
var artifact = CultMeshCdn.PackArtifact(
|
||||
generation.BodyId,
|
||||
body,
|
||||
new CultMeshCdnPackOptions
|
||||
{
|
||||
ChunkSizeBytes = _chunkSizeBytes,
|
||||
Kind = CultMeshCdnArtifactKinds.Package,
|
||||
Version = generation.ProducerEpoch.ToString(CultureInfo.InvariantCulture) + "." + generation.Sequence.ToString(CultureInfo.InvariantCulture)
|
||||
});
|
||||
await CultMeshCdn.PublishAsync(_cache, artifact).ConfigureAwait(false);
|
||||
|
||||
var document = new CultMeshNetworkBodyDocument
|
||||
{
|
||||
CapabilityToken = capabilityToken,
|
||||
BodyId = generation.BodyId,
|
||||
ProducerId = generation.ProducerId,
|
||||
SchemaId = generation.SchemaId,
|
||||
LayoutVersion = generation.LayoutVersion,
|
||||
ByteSize = body.LongLength,
|
||||
Capacity = generation.Capacity,
|
||||
ProducerEpoch = generation.ProducerEpoch,
|
||||
Sequence = generation.Sequence,
|
||||
Synchronization = generation.Synchronization,
|
||||
LeaseExpiresAtUnixMs = generation.LeaseExpiresAtUnixMs,
|
||||
SemanticHash = semanticHash,
|
||||
ManifestRecordKey = artifact.ManifestKey.Value
|
||||
};
|
||||
await _cache.UpsertAsync(document, new CultRecordHandle<CultMeshNetworkBodyDocument>(recordKey)).ConfigureAwait(false);
|
||||
return CreateDescriptor(document);
|
||||
}
|
||||
|
||||
private static void ValidateGeneration(CultMeshBodyGeneration generation)
|
||||
{
|
||||
if (generation == null) throw new ArgumentNullException(nameof(generation));
|
||||
if (string.IsNullOrWhiteSpace(generation.BodyId)) throw new ArgumentException("Body identity is required.", nameof(generation));
|
||||
if (string.IsNullOrWhiteSpace(generation.ProducerId)) throw new ArgumentException("Producer identity is required.", nameof(generation));
|
||||
if (string.IsNullOrWhiteSpace(generation.SchemaId)) throw new ArgumentException("Schema identity is required.", nameof(generation));
|
||||
if (generation.LayoutVersion < 0 || generation.Capacity < 0 || generation.ProducerEpoch < 0 || generation.Sequence < 0)
|
||||
throw new ArgumentOutOfRangeException(nameof(generation), "Body generation values must be non-negative.");
|
||||
if (generation.LeaseExpiresAtUnixMs <= 0) throw new ArgumentOutOfRangeException(nameof(generation), "Body generation requires a positive lease expiry.");
|
||||
}
|
||||
|
||||
private static string CreateCapabilityToken(CultMeshBodyGeneration generation)
|
||||
{
|
||||
using var stream = new MemoryStream();
|
||||
using (var writer = new BinaryWriter(stream, Encoding.UTF8, leaveOpen: true))
|
||||
{
|
||||
writer.Write(generation.BodyId);
|
||||
writer.Write(generation.ProducerEpoch);
|
||||
writer.Write(generation.Sequence);
|
||||
}
|
||||
using var sha256 = SHA256.Create();
|
||||
return string.Concat(sha256.ComputeHash(stream.ToArray()).Select(value => value.ToString("x2", CultureInfo.InvariantCulture)));
|
||||
}
|
||||
|
||||
private static void ValidateImmutableGeneration(
|
||||
CultMeshNetworkBodyDocument existing,
|
||||
CultMeshBodyGeneration generation,
|
||||
long byteSize,
|
||||
string semanticHash)
|
||||
{
|
||||
if (!string.Equals(existing.BodyId, generation.BodyId, StringComparison.Ordinal) ||
|
||||
!string.Equals(existing.ProducerId, generation.ProducerId, StringComparison.Ordinal) ||
|
||||
!string.Equals(existing.SchemaId, generation.SchemaId, StringComparison.Ordinal) ||
|
||||
existing.LayoutVersion != generation.LayoutVersion || existing.ByteSize != byteSize ||
|
||||
existing.Capacity != generation.Capacity || existing.ProducerEpoch != generation.ProducerEpoch ||
|
||||
existing.Sequence != generation.Sequence || existing.Synchronization != generation.Synchronization ||
|
||||
existing.LeaseExpiresAtUnixMs != generation.LeaseExpiresAtUnixMs ||
|
||||
!string.Equals(existing.SemanticHash, semanticHash, StringComparison.Ordinal))
|
||||
throw new InvalidOperationException("CultMesh body generation is immutable and was already published with different content or metadata.");
|
||||
}
|
||||
|
||||
private static CultMeshBodyDescriptor CreateDescriptor(CultMeshNetworkBodyDocument document) => new()
|
||||
{
|
||||
BodyId = document.BodyId,
|
||||
SchemaId = document.SchemaId,
|
||||
LayoutVersion = document.LayoutVersion,
|
||||
ByteSize = document.ByteSize,
|
||||
Capacity = document.Capacity,
|
||||
ProducerEpoch = document.ProducerEpoch,
|
||||
Sequence = document.Sequence,
|
||||
AccessMode = CultMeshBodyAccessMode.ReadOnly,
|
||||
Synchronization = document.Synchronization,
|
||||
LeaseExpiresAtUnixMs = document.LeaseExpiresAtUnixMs,
|
||||
TransportKind = CultMeshBodyTransportKind.Network,
|
||||
CapabilityToken = document.CapabilityToken,
|
||||
SemanticHash = document.SemanticHash
|
||||
};
|
||||
}
|
||||
|
||||
public sealed class CultMeshNetworkBodyResolver
|
||||
{
|
||||
private readonly CultCache _cache;
|
||||
|
||||
public CultMeshNetworkBodyResolver(CultCache cache) =>
|
||||
_cache = cache ?? throw new ArgumentNullException(nameof(cache));
|
||||
|
||||
public Func<CultMeshBodyDescriptor, byte[]> CreateFetchDelegate() => Fetch;
|
||||
|
||||
public byte[] Fetch(CultMeshBodyDescriptor descriptor)
|
||||
{
|
||||
if (descriptor == null) throw new ArgumentNullException(nameof(descriptor));
|
||||
if (descriptor.TransportKind != CultMeshBodyTransportKind.Network)
|
||||
throw new NotSupportedException("CultMesh network body resolver requires a network descriptor.");
|
||||
var document = _cache.Get<CultMeshNetworkBodyDocument>(
|
||||
CultMeshNetworkBodyDocument.CreateRecordKey(descriptor.CapabilityToken))
|
||||
?? throw new FileNotFoundException("CultMesh network body capability is missing.", descriptor.CapabilityToken);
|
||||
ValidateDescriptor(document, descriptor);
|
||||
var manifestKey = new CultRecordKey(document.ManifestRecordKey);
|
||||
var manifest = _cache.Get<CultMeshCdnArtifactManifest>(manifestKey)
|
||||
?? throw new FileNotFoundException("CultMesh network body manifest is missing.", manifestKey.Value);
|
||||
if (!string.Equals(CultMeshCdnArtifactManifest.CreateRecordKey(manifest).Value, manifestKey.Value, StringComparison.Ordinal))
|
||||
throw new InvalidDataException("CultMesh network body manifest identity does not match its capability binding.");
|
||||
if (manifest.SizeBytes != document.ByteSize ||
|
||||
!string.Equals(CultMeshCdn.NormalizeHash(manifest.ContentHash, nameof(manifest.ContentHash)), document.SemanticHash, StringComparison.Ordinal))
|
||||
throw new InvalidDataException("CultMesh network body manifest disagrees with its capability binding.");
|
||||
return CultMeshCdn.ReadArtifact(_cache, manifest);
|
||||
}
|
||||
|
||||
private static void ValidateDescriptor(CultMeshNetworkBodyDocument document, CultMeshBodyDescriptor descriptor)
|
||||
{
|
||||
if (!string.Equals(document.CapabilityToken, descriptor.CapabilityToken, StringComparison.Ordinal) ||
|
||||
!string.Equals(document.BodyId, descriptor.BodyId, StringComparison.Ordinal) ||
|
||||
!string.Equals(document.SchemaId, descriptor.SchemaId, StringComparison.Ordinal) ||
|
||||
document.LayoutVersion != descriptor.LayoutVersion || document.ByteSize != descriptor.ByteSize ||
|
||||
document.Capacity != descriptor.Capacity || document.ProducerEpoch != descriptor.ProducerEpoch ||
|
||||
document.Sequence != descriptor.Sequence || document.Synchronization != descriptor.Synchronization ||
|
||||
document.LeaseExpiresAtUnixMs != descriptor.LeaseExpiresAtUnixMs ||
|
||||
!string.Equals(document.SemanticHash, descriptor.SemanticHash, StringComparison.Ordinal))
|
||||
throw new InvalidDataException("CultMesh network body descriptor disagrees with its capability binding.");
|
||||
}
|
||||
}
|
||||
|
||||
[MessagePackObject]
|
||||
[CultDocument("gamecult.mesh.body_publication", CultMeshBodyPublicationSchemaVersions.Publication)]
|
||||
public sealed class CultMeshBodyPublicationDocument
|
||||
|
|
@ -393,7 +194,6 @@ namespace GameCult.Mesh
|
|||
documents ??= CultDocumentRegistry.Shared;
|
||||
return new CultNetDocumentRegistry(documents)
|
||||
.Register(CultNetDocumentBinding.ForDocument<CultMeshBodyPublicationDocument>(documents))
|
||||
.Register(CultNetDocumentBinding.ForDocument<CultMeshNetworkBodyDocument>(documents))
|
||||
.Register(CultNetDocumentBinding.ForDocument<CultMeshCdnArtifactManifest>(documents))
|
||||
.Register(CultNetDocumentBinding.ForDocument<CultMeshCdnArtifactChunk>(documents));
|
||||
}
|
||||
|
|
|
|||
342
src/GameCult.Mesh/CultMeshBodySessions.cs
Normal file
342
src/GameCult.Mesh/CultMeshBodySessions.cs
Normal file
|
|
@ -0,0 +1,342 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using GameCult.Networking;
|
||||
|
||||
#nullable enable
|
||||
|
||||
namespace GameCult.Mesh
|
||||
{
|
||||
/// <summary>
|
||||
/// Owns a bounded set of immutable network body generations. This is ephemeral transport state,
|
||||
/// not a CultCache document store and not world truth.
|
||||
/// </summary>
|
||||
public sealed class CultMeshNetworkBodyStore : IDisposable
|
||||
{
|
||||
public const long DefaultMaximumBodyBytes = 16L * 1024L * 1024L;
|
||||
public const int DefaultRetainedGenerations = 8;
|
||||
|
||||
private readonly object _gate = new();
|
||||
private readonly Dictionary<string, Entry> _entries = new(StringComparer.Ordinal);
|
||||
private readonly byte[] _capabilitySecret = new byte[32];
|
||||
private readonly long _maximumBodyBytes;
|
||||
private readonly int _retainedGenerations;
|
||||
private bool _disposed;
|
||||
|
||||
public CultMeshNetworkBodyStore(
|
||||
long maximumBodyBytes = DefaultMaximumBodyBytes,
|
||||
int retainedGenerations = DefaultRetainedGenerations)
|
||||
{
|
||||
if (maximumBodyBytes <= 0) throw new ArgumentOutOfRangeException(nameof(maximumBodyBytes));
|
||||
if (retainedGenerations <= 0) throw new ArgumentOutOfRangeException(nameof(retainedGenerations));
|
||||
_maximumBodyBytes = maximumBodyBytes;
|
||||
_retainedGenerations = retainedGenerations;
|
||||
using var random = RandomNumberGenerator.Create();
|
||||
random.GetBytes(_capabilitySecret);
|
||||
}
|
||||
|
||||
public CultMeshBodyDescriptor Publish(CultMeshBodyGeneration generation, ReadOnlySpan<byte> body)
|
||||
{
|
||||
return PublishCore(generation, body.ToArray());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Publishes a body buffer by transferring its ownership to the ephemeral store. The caller
|
||||
/// must not mutate the array after this call; this avoids an extra full-frame copy.
|
||||
/// </summary>
|
||||
public CultMeshBodyDescriptor PublishOwned(CultMeshBodyGeneration generation, byte[] body)
|
||||
{
|
||||
if (body == null) throw new ArgumentNullException(nameof(body));
|
||||
return PublishCore(generation, body);
|
||||
}
|
||||
|
||||
private CultMeshBodyDescriptor PublishCore(CultMeshBodyGeneration generation, byte[] bytes)
|
||||
{
|
||||
ValidateGeneration(generation);
|
||||
if (bytes.LongLength > _maximumBodyBytes)
|
||||
throw new InvalidOperationException(
|
||||
$"CultMesh network body exceeds the configured {_maximumBodyBytes}-byte live-body bound.");
|
||||
|
||||
var hash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(bytes);
|
||||
var token = CapabilityToken(generation);
|
||||
var descriptor = new CultMeshBodyDescriptor
|
||||
{
|
||||
BodyId = generation.BodyId,
|
||||
SchemaId = generation.SchemaId,
|
||||
LayoutVersion = generation.LayoutVersion,
|
||||
ByteSize = bytes.LongLength,
|
||||
Capacity = generation.Capacity,
|
||||
ProducerEpoch = generation.ProducerEpoch,
|
||||
Sequence = generation.Sequence,
|
||||
AccessMode = CultMeshBodyAccessMode.ReadOnly,
|
||||
Synchronization = generation.Synchronization,
|
||||
LeaseExpiresAtUnixMs = generation.LeaseExpiresAtUnixMs,
|
||||
TransportKind = CultMeshBodyTransportKind.Network,
|
||||
CapabilityToken = token,
|
||||
SemanticHash = hash
|
||||
};
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
PruneExpired(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());
|
||||
if (_entries.TryGetValue(token, out var existing))
|
||||
{
|
||||
if (!SameDescriptor(existing.Descriptor, descriptor) ||
|
||||
!string.Equals(existing.Descriptor.SemanticHash, hash, StringComparison.Ordinal))
|
||||
throw new InvalidOperationException("CultMesh network body generation is immutable.");
|
||||
Array.Clear(bytes, 0, bytes.Length);
|
||||
return Clone(existing.Descriptor);
|
||||
}
|
||||
|
||||
_entries.Add(token, new Entry(Clone(descriptor), bytes));
|
||||
TrimOldest();
|
||||
return descriptor;
|
||||
}
|
||||
}
|
||||
|
||||
public bool TryRead(CultMeshBodyReadRequestMessage request, DateTimeOffset nowUtc, out CultMeshBodyDescriptor descriptor, out byte[] body)
|
||||
{
|
||||
if (request == null) throw new ArgumentNullException(nameof(request));
|
||||
lock (_gate)
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
PruneExpired(nowUtc.ToUnixTimeMilliseconds());
|
||||
if (!_entries.TryGetValue(request.CapabilityToken ?? string.Empty, out var entry) ||
|
||||
!MatchesRequest(entry.Descriptor, request))
|
||||
{
|
||||
descriptor = new CultMeshBodyDescriptor();
|
||||
body = Array.Empty<byte>();
|
||||
return false;
|
||||
}
|
||||
descriptor = Clone(entry.Descriptor);
|
||||
body = (byte[])entry.Body.Clone();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
foreach (var entry in _entries.Values) Array.Clear(entry.Body, 0, entry.Body.Length);
|
||||
_entries.Clear();
|
||||
Array.Clear(_capabilitySecret, 0, _capabilitySecret.Length);
|
||||
}
|
||||
}
|
||||
|
||||
private string CapabilityToken(CultMeshBodyGeneration generation)
|
||||
{
|
||||
var input = Encoding.UTF8.GetBytes(string.Join("\u001f", generation.ProducerId, generation.BodyId,
|
||||
generation.ProducerEpoch, generation.Sequence));
|
||||
using var hmac = new HMACSHA256(_capabilitySecret);
|
||||
return string.Concat(hmac.ComputeHash(input).Select(value => value.ToString("x2")));
|
||||
}
|
||||
|
||||
private void PruneExpired(long nowUnixMs)
|
||||
{
|
||||
foreach (var key in _entries.Where(pair => pair.Value.Descriptor.LeaseExpiresAtUnixMs <= nowUnixMs)
|
||||
.Select(pair => pair.Key).ToArray())
|
||||
Remove(key);
|
||||
}
|
||||
|
||||
private void TrimOldest()
|
||||
{
|
||||
while (_entries.Count > _retainedGenerations)
|
||||
{
|
||||
var oldest = _entries.OrderBy(pair => pair.Value.Descriptor.ProducerEpoch)
|
||||
.ThenBy(pair => pair.Value.Descriptor.Sequence)
|
||||
.First().Key;
|
||||
Remove(oldest);
|
||||
}
|
||||
}
|
||||
|
||||
private void Remove(string token)
|
||||
{
|
||||
if (!_entries.Remove(token, out var removed)) return;
|
||||
Array.Clear(removed.Body, 0, removed.Body.Length);
|
||||
}
|
||||
|
||||
private static bool MatchesRequest(CultMeshBodyDescriptor descriptor, CultMeshBodyReadRequestMessage request) =>
|
||||
string.Equals(descriptor.CapabilityToken, request.CapabilityToken, StringComparison.Ordinal) &&
|
||||
string.Equals(descriptor.BodyId, request.BodyId, StringComparison.Ordinal) &&
|
||||
string.Equals(descriptor.SchemaId, request.BodySchemaId, StringComparison.Ordinal) &&
|
||||
descriptor.LayoutVersion == request.LayoutVersion &&
|
||||
descriptor.ProducerEpoch == request.ProducerEpoch && descriptor.Sequence == request.Sequence &&
|
||||
descriptor.ByteSize == request.ExpectedSizeBytes &&
|
||||
string.Equals(descriptor.SemanticHash, request.SemanticHash, StringComparison.Ordinal);
|
||||
|
||||
private static bool SameDescriptor(CultMeshBodyDescriptor left, CultMeshBodyDescriptor right) =>
|
||||
left.BodyId == right.BodyId && left.SchemaId == right.SchemaId &&
|
||||
left.LayoutVersion == right.LayoutVersion && left.ByteSize == right.ByteSize &&
|
||||
left.Capacity == right.Capacity && left.ProducerEpoch == right.ProducerEpoch &&
|
||||
left.Sequence == right.Sequence && left.Synchronization == right.Synchronization &&
|
||||
left.LeaseExpiresAtUnixMs == right.LeaseExpiresAtUnixMs;
|
||||
|
||||
private static CultMeshBodyDescriptor Clone(CultMeshBodyDescriptor value) => new()
|
||||
{
|
||||
BodyId = value.BodyId, SchemaId = value.SchemaId, LayoutVersion = value.LayoutVersion,
|
||||
ByteSize = value.ByteSize, Capacity = value.Capacity, ProducerEpoch = value.ProducerEpoch,
|
||||
Sequence = value.Sequence, AccessMode = value.AccessMode, Synchronization = value.Synchronization,
|
||||
LeaseExpiresAtUnixMs = value.LeaseExpiresAtUnixMs, TransportKind = value.TransportKind,
|
||||
CapabilityToken = value.CapabilityToken, SemanticHash = value.SemanticHash
|
||||
};
|
||||
|
||||
private static void ValidateGeneration(CultMeshBodyGeneration generation)
|
||||
{
|
||||
if (generation == null) throw new ArgumentNullException(nameof(generation));
|
||||
if (string.IsNullOrWhiteSpace(generation.BodyId) || string.IsNullOrWhiteSpace(generation.ProducerId) ||
|
||||
string.IsNullOrWhiteSpace(generation.SchemaId))
|
||||
throw new ArgumentException("Body, producer, and schema identities are required.", nameof(generation));
|
||||
if (generation.LayoutVersion < 0 || generation.Capacity < 0 || generation.ProducerEpoch < 0 ||
|
||||
generation.Sequence < 0 || generation.LeaseExpiresAtUnixMs <= 0)
|
||||
throw new ArgumentOutOfRangeException(nameof(generation));
|
||||
}
|
||||
|
||||
private void ThrowIfDisposed()
|
||||
{
|
||||
if (_disposed) throw new ObjectDisposedException(nameof(CultMeshNetworkBodyStore));
|
||||
}
|
||||
|
||||
private sealed class Entry
|
||||
{
|
||||
public Entry(CultMeshBodyDescriptor descriptor, byte[] body) { Descriptor = descriptor; Body = body; }
|
||||
public CultMeshBodyDescriptor Descriptor { get; }
|
||||
public byte[] Body { get; }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Serves capability-bound live bodies directly over an existing CultNet host.</summary>
|
||||
public sealed class CultMeshBodyServer : IDisposable
|
||||
{
|
||||
private readonly ICultNetSchemaServer _server;
|
||||
private readonly CultMeshNetworkBodyStore _bodies;
|
||||
private readonly Func<CultMeshBodyReadRequestMessage, ICultNetSchemaServerPeer, Task> _handler;
|
||||
private bool _disposed;
|
||||
|
||||
public CultMeshBodyServer(ICultNetSchemaServer server, CultMeshNetworkBodyStore bodies)
|
||||
{
|
||||
_server = server ?? throw new ArgumentNullException(nameof(server));
|
||||
_bodies = bodies ?? throw new ArgumentNullException(nameof(bodies));
|
||||
_handler = HandleAsync;
|
||||
_server.OnCultNet(_handler);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
_server.RemoveCultNetMessageListener<CultMeshBodyReadRequestMessage>(_handler);
|
||||
}
|
||||
|
||||
private Task HandleAsync(CultMeshBodyReadRequestMessage request, ICultNetSchemaServerPeer peer)
|
||||
{
|
||||
var response = new CultMeshBodyReadResponseMessage
|
||||
{
|
||||
MessageId = request?.MessageId ?? string.Empty,
|
||||
CapabilityToken = request?.CapabilityToken ?? string.Empty
|
||||
};
|
||||
try
|
||||
{
|
||||
if (request == null || string.IsNullOrWhiteSpace(request.MessageId))
|
||||
throw new InvalidDataException("Body read request requires a message identity.");
|
||||
if (!_bodies.TryRead(request, DateTimeOffset.UtcNow, out var descriptor, out var bytes))
|
||||
throw new FileNotFoundException("Body capability is unavailable or does not match its generation.");
|
||||
response.Found = true;
|
||||
response.BodyId = descriptor.BodyId;
|
||||
response.ProducerEpoch = descriptor.ProducerEpoch;
|
||||
response.Sequence = descriptor.Sequence;
|
||||
response.SizeBytes = descriptor.ByteSize;
|
||||
response.SemanticHash = descriptor.SemanticHash;
|
||||
response.Payload = bytes;
|
||||
}
|
||||
catch (Exception error)
|
||||
{
|
||||
response.Found = false;
|
||||
response.Error = error.GetType().Name + ": " + error.Message;
|
||||
response.Payload = Array.Empty<byte>();
|
||||
}
|
||||
peer.SendCultNet(response);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
public sealed class CultMeshSessionBodyProviderOptions
|
||||
{
|
||||
public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance;
|
||||
public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(2);
|
||||
}
|
||||
|
||||
/// <summary>Reads live body generations through one reusable identity-first body session.</summary>
|
||||
public sealed class CultMeshSessionBodyProvider
|
||||
{
|
||||
private readonly CultMeshSessionManager _sessions;
|
||||
private readonly CultMeshEndpointId _endpointId;
|
||||
private readonly CultMeshSessionBodyProviderOptions _options;
|
||||
|
||||
public CultMeshSessionBodyProvider(string providerId, CultMeshSessionManager sessions,
|
||||
CultMeshEndpointId endpointId, CultMeshSessionBodyProviderOptions? options = null)
|
||||
{
|
||||
ProviderId = string.IsNullOrWhiteSpace(providerId) ? throw new ArgumentException("Provider identity is required.", nameof(providerId)) : providerId;
|
||||
_sessions = sessions ?? throw new ArgumentNullException(nameof(sessions));
|
||||
_endpointId = endpointId ?? throw new ArgumentNullException(nameof(endpointId));
|
||||
_options = options ?? new CultMeshSessionBodyProviderOptions();
|
||||
if (_options.ResponseTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(options));
|
||||
}
|
||||
|
||||
public string ProviderId { get; }
|
||||
public Func<CultMeshBodyDescriptor, byte[]> CreateFetchDelegate() =>
|
||||
descriptor => FetchAsync(descriptor).GetAwaiter().GetResult();
|
||||
|
||||
public async Task<byte[]> FetchAsync(CultMeshBodyDescriptor descriptor, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (descriptor == null) throw new ArgumentNullException(nameof(descriptor));
|
||||
if (descriptor.TransportKind != CultMeshBodyTransportKind.Network)
|
||||
throw new NotSupportedException("Direct body provider requires a network descriptor.");
|
||||
var session = await _sessions.ConnectAsync(_endpointId, CultMeshProtocols.Bodies, cancellationToken).ConfigureAwait(false);
|
||||
var messageId = Guid.NewGuid().ToString("N");
|
||||
var completion = new TaskCompletionSource<CultMeshBodyReadResponseMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using var responseSubscription = session.OnCultNet<CultMeshBodyReadResponseMessage>(response =>
|
||||
{
|
||||
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) completion.TrySetResult(response);
|
||||
});
|
||||
using var errorSubscription = session.OnCultNet<CultNetErrorMessage>(error => completion.TrySetException(new IOException(error.Error)));
|
||||
session.SendCultNet(new CultMeshBodyReadRequestMessage
|
||||
{
|
||||
MessageId = messageId, CapabilityToken = descriptor.CapabilityToken, BodyId = descriptor.BodyId,
|
||||
BodySchemaId = descriptor.SchemaId, LayoutVersion = descriptor.LayoutVersion,
|
||||
ProducerEpoch = descriptor.ProducerEpoch, Sequence = descriptor.Sequence,
|
||||
ExpectedSizeBytes = descriptor.ByteSize, SemanticHash = descriptor.SemanticHash
|
||||
});
|
||||
var response = await WaitAsync(completion.Task, cancellationToken).ConfigureAwait(false);
|
||||
if (!response.Found) throw new FileNotFoundException(
|
||||
$"Body provider '{ProviderId}' rejected '{descriptor.BodyId}' generation {descriptor.Sequence}: {response.Error}");
|
||||
var body = response.Payload ?? Array.Empty<byte>();
|
||||
if (!string.Equals(response.CapabilityToken, descriptor.CapabilityToken, StringComparison.Ordinal) ||
|
||||
!string.Equals(response.BodyId, descriptor.BodyId, StringComparison.Ordinal) ||
|
||||
response.ProducerEpoch != descriptor.ProducerEpoch || response.Sequence != descriptor.Sequence ||
|
||||
response.SizeBytes != descriptor.ByteSize || body.LongLength != descriptor.ByteSize ||
|
||||
!string.Equals(response.SemanticHash, descriptor.SemanticHash, StringComparison.Ordinal) ||
|
||||
!string.Equals(CultMeshBodyDescriptorValidator.ComputeSemanticHash(body), descriptor.SemanticHash, StringComparison.Ordinal))
|
||||
throw new InvalidDataException("Direct body response disagrees with its advertised generation.");
|
||||
return body;
|
||||
}
|
||||
|
||||
private async Task<CultMeshBodyReadResponseMessage> WaitAsync(Task<CultMeshBodyReadResponseMessage> response, CancellationToken cancellationToken)
|
||||
{
|
||||
using var deadlineCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
var deadline = _options.Clock.DelayAsync(_options.ResponseTimeout, deadlineCancellation.Token);
|
||||
var completed = await Task.WhenAny(response, deadline).ConfigureAwait(false);
|
||||
if (completed == response) { deadlineCancellation.Cancel(); return await response.ConfigureAwait(false); }
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
throw new TimeoutException($"Timed out fetching live body from provider '{ProviderId}'.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -108,6 +108,20 @@ namespace GameCult.Mesh
|
|||
options);
|
||||
}
|
||||
|
||||
/// <summary>Creates a direct body provider over this client's reusable negotiated session owner.</summary>
|
||||
public CultMeshSessionBodyProvider BodyProvider(
|
||||
string providerId,
|
||||
string endpointId,
|
||||
CultMeshSessionBodyProviderOptions? options = null)
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
return new CultMeshSessionBodyProvider(
|
||||
providerId,
|
||||
_sessions,
|
||||
CultMeshEndpointId.Parse(endpointId),
|
||||
options);
|
||||
}
|
||||
|
||||
/// <summary>Opens one live typed document by stable provider identity and record key.</summary>
|
||||
public async Task<CultMeshDocumentHandle<TDocument>> DocumentAsync<TDocument>(
|
||||
string endpointId,
|
||||
|
|
|
|||
|
|
@ -35,6 +35,7 @@ namespace GameCult.Mesh
|
|||
{
|
||||
public static CultMeshProtocolId Documents { get; } = new CultMeshProtocolId("cultmesh.documents.v1");
|
||||
public static CultMeshProtocolId Content { get; } = new CultMeshProtocolId("cultmesh.content.v1");
|
||||
public static CultMeshProtocolId Bodies { get; } = new CultMeshProtocolId("cultmesh.bodies.v1");
|
||||
public static CultMeshProtocolId Discovery { get; } = new CultMeshProtocolId("cultmesh.discovery.v1");
|
||||
public static CultMeshProtocolId PeerExchange { get; } = new CultMeshProtocolId("cultmesh.peer_exchange.v1");
|
||||
public static CultMeshProtocolId Subscriptions { get; } = new CultMeshProtocolId("cultmesh.subscriptions.v1");
|
||||
|
|
|
|||
|
|
@ -525,7 +525,11 @@ and own their unsubscribe lifetime. Hot body state uses
|
|||
`CultMesh.SubscribeLiveBodyAsync(...)`: the reactive value is only the current
|
||||
body descriptor, while `OpenCurrentReadOnly()` negotiates shared memory,
|
||||
shared-file mapping, or network from observed locality and available adapters.
|
||||
Body bytes never become subscription snapshot payload. Either subscription may
|
||||
Same-machine bodies are opened from the mapped producer slab. Remote live bodies
|
||||
are capability-bound and read through one reusable `cultmesh.bodies.v1` session;
|
||||
they are not packed into CultCache or CDN records. `CultMeshClient.BodyProvider(...)`
|
||||
creates that verified network adapter without exposing session machinery to the
|
||||
consumer. Body bytes never become subscription snapshot payload. Either subscription may
|
||||
legitimately start with `HasValue == false`; it remains active until the provider
|
||||
publishes the exact record, avoiding a startup retry loop or broad snapshot.
|
||||
`CultNetDatabaseSubscriptionServer.DemandChanged` exposes the exact requested
|
||||
|
|
|
|||
|
|
@ -76,6 +76,8 @@ namespace GameCult.Networking
|
|||
CultNetSchemaVersions.PeerExchangeResponse => Deserialize<CultMeshPeerExchangeResponseMessage>(payload),
|
||||
CultNetSchemaVersions.ContentChunkRequest => Deserialize<CultMeshContentChunkRequestMessage>(payload),
|
||||
CultNetSchemaVersions.ContentChunkResponse => Deserialize<CultMeshContentChunkResponseMessage>(payload),
|
||||
CultNetSchemaVersions.BodyReadRequest => Deserialize<CultMeshBodyReadRequestMessage>(payload),
|
||||
CultNetSchemaVersions.BodyReadResponse => Deserialize<CultMeshBodyReadResponseMessage>(payload),
|
||||
CultNetSchemaVersions.SchemaCatalogRequest => Deserialize<CultNetSchemaCatalogRequestMessage>(payload),
|
||||
CultNetSchemaVersions.SchemaCatalogResponse => Deserialize<CultNetSchemaCatalogResponseMessage>(payload),
|
||||
_ => throw new MessagePackSerializationException(
|
||||
|
|
|
|||
|
|
@ -151,6 +151,14 @@ namespace GameCult.Networking
|
|||
/// CultMesh content chunk response contract identifier.
|
||||
/// </summary>
|
||||
public const string ContentChunkResponse = "cultmesh.content_chunk_response.v1";
|
||||
/// <summary>
|
||||
/// CultMesh ephemeral body read request contract identifier.
|
||||
/// </summary>
|
||||
public const string BodyReadRequest = "cultmesh.body_read_request.v1";
|
||||
/// <summary>
|
||||
/// CultMesh ephemeral body read response contract identifier.
|
||||
/// </summary>
|
||||
public const string BodyReadResponse = "cultmesh.body_read_response.v1";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -1363,6 +1371,42 @@ namespace GameCult.Networking
|
|||
[Key("error")] public string Error { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Requests one immutable body generation from its capability owner. Body bytes travel on this
|
||||
/// dedicated plane; they are never database snapshot records.
|
||||
/// </summary>
|
||||
[MessagePackObject]
|
||||
public class CultMeshBodyReadRequestMessage : ICultNetSchemaMessage
|
||||
{
|
||||
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.BodyReadRequest;
|
||||
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
|
||||
[Key("capabilityToken")] public string CapabilityToken { get; set; } = string.Empty;
|
||||
[Key("bodyId")] public string BodyId { get; set; } = string.Empty;
|
||||
[Key("bodySchemaId")] public string BodySchemaId { get; set; } = string.Empty;
|
||||
[Key("layoutVersion")] public int LayoutVersion { get; set; }
|
||||
[Key("producerEpoch")] public long ProducerEpoch { get; set; }
|
||||
[Key("sequence")] public long Sequence { get; set; }
|
||||
[Key("expectedSizeBytes")] public long ExpectedSizeBytes { get; set; }
|
||||
[Key("semanticHash")] public string SemanticHash { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
/// <summary>Returns one bounded immutable body generation directly to its subscriber.</summary>
|
||||
[MessagePackObject]
|
||||
public class CultMeshBodyReadResponseMessage : ICultNetSchemaMessage
|
||||
{
|
||||
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.BodyReadResponse;
|
||||
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
|
||||
[Key("found")] public bool Found { get; set; }
|
||||
[Key("capabilityToken")] public string CapabilityToken { get; set; } = string.Empty;
|
||||
[Key("bodyId")] public string BodyId { get; set; } = string.Empty;
|
||||
[Key("producerEpoch")] public long ProducerEpoch { get; set; }
|
||||
[Key("sequence")] public long Sequence { get; set; }
|
||||
[Key("sizeBytes")] public long SizeBytes { get; set; }
|
||||
[Key("semanticHash")] public string SemanticHash { get; set; } = string.Empty;
|
||||
[Key("payload")] public byte[] Payload { get; set; } = Array.Empty<byte>();
|
||||
[Key("error")] public string Error { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Wire representation of a CultMesh peer candidate.
|
||||
/// </summary>
|
||||
|
|
|
|||
218
tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs
Normal file
218
tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs
Normal file
|
|
@ -0,0 +1,218 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Diagnostics;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using FluentAssertions;
|
||||
using GameCult.Networking;
|
||||
using NUnit.Framework;
|
||||
|
||||
namespace GameCult.Mesh.Tests;
|
||||
|
||||
[TestFixture]
|
||||
public sealed class CultMeshBodySessionTests
|
||||
{
|
||||
[Test]
|
||||
public async Task LiveBodySessionTransfersVerifiedGenerationWithoutSnapshotRecords()
|
||||
{
|
||||
var bytes = new byte[384_000];
|
||||
for (var index = 0; index < bytes.Length; index++) bytes[index] = (byte)(index % 251);
|
||||
using var store = new CultMeshNetworkBodyStore();
|
||||
var descriptor = store.Publish(Generation(sequence: 7), bytes);
|
||||
var server = new LoopbackServer();
|
||||
using var bodyServer = new CultMeshBodyServer(server, store);
|
||||
var client = new LoopbackClient(server);
|
||||
var connector = new LoopbackConnector(client);
|
||||
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource() });
|
||||
using var sessions = new CultMeshSessionManager(discovery, new[] { connector });
|
||||
var provider = new CultMeshSessionBodyProvider(
|
||||
"aetheria.daemon", sessions, CultMeshEndpointId.Parse("service:aetheria.daemon"));
|
||||
|
||||
var received = await provider.FetchAsync(descriptor);
|
||||
|
||||
received.Should().Equal(bytes);
|
||||
client.BodyRequestCount.Should().Be(1);
|
||||
client.SnapshotRequestCount.Should().Be(0,
|
||||
"live body bytes must never enter database snapshot machinery");
|
||||
connector.ConnectCount.Should().Be(1);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task RudpBodySessionTransfersOneHotGenerationWithoutDatabaseRoundTrip()
|
||||
{
|
||||
var bytes = new byte[384_000];
|
||||
for (var index = 0; index < bytes.Length; index++) bytes[index] = (byte)(index % 251);
|
||||
using var store = new CultMeshNetworkBodyStore();
|
||||
var descriptor = store.Publish(Generation(sequence: 11), bytes);
|
||||
using var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
|
||||
socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
|
||||
{
|
||||
RuntimeId = "cultmesh-body-rudp-test", Socket = socket, MaxFragmentBytes = 1024,
|
||||
MaxPendingReliablePackets = 8192
|
||||
});
|
||||
using var bodyServer = new CultMeshBodyServer(server, store);
|
||||
using var pumpCancellation = new CancellationTokenSource();
|
||||
var pump = Task.Run(async () =>
|
||||
{
|
||||
while (!pumpCancellation.IsCancellationRequested)
|
||||
{
|
||||
var progress = await server.PollAvailableAsync(256);
|
||||
server.PollResends();
|
||||
if (progress.TransportItemsConsumed == 0) await Task.Delay(1, pumpCancellation.Token);
|
||||
}
|
||||
});
|
||||
var endpoint = $"rudp://127.0.0.1:{server.LocalEndPoint.Port}";
|
||||
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) });
|
||||
using var sessions = new CultMeshSessionManager(discovery,
|
||||
new ICultMeshTransportConnector[] { new CultMeshSchemaTransportConnector() });
|
||||
var provider = new CultMeshSessionBodyProvider("aetheria.daemon", sessions,
|
||||
CultMeshEndpointId.Parse("service:aetheria.daemon"),
|
||||
new CultMeshSessionBodyProviderOptions { ResponseTimeout = TimeSpan.FromSeconds(10) });
|
||||
|
||||
var elapsed = Stopwatch.StartNew();
|
||||
byte[] received;
|
||||
try { received = await provider.FetchAsync(descriptor); }
|
||||
finally
|
||||
{
|
||||
pumpCancellation.Cancel();
|
||||
try { await pump; } catch (OperationCanceledException) { }
|
||||
}
|
||||
elapsed.Stop();
|
||||
|
||||
received.Should().Equal(bytes);
|
||||
elapsed.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(5));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void NetworkBodyStoreRejectsGenerationMetadataThatDoesNotMatchCapability()
|
||||
{
|
||||
using var store = new CultMeshNetworkBodyStore();
|
||||
var descriptor = store.Publish(Generation(sequence: 3), new byte[] { 1, 2, 3 });
|
||||
var request = Request(descriptor);
|
||||
request.Sequence++;
|
||||
|
||||
store.TryRead(request, DateTimeOffset.UtcNow, out _, out _).Should().BeFalse();
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void BodyWireMessagesRoundTripThroughSchemaDispatcher()
|
||||
{
|
||||
using var store = new CultMeshNetworkBodyStore();
|
||||
var descriptor = store.Publish(Generation(sequence: 9), new byte[] { 1, 2, 3 });
|
||||
var request = Request(descriptor);
|
||||
var response = new CultMeshBodyReadResponseMessage
|
||||
{
|
||||
MessageId = request.MessageId, Found = true, CapabilityToken = descriptor.CapabilityToken,
|
||||
BodyId = descriptor.BodyId, ProducerEpoch = descriptor.ProducerEpoch,
|
||||
Sequence = descriptor.Sequence, SizeBytes = 3, SemanticHash = descriptor.SemanticHash,
|
||||
Payload = new byte[] { 1, 2, 3 }
|
||||
};
|
||||
|
||||
CultNetSchemaMessageSerialization.Deserialize(CultNetSchemaMessageSerialization.Serialize(request))
|
||||
.Should().BeEquivalentTo(request);
|
||||
CultNetSchemaMessageSerialization.Deserialize(CultNetSchemaMessageSerialization.Serialize(response))
|
||||
.Should().BeEquivalentTo(response);
|
||||
}
|
||||
|
||||
private static CultMeshBodyGeneration Generation(long sequence) => new()
|
||||
{
|
||||
BodyId = "eve:entity-soa:aetheria.daemon:pilot", ProducerId = "aetheria.daemon",
|
||||
SchemaId = "gamecult.eve.entity_soa.body.v2", LayoutVersion = 3, Capacity = 4096,
|
||||
ProducerEpoch = 2, Sequence = sequence, Synchronization = CultMeshBodySynchronization.TripleBuffer,
|
||||
LeaseExpiresAtUnixMs = DateTimeOffset.UtcNow.AddMinutes(1).ToUnixTimeMilliseconds()
|
||||
};
|
||||
|
||||
private static CultMeshBodyReadRequestMessage Request(CultMeshBodyDescriptor descriptor) => new()
|
||||
{
|
||||
MessageId = Guid.NewGuid().ToString("N"), CapabilityToken = descriptor.CapabilityToken,
|
||||
BodyId = descriptor.BodyId, BodySchemaId = descriptor.SchemaId, LayoutVersion = descriptor.LayoutVersion,
|
||||
ProducerEpoch = descriptor.ProducerEpoch, Sequence = descriptor.Sequence,
|
||||
ExpectedSizeBytes = descriptor.ByteSize, SemanticHash = descriptor.SemanticHash
|
||||
};
|
||||
|
||||
private sealed class RouteSource : ICultMeshLookupSource
|
||||
{
|
||||
private readonly string _endpoint;
|
||||
public RouteSource(string endpoint = "rudp://body.test:3076") => _endpoint = endpoint;
|
||||
public string SourceId => "test";
|
||||
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"),
|
||||
new[] { _endpoint }), SourceId, now, now.AddMinutes(1), CultMeshDiscoveryTrust.Signed)
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class LoopbackConnector : ICultMeshTransportConnector
|
||||
{
|
||||
private readonly ICultNetSchemaClient _client;
|
||||
public LoopbackConnector(ICultNetSchemaClient client) => _client = client;
|
||||
public string ConnectorId => "loopback";
|
||||
public int ConnectCount { get; private set; }
|
||||
public bool CanConnect(CultMeshTransportCandidate candidate) => true;
|
||||
public Task<ICultNetSchemaClient> ConnectAsync(CultMeshTransportCandidate candidate, CultMeshProtocolId protocol,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
protocol.Should().Be(CultMeshProtocols.Bodies);
|
||||
ConnectCount++;
|
||||
return Task.FromResult(_client);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class LoopbackServer : ICultNetSchemaServer
|
||||
{
|
||||
private readonly Dictionary<Type, Delegate> _handlers = new();
|
||||
public void OnCultNet<T>(Func<T, ICultNetSchemaServerPeer, Task> callback) where T : ICultNetSchemaMessage =>
|
||||
_handlers[typeof(T)] = callback;
|
||||
public void RemoveCultNetMessageListener<T>(Delegate callback) where T : ICultNetSchemaMessage =>
|
||||
_handlers.Remove(typeof(T));
|
||||
public Task Dispatch<T>(T message, ICultNetSchemaServerPeer peer) where T : ICultNetSchemaMessage =>
|
||||
((Func<T, ICultNetSchemaServerPeer, Task>)_handlers[typeof(T)])(message, peer);
|
||||
}
|
||||
|
||||
private sealed class LoopbackClient : ICultNetSchemaClient
|
||||
{
|
||||
private readonly LoopbackServer _server;
|
||||
private readonly Dictionary<Type, List<Delegate>> _handlers = new();
|
||||
public LoopbackClient(LoopbackServer server) => _server = server;
|
||||
public bool Connected => true;
|
||||
public int BodyRequestCount { 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 CultMeshBodyReadRequestMessage body)
|
||||
{
|
||||
BodyRequestCount++;
|
||||
_server.Dispatch(body, new LoopbackPeer(this)).GetAwaiter().GetResult();
|
||||
}
|
||||
else if (message is CultNetSnapshotRequestMessage) SnapshotRequestCount++;
|
||||
}
|
||||
public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
|
||||
{
|
||||
if (!_handlers.TryGetValue(typeof(T), out var values)) _handlers[typeof(T)] = values = new List<Delegate>();
|
||||
values.Add(callback);
|
||||
}
|
||||
public void Emit<T>(T message) where T : ICultNetSchemaMessage
|
||||
{
|
||||
if (_handlers.TryGetValue(typeof(T), out var values))
|
||||
foreach (var value in values.ToArray()) ((Action<T>)value)(message);
|
||||
}
|
||||
public void Dispose() { }
|
||||
}
|
||||
|
||||
private sealed class LoopbackPeer : ICultNetSchemaServerPeer
|
||||
{
|
||||
private readonly LoopbackClient _client;
|
||||
public LoopbackPeer(LoopbackClient client) => _client = client;
|
||||
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage => _client.Emit(message);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,158 +0,0 @@
|
|||
using System;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using FluentAssertions;
|
||||
using GameCult.Caching;
|
||||
using NUnit.Framework;
|
||||
|
||||
namespace GameCult.Mesh.Tests;
|
||||
|
||||
public sealed class CultMeshNetworkBodyPublicationTests
|
||||
{
|
||||
private CultCache _cache = null!;
|
||||
private string _mappedRoot = null!;
|
||||
private DateTimeOffset _now;
|
||||
|
||||
[SetUp]
|
||||
public void SetUp()
|
||||
{
|
||||
_cache = new CultCache();
|
||||
_mappedRoot = Path.Combine(Path.GetTempPath(), "cultmesh-network-body-tests", Guid.NewGuid().ToString("N"));
|
||||
_now = DateTimeOffset.UtcNow;
|
||||
}
|
||||
|
||||
[TearDown]
|
||||
public void TearDown()
|
||||
{
|
||||
_cache.Dispose();
|
||||
if (Directory.Exists(_mappedRoot)) Directory.Delete(_mappedRoot, true);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task PublishAndFetch_RoundTripsThroughTypedCdnDocuments()
|
||||
{
|
||||
var bytes = Enumerable.Range(0, 41).Select(value => (byte)value).ToArray();
|
||||
var descriptor = await Publisher(chunkSizeBytes: 8).PublishAsync(Generation(), bytes);
|
||||
var resolver = new CultMeshNetworkBodyResolver(_cache);
|
||||
|
||||
var fetched = resolver.Fetch(descriptor);
|
||||
|
||||
descriptor.TransportKind.Should().Be(CultMeshBodyTransportKind.Network);
|
||||
descriptor.CapabilityToken.Should().NotBeNullOrWhiteSpace();
|
||||
fetched.Should().Equal(bytes);
|
||||
_cache.Get<CultMeshNetworkBodyDocument>(CultMeshNetworkBodyDocument.CreateRecordKey(descriptor.CapabilityToken))
|
||||
.Should().NotBeNull();
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task LocalAndNetworkDescriptors_AreAcceptedAsOneLogicalGeneration()
|
||||
{
|
||||
var bytes = BitConverter.GetBytes(42.25f);
|
||||
var generation = Generation();
|
||||
var local = new CultMeshMappedBodyPublisher(_mappedRoot, TimeSpan.FromMinutes(1)).Publish(
|
||||
generation.BodyId, generation.SchemaId, generation.LayoutVersion, generation.Capacity,
|
||||
generation.ProducerEpoch, generation.Sequence, bytes, _now);
|
||||
generation.LeaseExpiresAtUnixMs = local.LeaseExpiresAtUnixMs;
|
||||
var network = await Publisher().PublishAsync(generation, bytes);
|
||||
var service = new CultMeshBodyTransportService(
|
||||
new ICultMeshBodyTransportAdapter[]
|
||||
{
|
||||
new CultMeshMappedBodyAdapter(_mappedRoot),
|
||||
new CultMeshNetworkBodyAdapter(new CultMeshNetworkBodyResolver(_cache).CreateFetchDelegate())
|
||||
},
|
||||
_ => true);
|
||||
|
||||
using var lease = service.NegotiateReadOnly(local, network, Request(network)).Lease;
|
||||
|
||||
lease.TransportKind.Should().Be(CultMeshBodyTransportKind.SharedFileMapping);
|
||||
lease.ReadSingle(0).Should().Be(42.25f);
|
||||
}
|
||||
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public async Task Fetch_RejectsMissingOrCorruptChunk(bool corrupt)
|
||||
{
|
||||
var descriptor = await Publisher(chunkSizeBytes: 2).PublishAsync(Generation(), new byte[] { 1, 2, 3, 4 });
|
||||
var binding = _cache.Get<CultMeshNetworkBodyDocument>(CultMeshNetworkBodyDocument.CreateRecordKey(descriptor.CapabilityToken))!;
|
||||
var manifest = _cache.Get<CultMeshCdnArtifactManifest>(new CultRecordKey(binding.ManifestRecordKey))!;
|
||||
var chunkKey = new CultRecordKey(manifest.Chunks[0].RecordKey);
|
||||
if (corrupt)
|
||||
{
|
||||
var chunk = _cache.Get<CultMeshCdnArtifactChunk>(chunkKey)!;
|
||||
chunk.Payload[0] ^= 0xff;
|
||||
await _cache.UpsertAsync(chunk, new CultRecordHandle<CultMeshCdnArtifactChunk>(chunkKey));
|
||||
}
|
||||
else
|
||||
{
|
||||
_cache.Remove(chunkKey);
|
||||
}
|
||||
|
||||
Action fetch = () => new CultMeshNetworkBodyResolver(_cache).Fetch(descriptor);
|
||||
|
||||
if (corrupt) fetch.Should().Throw<InvalidDataException>();
|
||||
else fetch.Should().Throw<FileNotFoundException>();
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task Fetch_RejectsStaleDescriptorGeneration()
|
||||
{
|
||||
var descriptor = await Publisher().PublishAsync(Generation(), new byte[] { 1, 2, 3, 4 });
|
||||
descriptor.Sequence--;
|
||||
|
||||
Action fetch = () => new CultMeshNetworkBodyResolver(_cache).Fetch(descriptor);
|
||||
|
||||
fetch.Should().Throw<InvalidDataException>().WithMessage("*capability binding*");
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task Publish_RejectsUnauthorizedProducerBeforeWritingDocuments()
|
||||
{
|
||||
var generation = Generation();
|
||||
var publisher = new CultMeshNetworkBodyPublisher(_cache, _ => false);
|
||||
|
||||
Func<Task> publish = () => publisher.PublishAsync(generation, new byte[] { 1, 2, 3, 4 });
|
||||
|
||||
await publish.Should().ThrowAsync<UnauthorizedAccessException>();
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task Publish_RejectsMutationOfAnExistingGeneration()
|
||||
{
|
||||
var publisher = Publisher();
|
||||
var generation = Generation();
|
||||
await publisher.PublishAsync(generation, new byte[] { 1, 2, 3, 4 });
|
||||
|
||||
Func<Task> republish = () => publisher.PublishAsync(generation, new byte[] { 1, 2, 3, 5 });
|
||||
|
||||
await republish.Should().ThrowAsync<InvalidOperationException>().WithMessage("*immutable*");
|
||||
}
|
||||
|
||||
private CultMeshNetworkBodyPublisher Publisher(int chunkSizeBytes = CultMeshCdnPackOptions.DefaultChunkSizeBytes) =>
|
||||
new(_cache, generation => generation.ProducerId == "aetheria", chunkSizeBytes);
|
||||
|
||||
private CultMeshBodyGeneration Generation() => new()
|
||||
{
|
||||
BodyId = "aetheria:entities",
|
||||
ProducerId = "aetheria",
|
||||
SchemaId = "eve.entity_soa.v1",
|
||||
LayoutVersion = 2,
|
||||
Capacity = 128,
|
||||
ProducerEpoch = 7,
|
||||
Sequence = 42,
|
||||
Synchronization = CultMeshBodySynchronization.ImmutableSequence,
|
||||
LeaseExpiresAtUnixMs = _now.AddMinutes(1).ToUnixTimeMilliseconds()
|
||||
};
|
||||
|
||||
private CultMeshBodyValidationRequest Request(CultMeshBodyDescriptor descriptor) => new()
|
||||
{
|
||||
BodyId = descriptor.BodyId,
|
||||
SchemaId = descriptor.SchemaId,
|
||||
LayoutVersion = descriptor.LayoutVersion,
|
||||
ProducerEpoch = descriptor.ProducerEpoch,
|
||||
Sequence = descriptor.Sequence,
|
||||
Capacity = descriptor.Capacity,
|
||||
AccessMode = CultMeshBodyAccessMode.ReadOnly,
|
||||
NowUtc = _now
|
||||
};
|
||||
}
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"name": "org.gamecult.cultlib",
|
||||
"displayName": "CultLib",
|
||||
"version": "1.0.30",
|
||||
"version": "1.0.31",
|
||||
"unity": "2021.3",
|
||||
"description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
|
||||
"license": "MIT",
|
||||
|
|
|
|||
Loading…
Reference in a new issue