Merge pull request #78 from systemli/feat/redis-rate-limiter

✨ Persist rate-limit state in Redis

Author: 0x46616c6b

.env.example

@@ -7,6 +7,7 @@ MAILBOX_LISTEN_ADDR=":10003"
 SENDERS_LISTEN_ADDR=":10004"
 METRICS_LISTEN_ADDR=":10005"
 RATE_LIMIT_MESSAGE="Rate limit exceeded, please try again later"
+REDIS_URL="redis://redis:6379/0"
 
 LOG_LEVEL=debug
 LOG_FORMAT=text

AGENTS.md

@@ -33,7 +33,7 @@ Only process at `protocol_state=END-OF-MESSAGE` for accurate counting.
 ## Pitfalls
 
 - `USERLI_TOKEN` is required — app exits immediately if missing
-- Rate limiter cleanup runs every 5 minutes in a background goroutine
+- `REDIS_URL` is required — rate-limit state lives in Redis (sliding-window sorted sets, key prefix `userli:ratelimit:sender:`, ~25h TTL); Redis errors fail open
 - Socketmap names must match exactly: `alias`, `domain`, `mailbox`, `senders`
 
 ## Docker Development

README.md

@@ -16,6 +16,7 @@ The adapter is configured via environment variables:
 - `POLICY_LISTEN_ADDR`: The address to listen on for policy requests (rate limiting). Default: `:10003`.
 - `METRICS_LISTEN_ADDR`: The address to listen on for metrics. Default: `:10002`.
 - `RATE_LIMIT_MESSAGE`: The rejection message returned when a sender exceeds their quota. Default: `Rate limit exceeded, please try again later`.
