Improve managed content throughput

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; }
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        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<string, SemaphoreSlim> _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;
+                    }
                 }
             }
 

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.

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 `<content-hash>.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<CultMeshCdnArtifactManifest>().SchemaId,
-        node.Cache.Registry.GetRequired<CultMeshCdnArtifactChunk>().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.

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
 

src/GameCult.Networking/CultNetRudpSchemaServer.cs

@@ -7,6 +7,23 @@ using System.Threading.Tasks;
 
 namespace GameCult.Networking
 {
+    /// <summary>
+    /// Result of a bounded non-blocking RUDP schema-server drain.
+    /// </summary>
+    public readonly struct RudpCultNetSchemaPollResult
+    {
+        public RudpCultNetSchemaPollResult(int transportItemsConsumed, int messagesDispatched)
+        {
+            TransportItemsConsumed = transportItemsConsumed;
+            MessagesDispatched = messagesDispatched;
+        }
+
+        /// <summary>Gets transport work consumed, including ACK and connection control packets.</summary>
+        public int TransportItemsConsumed { get; }
+        /// <summary>Gets application schema messages dispatched.</summary>
+        public int MessagesDispatched { get; }
+    }
+
     /// <summary>
     /// Options for hosting schema-v0 messages over the native CultNet RUDP listener.
     /// </summary>
@@ -223,12 +240,37 @@ namespace GameCult.Networking
         /// </summary>
         public async Task<bool> 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);
+        }
+
+        /// <summary>
+        /// Drains up to <paramref name="maxTransportItems"/> immediately available transport work items.
+        /// Control packets count as progress so callers only idle when the transport is actually empty.
+        /// </summary>
+        public async Task<RudpCultNetSchemaPollResult> 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<bool> 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)
             {

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.
         /// </summary>
         public CultNetRudpSocketServerFrame? ReceiveOnce()
+        {
+            TryReceiveOnce(out var delivered);
+            return delivered;
+        }
+
+        /// <summary>
+        /// Consumes one available transport work item and reports whether transport progress was made,
+        /// independently of whether that work item completed an application frame.
+        /// </summary>
+        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;
         }
 
         /// <summary>

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<CultMeshContentTransferStateDocument>(StateKey(artifact.Manifest));
+            (pendingState?.VerifiedChunkIndexes ?? Array.Empty<int>()).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<CultMeshContentTransferStateDocument>(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<CultMeshCdnArtifactChunk>()) },
+                options))
+            .Should().Throw<ArgumentOutOfRangeException>();
+    }
+
+    [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<InvalidDataException>();
+
+        provider.ActiveRequests.Should().Be(0, "the failed window must observe every task before returning");
+        cache.Get<CultMeshContentTransferStateDocument>(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<string, CultMeshCdnArtifactChunk> _chunks;
         private int _requestCount;
+        private int _activeRequests;
+        private int _maxConcurrentRequests;
+        private readonly TaskCompletionSource<bool> _releaseBlockedChunk = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
         public FakeProvider(string providerId, IEnumerable<CultMeshCdnArtifactChunk> chunks)
         {
@@ -278,16 +354,50 @@ public sealed class CultMeshContentTransferTests
         public ConcurrentDictionary<string, int> 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<bool> 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<CultMeshCdnArtifactChunk?> 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;
+            }
         }
     }
 }

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<CultNetSchemaCatalogRequestMessage>((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<ArgumentOutOfRangeException>(() => server.PollAvailableAsync(0));
         }
 
         [Test]