From a311cbe8cb096600a3fbc1224f9815373c827d13 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 14 Jul 2026 03:26:43 +0100 Subject: [PATCH] Improve managed content throughput --- src/GameCult.Mesh/CultMeshContentTransfer.cs | 53 ++++++-- src/GameCult.Mesh/docs/content-sessions.md | 12 ++ src/GameCult.Mesh/docs/distributed-cdn.md | 48 +++---- .../docs/reliability-migration-plan.md | 19 ++- .../CultNetRudpSchemaServer.cs | 46 ++++++- src/GameCult.Networking/CultNetTransport.cs | 35 +++-- .../CultMeshContentTransferTests.cs | 120 +++++++++++++++++- .../NetworkingTests.cs | 21 ++- 8 files changed, 292 insertions(+), 62 deletions(-) diff --git a/src/GameCult.Mesh/CultMeshContentTransfer.cs b/src/GameCult.Mesh/CultMeshContentTransfer.cs index 349c8736..cbcf641d 100644 --- a/src/GameCult.Mesh/CultMeshContentTransfer.cs +++ b/src/GameCult.Mesh/CultMeshContentTransfer.cs @@ -61,6 +61,9 @@ namespace GameCult.Mesh public sealed class CultMeshContentTransferOptions { + public const int DefaultMaxConcurrentChunkRequests = 4; + public const int MaximumConcurrentChunkRequests = 32; + public CultMeshContentTransferOptions(string cacheDirectory) { CacheDirectory = string.IsNullOrWhiteSpace(cacheDirectory) @@ -69,6 +72,12 @@ namespace GameCult.Mesh } public string CacheDirectory { get; } + + /// + /// Gets or sets the bounded number of chunk requests that may be in flight per + /// content hash while the transfer owner writes and checkpoints in manifest order. + /// + public int MaxConcurrentChunkRequests { get; set; } = DefaultMaxConcurrentChunkRequests; } public sealed class CultMeshContentTransferService @@ -76,6 +85,7 @@ namespace GameCult.Mesh private readonly CultCache _stateCache; private readonly ICultMeshContentProvider[] _providers; private readonly string _cacheDirectory; + private readonly int _maxConcurrentChunkRequests; private readonly CultMeshVerifiedBodyMappingBroker? _mappedBodies; private readonly ConcurrentDictionary _contentLocks = new(StringComparer.Ordinal); @@ -89,7 +99,13 @@ namespace GameCult.Mesh _providers = providers?.ToArray() ?? throw new ArgumentNullException(nameof(providers)); if (_providers.Length == 0) throw new ArgumentException("At least one content provider is required.", nameof(providers)); if (_providers.Any(provider => provider == null)) throw new ArgumentException("Content providers cannot contain null entries.", nameof(providers)); - _cacheDirectory = (options ?? throw new ArgumentNullException(nameof(options))).CacheDirectory; + if (options == null) throw new ArgumentNullException(nameof(options)); + if (options.MaxConcurrentChunkRequests <= 0 || + options.MaxConcurrentChunkRequests > CultMeshContentTransferOptions.MaximumConcurrentChunkRequests) + throw new ArgumentOutOfRangeException(nameof(options), + $"Concurrent chunk request bound must be between 1 and {CultMeshContentTransferOptions.MaximumConcurrentChunkRequests}."); + _cacheDirectory = options.CacheDirectory; + _maxConcurrentChunkRequests = options.MaxConcurrentChunkRequests; _mappedBodies = mappedBodies; } @@ -168,15 +184,36 @@ namespace GameCult.Mesh .OrderBy(item => item.Chunk.Offset) .ToArray(); - foreach (var item in ordered.Where(item => !verified.Contains(item.Index))) + var remaining = ordered.Where(item => !verified.Contains(item.Index)).ToArray(); + for (var windowStart = 0; windowStart < remaining.Length; windowStart += _maxConcurrentChunkRequests) { cancellationToken.ThrowIfCancellationRequested(); - var chunk = await FetchVerifiedChunkAsync(item.Chunk, cancellationToken).ConfigureAwait(false); - stream.Position = item.Chunk.Offset; - await stream.WriteAsync(chunk.Payload, 0, chunk.Payload.Length, cancellationToken).ConfigureAwait(false); - await stream.FlushAsync(cancellationToken).ConfigureAwait(false); - verified.Add(item.Index); - await SaveStateAsync(contentHash, fingerprint, manifest.SizeBytes, verified, stateKey).ConfigureAwait(false); + var window = remaining + .Skip(windowStart) + .Take(_maxConcurrentChunkRequests) + .Select(item => (Item: item, Fetch: FetchVerifiedChunkAsync(item.Chunk, cancellationToken))) + .ToArray(); + try + { + foreach (var pending in window) + { + var chunk = await pending.Fetch.ConfigureAwait(false); + stream.Position = pending.Item.Chunk.Offset; + await stream.WriteAsync(chunk.Payload, 0, chunk.Payload.Length, cancellationToken).ConfigureAwait(false); + await stream.FlushAsync(cancellationToken).ConfigureAwait(false); + verified.Add(pending.Item.Index); + await SaveStateAsync(contentHash, fingerprint, manifest.SizeBytes, verified, stateKey).ConfigureAwait(false); + } + } + catch + { + foreach (var pending in window) + { + try { await pending.Fetch.ConfigureAwait(false); } + catch { } + } + throw; + } } } diff --git a/src/GameCult.Mesh/docs/content-sessions.md b/src/GameCult.Mesh/docs/content-sessions.md index 30d98562..f3ea98dd 100644 --- a/src/GameCult.Mesh/docs/content-sessions.md +++ b/src/GameCult.Mesh/docs/content-sessions.md @@ -41,3 +41,15 @@ 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. + +The transfer owner may keep a bounded window of requests in flight per content +hash (four by default, never more than 32). Candidate chunks can arrive out of +order, but only the transfer owner writes them, and it flushes and checkpoints +them in manifest-offset order. If one request fails, the owner observes the +rest of that window before returning; fetched-but-uncommitted bytes never become +trusted resume state. Concurrent artifacts have independent windows. + +RUDP schema hosts should drain immediately available transport work with +`PollAvailableAsync`. Its result distinguishes transport items consumed from +application messages dispatched, so ACK and connection traffic counts as +progress without impersonating a message or triggering an idle delay. diff --git a/src/GameCult.Mesh/docs/distributed-cdn.md b/src/GameCult.Mesh/docs/distributed-cdn.md index fef4e276..85acfb31 100644 --- a/src/GameCult.Mesh/docs/distributed-cdn.md +++ b/src/GameCult.Mesh/docs/distributed-cdn.md @@ -1,10 +1,10 @@ # Distributed CDN Documents -CultMesh CDN artifacts are typed CultCache documents, not a separate asset -transport. The same manifest and chunk records can be written by an asset -pipeline daemon, replicated through CultNet raw snapshots or shard logs, and -read by Unity, TypeScript, or another runtime that implements the CultCache -document contract. +CultMesh CDN artifacts use typed CultCache storage and a separate managed +content-delivery session. Manifests may travel through snapshots or shard logs; +artifact payload chunks remain in provider storage and cross the network through +`cultmesh.content.v1`. Unity, TypeScript, and other runtimes consume the same +manifest and verification contract without making snapshots a bulk byte lane. This is the canonical path for moving Aetheria assets into CultLib. Starbridge Unity and Starbridge Electron should consume the same records so UI textures, @@ -67,11 +67,11 @@ updates. Clients should: 1. Fetch or receive a manifest by record key, tag, or index. -2. Fetch every chunk record referenced by `manifest.Chunks`. -3. Verify each chunk hash before copying bytes. -4. Reassemble chunks by `Offset`. -5. Verify the full `ContentHash`. -6. Interpret bytes according to `MimeType` plus domain metadata. +2. Give it to `CultMeshContentTransferService` with one or more content providers. +3. Let that owner request missing chunks through managed content sessions. +4. Use only the atomically promoted `.body` returned after chunk + and full-body verification. +5. Interpret bytes according to `MimeType` plus domain metadata. Single-channel data stays single-channel. For example, a gravity map can use raw bytes with metadata such as `channels=r32` instead of pretending to be an @@ -147,27 +147,20 @@ await CultMesh.PublishCdnArtifactAsync(node.Cache, artifact); For build distribution, use the same API with `Kind = CultMeshCdnArtifactKinds.Build` and metadata such as `platform=windows-x64` or `channel=internal`. -## CultNet Replication +## CultNet Publication And Delivery -CDN documents use the regular raw document lane: +Publish manifests and chunks into the provider cache: ```csharp -var registry = CultMesh.CreateCdnDocumentRegistry(node.Cache.Registry); -var request = registry.CreateSnapshotRequest( - "cdn-request", - schemaIds: - [ - node.Cache.Registry.GetRequired().SchemaId, - node.Cache.Registry.GetRequired().SchemaId - ], - recordKeys: artifact.RecordKeys.Select(key => key.Value)); - -var response = registry.CreateRawSnapshotResponse(node.Cache, "cdn-response", request); +await CultMesh.PublishCdnArtifactAsync(providerCache, artifact); +using var contentServer = new CultMeshContentServer(schemaServer, providerCache); ``` -The TypeScript client should mirror the C# schema names, schema versions, -record-key rules, chunk hashing, and verification order. C# remains the -reference implementation. +Snapshots may publish the manifest record, catalog, and body-control documents. +They must not publish `CultMeshCdnArtifactChunk` payload records for bulk +delivery. Consumers use `CultMeshSessionContentProvider` and +`CultMeshContentTransferService`; C# remains the reference implementation for +record keys, hashing, resumability, and atomic promotion. ## Entity Prefabs @@ -187,7 +180,8 @@ the same package graph into their native entity/component model. Unity and TypeScript clients should implement the same rules: - Fetch prefab packages by record key, tag, or version catalog. -- Fetch every referenced artifact manifest and chunk. +- Fetch every referenced artifact manifest, then delegate payload transfer to + the generic content-transfer owner. - Verify signed manifests before trusting package membership. - Verify chunk hashes and full artifact hashes before import. - Cache artifacts locally by content hash. diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index c8738988..8b4736ab 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -551,9 +551,10 @@ atomic promotion primitive. Concurrent requests for one hash share the same per-content transfer lock and verified work. **Cut line:** the legacy distributed reader no longer owns a sequential chunk -loop or whole-artifact allocation. The first slice intentionally leaves -adaptive parallelism, quotas, pinning, signer policy, arbitrary streaming -destinations, and cleanup scheduling to later Phase 4 work. +loop or whole-artifact allocation. The transfer owner now opens a bounded +per-content request window while preserving one ordered writer/checkpointer. +Adaptive scheduling, quotas, pinning, signer policy, arbitrary streaming +destinations, and cleanup scheduling remain later Phase 4 work. **State transition:** each verified chunk flushes a typed checkpoint. Restart rehydrates that record, rehashes every claimed range, and fetches corrupt or @@ -563,8 +564,10 @@ partial state. **Verification layer:** focused tests cover durable cancellation/restart, provider corruption and failover, forged checkpoint rejection, final-hash -failure, atomic visibility, and concurrent request coalescing. Existing CDN -tests cover the compatibility reader. +failure, atomic visibility, concurrent request coalescing, bounded concurrent +fetch, ordered checkpointing under out-of-order completion, and observation of +the whole request window after an early failure. Existing CDN tests cover the +compatibility reader. ### Phase 4 managed content-session slice @@ -600,8 +603,10 @@ 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. +schema dispatcher. RUDP schema-host tests distinguish control-packet progress +from dispatched messages and drain bounded available work without idle sleeps. +The released EveUnity/Aetheria cold witness remains the real consumer proof +required to close the slice. ### Phase 5 verified CDN mapped-body slice diff --git a/src/GameCult.Networking/CultNetRudpSchemaServer.cs b/src/GameCult.Networking/CultNetRudpSchemaServer.cs index 28063bdd..e6236943 100644 --- a/src/GameCult.Networking/CultNetRudpSchemaServer.cs +++ b/src/GameCult.Networking/CultNetRudpSchemaServer.cs @@ -7,6 +7,23 @@ using System.Threading.Tasks; namespace GameCult.Networking { + /// + /// Result of a bounded non-blocking RUDP schema-server drain. + /// + public readonly struct RudpCultNetSchemaPollResult + { + public RudpCultNetSchemaPollResult(int transportItemsConsumed, int messagesDispatched) + { + TransportItemsConsumed = transportItemsConsumed; + MessagesDispatched = messagesDispatched; + } + + /// Gets transport work consumed, including ACK and connection control packets. + public int TransportItemsConsumed { get; } + /// Gets application schema messages dispatched. + public int MessagesDispatched { get; } + } + /// /// Options for hosting schema-v0 messages over the native CultNet RUDP listener. /// @@ -223,12 +240,37 @@ namespace GameCult.Networking /// public async Task PollOnceAsync() { - var delivered = _transport.ReceiveOnce(); - if (delivered == null || !string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal)) + if (!_transport.TryReceiveOnce(out var delivered) || delivered == null) { return false; } + return await DispatchAsync(delivered).ConfigureAwait(false); + } + + /// + /// Drains up to immediately available transport work items. + /// Control packets count as progress so callers only idle when the transport is actually empty. + /// + public async Task PollAvailableAsync(int maxTransportItems) + { + if (maxTransportItems <= 0) throw new ArgumentOutOfRangeException(nameof(maxTransportItems)); + var consumed = 0; + var dispatched = 0; + while (consumed < maxTransportItems && _transport.TryReceiveOnce(out var delivered)) + { + consumed++; + if (delivered != null && await DispatchAsync(delivered).ConfigureAwait(false)) + dispatched++; + } + return new RudpCultNetSchemaPollResult(consumed, dispatched); + } + + private async Task DispatchAsync(CultNetRudpSocketServerFrame delivered) + { + if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal)) + return false; + var message = CultNetSchemaMessageSerialization.Deserialize(delivered.Frame.Payload); if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null) { diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index 8e1f15a8..4e61741b 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -2196,12 +2196,25 @@ namespace GameCult.Networking /// Polls the UDP socket once and returns the next delivered peer frame if one is available. /// public CultNetRudpSocketServerFrame? ReceiveOnce() + { + TryReceiveOnce(out var delivered); + return delivered; + } + + /// + /// Consumes one available transport work item and reports whether transport progress was made, + /// independently of whether that work item completed an application frame. + /// + public bool TryReceiveOnce(out CultNetRudpSocketServerFrame? delivered) { if (_deliveredFrames.Count > 0) { - return _deliveredFrames.Dequeue(); + delivered = _deliveredFrames.Dequeue(); + return true; } + delivered = null; + var buffer = new byte[65535]; EndPoint remote = new IPEndPoint(IPAddress.Any, 0); int received; @@ -2209,7 +2222,7 @@ namespace GameCult.Networking { if (!_socket.Poll(0, SelectMode.SelectRead)) { - return null; + return false; } received = _socket.ReceiveFrom(buffer, ref remote); @@ -2218,14 +2231,14 @@ namespace GameCult.Networking error.SocketErrorCode == SocketError.WouldBlock || error.SocketErrorCode == SocketError.TimedOut) { - return null; + return false; } catch (SocketException error) when ( error.SocketErrorCode == SocketError.ConnectionReset || error.SocketErrorCode == SocketError.ConnectionAborted || error.SocketErrorCode == SocketError.Shutdown) { - return null; + return false; } _stats.BytesReceived += received; @@ -2239,12 +2252,12 @@ namespace GameCult.Networking } catch (InvalidOperationException) { - return null; + return true; } if (packet.ConnectionId != _connectionId) { - return null; + return true; } var peerKey = RemoteKey(remote); @@ -2261,12 +2274,12 @@ namespace GameCult.Networking })); _peers[peerKey] = peer; SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload)); - return null; + return true; } if (!_peers.TryGetValue(peerKey, out var existingPeer)) { - return null; + return true; } var result = existingPeer.Session.Receive(packet, NowMs()); @@ -2278,7 +2291,7 @@ namespace GameCult.Networking { existingPeer.DisconnectReason = result.DisconnectReason; _peers.Remove(peerKey); - return null; + return true; } foreach (var frame in result.Delivered) @@ -2295,13 +2308,13 @@ namespace GameCult.Networking _stats.FramesReceived++; } - var delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null; + delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null; if (packet.PacketType == CultNetRudpPacketType.Data || delivered != null) { SendPacket(existingPeer.RemoteEndPoint, existingPeer.Session.CreateAck()); } - return delivered; + return true; } /// diff --git a/tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs b/tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs index 70e0ee03..bad8ac74 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs @@ -141,6 +141,79 @@ public sealed class CultMeshContentTransferTests provider.Requests.Values.Sum().Should().Be(artifact.Chunks.Count); } + [Test] + public async Task FetchAsync_PipelinesRequestsWithinConfiguredBoundAndCommitsInManifestOrder() + { + var artifact = Artifact("pipelined", 2); + using var cache = Cache(); + var provider = new FakeProvider("source", artifact.Chunks) + { + DelayMilliseconds = 40, + BlockedChunkHash = artifact.Chunks[0].ChunkHash + }; + var service = new CultMeshContentTransferService( + cache, + new[] { provider }, + new CultMeshContentTransferOptions(_directory) { MaxConcurrentChunkRequests = 3 }); + + var fetch = service.FetchAsync(artifact.Manifest); + try + { + await provider.BlockedChunkRequested.Task.WaitAsync(TimeSpan.FromSeconds(2)); + await WaitUntilAsync(() => provider.Requests.Values.Sum() >= 3 && provider.ActiveRequests == 1); + + var pendingState = cache.Get(StateKey(artifact.Manifest)); + (pendingState?.VerifiedChunkIndexes ?? Array.Empty()).Should().BeEmpty( + "later chunks may finish fetching but cannot be checkpointed ahead of the first manifest chunk"); + } + finally + { + provider.ReleaseBlockedChunk(); + } + var path = await fetch; + + File.ReadAllBytes(path).Should().Equal(Payload()); + provider.MaxConcurrentRequests.Should().Be(3); + provider.Requests.Values.Sum().Should().Be(artifact.Chunks.Count); + cache.Get(StateKey(artifact.Manifest)).Should().BeNull(); + } + + [Test] + public void Constructor_RejectsNonPositiveConcurrentChunkBound() + { + using var cache = Cache(); + var options = new CultMeshContentTransferOptions(_directory) { MaxConcurrentChunkRequests = 0 }; + + FluentActions.Invoking(() => new CultMeshContentTransferService( + cache, + new[] { new FakeProvider("source", Array.Empty()) }, + options)) + .Should().Throw(); + } + + [Test] + public async Task FetchAsync_ObservesRemainingWindowTasksWhenEarlierFetchFails() + { + var artifact = Artifact("failed-window", 2); + using var cache = Cache(); + var provider = new FakeProvider("source", artifact.Chunks) + { + DelayMilliseconds = 80, + FailedChunkHash = artifact.Chunks[0].ChunkHash + }; + var service = new CultMeshContentTransferService( + cache, + new[] { provider }, + new CultMeshContentTransferOptions(_directory) { MaxConcurrentChunkRequests = 3 }); + + await FluentActions.Awaiting(() => service.FetchAsync(artifact.Manifest)) + .Should().ThrowAsync(); + + provider.ActiveRequests.Should().Be(0, "the failed window must observe every task before returning"); + cache.Get(StateKey(artifact.Manifest)).Should().BeNull(); + File.Exists(Path.Combine(_directory, artifact.Manifest.ContentHash + ".body")).Should().BeFalse(); + } + [Test] public async Task FetchMappedBodyAsync_MapsOnlyCommittedBodyForMultipleConsumersAndFallsBackToNetwork() { @@ -267,6 +340,9 @@ public sealed class CultMeshContentTransferTests { private readonly Dictionary _chunks; private int _requestCount; + private int _activeRequests; + private int _maxConcurrentRequests; + private readonly TaskCompletionSource _releaseBlockedChunk = new(TaskCreationOptions.RunContinuationsAsynchronously); public FakeProvider(string providerId, IEnumerable chunks) { @@ -278,16 +354,50 @@ public sealed class CultMeshContentTransferTests public ConcurrentDictionary Requests { get; } = new(StringComparer.Ordinal); public int BlockFromRequest { get; set; } = int.MaxValue; public int DelayMilliseconds { get; set; } + public string? BlockedChunkHash { get; set; } + public string? FailedChunkHash { get; set; } + public TaskCompletionSource BlockedChunkRequested { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public int ActiveRequests => Volatile.Read(ref _activeRequests); + public int MaxConcurrentRequests => Volatile.Read(ref _maxConcurrentRequests); + + public void ReleaseBlockedChunk() => _releaseBlockedChunk.TrySetResult(true); public async Task GetChunkAsync(CultMeshCdnChunkRef chunk, CancellationToken cancellationToken = default) { Requests.AddOrUpdate(chunk.ChunkHash, 1, (_, count) => count + 1); var request = Interlocked.Increment(ref _requestCount); - if (request >= BlockFromRequest) - await Task.Delay(Timeout.Infinite, cancellationToken); - if (DelayMilliseconds > 0) - await Task.Delay(DelayMilliseconds, cancellationToken); - return _chunks.TryGetValue(chunk.ChunkHash, out var value) ? Clone(value) : null; + var active = Interlocked.Increment(ref _activeRequests); + UpdateMaximum(ref _maxConcurrentRequests, active); + try + { + if (request >= BlockFromRequest) + await Task.Delay(Timeout.Infinite, cancellationToken); + if (string.Equals(chunk.ChunkHash, FailedChunkHash, StringComparison.Ordinal)) + throw new InvalidDataException("Synthetic chunk failure."); + if (string.Equals(chunk.ChunkHash, BlockedChunkHash, StringComparison.Ordinal)) + { + BlockedChunkRequested.TrySetResult(true); + await _releaseBlockedChunk.Task.WaitAsync(cancellationToken); + } + if (DelayMilliseconds > 0) + await Task.Delay(DelayMilliseconds, cancellationToken); + return _chunks.TryGetValue(chunk.ChunkHash, out var value) ? Clone(value) : null; + } + finally + { + Interlocked.Decrement(ref _activeRequests); + } + } + + private static void UpdateMaximum(ref int maximum, int candidate) + { + var observed = Volatile.Read(ref maximum); + while (candidate > observed) + { + var previous = Interlocked.CompareExchange(ref maximum, candidate, observed); + if (previous == observed) return; + observed = previous; + } } } } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index bae6f5e5..07848be5 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1203,6 +1203,7 @@ namespace GameCult.Networking.Tests MaxFragmentBytes = 1024 }); var serverDone = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var controlProgressObserved = 0; server.OnCultNet((request, peer) => { peer.SendCultNet(new CultNetSchemaCatalogResponseMessage @@ -1228,8 +1229,11 @@ namespace GameCult.Networking.Tests { while (!serverDone.Task.IsCompleted) { - _ = server.PollOnceAsync().GetAwaiter().GetResult(); - Thread.Sleep(5); + var poll = server.PollAvailableAsync(64).GetAwaiter().GetResult(); + if (poll.TransportItemsConsumed > poll.MessagesDispatched) + Interlocked.Exchange(ref controlProgressObserved, 1); + if (poll.TransportItemsConsumed == 0) + Thread.Sleep(5); } } catch (Exception error) @@ -1260,6 +1264,19 @@ namespace GameCult.Networking.Tests Assert.That(response.MessageId, Is.EqualTo("rudp-schema-host-test")); Assert.That(response.Schemas.Single().SchemaId, Is.EqualTo("rudp.schema.host")); Assert.That(server.Profile.Transports[0].Protocol, Is.EqualTo("rudp")); + Assert.That(Volatile.Read(ref controlProgressObserved), Is.EqualTo(1), + "connection and ACK traffic must count as transport progress without impersonating an application message"); + } + + [Test] + public void RudpCultNetSchemaServer_RejectsNonPositiveDrainBound() + { + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + Socket = BindUdpSocket() + }); + + Assert.ThrowsAsync(() => server.PollAvailableAsync(0)); } [Test]