feat: implement is_negative_hits in rate limit descriptor

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874
 	github.com/coocood/freecache v1.2.4
 	github.com/envoyproxy/go-control-plane v0.14.0
-	github.com/envoyproxy/go-control-plane/envoy v1.37.0
+	github.com/envoyproxy/go-control-plane/envoy v1.37.1-0.20260409083702-98966259b99a
 	github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20260629194254-5581fade6089
 	github.com/go-kit/log v0.2.1
 	github.com/golang/mock v1.6.0

go.sum

@@ -35,8 +35,8 @@ github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.m
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/go-control-plane v0.14.0 h1:hbG2kr4RuFj222B6+7T83thSPqLjwBIfQawTkC++2HA=
 github.com/envoyproxy/go-control-plane v0.14.0/go.mod h1:NcS5X47pLl/hfqxU70yPwL9ZMkUlwlKxtAohpi2wBEU=
-github.com/envoyproxy/go-control-plane/envoy v1.37.0 h1:u3riX6BoYRfF4Dr7dwSOroNfdSbEPe9Yyl09/B6wBrQ=
-github.com/envoyproxy/go-control-plane/envoy v1.37.0/go.mod h1:DReE9MMrmecPy+YvQOAOHNYMALuowAnbjjEMkkWOi6A=
+github.com/envoyproxy/go-control-plane/envoy v1.37.1-0.20260409083702-98966259b99a h1:qrP4J6AWJ9yd6CINhPMRL/MbFXNiV7qimRsCDTOV0a0=
+github.com/envoyproxy/go-control-plane/envoy v1.37.1-0.20260409083702-98966259b99a/go.mod h1:5yRfenlmRH8sxKrhXyiFtK8BDz3syDWcFm81rkCcATM=
 github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20260629194254-5581fade6089 h1:tft2Nib7aYTUNFyDuYxeERxPIQxGtI1FScjT5RYDMnI=
 github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20260629194254-5581fade6089/go.mod h1:YySqCcozu0HwklKZzeX6N98q+TyqEkgX2sg7DQqiJfU=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=

src/limiter/base_limiter.go

@@ -14,6 +14,18 @@ import (
 	"github.com/envoyproxy/ratelimit/src/utils"
 )
 
