Add gateway retries and rate limiter

Documentation/guides/advanced/Voice/VoiceModule.cs

@@ -34,9 +34,6 @@ public async Task PlayAsync(string track)
         // Connect
         await voiceClient.StartAsync();
 
-        // Wait for ready
-        await voiceClient.ReadyAsync;
-
         // Enter speaking state, to be able to send voice
         await voiceClient.EnterSpeakingStateAsync(SpeakingFlags.Microphone);
 
@@ -124,9 +121,6 @@ public async Task EchoAsync()
         // Connect
         await voiceClient.StartAsync();
 
-        // Wait for ready
-        await voiceClient.ReadyAsync;
-
         // Enter speaking state, to be able to send voice
         await voiceClient.EnterSpeakingStateAsync(SpeakingFlags.Microphone);
 

Documentation/guides/advanced/voice.md

@@ -10,7 +10,7 @@ Follow the [installation guide](installing-native-dependencies.md) to install th
 > In the following examples streams and @NetCord.Gateway.Voice.VoiceClient instances are not disposed because they should be stored somewhere and disposed later.
 
 ### Sending Voice
-[!code-cs[VoiceModule.cs](Voice/VoiceModule.cs#L12-L102)]
+[!code-cs[VoiceModule.cs](Voice/VoiceModule.cs#L12-L99)]
 
 ### Receiving Voice
-[!code-cs[VoiceModule.cs](Voice/VoiceModule.cs#L104-L146)]
+[!code-cs[VoiceModule.cs](Voice/VoiceModule.cs#L101-L140)]

NetCord/Gateway/CancellationTokenProvider.cs

@@ -4,6 +4,8 @@ internal sealed class CancellationTokenProvider : IDisposable
 {
     private readonly CancellationTokenSource _source;
 
+    public bool IsCancellationRequested => _source.IsCancellationRequested;
+
     public CancellationToken Token { get; }
 
     public CancellationTokenProvider()

NetCord/Gateway/GatewayClient.cs

@@ -9,7 +9,7 @@
 namespace NetCord.Gateway;
 
 /// <summary>
-/// The GatewayClient class allows applications to send and receive data from the Discord Gateway, such as events and resource requests, via a WebSocket client.
+/// The <see cref="GatewayClient"/> class allows applications to send and receive data from the Discord Gateway, such as events and resource requests.
 /// </summary>
 public partial class GatewayClient : WebSocketClient, IEntity
 {
@@ -370,7 +370,7 @@ public partial class GatewayClient : WebSocketClient, IEntity
     public event Func<GuildUser, ValueTask>? GuildUserUpdate;
 
     /// <summary>
-    /// Sent in response to <see cref="RequestGuildUsersAsync(GuildUsersRequestProperties, CancellationToken)"/>. You can use the <see cref="GuildUserChunkEventArgs.ChunkIndex"/> and <see cref="GuildUserChunkEventArgs.ChunkCount"/> to calculate how many chunks are left for your request.<br/>
+    /// Sent in response to <see cref="RequestGuildUsersAsync(GuildUsersRequestProperties, WebSocketPayloadProperties, CancellationToken)"/>. You can use the <see cref="GuildUserChunkEventArgs.ChunkIndex"/> and <see cref="GuildUserChunkEventArgs.ChunkCount"/> to calculate how many chunks are left for your request.<br/>
     /// </summary>
     /// <remarks>
     /// <br/> Required Intents: None
@@ -836,7 +836,7 @@ private protected override void OnConnected()
         _compression.Initialize();
     }
 
-    private ValueTask SendIdentifyAsync(PresenceProperties? presence = null, CancellationToken cancellationToken = default)
+    private ValueTask SendIdentifyAsync(ConnectionState connectionState, PresenceProperties? presence = null, CancellationToken cancellationToken = default)
     {
         var serializedPayload = new GatewayPayloadProperties<GatewayIdentifyProperties>(GatewayOpcode.Identify, new(Token.RawToken)
         {
@@ -847,7 +847,7 @@ private ValueTask SendIdentifyAsync(PresenceProperties? presence = null, Cancell
             Intents = _configuration.Intents,
         }).Serialize(Serialization.Default.GatewayPayloadPropertiesGatewayIdentifyProperties);
         _latencyTimer.Start();
-        return SendPayloadAsync(serializedPayload, cancellationToken);
+        return SendConnectionPayloadAsync(connectionState, serializedPayload, _internalPayloadProperties, cancellationToken);
     }
 
     /// <summary>
@@ -858,8 +858,8 @@ private ValueTask SendIdentifyAsync(PresenceProperties? presence = null, Cancell
     /// <returns></returns>
     public async Task StartAsync(PresenceProperties? presence = null, CancellationToken cancellationToken = default)
     {
-        await StartAsync(cancellationToken).ConfigureAwait(false);
-        await SendIdentifyAsync(presence, cancellationToken).ConfigureAwait(false);
+        var connectionState = await StartAsync(cancellationToken).ConfigureAwait(false);
+        await SendIdentifyAsync(connectionState, presence, cancellationToken).ConfigureAwait(false);
     }
 
     /// <summary>
@@ -871,43 +871,43 @@ public async Task StartAsync(PresenceProperties? presence = null, CancellationTo
     /// <returns></returns>
     public async Task ResumeAsync(string sessionId, int sequenceNumber, CancellationToken cancellationToken = default)
     {
-        await ConnectAsync(cancellationToken).ConfigureAwait(false);
-        await TryResumeAsync(SessionId = sessionId, SequenceNumber = sequenceNumber, cancellationToken).ConfigureAwait(false);
+        var connectionState = await StartAsync(cancellationToken).ConfigureAwait(false);
+        await TryResumeAsync(connectionState, SessionId = sessionId, SequenceNumber = sequenceNumber, cancellationToken).ConfigureAwait(false);
     }
 
     private protected override bool Reconnect(WebSocketCloseStatus? status, string? description)
         => status is not ((WebSocketCloseStatus)4004 or (WebSocketCloseStatus)4010 or (WebSocketCloseStatus)4011 or (WebSocketCloseStatus)4012 or (WebSocketCloseStatus)4013 or (WebSocketCloseStatus)4014);
 
-    private protected override ValueTask TryResumeAsync(CancellationToken cancellationToken = default)
+    private protected override ValueTask TryResumeAsync(ConnectionState connectionState, CancellationToken cancellationToken = default)
     {
-        return TryResumeAsync(SessionId!, SequenceNumber, cancellationToken);
+        return TryResumeAsync(connectionState, SessionId!, SequenceNumber, cancellationToken);
     }
 
-    private ValueTask TryResumeAsync(string sessionId, int sequenceNumber, CancellationToken cancellationToken = default)
+    private ValueTask TryResumeAsync(ConnectionState connectionState, string sessionId, int sequenceNumber, CancellationToken cancellationToken = default)
     {
         var serializedPayload = new GatewayPayloadProperties<GatewayResumeProperties>(GatewayOpcode.Resume, new(Token.RawToken, sessionId, sequenceNumber)).Serialize(Serialization.Default.GatewayPayloadPropertiesGatewayResumeProperties);
         _latencyTimer.Start();
-        return SendPayloadAsync(serializedPayload, cancellationToken);
+        return SendConnectionPayloadAsync(connectionState, serializedPayload, _internalPayloadProperties, cancellationToken);
     }
 
-    private protected override ValueTask HeartbeatAsync(CancellationToken cancellationToken = default)
+    private protected override ValueTask HeartbeatAsync(ConnectionState connectionState, CancellationToken cancellationToken = default)
     {
         var serializedPayload = new GatewayPayloadProperties<int>(GatewayOpcode.Heartbeat, SequenceNumber).Serialize(Serialization.Default.GatewayPayloadPropertiesInt32);
         _latencyTimer.Start();
-        return SendPayloadAsync(serializedPayload, cancellationToken);
+        return SendConnectionPayloadAsync(connectionState, serializedPayload, _internalPayloadProperties, cancellationToken);
     }
 
     private protected override JsonPayload CreatePayload(ReadOnlyMemory<byte> payload) => JsonSerializer.Deserialize(_compression.Decompress(payload).Span, Serialization.Default.JsonPayload)!;
 
-    private protected override async Task ProcessPayloadAsync(JsonPayload payload)
+    private protected override async Task ProcessPayloadAsync(State state, ConnectionState connectionState, JsonPayload payload)
     {
         switch ((GatewayOpcode)payload.Opcode)
         {
             case GatewayOpcode.Dispatch:
                 SequenceNumber = payload.SequenceNumber.GetValueOrDefault();
                 try
                 {
-                    await ProcessEventAsync(payload).ConfigureAwait(false);
+                    await ProcessEventAsync(state, connectionState, payload).ConfigureAwait(false);
                 }
                 catch (Exception ex)
                 {
@@ -918,21 +918,21 @@ private protected override async Task ProcessPayloadAsync(JsonPayload payload)
                 break;
             case GatewayOpcode.Reconnect:
                 InvokeLog(LogMessage.Info("Reconnect request"));
-                await AbortAndReconnectAsync().ConfigureAwait(false);
+                await AbortAndReconnectAsync(state, connectionState).ConfigureAwait(false);
                 break;
             case GatewayOpcode.InvalidSession:
                 InvokeLog(LogMessage.Info("Invalid session"));
                 try
                 {
-                    await SendIdentifyAsync().ConfigureAwait(false);
+                    await SendIdentifyAsync(connectionState).ConfigureAwait(false);
                 }
                 catch (Exception ex)
                 {
                     InvokeLog(LogMessage.Error(ex));
                 }
                 break;
             case GatewayOpcode.Hello:
-                StartHeartbeating(payload.Data.GetValueOrDefault().ToObject(Serialization.Default.JsonHello).HeartbeatInterval);
+                StartHeartbeating(connectionState, payload.Data.GetValueOrDefault().ToObject(Serialization.Default.JsonHello).HeartbeatInterval);
                 break;
             case GatewayOpcode.HeartbeatACK:
                 await UpdateLatencyAsync(_latencyTimer.Elapsed).ConfigureAwait(false);
@@ -943,33 +943,34 @@ private protected override async Task ProcessPayloadAsync(JsonPayload payload)
     /// <summary>
     /// Joins, moves, or disconnects the app from a voice channel.
     /// </summary>
-    public ValueTask UpdateVoiceStateAsync(VoiceStateProperties voiceState, CancellationToken cancellationToken = default)
+    public ValueTask UpdateVoiceStateAsync(VoiceStateProperties voiceState, WebSocketPayloadProperties? properties = null, CancellationToken cancellationToken = default)
     {
         GatewayPayloadProperties<VoiceStateProperties> payload = new(GatewayOpcode.VoiceStateUpdate, voiceState);
-        return SendPayloadAsync(payload.Serialize(Serialization.Default.GatewayPayloadPropertiesVoiceStateProperties), cancellationToken);
+        return SendPayloadAsync(payload.Serialize(Serialization.Default.GatewayPayloadPropertiesVoiceStateProperties), properties, cancellationToken);
     }
 
     /// <summary>
     /// Updates an app's presence.
     /// </summary>
     /// <param name="presence">The presence to set.</param>
+    /// <param name="properties"></param>
     /// <param name="cancellationToken">The cancellation token to cancel the operation.</param>
-    public ValueTask UpdatePresenceAsync(PresenceProperties presence, CancellationToken cancellationToken = default)
+    public ValueTask UpdatePresenceAsync(PresenceProperties presence, WebSocketPayloadProperties? properties = null, CancellationToken cancellationToken = default)
     {
         GatewayPayloadProperties<PresenceProperties> payload = new(GatewayOpcode.PresenceUpdate, presence);
-        return SendPayloadAsync(payload.Serialize(Serialization.Default.GatewayPayloadPropertiesPresenceProperties), cancellationToken);
+        return SendPayloadAsync(payload.Serialize(Serialization.Default.GatewayPayloadPropertiesPresenceProperties), properties, cancellationToken);
     }
 
     /// <summary>
-    /// Requests user for a guild.
+    /// Requests users for a guild.
     /// </summary>
-    public ValueTask RequestGuildUsersAsync(GuildUsersRequestProperties requestProperties, CancellationToken cancellationToken = default)
+    public ValueTask RequestGuildUsersAsync(GuildUsersRequestProperties requestProperties, WebSocketPayloadProperties? properties = null, CancellationToken cancellationToken = default)
     {
         GatewayPayloadProperties<GuildUsersRequestProperties> payload = new(GatewayOpcode.RequestGuildUsers, requestProperties);
-        return SendPayloadAsync(payload.Serialize(Serialization.Default.GatewayPayloadPropertiesGuildUsersRequestProperties), cancellationToken);
+        return SendPayloadAsync(payload.Serialize(Serialization.Default.GatewayPayloadPropertiesGuildUsersRequestProperties), properties, cancellationToken);
     }
 
-    private async Task ProcessEventAsync(JsonPayload payload)
+    private async Task ProcessEventAsync(State state, ConnectionState connectionState, JsonPayload payload)
     {
         var data = payload.Data.GetValueOrDefault();
         var name = payload.Event!;
@@ -991,7 +992,7 @@ await InvokeEventAsync(Ready, args, data =>
                         SessionId = args.SessionId;
                         ApplicationFlags = args.ApplicationFlags;
 
-                        _readyCompletionSource.TrySetResult();
+                        state.IndicateReady(connectionState);
                     }).ConfigureAwait(false);
                     await updateLatencyTask.ConfigureAwait(false);
                 }
@@ -1003,7 +1004,7 @@ await InvokeEventAsync(Ready, args, data =>
                     var updateLatencyTask = UpdateLatencyAsync(latency);
                     var resumeTask = InvokeResumeEventAsync();
 
-                    _readyCompletionSource.TrySetResult();
+                    state.IndicateReady(connectionState);
 
                     await updateLatencyTask.ConfigureAwait(false);
                     await resumeTask.ConfigureAwait(false);

NetCord/Gateway/GatewayClientConfiguration.cs

@@ -7,7 +7,9 @@ namespace NetCord.Gateway;
 
 public class GatewayClientConfiguration : IWebSocketClientConfiguration
 {
-    public IWebSocket? WebSocket { get; init; }
+    public IWebSocketConnectionProvider? WebSocketConnectionProvider { get; init; }
+    public IRateLimiterProvider? RateLimiterProvider { get; init; }
+    public WebSocketPayloadProperties? DefaultPayloadProperties { get; init; }
     public IReconnectStrategy? ReconnectStrategy { get; init; }
     public ILatencyTimer? LatencyTimer { get; init; }
     public ApiVersion Version { get; init; } = ApiVersion.V10;
@@ -21,4 +23,6 @@ public class GatewayClientConfiguration : IWebSocketClientConfiguration
     public Shard? Shard { get; init; }
     public bool CacheDMChannels { get; init; } = true;
     public Rest.RestClientConfiguration? RestClientConfiguration { get; init; }
+
+    IRateLimiterProvider? IWebSocketClientConfiguration.RateLimiterProvider => RateLimiterProvider is { } rateLimiter ? rateLimiter : new GatewayRateLimiterProvider(120, 60_000);
 }

NetCord/Gateway/GatewayRateLimiterProvider.cs

@@ -0,0 +1,59 @@
+namespace NetCord.Gateway;
+
+public class GatewayRateLimiterProvider(int limit, long duration) : IRateLimiterProvider
+{
+    public IRateLimiter CreateRateLimiter() => new GatewayRateLimiter(limit, duration);
+
+    private sealed class GatewayRateLimiter(int limit, long duration) : IRateLimiter
+    {
+        private readonly object _lock = new();
+        private readonly int _limit = limit;
+        private int _remaining = limit;
+        private long _reset;
+
+        public ValueTask<RateLimitAcquisitionResult> TryAcquireAsync(CancellationToken cancellationToken = default)
+        {
+            var timestamp = Environment.TickCount64;
+            lock (_lock)
+            {
+                var diff = _reset - timestamp;
+                if (diff <= 0)
+                {
+                    _remaining = _limit - 1;
+                    _reset = timestamp + duration;
+                }
+                else
+                {
+                    if (_remaining == 0)
+                        return new(RateLimitAcquisitionResult.RateLimit((int)diff));
+                    else
+                        _remaining--;
+                }
+            }
+
+            return new(RateLimitAcquisitionResult.NoRateLimit);
+        }
+
+        public ValueTask CancelAcquireAsync(long acquisitionTimestamp, CancellationToken cancellationToken = default)
+        {
+            var currentTimestamp = Environment.TickCount64;
+            lock (_lock)
+            {
+                var reset = _reset;
+                var start = reset - duration;
+                if (acquisitionTimestamp <= reset
+                    && acquisitionTimestamp >= start
+                    && currentTimestamp <= reset
+                    && currentTimestamp >= start
+                    && _remaining < _limit)
+                    _remaining++;
+            }
+
+            return default;
+        }
+
+        public void Dispose()
+        {
+        }
+    }
+}

NetCord/Gateway/IRateLimiter.cs

@@ -0,0 +1,8 @@
+namespace NetCord.Gateway;
+
+public interface IRateLimiter : IDisposable
+{
+    public ValueTask<RateLimitAcquisitionResult> TryAcquireAsync(CancellationToken cancellationToken = default);
+
+    public ValueTask CancelAcquireAsync(long acquisitionTimestamp, CancellationToken cancellationToken = default);
+}

NetCord/Gateway/IRateLimiterProvider.cs

@@ -0,0 +1,6 @@
+namespace NetCord.Gateway;
+
+public interface IRateLimiterProvider
+{
+    public IRateLimiter CreateRateLimiter();
+}

NetCord/Gateway/IWebSocketClientConfiguration.cs

@@ -6,7 +6,9 @@ namespace NetCord.Gateway;
 
 internal interface IWebSocketClientConfiguration
 {
-    public IWebSocket? WebSocket { get; }
+    public IWebSocketConnectionProvider? WebSocketConnectionProvider { get; }
     public IReconnectStrategy? ReconnectStrategy { get; }
     public ILatencyTimer? LatencyTimer { get; }
+    public IRateLimiterProvider? RateLimiterProvider { get; }
+    public WebSocketPayloadProperties? DefaultPayloadProperties { get; }
 }

NetCord/Gateway/NullRateLimiterProvider.cs

@@ -0,0 +1,25 @@
+namespace NetCord.Gateway;
+
+public class NullRateLimiterProvider : IRateLimiterProvider
+{
+    public static NullRateLimiterProvider Instance { get; } = new();
+
+    public IRateLimiter CreateRateLimiter() => NullRateLimiter.Instance;
+
+    private sealed class NullRateLimiter : IRateLimiter
+    {
+        public static NullRateLimiter Instance { get; } = new();
+
+        private NullRateLimiter()
+        {
+        }
+
+        public ValueTask<RateLimitAcquisitionResult> TryAcquireAsync(CancellationToken cancellationToken = default) => new(RateLimitAcquisitionResult.NoRateLimit);
+
+        public ValueTask CancelAcquireAsync(long acquisitionTimestamp, CancellationToken cancellationToken = default) => default;
+
+        public void Dispose()
+        {
+        }
+    }
+}

NetCord/Gateway/RateLimitAcquisitionResult.cs

@@ -0,0 +1,18 @@
+namespace NetCord.Gateway;
+
+public readonly struct RateLimitAcquisitionResult
+{
+    private RateLimitAcquisitionResult(int resetAfter, bool rateLimited)
+    {
+        ResetAfter = resetAfter;
+        RateLimited = rateLimited;
+    }
+
+    public static RateLimitAcquisitionResult NoRateLimit { get; } = new(0, false);
+
+    public static RateLimitAcquisitionResult RateLimit(int resetAfter) => new(resetAfter, true);
+
+    public int ResetAfter { get; }
+
+    public bool RateLimited { get; }
+}

NetCord/Gateway/RentedArrayBufferWriter.cs

@@ -57,7 +57,7 @@ private void ResizeBuffer(int sizeHint)
         {
             var pool = ArrayPool<T>.Shared;
             var newBuffer = pool.Rent(sum);
-            Array.Copy(buffer, newBuffer, index);
+            buffer.AsSpan(0, index).CopyTo(newBuffer);
             _buffer = newBuffer;
             pool.Return(buffer);
         }