Implementing Redis Sliding Window Rate Limiting

· 9 min read ·
Implementing Redis Sliding Window Rate Limiting

What You’ll Need

  • n8n Cloud or self-hosted n8n instance
  • Hetzner VPS or Contabo VPS for hosting Redis
  • DigitalOcean as an alternative hosting provider
  • Redis server (6.0 or later recommended)
  • Python 3.8 or higher
  • Basic knowledge of HTTP requests and time-based algorithms

Table of Contents

Understanding Sliding Window Rate Limiting

Rate limiting protects your API and services from abuse, brute force attacks, and resource exhaustion. I’ve implemented dozens of rate limiting strategies, and the sliding window approach remains one of the most effective for balancing accuracy with performance.

Unlike fixed window counters that reset at specific intervals (often causing burst problems at window boundaries), sliding window rate limiting maintains a continuous, time-based perspective. Imagine a 60-second window that moves forward with each request, always looking back exactly 60 seconds. This eliminates the “window boundary reset” vulnerability where attackers can make twice as many requests by timing them around the reset moment.

The sliding window approach counts all requests within the current time window. If you allow 100 requests per minute, the system tracks when the oldest request within your current 60-second window occurred. New requests are rejected if adding them would exceed your limit within that continuous window.

This technique is particularly valuable when protecting endpoints that process payment transactions, authenticate users, or trigger resource-intensive operations. Combined with proper monitoring via systems like centralized server logging with Grafana Loki, you can identify attack patterns and adjust thresholds in real time.

Redis Architecture for Rate Limiting

Redis excels at rate limiting because it’s in-memory, blazingly fast, and offers atomic operations. I typically deploy Redis on a Hetzner VPS or Contabo VPS for cost-effective, low-latency access.

The core Redis data structure for sliding window is a sorted set, where the score represents the timestamp of each request. Here’s the conceptual flow:

  1. Store each request timestamp as a member in a Redis sorted set, with the timestamp as both the member and score
  2. When a new request arrives, remove all members with scores older than (current_time - window_duration)
  3. Check if the remaining member count is below your limit
  4. If below limit, add the new timestamp and allow the request
  5. If at or above limit, reject the request

This approach requires minimal memory (storing only timestamps) and executes in O(log n) time for most operations.

For applications handling millions of requests, if you’re concerned about memory constraints on smaller VPS instances, reviewing swap space configuration on low memory VPS helps prevent out-of-memory scenarios during traffic spikes.

Let me show you the actual implementation.

Implementing Sliding Window in Python

I’ll walk you through a production-ready Python implementation using the redis-py library.

First, install the required package:

pip install redis

Here’s the core rate limiter class:

import redis
import time
from typing import Tuple

