Determine which limiter rejects when using both ConcurrenceLimiter and RateLimiter

[eisendle]

Hi,

I'm creating following pipeline which consists of a ConcurrencyLimiter and a RateLimiter:

 var limiter = PartitionedRateLimiter.Create<ResilienceContext, string>(context =>
 {
     string partitionKey = GetPartitionKey(context);

     return RateLimitPartition.GetTokenBucketLimiter(
         partitionKey,
         key => options.Value.RateLimiterOptions);
 });

 return new ResiliencePipelineBuilder()
             .AddConcurrencyLimiter(100,10)
             .AddRateLimiter(new RateLimiterStrategyOptions
             {
                 RateLimiter = args => limiter.AcquireAsync(args.Context, 1, args.Context.CancellationToken),
                 OnRejected = x => OnRejected(x,logger),
                 Name="RateLimiter"
             }).Build();

 static string GetPartitionKey(ResilienceContext context) 
     => context.Properties.GetValue(UserProperty, "Default");

When the pipeline rejects a request by throwing a RateLimiterRejectedException there seems no way to find out which of the limiters caused the rejection. In my use case this is important as I want to send different replies when the concurrency limit is hit and the rate limit of for a user has been exceeded.

Is there a way to know which limiter caused the rejection?

[eisendle]

Found a solution:

In the OnRejected callback, set a ResilienceContext property indicating the rate limit has been hit. If the pipeline throws a RateLimiterRejectedException check the ResilienceContext if the property exists. If it exists, the rate limit has been hit, otherwise the concurrency limit has been hit.

Example:

private static readonly ResiliencePropertyKey<string> UserProperty = new("UserName");
private static readonly ResiliencePropertyKey<bool> RateLimitHitProperty = new("RateLimitHit");
private readonly ResiliencePipeline _resiliencePipeline = CreateResiliencePipeline(options);

private static ResiliencePipeline CreateResiliencePipeline(IOptions<MyOptions> options, ILogger logger)
{
    var limiter = PartitionedRateLimiter.Create<ResilienceContext, string>(context =>
    {
        string partitionKey = GetPartitionKey(context);

        return RateLimitPartition.GetTokenBucketLimiter(
            partitionKey,
            key => options.Value.RateLimiterOptions);
    });

    return new ResiliencePipelineBuilder()
                .AddConcurrencyLimiter(options.Value.ConcurrencyLimiterOptions)
                .AddRateLimiter(new RateLimiterStrategyOptions
                {
                    RateLimiter = args => limiter.AcquireAsync(args.Context, 1, args.Context.CancellationToken),
                    OnRejected = OnRejected,
                    Name="RateLimiter"
                }).Build();

    static string GetPartitionKey(ResilienceContext context) 
        => context.Properties.GetValue(UserProperty, "Default");
}

private static ValueTask OnRejected(OnRateLimiterRejectedArguments arguments)
{
    //Set property to indicate that the RateLimit has been hit 
    arguments.Context.Properties.Set(RateLimitHitProperty, true);
    return ValueTask.CompletedTask;
}
  

public async Task Handle(Request req)
{
    var resilienceContext = ResilienceContextPool.Shared.Get(CancellationToken.None);

    resilienceContext.Properties.Set(UserProperty, request.User!);

    try
    {
        var outcome = await _resiliencePipeline.ExecuteOutcomeAsync(static async (ctx, state) =>
        {
            try
            {
                await HandleRequest(req); //Code for HandleRequest omitted for brevity
                return Outcome.FromResult<int>(0);
            }
            catch (Exception ex)
            {
                return Outcome.FromException<Unit>(ex);
            }
        }, resilienceContext, string.Empty);

        if (outcome.Exception is RateLimiterRejectedException rex)
        {
            if (resilienceContext.Properties.TryGetValue(RateLimitHitProperty, out var _))
                await req.SendErrorReply(-1, $"Rate limit exceeded, retry after: {rex.RetryAfter}.");
            else
                await req.SendErrorReply(-1, $"Service is busy, try again later.");
        }
    }
    finally
    {
        ResilienceContextPool.Shared.Return(resilienceContext);
    }
}

[martintmk]

This is perfectly valid solution. Alternatively, we can include the ResilienceSource property of type ResilienceTelemetrySource to ExecutionRejectedException that allow uniquely identify the strategy that caused the exception.

We can consider this of more requests like this come up.

Polly/src/Polly.Core/Telemetry/ResilienceTelemetrySource.cs

Line 9 in 41fd38c

public sealed class ResilienceTelemetrySource

[peter-csala]

Can you please elaborate why do you need both kind of limiters in the same pipeline?

[eisendle]

The ConcurrencyLimiter is used to prevent overload of the service by limiting the number of globally concurrently handled requests. The RateLimiter limits the number of requests per user.

Additional context: A request in my use case is a incoming messages from a message queue