fix(3scale_batcher): read batched report keys while holding the lock

CHANGELOG.md

@@ -28,6 +28,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - Set request timeout for OIDC query. [PR #1601](https://github.com/3scale/APIcast/pull/1601) [THREESCALE-8006](https://redhat.atlassian.net/browse/THREESCALE-8006)
 - Set upstream metrics when sending request via proxy. [PR #1598](https://github.com/3scale/APIcast/pull/1598) [THREESCALE-10571](https://redhat.atlassian.net/browse/THREESCALE-15071)
 - Certificate is never verified in API request for https backend via proxy [PR #1573](https://github.com/3scale/APIcast/pull/1568) [THREESCALE-11944](https://redhat.atlassian.net/browse/THREESCALE-11944)
+- 3scale Batcher Policy: read the batched report keys while holding the lock, so a concurrent flush cannot leave the lock orphaned and stall requests for the lock timeout [Issue #1610](https://github.com/3scale/APIcast/issues/1610)
+- TimerTask: run the task inside a `pcall` so a failing run cannot stop the task from being scheduled again [Issue #1610](https://github.com/3scale/APIcast/issues/1610)
 
 ### Added
 - Update APIcast schema manifest [PR #1550](https://github.com/3scale/APIcast/pull/1550)

gateway/src/apicast/policy/3scale_batcher/reports_batcher.lua

@@ -72,8 +72,6 @@ end
 function _M:get_all(service_id)
   local cached_reports = {}
 
-  local cached_report_keys = self.storage:get_keys()
-
   local lock, new_lock_err = resty_lock:new(self.lock_shdict_name, lock_options)
   if not lock then
     ngx.log(ngx.ERR, 'failed to create lock: ', new_lock_err)
@@ -86,17 +84,30 @@ function _M:get_all(service_id)
     return {}
   end
 
+  -- The keys need to be read while holding the lock. There is one reporting
+  -- timer per worker, so reading them before acquiring it allows another
+  -- worker to delete a key in between, and then storage:get() below returns
+  -- nil for it.
+  local cached_report_keys = self.storage:get_keys()
+
   for _, key in ipairs(cached_report_keys) do
     local value = self.storage:get(key)
-    local report, err = keys_helper.report_from_key_batched_report(key, value)
 
-    if not err then
-      if value > 0 and report and report.service_id == service_id then
-        insert(cached_reports, report)
-        self.storage:delete(key)
+    -- Deletes only happen while holding the lock, but the entry can still be
+    -- gone, for example if it was evicted. Skip it rather than raising, an
+    -- error here would leave the lock to expire on its own and would prevent
+    -- the caller from scheduling the next run.
+    if value then
+      local report, err = keys_helper.report_from_key_batched_report(key, value)
+
+      if not err then
+        if value > 0 and report and report.service_id == service_id then
+          insert(cached_reports, report)
+          self.storage:delete(key)
+        end
+      else
+        ngx.log(ngx.WARN, 'failed to get report for key ', key, ' err: ', err)
       end
-    else
-      ngx.log(ngx.WARN, 'failed to get report for key ', key, ' err: ', err)
     end
   end
 

gateway/src/resty/concurrent/timer_task.lua

@@ -2,6 +2,7 @@ local GC = require 'apicast.gc'
 
 local uuid = require 'resty.jit-uuid'
 local unpack = unpack
+local pcall = pcall
 
 local _M = {}
 
@@ -68,7 +69,15 @@ run_periodic = function(run_now, id, func, args, interval)
   end
 
   if run_now then
-    func(unpack(args))
+    -- The task is run inside a pcall so that a failing run cannot prevent the
+    -- next one from being scheduled. Letting the error propagate aborts the
+    -- timer thread before schedule_next() below, which stops the task from
+    -- ever running again on this worker.
+    local ok, err = pcall(func, unpack(args))
+
+    if not ok then
+      ngx.log(ngx.ERR, 'timer task failed to run: ', err)
+    end
   end
 
   if not _M.last_one[id] then

spec/policy/3scale_batcher/reports_batcher_spec.lua

@@ -5,11 +5,16 @@ local Transaction = require 'apicast.policy.3scale_batcher.transaction'
 local resty_lock = require 'resty.lock'
 
 describe('reports batcher', function()
+  -- Records lock/unlock calls so the tests that care about them can assert on
+  -- the order. Not relevant for most of these tests.
+  local events
+
   before_each(function()
-    -- Stub locks. Not relevant for these tests.
+    events = {}
+
     stub(resty_lock, 'new').returns({
-      lock = function() return true end,
-      unlock = function() return true end
+      lock = function() table.insert(events, 'lock') return true end,
+      unlock = function() table.insert(events, 'unlock') return true end
     })
   end)
 
@@ -93,5 +98,55 @@ describe('reports batcher', function()
         assert.same({}, reports_batcher:get_all('1'))
       end)
     end)
+
+    it('reads the keys while holding the lock', function()
+      -- There is one reporting timer per worker. Listing the keys before
+      -- taking the lock lets another worker delete them in between.
+      local test_dict = {
+        get_keys = function() table.insert(events, 'get_keys') return {} end,
+        get = function() end,
+        delete = function() end
+      }
+
+      ReportsBatcher.new(test_dict, 'batched_reports_locks'):get_all('1')
+
+      assert.same({ 'lock', 'get_keys', 'unlock' }, events)
+    end)
+
+    describe('when a key is gone after it has been listed', function()
+      local service_id = '1'
+      local credentials = { user_key = 'uk' }
+
+      local missing_key = keys_helper.key_for_batched_report(
+        service_id, credentials, 'missing')
+      local existing_key = keys_helper.key_for_batched_report(
+        service_id, credentials, 'existing')
+
+      local function build_dict()
+        local storage = { [existing_key] = 1 }
+
+        return {
+          get_keys = function() return { missing_key, existing_key } end,
+          get = function(_, key) return storage[key] end,
+          delete = function(_, key) storage[key] = nil end
+        }
+      end
+
+      it('skips it and returns the remaining reports', function()
+        local reports_batcher = ReportsBatcher.new(
+          build_dict(), 'batched_reports_locks')
+
+        local reports = reports_batcher:get_all(service_id)
+
+        assert.equals(1, #reports)
+        assert.equals('existing', reports[1].metric)
+      end)
+
+      it('releases the lock', function()
+        ReportsBatcher.new(build_dict(), 'batched_reports_locks'):get_all(service_id)
+
+        assert.same({ 'lock', 'unlock' }, events)
+      end)
+    end)
   end)
 end)

spec/resty/concurrent/timer_task_spec.lua

@@ -132,6 +132,24 @@ describe('TimerTask', function()
       end)
     end)
 
+    describe('when the task raises an error', function()
+      local failing_task = function() error('something went wrong') end
+
+      it('does not propagate the error', function()
+        local timer_task = TimerTask.new(failing_task, { args = args, interval = interval })
+
+        assert.has_no.errors(function() timer_task:execute(true) end)
+      end)
+
+      it('schedules the next one', function()
+        local timer_task = TimerTask.new(failing_task, { args = args, interval = interval })
+
+        timer_task:execute(true)
+
+        assert.stub(ngx_timer_stub).was_called()
+      end)
+    end)
+
     describe('when the task is not active', function()
       it('does not run the task', function()
         local timer_task = TimerTask.new(func, { args = args, interval = interval })