fix: batch-processor infinite timer loop prevents graceful shutdown (#13288)

Author: nic-6443

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 ",

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
 
 
 

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
 
 
 

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