This is an automated email from the ASF dual-hosted git repository.
shreemaan-abhishek pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new 6e877eb44 fix(limit): atomic redis commits and resolved-var validation
(#13467)
6e877eb44 is described below
commit 6e877eb44f68d0a34385cf42fcff403d10acc97e
Author: Shreemaan Abhishek <[email protected]>
AuthorDate: Mon Jun 8 18:52:38 2026 +0800
fix(limit): atomic redis commits and resolved-var validation (#13467)
---
apisix/plugins/limit-conn/init.lua | 33 ++++-
apisix/plugins/limit-req/util.lua | 120 +++++++++++-----
t/plugin/limit-conn-variable.t | 117 ++++++++++++++++
t/plugin/limit-req-redis.t | 272 +++++++++++++++++++++++++++++++++++++
4 files changed, 506 insertions(+), 36 deletions(-)
diff --git a/apisix/plugins/limit-conn/init.lua
b/apisix/plugins/limit-conn/init.lua
index b45ff567c..0d45bd290 100644
--- 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
index 6724889f1..e12660453 100644
--- 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
index e1db32c1e..0b6dd2287 100644
--- 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
index fd67468f2..ad5468a4b 100644
--- a/t/plugin/limit-req-redis.t
+++ b/t/plugin/limit-req-redis.t
@@ -667,3 +667,275 @@ passed
GET /t
--- response_body eval
qr/property \"rate\" validation failed: expected 0 to be greater than 0/
+
+
+
+=== TEST 25: set route for hash-tag key-format tests (rate=100, burst=0)
+--- 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-req": {
+ "rate": 100,
+ "burst": 0,
+ "rejected_code": 503,
+ "key": "remote_addr",
+ "policy": "redis",
+ "redis_host": "127.0.0.1"
+ }
+ },
+ "upstream": {
+ "nodes": {
+ "127.0.0.1:1980": 1
+ },
+ "type": "roundrobin"
+ },
+ "uri": "/hello"
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ end
+ ngx.say(body)
+ }
+ }
+--- request
+GET /t
+--- response_body
+passed
+
+
+
+=== TEST 26: atomic script stores state in a single hash key
+--- config
+ location /t {
+ content_by_lua_block {
+ -- Flush all Redis state so this test starts clean.
+ local redis = require "resty.redis"
+ local red = redis:new()
+ red:set_timeout(1000)
+ local ok, err = red:connect("127.0.0.1", 6379)
+ if not ok then
+ ngx.say("connect failed: ", err)
+ return
+ end
+ local ok_flush, err_flush = red:flushall()
+ if not ok_flush then
+ ngx.say("flushall failed: ", err_flush)
+ return
+ end
+
+ -- Make one request to trigger the atomic script.
+ local httpc = require "resty.http"
+ local hc = httpc:new()
+ local port = ngx.var.server_port
+ local res, err2 = hc:request_uri("http://127.0.0.1:"; .. port ..
"/hello")
+ if not res then
+ ngx.say("request failed: ", err2)
+ return
+ end
+ ngx.say("status: ", res.status)
+
+ -- Verify the state lives in a single hash key limit_req:<key>
+ -- holding both the "excess" and "last" fields.
+ local keys, err3 = red:keys("limit_req:*")
+ if not keys then
+ ngx.say("keys cmd failed: ", err3)
+ return
+ end
+ ngx.say("state keys found: ", #keys)
+
+ local fields, err4 = red:hmget(keys[1], "excess", "last")
+ if not fields then
+ ngx.say("hmget failed: ", err4)
+ return
+ end
+ local has_fields = fields[1] ~= ngx.null and fields[2] ~= ngx.null
+ ngx.say("excess and last fields present: ", has_fields and "yes"
or "no")
+ }
+ }
+--- request
+GET /t
+--- response_body
+status: 200
+state keys found: 1
+excess and last fields present: yes
+--- no_error_log
+[error]
+
+
+
+=== TEST 27: set route for first-request tests (rate=1, burst=0)
+--- config
+ location /t {
+ content_by_lua_block {
+ -- Flush so no stale excess remains from previous tests.
+ local redis = require "resty.redis"
+ local red = redis:new()
+ red:set_timeout(1000)
+ local ok, err = red:connect("127.0.0.1", 6379)
+ if not ok then
+ ngx.say("connect failed: ", err)
+ return
+ end
+ local ok_flush, err_flush = red:flushall()
+ if not ok_flush then
+ ngx.say("flushall failed: ", err_flush)
+ return
+ end
+
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1',
+ ngx.HTTP_PUT,
+ [[{
+ "plugins": {
+ "limit-req": {
+ "rate": 1,
+ "burst": 0,
+ "rejected_code": 503,
+ "key": "remote_addr",
+ "policy": "redis",
+ "redis_host": "127.0.0.1"
+ }
+ },
+ "upstream": {
+ "nodes": {
+ "127.0.0.1:1980": 1
+ },
+ "type": "roundrobin"
+ },
+ "uri": "/hello"
+ }]]
+ )
+ if code >= 300 then
+ ngx.status = code
+ ngx.say("route update failed: ", body)
+ return
+ end
+ ngx.say(body)
+ }
+ }
+--- request
+GET /t
+--- response_body
+passed
+
+
+
+=== TEST 28: first request is always admitted even when burst=0 (no prior
state)
+--- pipelined_requests eval
+["GET /hello", "GET /hello"]
+--- error_code eval
+[200, 503]
+
+
+
+=== TEST 29: set route for atomic-script concurrency tests (rate=1, burst=0)
+--- config
+ location /t {
+ content_by_lua_block {
+ -- Flush so no stale excess remains from previous tests.
+ local redis = require "resty.redis"
+ local red = redis:new()
+ red:set_timeout(1000)
+ local ok, err = red:connect("127.0.0.1", 6379)
+ if not ok then
+ ngx.say("connect failed: ", err)
+ return
+ end
+ local ok_flush, err_flush = red:flushall()
+ if not ok_flush then
+ ngx.say("flushall failed: ", err_flush)
+ return
+ end
+
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1',
+ ngx.HTTP_PUT,
+ [[{
+ "plugins": {
+ "limit-req": {
+ "rate": 1,
+ "burst": 0,
+ "rejected_code": 503,
+ "key": "remote_addr",
+ "policy": "redis",
+ "redis_host": "127.0.0.1"
+ }
+ },
+ "upstream": {
+ "nodes": {
+ "127.0.0.1:1980": 1
+ },
+ "type": "roundrobin"
+ },
+ "uri": "/hello"
+ }]]
+ )
+ if code >= 300 then
+ ngx.status = code
+ ngx.say("route update failed: ", body)
+ return
+ end
+ ngx.say(body)
+ }
+ }
+--- request
+GET /t
+--- response_body
+passed
+
+
+
+=== TEST 30: atomic script prevents concurrent requests from bypassing rate
limit
+--- config
+ location /hello_proxy {
+ content_by_lua_block {
+ local httpc = require "resty.http"
+ local hc = httpc:new()
+ local port = ngx.var.server_port
+ local res, err = hc:request_uri("http://127.0.0.1:"; .. port ..
"/hello")
+ if not res then
+ ngx.exit(500)
+ end
+ ngx.exit(res.status)
+ }
+ }
+
+ location /t {
+ content_by_lua_block {
+ -- Fire 5 concurrent sub-requests via capture_multi.
+ -- Each sub-request calls /hello through a separate resty.http
connection,
+ -- so the atomic Redis Lua script is exercised under concurrent
load.
+ -- With rate=1, burst=0 and a clean Redis state, exactly 1 request
+ -- should pass (status 200) and the remaining 4 should be rejected
(503).
+ local reqs = {}
+ for i = 1, 5 do
+ reqs[i] = { "/hello_proxy" }
+ end
+ local resps = { ngx.location.capture_multi(reqs) }
+ local ok_count = 0
+ local reject_count = 0
+ for _, resp in ipairs(resps) do
+ if resp.status == 200 then
+ ok_count = ok_count + 1
+ elseif resp.status == 503 then
+ reject_count = reject_count + 1
+ end
+ end
+ ngx.say("admitted: ", ok_count)
+ ngx.say("rejected: ", reject_count)
+ }
+ }
+--- request
+GET /t
+--- response_body
+admitted: 1
+rejected: 4
+--- no_error_log
+[error]