class SlidingWindowRateLimiter:
    def __init__(self, redis_host: str = 'localhost', redis_port: int = 6379, 
                 redis_db: int = 0, redis_password: str = None):
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port,
            db=redis_db,
            password=redis_password,
            decode_responses=True,
            socket_connect_timeout=5,
            socket_keepalive=True
        )
        self.redis_client.ping()
    
    def is_allowed(self, identifier: str, max_requests: int, 
                   window_seconds: int) -> Tuple[bool, dict]:
        """
        Check if a request is allowed under the sliding window rate limit.
        
        Args:
            identifier: Unique identifier (IP, user ID, API key)
            max_requests: Maximum requests allowed in window
            window_seconds: Time window in seconds
            
        Returns:
            Tuple of (allowed: bool, info: dict with remaining requests and reset time)
        """
        current_time = time.time()
        window_start = current_time - window_seconds
        key = f"rate_limit:{identifier}"
        
        pipe = self.redis_client.pipeline()
        
        pipe.zremrangebyscore(key, 0, window_start)
        pipe.zcard(key)
        pipe.zadd(key, {str(current_time): current_time})
        pipe.expire(key, window_seconds + 1)
        
        results = pipe.execute()
        
        request_count = results[1]
        
        allowed = request_count < max_requests
        remaining = max(0, max_requests - request_count - 1)
        reset_time = current_time + window_seconds
        
        return allowed, {
            'remaining': remaining,
            'reset_time': reset_time,
            'reset_in_seconds': window_seconds,
            'limit': max_requests,
            'current_count': request_count + 1
        }
    
    def get_status(self, identifier: str, window_seconds: int) -> dict:
        """Get current status without making a request."""
        current_time = time.time()
        window_start = current_time - window_seconds
        key = f"rate_limit:{identifier}"
        
        current_count = self.redis_client.zcount(key, window_start, current_time)
        
        oldest_request = self.redis_client.zrange(key, 0, 0, withscores=True)
        reset_time = None
        if oldest_request:
            reset_time = oldest_request[0][1] + window_seconds
        
        return {
            'current_count': current_count,
            'reset_time': reset_time,
            'reset_in_seconds': reset_time - current_time if reset_time else None
        }
    
    def reset(self, identifier: str) -> None:
        """Manually reset rate limit for an identifier."""
        key = f"rate_limit:{identifier}"
        self.redis_client.delete(key)
    
    def close(self):
        """Close Redis connection."""
        self.redis_client.close()

Now let’s integrate this into a Flask application:

from flask import Flask, request, jsonify, Response
from functools import wraps
import logging

