* 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 ResetAfterpull/1614/head
@@ -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(); | |||||
} | |||||
} | |||||
} |
@@ -1,3 +1,4 @@ | |||||
using Discord.Net; | |||||
using System.Threading; | using System.Threading; | ||||
namespace Discord | namespace Discord | ||||
@@ -57,7 +58,7 @@ namespace Discord | |||||
public bool? UseSystemClock { get; set; } | public bool? UseSystemClock { get; set; } | ||||
internal bool IgnoreState { get; set; } | internal bool IgnoreState { get; set; } | ||||
internal string BucketId { get; set; } | |||||
internal BucketId BucketId { get; set; } | |||||
internal bool IsClientBucket { get; set; } | internal bool IsClientBucket { get; set; } | ||||
internal bool IsReactionBucket { get; set; } | internal bool IsReactionBucket { get; set; } | ||||
@@ -49,9 +49,9 @@ namespace Discord.Rest | |||||
ApiClient.RequestQueue.RateLimitTriggered += async (id, info) => | ApiClient.RequestQueue.RateLimitTriggered += async (id, info) => | ||||
{ | { | ||||
if (info == null) | 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 | 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); | ApiClient.SentRequest += async (method, endpoint, millis) => await _restLogger.VerboseAsync($"{method} {endpoint}: {millis} ms").ConfigureAwait(false); | ||||
} | } | ||||
@@ -24,7 +24,7 @@ namespace Discord.API | |||||
{ | { | ||||
internal class DiscordRestApiClient : IDisposable | 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); } } | 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>>(); | private readonly AsyncEvent<Func<string, string, double, Task>> _sentRequestEvent = new AsyncEvent<Func<string, string, double, Task>>(); | ||||
@@ -176,9 +176,9 @@ namespace Discord.API | |||||
//Core | //Core | ||||
internal Task SendAsync(string method, Expression<Func<string>> endpointExpr, BucketIds ids, | internal Task SendAsync(string method, Expression<Func<string>> endpointExpr, BucketIds ids, | ||||
ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null, [CallerMemberName] string funcName = null) | 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, | 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 = options ?? new RequestOptions(); | ||||
options.HeaderOnly = true; | options.HeaderOnly = true; | ||||
@@ -190,9 +190,9 @@ namespace Discord.API | |||||
internal Task SendJsonAsync(string method, Expression<Func<string>> endpointExpr, object payload, BucketIds ids, | internal Task SendJsonAsync(string method, Expression<Func<string>> endpointExpr, object payload, BucketIds ids, | ||||
ClientBucketType clientBucket = ClientBucketType.Unbucketed, RequestOptions options = null, [CallerMemberName] string funcName = null) | 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, | 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 = options ?? new RequestOptions(); | ||||
options.HeaderOnly = true; | 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, | 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) | 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, | 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 = options ?? new RequestOptions(); | ||||
options.HeaderOnly = true; | options.HeaderOnly = true; | ||||
@@ -219,9 +219,9 @@ namespace Discord.API | |||||
internal Task<TResponse> SendAsync<TResponse>(string method, Expression<Func<string>> endpointExpr, BucketIds ids, | 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 | 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, | 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 = options ?? new RequestOptions(); | ||||
options.BucketId = bucketId; | 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, | 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 | 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, | 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 = options ?? new RequestOptions(); | ||||
options.BucketId = bucketId; | 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, | 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) | 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, | 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 = options ?? new RequestOptions(); | ||||
options.BucketId = bucketId; | 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)); | 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); | 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> | /// <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) | 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)); | 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) | 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 GuildId { get; internal set; } | ||||
public ulong ChannelId { 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; | GuildId = guildId; | ||||
ChannelId = channelId; | ChannelId = channelId; | ||||
WebhookId = webhookId; | |||||
} | } | ||||
internal object[] ToArray() | 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) | internal static int? GetIndex(string name) | ||||
{ | { | ||||
switch (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: | default: | ||||
return null; | return null; | ||||
} | } | ||||
@@ -1491,18 +1511,19 @@ namespace Discord.API | |||||
{ | { | ||||
return endpointExpr.Compile()(); | 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); | 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 | try | ||||
{ | { | ||||
//Is this a constant string? | //Is this a constant string? | ||||
if (endpoint.Body.NodeType == ExpressionType.Constant) | 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 builder = new StringBuilder(); | ||||
var methodCall = endpoint.Body as MethodCallExpression; | var methodCall = endpoint.Body as MethodCallExpression; | ||||
@@ -1539,7 +1560,7 @@ namespace Discord.API | |||||
var mappedId = BucketIds.GetIndex(fieldName); | 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++; | rightIndex++; | ||||
if (mappedId.HasValue) | if (mappedId.HasValue) | ||||
@@ -1552,7 +1573,7 @@ namespace Discord.API | |||||
format = builder.ToString(); | 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) | catch (Exception ex) | ||||
{ | { | ||||
@@ -10,14 +10,14 @@ namespace Discord.Net.Queue | |||||
internal struct ClientBucket | internal struct ClientBucket | ||||
{ | { | ||||
private static readonly ImmutableDictionary<ClientBucketType, ClientBucket> DefsByType; | private static readonly ImmutableDictionary<ClientBucketType, ClientBucket> DefsByType; | ||||
private static readonly ImmutableDictionary<string, ClientBucket> DefsById; | |||||
private static readonly ImmutableDictionary<BucketId, ClientBucket> DefsById; | |||||
static ClientBucket() | static ClientBucket() | ||||
{ | { | ||||
var buckets = new[] | 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>(); | var builder = ImmutableDictionary.CreateBuilder<ClientBucketType, ClientBucket>(); | ||||
@@ -25,21 +25,21 @@ namespace Discord.Net.Queue | |||||
builder.Add(bucket.Type, bucket); | builder.Add(bucket.Type, bucket); | ||||
DefsByType = builder.ToImmutable(); | DefsByType = builder.ToImmutable(); | ||||
var builder2 = ImmutableDictionary.CreateBuilder<string, ClientBucket>(); | |||||
var builder2 = ImmutableDictionary.CreateBuilder<BucketId, ClientBucket>(); | |||||
foreach (var bucket in buckets) | foreach (var bucket in buckets) | ||||
builder2.Add(bucket.Id, bucket); | builder2.Add(bucket.Id, bucket); | ||||
DefsById = builder2.ToImmutable(); | DefsById = builder2.ToImmutable(); | ||||
} | } | ||||
public static ClientBucket Get(ClientBucketType type) => DefsByType[type]; | 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 ClientBucketType Type { get; } | ||||
public string Id { get; } | |||||
public BucketId Id { get; } | |||||
public int WindowCount { get; } | public int WindowCount { get; } | ||||
public int WindowSeconds { 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; | Type = type; | ||||
Id = id; | Id = id; | ||||
@@ -12,9 +12,9 @@ namespace Discord.Net.Queue | |||||
{ | { | ||||
internal class RequestQueue : IDisposable | 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 SemaphoreSlim _tokenLock; | ||||
private readonly CancellationTokenSource _cancelTokenSource; //Dispose token | private readonly CancellationTokenSource _cancelTokenSource; //Dispose token | ||||
private CancellationTokenSource _clearToken; | private CancellationTokenSource _clearToken; | ||||
@@ -34,7 +34,7 @@ namespace Discord.Net.Queue | |||||
_requestCancelToken = CancellationToken.None; | _requestCancelToken = CancellationToken.None; | ||||
_parentToken = CancellationToken.None; | _parentToken = CancellationToken.None; | ||||
_buckets = new ConcurrentDictionary<string, RequestBucket>(); | |||||
_buckets = new ConcurrentDictionary<BucketId, object>(); | |||||
_cleanupTask = RunCleanup(); | _cleanupTask = RunCleanup(); | ||||
} | } | ||||
@@ -82,7 +82,7 @@ namespace Discord.Net.Queue | |||||
else | else | ||||
request.Options.CancelToken = _requestCancelToken; | 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); | var result = await bucket.SendAsync(request).ConfigureAwait(false); | ||||
createdTokenSource?.Dispose(); | createdTokenSource?.Dispose(); | ||||
return result; | return result; | ||||
@@ -110,14 +110,32 @@ namespace Discord.Net.Queue | |||||
_waitUntil = DateTimeOffset.UtcNow.AddMilliseconds(info.RetryAfter.Value + (info.Lag?.TotalMilliseconds ?? 0.0)); | _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); | 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() | private async Task RunCleanup() | ||||
{ | { | ||||
@@ -126,10 +144,15 @@ namespace Discord.Net.Queue | |||||
while (!_cancelTokenSource.IsCancellationRequested) | while (!_cancelTokenSource.IsCancellationRequested) | ||||
{ | { | ||||
var now = DateTimeOffset.UtcNow; | 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 ((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 _); | _buckets.TryRemove(bucket.Id, out _); | ||||
} | |||||
} | } | ||||
await Task.Delay(60000, _cancelTokenSource.Token).ConfigureAwait(false); //Runs each minute | await Task.Delay(60000, _cancelTokenSource.Token).ConfigureAwait(false); //Runs each minute | ||||
} | } | ||||
@@ -19,12 +19,13 @@ namespace Discord.Net.Queue | |||||
private readonly RequestQueue _queue; | private readonly RequestQueue _queue; | ||||
private int _semaphore; | private int _semaphore; | ||||
private DateTimeOffset? _resetTick; | 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 int WindowCount { get; private set; } | ||||
public DateTimeOffset LastAttemptAt { 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; | _queue = queue; | ||||
Id = id; | Id = id; | ||||
@@ -32,7 +33,7 @@ namespace Discord.Net.Queue | |||||
_lock = new object(); | _lock = new object(); | ||||
if (request.Options.IsClientBucket) | if (request.Options.IsClientBucket) | ||||
WindowCount = ClientBucket.Get(request.Options.BucketId).WindowCount; | |||||
WindowCount = ClientBucket.Get(Id).WindowCount; | |||||
else | else | ||||
WindowCount = 1; //Only allow one request until we get a header back | WindowCount = 1; //Only allow one request until we get a header back | ||||
_semaphore = WindowCount; | _semaphore = WindowCount; | ||||
@@ -52,6 +53,8 @@ namespace Discord.Net.Queue | |||||
{ | { | ||||
await _queue.EnterGlobalAsync(id, request).ConfigureAwait(false); | await _queue.EnterGlobalAsync(id, request).ConfigureAwait(false); | ||||
await EnterAsync(id, request).ConfigureAwait(false); | await EnterAsync(id, request).ConfigureAwait(false); | ||||
if (_redirectBucket != null) | |||||
return await _redirectBucket.SendAsync(request); | |||||
#if DEBUG_LIMITS | #if DEBUG_LIMITS | ||||
Debug.WriteLine($"[{id}] Sending..."); | Debug.WriteLine($"[{id}] Sending..."); | ||||
@@ -160,6 +163,9 @@ namespace Discord.Net.Queue | |||||
while (true) | while (true) | ||||
{ | { | ||||
if (_redirectBucket != null) | |||||
break; | |||||
if (DateTimeOffset.UtcNow > request.TimeoutAt || request.Options.CancelToken.IsCancellationRequested) | if (DateTimeOffset.UtcNow > request.TimeoutAt || request.Options.CancelToken.IsCancellationRequested) | ||||
{ | { | ||||
if (!isRateLimited) | if (!isRateLimited) | ||||
@@ -175,7 +181,8 @@ namespace Discord.Net.Queue | |||||
} | } | ||||
DateTimeOffset? timeoutAt = request.TimeoutAt; | 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) | if (!isRateLimited) | ||||
{ | { | ||||
@@ -210,20 +217,52 @@ namespace Discord.Net.Queue | |||||
} | } | ||||
#if DEBUG_LIMITS | #if DEBUG_LIMITS | ||||
else | else | ||||
Debug.WriteLine($"[{id}] Entered Semaphore ({_semaphore}/{WindowCount} remaining)"); | |||||
Debug.WriteLine($"[{id}] Entered Semaphore ({semaphore}/{WindowCount} remaining)"); | |||||
#endif | #endif | ||||
break; | 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) | if (WindowCount == 0) | ||||
return; | return; | ||||
lock (_lock) | 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; | 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) | if (info.Limit.HasValue && WindowCount != info.Limit.Value) | ||||
{ | { | ||||
WindowCount = info.Limit.Value; | WindowCount = info.Limit.Value; | ||||
@@ -233,7 +272,6 @@ namespace Discord.Net.Queue | |||||
#endif | #endif | ||||
} | } | ||||
var now = DateTimeOffset.UtcNow.ToUnixTimeSeconds(); | |||||
DateTimeOffset? resetTick = null; | DateTimeOffset? resetTick = null; | ||||
//Using X-RateLimit-Remaining causes a race condition | //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)"); | Debug.WriteLine($"[{id}] Retry-After: {info.RetryAfter.Value} ({info.RetryAfter.Value} ms)"); | ||||
#endif | #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) | else if (info.Reset.HasValue) | ||||
{ | { | ||||
resetTick = info.Reset.Value.AddSeconds(info.Lag?.TotalSeconds ?? 1.0); | 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) | if (request.Options.IsReactionBucket) | ||||
resetTick = DateTimeOffset.Now.AddMilliseconds(250); | 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)"); | Debug.WriteLine($"[{id}] X-RateLimit-Reset: {info.Reset.Value.ToUnixTimeSeconds()} ({diff} ms, {info.Lag?.TotalMilliseconds} ms lag)"); | ||||
#endif | #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 | #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 | #endif | ||||
} | } | ||||
if (resetTick == null) | 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 | #if DEBUG_LIMITS | ||||
Debug.WriteLine($"[{id}] Disabled Semaphore"); | Debug.WriteLine($"[{id}] Disabled Semaphore"); | ||||
#endif | #endif | ||||
@@ -11,7 +11,8 @@ namespace Discord.Net | |||||
public int? Remaining { get; } | public int? Remaining { get; } | ||||
public int? RetryAfter { get; } | public int? RetryAfter { get; } | ||||
public DateTimeOffset? Reset { get; } | public DateTimeOffset? Reset { get; } | ||||
public TimeSpan? ResetAfter { get; } | |||||
public TimeSpan? ResetAfter { get; } | |||||
public string Bucket { get; } | |||||
public TimeSpan? Lag { get; } | public TimeSpan? Lag { get; } | ||||
internal RateLimitInfo(Dictionary<string, string> headers) | 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; | double.TryParse(temp, NumberStyles.AllowDecimalPoint, CultureInfo.InvariantCulture, out var reset) ? DateTimeOffset.FromUnixTimeMilliseconds((long)(reset * 1000)) : (DateTimeOffset?)null; | ||||
RetryAfter = headers.TryGetValue("Retry-After", out temp) && | RetryAfter = headers.TryGetValue("Retry-After", out temp) && | ||||
int.TryParse(temp, NumberStyles.None, CultureInfo.InvariantCulture, out var retryAfter) ? retryAfter : (int?)null; | 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) && | Lag = headers.TryGetValue("Date", out temp) && | ||||
DateTimeOffset.TryParse(temp, CultureInfo.InvariantCulture, DateTimeStyles.None, out var date) ? DateTimeOffset.UtcNow - date : (TimeSpan?)null; | DateTimeOffset.TryParse(temp, CultureInfo.InvariantCulture, DateTimeStyles.None, out var date) ? DateTimeOffset.UtcNow - date : (TimeSpan?)null; | ||||
} | } | ||||
@@ -77,9 +77,9 @@ namespace Discord.Webhook | |||||
ApiClient.RequestQueue.RateLimitTriggered += async (id, info) => | ApiClient.RequestQueue.RateLimitTriggered += async (id, info) => | ||||
{ | { | ||||
if (info == null) | 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 | 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); | ApiClient.SentRequest += async (method, endpoint, millis) => await _restLogger.VerboseAsync($"{method} {endpoint}: {millis} ms").ConfigureAwait(false); | ||||
} | } | ||||