Retire legacy implementations and flatten managed layout

tests/VoiceCat.Tests/MediaFanoutTests.cs

@@ -0,0 +1,110 @@
+using System.Diagnostics;
+using System.Net;
+using System.Net.Sockets;
+using System.Security.Cryptography;
+using VoiceCat.Crypto;
+using VoiceCat.Protocol;
+using VoiceCat.Server.Transport;
+using Xunit.Abstractions;
+
+namespace VoiceCat.Tests;
+
+public sealed class MediaFanoutTests(ITestOutputHelper output)
+{
+    [Fact]
+    public async Task UdpRelayDeliversFiftyPacketsPerSecondToFiftySubscribers()
+    {
+        await using var relay = new MediaRelay(new(IPAddress.Loopback, 0));
+        byte[] key = Enumerable.Range(0, 32).Select(i => (byte)i).ToArray();
+        Socket[] sockets = Enumerable.Range(0, 51).Select(_ => new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp)).ToArray();
+        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(20));
+        try
+        {
+            foreach (Socket socket in sockets)
+            {
+                socket.ReceiveBufferSize = 1024 * 1024;
+                socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+            }
+            MediaRoute[] routes = sockets.Select(socket => new MediaRoute(
+                new(new byte[16], new(new(key), new(key))) { Endpoint = ((IPEndPoint)socket.LocalEndPoint!).Serialize() },
+                1, true, false, false, [42])).ToArray();
+            relay.Publish(routes);
+            byte[] payload = Enumerable.Range(0, 120).Select(i => (byte)i).ToArray();
+            Task[] receivers = sockets.Skip(1).Select(async socket =>
+            {
+                using var decryptor = new MediaDecryptor(key);
+                byte[] packet = new byte[4000];
+                byte[] decoded = new byte[4000];
+                for (ulong sequence = 0; sequence < 50; sequence++)
+                {
+                    int length = await socket.ReceiveAsync(packet, SocketFlags.None, timeout.Token);
+                    Assert.True(decryptor.TryDecrypt(packet.AsSpan(0, length), decoded, out var header, out int bytes));
+                    Assert.Equal(sequence, header.Sequence);
+                    Assert.Equal(payload, decoded[..bytes]);
+                }
+            }).ToArray();
+            using var encryptor = new MediaEncryptor(key);
+            byte[] outgoing = new byte[VoiceFrameHeader.Size + payload.Length + 16];
+            var elapsed = Stopwatch.StartNew();
+            for (uint index = 0; index < 50; index++)
+            {
+                encryptor.Encrypt(new(MediaFrameType.Voice, 0, 0, 42, 0, index * 960), payload, outgoing);
+                await sockets[0].SendToAsync(outgoing, SocketFlags.None, relay.EndPoint, timeout.Token);
+                await Task.Delay(20, timeout.Token);
+            }
+            await Task.WhenAll(receivers);
+            output.WriteLine($"Delivered all 2,500 recipient packets in {elapsed.Elapsed.TotalMilliseconds:F1} ms at a paced 50 pps input.");
+        }
+        finally { foreach (Socket socket in sockets) socket.Dispose(); }
+    }
+
+    [PlatformCipherFact]
+    public void FiftySubscriberFanoutAllocatesNoManagedMemoryAndPreservesPayload()
+    {
+        byte[] key = Enumerable.Range(0, 32).Select(i => (byte)i).ToArray();
+        MediaRoute[] routes = Enumerable.Range(0, 51).Select(i => new MediaRoute(
+            new(new byte[16], new(new(key), new(key))) { Endpoint = new IPEndPoint(IPAddress.Loopback, 10000 + i).Serialize() },
+            1, true, false, false, [42])).ToArray();
+        using var sender = new MediaEncryptor(key);
+        using var receiver = new MediaDecryptor(key);
+        using var fanout = new MediaFanout();
+        byte[] payload = Enumerable.Range(0, 120).Select(i => (byte)i).ToArray();
+        byte[] packet = new byte[VoiceFrameHeader.Size + payload.Length + 16];
+        byte[] decoded = new byte[payload.Length];
+        ReadOnlyMemory<byte> last = default;
+        try
+        {
+            for (int i = 0; i < 100; i++) Cycle();
+            long before = GC.GetAllocatedBytesForCurrentThread();
+            long started = Stopwatch.GetTimestamp();
+            for (int i = 0; i < 1000; i++) Cycle();
+            TimeSpan elapsed = Stopwatch.GetElapsedTime(started);
+            long allocated = GC.GetAllocatedBytesForCurrentThread() - before;
+            Assert.Equal(0, allocated);
+            Assert.True(receiver.TryDecrypt(last.Span, decoded, out var header, out int length));
+            Assert.Equal(payload.Length, length);
+            Assert.Equal(payload, decoded);
+            Assert.Equal(42U, header.Ssrc);
+            Assert.Equal(1099UL, header.Sequence);
+            output.WriteLine($"50,000 recipient seals in {elapsed.TotalMilliseconds:F1} ms; {allocated} managed bytes. Transport scheduling is excluded.");
+        }
+        finally { foreach (MediaRoute route in routes) route.Peer.Dispose(); }
+
+        void Cycle()
+        {
+            sender.Encrypt(new(MediaFrameType.Voice, 0, 0, 42, 0, 960), payload, packet);
+            if (!fanout.TryStart(packet, routes[0], routes)) throw new InvalidOperationException("Valid packet rejected.");
+            int recipients = 0;
+            while (fanout.TryNext(out var next, out _)) { last = next; recipients++; }
+            if (recipients != 50) throw new InvalidOperationException("Incorrect fanout.");
+        }
+    }
+
+    private sealed class PlatformCipherFactAttribute : FactAttribute
+    {
+        public PlatformCipherFactAttribute()
+        {
+            if (!ChaCha20Poly1305.IsSupported) Skip = "The allocation guarantee requires platform ChaCha20-Poly1305; fallback conformance is tested separately.";
+        }
+    }
+}