Task Rate Limiting

Example of Configuring asynq Server to Implement Task Processing Rate Limiting

This page shows how to configure the asynq server to implement task processing rate limiting.

Please note that this is the rate limit for each server instance, not a global rate limit.

In this example, we will use the golang.org/x/time/rate package to demonstrate rate limiting. The key configurations in your server initialization settings are IsFailure and RetryDelayFunc. We will create a custom error type and type assert the given error in the IsFailure and RetryDelayFunc functions.

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"math/rand"
	"time"

	"golang.org/x/time/rate"
	"github.com/hibiken/asynq"
)

func main() {
	srv := asynq.NewServer(
		asynq.RedisClientOpt{Addr: ":6379"},
		asynq.Config{
			Concurrency:    10,
			// Do not count an error as a failure if it is due to rate limiting.
			IsFailure:      func(err error) bool { return !IsRateLimitError(err) },
			RetryDelayFunc: retryDelay,
		},
	)

	if err := srv.Run(asynq.HandlerFunc(handler)); err != nil {
		log.Fatal(err)
	}
}

type RateLimitError struct {
	RetryIn time.Duration
}

func (e *RateLimitError) Error() string {
	return fmt.Sprintf("Rate limit reached (retry in %v)", e.RetryIn)
}

func IsRateLimitError(err error) bool {
	_, ok := err.(*RateLimitError)
	return ok
}

func retryDelay(n int, err error, task *asynq.Task) time.Duration {
	var ratelimitErr *RateLimitError
	if errors.As(err, &ratelimitErr) {
		return ratelimitErr.RetryIn
	}
	return asynq.DefaultRetryDelayFunc(n, err, task)
}

// Rate limit of 10 events per second, allowing bursts of up to 30 events.
var limiter = rate.NewLimiter(10, 30)

func handler(ctx context.Context, task *asynq.Task) error {
	if !limiter.Allow() {
		return &RateLimitError{
			RetryIn: time.Duration(rand.Intn(10)) * time.Second,
		}
	}
	log.Printf("[*] Handling task %s", task.Payload())
	return nil
}