fix(queues): continue processing after hard error in handler (#11423)

There was an issue with the queue implementation where processing would stop when encountering a hard error in the handler function. The changes in the code fix this problem.

Now, the handler function is called using pcall to catch any errors that occur. If a hard error occurs, the error is logged but the processing of the remaining entries in the queue continues.

CHANGELOG.md

@@ -40,6 +40,8 @@
   [#11385](https://github.com/Kong/kong/pull/11385)
 - Fix cache warmup mechanism not working in `acls` plugin groups config entity scenario.
   [#11414](https://github.com/Kong/kong/pull/11414)
+- Fix an issue that queue stops processing when a hard error is encountered in the handler function.
+  [#11423](https://github.com/Kong/kong/pull/11423)
 
 #### Plugins
 

kong/tools/queue.lua

@@ -267,12 +267,20 @@ function Queue:process_once()
   local retry_count = 0
   while true do
     self:log_debug("passing %d entries to handler", entry_count)
-    ok, err = self.handler(self.handler_conf, batch)
-    if ok then
-      self:log_debug("handler processed %d entries sucessfully", entry_count)
+    local status
+    status, ok, err = pcall(self.handler, self.handler_conf, batch)
+    if status and ok == true then
+      self:log_debug("handler processed %d entries successfully", entry_count)
       break
     end
 
+    if not status then
+      -- protected call failed, ok is the error message
+      err = ok
+    end
+
+    self:log_warn("handler could not process entries: %s", tostring(err or "no error details returned by handler"))
+
     if not err then
       self:log_err("handler returned falsy value but no error information")
     end
@@ -284,8 +292,6 @@ function Queue:process_once()
       break
     end
 
-    self:log_warn("handler could not process entries: %s", tostring(err))
-
     -- Delay before retrying.  The delay time is calculated by multiplying the configured initial_retry_delay with
     -- 2 to the power of the number of retries, creating an exponential increase over the course of each retry.
     -- The maximum time between retries is capped by the max_retry_delay configuration parameter.

spec/01-unit/27-queue_spec.lua

@@ -755,4 +755,36 @@ describe("plugin queue", function()
     assert.equals(123, converted_parameters.max_batch_size)
     assert.equals(234, converted_parameters.max_coalescing_delay)
   end)
+
+  it("continue processing after hard error in handler", function()
+    local processed = {}
+    local function enqueue(entry)
+      Queue.enqueue(
+        queue_conf({
+          name = "continue-processing",
+          max_batch_size = 1,
+          max_entries = 5,
+          max_coalescing_delay = 0.1,
+          max_retry_time = 3,
+        }),
+        function(_, batch)
+          if batch[1] == "Two" then
+            error("hard error")
+          end
+          table.insert(processed, batch[1])
+          return true
+        end,
+        nil,
+        entry
+      )
+    end
+    enqueue("One")
+    enqueue("Two")
+    enqueue("Three")
+    wait_until_queue_done("continue-processing")
+    assert.equal("One", processed[1])
+    assert.equal("Three", processed[2])
+    assert.match_re(log_messages, 'WARN \\[\\] queue continue-processing: handler could not process entries: .*: hard error')
+    assert.match_re(log_messages, 'ERR \\[\\] queue continue-processing: could not send entries, giving up after \\d retries.  1 queue entries were lost')  
+  end)
 end)