diff --git a/src/GameCult.Mesh/CultMeshContentSessions.cs b/src/GameCult.Mesh/CultMeshContentSessions.cs new file mode 100644 index 00000000..99687358 --- /dev/null +++ b/src/GameCult.Mesh/CultMeshContentSessions.cs @@ -0,0 +1,191 @@ +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Caching; +using GameCult.Networking; + +namespace GameCult.Mesh +{ + /// + /// Serves bounded, content-addressed CDN chunks over the CultMesh content protocol. + /// Manifests and descriptors remain typed document state; payload bytes do not. + /// + public sealed class CultMeshContentServer : IDisposable + { + private readonly ICultNetSchemaServer _server; + private readonly CultCache _content; + private readonly Func _canServeHash; + private readonly Func _handler; + private bool _disposed; + + /// Attaches content serving to an existing schema host and provider cache. + public CultMeshContentServer( + ICultNetSchemaServer server, + CultCache content, + Func? canServeHash = null) + { + _server = server ?? throw new ArgumentNullException(nameof(server)); + _content = content ?? throw new ArgumentNullException(nameof(content)); + _canServeHash = canServeHash ?? (_ => true); + _handler = HandleAsync; + _server.OnCultNet(_handler); + } + + /// + public void Dispose() + { + if (_disposed) return; + _disposed = true; + _server.RemoveCultNetMessageListener(_handler); + } + + private Task HandleAsync( + CultMeshContentChunkRequestMessage request, + ICultNetSchemaServerPeer peer) + { + var response = new CultMeshContentChunkResponseMessage + { + MessageId = request?.MessageId ?? string.Empty, + ChunkHash = request?.ChunkHash ?? string.Empty + }; + try + { + if (request == null || string.IsNullOrWhiteSpace(request.MessageId)) + throw new InvalidDataException("Content chunk request requires a message identity."); + var hash = CultMeshCdn.NormalizeHash(request.ChunkHash, nameof(request.ChunkHash)); + if (!_canServeHash(hash)) + throw new UnauthorizedAccessException("Content access policy rejected the requested hash."); + var canonicalKey = CultMeshCdnArtifactChunk.CreateRecordKey(hash); + if (!string.IsNullOrWhiteSpace(request.RecordKey) && + !string.Equals(request.RecordKey, canonicalKey.Value, StringComparison.Ordinal)) + throw new InvalidDataException("Content chunk record key disagrees with its content hash."); + var chunk = _content.Get(canonicalKey) + ?? throw new FileNotFoundException("Content chunk is not available.", canonicalKey.Value); + var reference = new CultMeshCdnChunkRef + { + ChunkHash = hash, + RecordKey = canonicalKey.Value, + SizeBytes = request.ExpectedSizeBytes + }; + CultMeshCdn.ValidateChunkPayload(reference, chunk); + response.Found = true; + response.ChunkHash = hash; + response.SizeBytes = chunk.SizeBytes; + response.Payload = chunk.Payload; + } + catch (Exception error) + { + response.Found = false; + response.Error = error.GetType().Name + ": " + error.Message; + response.Payload = Array.Empty(); + } + peer.SendCultNet(response); + return Task.CompletedTask; + } + } + + /// Configures bounded response waiting for session-backed content requests. + public sealed class CultMeshSessionContentProviderOptions + { + /// Gets or sets the injected reliability clock. + public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance; + /// Gets or sets the maximum time to wait for one chunk response. + public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(10); + } + + /// + /// Retrieves verified CDN chunks through one reusable identity-first CultMesh session. + /// Transfer verification and final cache promotion remain owned by CultMeshContentTransferService. + /// + public sealed class CultMeshSessionContentProvider : ICultMeshContentProvider + { + private readonly CultMeshSessionManager _sessions; + private readonly CultMeshEndpointId _endpointId; + private readonly CultMeshSessionContentProviderOptions _options; + + /// Creates a content provider backed by a reusable identity-first session. + public CultMeshSessionContentProvider( + string providerId, + CultMeshSessionManager sessions, + CultMeshEndpointId endpointId, + CultMeshSessionContentProviderOptions? 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 CultMeshSessionContentProviderOptions(); + if (_options.ResponseTimeout <= TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(options), "Content response timeout must be positive."); + } + + /// + public string ProviderId { get; } + + /// + public async Task GetChunkAsync( + CultMeshCdnChunkRef chunk, + CancellationToken cancellationToken = default) + { + if (chunk == null) throw new ArgumentNullException(nameof(chunk)); + var hash = CultMeshCdn.NormalizeHash(chunk.ChunkHash, nameof(chunk.ChunkHash)); + var session = await _sessions.ConnectAsync( + _endpointId, + CultMeshProtocols.Content, + 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 CultMeshContentChunkRequestMessage + { + MessageId = messageId, + ChunkHash = hash, + RecordKey = string.IsNullOrWhiteSpace(chunk.RecordKey) + ? CultMeshCdnArtifactChunk.CreateRecordKey(hash).Value + : chunk.RecordKey, + ExpectedSizeBytes = chunk.SizeBytes + }); + + var response = await WaitForResponseAsync(completion.Task, cancellationToken).ConfigureAwait(false); + if (!response.Found) + throw new FileNotFoundException( + "Content provider '" + ProviderId + "' rejected chunk '" + hash + "': " + response.Error, + chunk.RecordKey); + var resolved = new CultMeshCdnArtifactChunk + { + ChunkHash = response.ChunkHash, + SizeBytes = response.SizeBytes, + Payload = response.Payload ?? Array.Empty() + }; + CultMeshCdn.ValidateChunkPayload(chunk, resolved); + return resolved; + } + + private async Task WaitForResponseAsync( + 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 content from provider '" + ProviderId + "' through endpoint identity '" + + _endpointId.Value + "'."); + } + } +} diff --git a/src/GameCult.Mesh/docs/content-sessions.md b/src/GameCult.Mesh/docs/content-sessions.md new file mode 100644 index 00000000..bd94bbae --- /dev/null +++ b/src/GameCult.Mesh/docs/content-sessions.md @@ -0,0 +1,44 @@ +# CultMesh content sessions + +Large immutable artifacts use CultMesh content sessions. Typed snapshots carry +manifests, body descriptors, leases, and progress; they do not carry artifact +payloads. + +The provider attaches one generic content server to its existing CultNet +schema host: + +```csharp +using var content = new CultMeshContentServer(schemaServer, providerCache); +``` + +The consumer creates a provider over its existing identity-first session +manager, then gives that provider to the transfer owner: + +```csharp +var provider = new CultMeshSessionContentProvider( + "aetheria.daemon", + sessions, + CultMeshEndpointId.Parse("service:aetheria.daemon")); + +var transfer = new CultMeshContentTransferService( + transferStateCache, + new[] { provider }, + new CultMeshContentTransferOptions(cacheDirectory)); + +var verifiedBodyPath = await transfer.FetchAsync(manifest, cancellationToken); +``` + +`CultMeshContentTransferService` remains the only owner of verified partial +state and final cache promotion. It validates each chunk, resumes durable work, +verifies the complete SHA-256 identity, and atomically publishes +`.body`. The session provider owns no cache truth and does not +implement a second retry, failover, or promotion policy. + +For local consumers, configure `CultMeshVerifiedBodyMappingBroker` and call +`FetchMappedBodyAsync`. The resulting descriptor maps only the completed body; +mapping failure remains observable and falls back through the equivalent +network descriptor without changing logical identity. + +The content protocol uses one bounded response per manifest chunk. A snapshot +response containing `gamecult.mesh.cdn_artifact_chunk.v1` payload records is a +legacy transport path and must not be used for bulk delivery. diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index 8f458646..c8738988 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -566,6 +566,43 @@ provider corruption and failover, forged checkpoint rejection, final-hash failure, atomic visibility, and concurrent request coalescing. Existing CDN tests cover the compatibility reader. +### Phase 4 managed content-session slice + +**Owner:** `CultMeshContentTransferService` still owns verification, resumable +state, provider failover, and atomic promotion. `CultMeshSessionContentProvider` +owns only one bounded request/response projection over a reusable +`cultmesh.content.v1` session; `CultMeshContentServer` owns only serving +canonical content-addressed chunks from provider storage. + +**Inputs:** stable provider endpoint identity, the shared session manager, a +validated manifest chunk reference, provider CultCache content, response +deadline, and caller cancellation. + +**Outputs:** one manifest-bound chunk response to the transfer service. The +session manager retains connection/reconnect/path authority, and the transfer +service decides whether the response becomes verified partial state. + +**Derived state:** request correlation identities and content-session physical +channels. Neither is content identity, authority evidence, or cache truth. + +**Forbidden writers:** snapshot servers cannot embed artifact payloads in raw +document responses; the session provider cannot write partial/final files, +mark ranges verified, choose a replacement endpoint, or authorize a producer. + +**Shared paths:** every cold network chunk enters the same incremental and +final verification path used by database and test providers. Concurrent chunks +borrow the same logical content session. + +**Cut line:** bulk CDN bytes no longer require +`CultNetSnapshotResponseRawMessage` records. Snapshots retain manifests and +descriptors; the managed content protocol carries bounded payload responses. + +**Verification layer:** Mesh tests transfer a multi-chunk body through one +managed session, assert zero snapshot requests, assert warm committed reuse +without another network request, and round-trip both wire messages through the +schema dispatcher. The released EveUnity/Aetheria cold witness remains the real +RUDP consumer proof required to close the slice. + ### Phase 5 verified CDN mapped-body slice **Owner:** `CultMeshContentTransferService` remains the sole writer and promoter diff --git a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs index a8487665..87697b66 100644 --- a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs +++ b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs @@ -74,6 +74,8 @@ namespace GameCult.Networking CultNetSchemaVersions.VerseCatalogResponse => Deserialize(payload), CultNetSchemaVersions.PeerExchangeRequest => Deserialize(payload), CultNetSchemaVersions.PeerExchangeResponse => Deserialize(payload), + CultNetSchemaVersions.ContentChunkRequest => Deserialize(payload), + CultNetSchemaVersions.ContentChunkResponse => 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 a37c2bc0..77151dd1 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -143,6 +143,14 @@ namespace GameCult.Networking /// CultMesh peer exchange response contract identifier. /// public const string PeerExchangeResponse = "cultmesh.peer_exchange_response.v0"; + /// + /// CultMesh content chunk request contract identifier. + /// + public const string ContentChunkRequest = "cultmesh.content_chunk_request.v1"; + /// + /// CultMesh content chunk response contract identifier. + /// + public const string ContentChunkResponse = "cultmesh.content_chunk_response.v1"; } /// @@ -1303,6 +1311,46 @@ namespace GameCult.Networking [Key("peers")] public CultMeshPeerCardMessage[] Peers { get; set; } = Array.Empty(); } + /// + /// Requests one bounded content-addressed chunk over a managed CultMesh content session. + /// + [MessagePackObject] + public class CultMeshContentChunkRequestMessage : ICultNetSchemaMessage + { + /// Gets or sets the wire schema identity. + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.ContentChunkRequest; + /// Gets or sets the request correlation identity. + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// Gets or sets the canonical SHA-256 chunk identity. + [Key("chunkHash")] public string ChunkHash { get; set; } = string.Empty; + /// Gets or sets the optional canonical CultCache record key. + [Key("recordKey")] public string RecordKey { get; set; } = string.Empty; + /// Gets or sets the manifest-bound expected payload size. + [Key("expectedSizeBytes")] public int ExpectedSizeBytes { get; set; } + } + + /// + /// Returns one bounded content-addressed chunk without publishing it as snapshot state. + /// + [MessagePackObject] + public class CultMeshContentChunkResponseMessage : ICultNetSchemaMessage + { + /// Gets or sets the wire schema identity. + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.ContentChunkResponse; + /// Gets or sets the matching request correlation identity. + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// Gets or sets whether a validated chunk is present. + [Key("found")] public bool Found { get; set; } + /// Gets or sets the canonical SHA-256 chunk identity. + [Key("chunkHash")] public string ChunkHash { get; set; } = string.Empty; + /// Gets or sets the payload byte count. + [Key("sizeBytes")] public int SizeBytes { get; set; } + /// Gets or sets the bounded content payload. + [Key("payload")] public byte[] Payload { get; set; } = Array.Empty(); + /// Gets or sets a typed-boundary failure description when the chunk is unavailable. + [Key("error")] public string Error { get; set; } = string.Empty; + } + /// /// Wire representation of a CultMesh peer candidate. /// diff --git a/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs b/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs new file mode 100644 index 00000000..9c6d0fba --- /dev/null +++ b/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs @@ -0,0 +1,197 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using GameCult.Caching; +using GameCult.Networking; +using NUnit.Framework; + +namespace GameCult.Mesh.Tests; + +[TestFixture] +public sealed class CultMeshContentSessionTests +{ + private string _directory = null!; + + [SetUp] + public void SetUp() => + _directory = Path.Combine(Path.GetTempPath(), "cultmesh-content-session-tests", Guid.NewGuid().ToString("N")); + + [TearDown] + public void TearDown() + { + if (Directory.Exists(_directory)) Directory.Delete(_directory, true); + } + + [Test] + public async Task ContentSessionTransfersLargeBodyWithoutSnapshotPayloadsAndReusesCommittedBody() + { + var payload = Enumerable.Range(0, 700_000).Select(value => (byte)(value % 251)).ToArray(); + var artifact = CultMesh.PackCdnArtifact("aetheria/world/windows", payload); + using var providerCache = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( + typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk))); + await CultMeshCdn.PublishAsync(providerCache, artifact); + var wireServer = new LoopbackSchemaServer(); + using var contentServer = new CultMeshContentServer(wireServer, providerCache); + var wireClient = new LoopbackSchemaClient(wireServer); + var connector = new LoopbackConnector(wireClient); + using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource() }); + using var sessions = new CultMeshSessionManager(discovery, new[] { connector }); + var provider = new CultMeshSessionContentProvider( + "aetheria.daemon", + sessions, + CultMeshEndpointId.Parse("service:aetheria.daemon")); + using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( + typeof(CultMeshContentTransferStateDocument))); + var transfer = new CultMeshContentTransferService( + transferState, + new[] { provider }, + new CultMeshContentTransferOptions(_directory)); + + var first = await transfer.FetchAsync(artifact.Manifest); + var requestsAfterCold = wireClient.ContentRequestCount; + var second = await transfer.FetchAsync(artifact.Manifest); + + File.ReadAllBytes(first).Should().Equal(payload); + second.Should().Be(first); + requestsAfterCold.Should().Be(artifact.Manifest.Chunks.Length); + wireClient.ContentRequestCount.Should().Be(requestsAfterCold, + "a committed verified body is the warm-cache authority"); + wireClient.SnapshotRequestCount.Should().Be(0, + "bulk content must never be represented as snapshot records"); + connector.ConnectCount.Should().Be(1, + "all chunk requests borrow one identity-first content session"); + } + + [Test] + public void ContentWireMessagesRoundTripThroughTheSchemaDispatcher() + { + var request = new CultMeshContentChunkRequestMessage + { + MessageId = "request-1", + ChunkHash = new string('a', 64), + RecordKey = "mesh:cdn:chunk:" + new string('a', 64), + ExpectedSizeBytes = 42 + }; + var response = new CultMeshContentChunkResponseMessage + { + MessageId = "request-1", + Found = true, + ChunkHash = new string('a', 64), + SizeBytes = 3, + Payload = new byte[] { 1, 2, 3 } + }; + + CultNetSchemaMessageSerialization.Deserialize(CultNetSchemaMessageSerialization.Serialize(request)) + .Should().BeEquivalentTo(request); + CultNetSchemaMessageSerialization.Deserialize(CultNetSchemaMessageSerialization.Serialize(response)) + .Should().BeEquivalentTo(response); + } + + private sealed class RouteSource : ICultMeshLookupSource + { + public string SourceId => "odin"; + + 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[] { "rudp://content.test:3076" }), + SourceId, + now, + now.AddMinutes(1), + CultMeshDiscoveryTrust.Signed) + }); + } + } + + private sealed class LoopbackConnector : ICultMeshTransportConnector + { + private readonly ICultNetSchemaClient _client; + private int _connectCount; + public LoopbackConnector(ICultNetSchemaClient client) => _client = client; + public string ConnectorId => "loopback"; + public int ConnectCount => Volatile.Read(ref _connectCount); + public bool CanConnect(CultMeshTransportCandidate candidate) => true; + public Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshProtocolId protocol, + CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _connectCount); + protocol.Should().Be(CultMeshProtocols.Content); + return Task.FromResult(_client); + } + } + + private sealed class LoopbackSchemaServer : ICultNetSchemaServer + { + private readonly Dictionary _handlers = new(); + + public void OnCultNet(Func callback) + where TMessage : ICultNetSchemaMessage => _handlers[typeof(TMessage)] = callback; + + public void RemoveCultNetMessageListener(Delegate callback) + where TMessage : ICultNetSchemaMessage => _handlers.Remove(typeof(TMessage)); + + public Task DispatchAsync(TMessage message, ICultNetSchemaServerPeer peer) + where TMessage : ICultNetSchemaMessage => + ((Func)_handlers[typeof(TMessage)])(message, peer); + } + + private sealed class LoopbackSchemaClient : ICultNetSchemaClient + { + private readonly LoopbackSchemaServer _server; + private readonly Dictionary> _handlers = new(); + public LoopbackSchemaClient(LoopbackSchemaServer server) => _server = server; + public bool Connected => true; + public int ContentRequestCount { get; private set; } + public int SnapshotRequestCount { get; private set; } + public void Connect(string host, int port) { } + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + if (message is CultMeshContentChunkRequestMessage content) + { + ContentRequestCount++; + _server.DispatchAsync(content, new LoopbackPeer(this)).GetAwaiter().GetResult(); + } + else if (message is CultNetSnapshotRequestMessage) + { + SnapshotRequestCount++; + throw new InvalidOperationException("Content session attempted to use a snapshot request."); + } + } + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (!_handlers.TryGetValue(typeof(T), out var handlers)) + _handlers[typeof(T)] = handlers = new List(); + handlers.Add(callback); + } + public void Dispose() { } + public void Emit(T message) where T : ICultNetSchemaMessage + { + if (!_handlers.TryGetValue(typeof(T), out var handlers)) return; + foreach (var handler in handlers.ToArray()) ((Action)handler)(message); + } + } + + private sealed class LoopbackPeer : ICultNetSchemaServerPeer + { + private readonly LoopbackSchemaClient _client; + public LoopbackPeer(LoopbackSchemaClient client) => _client = client; + public void SendCultNet(TMessage message) where TMessage : ICultNetSchemaMessage => + _client.Emit(message); + } +}