Sliding Window Counter using Redis Streams

The Case for Sliding Window Counters

A sliding window counter is used to track the frequency of recurring actions that occur within a specific period. One common application of a sliding window counter is in rate-limiting algorithms, where it is used to keep track of the number of API calls made by a client. By comparing this count against a predetermined threshold, excessive calls from the same client can be limited or throttled.

The counter-sliding-window Node package provides an implementation of both a sliding window counter and a distributed sliding window counter.

The basic sliding window counter functions by keeping a record of the number of API invocation calls per client in memory. This record is maintained in a list, which is ordered based on the timestamp of each call in descending order. Whenever a client makes a new call, a new entry is added to the beginning of their corresponding list.

To determine the number of API invocations made by a client within the sliding window period, the get method traverses the list of calls and sums the entities that fall within the current sliding window period. To avoid memory usage issues, the counter automatically clears entities that are older than the sliding window period whenever the get or add methods are called.

import { SlidingWindowCounterLocal } from "counter-sliding-window"

const sliding = new SlidingWindowCounterLocal(5, "seconds")
// [Time 0]
sliding.add(1)
// After 3 seconds [Time 3]
sliding.add(2)

// After 1 more second [Time 4]
console.log(sliding.get()) // prints 3

// After 3 more seconds [Time 7]
console.log(sliding.get()) // prints 2

// After 2 more seconds [Time 9]
console.log(sliding.get()) // prints 0

Redis Implementation

An in-memory counter can effectively handle the task of counting actions in a single server. However, it is not suitable for a distributed system where multiple servers are processing incoming requests. In this scenario, a shared storage solution is necessary, and Redis is an excellent choice to manage these counters due to its shared nature and high speed.

Sorted Set implementation

To implement a sliding window counter in Redis, we can use Sorted Sets, which are commonly used in Redis to represent time series data. Sorted Sets store unique members with a score per member, allowing for easy counting and fetching of elements from the set within a specified score range. Since timestamps are naturally ordered, we can use the score part of the Sorted Set to store the timestamp of each API call invocation, making it easy to count and fetch calls made within a timestamp range equal to the sliding window period.

To add a new API call invocation count to our counter Redis key, we can use the ZADD command, as shown below:

-- ZADD counter [timestamp] [value]
ZADD counter 1657961915 1

To count the number of API call invocations in the last sliding window period, we can use the ZCOUNT command with a max score of now in milliseconds and a min score of now - window_size, where now is the current timestamp and window_size is the duration of the sliding window period in milliseconds, as shown below:

-- ZCOUNT counter [now timestamp] [now timestamp minus window period size of 1 minute]
ZCOUNT counter 1672499174 1672499114

Sorted Set Limitations

One of the limitations of using a sorted set is that it is primarily a set, which means that it can store a value only once. For example, if we count API call invocations in a one-minute sliding window, with the timestamp as the score and the count number as the value, we may miss API invocations. If we add two calls within a minute, with each call representing a single API invocation, we expect to have two separate elements, one for each timestamp, with a value of 1. However, since it is a set, the value 1 can only be stored once, causing the second element to override the first and resulting in a missed request count.

ZADD counter 1657961915 1
ZADD counter 1657961937 1

To address this issue, we can add the timestamp as a prefix to the value, enabling us to store the same value more than once. However, it may still encounter problems when receiving more than one request at the same timestamp, which is possible in a distributed environment. A complete solution requires mixing the value with some random prefix.

ZADD counter 1657961915 1657961915:1
ZADD counter 1657961937 1657961937:1

Cleaning elements that are older than the current sliding window is also a challenge with the Sorted Set implementation. For a comprehensive and detailed implementation of a sliding window counter using Sorted Sets, you can refer to How to implement Sliding Window Rate Limiting app using ASP.NET Core & Redis.

Using Redis Streams

Redis Streams were introduced with version 5 and are an append-only data structure designed for time-series data, such as logs or events. They offer a parallel data consumption model similar to Kafka topics and provide a more robust pub-sub implementation than the SUBSCRIBE/PUBLISH alternative.

Redis Streams are also suitable for implementing sliding window counters since they support adding entries sorted by time and querying entries by a time range.

To add an entry to a stream, we can use the following command:

XADD couter_key * count "1"

This adds an entry to the counter_key stream with a structure similar to count=1. The * tells Redis to generate an automatic ID for the entry, with a structure like [redis current Unix timestamp]-[sequence] (for example, 1657985420-0). By calling this command for every incoming request, we can log count=1 at the Redis server Unix timestamp.

Redis Streams allow us to fetch entries by a time range. For a time window of 5 seconds (5000 ms), we can request the range of entries between now and now - 5000 with the following command:

XRANGE couter_key 1657987420 +

The 1657987420 + specifies an entry ID range starting at 1657987420 and ending with the maximum ID in the stream. Because we let Redis auto-generate IDs with Unix timestamps, this is a query on the time range of [1657987420, now].

The XRANGE command returns a list of entries that look like this:

1) 1) 1657985420-0
   2) 1) "count"
      2) "1"
2) 1) 1657985820-0
   2) 1) "count"
      2) "1"

By summing the count fields of the entries in the time window range, we can determine the request count we’re looking for. We’ll discuss the actual implementation of this later. First, let’s talk about cleaning the stream from old entries.

