This is an automated email from the ASF dual-hosted git repository.
nic-6443 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 88c4b8c47 fix(limit-count): refresh the delayed-sync quota after the
local delta expired (#14015)
88c4b8c47 is described below
commit 88c4b8c47d9d59725d22e02aca0364388c76e51f
Author: Nic <[email protected]>
AuthorDate: Thu Oct 8 18:08:52 2026 +0800
fix(limit-count): refresh the delayed-sync quota after the local delta
expired (#14015)
---
apisix/plugins/limit-count/delayed-syncer.lua | 22 +-
t/plugin/limit-count-redis-delayed-sync3.t | 350 ++++++++++++++++++++++++++
2 files changed, 366 insertions(+), 6 deletions(-)
diff --git a/apisix/plugins/limit-count/delayed-syncer.lua
b/apisix/plugins/limit-count/delayed-syncer.lua
index 369975925..acb3689ff 100644
--- a/apisix/plugins/limit-count/delayed-syncer.lua
+++ b/apisix/plugins/limit-count/delayed-syncer.lua
@@ -125,10 +125,14 @@ function _M.sync_to_shm(self, key, remaining, reset,
local_delta, info)
return err
end
- _, err = self.shd:incr(self:key_local_delta(key), -local_delta, 0, 2 *
self.window)
- if err then
- core.log.error("incr local delta shm to failed: ", err, ", key: ", key)
- return err
+ -- nothing was flushed: don't create an empty local delta entry, a failed
+ -- sync would take it for a pending delta and charge the fallback limiter
+ if local_delta ~= 0 then
+ _, err = self.shd:incr(self:key_local_delta(key), -local_delta, 0, 2 *
self.window)
+ if err then
+ core.log.error("incr local delta shm to failed: ", err, ", key: ",
key)
+ return err
+ end
end
return nil, quota
@@ -357,7 +361,12 @@ local function sync_key(self, key)
core.log.error("get local delta from shm failed: ", err)
end
- if delta then
+ -- the key expires two windows after its creation and is not renewed, so a
+ -- node that only rejects loses it; flush anyway, otherwise that node keeps
+ -- judging on the same cached quota until the quota itself expires
+ if not err then
+ local had_delta = delta ~= nil
+ delta = delta or 0
local flush = self.limiter.commit or self.limiter.incoming
local _, remaining_or_err, reset, info = flush(self.limiter, key,
delta)
-- compat
@@ -365,7 +374,8 @@ local function sync_key(self, key)
self:sync_to_shm(key, remaining_or_err, reset, delta, info)
elseif remaining_or_err ~= "rejected" then
core.log.error("sync to redis failed: ", remaining_or_err, ", key:
", key)
- if self.limiter.fallback_limiter then
+ -- without a local delta there is nothing to account for locally
+ if self.limiter.fallback_limiter and had_delta then
core.log.warn("try use fallback limiter to do rate limiting")
if delta < 1 then
delta = 1
diff --git a/t/plugin/limit-count-redis-delayed-sync3.t
b/t/plugin/limit-count-redis-delayed-sync3.t
new file mode 100644
index 000000000..2c0f4d5b9
--- /dev/null
+++ b/t/plugin/limit-count-redis-delayed-sync3.t
@@ -0,0 +1,350 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+use t::APISIX 'no_plan';
+
+repeat_each(1);
+no_long_string();
+no_shuffle();
+no_root_location();
+
+add_block_preprocessor(sub {
+ my ($block) = @_;
+
+ if (!$block->request) {
+ $block->set_value("request", "GET /t");
+ }
+
+ if (!$block->error_log && !$block->no_error_log) {
+ $block->set_value("no_error_log", "[error]\n[alert]");
+ }
+});
+
+run_tests;
+
+__DATA__
+
+=== TEST 1: sliding window, redis, delayed sync
+--- 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-count": {
+ "count": 2,
+ "time_window": 60,
+ "window_type": "sliding",
+ "key_type": "constant",
+ "key": "delayed-sync-stale-snapshot",
+ "policy": "redis",
+ "redis_host": "127.0.0.1",
+ "sync_interval": 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)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 2: a node that only rejects still refreshes its quota after its local
delta expired
+--- config
+ location /t {
+ content_by_lua_block {
+ local http = require("resty.http")
+ local redis = require("resty.redis")
+ local uri = "http://127.0.0.1:"; .. ngx.var.server_port .. "/hello"
+
+ local function hit()
+ local res, err = http.new():request_uri(uri)
+ return res and res.status or err
+ end
+
+ local function drop_redis_counters()
+ local red = redis:new()
+ assert(red:connect("127.0.0.1", 6379))
+ local keys = assert(red:keys("*delayed-sync-stale-snapshot*"))
+ for _, key in ipairs(keys) do
+ red:del(key)
+ end
+ red:close()
+ end
+
+ -- keep all requests inside one 60s sliding window
+ local left = 60 - ngx.now() % 60
+ if left < 5 then
+ ngx.sleep(left + 0.1)
+ end
+ drop_redis_counters()
+
+ local codes = {hit(), hit(), hit()}
+ -- let the delayed sync flush the local delta
+ ngx.sleep(0.3)
+
+ -- the local delta key lives two windows from its creation and is
+ -- not renewed; drop it as if that time had passed
+ local dict = ngx.shared["plugin-limit-count"]
+ for _, key in ipairs(dict:get_keys(0)) do
+ if key:find("^local_delta#") then
+ dict:delete(key)
+ end
+ end
+
+ -- free the quota in Redis, as the previous window's weight
+ -- decaying or another node's view would
+ drop_redis_counters()
+
+ -- judged on the cached quota, which is still exhausted
+ codes[4] = hit()
+ -- the sync this rejection scheduled must refresh the cached quota
+ ngx.sleep(0.3)
+ codes[5] = hit()
+
+ ngx.say(table.concat(codes, " "))
+ }
+ }
+--- response_body
+200 200 503 503 200
+--- timeout: 10
+
+
+
+=== TEST 3: fixed window, unreachable redis, delayed sync
+--- 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-count": {
+ "count": 1,
+ "time_window": 60,
+ "window_type": "fixed",
+ "key_type": "constant",
+ "key": "delayed-sync-no-redis",
+ "policy": "redis",
+ "redis_host": "127.0.0.1",
+ "redis_port": 16399,
+ "sync_interval": 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)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 4: without a local delta, a failed sync does not charge the fallback
limiter
+--- config
+ location /t {
+ content_by_lua_block {
+ local http = require("resty.http")
+ local uri = "http://127.0.0.1:"; .. ngx.var.server_port .. "/hello"
+ local dict = ngx.shared["plugin-limit-count"]
+
+ local function hit()
+ local res, err = http.new():request_uri(uri)
+ return res and res.status or err
+ end
+
+ -- what the fallback limiter has counted for this route
+ local function fallback_count()
+ local sum = 0
+ for _, key in ipairs(dict:get_keys(0)) do
+ -- the fixed window counter is stored under the limit key
+ if key:find("^/apisix/routes/")
+ and key:find("delayed-sync-no-redis", 1, true) then
+ sum = sum + dict:get(key)
+ end
+ end
+ return sum
+ end
+
+ -- admitted on the fallback limiter, then synced to it
+ local codes = {hit()}
+ ngx.sleep(0.3)
+ local before = fallback_count()
+
+ -- the local delta key expired: the node has nothing to sync
+ for _, key in ipairs(dict:get_keys(0)) do
+ if key:find("^local_delta#") then
+ dict:delete(key)
+ end
+ end
+
+ -- rejected, and the sync it schedules fails against Redis
+ codes[2] = hit()
+ ngx.sleep(0.3)
+
+ ngx.say(table.concat(codes, " "), ", fallback count ", before, "
-> ",
+ fallback_count())
+ }
+ }
+--- response_body
+200 503, fallback count 1 -> 1
+--- error_log
+sync to redis failed
+--- timeout: 10
+
+
+
+=== TEST 5: fixed window, redis, delayed sync
+--- 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-count": {
+ "count": 1,
+ "time_window": 60,
+ "window_type": "fixed",
+ "key_type": "constant",
+ "key": "delayed-sync-redis-error",
+ "policy": "redis",
+ "redis_host": "127.0.0.1",
+ "sync_interval": 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)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 6: a quota refresh without a local delta leaves nothing to charge
when redis fails later
+--- config
+ location /t {
+ content_by_lua_block {
+ local http = require("resty.http")
+ local redis = require("resty.redis")
+ local uri = "http://127.0.0.1:"; .. ngx.var.server_port .. "/hello"
+ local dict = ngx.shared["plugin-limit-count"]
+
+ local function hit()
+ local res, err = http.new():request_uri(uri)
+ return res and res.status or err
+ end
+
+ local function redis_keys()
+ local red = redis:new()
+ assert(red:connect("127.0.0.1", 6379))
+ return red, assert(red:keys("*delayed-sync-redis-error*"))
+ end
+
+ -- what the fallback limiter has counted for this route
+ local function fallback_count()
+ local sum = 0
+ for _, key in ipairs(dict:get_keys(0)) do
+ if key:find("^/apisix/routes/")
+ and key:find("delayed-sync-redis-error", 1, true) then
+ sum = sum + dict:get(key)
+ end
+ end
+ return sum
+ end
+
+ local red, keys = redis_keys()
+ for _, key in ipairs(keys) do
+ red:del(key)
+ end
+
+ local codes = {hit()}
+ ngx.sleep(0.3)
+
+ -- the local delta key expired
+ for _, key in ipairs(dict:get_keys(0)) do
+ if key:find("^local_delta#") then
+ dict:delete(key)
+ end
+ end
+
+ -- rejected; its sync refreshes the quota from Redis with nothing
+ -- to flush
+ codes[2] = hit()
+ ngx.sleep(0.3)
+
+ -- make the next sync fail: the counter is no longer an integer
+ red, keys = redis_keys()
+ for _, key in ipairs(keys) do
+ red:set(key, "not-a-number", "KEEPTTL")
+ end
+
+ local before = fallback_count()
+ codes[3] = hit()
+ ngx.sleep(0.3)
+
+ ngx.say(table.concat(codes, " "), ", fallback count ", before, "
-> ",
+ fallback_count())
+ }
+ }
+--- response_body
+200 503 503, fallback count 0 -> 0
+--- error_log
+sync to redis failed
+--- timeout: 10
