Talon/voice-cat
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Add configurable media-aware managed session reaping

Browse Source
This commit is contained in:
Talon
2026-09-15 22:58:16 +02:00
parent 05eacb3092
commit 274b85025c
14 changed files with 291 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/README.md
+6 -2
View File
@@ -140,8 +140,12 @@ the optional `VOICECAT_BUILD_DOTNET_ORACLE=ON` configure flag and a real-deps bu
The server also advertises UDP on the TCP port number, supports voice subscription
and stream signaling, and reseals encoded audio for subscribers in the same channel.
UDP binding fixes the first endpoint for the session; reconnect after endpoint changes.
Protected joins, administration, moderation, production configuration and full
media-aware reaping remain before Phase 4 completion.
Protected joins, administration, moderation and production configuration remain
before Phase 4 completion. The server's media-aware reaper defaults to 45 seconds
of inactivity with a 15-second sweep. Parsed control envelopes, valid encrypted
voice and keepalives from bound endpoints refresh activity; invalid media does not.
`VoiceServerOptions` configures timeouts and capacity; zero idle timeout disables
reaping. The constructor overload accepts `TimeProvider` for deterministic expiry tests.
Enable deterministic native voice interoperability (no audio hardware required):
dotnet/src/VoiceCat.Server/Transport/MediaRelay.cs
+8 -2
View File
@@ -7,10 +7,11 @@ using VoiceCat.Protocol;
namespace VoiceCat.Server.Transport;
internal sealed class MediaPeer(byte[] token, MediaSessionCrypto crypto)
internal sealed class MediaPeer(byte[] token, MediaSessionCrypto crypto, SessionActivity? activity = null)
{
public byte[] Token { get; } = token;
public MediaSessionCrypto Crypto { get; } = crypto;
public SessionActivity Activity { get; } = activity ?? new(TimeProvider.System);
// Only the UDP loop reads or changes the endpoint and binding state.
public SocketAddress? Endpoint { get; set; }
public void Dispose() { Crypto.Dispose(); CryptographicOperations.ZeroMemory(Token); }
@@ -101,10 +102,15 @@ internal sealed class MediaRelay : IAsyncDisposable
if (source is null) continue;
if (header.Type == MediaFrameType.Keepalive)
{
if (length == VoiceFrameHeader.Size) await SendAsync(input.AsMemory(0, length), sender).ConfigureAwait(false);
if (length == VoiceFrameHeader.Size)
{
source.Peer.Activity.Touch();
await SendAsync(input.AsMemory(0, length), sender).ConfigureAwait(false);
}
continue;
}
if (!fanout.TryStart(input.AsSpan(0, length), source, current)) continue;
source.Peer.Activity.Touch();
while (fanout.TryNext(out ReadOnlyMemory<byte> packet, out SocketAddress? endpoint))
await SendAsync(packet, endpoint!).ConfigureAwait(false);
}
dotnet/src/VoiceCat.Server/Transport/SessionActivity.cs
+18
View File
@@ -0,0 +1,18 @@
namespace VoiceCat.Server.Transport;
internal sealed class SessionActivity(TimeProvider clock)
{
private long lastSeen = clock.GetTimestamp();
public void Touch()
{
long now = clock.GetTimestamp();
long previous = Volatile.Read(ref lastSeen);
while (now > previous)
{
long observed = Interlocked.CompareExchange(ref lastSeen, now, previous);
if (observed == previous) return;
previous = observed;
}
}
public bool IsExpired(TimeSpan timeout) => clock.GetElapsedTime(Volatile.Read(ref lastSeen)) >= timeout;
}
dotnet/src/VoiceCat.Server/Transport/TlsControlConnection.cs
+3 -3
View File
@@ -27,12 +27,12 @@ internal sealed class TlsControlConnection : IAsyncDisposable
public Task Completion { get; }
public CancellationToken CancellationToken => lifetime.Token;
internal TlsControlConnection(Socket socket, TlsSession tls, CancellationToken cancellationToken)
internal TlsControlConnection(Socket socket, TlsSession tls, CancellationToken cancellationToken, TimeSpan? handshakeTimeout = null)
{
this.socket = socket;
this.tls = tls;
lifetime = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
lifetime.CancelAfter(TimeSpan.FromSeconds(15));
lifetime.CancelAfter(handshakeTimeout ?? TimeSpan.FromSeconds(15));
Completion = RunAsync();
}
@@ -94,8 +94,8 @@ internal sealed class TlsControlConnection : IAsyncDisposable
try { mediaCrypto = new(encryptor, tls.CreateMediaDecryptor()); }
catch { encryptor.Dispose(); throw; }
mediaReady.SetResult();
lifetime.CancelAfter(Timeout.InfiniteTimeSpan);
}
if (tls.IsReady) lifetime.CancelAfter(TimeSpan.FromSeconds(60));
while ((count = tls.ReadPlaintext(plaintext)) > 0) Parse(plaintext.AsSpan(0, count));
await FlushAsync(sendBuffer, cancellationToken).ConfigureAwait(false);
receive = socket.ReceiveAsync(ciphertext, SocketFlags.None, cancellationToken).AsTask();
dotnet/src/VoiceCat.Server/VoiceServer.cs
+56 -14
View File
@@ -19,6 +19,8 @@ public sealed class VoiceServer : IAsyncDisposable
private readonly IReadOnlyList<Voicecat.V1.Channel> channels;
private readonly bool allowGuests;
private readonly string name;
private readonly VoiceServerOptions options;
private readonly TimeProvider clock;
private readonly CancellationTokenSource shutdown = new();
private readonly object gate = new();
private readonly Dictionary<ulong, Session> sessions = [];
@@ -27,6 +29,7 @@ public sealed class VoiceServer : IAsyncDisposable
private uint nextUser;
private uint nextSsrc;
private readonly Task accepting;
private readonly Task reaping;
private int disposed;
public IPEndPoint EndPoint => (IPEndPoint)listener.LocalEndPoint!;
@@ -34,9 +37,16 @@ public sealed class VoiceServer : IAsyncDisposable
public event Action<Exception>? ConnectionFailed;
public VoiceServer(string directory, IPEndPoint endpoint, bool allowGuests = true, string name = "VoiceCat Server")
: this(directory, endpoint, new VoiceServerOptions { AllowGuests = allowGuests, Name = name }) { }
public VoiceServer(string directory, IPEndPoint endpoint, VoiceServerOptions options, TimeProvider? timeProvider = null)
{
this.allowGuests = allowGuests;
this.name = name;
ArgumentNullException.ThrowIfNull(options);
options.Validate();
this.options = options;
clock = timeProvider ?? TimeProvider.System;
allowGuests = options.AllowGuests;
name = options.Name;
credentials = ServerCredentials.LoadOrCreate(directory, name);
try
{
@@ -44,7 +54,7 @@ public sealed class VoiceServer : IAsyncDisposable
channels = accounts.LoadChannels();
listener = new Socket(endpoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
listener.Bind(endpoint);
listener.Listen(64);
listener.Listen(options.MaximumConnections);
media = new((IPEndPoint)listener.LocalEndPoint!);
media.Failed += exception => ConnectionFailed?.Invoke(exception);
}
@@ -57,6 +67,7 @@ public sealed class VoiceServer : IAsyncDisposable
throw;
}
accepting = AcceptAsync();
reaping = ReapAsync();
}
private async Task AcceptAsync()
@@ -68,11 +79,11 @@ public sealed class VoiceServer : IAsyncDisposable
Socket socket = await listener.AcceptAsync(shutdown.Token).ConfigureAwait(false);
lock (gate)
{
if (sessions.Count >= 64) { socket.Dispose(); continue; }
if (sessions.Count >= options.MaximumConnections) { socket.Dispose(); continue; }
socket.NoDelay = true;
string address = ((IPEndPoint)socket.RemoteEndPoint!).Address.ToString();
var connection = new TlsControlConnection(socket, credentials.CreateTlsSession(), shutdown.Token);
var session = new Session(++nextSession, connection, address);
var connection = new TlsControlConnection(socket, credentials.CreateTlsSession(), shutdown.Token, options.HandshakeTimeout);
var session = new Session(++nextSession, connection, address, new(clock));
sessions.Add(session.Id, session);
connections.RemoveAll(task => task.IsCompleted);
connections.Add(HandleAsync(session));
@@ -88,6 +99,7 @@ public sealed class VoiceServer : IAsyncDisposable
{
await foreach (Envelope envelope in session.Connection.ReadAsync(shutdown.Token).ConfigureAwait(false))
{
session.Activity.Touch();
if (envelope.Ping is not null)
{
session.Connection.TrySend(new() { RequestId = envelope.RequestId, Pong = new() { Nonce = envelope.Ping.Nonce } });
@@ -101,7 +113,7 @@ public sealed class VoiceServer : IAsyncDisposable
Reject(session, "Unsupported protocol version or banned address.");
break;
}
session.Media = new(RandomNumberGenerator.GetBytes(16), await session.Connection.TakeMediaCryptoAsync(shutdown.Token).ConfigureAwait(false));
session.Media = new(RandomNumberGenerator.GetBytes(16), await session.Connection.TakeMediaCryptoAsync(shutdown.Token).ConfigureAwait(false), session.Activity);
var hello = new ServerHello { ProtoVersion = 2, ServerName = name, ServerVersion = "0.1.0-dotnet", UdpPort = checked((uint)media.EndPoint.Port), ServerIdentityFingerprint = ByteString.CopyFrom(SHA256.HashData(credentials.Identity.PublicKey)) };
if (allowGuests) hello.AuthMethods.Add("guest");
hello.AuthMethods.Add("password");
@@ -159,6 +171,28 @@ public sealed class VoiceServer : IAsyncDisposable
session.Connection.CompleteWrites();
}
private async Task ReapAsync()
{
if (options.IdleTimeout == TimeSpan.Zero) return;
using var timer = new PeriodicTimer(options.ReaperInterval, clock);
try
{
while (await timer.WaitForNextTickAsync(shutdown.Token).ConfigureAwait(false))
{
lock (gate)
{
foreach (Session session in sessions.Values)
{
if (session.Closing || !session.Activity.IsExpired(options.IdleTimeout)) continue;
session.Closing = true;
Reject(session, "Receive idle timeout.");
}
}
}
}
catch (OperationCanceledException) when (shutdown.IsCancellationRequested) { }
}
private async Task AuthenticateAsync(Session session, ulong requestId, AuthRequest request)
{
User? user = null;
@@ -321,23 +355,31 @@ public sealed class VoiceServer : IAsyncDisposable
listener.Dispose();
try
{
await accepting.ConfigureAwait(false);
Task[] pending;
lock (gate) pending = connections.ToArray();
await Task.WhenAll(pending).ConfigureAwait(false);
await Task.WhenAll(accepting, reaping).ConfigureAwait(false);
}
finally
{
try { await media.DisposeAsync().ConfigureAwait(false); }
finally { accounts.Dispose(); credentials.Dispose(); shutdown.Dispose(); }
try
{
Task[] pending;
lock (gate) pending = connections.ToArray();
await Task.WhenAll(pending).ConfigureAwait(false);
}
finally
{
try { await media.DisposeAsync().ConfigureAwait(false); }
finally { accounts.Dispose(); credentials.Dispose(); shutdown.Dispose(); }
}
}
}
private sealed class Session(ulong id, TlsControlConnection connection, string address)
private sealed class Session(ulong id, TlsControlConnection connection, string address, SessionActivity activity)
{
public ulong Id { get; } = id;
public TlsControlConnection Connection { get; } = connection;
public string Address { get; } = address;
public SessionActivity Activity { get; } = activity;
public bool Closing { get; set; }
public bool HelloReceived { get; set; }
public User? User { get; set; }
public MediaPeer? Media { get; set; }
dotnet/src/VoiceCat.Server/VoiceServerOptions.cs
+21
View File
@@ -0,0 +1,21 @@
namespace VoiceCat.Server;
public sealed record VoiceServerOptions
{
public string Name { get; init; } = "VoiceCat Server";
public bool AllowGuests { get; init; } = true;
public int MaximumConnections { get; init; } = 64;
public TimeSpan HandshakeTimeout { get; init; } = TimeSpan.FromSeconds(15);
public TimeSpan IdleTimeout { get; init; } = TimeSpan.FromSeconds(45);
public TimeSpan ReaperInterval { get; init; } = TimeSpan.FromSeconds(15);
internal void Validate()
{
ArgumentException.ThrowIfNullOrWhiteSpace(Name);
ArgumentOutOfRangeException.ThrowIfLessThan(MaximumConnections, 1);
if (HandshakeTimeout <= TimeSpan.Zero || HandshakeTimeout.TotalMilliseconds > uint.MaxValue - 1) throw new ArgumentOutOfRangeException(nameof(HandshakeTimeout));
if (IdleTimeout < TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(IdleTimeout));
if (ReaperInterval < TimeSpan.Zero || ReaperInterval.TotalMilliseconds > uint.MaxValue - 1 || IdleTimeout > TimeSpan.Zero && ReaperInterval == TimeSpan.Zero)
throw new ArgumentOutOfRangeException(nameof(ReaperInterval));
}
}
dotnet/tests/VoiceCat.Tests/ReaperTests.cs
+129
View File
@@ -0,0 +1,129 @@
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);
}
}
dotnet/tests/VoiceCat.Tests/ServerTests.cs
+2 -2
View File
@@ -136,10 +136,10 @@ public sealed class ServerTests
public string Directory { get; } = Path.Combine(Path.GetTempPath(), "voicecat-server-" + Guid.NewGuid().ToString("N"));
public VoiceServer Server { get; }
private readonly string fingerprint;
public ServerFixture(bool guests = true)
public ServerFixture(bool guests = true, VoiceServerOptions? options = null, TimeProvider? timeProvider = null)
{
System.IO.Directory.CreateDirectory(Directory);
Server = new(Directory, new(IPAddress.Loopback, 0), guests);
Server = new(Directory, new(IPAddress.Loopback, 0), options ?? new() { AllowGuests = guests }, timeProvider);
using var credentials = ServerCredentials.LoadOrCreate(Directory, "VoiceCat Server");
fingerprint = credentials.CertificateFingerprint;
}