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

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - Correct FAPI header to `x-fapi-interaction-id` [PR #1557](https://github.com/3scale/APIcast/pull/1557) [THREESCALE-11957](https://issues.redhat.com/browse/THREESCALE-11957)
 - Only validate oidc setting if authentication method is set to oidc [PR #1568](https://github.com/3scale/APIcast/pull/1568) [THREESCALE-11441](https://issues.redhat.com/browse/THREESCALE-11441)
 - Reduce memory consumption when returning large response that has been routed through a proxy server. [PR #1572](https://github.com/3scale/APIcast/pull/1572) [THREESCALE-12258](https://issues.redhat.com/browse/THREESCALE-12258)
+- 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 [PR #1611](https://github.com/3scale/APIcast/pull/1611) [THREESCALE-16640](https://issues.redhat.com/browse/THREESCALE-16640)
 
 ### 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
 

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)