From 091200c09d05e7cbb75eeed7513f1ea8edb05c94 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 19 Jul 2026 13:57:06 +0100 Subject: [PATCH] Move hot bodies off snapshot state --- Directory.Build.props | 2 +- src/GameCult.Mesh/CultMeshBodies.cs | 18 +- src/GameCult.Mesh/CultMeshBodyPublications.cs | 200 ---------- src/GameCult.Mesh/CultMeshBodySessions.cs | 342 ++++++++++++++++++ src/GameCult.Mesh/CultMeshClient.cs | 14 + src/GameCult.Mesh/CultMeshSessions.cs | 1 + src/GameCult.Mesh/docs/public-api.md | 6 +- .../CultNetSchemaMessageSerialization.cs | 2 + .../CultNetSchemaMessages.cs | 44 +++ .../CultMeshBodySessionTests.cs | 218 +++++++++++ .../CultMeshNetworkBodyPublicationTests.cs | 158 -------- unity/org.gamecult.cultlib/package.json | 2 +- 12 files changed, 643 insertions(+), 364 deletions(-) create mode 100644 src/GameCult.Mesh/CultMeshBodySessions.cs create mode 100644 tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs delete mode 100644 tests/GameCult.Mesh.Tests/CultMeshNetworkBodyPublicationTests.cs diff --git a/Directory.Build.props b/Directory.Build.props index 45b0cfa3..28bdff53 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -14,7 +14,7 @@ MIT © Meta Vens and contributors true - 1.0.30 + 1.0.31 true diff --git a/src/GameCult.Mesh/CultMeshBodies.cs b/src/GameCult.Mesh/CultMeshBodies.cs index ff12e3b9..0c740c18 100644 --- a/src/GameCult.Mesh/CultMeshBodies.cs +++ b/src/GameCult.Mesh/CultMeshBodies.cs @@ -572,10 +572,19 @@ namespace GameCult.Mesh public sealed class CultMeshNetworkBodyAdapter : ICultMeshBodyTransportAdapter { private readonly Func _fetch; + private readonly bool _fetchIsVerified; public CultMeshNetworkBodyAdapter(Func fetch) => _fetch = fetch ?? throw new ArgumentNullException(nameof(fetch)); + /// Uses a session provider that already verifies the response against the descriptor. + 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); } } diff --git a/src/GameCult.Mesh/CultMeshBodyPublications.cs b/src/GameCult.Mesh/CultMeshBodyPublications.cs index 673465cc..2358c277 100644 --- a/src/GameCult.Mesh/CultMeshBodyPublications.cs +++ b/src/GameCult.Mesh/CultMeshBodyPublications.cs @@ -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 _authorizePublication; - private readonly int _chunkSizeBytes; - - public CultMeshNetworkBodyPublisher( - CultCache cache, - Func 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 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(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(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 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.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(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(documents)) - .Register(CultNetDocumentBinding.ForDocument(documents)) .Register(CultNetDocumentBinding.ForDocument(documents)) .Register(CultNetDocumentBinding.ForDocument(documents)); } diff --git a/src/GameCult.Mesh/CultMeshBodySessions.cs b/src/GameCult.Mesh/CultMeshBodySessions.cs new file mode 100644 index 00000000..cc45277e --- /dev/null +++ b/src/GameCult.Mesh/CultMeshBodySessions.cs @@ -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 +{ + /// + /// Owns a bounded set of immutable network body generations. This is ephemeral transport state, + /// not a CultCache document store and not world truth. + /// + 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 _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 body) + { + return PublishCore(generation, body.ToArray()); + } + + /// + /// 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. + /// + 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(); + 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; } + } + } + + /// Serves capability-bound live bodies directly over an existing CultNet host. + public sealed class CultMeshBodyServer : IDisposable + { + private readonly ICultNetSchemaServer _server; + private readonly CultMeshNetworkBodyStore _bodies; + private readonly Func _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(_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(); + } + 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); + } + + /// Reads live body generations through one reusable identity-first body session. + 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 CreateFetchDelegate() => + descriptor => FetchAsync(descriptor).GetAwaiter().GetResult(); + + public async Task 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(TaskCreationOptions.RunContinuationsAsynchronously); + using var responseSubscription = session.OnCultNet(response => + { + if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) completion.TrySetResult(response); + }); + using var errorSubscription = session.OnCultNet(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(); + 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 WaitAsync(Task 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}'."); + } + } +} diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index e5591397..2be64288 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -108,6 +108,20 @@ namespace GameCult.Mesh options); } + /// Creates a direct body provider over this client's reusable negotiated session owner. + public CultMeshSessionBodyProvider BodyProvider( + string providerId, + string endpointId, + CultMeshSessionBodyProviderOptions? options = null) + { + ThrowIfDisposed(); + return new CultMeshSessionBodyProvider( + providerId, + _sessions, + CultMeshEndpointId.Parse(endpointId), + options); + } + /// Opens one live typed document by stable provider identity and record key. public async Task> DocumentAsync( string endpointId, diff --git a/src/GameCult.Mesh/CultMeshSessions.cs b/src/GameCult.Mesh/CultMeshSessions.cs index e654fddc..6c10e263 100644 --- a/src/GameCult.Mesh/CultMeshSessions.cs +++ b/src/GameCult.Mesh/CultMeshSessions.cs @@ -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"); diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index 5804b59d..174ff84a 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -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 diff --git a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs index 87697b66..386bc99b 100644 --- a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs +++ b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs @@ -76,6 +76,8 @@ namespace GameCult.Networking CultNetSchemaVersions.PeerExchangeResponse => Deserialize(payload), CultNetSchemaVersions.ContentChunkRequest => Deserialize(payload), CultNetSchemaVersions.ContentChunkResponse => Deserialize(payload), + CultNetSchemaVersions.BodyReadRequest => Deserialize(payload), + CultNetSchemaVersions.BodyReadResponse => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogRequest => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogResponse => Deserialize(payload), _ => throw new MessagePackSerializationException( diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 2dc38ffa..9a8800fc 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -151,6 +151,14 @@ namespace GameCult.Networking /// CultMesh content chunk response contract identifier. /// public const string ContentChunkResponse = "cultmesh.content_chunk_response.v1"; + /// + /// CultMesh ephemeral body read request contract identifier. + /// + public const string BodyReadRequest = "cultmesh.body_read_request.v1"; + /// + /// CultMesh ephemeral body read response contract identifier. + /// + public const string BodyReadResponse = "cultmesh.body_read_response.v1"; } /// @@ -1363,6 +1371,42 @@ namespace GameCult.Networking [Key("error")] public string Error { get; set; } = string.Empty; } + /// + /// Requests one immutable body generation from its capability owner. Body bytes travel on this + /// dedicated plane; they are never database snapshot records. + /// + [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; + } + + /// Returns one bounded immutable body generation directly to its subscriber. + [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(); + [Key("error")] public string Error { get; set; } = string.Empty; + } + /// /// Wire representation of a CultMesh peer candidate. /// diff --git a/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs b/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs new file mode 100644 index 00000000..55e45207 --- /dev/null +++ b/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs @@ -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> LookupAsync(CultMeshDiscoveryQuery query, + CancellationToken cancellationToken = default) + { + var now = DateTimeOffset.UtcNow; + return Task.FromResult>(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 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 _handlers = new(); + public void OnCultNet(Func callback) where T : ICultNetSchemaMessage => + _handlers[typeof(T)] = callback; + public void RemoveCultNetMessageListener(Delegate callback) where T : ICultNetSchemaMessage => + _handlers.Remove(typeof(T)); + public Task Dispatch(T message, ICultNetSchemaServerPeer peer) where T : ICultNetSchemaMessage => + ((Func)_handlers[typeof(T)])(message, peer); + } + + private sealed class LoopbackClient : ICultNetSchemaClient + { + private readonly LoopbackServer _server; + private readonly Dictionary> _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 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(Action callback) where T : ICultNetSchemaMessage + { + if (!_handlers.TryGetValue(typeof(T), out var values)) _handlers[typeof(T)] = values = new List(); + values.Add(callback); + } + public void Emit(T message) where T : ICultNetSchemaMessage + { + if (_handlers.TryGetValue(typeof(T), out var values)) + foreach (var value in values.ToArray()) ((Action)value)(message); + } + public void Dispose() { } + } + + private sealed class LoopbackPeer : ICultNetSchemaServerPeer + { + private readonly LoopbackClient _client; + public LoopbackPeer(LoopbackClient client) => _client = client; + public void SendCultNet(T message) where T : ICultNetSchemaMessage => _client.Emit(message); + } +} diff --git a/tests/GameCult.Mesh.Tests/CultMeshNetworkBodyPublicationTests.cs b/tests/GameCult.Mesh.Tests/CultMeshNetworkBodyPublicationTests.cs deleted file mode 100644 index 132714fe..00000000 --- a/tests/GameCult.Mesh.Tests/CultMeshNetworkBodyPublicationTests.cs +++ /dev/null @@ -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.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.CreateRecordKey(descriptor.CapabilityToken))!; - var manifest = _cache.Get(new CultRecordKey(binding.ManifestRecordKey))!; - var chunkKey = new CultRecordKey(manifest.Chunks[0].RecordKey); - if (corrupt) - { - var chunk = _cache.Get(chunkKey)!; - chunk.Payload[0] ^= 0xff; - await _cache.UpsertAsync(chunk, new CultRecordHandle(chunkKey)); - } - else - { - _cache.Remove(chunkKey); - } - - Action fetch = () => new CultMeshNetworkBodyResolver(_cache).Fetch(descriptor); - - if (corrupt) fetch.Should().Throw(); - else fetch.Should().Throw(); - } - - [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().WithMessage("*capability binding*"); - } - - [Test] - public async Task Publish_RejectsUnauthorizedProducerBeforeWritingDocuments() - { - var generation = Generation(); - var publisher = new CultMeshNetworkBodyPublisher(_cache, _ => false); - - Func publish = () => publisher.PublishAsync(generation, new byte[] { 1, 2, 3, 4 }); - - await publish.Should().ThrowAsync(); - } - - [Test] - public async Task Publish_RejectsMutationOfAnExistingGeneration() - { - var publisher = Publisher(); - var generation = Generation(); - await publisher.PublishAsync(generation, new byte[] { 1, 2, 3, 4 }); - - Func republish = () => publisher.PublishAsync(generation, new byte[] { 1, 2, 3, 5 }); - - await republish.Should().ThrowAsync().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 - }; -} diff --git a/unity/org.gamecult.cultlib/package.json b/unity/org.gamecult.cultlib/package.json index f80cda7b..e51e5444 100644 --- a/unity/org.gamecult.cultlib/package.json +++ b/unity/org.gamecult.cultlib/package.json @@ -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",