Browse Source

fix: Different ratelimits for the same route (implement discord buckets) (#1546)

* Don't disable when there's no resetTick

Sometimes Discord won't send any ratelimit headers, disabling the semaphore for endpoints that should have them.

* Undo changes and change comment

* Add HttpMethod to BucketIds

* Add X-RateLimit-Bucket

* BucketId changes

- BucketId is it's own class now
- Add WebhookId as a major parameter
- Add shared buckets using the hash and major parameters

* Add webhookId to BucketIds

* Update BucketId and redirect requests

* General bugfixes

* Assign semaphore and follow the same standard as Reset for ResetAfter
tags/2.3.0
Paulo GitHub 4 years ago
parent
commit
2f6c0175c8
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 269 additions and 64 deletions
  1. +118
    -0
      src/Discord.Net.Core/Net/BucketId.cs
  2. +2
    -1
      src/Discord.Net.Core/RequestOptions.cs
  3. +2
    -2
      src/Discord.Net.Rest/BaseDiscordClient.cs
  4. +45
    -24
      src/Discord.Net.Rest/DiscordRestApiClient.cs
  5. +7
    -7
      src/Discord.Net.Rest/Net/Queue/ClientBucket.cs
  6. +31
    -8
      src/Discord.Net.Rest/Net/Queue/RequestQueue.cs
  7. +57
    -17
      src/Discord.Net.Rest/Net/Queue/RequestQueueBucket.cs
  8. +5
    -3
      src/Discord.Net.Rest/Net/RateLimitInfo.cs
  9. +2
    -2
      src/Discord.Net.Webhook/DiscordWebhookClient.cs

+ 118
- 0
src/Discord.Net.Core/Net/BucketId.cs View File

@@ -0,0 +1,118 @@
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Linq;

namespace Discord.Net
{
/// <summary>
/// Represents a ratelimit bucket.
/// </summary>
public class BucketId : IEquatable<BucketId>
{
/// <summary>
/// Gets the http method used to make the request if available.
/// </summary>
public string HttpMethod { get; }
/// <summary>
/// Gets the endpoint that is going to be requested if available.
/// </summary>
public string Endpoint { get; }
/// <summary>
/// Gets the major parameters of the route.
/// </summary>
public IOrderedEnumerable<KeyValuePair<string, string>> MajorParameters { get; }
/// <summary>
/// Gets the hash of this bucket.
/// </summary>
/// <remarks>
/// The hash is provided by Discord to group ratelimits.
/// </remarks>
public string BucketHash { get; }
/// <summary>
/// Gets if this bucket is a hash type.
/// </summary>
public bool IsHashBucket { get => BucketHash != null; }

private BucketId(string httpMethod, string endpoint, IEnumerable<KeyValuePair<string, string>> majorParameters, string bucketHash)
{
HttpMethod = httpMethod;
Endpoint = endpoint;
MajorParameters = majorParameters.OrderBy(x => x.Key);
BucketHash = bucketHash;
}

/// <summary>
/// Creates a new <see cref="BucketId"/> based on the
/// <see cref="HttpMethod"/> and <see cref="Endpoint"/>.
/// </summary>
/// <param name="httpMethod">Http method used to make the request.</param>
/// <param name="endpoint">Endpoint that is going to receive requests.</param>
/// <param name="majorParams">Major parameters of the route of this endpoint.</param>
/// <returns>
/// A <see cref="BucketId"/> based on the <see cref="HttpMethod"/>
/// and the <see cref="Endpoint"/> with the provided data.
/// </returns>
public static BucketId Create(string httpMethod, string endpoint, Dictionary<string, string> majorParams)
{
Preconditions.NotNullOrWhitespace(endpoint, nameof(endpoint));
majorParams ??= new Dictionary<string, string>();
return new BucketId(httpMethod, endpoint, majorParams, null);
}

/// <summary>
/// Creates a new <see cref="BucketId"/> based on a
/// <see cref="BucketHash"/> and a previous <see cref="BucketId"/>.
/// </summary>
/// <param name="hash">Bucket hash provided by Discord.</param>
/// <param name="oldBucket"><see cref="BucketId"/> that is going to be upgraded to a hash type.</param>
/// <returns>
/// A <see cref="BucketId"/> based on the <see cref="BucketHash"/>
/// and <see cref="MajorParameters"/>.
/// </returns>
public static BucketId Create(string hash, BucketId oldBucket)
{
Preconditions.NotNullOrWhitespace(hash, nameof(hash));
Preconditions.NotNull(oldBucket, nameof(oldBucket));
return new BucketId(null, null, oldBucket.MajorParameters, hash);
}

/// <summary>
/// Gets the string that will define this bucket as a hash based one.
/// </summary>
/// <returns>
/// A <see cref="string"/> that defines this bucket as a hash based one.
/// </returns>
public string GetBucketHash()
=> IsHashBucket ? $"{BucketHash}:{string.Join("/", MajorParameters.Select(x => x.Value))}" : null;

/// <summary>
/// Gets the string that will define this bucket as an endpoint based one.
/// </summary>
/// <returns>
/// A <see cref="string"/> that defines this bucket as an endpoint based one.
/// </returns>
public string GetUniqueEndpoint()
=> HttpMethod != null ? $"{HttpMethod} {Endpoint}" : Endpoint;

public override bool Equals(object obj)
=> Equals(obj as BucketId);

public override int GetHashCode()
=> IsHashBucket ? (BucketHash, string.Join("/", MajorParameters.Select(x => x.Value))).GetHashCode() : (HttpMethod, Endpoint).GetHashCode();

public override string ToString()
=> GetBucketHash() ?? GetUniqueEndpoint();

public bool Equals(BucketId other)
{
if (other is null)
return false;
if (ReferenceEquals(this, other))
return true;
if (GetType() != other.GetType())
return false;
return ToString() == other.ToString();
}
}
}

+ 2
- 1
src/Discord.Net.Core/RequestOptions.cs View File

@@ -1,3 +1,4 @@
using Discord.Net;
using System.Threading;

namespace Discord
@@ -57,7 +58,7 @@ namespace Discord
public bool? UseSystemClock { get; set; }

internal bool IgnoreState { get; set; }
internal string BucketId { get; set; }
internal BucketId BucketId { get; set; }
internal bool IsClientBucket { get; set; }
internal bool IsReactionBucket { get; set; }



+ 2
- 2
src/Discord.Net.Rest/BaseDiscordClient.cs View File

@@ -49,9 +49,9 @@ namespace Discord.Rest
ApiClient.RequestQueue.RateLimitTriggered += async (id, info) =>
{
if (info == null)
await _restLogger.VerboseAsync($"Preemptive Rate limit triggered: {id ?? "null"}").ConfigureAwait(false);
await _restLogger.VerboseAsync($"Preemptive Rate limit triggered: {id?.ToString() ?? "null"}").ConfigureAwait(false);
else
await _restLogger.WarningAsync($"Rate limit triggered: {id ?? "null"}").ConfigureAwait(false);
await _restLogger.WarningAsync($"Rate limit triggered: {id?.ToString() ?? "null"}").ConfigureAwait(false);
};
ApiClient.SentRequest += async (method, endpoint, millis) => await _restLogger.VerboseAsync($"{method} {endpoint}: {millis} ms").ConfigureAwait(false);
}