+// DecrementScript atomically decrements a rate limit counter, floored at 0.
+// If the key does not exist there is nothing to refund, so it returns 0 without
+// creating a phantom key.
+const DecrementScript = `
+local current = redis.call('GET', KEYS[1])                   -- get current count
+if current == false then return 0 end                        -- key absent: nothing to refund
+local new_val = math.floor(math.max(0, tonumber(current) - tonumber(ARGV[1]))) -- subtract hits, floor at 0
+redis.call('SET', KEYS[1], tostring(new_val))                -- persist new value
+redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))             -- reset TTL
+return new_val                                               -- return count after decrement
+`
+
 type BaseRateLimiter struct {
 	timeSource                 utils.TimeSource
 	JitterRand                 *rand.Rand
@@ -44,7 +56,7 @@ func NewRateLimitInfo(limit *config.RateLimit, limitBeforeIncrease uint64, limit
 // Generates cache keys for given rate limit request. Each cache key is represented by a concatenation of
 // domain, descriptor and current timestamp.
 func (this *BaseRateLimiter) GenerateCacheKeys(request *pb.RateLimitRequest,
-	limits []*config.RateLimit, hitsAddends []uint64,
+	limits []*config.RateLimit, hitsAddends []utils.HitsAddend,
 ) []CacheKey {
 	assert.Assert(len(request.Descriptors) == len(limits))
 	cacheKeys := make([]CacheKey, len(request.Descriptors))
@@ -55,7 +67,11 @@ func (this *BaseRateLimiter) GenerateCacheKeys(request *pb.RateLimitRequest,
 		cacheKeys[i] = this.cacheKeyGenerator.GenerateCacheKey(request.Domain, request.Descriptors[i], limits[i], now)
 		// Increase statistics for limits hit by their respective requests.
 		if limits[i] != nil {
-			limits[i].Stats.TotalHits.Add(hitsAddends[i])
+			if hitsAddends[i].IsNegative {
+				limits[i].Stats.TotalNegativeHits.Add(hitsAddends[i].Value)
+			} else {
+				limits[i].Stats.TotalHits.Add(hitsAddends[i].Value)
+			}
 		}
 	}
 	return cacheKeys
@@ -142,6 +158,28 @@ func (this *BaseRateLimiter) GetResponseDescriptorStatus(key string, limitInfo *
 	return responseDescriptorStatus
 }
 
+// GetResponseDescriptorStatusForNegativeHits generates a response for a negative-hit
+// (decrement/refund) request. Refunds release capacity rather than consuming it, so they
+// always return OK regardless of the counter value and never trigger over-limit side
+// effects (over-limit stats, local-cache poisoning). currentValue is the counter value
+// after the decrement; LimitRemaining is reported as the remaining capacity, clamped at 0
+// in case the counter is still above the limit.
+func (this *BaseRateLimiter) GetResponseDescriptorStatusForNegativeHits(key string, limit *config.RateLimit,
+	currentValue uint64,
+) *pb.RateLimitResponse_DescriptorStatus {
+	if key == "" {
+		return this.generateResponseDescriptorStatus(pb.RateLimitResponse_OK, nil, 0)
+	}
+
+	overLimitThreshold := uint64(limit.Limit.RequestsPerUnit)
+	var limitRemaining uint64
+	if overLimitThreshold > currentValue {
+		limitRemaining = overLimitThreshold - currentValue
+	}
+
+	return this.generateResponseDescriptorStatus(pb.RateLimitResponse_OK, limit.Limit, uint32(limitRemaining))
+}
+
 func NewBaseRateLimit(timeSource utils.TimeSource, jitterRand *rand.Rand, expirationJitterMaxSeconds int64,
 	localCache *freecache.Cache, nearLimitRatio float32, cacheKeyPrefix string, statsManager stats.Manager,
 ) *BaseRateLimiter {

src/memcached/cache_impl.go

@@ -85,6 +85,12 @@ func (this *rateLimitMemcacheImpl) DoLimit(
 			continue
 		}
 
+		// Negative hits skip the over-limit check — they always proceed.
+		if hitsAddends[i].IsNegative {
+			keysToGet = append(keysToGet, cacheKey.Key)
+			continue
+		}
+
 		// Check if key is over the limit in local cache.
 		if this.baseRateLimiter.IsOverLimitWithLocalCache(cacheKey.Key) {
 			isOverLimitWithLocalCache[i] = true
@@ -130,15 +136,26 @@ func (this *rateLimitMemcacheImpl) DoLimit(
 			} else {
 				limitBeforeIncrease = uint64(decoded)
 			}
-
 		}
 
-		limitAfterIncrease := limitBeforeIncrease + hitsAddends[i]
+		if hitsAddends[i].IsNegative {
+			// Predict the post-decrement value (guard against uint64 underflow).
+			var limitAfterDecrease uint64
+			if limitBeforeIncrease > hitsAddends[i].Value {
+				limitAfterDecrease = limitBeforeIncrease - hitsAddends[i].Value
+			}
+			// Negative hits (refunds) always return OK with the remaining capacity,
+			// even if the post-decrement counter is still above the limit.
+			responseDescriptorStatuses[i] = this.baseRateLimiter.GetResponseDescriptorStatusForNegativeHits(
+				cacheKey.Key, limits[i], limitAfterDecrease)
+		} else {
+			limitAfterIncrease := limitBeforeIncrease + hitsAddends[i].Value
 
-		limitInfo := limiter.NewRateLimitInfo(limits[i], limitBeforeIncrease, limitAfterIncrease, 0, 0)
+			limitInfo := limiter.NewRateLimitInfo(limits[i], limitBeforeIncrease, limitAfterIncrease, 0, 0)
 
-		responseDescriptorStatuses[i] = this.baseRateLimiter.GetResponseDescriptorStatus(cacheKey.Key,
-			limitInfo, isOverLimitWithLocalCache[i], hitsAddends[i])
+			responseDescriptorStatuses[i] = this.baseRateLimiter.GetResponseDescriptorStatus(cacheKey.Key,
+				limitInfo, isOverLimitWithLocalCache[i], hitsAddends[i].Value)
+		}
 	}
 
 	this.waitGroup.Add(1)
@@ -151,15 +168,24 @@ func (this *rateLimitMemcacheImpl) DoLimit(
 }
 
 func (this *rateLimitMemcacheImpl) increaseAsync(cacheKeys []limiter.CacheKey, isOverLimitWithLocalCache []bool,
-	limits []*config.RateLimit, hitsAddends []uint64,
+	limits []*config.RateLimit, hitsAddends []utils.HitsAddend,
 ) {
 	defer this.waitGroup.Done()
 	for i, cacheKey := range cacheKeys {
 		if cacheKey.Key == "" || isOverLimitWithLocalCache[i] {
 			continue
 		}
 
-		_, err := this.client.Increment(cacheKey.Key, hitsAddends[i])
+		if hitsAddends[i].IsNegative {
+			// Memcached Decrement natively floors at 0.
+			_, err := this.client.Decrement(cacheKey.Key, hitsAddends[i].Value)
+			if err != nil && err != memcache.ErrCacheMiss {
+				logger.Errorf("Failed to decrement key %s: %s", cacheKey.Key, err)
+			}
+			continue
+		}
+
+		_, err := this.client.Increment(cacheKey.Key, hitsAddends[i].Value)
 		if err == memcache.ErrCacheMiss {
 			expirationSeconds := utils.UnitToDivider(limits[i].Limit.Unit)
 			if this.expirationJitterMaxSeconds > 0 {
@@ -169,13 +195,13 @@ func (this *rateLimitMemcacheImpl) increaseAsync(cacheKeys []limiter.CacheKey, i
 			// Need to add instead of increment.
 			err = this.client.Add(&memcache.Item{
 				Key:        cacheKey.Key,
-				Value:      []byte(strconv.FormatUint(hitsAddends[i], 10)),
+				Value:      []byte(strconv.FormatUint(hitsAddends[i].Value, 10)),
 				Expiration: int32(expirationSeconds),
 			})
 			if err == memcache.ErrNotStored {
 				// There was a race condition to do this add. We should be able to increment
 				// now instead.
-				_, err := this.client.Increment(cacheKey.Key, hitsAddends[i])
+				_, err := this.client.Increment(cacheKey.Key, hitsAddends[i].Value)
 				if err != nil {
 					logger.Errorf("Failed to increment key %s after failing to add: %s", cacheKey.Key, err)
 					continue

src/memcached/client.go

@@ -17,5 +17,6 @@ var _ Client = (*memcache.Client)(nil)
 type Client interface {
 	GetMulti(keys []string) (map[string]*memcache.Item, error)
 	Increment(key string, delta uint64) (newValue uint64, err error)
+	Decrement(key string, delta uint64) (newValue uint64, err error)
 	Add(item *memcache.Item) error
 }

src/memcached/stats_collecting_client.go

@@ -13,6 +13,9 @@ type statsCollectingClient struct {
 	incrementSuccess stats.Counter
 	incrementMiss    stats.Counter
 	incrementError   stats.Counter
+	decrementSuccess stats.Counter
+	decrementMiss    stats.Counter
+	decrementError   stats.Counter
 	addSuccess       stats.Counter
 	addError         stats.Counter
 	addNotStored     stats.Counter
@@ -28,6 +31,9 @@ func CollectStats(c Client, scope stats.Scope) Client {
 		incrementSuccess: scope.NewCounterWithTags("increment", map[string]string{"code": "success"}),
 		incrementMiss:    scope.NewCounterWithTags("increment", map[string]string{"code": "miss"}),
 		incrementError:   scope.NewCounterWithTags("increment", map[string]string{"code": "error"}),
+		decrementSuccess: scope.NewCounterWithTags("decrement", map[string]string{"code": "success"}),
+		decrementMiss:    scope.NewCounterWithTags("decrement", map[string]string{"code": "miss"}),
+		decrementError:   scope.NewCounterWithTags("decrement", map[string]string{"code": "error"}),
 		addSuccess:       scope.NewCounterWithTags("add", map[string]string{"code": "success"}),
 		addError:         scope.NewCounterWithTags("add", map[string]string{"code": "error"}),
 		addNotStored:     scope.NewCounterWithTags("add", map[string]string{"code": "not_stored"}),
@@ -64,6 +70,19 @@ func (scc statsCollectingClient) Increment(key string, delta uint64) (newValue u
 	return
 }
 
+func (scc statsCollectingClient) Decrement(key string, delta uint64) (newValue uint64, err error) {
+	newValue, err = scc.c.Decrement(key, delta)
+	switch err {
+	case memcache.ErrCacheMiss:
+		scc.decrementMiss.Inc()
+	case nil:
+		scc.decrementSuccess.Inc()
+	default:
+		scc.decrementError.Inc()
+	}
+	return
+}
+
 func (scc statsCollectingClient) Add(item *memcache.Item) error {
 	err := scc.c.Add(item)
 

src/redis/driver.go

@@ -32,6 +32,21 @@ type Client interface {
 	// @param args supplies the additional arguments.
 	PipeAppend(pipeline Pipeline, rcv interface{}, cmd, key string, args ...interface{}) Pipeline
 
+	// PipeAppendWithRoutingKey appends a command onto the pipeline queue whose
+	// Redis key differs from the command's first positional argument. This is
+	// needed for commands like EVAL, where the first argument is the script body
+	// and the actual key appears later in the argument list. In Redis Cluster
+	// mode the routing key determines which slot/node the command is sent to, so
+	// it must be the real key rather than the script text.
+	//
+	// @param pipeline supplies the queue for pending commands.
+	// @param routingKey supplies the key used for cluster slot routing/grouping.
+	// @param rcv supplies receiver for the result.
+	// @param cmd supplies the command to append.
+	// @param key supplies the first positional argument of the command.
+	// @param args supplies the additional arguments.
+	PipeAppendWithRoutingKey(pipeline Pipeline, routingKey string, rcv interface{}, cmd, key string, args ...interface{}) Pipeline
+
 	// PipeDo writes multiple commands to a Conn in
 	// a single write, then reads their responses in a single read. This reduces
 	// network delay into a single round-trip.

src/redis/driver_impl.go

@@ -353,13 +353,17 @@ func (c *clientImpl) NumActiveConns() int {
 }
 
 func (c *clientImpl) PipeAppend(pipeline Pipeline, rcv interface{}, cmd, key string, args ...interface{}) Pipeline {
+	return c.PipeAppendWithRoutingKey(pipeline, key, rcv, cmd, key, args...)
+}
+
+func (c *clientImpl) PipeAppendWithRoutingKey(pipeline Pipeline, routingKey string, rcv interface{}, cmd, key string, args ...interface{}) Pipeline {
 	// Combine key and args into a single slice
 	allArgs := make([]interface{}, 0, 1+len(args))
 	allArgs = append(allArgs, key)
 	allArgs = append(allArgs, args...)
 	return append(pipeline, PipelineAction{
 		Action: radix.FlatCmd(rcv, cmd, allArgs...),
-		Key:    key,
+		Key:    routingKey,
 	})
 }
 

src/redis/fixed_cache_impl.go

@@ -37,11 +37,32 @@ func pipelineAppend(client Client, pipeline *Pipeline, key string, hitsAddend ui
 	*pipeline = client.PipeAppend(*pipeline, nil, "EXPIRE", key, expirationSeconds)
 }
 
+func pipelineAppendDecrement(client Client, pipeline *Pipeline, key string, hitsAddend uint64, result *uint64, expirationSeconds int64) {
+	// EVAL's first positional argument is the script body, not the key, so the
+	// real cache key must be passed explicitly as the routing key. Otherwise, in
+	// Redis Cluster mode the command would be routed using the script text,
+	// causing MOVED/CROSSSLOT errors or misrouting.
+	*pipeline = client.PipeAppendWithRoutingKey(*pipeline, key, result, "EVAL", limiter.DecrementScript, 1, key, hitsAddend, expirationSeconds)
+}
+
+func (this *fixedRateLimitCacheImpl) selectPipeline(cacheKey limiter.CacheKey, pipeline *Pipeline, perSecondPipeline *Pipeline) (Client, *Pipeline) {
+	if this.perSecondClient != nil && cacheKey.PerSecond {
+		if *perSecondPipeline == nil {
+			*perSecondPipeline = Pipeline{}
+		}
+		return this.perSecondClient, perSecondPipeline
+	}
+	if *pipeline == nil {
+		*pipeline = Pipeline{}
+	}
+	return this.client, pipeline
+}
+
 func pipelineAppendtoGet(client Client, pipeline *Pipeline, key string, result *uint64) {
 	*pipeline = client.PipeAppend(*pipeline, result, "GET", key)
 }
 
-func (this *fixedRateLimitCacheImpl) getHitsAddend(hitsAddend uint64, isCacheKeyOverlimit, isCacheKeyNearlimit,
+func (this *fixedRateLimitCacheImpl) getHitsAddendValue(hitsAddend uint64, isCacheKeyOverlimit, isCacheKeyNearlimit,
 	isNearLimit bool,
 ) uint64 {
 	// If stopCacheKeyIncrementWhenOverlimit is false, then we always increment the cache key.
@@ -94,8 +115,9 @@ func (this *fixedRateLimitCacheImpl) DoLimit(
 	isCacheKeyNearlimit := false
 
 	// Check if any of the keys are already to the over limit in cache.
+	// Negative hits (decrements) skip this check — they always proceed.
 	for i, cacheKey := range cacheKeys {
-		if cacheKey.Key == "" {
+		if cacheKey.Key == "" || hitsAddends[i].IsNegative {
 			continue
 		}
 
@@ -116,7 +138,7 @@ func (this *fixedRateLimitCacheImpl) DoLimit(
 	// then we check if any of the keys are near limit in redis cache.
 	if this.stopCacheKeyIncrementWhenOverlimit && !isCacheKeyOverlimit {
 		for i, cacheKey := range cacheKeys {
-			if cacheKey.Key == "" {
+			if cacheKey.Key == "" || hitsAddends[i].IsNegative {
 				continue
 			}
 
@@ -141,12 +163,12 @@ func (this *fixedRateLimitCacheImpl) DoLimit(
 		}
 
 		for i, cacheKey := range cacheKeys {
-			if cacheKey.Key == "" {
+			if cacheKey.Key == "" || hitsAddends[i].IsNegative {
 				continue
 			}
 			// Now fetch the pipeline.
 			limitBeforeIncrease := currentCount[i]
-			limitAfterIncrease := limitBeforeIncrease + hitsAddends[i]
+			limitAfterIncrease := limitBeforeIncrease + hitsAddends[i].Value
 
 			limitInfo := limiter.NewRateLimitInfo(limits[i], limitBeforeIncrease, limitAfterIncrease, 0, 0)
 
@@ -157,7 +179,7 @@ func (this *fixedRateLimitCacheImpl) DoLimit(
 		}
 	}
 
-	// Now, actually setup the pipeline to increase the usage of cache key, skipping empty cache keys.
+	// Now, actually setup the pipeline to increase/decrease the usage of cache key, skipping empty cache keys.
 	for i, cacheKey := range cacheKeys {
 		if cacheKey.Key == "" || overlimitIndexes[i] {
 			continue
@@ -170,19 +192,12 @@ func (this *fixedRateLimitCacheImpl) DoLimit(
 			expirationSeconds += this.baseRateLimiter.JitterRand.Int63n(this.baseRateLimiter.ExpirationJitterMaxSeconds)
 		}
 
-		// Use the perSecondConn if it is not nil and the cacheKey represents a per second Limit.
-		if this.perSecondClient != nil && cacheKey.PerSecond {
-			if perSecondPipeline == nil {
-				perSecondPipeline = Pipeline{}
-			}
-			pipelineAppend(this.perSecondClient, &perSecondPipeline, cacheKey.Key, this.getHitsAddend(hitsAddends[i],
-				isCacheKeyOverlimit, isCacheKeyNearlimit, nearlimitIndexes[i]), &results[i], expirationSeconds)
+		client, p := this.selectPipeline(cacheKey, &pipeline, &perSecondPipeline)
+		if hitsAddends[i].IsNegative {
+			pipelineAppendDecrement(client, p, cacheKey.Key, hitsAddends[i].Value, &results[i], expirationSeconds)
 		} else {
-			if pipeline == nil {
-				pipeline = Pipeline{}
-			}
-			pipelineAppend(this.client, &pipeline, cacheKey.Key, this.getHitsAddend(hitsAddends[i], isCacheKeyOverlimit,
-				isCacheKeyNearlimit, nearlimitIndexes[i]), &results[i], expirationSeconds)
+			pipelineAppend(client, p, cacheKey.Key, this.getHitsAddendValue(hitsAddends[i].Value,
+				isCacheKeyOverlimit, isCacheKeyNearlimit, nearlimitIndexes[i]), &results[i], expirationSeconds)
 		}
 	}
 
@@ -207,14 +222,20 @@ func (this *fixedRateLimitCacheImpl) DoLimit(
 	responseDescriptorStatuses := make([]*pb.RateLimitResponse_DescriptorStatus,
 		len(request.Descriptors))
 	for i, cacheKey := range cacheKeys {
+		if hitsAddends[i].IsNegative {
+			// Negative hits (refunds) always return OK with the remaining capacity,
+			// even if the post-decrement counter is still above the limit.
+			responseDescriptorStatuses[i] = this.baseRateLimiter.GetResponseDescriptorStatusForNegativeHits(
+				cacheKey.Key, limits[i], results[i])
+		} else {
+			limitAfterIncrease := results[i]
+			limitBeforeIncrease := limitAfterIncrease - hitsAddends[i].Value
 
-		limitAfterIncrease := results[i]
-		limitBeforeIncrease := limitAfterIncrease - hitsAddends[i]
-
-		limitInfo := limiter.NewRateLimitInfo(limits[i], limitBeforeIncrease, limitAfterIncrease, 0, 0)
+			limitInfo := limiter.NewRateLimitInfo(limits[i], limitBeforeIncrease, limitAfterIncrease, 0, 0)
 
-		responseDescriptorStatuses[i] = this.baseRateLimiter.GetResponseDescriptorStatus(cacheKey.Key,
-			limitInfo, isOverLimitWithLocalCache[i], hitsAddends[i])
+			responseDescriptorStatuses[i] = this.baseRateLimiter.GetResponseDescriptorStatus(cacheKey.Key,
+				limitInfo, isOverLimitWithLocalCache[i], hitsAddends[i].Value)
+		}
 	}
 
 	return responseDescriptorStatuses