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 4d8106839 fix: batch-processor infinite timer loop prevents graceful
shutdown (#13288)
4d8106839 is described below
commit 4d81068390a963e6c7c954fb3c44d3caaf2214a3
Author: Nic <[email protected]>
AuthorDate: Mon Apr 27 10:06:58 2026 +0800
fix: batch-processor infinite timer loop prevents graceful shutdown (#13288)
---
apisix/utils/batch-processor.lua | 13 +++++++----
t/plugin/prometheus2.t | 8 +++++++
t/plugin/prometheus3.t | 2 ++
t/utils/batch-processor.t | 50 ++++++++++++++++++++++++++++++++++++++++
4 files changed, 69 insertions(+), 4 deletions(-)
diff --git a/apisix/utils/batch-processor.lua b/apisix/utils/batch-processor.lua
index cd2421af6..4ede38674 100644
--- a/apisix/utils/batch-processor.lua
+++ b/apisix/utils/batch-processor.lua
@@ -21,6 +21,7 @@ local ipairs = ipairs
local table = table
local now = ngx.now
local type = type
+local exiting = ngx.worker.exiting
local batch_processor = {}
local batch_processor_mt = {
__index = batch_processor
@@ -52,9 +53,12 @@ local function schedule_func_exec(self, delay, batch)
local hdl, err = timer_at(delay, execute_func, self, batch)
if not hdl then
if err == "process exiting" then
- -- it is allowed to create zero-delay timers even when
- -- the Nginx worker process starts shutting down
- timer_at(0, execute_func, self)
+ local hdl2, err2 = timer_at(0, execute_func, self, batch)
+ if not hdl2 then
+ core.log.error("failed to create fallback process timer ",
+ "while exiting: ", err2)
+ return
+ end
else
core.log.error("failed to create process timer: ", err)
return
@@ -118,7 +122,8 @@ end
local function flush_buffer(premature, self)
- if now() - self.last_entry_t >= self.inactive_timeout or
+ if premature or exiting() or
+ now() - self.last_entry_t >= self.inactive_timeout or
now() - self.first_entry_t >= self.buffer_duration
then
core.log.debug("Batch Processor[", self.name ,"] buffer ",
diff --git a/t/plugin/prometheus2.t b/t/plugin/prometheus2.t
index cf303ce77..bcc1e915f 100644
--- a/t/plugin/prometheus2.t
+++ b/t/plugin/prometheus2.t
@@ -458,6 +458,8 @@ GET /apisix/prometheus/metrics
--- error_code: 200
--- response_body_like eval
qr/apisix_batch_process_entries\{name="zipkin_report",route_id="9",server_addr="127.0.0.1"\}
\d+/
+--- error_log
+Batch Processor[zipkin_report] failed to process entries
@@ -519,6 +521,8 @@ GET /apisix/prometheus/metrics
--- error_code: 200
--- response_body_like eval
qr/apisix_batch_process_entries\{name="http-logger",route_id="9",server_addr="127.0.0.1"\}
\d+/
+--- error_log
+Batch Processor[http-logger] failed to process entries
@@ -581,6 +585,8 @@ GET /apisix/prometheus/metrics
--- error_code: 200
--- response_body_like eval
qr/apisix_batch_process_entries\{name="tcp-logger",route_id="10",server_addr="127.0.0.1"\}
\d+/
+--- error_log
+Batch Processor[tcp-logger] failed to process entries
@@ -704,6 +710,8 @@ GET /apisix/prometheus/metrics
--- error_code: 200
--- response_body_like eval
qr/apisix_batch_process_entries\{name="sls-logger",route_id="10",server_addr="127.0.0.1"\}
\d+/
+--- error_log
+Batch Processor[sls-logger] failed to process entries
diff --git a/t/plugin/prometheus3.t b/t/plugin/prometheus3.t
index 57c2f73df..21c0d7bab 100644
--- a/t/plugin/prometheus3.t
+++ b/t/plugin/prometheus3.t
@@ -255,6 +255,8 @@ qr/apisix_batch_process_entries\{name="http
logger",route_id="1",server_addr="12
}
--- response_body
passed
+--- error_log
+Batch Processor[error-log-logger] failed to process entries
diff --git a/t/utils/batch-processor.t b/t/utils/batch-processor.t
index e1ce83b76..5792329cf 100644
--- a/t/utils/batch-processor.t
+++ b/t/utils/batch-processor.t
@@ -481,3 +481,53 @@ Batch Processor[log buffer] failed to process entries
[2/3]: error after consumi
Batch Processor[log buffer] failed to process entries [1/2]: error after
consuming single entry
[{"msg":"4"}]
--- wait: 2
+
+
+
+=== TEST 13: batch processor exits cleanly during shutdown with active entries
+# The real bug triggers when a buffer timer naturally expires (not via
abort_pending_timers)
+# during shutdown. With short inactive_timeout, the timer fires with
premature=false while
+# ngx.worker.exiting() is true. Without the exiting() check, flush_buffer
would re-arm a
+# zero-delay timer in an infinite loop, preventing worker exit.
+# Using short timeouts + continuous pushing to create the race condition.
+--- config
+ location /t {
+ content_by_lua_block {
+ local Batch = require("apisix.utils.batch-processor")
+ local func_to_send = function(elements)
+ return true
+ end
+
+ local config = {
+ max_retry_count = 0,
+ batch_max_size = 1000,
+ buffer_duration = 60,
+ inactive_timeout = 1,
+ retry_delay = 0,
+ }
+
+ local log_buffer, err = Batch:new(func_to_send, config)
+ if not log_buffer then
+ ngx.say(err)
+ return
+ end
+
+ -- push entries with short gaps to keep last_entry_t fresh
+ local count = 0
+ local function keep_pushing(premature)
+ if premature then return end
+ log_buffer:push({msg = "entry-" .. count})
+ count = count + 1
+ if count < 5 then
+ ngx.timer.at(0.3, keep_pushing)
+ end
+ end
+ keep_pushing(false)
+ ngx.say("done")
+ }
+ }
+--- request
+GET /t
+--- response_body
+done
+--- wait: 2
