Designing a Distributed Rate Limiter in Go Using Sliding Window Logs and Redis Cluster — code snippets

snippet-1.go

package ratelimit

import (
	"context"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"
)

type PipelineLimiter struct {
	client *redis.ClusterClient
	limit  int64
	window time.Duration
}

func NewPipelineLimiter(client *redis.ClusterClient, limit int64, window time.Duration) *PipelineLimiter {
	return &PipelineLimiter{
		client: client,
		limit:  limit,
		window: window,
	}
}

// Allow evaluates the rate limit for a given key.
// WARNING: This implementation suffers from race conditions and slot routing issues in a Redis Cluster.
func (p *PipelineLimiter) Allow(ctx context.Context, key string) (bool, error) {
	now := time.Now()
	nowMs := now.UnixNano() / int64(time.Millisecond)
	clearBefore := nowMs - p.window.Milliseconds()
	
	// Create a unique member to prevent collision of concurrent requests at the same millisecond
	member := fmt.Sprintf("%d:%s", nowMs, uuid.NewString()[:8])

	// In a Redis Cluster, this pipeline will throw a CROSSSLOT error unless the key
	// is explicitly routed to a single slot using hash tags, e.g., "{user_123}:ratelimit".
	pipe := p.client.TxPipeline()
	
	// 1. Evict expired entries
	pipe.ZRemRangeByScore(ctx, key, "-inf", fmt.Sprintf("%d", clearBefore))
	
	// 2. Count active entries
	cardCmd := pipe.ZCard(ctx, key)
	
	// 3. Add the current entry
	pipe.ZAdd(ctx, key, redis.Z{Score: float64(nowMs), Member: member})
	
	// 4. Update Key TTL
	pipe.Expire(ctx, key, p.window)
	
	// Execute the transaction pipeline
	_, err := pipe.Exec(ctx)
	if err != nil {
		return false, fmt.Errorf("failed to execute rate limit pipeline: %w", err)
	}

	// Naive check-then-act issue: The ZADD was already executed.
	// If the cardCmd returned a value greater than our limit, the client exceeded the limit,
	// but we have already added their request to the log.
	if cardCmd.Val() > p.limit {
		return false, nil
	}

	return true, nil
}

snippet-3.go

package ratelimit

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"
)

// LuaRateLimiter implements a sliding window rate limiter using Redis Lua scripts.
type LuaRateLimiter struct {
	client *redis.ClusterClient
	limit  int
	window time.Duration
	script *redis.Script
}

// NewLuaRateLimiter compiles and prepares the Lua rate limiter.
func NewLuaRateLimiter(client *redis.ClusterClient, limit int, window time.Duration) *LuaRateLimiter {
	const luaScript = `
		local key = KEYS[1]
		local now = tonumber(ARGV[1])
		local window = tonumber(ARGV[2])
		local limit = tonumber(ARGV[3])
		local member = ARGV[4]
		local clear_before = now - window

		redis.call('ZREMRANGEBYSCORE', key, '-inf', clear_before)
		local current_requests = redis.call('ZCARD', key)

		if current_requests < limit then
			redis.call('ZADD', key, now, member)
			redis.call('EXPIRE', key, math.ceil(window / 1000))
			return {1, limit - current_requests - 1}
		else
			return {0, 0}
		end
	`
	return &LuaRateLimiter{
		client: client,
		limit:  limit,
		window: window,
		script: redis.NewScript(luaScript),
	}
}

// Result holds the rate limiting verdict and metadata.
type Result struct {
	Allowed   bool
	Remaining int
}

// Allow checks if a request is allowed for the given key.
// The key must use hash tags to enforce single-slot routing in Redis Cluster (e.g., "{user_123}:ratelimit").
func (l *LuaRateLimiter) Allow(ctx context.Context, key string) (Result, error) {
	now := time.Now()
	nowMs := now.UnixNano() / int64(time.Millisecond)
	windowMs := l.window.Milliseconds()
	member := fmt.Sprintf("%d:%s", nowMs, uuid.NewString()[:8])

	// Executing script. KEYS[1] ensures slot routing is handled by go-redis ClusterClient.
	res, err := l.script.Run(ctx, l.client, []string{key}, nowMs, windowMs, l.limit, member).Result()
	if err != nil {
		return Result{}, fmt.Errorf("redis script execution failed: %w", err)
	}

	// Parse Lua multi-bulk response: array of two integers
	values, ok := res.([]interface{})
	if !ok || len(values) < 2 {
		return Result{}, errors.New("unexpected return type from Redis Lua script")
	}

	allowedVal, ok1 := values[0].(int64)
	remainingVal, ok2 := values[1].(int64)
	if !ok1 || !ok2 {
		return Result{}, errors.New("unexpected value types in Lua response slice")
	}

	return Result{
		Allowed:   allowedVal == 1,
		Remaining: int(remainingVal),
	}, nil
}

snippet-4.go

package ratelimit

import (
	"context"
	"log"
	"net"
	"net/http"
	"strconv"
	"time"

	"github.com/gin-gonic/gin"
)

type Limiter interface {
	Allow(ctx context.Context, key string) (Result, error)
}

