imp(): mv rate limiting to grouper

.github/workflows/lint.yml

@@ -5,10 +5,19 @@ on: [push, pull_request]
 jobs:
   lint:
     runs-on: ubuntu-latest
+
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.21'
+          cache: true
+
       - name: GolangCI Lint
-        uses: golangci/golangci-lint-action@v2
+        uses: golangci/golangci-lint-action@v6
         with:
-          version: latest
+          version: v1.63.4
+          args: --verbose

README.md

@@ -165,10 +165,15 @@ Basic configuration is taken from `.env` file.
 | NOTIFY_URL | https://notify.bot.ifmo.su/u/ABCD1234 | Address to send alerts in case of too many requests |
 | TOKEN_UPDATE_PERIOD | 10s | Time interval to update token cache |
 | PROJECTS_LIMITS_UPDATE_PERIOD | 3600 | Time interval to update projects limits cache (in seconds) |
+
 # Rate Limiting
 
 Rate limiting is implemented using Redis to track and enforce request limits per project. The system supports configurable limits at the project, workspace and plan level.
 
+Collector performs a **read-only** check against the shared Redis counters and rejects requests early when a project is already at or over its limit. Counter increments are owned by the grouper worker. When collector rejects a rate-limited request, it also records the `events-rate-limited` project metric in Redis TimeSeries.
+
+Collector also rejects requests from workspace-blocked projects via `DisabledProjectsSet`.
+
 ## Configuration
 
 Rate limits can be configured at multiple levels and applied in the following order (highest to lowest):
@@ -179,7 +184,7 @@ Rate limits can be configured at multiple levels and applied in the following or
 
 ## Implementation
 
-Rate limits are tracked in `rate_limit` Redis set with the following pattern:
+Rate limits are tracked in the `rate_limits` Redis hash with the following pattern:
 
 ```go
 // Key: "project_id" -> value: "timestamp:count"
@@ -218,7 +223,7 @@ Rate limits are fetched from MongoDB. You can find them in the `rateLimitSetting
 }
 ```
 
-Rate limits are automatically enforced for all incoming error and release events. No additional configuration is needed at the client level.
+Rate limits are automatically enforced for all incoming error events. No additional configuration is needed at the client level.
 
 When a rate limit is exceeded, clients will receive a response like:
 

pkg/redis/client.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
-	"github.com/go-redis/redis/v8"
+	redis "github.com/go-redis/redis/v8"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -224,14 +224,15 @@ func (r *RedisClient) CheckAvailability() bool {
 	return pong == "PONG"
 }
 
-// UpdateRateLimit checks and updates the rate limit for a project using a Lua script
-func (r *RedisClient) UpdateRateLimit(projectID string, eventsLimit int64, eventsPeriod int64) (bool, error) {
-	// If eventsLimit is 0, we don't need to update the rate limit
+// CheckRateLimit reads the per-project rate limit counter without incrementing.
+// Grouper owns increments; collector only rejects early when already at/over limit.
+// Expired or malformed windows are cleared to "{now}:0".
+func (r *RedisClient) CheckRateLimit(projectID string, eventsLimit int64, eventsPeriod int64) (bool, error) {
 	if eventsLimit == 0 {
 		return true, nil
 	}
 
-	// Lua script for atomic rate limit check and update
+	// Read-only check (no increment). Clear expired/malformed windows to now:0.
 	script := `
 		local key = KEYS[1]
 		local field = ARGV[1]
