voice-cat/tests/VoiceCat.Tests/MediaFanoutTests.cs

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, out _)) 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.";
        }
    }
}