// RateLimitMiddleware returns a Gin middleware that enforces a sliding window rate limit.
func RateLimitMiddleware(limiter Limiter, limit int, window time.Duration) gin.HandlerFunc {
	return func(c *gin.Context) {
		// 1. Identify the user. Prefer API key tenant ID, fallback to client IP.
		tenantID := c.GetString("tenant_id")
		if tenantID == "" {
			ip, _, err := net.SplitHostPort(c.Request.RemoteAddr)
			if err != nil {
				ip = c.ClientIP()
			}
			tenantID = ip
		}

		// 2. Construct slot-bracketed key to ensure all requests route to a single Redis Cluster node.
		// Curly braces define the hash tag: {tenantID} determines the cluster routing slot.
		key := "{" + tenantID + "}:ratelimit"

		// 3. Evaluate the rate limit
		result, err := limiter.Allow(c.Request.Context(), key)
		if err != nil {
			// Production Fail-Open Pattern:
			// Rate limiting is a non-functional protection layer. We do not block legitimate traffic
			// if our Redis Cluster degrades. Log the incident and continue request processing.
			log.Printf("[RateLimiter] Error evaluating rate limit for key %s: %v", key, err)
			c.Header("X-RateLimit-Degraded", "true")
			c.Next()
			return
		}

		// 4. Inject standard IETF rate limiting headers
		c.Header("X-RateLimit-Limit", strconv.Itoa(limit))
		c.Header("X-RateLimit-Remaining", strconv.Itoa(result.Remaining))
		c.Header("X-RateLimit-Reset", strconv.FormatInt(time.Now().Add(window).Unix(), 10))

		// 5. Reject request if limit was exceeded
		if !result.Allowed {
			c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{
				"error":       "Too Many Requests",
				"message":     "Rate limit exceeded. Please retry later.",
				"retry_after": int(window.Seconds()),
			})
			return
		}

		c.Next()
	}
}

snippet-5.go

package ratelimit

import (
	"context"
	"crypto/tls"
	"net"
	"time"

	"github.com/redis/go-redis/v9"
)

// NewProductionRedisCluster creates a robustly configured Redis Cluster client.
func NewProductionRedisCluster(addrs []string, password string) *redis.ClusterClient {
	return redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:    addrs,
		Password: password,

		// Strict Timeouts to prevent goroutine leaks under latency spikes
		DialTimeout:  1 * time.Second,   // Timeout to establish physical connection
		ReadTimeout:  30 * time.Millisecond,  // Strict read limit (Lua execution shouldn't exceed 10ms)
		WriteTimeout: 30 * time.Millisecond,  // Strict write limit

		// Connection Pool Tuning
		PoolSize:        100,             // Max connection count per master node
		MinIdleConns:    20,              // Keep warm connections ready
		MaxActiveConns:  200,             // Protect client OS file descriptors
		ConnMaxIdleTime: 5 * time.Minute, // Recycle inactive sockets
		ConnMaxLifetime: 1 * time.Hour,   // Refresh connections to rotate load balanced nodes

		// Retry Logic during topology updates (resharding/failover)
		MaxRedirects: 3,             // Max retries on MOVED/ASK responses
		MaxRetries:   2,             // Max socket failure retries
		MinRetryBackoff: 8 * time.Millisecond,
		MaxRetryBackoff: 50 * time.Millisecond,

		// KeepAlive probing for TCP socket health check
		Dialer: func(ctx context.Context, network, addr string) (net.Conn, error) {
			dialer := &net.Dialer{
				Timeout:   1 * time.Second,
				KeepAlive: 30 * time.Second,
			}
			return dialer.DialContext(ctx, network, addr)
		},

		// TLS settings (mandatory for cloud Redis systems like AWS ElastiCache)
		TLSConfig: &tls.Config{
			InsecureSkipVerify: false,
		},
	})
}

snippet-6.go

package ratelimit

import (
	"context"
	"sync"
	"time"

	"golang.org/x/time/rate"
)

type FallbackLuaLimiter struct {
	primary   *LuaRateLimiter
	fallback  sync.Map // In-memory rate limiters for local tracking
	localRate rate.Limit
	localBurst int
}

func NewFallbackLuaLimiter(primary *LuaRateLimiter, limit int, window time.Duration) *FallbackLuaLimiter {
	// Convert sliding window constraints to equivalent local token bucket limits
	// e.g. 100 requests per 60 seconds -> 1.66 tokens per second
	r := rate.Limit(float64(limit) / window.Seconds())
	return &FallbackLuaLimiter{
		primary:    primary,
		localRate:  r,
		localBurst: limit,
	}
}

// Allow executes rate limiting checking, falling back locally on Redis failures.
func (f *FallbackLuaLimiter) Allow(ctx context.Context, key string) (Result, error) {
	// Attempt primary evaluation via Redis Cluster
	result, err := f.primary.Allow(ctx, key)
	if err == nil {
		return result, nil
	}

	// Fallback Path: Redis execution failed (network timeout, node down)
	// Retrieve or initialize a local in-memory token bucket limiter for the key.
	limiterObj, _ := f.fallback.LoadOrStore(key, rate.NewLimiter(f.localRate, f.localBurst))
	limiter := limiterObj.(*rate.Limiter)

	// Evaluate request against local memory
	if !limiter.Allow() {
		return Result{
			Allowed:   false,
			Remaining: 0,
		}, err // Return the original Redis error along with rejection
	}

	// Request allowed locally under degraded circumstances
	return Result{
		Allowed:   true,
		Remaining: int(limiter.Tokens()),
	}, nil
}