@@ -241,33 +242,29 @@ func (r *RedisClient) UpdateRateLimit(projectID string, eventsLimit int64, event
 
 		local current = redis.call('HGET', key, field)
 		if not current then
-			-- No existing record, create new window
-			redis.call('HSET', key, field, now .. ':1')
 			return 1
 		end
 
-		local timestamp, count = string.match(current, '(%d+):(%d+)')
+		local timestamp, count = string.match(current, '^(%d+):(%d+)$')
+		if not timestamp then
+			redis.call('HSET', key, field, now .. ':0')
+			return 1
+		end
 		timestamp = tonumber(timestamp)
 		count = tonumber(count)
 
-		-- Check if we're in a new time window
 		if now - timestamp >= period then
-			-- Reset for new window
-			redis.call('HSET', key, field, now .. ':1')
+			redis.call('HSET', key, field, now .. ':0')
 			return 1
 		end
 
-		-- Check if incrementing would exceed limit
-		if count + 1 > limit then
+		if count >= limit then
 			return 0
 		end
 
-		-- Increment counter
-		redis.call('HSET', key, field, timestamp .. ':' .. (count + 1))
 		return 1
 	`
 
-	// Run the script
 	result, err := r.rdb.Eval(
 		r.ctx,
 		script,
@@ -277,9 +274,8 @@ func (r *RedisClient) UpdateRateLimit(projectID string, eventsLimit int64, event
 		eventsLimit,             // limit (ARGV[3])
 		eventsPeriod,            // period (ARGV[4])
 	).Result()
-
 	if err != nil {
-		return false, fmt.Errorf("failed to execute rate limit script: %w", err)
+		return false, fmt.Errorf("failed to execute rate limit check script: %w", err)
 	}
 
 	// Script returns 1 if rate limit is not exceeded, 0 if it is

pkg/redis/client_test.go

@@ -7,20 +7,20 @@ import (
 	"time"
 
 	"github.com/alicebob/miniredis/v2"
-	"github.com/go-redis/redis/v8"
+	goredis "github.com/go-redis/redis/v8"
 	"github.com/stretchr/testify/assert"
 )
 
 func setupTestRedis(t *testing.T) (*RedisClient, *miniredis.Miniredis) {
-	// Create a mock Redis server
+	t.Helper()
+
 	mr, err := miniredis.Run()
 	if err != nil {
 		t.Fatalf("Failed to create mock redis: %v", err)
 	}
 
-	// Create Redis client connected to mock server
 	client := &RedisClient{
-		rdb: redis.NewClient(&redis.Options{
+		rdb: goredis.NewClient(&goredis.Options{
 			Addr: mr.Addr(),
 		}),
 		ctx: context.Background(),
@@ -29,7 +29,7 @@ func setupTestRedis(t *testing.T) (*RedisClient, *miniredis.Miniredis) {
 	return client, mr
 }
 
-func TestUpdateRateLimit(t *testing.T) {
+func TestCheckRateLimit(t *testing.T) {
 	client, mr := setupTestRedis(t)
 	defer mr.Close()
 
@@ -39,99 +39,99 @@ func TestUpdateRateLimit(t *testing.T) {
 		eventsLimit  int64
 		eventsPeriod int64
 		setup        func()
-		calls        int
 		wantAllowed  bool
-		wantErr      bool
+		wantValue    string
+		checkValue   bool
 	}{
 		{
-			name:         "should allow when no previous events",
+			name:         "allows when no previous events",
 			projectID:    "project1",
 			eventsLimit:  10,
 			eventsPeriod: 60,
-			calls:        1,
 			wantAllowed:  true,
-			wantErr:      false,
+			checkValue:   false,
 		},
 		{
-			name:         "should allow when under limit",
+			name:         "allows when under limit without incrementing",
 			projectID:    "project2",
 			eventsLimit:  10,
 			eventsPeriod: 60,
 			setup: func() {
 				client.rdb.HSet(client.ctx, "rate_limits", "project2",
 					fmt.Sprintf("%d:%d", time.Now().Unix()-30, 5))
 			},
-			calls:       1,
 			wantAllowed: true,
-			wantErr:     false,
+			wantValue:   fmt.Sprintf("%d:%d", time.Now().Unix()-30, 5),
+			checkValue:  true,
 		},
 		{
-			name:         "should deny when at limit",
+			name:         "denies when at limit without changing counter",
 			projectID:    "project3",
 			eventsLimit:  5,
 			eventsPeriod: 60,
 			setup: func() {
 				client.rdb.HSet(client.ctx, "rate_limits", "project3",
 					fmt.Sprintf("%d:%d", time.Now().Unix()-30, 5))
 			},
-			calls:       1,
 			wantAllowed: false,
-			wantErr:     false,
+			checkValue:  true,
 		},
 		{
-			name:         "should reset count after period expires",
+			name:         "clears expired window to zero and allows",
 			projectID:    "project4",
 			eventsLimit:  5,
 			eventsPeriod: 60,
 			setup: func() {
 				client.rdb.HSet(client.ctx, "rate_limits", "project4",
 					fmt.Sprintf("%d:%d", time.Now().Unix()-61, 5))
 			},
-			calls:       1,
 			wantAllowed: true,
-			wantErr:     false,
+			checkValue:  true,
 		},
 		{
-			name:         "should allow all when limit is 0",
+			name:         "allows all when limit is 0",
 			projectID:    "project5",
 			eventsLimit:  0,
 			eventsPeriod: 60,
-			calls:        5,
 			wantAllowed:  true,
-			wantErr:      false,
+			checkValue:   false,
 		},
 		{
-			name:         "should handle multiple calls up to limit",
+			name:         "clears malformed value to zero and allows",
 			projectID:    "project6",
-			eventsLimit:  3,
+			eventsLimit:  5,
 			eventsPeriod: 60,
-			calls:        4,
-			wantAllowed:  false, // Last call should be denied
-			wantErr:      false,
+			setup: func() {
+				client.rdb.HSet(client.ctx, "rate_limits", "project6", "broken")
+			},
+			wantAllowed: true,
+			checkValue:  true,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			// Run setup if provided
 			if tt.setup != nil {
 				tt.setup()
 			}
 
-			var lastAllowed bool
-			var lastErr error
+			before, _ := client.rdb.HGet(client.ctx, "rate_limits", tt.projectID).Result()
 
-			// Make the specified number of calls
-			for i := 0; i < tt.calls; i++ {
-				lastAllowed, lastErr = client.UpdateRateLimit(tt.projectID, tt.eventsLimit, tt.eventsPeriod)
+			allowed, err := client.CheckRateLimit(tt.projectID, tt.eventsLimit, tt.eventsPeriod)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.wantAllowed, allowed)
+
+			after, _ := client.rdb.HGet(client.ctx, "rate_limits", tt.projectID).Result()
+			if !tt.checkValue {
+				return
 			}
 
-			if tt.wantErr {
-				assert.Error(t, lastErr)
-			} else {
-				assert.NoError(t, lastErr)
+			switch tt.name {
+			case "allows when under limit without incrementing", "denies when at limit without changing counter":
+				assert.Equal(t, before, after, "counter must not be incremented")
+			case "clears expired window to zero and allows", "clears malformed value to zero and allows":
+				assert.Regexp(t, `^\d+:0$`, after)
 			}
-			assert.Equal(t, tt.wantAllowed, lastAllowed)
 		})
 	}
 }
@@ -159,53 +159,3 @@ func TestLoadBlockedIDsLargeSet(t *testing.T) {
 	}
 	assert.False(t, client.IsBlocked("not-in-set"))
 }
-
-func TestUpdateRateLimitConcurrent(t *testing.T) {
-	client, mr := setupTestRedis(t)
-	defer mr.Close()
-
-	const (
-		projectID       = "concurrent-project"
-		eventsLimit     = 90
-		eventsPeriod    = 60
-		goroutines      = 10
-		callsPerRoutine = 20
-	)
-
-	var rejectedCount int = 0
-
-	done := make(chan bool)
-
-	// Launch multiple goroutines to test concurrent access
-	for i := 0; i < goroutines; i++ {
-		go func() {
-			for j := 0; j < callsPerRoutine; j++ {
-				allowed, err := client.UpdateRateLimit(projectID, eventsLimit, eventsPeriod)
-				assert.NoError(t, err)
-				if !allowed {
-					rejectedCount++
-				}
-			}
-			done <- true
-		}()
-	}
-
-	// Wait for all goroutines to complete
-	for i := 0; i < goroutines; i++ {
-		<-done
-	}
-
-	// Verify the total number of successful updates doesn't exceed the limit
-	val, err := client.rdb.HGet(client.ctx, "rate_limits", projectID).Result()
-	assert.NoError(t, err)
-	assert.NotEmpty(t, val)
-
-	// The total count should not exceed the events limit
-	count := 0
-	_, err = fmt.Sscanf(val, "%d:%d", &count, &count)
-	assert.NoError(t, err)
-	assert.Equal(t, count, eventsLimit)
-	assert.Equal(t, rejectedCount, goroutines*callsPerRoutine-eventsLimit)
-	t.Logf("count: %d", count)
-	t.Logf("rejectedCount: %d", rejectedCount)
-}

pkg/server/errorshandler/handler.go

@@ -78,10 +78,10 @@ func (handler *Handler) process(body []byte) ResponseMessage {
 		return ResponseMessage{402, true, "Project has exceeded the events limit"}
 	}
 
-	rateWithinLimit, err := handler.RedisClient.UpdateRateLimit(projectId, projectLimits.EventsLimit, projectLimits.EventsPeriod)
+	rateWithinLimit, err := handler.RedisClient.CheckRateLimit(projectId, projectLimits.EventsLimit, projectLimits.EventsPeriod)
 	if err != nil {
-		log.Errorf("Failed to update rate limit: %s", err)
-		return ResponseMessage{402, true, "Failed to update rate limit"}
+		log.Errorf("Failed to check rate limit: %s", err)
+		return ResponseMessage{402, true, "Failed to check rate limit"}
 	}
 	if !rateWithinLimit {
 		handler.recordProjectMetrics(projectId, "events-rate-limited", false)
@@ -110,9 +110,6 @@ func (handler *Handler) process(body []byte) ResponseMessage {
 	// increment processed errors counter
 	handler.ErrorsProcessed.Inc()
 
-	// record project metrics
-	handler.recordProjectMetrics(projectId, "events-accepted", true)
-
 	return ResponseMessage{200, false, "OK"}
 }