fix(limit): atomic redis commits and resolved-var validation (#13467)

shreemaan-abhishek · web-flow · commit 6e877eb44f68 · 2026-06-08T16:37:38.000+05:45
diff --git a/apisix/plugins/limit-conn/init.lua b/apisix/plugins/limit-conn/init.lua
@@ -22,6 +22,7 @@ local tonumber = tonumber
 local type = type
 local tostring = tostring
 local ipairs = ipairs
+local floor = math.floor
 local shdict_name = "plugin-limit-conn"
 if ngx.config.subsystem == "stream" then
     shdict_name = shdict_name .. "-stream"
@@ -41,16 +42,38 @@ end
 local _M = {}
 
 
-local function resolve_var(ctx, value)
+local function resolve_var(ctx, value, allow_zero)
     if type(value) == "string" then
+        local original = value
         local err, _
         value, err, _ = core.utils.resolve_var(value, ctx.var)
         if err then
-            return nil, "could not resolve var for value: " .. value .. ", err: " .. err
+            return nil, "could not resolve var for value: " .. original .. ", err: " .. err
         end
+        local resolved = value
         value = tonumber(value)
         if not value then
-            return nil, "resolved value is not a number: " .. tostring(value)
+            return nil, "resolved value is not a number: " .. tostring(resolved)
+        end
+        -- conn must be positive; burst may be zero, mirroring the schema where
+        -- conn has exclusiveMinimum 0 and burst has minimum 0
+        if allow_zero then
+            if value < 0 then
+                return nil, "resolved value must be a non-negative number, got: "
+                            .. tostring(value)
+            end
+        elseif value <= 0 then
+            return nil, "resolved value must be a positive number, got: " .. tostring(value)
+        end
+        -- conn/burst are used as integer operands in floor((conn - 1) / max);
+        -- fractional values produce wrong delay calculations
+        if value ~= floor(value) then
+            return nil, "resolved value must be an integer, got: " .. tostring(value)
+        end
+        -- LuaJIT doubles lose integer precision above 2^53 (9007199254740992)
+        if value > 9007199254740991 then
+            return nil, "resolved value exceeds safe integer range (2^53-1), got: "
+                        .. tostring(value)
         end
     end
     return value
@@ -63,7 +86,7 @@ local function get_rules(ctx, conf)
         if err then
             return nil, err
         end
-        local burst, err2 = resolve_var(ctx, conf.burst)
+        local burst, err2 = resolve_var(ctx, conf.burst, true)
         if err2 then
             return nil, err2
         end
@@ -83,7 +106,7 @@ local function get_rules(ctx, conf)
         if err then
             goto CONTINUE
         end
-        local burst, err2 = resolve_var(ctx, rule.burst)
+        local burst, err2 = resolve_var(ctx, rule.burst, true)
         if err2 then
             goto CONTINUE
         end
diff --git a/apisix/plugins/limit-req/util.lua b/apisix/plugins/limit-req/util.lua
@@ -20,55 +20,113 @@ local max               = math.max
 local ngx_now           = ngx.now
 local ngx_null          = ngx.null
 local tonumber          = tonumber
+local core              = require("apisix.core")
 
 
 local _M = {version = 0.1}
 
 
--- the "commit" argument controls whether should we record the event in shm.
-function _M.incoming(self, red, key, commit)
-    local rate = self.rate
-    local now = ngx_now() * 1000
-
-    key = "limit_req" .. ":" .. key
-    local excess_key = key .. "excess"
-    local last_key = key .. "last"
+-- Redis Lua script that reads, decays and persists the leaky-bucket state in a
+-- single atomic step. With separate GET/GET/SET/SET commands, concurrent
+-- requests can each read the same stale excess value, all conclude they are
+-- within limits, and all get admitted, exceeding the configured rate.
+--
+-- The whole state lives in a single Redis hash (fields "excess" and "last")
+-- so the script only needs one key. lua-resty-rediscluster refuses to run
+-- EVAL with more than one key, and a single key is trivially slot-safe in
+-- Redis Cluster.
+--
+-- KEYS[1] = state key (hash with fields "excess" and "last")
+-- ARGV[1] = rate (req/s * 1000), ARGV[2] = burst (* 1000), ARGV[3] = now (ms),
+-- ARGV[4] = ttl (seconds)
+-- Returns {1, excess} on allow  (state already stored)
+--         {0, excess} on reject (nothing stored)
+-- excess is returned as a string to keep its fractional part: Redis truncates
+-- Lua numbers to integers when converting them to a reply.
+local redis_commit_script = core.string.compress_script([=[
+    local state_key  = KEYS[1]
+    local rate       = tonumber(ARGV[1])
+    local burst      = tonumber(ARGV[2])
+    local now        = tonumber(ARGV[3])
+    local ttl        = tonumber(ARGV[4])
 
-    local excess, err = red:get(excess_key)
-    if err then
-        return nil, err
-    end
-    local last, err = red:get(last_key)
-    if err then
-        return nil, err
-    end
+    local state = redis.call('hmget', state_key, 'excess', 'last')
+    local excess_raw = state[1]
+    local last_raw   = state[2]
 
-    if excess ~= ngx_null and last ~= ngx_null then
-        excess = tonumber(excess)
-        last = tonumber(last)
-        local elapsed = now - last
-        excess = max(excess - rate * abs(elapsed) / 1000 + 1000, 0)
+    local excess
+    if excess_raw and last_raw then
+        -- state exists: apply leaky-bucket decay then add one request-unit (1000)
+        local elapsed = now - tonumber(last_raw)
+        excess = math.max(tonumber(excess_raw) - rate * math.abs(elapsed) / 1000 + 1000, 0)
 
-        if excess > self.burst then
-            return nil, "rejected"
+        if excess > burst then
+            return {0, tostring(excess)}
         end
     else
+        -- no prior state: mirror the original behaviour, which skips the
+        -- leaky-bucket formula and starts with excess = 0 so the very first
+        -- request is always admitted
         excess = 0
     end
 
-    if commit then
-        local ttl = math.ceil(self.burst / self.rate) + 1
-        local ok, err
+    redis.call('hset', state_key, 'excess', excess, 'last', now)
+    redis.call('expire', state_key, ttl)
+    return {1, tostring(excess)}
+]=])
 
-        ok, err = red:set(excess_key, excess, "EX", ttl)
-        if not ok then
+
+-- the "commit" argument controls whether should we record the event in shm.
+function _M.incoming(self, red, key, commit)
+    local rate = self.rate
+    local now = ngx_now() * 1000
+
+    -- all leaky-bucket state lives in a single Redis hash so that the commit
+    -- script can run with one key on both Redis and Redis Cluster
+    local state_key = "limit_req" .. ":" .. key
+
+    if not commit then
+        -- read-only path: a plain HMGET is fine here because nothing is
+        -- written back, so a stale read only affects this advisory check
+        local state, err = red:hmget(state_key, "excess", "last")
+        if not state then
             return nil, err
         end
 
-        ok, err = red:set(last_key, now, "EX", ttl)
-        if not ok then
-            return nil, err
+        local excess = state[1]
+        local last = state[2]
+        local excess_val
+        if excess ~= ngx_null and last ~= ngx_null then
+            excess_val = tonumber(excess)
+            local last_val = tonumber(last)
+            local elapsed = now - last_val
+            excess_val = max(excess_val - rate * abs(elapsed) / 1000 + 1000, 0)
+
+            if excess_val > self.burst then
+                return nil, "rejected"
+            end
+        else
+            excess_val = 0
         end
+
+        return excess_val / rate, excess_val / 1000
+    end
+
+    -- commit path: run the read-compute-write cycle atomically in Redis so
+    -- that concurrent requests cannot be admitted based on the same stale
+    -- state
+    local ttl = math.ceil(self.burst / self.rate) + 1
+
+    local res, err = red:eval(redis_commit_script, 1, state_key,
+                              rate, self.burst, now, ttl)
+    if not res then
+        return nil, err
+    end
+
+    local allowed = res[1]
+    local excess = tonumber(res[2])
+    if allowed == 0 then
+        return nil, "rejected"
     end
 
     -- return the delay in seconds, as well as excess
diff --git a/t/plugin/limit-conn-variable.t b/t/plugin/limit-conn-variable.t
@@ -351,3 +351,120 @@ GET /test_concurrency
 500
 --- error_log
 failed to get limit conn rules
+
+
+
+=== TEST 11: restore route for invalid-variable tests (conn/burst as dynamic header)
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local code, body = t('/apisix/admin/routes/1',
+                 ngx.HTTP_PUT,
+                 [[{
+                        "plugins": {
+                            "limit-conn": {
+                                "conn": "${http_conn ?? 5}",
+                                "burst": "${http_burst ?? 2}",
+                                "default_conn_delay": 0.1,
+                                "rejected_code": 503,
+                                "key": "remote_addr"
+                            }
+                        },
+                        "upstream": {
+                            "nodes": {
+                                "127.0.0.1:1980": 1
+                            },
+                            "type": "roundrobin"
+                        },
+                        "uri": "/limit_conn"
+                }]]
+                )
+
+            if code >= 300 then
+                ngx.status = code
+            end
+            ngx.say(body)
+        }
+    }
+--- request
+GET /t
+--- response_body
+passed
+
+
+
+=== TEST 12: zero conn header value is rejected with 500 and error log
+--- request
+GET /limit_conn
+--- more_headers
+conn: 0
+--- error_code: 500
+--- error_log
+resolved value must be a positive number
+
+
+
+=== TEST 13: negative conn header value is rejected with 500 and error log
+--- request
+GET /limit_conn
+--- more_headers
+conn: -1
+--- error_code: 500
+--- error_log
+resolved value must be a positive number
+
+
+
+=== TEST 14: fractional conn header value is rejected with 500 and error log
+--- request
+GET /limit_conn
+--- more_headers
+conn: 1.5
+--- error_code: 500
+--- error_log
+resolved value must be an integer
+
+
+
+=== TEST 15: fractional burst header value is rejected with 500 and error log
+--- request
+GET /limit_conn
+--- more_headers
+burst: 1.5
+--- error_code: 500
+--- error_log
+resolved value must be an integer
+
+
+
+=== TEST 16: conn header value above 2^53-1 is rejected with 500 and error log
+--- request
+GET /limit_conn
+--- more_headers
+conn: 99007199254740993
+--- error_code: 500
+--- error_log
+resolved value exceeds safe integer range
+
+
+
+=== TEST 17: dynamic burst resolving to 0 is accepted (schema allows burst minimum 0)
+--- request
+GET /limit_conn
+--- more_headers
+burst: 0
+--- error_code: 200
+--- no_error_log
+resolved value must be
+
+
+
+=== TEST 18: negative burst header value is rejected with 500 and error log
+--- request
+GET /limit_conn
+--- more_headers
+burst: -1
+--- error_code: 500
+--- error_log
+resolved value must be a non-negative number
diff --git a/t/plugin/limit-req-redis.t b/t/plugin/limit-req-redis.t