+- `REDIS_URL`: Connection URL for Redis (required). Format follows [`redis.ParseURL`](https://pkg.go.dev/github.com/redis/go-redis/v9#ParseURL), e.g. `redis://[user:password@]host:port/db`. Rate-limit state is stored in Redis so it survives restarts.
 
 In Postfix, you can configure the adapter using the socketmap protocol like this:
 
@@ -48,6 +49,8 @@ The Userli API endpoint `/api/postfix/smtp_quota/{email}` returns:
 
 Where `0` means unlimited. If the API is unreachable, messages are allowed (fail-open).
 
+Rate-limit state is persisted to Redis (`REDIS_URL`) so it survives restarts. If Redis is unreachable, messages are also allowed (fail-open) and the `userli_postfix_adapter_ratelimit_backend_errors_total` counter is incremented.
+
 ## Docker
 
 You can run the adapter using Docker.
@@ -172,7 +175,7 @@ The adapter exposes Prometheus metrics on `/metrics` (port 10002) and provides h
 - `userli_postfix_adapter_policy_request_duration_seconds` - Policy request duration histogram
 - `userli_postfix_adapter_quota_exceeded_total` - Total messages rejected due to quota
 - `userli_postfix_adapter_quota_checks_total` - Total quota checks performed
-- `userli_postfix_adapter_tracked_senders` - Number of senders tracked by rate limiter
+- `userli_postfix_adapter_ratelimit_backend_errors_total` - Total Redis errors hit by the rate limiter, labelled by `operation`
 
 All metrics include relevant labels (handler, status, endpoint, etc.).
 

config.go

@@ -27,6 +27,9 @@ type Config struct {
 
 	// RateLimitMessage is the message returned when rate limit is exceeded.
 	RateLimitMessage string
+
+	// RedisURL is the connection URL for Redis (used to persist rate-limit state).
+	RedisURL string
 }
 
 // NewConfig creates a new Config with default values.
@@ -63,6 +66,11 @@ func NewConfig() (*Config, error) {
 		rateLimitMessage = "Rate limit exceeded, please try again later"
 	}
 
+	redisURL := os.Getenv("REDIS_URL")
+	if redisURL == "" {
+		return nil, fmt.Errorf("REDIS_URL is required")
+	}
+
 	return &Config{
 		UserliBaseURL:             userliBaseURL,
 		UserliToken:               userliToken,
@@ -71,5 +79,6 @@ func NewConfig() (*Config, error) {
 		PolicyListenAddr:          policyListenAddr,
 		MetricsListenAddr:         metricsListenAddr,
 		RateLimitMessage:          rateLimitMessage,
+		RedisURL:                  redisURL,
 	}, nil
 }

config_test.go

@@ -20,6 +20,7 @@ func (s *ConfigTestSuite) SetupTest() {
 func (s *ConfigTestSuite) TestNewConfig() {
 	s.Run("fail when userli token not set", func() {
 		os.Unsetenv("USERLI_TOKEN")
+		os.Unsetenv("REDIS_URL")
 
 		config, err := NewConfig()
 
@@ -28,8 +29,20 @@ func (s *ConfigTestSuite) TestNewConfig() {
 		s.Contains(err.Error(), "USERLI_TOKEN is required")
 	})
 
+	s.Run("fail when redis url not set", func() {
+		os.Setenv("USERLI_TOKEN", "token")
+		os.Unsetenv("REDIS_URL")
+
+		config, err := NewConfig()
+
+		s.Nil(config)
+		s.Error(err)
+		s.Contains(err.Error(), "REDIS_URL is required")
+	})
+
 	s.Run("default config", func() {
 		os.Setenv("USERLI_TOKEN", "token")
+		os.Setenv("REDIS_URL", "redis://localhost:6379/0")
 
 		config, err := NewConfig()
 
@@ -40,6 +53,7 @@ func (s *ConfigTestSuite) TestNewConfig() {
 		s.Equal(":10001", config.SocketmapListenAddr)
 		s.Equal(":10002", config.MetricsListenAddr)
 		s.Equal("Rate limit exceeded, please try again later", config.RateLimitMessage)
+		s.Equal("redis://localhost:6379/0", config.RedisURL)
 	})
 
 	s.Run("custom config", func() {
@@ -49,6 +63,7 @@ func (s *ConfigTestSuite) TestNewConfig() {
 		os.Setenv("SOCKETMAP_LISTEN_ADDR", ":20001")
 		os.Setenv("METRICS_LISTEN_ADDR", ":20002")
 		os.Setenv("RATE_LIMIT_MESSAGE", "Too many emails")
+		os.Setenv("REDIS_URL", "redis://redis:6379/1")
 
 		config, err := NewConfig()
 
@@ -59,6 +74,7 @@ func (s *ConfigTestSuite) TestNewConfig() {
 		s.Equal(":20001", config.SocketmapListenAddr)
 		s.Equal(":20002", config.MetricsListenAddr)
 		s.Equal("Too many emails", config.RateLimitMessage)
+		s.Equal("redis://redis:6379/1", config.RedisURL)
 	})
 }
 

docker-compose.yml

@@ -10,6 +10,20 @@ services:
       - "10003:10003"
     env_file:
       - .env
+    depends_on:
+      redis:
+        condition: service_healthy
+    networks:
+      - userli
+
+  redis:
+    hostname: redis
+    image: redis:7-alpine
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 5s
+      timeout: 3s
+      retries: 5
     networks:
       - userli
 

go.mod

@@ -3,9 +3,11 @@ module github.com/systemli/userli-postfix-adapter
 go 1.23.1
 
 require (
+	github.com/alicebob/miniredis/v2 v2.37.0
 	github.com/h2non/gock v1.2.0
 	github.com/markdingo/netstring v1.0.2
 	github.com/prometheus/client_golang v1.23.2
+	github.com/redis/go-redis/v9 v9.18.0
 	github.com/stretchr/testify v1.11.1
 	go.uber.org/zap v1.27.1
 )
@@ -14,13 +16,16 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.66.1 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
+	github.com/yuin/gopher-lua v1.1.1 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.10.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	golang.org/x/sys v0.35.0 // indirect

go.sum

@@ -1,9 +1,17 @@
+github.com/alicebob/miniredis/v2 v2.37.0 h1:RheObYW32G1aiJIj81XVt78ZHJpHonHLHW7OLIshq68=
+github.com/alicebob/miniredis/v2 v2.37.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/h2non/gock v1.2.0 h1:K6ol8rfrRkUOefooBC8elXoaNGYkpp7y2qcxGG6BzUE=
@@ -12,6 +20,8 @@ github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542 h1:2VTzZjLZBgl62/EtslC
 github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542/go.mod h1:Ow0tF8D4Kplbc8s8sSb3V2oUCygFHVp8gC3Dn6U4MNI=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
+github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4=
+github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -34,12 +44,20 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z
 github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
 github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
+github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs=
+github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
 github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
+github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
+github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
+github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=

main.go

@@ -47,15 +47,19 @@ func main() {
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
 
-	rateLimiter := NewRateLimiter(ctx)
+	rateLimiter, err := NewRateLimiter(ctx, config.RedisURL, logger.Named("ratelimit"))
+	if err != nil {
+		logger.Fatal("Failed to initialize rate limiter", zap.Error(err))
+	}
+	defer func() { _ = rateLimiter.Close() }()
 	policyServer := NewPolicyServer(userli, rateLimiter, config.RateLimitMessage, logger.Named("policy"))
 
 	var wg sync.WaitGroup
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		StartMetricsServer(ctx, config.MetricsListenAddr, userli, rateLimiter)
+		StartMetricsServer(ctx, config.MetricsListenAddr, userli)
 	}()
 
 	wg.Add(1)

policy.go

@@ -219,7 +219,7 @@ func (p *PolicyServer) handleRequest(ctx context.Context, req *PolicyRequest) st
 	}
 
 	// Check rate limit
-	allowed, hourCount, dayCount := p.rateLimiter.CheckAndIncrement(sender, quota)
+	allowed, hourCount, dayCount := p.rateLimiter.CheckAndIncrement(quotaCtx, sender, quota)
 
 	// Update metrics
 	quotaChecksTotal.WithLabelValues("checked").Inc()