Cleaning old entries

When a heavily loaded system receives a lot of requests, Redis can quickly fill up with counter entries. It is essential to keep the stream focused around the relevant time range entries.

To remove all entries with IDs before 1657987420 Unix timestamp (the MINID), we use the XTRIM command.

XTRIM counter_key MINID ~ 1657987420

The ~ operand allows Redis to optimize for the actual number of entities to remove, not removing every old entry. We’re not concerned with removing every old entry since the XRANGE returns only the relevant ones anyway.

We can combine adding new entries and cleaning old entries in the same XADD command:

XADD MINID ~ 1657987420 couter_key * count "1"

Another critical cleanup involves removing old counter streams that store Redis keys older than the time window range. We use the Redis EXPIRE command to set a timeout for the Redis key holding the stream. This command tells Redis to remove the stream once the timeout has passed.

To ensure that the key expiration remains current, we set EXPIRE on the stream key with a timeout value equal to the sliding window period. When we post within the sliding window period, the new EXPIRE timeout value replaces the previous one, directing Redis to keep the key around for at least the next sliding window period. If we don’t post to the stream within the next sliding window period, Redis will remove the key automatically.

We can combine XADD with EXPIRE to ensure that unused streams disappear automatically within a time window period of 5 seconds (5000 ms):

XADD MINID ~ 1657987420 couter_key * count "1"
EXPIRE couter_key 5000

Lua script

To ensure accurate results, the calculation of the time window range, rate limit check, and logging of a new API invocation must be executed atomically. This can be accomplished by using a Lua script, which, by nature, runs within a boundary of a Redis transaction. The provided Lua script fetches a stream range from the start window id until the latest id, sums the range counters, checks if the count exceeds the rate limitation, adds the counter while keeping the stream free of old entries, and sets the stream expiration.

-- `now` is an array with first item as the linux timestamp
local now = redis.call('TIME')
-- ARGV[1] holds the time window size in milliseconds
local start_window = tonumber(now[1]) - tonumber(ARGV[1])

-- Fetching stream range from start window id till latest id
local range = redis.call('XRANGE', KEYS[1], start_window, '+')

-- Summing the range counters
local count = 0;
for _, item in ipairs(range) do
    count = count + tonumber(item[2][2])
end

-- ARGV[3] holds the rate limitation
if tonumber(ARGV[3]) > 0 and count >= ARGV[3] then
    return -1
end

-- Adding the counter and keeping the stream cleaned of old entries
redis.call('XADD', KEYS[1], 'MINID', '~', start_window, '*', 'count', ARGV[3])
-- Setting stream expiration
redis.call('EXPIRE', KEYS[1], ARGV[1])
return 1

Usage

For a NodeJS implementation with ioredis library, a new command streamCounterAdd is defined to execute the Lua script.

import Redis, { Callback, RedisOptions, Result } from "ioredis"

declare module "ioredis" {
  interface RedisCommander<Context> {
    streamCounterAdd(
      key: string,
      windowMs: string,
      count: string,
      limit: string,
      callback?: Callback<string>
    ): Result<string, Context>
  }
}

const redis = new Redis({
  scripts: {
    streamCounterAdd: {
      numberOfKeys: 1,
      lua: `
                local now = redis.call('TIME')
                local start_window = tonumber(now[1]) - tonumber(ARGV[1])
                local range = redis.call('XRANGE', KEYS[1], start_window, '+')

                local count = 0;
                for _, item in ipairs(range) do
                    count = count + tonumber(item[2][2])
                end

                if tonumber(ARGV[3]) > 0 and count >= ARGV[3] then
                    return -1
                end

                redis.call('XADD', KEYS[1], 'MINID', '~', start_window, '*', 'count', ARGV[3])
                redis.call('EXPIRE', KEYS[1], ARGV[1])
                return 1
            `,
    },
  },
})

The function throttleRequest uses the streamCounterAdd command to apply rate limitation per client and returns true if the request should be throttled.

async function throttleRequest(
  counterName: string,
  windowPeriondMS: number,
  limit: number
): Promise<bool> {
  const res = await this.redis.streamCounterAdd(
    counterName,
    windowPeriondMS.toString(),
    "1",
    limit.toString()
  )
  return parseInt(res) < 0
}

Rate limiter

An example implementation of API rate limitation for Express using counter-sliding-window library uses the client token as the counter name:

import { SlidingWindowCounterRedis } from "counter-sliding-window"

app.get("/do-something", async (req, res) => {
  const authToken = req.headers["authorization"]
  if (await throttleRequest(authToken, 5000, 10)) {
    // HTTP 429 Too Many Requests
    return res.sendStatus(429)
  }

  // Do the actual stuff ...
})

A middleware rateLimiter can be defined to apply the rate limitation to all the handlers. The middleware calls the throttleRequest function to check if the request should be throttled and responds with HTTP 429 Too Many Requests if the request is throttled.

function rateLimiter(windowPeriondMS: number, limit: number) {
  return async function (
    request: Request,
    response: Response,
    next: NextFunction
  ): Promise<void> {
    const authToken = req.headers["authorization"]
    if (await throttleRequest(authToken, windowPeriondMS, limit)) {
      // HTTP 429 Too Many Requests
      return res.sendStatus(429)
    }
    next()
  }
}

app.use(rateLimiter(5000, 10))