app = Flask(__name__)
limiter = SlidingWindowRateLimiter(
    redis_host='localhost',
    redis_port=6379,
    redis_db=0
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def rate_limit(max_requests: int = 100, window_seconds: int = 60):
    """Decorator for rate limiting endpoints."""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            identifier = request.remote_addr
            api_key = request.headers.get('X-API-Key')
            if api_key:
                identifier = f"api_key:{api_key}"
            
            allowed, info = limiter.is_allowed(identifier, max_requests, window_seconds)
            
            response = Response(
                response=None,
                status=200,
                headers={
                    'X-RateLimit-Limit': str(info['limit']),
                    'X-RateLimit-Remaining': str(info['remaining']),
                    'X-RateLimit-Reset': str(int(info['reset_time']))
                }
            )
            
            if not allowed:
                logger.warning(f"Rate limit exceeded for {identifier}")
                response.status_code = 429
                response.data = jsonify({
                    'error': 'Rate limit exceeded',
                    'retry_after': info['reset_in_seconds']
                }).get_data()
                return response
            
            response.status_code = 200
            return f(*args, **kwargs)
        
        return decorated_function
    return decorator

@app.route('/api/data', methods=['GET'])
@rate_limit(max_requests=100, window_seconds=60)
def get_data():
    return jsonify({'data': 'This is protected data'})

@app.route('/api/status', methods=['GET'])
def status():
    identifier = request.remote_addr
    api_key = request.headers.get('X-API-Key')
    if api_key:
        identifier = f"api_key:{api_key}"
    
    status_info = limiter.get_status(identifier, window_seconds=60)
    return jsonify(status_info)

@app.route('/api/reset', methods=['POST'])
def reset_limit():
    identifier = request.remote_addr
    api_key = request.headers.get('X-API-Key')
    if api_key:
        identifier = f"api_key:{api_key}"
    
    limiter.reset(identifier)
    logger.info(f"Rate limit reset for {identifier}")
    return jsonify({'message': 'Rate limit reset'})

if __name__ == '__main__':
    app.run(debug=False, host='0.0.0.0', port=5000)

The implementation uses Redis pipelining to execute multiple commands atomically. The zremrangebyscore removes expired entries, zcard counts remaining requests, and zadd records the new request timestamp.

💡 Fast-Track Your Project: Don’t want to configure this yourself? I build custom n8n pipelines and bots. Message me with code SYS3-HUGO.

Building an n8n Workflow

n8n Cloud lets you automate rate limiting checks across multiple services without code. Here’s how I structure a production workflow:

{
  "name": "Rate Limit Enforcement Workflow",
  "nodes": [
    {
      "parameters": {},
      "name": "Webhook Trigger",
      "type": "n8n-nodes-base.webhook",
      "typeVersion": 1,
      "position": [250, 300],
      "webhookId": "your-webhook-id"
    },
    {
      "parameters": {
        "resource": "command",
        "command": "custom",
        "customCommand": "SCRIPT LOAD 'return redis.call(\"ZREMRANGEBYSCORE\", KEYS[1], 0, ARGV[1]) or 0'"
      },
      "name": "Redis Remove Old Entries",
      "type": "n8n-nodes-base.redis",
      "typeVersion": 1,
      "position": [450, 250]
    },
    {
      "parameters": {
        "resource": "command",
        "command": "custom",
        "customCommand": "ZCARD rate_limit:{{ $node[\"Webhook Trigger\"].json[\"ip\"] }}"
      },
      "name": "Redis Count Current Requests",
      "type": "n8n-nodes-base.redis",
      "typeVersion": 1,
      "position": [450, 350]
    },
    {
      "parameters": {
        "conditions": {
          "number": [
            {
              "value1": "{{ $node[\"Redis Count Current Requests\"].json[\"response\"] }}",
              "operation": "lessThan",
              "value2": 100
            }
          ]
        }
      },
      "name": "Check Rate Limit",
      "type": "n8n-nodes-base.if",
      "typeVersion": 1,
      "position": [650, 300]
    },
    {
      "parameters": {
        "resource": "command",
        "command": "zadd",
        "key": "rate_limit:{{ $node[\"Webhook Trigger\"].json[\"ip\"] }}",
        "score": "{{ Date.now() / 1000 }}",
        "member": "{{ Date.now() / 1000 }}"
      },
      "name": "Record Request Timestamp",
      "type": "n8n-nodes-base.redis",
      "typeVersion": 1,
      "position": [850, 250]
    },
    {
      "parameters": {
        "resource": "command",
        "command": "expire",
        "key": "rate_limit:{{ $node[\"Webhook Trigger\"].json[\"ip\"] }}",
        "ttl": 61
      },
      "name": "Set Key Expiration",
      "type": "n8n-nodes-base.redis",
      "typeVersion": 1,
      "position": [850, 350]
    },
    {
      "parameters": {
        "responseCode": 200,
        "responseBody": "{\"allowed\": true, \"remaining\": {{ 100 - $node[\"Redis Count Current Requests\"].json[\"response\"] - 1 }} }"
      },
      "name": "Allow Request",
      "type": "n8n-nodes-base.respondToWebhook",
      "typeVersion": 1,
      "position": [1050, 250]
    },
    {
      "parameters": {
        "responseCode": 429,
        "responseBody": "{\"error\": \"Rate limit exceeded\", \"retry_after\": 60}"
      },
      "name": "Reject Request",
      "type": "n8n-nodes-base.respondToWebhook",
      "typeVersion": 1,
      "position": [1050, 350]
    }
  ],
  "connections": {
    "Webhook Trigger": {
      "main": [
        [
          {
            "node": "Redis Remove Old Entries",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Redis Remove Old Entries": {
      "main": [
        [
          {
            "node": "Redis Count Current Requests",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Redis Count Current Requests": {
      "main": [
        [
          {
            "node": "Check Rate Limit",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Check Rate Limit": {
      "main": [
        [
          {
            "node": "Record Request Timestamp",
            "type": "main",
            "index": 0
          }
        ],
        [
          {
            "node": "Reject Request",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Record Request Timestamp": {
      "main": [
        [
          {
            "node": "Set Key Expiration",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Set Key Expiration": {
      "main": [
        [
          {
            "node": "Allow Request",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  }
}

To deploy this on n8n Cloud, create a new workflow, add the Redis node, and configure your Redis connection credentials. For self-hosted setups on a Hetzner VPS, ensure Redis runs on the same internal network for minimal latency.

Testing and Monitoring

I always test rate limiting under load to validate behavior. Here’s a test script:

import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

BASE_URL = 'http://localhost:5000'
HEADERS = {'X-API-Key': 'test-key-123'}

def make_request(request_number: int) -> dict:
    """Make a single API request and return response details."""
    start_time = time.time()
    try:
        response = requests.get(
            f'{BASE_URL}/api/data',
            headers=HEADERS,
            timeout=5
        )
        elapsed = time.time() - start_time
        return {
            'request_num': request_number,
            'status': response.status_code,
            'remaining': response.headers.get('X-RateLimit-Remaining'),
            'reset': response.headers.get('X-RateLimit-Reset'),
            'elapsed': elapsed,
            'success': response.status_code == 200
        }
    except Exception as e:
        return {
            'request_num': request_number,
            'error': str(e),
            'success': False,
            'elapsed': time.time() - start_time
        }

def test_rate_limit(num_requests: int = 150, workers: int = 10):
    """Test rate limiting with concurrent requests."""
    print(f"Testing rate limiting with {num_requests} concurrent requests...")
    print(f"Using {workers} worker threads\n")
    
    results = []
    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = [executor.submit(make_request, i) for i in range(num_requests)]
        
        for future in as_completed(futures):
            result = future.result()
            results.append(result)
            if result['success']:
                print(f"Request {result['request_num']:3d}: ALLOWED   | "
                      f"Remaining: {result['remaining']:3s} | Elapsed: {result['elapsed']:.3f}s")
            else:
                print(f"Request {result['request_num']:3d}: REJECTED  | "
                      f"Status: {result['status']} | Elapsed: {result['elapsed']:.3f}s")
    
    allowed = sum(1 for r in results if r['success'])
    rejected = sum(1 for r in results if not r['success'])
    avg_time = sum(r.get('elapsed', 0) for r in results) / len(results)
    
    print(f"\n--- Test Summary ---")
    print(f"Total Requests: {num_requests}")
    print(f"Allowed: {allowed}")
    print(f"Rejected: {rejected}")
    print(f"Average Response Time: {avg_time:.3f}s")
    print(f"Success Rate: {(allowed/num_requests)*100:.1f}%")
    
    return results

if __name__ == '__main__':
    test_rate_limit(num_requests=150, workers=10)

For ongoing monitoring, I recommend setting up centralized server logging with Grafana Loki to track rate limit events, rejected requests by client, and historical patterns. This helps identify whether specific clients need higher limits or if your thresholds need adjustment.

When building large-scale scraping projects, rate limiting protects both your service and upstream APIs. Review production web scraping pipelines with Python for integration patterns that respect rate limits while maximizing throughput.

Getting Started

To implement this yourself:

  1. Deploy Redis on a Hetzner VPS, Contabo VPS, or DigitalOcean instance
  2. Install the redis-py library via pip
  3. Copy the Python class into your application
  4. Decorate your endpoints with the rate_limit decorator
  5. Monitor Redis memory usage and adjust retention policies
  6. For no-code implementation, set up a n8n Cloud workflow with Redis nodes

Start with a 100 requests per 60-second window and adjust based on your user behavior data. Test aggressively before production deployment using the load test script above.

Outsource Your Automation

Don’t have time? I build production n8n workflows, WhatsApp bots, and fully automated YouTube Shorts pipelines. Hire me on Fiverr, mention SYS3-HUGO for priority. Or DM at chasebot.online.

Want to automate this yourself?

Start with n8n Cloud (free tier available) or self-host on a Hetzner VPS for full control.

Want this engine running on your own VPS?

This blog publishes itself — daily, unattended, on free API tiers. The full engine, Hugo theme, and setup guide are available as System 3.

Get System 3

Related Guides

system online