+ 45
- 24
src/Discord.Net.Rest/DiscordRestApiClient.cs View File

@@ -24,7 +24,7 @@ namespace Discord.API
{
internal class DiscordRestApiClient : IDisposable
{
private static readonly ConcurrentDictionary<string, Func<BucketIds, string>> _bucketIdGenerators = new ConcurrentDictionary<string, Func<BucketIds, string>>();
private static readonly ConcurrentDictionary<string, Func<BucketIds, BucketId>> _bucketIdGenerators = new ConcurrentDictionary<string, Func<BucketIds, BucketId>>();

public event Func<string, string, double, Task> SentRequest { add { _sentRequestEvent.Add(value); } remove { _sentRequestEvent.Remove(value); } }
private readonly AsyncEvent<Func<string, string, double, Task>> _sentRequestEvent = new AsyncEvent<Func<string, string, double, Task>>();
@@ -176,9 +176,9 @@ namespace Discord.API
//Core
internal Task SendAsync(string method, Expression<Func<string>> endpointExpr, BucketIds ids,
ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null, [CallerMemberName] string funcName = null)
=> SendAsync(method, GetEndpoint(endpointExpr), GetBucketId(ids, endpointExpr, funcName), clientBucket, options);
=> SendAsync(method, GetEndpoint(endpointExpr), GetBucketId(method, ids, endpointExpr, funcName), clientBucket, options);
public async Task SendAsync(string method, string endpoint,
string bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null)
BucketId bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null)
{
options = options ?? new RequestOptions();
options.HeaderOnly = true;
@@ -190,9 +190,9 @@ namespace Discord.API

internal Task SendJsonAsync(string method, Expression<Func<string>> endpointExpr, object payload, BucketIds ids,
ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null, [CallerMemberName] string funcName = null)
=> SendJsonAsync(method, GetEndpoint(endpointExpr), payload, GetBucketId(ids, endpointExpr, funcName), clientBucket, options);
=> SendJsonAsync(method, GetEndpoint(endpointExpr), payload, GetBucketId(method, ids, endpointExpr, funcName), clientBucket, options);
public async Task SendJsonAsync(string method, string endpoint, object payload,
string bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null)
BucketId bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null)
{
options = options ?? new RequestOptions();
options.HeaderOnly = true;
@@ -205,9 +205,9 @@ namespace Discord.API

internal Task SendMultipartAsync(string method, Expression<Func<string>> endpointExpr, IReadOnlyDictionary<string, object> multipartArgs, BucketIds ids,
ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null, [CallerMemberName] string funcName = null)
=> SendMultipartAsync(method, GetEndpoint(endpointExpr), multipartArgs, GetBucketId(ids, endpointExpr, funcName), clientBucket, options);
=> SendMultipartAsync(method, GetEndpoint(endpointExpr), multipartArgs, GetBucketId(method, ids, endpointExpr, funcName), clientBucket, options);
public async Task SendMultipartAsync(string method, string endpoint, IReadOnlyDictionary<string, object> multipartArgs,
string bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null)
BucketId bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null)
{
options = options ?? new RequestOptions();
options.HeaderOnly = true;
@@ -219,9 +219,9 @@ namespace Discord.API

internal Task<TResponse> SendAsync<TResponse>(string method, Expression<Func<string>> endpointExpr, BucketIds ids,
ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null, [CallerMemberName] string funcName = null) where TResponse : class
=> SendAsync<TResponse>(method, GetEndpoint(endpointExpr), GetBucketId(ids, endpointExpr, funcName), clientBucket, options);
=> SendAsync<TResponse>(method, GetEndpoint(endpointExpr), GetBucketId(method, ids, endpointExpr, funcName), clientBucket, options);
public async Task<TResponse> SendAsync<TResponse>(string method, string endpoint,
string bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null) where TResponse : class
BucketId bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null) where TResponse : class
{
options = options ?? new RequestOptions();
options.BucketId = bucketId;
@@ -232,9 +232,9 @@ namespace Discord.API

internal Task<TResponse> SendJsonAsync<TResponse>(string method, Expression<Func<string>> endpointExpr, object payload, BucketIds ids,
ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null, [CallerMemberName] string funcName = null) where TResponse : class
=> SendJsonAsync<TResponse>(method, GetEndpoint(endpointExpr), payload, GetBucketId(ids, endpointExpr, funcName), clientBucket, options);
=> SendJsonAsync<TResponse>(method, GetEndpoint(endpointExpr), payload, GetBucketId(method, ids, endpointExpr, funcName), clientBucket, options);
public async Task<TResponse> SendJsonAsync<TResponse>(string method, string endpoint, object payload,
string bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null) where TResponse : class
BucketId bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null) where TResponse : class
{
options = options ?? new RequestOptions();
options.BucketId = bucketId;
@@ -246,9 +246,9 @@ namespace Discord.API

internal Task<TResponse> SendMultipartAsync<TResponse>(string method, Expression<Func<string>> endpointExpr, IReadOnlyDictionary<string, object> multipartArgs, BucketIds ids,
ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null, [CallerMemberName] string funcName = null)
=> SendMultipartAsync<TResponse>(method, GetEndpoint(endpointExpr), multipartArgs, GetBucketId(ids, endpointExpr, funcName), clientBucket, options);
=> SendMultipartAsync<TResponse>(method, GetEndpoint(endpointExpr), multipartArgs, GetBucketId(method, ids, endpointExpr, funcName), clientBucket, options);
public async Task<TResponse> SendMultipartAsync<TResponse>(string method, string endpoint, IReadOnlyDictionary<string, object> multipartArgs,
string bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null)
BucketId bucketId = null, ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null)
{
options = options ?? new RequestOptions();
options.BucketId = bucketId;
@@ -520,7 +520,8 @@ namespace Discord.API
throw new ArgumentException(message: $"Message content is too long, length must be less or equal to {DiscordConfig.MaxMessageSize}.", paramName: nameof(args.Content));
options = RequestOptions.CreateOrClone(options);

return await SendJsonAsync<Message>("POST", () => $"webhooks/{webhookId}/{AuthToken}?wait=true", args, new BucketIds(), clientBucket: ClientBucketType.SendEdit, options: options).ConfigureAwait(false);
var ids = new BucketIds(webhookId: webhookId);
return await SendJsonAsync<Message>("POST", () => $"webhooks/{webhookId}/{AuthToken}?wait=true", args, ids, clientBucket: ClientBucketType.SendEdit, options: options).ConfigureAwait(false);
}
/// <exception cref="ArgumentOutOfRangeException">Message content is too long, length must be less or equal to <see cref="DiscordConfig.MaxMessageSize"/>.</exception>
public async Task<Message> UploadFileAsync(ulong channelId, UploadFileParams args, RequestOptions options = null)
@@ -559,7 +560,8 @@ namespace Discord.API
throw new ArgumentOutOfRangeException($"Message content is too long, length must be less or equal to {DiscordConfig.MaxMessageSize}.", nameof(args.Content));
}

