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

using VoiceCat.Protocol;
using System.Net.Sockets;
using VoiceCat.Server;
using Voicecat.V1;
using static VoiceCat.Tests.ServerTests;
using static VoiceCat.Tests.MediaRelayTests;

namespace VoiceCat.Tests;

public sealed class ReaperTests
{
    private static readonly VoiceServerOptions Options = new()
    {
        IdleTimeout = TimeSpan.FromSeconds(10), ReaperInterval = TimeSpan.FromMilliseconds(20)
    };

    [Fact]
    public async Task SilentPeerIsReapedWhileTcpActivityKeepsObserverAlive()
    {
        var clock = new ManualClock();
        await using var fixture = new ServerFixture(options: Options, timeProvider: clock);
        await using var alice = await fixture.ConnectAsync();
        await alice.LoginAsync("Alice");
        await using var bob = await fixture.ConnectAsync();
        User self = await bob.LoginAsync("Bob");
        clock.Advance(9);
        alice.Send(new() { Ping = new() { Nonce = 99 } });
        await alice.ReadUntilAsync(e => e.Pong?.Nonce == 99);
        clock.Advance(2);
        Assert.Equal(self.Id, (await alice.ReadUntilAsync(e => e.UserEvent?.Kind == UserEvent.Types.Kind.Left)).UserEvent.LeftId);
        Assert.Equal("Receive idle timeout.", (await bob.ReadUntilAsync(e => e.Disconnect is not null)).Disconnect.Reason);
        alice.Send(new() { Subscribe = new() });
        int additionalDepartures = 0;
        var snapshot = await alice.ReadUntilAsync(e =>
        {
            if (e.UserEvent?.Kind == UserEvent.Types.Kind.Left) additionalDepartures++;
            return e.ServerState is not null;
        });
        Assert.Equal(0, additionalDepartures);
        Assert.DoesNotContain(snapshot.ServerState.Users, user => user.Id == self.Id);
        alice.Send(new() { Ping = new() { Nonce = 100 } });
        await alice.ReadUntilAsync(e => e.Pong?.Nonce == 100);
    }

    [Theory]
    [InlineData(true)]
    [InlineData(false)]
    public async Task ValidUdpActivityKeepsTcpIdleClientAlive(bool voice)
    {
        var clock = new ManualClock();
        await using var fixture = new ServerFixture(options: Options, timeProvider: clock);
        await using var alice = await VoicePeer.ConnectAsync(fixture, "Alice");
        await using var bob = await VoicePeer.ConnectAsync(fixture, "Bob");
        uint ssrc = voice ? (await alice.AnnounceAsync(StreamKind.StreamMic)).Ssrc : 0;
        clock.Advance(9);
        if (voice)
        {
            await alice.SendAsync(alice.Seal(ssrc, [1, 2, 3]));
            await bob.ReceiveVoiceAsync();
        }
        else
        {
            byte[] keepalive = new byte[VoiceFrameHeader.Size];
            new VoiceFrameHeader(MediaFrameType.Keepalive, 0, 0, 0, 0, 0).Write(keepalive);
            await alice.SendAsync(keepalive);
            Assert.Equal(keepalive, await alice.ReceivePacketAsync());
        }
        clock.Advance(2);
        Assert.Equal(bob.Client.Authentication!.Self.Id,
            (await alice.Client.ReadUntilAsync(e => e.UserEvent?.Kind == UserEvent.Types.Kind.Left)).UserEvent.LeftId);
        alice.Client.Send(new() { Ping = new() { Nonce = 42 } });
        await alice.Client.ReadUntilAsync(e => e.Pong?.Nonce == 42);
    }

    [Fact]
    public async Task InvalidVoiceCannotKeepSilentSessionAlive()
    {
        var clock = new ManualClock();
        await using var fixture = new ServerFixture(options: Options, timeProvider: clock);
        await using var alice = await VoicePeer.ConnectAsync(fixture, "Alice");
        await using var bob = await VoicePeer.ConnectAsync(fixture, "Bob");
        var stream = await bob.AnnounceAsync(StreamKind.StreamMic);
        clock.Advance(9);
        byte[] forged = bob.Seal(stream.Ssrc, [1]);
        forged[^1] ^= 1;
        await bob.SendAsync(forged);
        alice.Client.Send(new() { Ping = new() { Nonce = 1 } });
        await alice.Client.ReadUntilAsync(e => e.Pong is not null);
        clock.Advance(2);
        Assert.Equal(bob.Client.Authentication!.Self.Id,
            (await alice.Client.ReadUntilAsync(e => e.UserEvent?.Kind == UserEvent.Types.Kind.Left)).UserEvent.LeftId);
    }

    [Fact]
    public async Task ShutdownAwaitsActiveVoiceAndUnfinishedHandshake()
    {
        await using var fixture = new ServerFixture(options: Options);
        await using var alice = await VoicePeer.ConnectAsync(fixture, "Alice");
        await using var bob = await VoicePeer.ConnectAsync(fixture, "Bob");
        var stream = await alice.AnnounceAsync(StreamKind.StreamMic);
        await alice.SendAsync(alice.Seal(stream.Ssrc, [1, 2]));
        await bob.ReceiveVoiceAsync();
        using var unfinished = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        await unfinished.ConnectAsync(fixture.Server.EndPoint);
        await fixture.Server.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(10));
        await fixture.Server.DisposeAsync();
    }

    [Fact]
    public async Task ReaperCanBeDisabled()
    {
        var clock = new ManualClock();
        await using var fixture = new ServerFixture(options: Options with { IdleTimeout = TimeSpan.Zero, ReaperInterval = TimeSpan.Zero }, timeProvider: clock);
        await using var client = await fixture.ConnectAsync();
        await client.LoginAsync("Alice");
        clock.Advance(1000);
        await Task.Delay(100);
        client.Send(new() { Ping = new() { Nonce = 1 } });
        await client.ReadUntilAsync(e => e.Pong is not null);
    }

    private sealed class ManualClock : TimeProvider
    {
        private long timestamp;
        public override long TimestampFrequency => TimeSpan.TicksPerSecond;
        public override long GetTimestamp() => Volatile.Read(ref timestamp);
        public void Advance(int seconds) => Interlocked.Add(ref timestamp, seconds * TimeSpan.TicksPerSecond);
    }
}