feat(client): add rendezvous traversal coordinators (#12)
quality-gate / quality (push) Successful in 56s
quality-gate / quality (push) Successful in 56s
This commit is contained in:
KyuubiYoru
@@ -0,0 +1,547 @@
|
||||
using System.Net;
|
||||
using FinalFactory.Rendezvous.Contracts;
|
||||
using LiteNetLib;
|
||||
|
||||
namespace FinalFactory.Rendezvous.Client;
|
||||
|
||||
public enum RendezvousHostState
|
||||
{
|
||||
Active = 1,
|
||||
ManagerStopped = 2,
|
||||
Disposed = 3,
|
||||
}
|
||||
|
||||
public sealed class RendezvousHostAttemptCompletedEventArgs(
|
||||
JoinAttemptId attemptId,
|
||||
RendezvousConnectionState state,
|
||||
NetPeer? peer = null) : EventArgs
|
||||
{
|
||||
public JoinAttemptId AttemptId { get; } = attemptId;
|
||||
public RendezvousConnectionState State { get; } = state;
|
||||
public NetPeer? Peer { get; } = peer;
|
||||
}
|
||||
|
||||
public sealed class RendezvousHostCoordinator : IDisposable
|
||||
{
|
||||
private readonly NetManager _manager;
|
||||
private readonly RendezvousNetListener _networkEvents;
|
||||
private readonly EventBasedNatPunchListener _punchEvents;
|
||||
private readonly IPEndPoint _mediator;
|
||||
private readonly PublishedSession _session;
|
||||
private readonly IRendezvousJoinClient _joinClient;
|
||||
private readonly RendezvousCoordinatorOptions _options;
|
||||
private readonly IRendezvousCoordinatorClock _clock;
|
||||
private readonly ConnectionTicketValidator _tickets;
|
||||
private readonly Dictionary<JoinAttemptId, PendingHostAttempt> _attempts = [];
|
||||
private readonly Dictionary<NetPeer, JoinAttemptId> _acceptedPeers = [];
|
||||
private readonly Dictionary<JoinAttemptId, DeferredConnectionRequest> _deferredRequests = [];
|
||||
private readonly Dictionary<JoinAttemptId, DateTimeOffset> _terminalAttempts = [];
|
||||
private readonly Queue<JoinAttemptId> _attemptSchedule = [];
|
||||
private readonly List<JoinAttemptId> _cleanupScratch = [];
|
||||
private HostJoinAttempt[]? _latestSnapshot;
|
||||
private DateTimeOffset _nextPresenceAt = DateTimeOffset.MinValue;
|
||||
private DateTimeOffset _nextTerminalCleanupAt = DateTimeOffset.MinValue;
|
||||
private int _refreshing;
|
||||
private int _polling;
|
||||
private bool _subscriptionsReleased;
|
||||
private int _disposed;
|
||||
|
||||
public RendezvousHostCoordinator(
|
||||
NetManager manager,
|
||||
RendezvousNetListener networkEvents,
|
||||
IPEndPoint mediator,
|
||||
PublishedSession session,
|
||||
IRendezvousJoinClient joinClient,
|
||||
RendezvousCoordinatorOptions? options = null)
|
||||
: this(
|
||||
manager,
|
||||
networkEvents,
|
||||
mediator,
|
||||
session,
|
||||
joinClient,
|
||||
options,
|
||||
new SystemRendezvousCoordinatorClock(),
|
||||
null)
|
||||
{
|
||||
}
|
||||
|
||||
internal RendezvousHostCoordinator(
|
||||
NetManager manager,
|
||||
RendezvousNetListener networkEvents,
|
||||
IPEndPoint mediator,
|
||||
PublishedSession session,
|
||||
IRendezvousJoinClient joinClient,
|
||||
RendezvousCoordinatorOptions? options,
|
||||
IRendezvousCoordinatorClock clock,
|
||||
ConnectionTicketValidator? tickets)
|
||||
{
|
||||
_manager = manager ?? throw new ArgumentNullException(nameof(manager));
|
||||
_networkEvents = networkEvents ?? throw new ArgumentNullException(nameof(networkEvents));
|
||||
_punchEvents = _networkEvents.PunchEvents;
|
||||
_mediator = mediator ?? throw new ArgumentNullException(nameof(mediator));
|
||||
_session = session ?? throw new ArgumentNullException(nameof(session));
|
||||
_joinClient = joinClient ?? throw new ArgumentNullException(nameof(joinClient));
|
||||
_options = (options ?? new RendezvousCoordinatorOptions()).CopyAndValidate();
|
||||
_clock = clock ?? throw new ArgumentNullException(nameof(clock));
|
||||
_tickets = tickets ?? new ConnectionTicketValidator();
|
||||
|
||||
RendezvousManagerGuard.Validate(_manager, _networkEvents);
|
||||
ValidateInputs();
|
||||
_networkEvents.RendezvousConnectionRequest += OnConnectionRequest;
|
||||
_networkEvents.RendezvousPeerConnected += OnPeerConnected;
|
||||
_networkEvents.RendezvousPeerDisconnected += OnPeerDisconnected;
|
||||
_punchEvents.NatIntroductionSuccess += OnNatIntroductionSuccess;
|
||||
}
|
||||
|
||||
public event EventHandler<RendezvousHostAttemptCompletedEventArgs>? AttemptCompleted;
|
||||
|
||||
public RendezvousHostState State { get; private set; } = RendezvousHostState.Active;
|
||||
public int PendingAttemptCount => _attempts.Count;
|
||||
internal int DeferredRequestCount => _deferredRequests.Count;
|
||||
|
||||
public async Task<RendezvousClientResult<int>> RefreshJoinAttemptsAsync(
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
if (Interlocked.Exchange(ref _refreshing, 1) != 0)
|
||||
{
|
||||
throw new InvalidOperationException("A host invitation refresh is already running.");
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
RendezvousClientResult<IReadOnlyList<HostJoinAttempt>> result =
|
||||
await _joinClient.BrowseAllForHostAsync(
|
||||
_session,
|
||||
cancellationToken: cancellationToken).ConfigureAwait(false);
|
||||
if (!result.IsSuccess || result.Value is null)
|
||||
{
|
||||
return RendezvousClientResult.Failure<int>(
|
||||
result.Error,
|
||||
result.Message,
|
||||
result.RetryAfterSeconds);
|
||||
}
|
||||
|
||||
HostJoinAttempt[] snapshot = result.Value.Select(CopyAttempt).ToArray();
|
||||
if (Volatile.Read(ref _disposed) != 0)
|
||||
{
|
||||
throw new ObjectDisposedException(nameof(RendezvousHostCoordinator));
|
||||
}
|
||||
|
||||
Interlocked.Exchange(ref _latestSnapshot, snapshot);
|
||||
if (Volatile.Read(ref _disposed) != 0)
|
||||
{
|
||||
Interlocked.Exchange(ref _latestSnapshot, null);
|
||||
throw new ObjectDisposedException(nameof(RendezvousHostCoordinator));
|
||||
}
|
||||
|
||||
return RendezvousClientResult.Success(snapshot.Length);
|
||||
}
|
||||
finally
|
||||
{
|
||||
Volatile.Write(ref _refreshing, 0);
|
||||
}
|
||||
}
|
||||
|
||||
public void Poll()
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
if (State != RendezvousHostState.Active)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (Interlocked.Exchange(ref _polling, 1) != 0)
|
||||
{
|
||||
throw new InvalidOperationException("The Rendezvous coordinator cannot be polled concurrently or recursively.");
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
ApplySnapshots();
|
||||
if (!_manager.IsRunning)
|
||||
{
|
||||
Stop(RendezvousHostState.ManagerStopped, RendezvousConnectionState.ManagerStopped);
|
||||
return;
|
||||
}
|
||||
|
||||
_manager.NatPunchModule.PollEvents();
|
||||
_manager.PollEvents();
|
||||
_manager.NatPunchModule.PollEvents();
|
||||
if (State != RendezvousHostState.Active)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
DateTimeOffset now = _clock.UtcNow;
|
||||
if (!_manager.IsRunning)
|
||||
{
|
||||
Stop(RendezvousHostState.ManagerStopped, RendezvousConnectionState.ManagerStopped);
|
||||
return;
|
||||
}
|
||||
|
||||
RefreshPresence(now);
|
||||
int checks = Math.Min(
|
||||
_attemptSchedule.Count,
|
||||
_options.MaximumAttemptChecksPerPoll);
|
||||
for (int index = 0; index < checks; index++)
|
||||
{
|
||||
JoinAttemptId attemptId = _attemptSchedule.Dequeue();
|
||||
if (!_attempts.TryGetValue(attemptId, out PendingHostAttempt? attempt))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
if (now >= attempt.Invitation.ExpiresAt)
|
||||
{
|
||||
CompleteAttempt(attemptId, RendezvousConnectionState.TimedOut);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (attempt.State == RendezvousConnectionState.Punching
|
||||
&& attempt.Retry.IsDue(now))
|
||||
{
|
||||
if (attempt.Retry.IsExhausted)
|
||||
{
|
||||
CompleteAttempt(attemptId, RendezvousConnectionState.TimedOut);
|
||||
continue;
|
||||
}
|
||||
|
||||
_manager.NatPunchModule.SendNatIntroduceRequest(
|
||||
_mediator,
|
||||
NatPunchRequestTokenCodec.Encode(
|
||||
NatPunchPeerRole.Host,
|
||||
attempt.Invitation.MediationHandle,
|
||||
attempt.Invitation.HostPunchCapability));
|
||||
attempt.Retry.RecordRequest();
|
||||
}
|
||||
|
||||
_attemptSchedule.Enqueue(attemptId);
|
||||
}
|
||||
|
||||
if (now >= _nextTerminalCleanupAt)
|
||||
{
|
||||
_cleanupScratch.Clear();
|
||||
foreach (KeyValuePair<JoinAttemptId, DateTimeOffset> terminal in _terminalAttempts)
|
||||
{
|
||||
if (terminal.Value <= now)
|
||||
{
|
||||
_cleanupScratch.Add(terminal.Key);
|
||||
}
|
||||
}
|
||||
|
||||
foreach (JoinAttemptId attemptId in _cleanupScratch)
|
||||
{
|
||||
_terminalAttempts.Remove(attemptId);
|
||||
}
|
||||
|
||||
_nextTerminalCleanupAt = now + TimeSpan.FromSeconds(1);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
Volatile.Write(ref _polling, 0);
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (Interlocked.Exchange(ref _disposed, 1) != 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Stop(RendezvousHostState.Disposed, RendezvousConnectionState.Disposed);
|
||||
Interlocked.Exchange(ref _latestSnapshot, null);
|
||||
_attemptSchedule.Clear();
|
||||
_terminalAttempts.Clear();
|
||||
_cleanupScratch.Clear();
|
||||
_tickets.Dispose();
|
||||
}
|
||||
|
||||
public override string ToString() =>
|
||||
$"[RendezvousHostCoordinator {_session.ListingId}; credentials redacted]";
|
||||
|
||||
private void ApplySnapshots()
|
||||
{
|
||||
HostJoinAttempt[]? latest = Interlocked.Exchange(ref _latestSnapshot, null);
|
||||
|
||||
if (latest is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
DateTimeOffset now = _clock.UtcNow;
|
||||
foreach (HostJoinAttempt invitation in latest)
|
||||
{
|
||||
if (invitation.AttemptId.Value == Guid.Empty
|
||||
|| invitation.MediationHandle.Value == Guid.Empty
|
||||
|| !ContractValidation.IsCapabilityValid(invitation.HostPunchCapability)
|
||||
|| !ContractValidation.IsConnectionTicketValid(
|
||||
invitation.ConnectionTicketDigest))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
if (invitation.IsCancelled)
|
||||
{
|
||||
if (_attempts.ContainsKey(invitation.AttemptId))
|
||||
{
|
||||
CompleteAttempt(
|
||||
invitation.AttemptId,
|
||||
RendezvousConnectionState.Cancelled);
|
||||
}
|
||||
|
||||
_terminalAttempts[invitation.AttemptId] = invitation.ExpiresAt;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (invitation.ExpiresAt <= now
|
||||
|| _attempts.ContainsKey(invitation.AttemptId)
|
||||
|| _terminalAttempts.ContainsKey(invitation.AttemptId))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
_attempts.Add(
|
||||
invitation.AttemptId,
|
||||
new PendingHostAttempt(
|
||||
CopyAttempt(invitation),
|
||||
new RendezvousPunchRetrySchedule(_options, _clock)));
|
||||
_attemptSchedule.Enqueue(invitation.AttemptId);
|
||||
}
|
||||
}
|
||||
|
||||
private void RefreshPresence(DateTimeOffset now)
|
||||
{
|
||||
if (now < _nextPresenceAt || now >= _session.ExpiresAt)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_manager.NatPunchModule.SendNatIntroduceRequest(
|
||||
_mediator,
|
||||
NatPunchRequestTokenCodec.Encode(
|
||||
NatPunchPeerRole.HostPresence,
|
||||
_session.HostPresenceHandle,
|
||||
_session.HostPresenceCapability));
|
||||
_nextPresenceAt = now + TimeSpan.FromSeconds(_session.HostPresenceRefreshAfterSeconds);
|
||||
}
|
||||
|
||||
private void OnNatIntroductionSuccess(
|
||||
IPEndPoint target,
|
||||
NatAddressType addressType,
|
||||
string encodedIntroduction)
|
||||
{
|
||||
_ = target;
|
||||
_ = addressType;
|
||||
if (!NatIntroductionTokenCodec.TryDecode(
|
||||
encodedIntroduction,
|
||||
out NatIntroductionToken? introduction)
|
||||
|| introduction is null
|
||||
|| !_attempts.TryGetValue(introduction.AttemptId, out PendingHostAttempt? attempt)
|
||||
|| !NatIntroductionTokenCodec.MatchesDigest(
|
||||
introduction.ConnectionTicket,
|
||||
attempt.Invitation.ConnectionTicketDigest)
|
||||
|| !_tickets.TryAuthorize(
|
||||
introduction.AttemptId,
|
||||
introduction.ConnectionTicket,
|
||||
Min(
|
||||
attempt.Invitation.ExpiresAt,
|
||||
_clock.UtcNow + _options.ConnectionTicketLifetime)))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
attempt.State = RendezvousConnectionState.Connecting;
|
||||
if (_deferredRequests.Remove(
|
||||
introduction.AttemptId,
|
||||
out DeferredConnectionRequest? deferred))
|
||||
{
|
||||
AcceptAuthorizedRequest(
|
||||
introduction.AttemptId,
|
||||
attempt,
|
||||
deferred.Request,
|
||||
deferred.ConnectionTicket);
|
||||
}
|
||||
}
|
||||
|
||||
private void OnConnectionRequest(ConnectionRequest request)
|
||||
{
|
||||
ReadOnlySpan<byte> data = request.Data.GetRemainingBytesSpan();
|
||||
if (!DirectConnectionRequestCodec.IsRendezvousRequest(data))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (!DirectConnectionRequestCodec.TryDecode(data, out DirectConnectionRequest? connection)
|
||||
|| connection is null
|
||||
|| !_attempts.TryGetValue(connection.AttemptId, out PendingHostAttempt? attempt)
|
||||
|| !NatIntroductionTokenCodec.MatchesDigest(
|
||||
connection.ConnectionTicket,
|
||||
attempt.Invitation.ConnectionTicketDigest))
|
||||
{
|
||||
request.RejectForce([]);
|
||||
return;
|
||||
}
|
||||
|
||||
if (attempt.State == RendezvousConnectionState.Punching)
|
||||
{
|
||||
_deferredRequests[connection.AttemptId] = new(
|
||||
request,
|
||||
connection.ConnectionTicket);
|
||||
return;
|
||||
}
|
||||
|
||||
if (attempt.State != RendezvousConnectionState.Connecting)
|
||||
{
|
||||
request.RejectForce([]);
|
||||
return;
|
||||
}
|
||||
|
||||
AcceptAuthorizedRequest(
|
||||
connection.AttemptId,
|
||||
attempt,
|
||||
request,
|
||||
connection.ConnectionTicket);
|
||||
}
|
||||
|
||||
private void OnPeerConnected(NetPeer peer)
|
||||
{
|
||||
if (_acceptedPeers.TryGetValue(peer, out JoinAttemptId attemptId))
|
||||
{
|
||||
CompleteAttempt(attemptId, RendezvousConnectionState.Connected, peer);
|
||||
}
|
||||
}
|
||||
|
||||
private void OnPeerDisconnected(NetPeer peer, DisconnectInfo disconnectInfo)
|
||||
{
|
||||
_ = disconnectInfo;
|
||||
if (_acceptedPeers.TryGetValue(peer, out JoinAttemptId attemptId))
|
||||
{
|
||||
CompleteAttempt(attemptId, RendezvousConnectionState.Rejected);
|
||||
}
|
||||
}
|
||||
|
||||
private void CompleteAttempt(
|
||||
JoinAttemptId attemptId,
|
||||
RendezvousConnectionState state,
|
||||
NetPeer? peer = null)
|
||||
{
|
||||
if (!_attempts.Remove(attemptId, out PendingHostAttempt? attempt))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (attempt.AcceptedPeer is not null)
|
||||
{
|
||||
_acceptedPeers.Remove(attempt.AcceptedPeer);
|
||||
}
|
||||
|
||||
_deferredRequests.Remove(attemptId);
|
||||
_tickets.Revoke(attemptId);
|
||||
_terminalAttempts[attemptId] = attempt.Invitation.ExpiresAt;
|
||||
AttemptCompleted?.Invoke(this, new(attemptId, state, peer));
|
||||
}
|
||||
|
||||
private void Stop(RendezvousHostState hostState, RendezvousConnectionState attemptState)
|
||||
{
|
||||
if (State != RendezvousHostState.Active)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
State = hostState;
|
||||
foreach (JoinAttemptId attemptId in _attempts.Keys.ToArray())
|
||||
{
|
||||
CompleteAttempt(attemptId, attemptState);
|
||||
}
|
||||
|
||||
ReleaseSubscriptions();
|
||||
}
|
||||
|
||||
private void ReleaseSubscriptions()
|
||||
{
|
||||
if (_subscriptionsReleased)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_networkEvents.RendezvousConnectionRequest -= OnConnectionRequest;
|
||||
_networkEvents.RendezvousPeerConnected -= OnPeerConnected;
|
||||
_networkEvents.RendezvousPeerDisconnected -= OnPeerDisconnected;
|
||||
_punchEvents.NatIntroductionSuccess -= OnNatIntroductionSuccess;
|
||||
_subscriptionsReleased = true;
|
||||
}
|
||||
|
||||
private void ValidateInputs()
|
||||
{
|
||||
if (_mediator.Port is < 1 or > 65_535
|
||||
|| _session.HostPresenceHandle.Value == Guid.Empty
|
||||
|| !ContractValidation.IsCapabilityValid(_session.HostPresenceCapability)
|
||||
|| _session.HostPresenceRefreshAfterSeconds < 1
|
||||
|| _session.ExpiresAt <= _clock.UtcNow)
|
||||
{
|
||||
throw new ArgumentException("The host traversal inputs are invalid.");
|
||||
}
|
||||
}
|
||||
|
||||
private static HostJoinAttempt CopyAttempt(HostJoinAttempt attempt) => new()
|
||||
{
|
||||
AttemptId = attempt.AttemptId,
|
||||
MediationHandle = attempt.MediationHandle,
|
||||
HostPunchCapability = attempt.HostPunchCapability,
|
||||
ConnectionTicketDigest = attempt.ConnectionTicketDigest,
|
||||
IsCancelled = attempt.IsCancelled,
|
||||
ExpiresAt = attempt.ExpiresAt,
|
||||
};
|
||||
|
||||
private static DateTimeOffset Min(DateTimeOffset left, DateTimeOffset right) =>
|
||||
left <= right ? left : right;
|
||||
|
||||
private void AcceptAuthorizedRequest(
|
||||
JoinAttemptId attemptId,
|
||||
PendingHostAttempt attempt,
|
||||
ConnectionRequest request,
|
||||
string connectionTicket)
|
||||
{
|
||||
ConnectionTicketConsumptionResult consumption = _tickets.Consume(
|
||||
attemptId,
|
||||
connectionTicket);
|
||||
if (consumption != ConnectionTicketConsumptionResult.Accepted)
|
||||
{
|
||||
request.RejectForce([]);
|
||||
return;
|
||||
}
|
||||
|
||||
NetPeer peer = request.Accept();
|
||||
attempt.AcceptedPeer = peer;
|
||||
_acceptedPeers[peer] = attemptId;
|
||||
}
|
||||
|
||||
private void ThrowIfDisposed()
|
||||
{
|
||||
if (Volatile.Read(ref _disposed) != 0)
|
||||
{
|
||||
throw new ObjectDisposedException(nameof(RendezvousHostCoordinator));
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class PendingHostAttempt(
|
||||
HostJoinAttempt invitation,
|
||||
RendezvousPunchRetrySchedule retry)
|
||||
{
|
||||
internal HostJoinAttempt Invitation { get; } = invitation;
|
||||
internal RendezvousPunchRetrySchedule Retry { get; } = retry;
|
||||
internal RendezvousConnectionState State { get; set; } = RendezvousConnectionState.Punching;
|
||||
internal NetPeer? AcceptedPeer { get; set; }
|
||||
}
|
||||
|
||||
private sealed class DeferredConnectionRequest(
|
||||
ConnectionRequest request,
|
||||
string connectionTicket)
|
||||
{
|
||||
internal ConnectionRequest Request { get; } = request;
|
||||
internal string ConnectionTicket { get; } = connectionTicket;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user