return await SendMultipartAsync<Message>("POST", () => $"webhooks/{webhookId}/{AuthToken}?wait=true", args.ToDictionary(), new BucketIds(), clientBucket: ClientBucketType.SendEdit, options: options).ConfigureAwait(false);
var ids = new BucketIds(webhookId: webhookId);
return await SendMultipartAsync<Message>("POST", () => $"webhooks/{webhookId}/{AuthToken}?wait=true", args.ToDictionary(), ids, clientBucket: ClientBucketType.SendEdit, options: options).ConfigureAwait(false);
}
public async Task DeleteMessageAsync(ulong channelId, ulong messageId, RequestOptions options = null)
{
@@ -1466,21 +1468,39 @@ namespace Discord.API
{
public ulong GuildId { get; internal set; }
public ulong ChannelId { get; internal set; }
public ulong WebhookId { get; internal set; }
public string HttpMethod { get; internal set; }

internal BucketIds(ulong guildId = 0, ulong channelId = 0)
internal BucketIds(ulong guildId = 0, ulong channelId = 0, ulong webhookId = 0)
{
GuildId = guildId;
ChannelId = channelId;
WebhookId = webhookId;
}

internal object[] ToArray()
=> new object[] { GuildId, ChannelId };
=> new object[] { HttpMethod, GuildId, ChannelId, WebhookId };

internal Dictionary<string, string> ToMajorParametersDictionary()
{
var dict = new Dictionary<string, string>();
if (GuildId != 0)
dict["GuildId"] = GuildId.ToString();
if (ChannelId != 0)
dict["ChannelId"] = ChannelId.ToString();
if (WebhookId != 0)
dict["WebhookId"] = WebhookId.ToString();
return dict;
}

internal static int? GetIndex(string name)
{
switch (name)
{
case "guildId": return 0;
case "channelId": return 1;
case "httpMethod": return 0;
case "guildId": return 1;
case "channelId": return 2;
case "webhookId": return 3;
default:
return null;
}
@@ -1491,18 +1511,19 @@ namespace Discord.API
{
return endpointExpr.Compile()();
}
private static string GetBucketId(BucketIds ids, Expression<Func<string>> endpointExpr, string callingMethod)
private static BucketId GetBucketId(string httpMethod, BucketIds ids, Expression<Func<string>> endpointExpr, string callingMethod)
{
ids.HttpMethod ??= httpMethod;
return _bucketIdGenerators.GetOrAdd(callingMethod, x => CreateBucketId(endpointExpr))(ids);
}

private static Func<BucketIds, string> CreateBucketId(Expression<Func<string>> endpoint)
private static Func<BucketIds, BucketId> CreateBucketId(Expression<Func<string>> endpoint)
{
try
{
//Is this a constant string?
if (endpoint.Body.NodeType == ExpressionType.Constant)
return x => (endpoint.Body as ConstantExpression).Value.ToString();
return x => BucketId.Create(x.HttpMethod, (endpoint.Body as ConstantExpression).Value.ToString(), x.ToMajorParametersDictionary());

var builder = new StringBuilder();
var methodCall = endpoint.Body as MethodCallExpression;
@@ -1539,7 +1560,7 @@ namespace Discord.API

var mappedId = BucketIds.GetIndex(fieldName);

if(!mappedId.HasValue && rightIndex != endIndex && format.Length > rightIndex + 1 && format[rightIndex + 1] == '/') //Ignore the next slash
if (!mappedId.HasValue && rightIndex != endIndex && format.Length > rightIndex + 1 && format[rightIndex + 1] == '/') //Ignore the next slash
rightIndex++;

if (mappedId.HasValue)
@@ -1552,7 +1573,7 @@ namespace Discord.API

format = builder.ToString();

return x => string.Format(format, x.ToArray());
return x => BucketId.Create(x.HttpMethod, string.Format(format, x.ToArray()), x.ToMajorParametersDictionary());
}
catch (Exception ex)
{


+ 7
- 7
src/Discord.Net.Rest/Net/Queue/ClientBucket.cs View File

@@ -10,14 +10,14 @@ namespace Discord.Net.Queue
internal struct ClientBucket
{
private static readonly ImmutableDictionary<ClientBucketType, ClientBucket> DefsByType;
private static readonly ImmutableDictionary<string, ClientBucket> DefsById;
private static readonly ImmutableDictionary<BucketId, ClientBucket> DefsById;

static ClientBucket()
{
var buckets = new[]
{
new ClientBucket(ClientBucketType.Unbucketed, "<unbucketed>", 10, 10),
new ClientBucket(ClientBucketType.SendEdit, "<send_edit>", 10, 10)
new ClientBucket(ClientBucketType.Unbucketed, BucketId.Create(null, "<unbucketed>", null), 10, 10),
new ClientBucket(ClientBucketType.SendEdit, BucketId.Create(null, "<send_edit>", null), 10, 10)
};

var builder = ImmutableDictionary.CreateBuilder<ClientBucketType, ClientBucket>();
@@ -25,21 +25,21 @@ namespace Discord.Net.Queue
builder.Add(bucket.Type, bucket);
DefsByType = builder.ToImmutable();

var builder2 = ImmutableDictionary.CreateBuilder<string, ClientBucket>();
var builder2 = ImmutableDictionary.CreateBuilder<BucketId, ClientBucket>();
foreach (var bucket in buckets)
builder2.Add(bucket.Id, bucket);
DefsById = builder2.ToImmutable();
}

public static ClientBucket Get(ClientBucketType type) => DefsByType[type];
public static ClientBucket Get(string id) => DefsById[id];
public static ClientBucket Get(BucketId id) => DefsById[id];
public ClientBucketType Type { get; }
public string Id { get; }
public BucketId Id { get; }
public int WindowCount { get; }
public int WindowSeconds { get; }

public ClientBucket(ClientBucketType type, string id, int count, int seconds)
public ClientBucket(ClientBucketType type, BucketId id, int count, int seconds)
{
Type = type;
Id = id;


+ 31
- 8
src/Discord.Net.Rest/Net/Queue/RequestQueue.cs View File

@@ -12,9 +12,9 @@ namespace Discord.Net.Queue
{
internal class RequestQueue : IDisposable
{
public event Func<string, RateLimitInfo?, Task> RateLimitTriggered;
public event Func<BucketId, RateLimitInfo?, Task> RateLimitTriggered;

private readonly ConcurrentDictionary<string, RequestBucket> _buckets;
private readonly ConcurrentDictionary<BucketId, object> _buckets;
private readonly SemaphoreSlim _tokenLock;
private readonly CancellationTokenSource _cancelTokenSource; //Dispose token
private CancellationTokenSource _clearToken;
@@ -34,7 +34,7 @@ namespace Discord.Net.Queue
_requestCancelToken = CancellationToken.None;
_parentToken = CancellationToken.None;

_buckets = new ConcurrentDictionary<string, RequestBucket>();
_buckets = new ConcurrentDictionary<BucketId, object>();

_cleanupTask = RunCleanup();
}
@@ -82,7 +82,7 @@ namespace Discord.Net.Queue
else
request.Options.CancelToken = _requestCancelToken;

var bucket = GetOrCreateBucket(request.Options.BucketId, request);
var bucket = GetOrCreateBucket(request.Options, request);
var result = await bucket.SendAsync(request).ConfigureAwait(false);
createdTokenSource?.Dispose();
return result;
@@ -110,14 +110,32 @@ namespace Discord.Net.Queue
_waitUntil = DateTimeOffset.UtcNow.AddMilliseconds(info.RetryAfter.Value + (info.Lag?.TotalMilliseconds ?? 0.0));
}

private RequestBucket GetOrCreateBucket(string id, RestRequest request)
private RequestBucket GetOrCreateBucket(RequestOptions options, RestRequest request)
{
return _buckets.GetOrAdd(id, x => new RequestBucket(this, request, x));
var bucketId = options.BucketId;
object obj = _buckets.GetOrAdd(bucketId, x => new RequestBucket(this, request, x));
if (obj is BucketId hashBucket)
{
options.BucketId = hashBucket;
return (RequestBucket)_buckets.GetOrAdd(hashBucket, x => new RequestBucket(this, request, x));
}
return (RequestBucket)obj;
}
internal async Task RaiseRateLimitTriggered(string bucketId, RateLimitInfo? info)
internal async Task RaiseRateLimitTriggered(BucketId bucketId, RateLimitInfo? info)
{
await RateLimitTriggered(bucketId, info).ConfigureAwait(false);
}
internal (RequestBucket, BucketId) UpdateBucketHash(BucketId id, string discordHash)
{
if (!id.IsHashBucket)
{
var bucket = BucketId.Create(discordHash, id);
var hashReqQueue = (RequestBucket)_buckets.GetOrAdd(bucket, _buckets[id]);
_buckets.AddOrUpdate(id, bucket, (oldBucket, oldObj) => bucket);
return (hashReqQueue, bucket);
}
return (null, null);
}

private async Task RunCleanup()
{
@@ -126,10 +144,15 @@ namespace Discord.Net.Queue
while (!_cancelTokenSource.IsCancellationRequested)
{
var now = DateTimeOffset.UtcNow;
foreach (var bucket in _buckets.Select(x => x.Value))
foreach (var bucket in _buckets.Where(x => x.Value is RequestBucket).Select(x => (RequestBucket)x.Value))
{
if ((now - bucket.LastAttemptAt).TotalMinutes > 1.0)
{
if (bucket.Id.IsHashBucket)
foreach (var redirectBucket in _buckets.Where(x => x.Value == bucket.Id).Select(x => (BucketId)x.Value))
_buckets.TryRemove(redirectBucket, out _); //remove redirections if hash bucket
_buckets.TryRemove(bucket.Id, out _);
}
}
await Task.Delay(60000, _cancelTokenSource.Token).ConfigureAwait(false); //Runs each minute
}


+ 57
- 17
src/Discord.Net.Rest/Net/Queue/RequestQueueBucket.cs View File

@@ -19,12 +19,13 @@ namespace Discord.Net.Queue
private readonly RequestQueue _queue;
private int _semaphore;
private DateTimeOffset? _resetTick;
private RequestBucket _redirectBucket;

public string Id { get; private set; }
public BucketId Id { get; private set; }
public int WindowCount { get; private set; }
public DateTimeOffset LastAttemptAt { get; private set; }

public RequestBucket(RequestQueue queue, RestRequest request, string id)
public RequestBucket(RequestQueue queue, RestRequest request, BucketId id)
{
_queue = queue;
Id = id;
@@ -32,7 +33,7 @@ namespace Discord.Net.Queue
_lock = new object();

if (request.Options.IsClientBucket)
WindowCount = ClientBucket.Get(request.Options.BucketId).WindowCount;
WindowCount = ClientBucket.Get(Id).WindowCount;
else
WindowCount = 1; //Only allow one request until we get a header back
_semaphore = WindowCount;
@@ -52,6 +53,8 @@ namespace Discord.Net.Queue
{
await _queue.EnterGlobalAsync(id, request).ConfigureAwait(false);
await EnterAsync(id, request).ConfigureAwait(false);
if (_redirectBucket != null)
return await _redirectBucket.SendAsync(request);

#if DEBUG_LIMITS
Debug.WriteLine($"[{id}] Sending...");
@@ -160,6 +163,9 @@ namespace Discord.Net.Queue

while (true)
{
if (_redirectBucket != null)
break;

if (DateTimeOffset.UtcNow > request.TimeoutAt || request.Options.CancelToken.IsCancellationRequested)
{
if (!isRateLimited)
@@ -175,7 +181,8 @@ namespace Discord.Net.Queue
}

DateTimeOffset? timeoutAt = request.TimeoutAt;
if (windowCount > 0 && Interlocked.Decrement(ref _semaphore) < 0)
int semaphore = Interlocked.Decrement(ref _semaphore);
if (windowCount > 0 && semaphore < 0)
{
if (!isRateLimited)
{
@@ -210,20 +217,52 @@ namespace Discord.Net.Queue
}
#if DEBUG_LIMITS
else
Debug.WriteLine($"[{id}] Entered Semaphore ({_semaphore}/{WindowCount} remaining)");
Debug.WriteLine($"[{id}] Entered Semaphore ({semaphore}/{WindowCount} remaining)");
#endif
break;
}
}

private void UpdateRateLimit(int id, RestRequest request, RateLimitInfo info, bool is429)
private void UpdateRateLimit(int id, RestRequest request, RateLimitInfo info, bool is429, bool redirected = false)
{
if (WindowCount == 0)
return;

lock (_lock)
{
if (redirected)
{
Interlocked.Decrement(ref _semaphore); //we might still hit a real ratelimit if all tickets were already taken, can't do much about it since we didn't know they were the same
#if DEBUG_LIMITS
Debug.WriteLine($"[{id}] Decrease Semaphore");
#endif
}
bool hasQueuedReset = _resetTick != null;

if (info.Bucket != null && !redirected)
{
(RequestBucket, BucketId) hashBucket = _queue.UpdateBucketHash(Id, info.Bucket);
if (!(hashBucket.Item1 is null) && !(hashBucket.Item2 is null))
{
if (hashBucket.Item1 == this) //this bucket got promoted to a hash queue
{
Id = hashBucket.Item2;
#if DEBUG_LIMITS
Debug.WriteLine($"[{id}] Promoted to Hash Bucket ({hashBucket.Item2})");
#endif
}
else
{
_redirectBucket = hashBucket.Item1; //this request should be part of another bucket, this bucket will be disabled, redirect everything
_redirectBucket.UpdateRateLimit(id, request, info, is429, redirected: true); //update the hash bucket ratelimit
#if DEBUG_LIMITS
Debug.WriteLine($"[{id}] Redirected to {_redirectBucket.Id}");
#endif
return;
}
}
}

if (info.Limit.HasValue && WindowCount != info.Limit.Value)
{
WindowCount = info.Limit.Value;
@@ -233,7 +272,6 @@ namespace Discord.Net.Queue
#endif
}

var now = DateTimeOffset.UtcNow.ToUnixTimeSeconds();
DateTimeOffset? resetTick = null;

//Using X-RateLimit-Remaining causes a race condition
@@ -250,16 +288,18 @@ namespace Discord.Net.Queue
Debug.WriteLine($"[{id}] Retry-After: {info.RetryAfter.Value} ({info.RetryAfter.Value} ms)");
#endif
}
else if (info.ResetAfter.HasValue && (request.Options.UseSystemClock.HasValue ? !request.Options.UseSystemClock.Value : false))
{
resetTick = DateTimeOffset.UtcNow.Add(info.ResetAfter.Value);
}
else if (info.ResetAfter.HasValue && (request.Options.UseSystemClock.HasValue ? !request.Options.UseSystemClock.Value : false))
{
resetTick = DateTimeOffset.UtcNow.Add(info.ResetAfter.Value);
#if DEBUG_LIMITS
Debug.WriteLine($"[{id}] Reset-After: {info.ResetAfter.Value} ({info.ResetAfter?.TotalMilliseconds} ms)");
#endif
}
else if (info.Reset.HasValue)
{
resetTick = info.Reset.Value.AddSeconds(info.Lag?.TotalSeconds ?? 1.0);

/* millisecond precision makes this unnecessary, retaining in case of regression

/* millisecond precision makes this unnecessary, retaining in case of regression
if (request.Options.IsReactionBucket)
resetTick = DateTimeOffset.Now.AddMilliseconds(250);
*/
@@ -269,17 +309,17 @@ namespace Discord.Net.Queue
Debug.WriteLine($"[{id}] X-RateLimit-Reset: {info.Reset.Value.ToUnixTimeSeconds()} ({diff} ms, {info.Lag?.TotalMilliseconds} ms lag)");
#endif
}
else if (request.Options.IsClientBucket && request.Options.BucketId != null)
else if (request.Options.IsClientBucket && Id != null)
{
resetTick = DateTimeOffset.UtcNow.AddSeconds(ClientBucket.Get(request.Options.BucketId).WindowSeconds);
resetTick = DateTimeOffset.UtcNow.AddSeconds(ClientBucket.Get(Id).WindowSeconds);
#if DEBUG_LIMITS
Debug.WriteLine($"[{id}] Client Bucket ({ClientBucket.Get(request.Options.BucketId).WindowSeconds * 1000} ms)");
Debug.WriteLine($"[{id}] Client Bucket ({ClientBucket.Get(Id).WindowSeconds * 1000} ms)");
#endif
}

if (resetTick == null)
{
WindowCount = 0; //No rate limit info, disable limits on this bucket (should only ever happen with a user token)
WindowCount = 0; //No rate limit info, disable limits on this bucket
#if DEBUG_LIMITS
Debug.WriteLine($"[{id}] Disabled Semaphore");
#endif


+ 5
- 3
src/Discord.Net.Rest/Net/RateLimitInfo.cs View File

@@ -11,7 +11,8 @@ namespace Discord.Net
public int? Remaining { get; }
public int? RetryAfter { get; }
public DateTimeOffset? Reset { get; }
public TimeSpan? ResetAfter { get; }
public TimeSpan? ResetAfter { get; }
public string Bucket { get; }
public TimeSpan? Lag { get; }

internal RateLimitInfo(Dictionary<string, string> headers)
@@ -26,8 +27,9 @@ namespace Discord.Net
double.TryParse(temp, NumberStyles.AllowDecimalPoint, CultureInfo.InvariantCulture, out var reset) ? DateTimeOffset.FromUnixTimeMilliseconds((long)(reset * 1000)) : (DateTimeOffset?)null;
RetryAfter = headers.TryGetValue("Retry-After", out temp) &&
int.TryParse(temp, NumberStyles.None, CultureInfo.InvariantCulture, out var retryAfter) ? retryAfter : (int?)null;
ResetAfter = headers.TryGetValue("X-RateLimit-Reset-After", out temp) &&
float.TryParse(temp, out var resetAfter) ? TimeSpan.FromMilliseconds((long)(resetAfter * 1000)) : (TimeSpan?)null;
ResetAfter = headers.TryGetValue("X-RateLimit-Reset-After", out temp) &&
double.TryParse(temp, NumberStyles.AllowDecimalPoint, CultureInfo.InvariantCulture, out var resetAfter) ? TimeSpan.FromMilliseconds((long)(resetAfter * 1000)) : (TimeSpan?)null;
Bucket = headers.TryGetValue("X-RateLimit-Bucket", out temp) ? temp : null;
Lag = headers.TryGetValue("Date", out temp) &&
DateTimeOffset.TryParse(temp, CultureInfo.InvariantCulture, DateTimeStyles.None, out var date) ? DateTimeOffset.UtcNow - date : (TimeSpan?)null;
}


+ 2
- 2
src/Discord.Net.Webhook/DiscordWebhookClient.cs View File

@@ -77,9 +77,9 @@ namespace Discord.Webhook
ApiClient.RequestQueue.RateLimitTriggered += async (id, info) =>
{
if (info == null)
await _restLogger.VerboseAsync($"Preemptive Rate limit triggered: {id ?? "null"}").ConfigureAwait(false);
await _restLogger.VerboseAsync($"Preemptive Rate limit triggered: {id?.ToString() ?? "null"}").ConfigureAwait(false);
else
await _restLogger.WarningAsync($"Rate limit triggered: {id ?? "null"}").ConfigureAwait(false);
await _restLogger.WarningAsync($"Rate limit triggered: {id?.ToString() ?? "null"}").ConfigureAwait(false);
};
ApiClient.SentRequest += async (method, endpoint, millis) => await _restLogger.VerboseAsync($"{method} {endpoint}: {millis} ms").ConfigureAwait(false);
}


Loading…
Cancel
Save