diff --git a/CHANGELOG.md b/CHANGELOG.md
index ddd13ba..b929e18 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,11 @@ of changes.
 - Reuse expired index slots while their dictionary nodes still exist, avoid
   expiry-driven broadcasts, renew known slots without synchronizing the full
   index, and detect a re-created `key_count` (#25).
+- Stagger worker cleanup timers, skip slots whose deadlines have not passed,
+  and yield between groups of TTL checks (#23).
+- Reclaim expired entries once per `remove_expired_keys_interval` for the
+  whole dictionary instead of once per worker; a value of 0 or less falls back
+  to the default (#23).
 
 ## 1.0.0
 
diff --git a/README.md b/README.md
index ff28be6..ec0c499 100644
--- a/README.md
+++ b/README.md
@@ -112,6 +112,17 @@ section of nginx configuration.
     RAM, you might want to increase this to avoid cache getting flushed too
     often. Decreasing this makes sense if you have a very large number of
     metrics or need to minimize memory usage of this library.
+  * `remove_expired_keys_interval` (number): cleanup interval in seconds.
+    Defaults to 3600, which is also the maximum: larger values are capped to
+    it. Cleanup checks slots whose recorded deadlines have passed and yields
+    between groups of TTL checks. Choose an interval based on dictionary
+    capacity and label churn: longer intervals retain more expired entries
+    until reclamation. A value of 0 or less is replaced by the default.
+    Worker cleanup timers are staggered across workers. Expired entries are
+    reclaimed once per interval for the whole dictionary, by the first worker
+    to finish its cleanup in that interval: the reclamation holds the
+    dictionary lock while it scans all entries, so its latency grows with the
+    number of entries in the dictionary.
 
 Returns a `prometheus` object that should be used to register metrics.
 
diff --git a/integration/cleanup.lua b/integration/cleanup.lua
new file mode 100644
index 0000000..99c7638
--- /dev/null
+++ b/integration/cleanup.lua
@@ -0,0 +1,32 @@
+local p = cleanup_prometheus
+local d = ngx.shared.cleanup_metrics
+local marker = "__ngx_prom__flush_expired"
+
+local function set_expired()
+  for i = 1, 10 do
+    assert(d:set("expired_" .. i, i, 0.01))
+  end
+  ngx.sleep(0.02)
+end
+
+-- start from an interval nobody has reclaimed in yet
+d:delete(marker)
+d:set("permanent", 42)
+set_expired()
+p.key_index:remove_expired_keys()
+for i = 1, 10 do
+  assert(d:get_stale("expired_" .. i) == nil, "did not reclaim the expired entries")
+end
+assert(d:get("permanent") == 42, "permanent entry was reclaimed")
+assert(d:get(marker), "did not take the interval")
+
+-- the interval is taken: another round leaves the reclamation alone
+set_expired()
+p.key_index:remove_expired_keys()
+assert(d:get_stale("expired_1") ~= nil, "reclaimed twice in one interval")
+
+-- once it has passed, the next round reclaims again
+d:delete(marker)
+p.key_index:remove_expired_keys()
+assert(d:get_stale("expired_1") == nil, "did not reclaim in the next interval")
+ngx.say("ok")
diff --git a/integration/nginx.conf b/integration/nginx.conf
index f8bc3ae..8baaa6b 100644
--- a/integration/nginx.conf
+++ b/integration/nginx.conf
@@ -11,6 +11,7 @@ http {
 
     lua_shared_dict prometheus_metrics 10M;
     lua_shared_dict index_metrics 1M;
+    lua_shared_dict cleanup_metrics 16M;
     lua_package_path "/nginx-lua-prometheus/?.lua;;";
 
     error_log stderr;
@@ -20,6 +21,7 @@ http {
 	  {sync_interval=0.4})
         index_writer = require("prometheus").init("index_metrics")
         index_reader = require("prometheus").init("index_metrics")
+        cleanup_prometheus = require("prometheus").init("cleanup_metrics")
         metric_requests = prometheus:counter("requests_total",
           "Number of HTTP requests", {"host", "status"})
         metric_latency = prometheus:histogram("request_duration_seconds",
@@ -44,6 +46,10 @@ http {
             log_by_lua_block { return }
             content_by_lua_file /nginx-lua-prometheus/integration/index.lua;
         }
+        location = /test_cleanup {
+            log_by_lua_block { return }
+            content_by_lua_file /nginx-lua-prometheus/integration/cleanup.lua;
+        }
         location /error {
             return 500;
         }
diff --git a/integration/test.go b/integration/test.go
index 07c8c76..1bce01d 100644
--- a/integration/test.go
+++ b/integration/test.go
@@ -124,6 +124,17 @@ func main() {
 		log.Fatalf("Key index regression test returned HTTP %d", indexResp.StatusCode)
 	}
 
+	cleanupResp, err := indexClient.Get("http://localhost:18001/test_cleanup")
+	if err != nil {
+		log.Fatalf("Cleanup regression test failed: %v", err)
+	}
+	if err := cleanupResp.Body.Close(); err != nil {
+		log.Fatalf("Could not close the cleanup regression response: %v", err)
+	}
+	if cleanupResp.StatusCode != http.StatusOK {
+		log.Fatalf("Cleanup regression test returned HTTP %d", cleanupResp.StatusCode)
+	}
+
 	// Use a custom http client with a lower idle connection timeout and a request timeout.
 	client := &http.Client{
 		Timeout:   500 * time.Millisecond,
diff --git a/prometheus.lua b/prometheus.lua
index 8556b8d..00bed8c 100644
--- a/prometheus.lua
+++ b/prometheus.lua
@@ -742,9 +742,15 @@ function Prometheus.init(dict_name, options_or_prefix)
       DEFAULT_SYNC_INTERVAL
     self.lookup_max_size = options_or_prefix.lookup_max_size or
       DEFAULT_LOOKUP_MAX_SIZE
-    self.remove_expired_keys_interval = options_or_prefix.remove_expired_keys_interval
-        and options_or_prefix.remove_expired_keys_interval < MAX_REMOVE_EXPIRED_KEYS_INTERVAL
-        and options_or_prefix.remove_expired_keys_interval
+    local interval = options_or_prefix.remove_expired_keys_interval
+    if interval and interval <= 0 then
+      ngx.log(ngx.WARN, "remove_expired_keys_interval must be positive, using ",
+              MAX_REMOVE_EXPIRED_KEYS_INTERVAL)
+      interval = nil
+    end
+    self.remove_expired_keys_interval = interval
+        and interval < MAX_REMOVE_EXPIRED_KEYS_INTERVAL
+        and interval
       or MAX_REMOVE_EXPIRED_KEYS_INTERVAL
   else
     self.prefix = options_or_prefix or ''
diff --git a/prometheus_keys.lua b/prometheus_keys.lua
index 90bffe0..badb0df 100644
--- a/prometheus_keys.lua
+++ b/prometheus_keys.lua
@@ -14,10 +14,40 @@ KeyIndex.__index = KeyIndex
 -- and the index converges even when far more slots need repairing.
 local MAX_KEY_COUNT_REPAIRS = 1000
 
+local DEFAULT_REMOVE_EXPIRED_KEYS_INTERVAL = 600
+
+-- Yield between groups of shared-dictionary TTL checks.
+local REMOVE_EXPIRED_KEYS_YIELD_EVERY = 500
+
+KeyIndex.REMOVE_EXPIRED_KEYS_YIELD_EVERY = REMOVE_EXPIRED_KEYS_YIELD_EVERY
+
 
 -- check and remove expired keys
-local function remove_expired_keys(_, self)
-  self:remove_expired_keys()
+local function remove_expired_keys(premature, self)
+  if not premature then
+    self:remove_expired_keys()
+  end
+end
+
+
+-- Stagger worker cleanup timers to avoid starting every scan together.
+local function start_remove_expired_keys_timer(self, interval)
+  local id = ngx.worker.id() or 0
+  local count = ngx.worker.count()
+
+  local offset = interval * id / count
+  if offset <= 0 then
+    ngx.timer.every(interval, remove_expired_keys, self)
+    return
+  end
+
+  ngx.timer.at(offset, function(premature, s)
+    if premature then
+      return
+    end
+    ngx.timer.every(interval, remove_expired_keys, s)
+    s:remove_expired_keys()
+  end, self)
 end
 
 
@@ -28,6 +58,7 @@ function KeyIndex.new(shared_dict, prefix, remove_expired_keys_interval)
   self.delete_count = prefix .. "delete_count"
   self.key_count = prefix .. "key_count"
   self.generation = prefix .. "generation"
+  self.flush_marker = prefix .. "flush_expired"
   self.last = 0
   self.deleted = 0
   self.keys = {}
@@ -35,7 +66,12 @@ function KeyIndex.new(shared_dict, prefix, remove_expired_keys_interval)
   self.aliases = {}
   self.expire_keys = {}
 
-  ngx.timer.every(remove_expired_keys_interval or 600, remove_expired_keys, self)
+  local interval = remove_expired_keys_interval
+  if not interval or interval <= 0 then
+    interval = DEFAULT_REMOVE_EXPIRED_KEYS_INTERVAL
+  end
+  self.remove_expired_keys_interval = interval
+  start_remove_expired_keys_timer(self, interval)
   return self
 end
 
@@ -46,14 +82,61 @@ local function slot_gone(self, i, err)
   return err == "not found" and self.dict:get_stale(self.key_prefix .. i) == nil
 end
 
+-- Drops the worker-local references to the slots that are gone.
+local function remove_expired_slots(self)
+  -- Snapshot due slots before yielding: add() may mutate expire_keys while
+  -- this coroutine is suspended. Future deadlines need no dictionary lookup.
+  local now = ngx.now()
+  local due = {}
+  for i, deadline in pairs(self.expire_keys) do
+    if deadline <= now then
+      due[#due + 1] = i
+    end
+  end
+
+  for j, i in ipairs(due) do
+    if j > 1 and (j - 1) % REMOVE_EXPIRED_KEYS_YIELD_EVERY == 0 then
+      ngx.sleep(0)
+      -- do not hold up a reload: what is left is gone with the worker anyway
+      if ngx.worker.exiting() then
+        return
+      end
+      now = ngx.now()
+    end
+
+    -- re-read: while this round yielded the slot may have been renewed, or
+    -- dropped by add()
+    local deadline = self.expire_keys[i]
+    if deadline and deadline <= now then
+      -- Keep expired nodes until reclamation: expire() can renew them in place.
+      local ttl, err = self.dict:ttl(self.key_prefix .. i)
+      if ttl and ttl > 0 then
+        self.expire_keys[i] = now + ttl
+      elseif slot_gone(self, i, err) then
+        self:forget_slot(i)
+      end
+    end
+  end
+end
+
 -- check and remove expired keys
 function KeyIndex:remove_expired_keys()
-  for i, _ in pairs(self.expire_keys) do
-    -- Keep expired nodes until reclamation: expire() can renew them in place.
-    local _, err = self.dict:ttl(self.key_prefix .. i)
-    if slot_gone(self, i, err) then
-      self:forget_slot(i)
-    end
+  -- A round yields, so the timer can fire again before the previous run
+  -- finished (ngx.timer.every does not wait for it). Two rounds would only ask
+  -- the dict about the same slots twice, so the second one steps aside.
+  if self.removing then
+    return
+  end
+
+  self.removing = true
+  -- pcall so that a throwing ngx.sleep -- a caller reaching this from a phase
+  -- that does not allow yielding -- cannot leave the guard set and stop the
+  -- local scan for the life of the worker.
+  local ok, err = pcall(remove_expired_slots, self)
+  self.removing = false
+
+  if not ok then
+    ngx.log(ngx.ERR, "failed to remove expired keys: ", tostring(err))
   end
 
   -- The loop above only drops worker-local references. The expired shared-dict
@@ -66,7 +149,21 @@ function KeyIndex:remove_expired_keys()
   -- up sitting there. Without this call the dict grows without bound under
   -- label churn: index slots are never reused, so free_space steps down on
   -- every new series and never recovers (apache/apisix#13658).
-  self.dict:flush_expired()
+  --
+  -- flush_expired() walks the whole dict under its lock, and one call reclaims
+  -- for every worker, so only the first process to finish its local cleanup in
+  -- an interval runs it. The marker expires after the interval and is never
+  -- deleted, so the next interval is open to whichever worker comes first.
+  if ngx.worker.exiting() then
+    return
+  end
+  -- safe_add: add() would force-evict a live entry to store the marker. A dict
+  -- too full to hold it is the one that most needs reclaiming, so flush anyway.
+  local added, add_err = self.dict:safe_add(self.flush_marker, true,
+                                            self.remove_expired_keys_interval)
+  if added or add_err == "no memory" then
+    self.dict:flush_expired()
+  end
 end
 
 -- Tokens for the generation key (see advance_key_count()), unique per
@@ -128,13 +225,13 @@ function KeyIndex:sync_range(first, last)
         local ttl, err = self.dict:ttl(self.key_prefix .. i)
         if ttl then
           if ttl ~= 0 then
-            self.expire_keys[i] = true
+            self.expire_keys[i] = ngx.now() + ttl
           end
         elseif slot_gone(self, i, err) then
           self:forget_slot(i)
         else
           -- Kept at the ttl() sentinel; check it again on the next round.
-          self.expire_keys[i] = true
+          self.expire_keys[i] = ngx.now()
         end
       end
     elseif self.keys[i] then
@@ -242,6 +339,10 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime)
     keys = { key_or_keys }
   end
 
+  -- when the slot's expiry is renewed or set below, remember when it lands, so
+  -- that remove_expired_keys() does not have to ask the dict about it
+  local now = exptime and exptime > 0 and ngx.now()
+
   for _, key in pairs(keys) do
     local retried = false
     local repairs = 0
@@ -255,6 +356,9 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime)
     if mine and exptime and self.dict:get(self.key_prefix .. mine) == key then
       local ok, err = self.dict:expire(self.key_prefix .. mine, exptime)
       if ok then
+        if now then
+          self.expire_keys[mine] = now + exptime
+        end
         goto renewed
       end
       if err ~= "not found" then
@@ -271,6 +375,9 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime)
         local expired = false
         if exptime then
           local ok, err = self.dict:expire(self.key_prefix .. self.index[key], exptime)
+          if ok and now then
+            self.expire_keys[self.index[key]] = now + exptime
+          end
           if not ok then
             if err == "not found" then
               -- The node is gone, so this number is no longer ours to renew.
@@ -309,8 +416,8 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime)
       if ok then
         local _, _, forcible2 = advance_key_count(self)
         self:remember_slot(N, key)
-        if exptime and exptime > 0 then
-          self.expire_keys[N] = true
+        if now then
+          self.expire_keys[N] = now + exptime
         end
         if forcible or forcible2 or repair_forcible then
           return (err_msg_lru_eviction .. "; key index: add key: idx=" ..
diff --git a/prometheus_test.lua b/prometheus_test.lua
index 22d2f91..08a35e7 100644
--- a/prometheus_test.lua
+++ b/prometheus_test.lua
@@ -32,6 +32,10 @@ function SimpleDict:add(k, v, exptime)
   self:set(k, v, exptime)
   return true, nil, forcible  -- ok, err, forcible
 end
+function SimpleDict:safe_add(k, v, exptime)
+  local ok, err = self:add(k, v, exptime)
+  return ok, err
+end
 function SimpleDict:incr(k, v, init)
   local forcible = false
   if k == "willnotfitk" or v == "willnotfitv" or
@@ -137,7 +141,13 @@ function Nginx.print(printed)
 end
 Nginx.worker = {}
 function Nginx.worker.id()
-  return 'testworker'
+  return 0
+end
+function Nginx.worker.count()
+  return 1
+end
+function Nginx.worker.exiting()
+  return false
 end
 function Nginx.sleep() end
 function Nginx.now()
@@ -146,6 +156,7 @@ function Nginx.now()
 end
 Nginx.timer = {}
 function Nginx.timer.every(_, _, _) end
+function Nginx.timer.at(_, _, _) end
 function Nginx.get_phase()
   return 'init_worker'
 end
@@ -231,6 +242,7 @@ function TestPrometheus:testInitOptions()
 
   luaunit.assertEquals(ngx.logs, nil)
 end
+
 function TestPrometheus:testInitWorker()
   self.dict = setmetatable({}, SimpleDict)
   ngx.shared.metrics = self.dict
@@ -793,6 +805,9 @@ function TestKeyIndex:setUp()
 end
 function TestKeyIndex.tearDown()
   ngx.logs = nil
+  -- the cleanup tests stub ngx.sleep; drop the stub here too, so a failing
+  -- assertion before their own restore cannot leak it into the next test
+  ngx.sleep = nil
 end
 function TestKeyIndex.testInit()
   luaunit.assertEquals(ngx.logs, nil)
@@ -933,7 +948,7 @@ function TestKeyIndex:testReclaimedCanonicalFallsBackToAlias()
   self.key_index:sync()
   luaunit.assertEquals(self.key_index:list(), {"target"})
   self.dict:delete("_prefix_key_2")
-  self.key_index.expire_keys[2] = true
+  self.key_index.expire_keys[2] = 0
   self.key_index:remove_expired_keys()
   luaunit.assertEquals(self.key_index:list(), {"target"})
   luaunit.assertEquals(self.key_index.index.target, 1)
@@ -1168,6 +1183,344 @@ function TestKeyIndex:testSyncTracksSlotsImportedAtTheTtlSentinel()
   luaunit.assertNil(reader.expire_keys[i])
 end
 
+function TestKeyIndex:testCleanupTimersAreStaggeredByWorker()
+  local id, count = ngx.worker.id, ngx.worker.count
+  local at, every = ngx.timer.at, ngx.timer.every
+  local scheduled
+  ngx.worker.count = function() return 4 end
+  ngx.timer.at = function(delay, callback, context)
+    scheduled = {kind="at", delay=delay, callback=callback, context=context}
+  end
+  ngx.timer.every = function(delay, callback, context)
+    scheduled = {kind="every", delay=delay, callback=callback, context=context}
+  end
+
+  local cases = {{id=0}, {id=2}, {}}
+  local results = {}
+  for i, case in ipairs(cases) do
+    ngx.worker.id = function() return case.id end
+    local key_index = require('prometheus_keys').new(self.dict, "_timer_", 60)
+    local calls = 0
+    key_index.remove_expired_keys = function() calls = calls + 1 end
+    local first = scheduled
+    if first.kind == "at" then
+      first.callback(true, first.context)
+      local after_shutdown = calls
+      first.callback(false, first.context)
+      results[i] = {first.kind, first.delay, scheduled.kind, scheduled.delay,
+                    after_shutdown, calls}
+    else
+      results[i] = {first.kind, first.delay}
+    end
+  end
+  ngx.worker.id, ngx.worker.count = id, count
+  ngx.timer.at, ngx.timer.every = at, every
+
+  luaunit.assertEquals(results, {{"every", 60},
+    {"at", 30, "every", 60, 0, 1}, {"every", 60}})
+end
+
+function TestKeyIndex:testCleanupTimerSkipsPrematureCallbacks()
+  local every = ngx.timer.every
+  local callback, context
+  ngx.timer.every = function(_, fn, key_index)
+    callback, context = fn, key_index
+  end
+  local key_index = require('prometheus_keys').new(self.dict, "_timer_", 60)
+  ngx.timer.every = every
+
+  local calls = 0
+  key_index.remove_expired_keys = function() calls = calls + 1 end
+  callback(true, context)
+  luaunit.assertEquals(calls, 0)
+  callback(false, context)
+  luaunit.assertEquals(calls, 1)
+end
+
+function TestKeyIndex:testRemoveExpiredKeysKeepsPermanentSlots()
+  luaunit.assertNil(self.key_index:add("permanent", "eviction_err", 600))
+  local i = self.key_index.index["permanent"]
+  local slot = self.key_index.key_prefix .. i
+  self.dict:set(slot, "permanent")
+  self.key_index.expire_keys[i] = 0
+
+  self.key_index:remove_expired_keys()
+  self.key_index:remove_expired_keys()
+
+  luaunit.assertEquals(self.key_index.index["permanent"], i)
+  luaunit.assertEquals(self.dict:get(slot), "permanent")
+end
+
+-- ttl() also reads 0 at a slot's exact expiry millisecond, so a zero keeps the
+-- slot tracked until its node is reclaimed.
+function TestKeyIndex:testRemoveExpiredKeysRechecksZeroTTLSlots()
+  luaunit.assertNil(self.key_index:add("boundary", "eviction_err", 600))
+  local i = self.key_index.index["boundary"]
+  local slot = self.key_index.key_prefix .. i
+  self.key_index.expire_keys[i] = 0
+
+  self.dict.ttl = function(_, key)
+    luaunit.assertEquals(key, slot)
+    return 0
+  end
+  self.key_index:remove_expired_keys()
+  self.dict.ttl = nil
+  luaunit.assertNotNil(self.key_index.expire_keys[i])
+
+  self.dict:delete(slot)
+  self.key_index:remove_expired_keys()
+  luaunit.assertNil(self.key_index.keys[i])
+  luaunit.assertNil(self.key_index.index["boundary"])
+end
+
+-- Metrics registered without an exptime are permanent; the workers that import
+-- their slots must not check them on every round.
+function TestKeyIndex:testSyncDoesNotTrackPermanentSlots()
+  luaunit.assertNil(self.key_index:add("permanent", "eviction_err"))
+  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
+  reader:sync()
+  local i = reader.index["permanent"]
+  luaunit.assertNotNil(i)
+  luaunit.assertNil(reader.expire_keys[i])
+end
+
+-- Future deadlines require no shared-dictionary TTL checks.
+function TestKeyIndex:testRemoveExpiredKeysSkipsSlotsThatCannotHaveLapsed()
+  local keys = {}
+  for i = 1, 100 do
+    keys[i] = "live_" .. i
+  end
+  luaunit.assertEquals(self.key_index:add(keys, "eviction_err", 600), nil)
+
+  local checked = 0
+  local ttl = SimpleDict.ttl
+  self.dict.ttl = function(dict, k)
+    checked = checked + 1
+    return ttl(dict, k)
+  end
+
+  self.key_index:remove_expired_keys()
+  luaunit.assertEquals(checked, 0)
+  luaunit.assertEquals(self.key_index.index["live_1"], 1)
+
+  self.dict.ttl = nil
+  luaunit.assertEquals(ngx.logs, nil)
+end
+
+-- A renewal on the fast path moves the deadline, and tracks a slot that was
+-- not tracked before.
+function TestKeyIndex:testRenewalRecordsTheNewDeadline()
+  luaunit.assertNil(self.key_index:add("renewed", "eviction_err", 1))
+  local i = self.key_index.index["renewed"]
+  self.key_index.expire_keys[i] = nil
+
+  local sync = self.key_index.sync
+  self.key_index.sync = function() error("renewal must not sync") end
+  luaunit.assertNil(self.key_index:add("renewed", "eviction_err", 600))
+  self.key_index.sync = sync
+
+  luaunit.assertTrue(self.key_index.expire_keys[i] >= ngx.now() + 600)
+end
+
+-- The slots whose deadline has passed do have to be asked about, and there can
+-- be as many of them as the index is large. One round asks about all of them,
+-- so the local index keeps up with the expiry rate, but yields every
+-- REMOVE_EXPIRED_KEYS_YIELD_EVERY of them so the worker keeps serving.
+local function lapse_all_slots(self, prefix, count)
+  local keys = {}
+  for i = 1, count do
+    keys[i] = prefix .. i
+  end
+  luaunit.assertEquals(self.key_index:add(keys, "eviction_err", 600), nil)
+
+  -- what a round running after the exptime sees, without waiting for it: every
+  -- recorded deadline has passed, so every slot has to be asked about
+  for i in pairs(self.key_index.expire_keys) do
+    self.key_index.expire_keys[i] = 0
+  end
+
+  local counter = {checked = 0}
+  local ttl = SimpleDict.ttl
+  self.dict.ttl = function(dict, k)
+    counter.checked = counter.checked + 1
+    return ttl(dict, k)
+  end
+  return counter
+end
+
+function TestKeyIndex:testRemoveExpiredKeysYieldsWhileScanning()
+  local every = self.key_index.REMOVE_EXPIRED_KEYS_YIELD_EVERY
+  local counter = lapse_all_slots(self, "yield_", 2 * every + 10)
+
+  local yields, checked_at_yield = 0, {}
+  ngx.sleep = function()
+    yields = yields + 1
+    checked_at_yield[yields] = counter.checked
+  end
+  self.key_index:remove_expired_keys()
+  ngx.sleep = nil
+
+  -- every lapsed slot is asked about in this one round, in steps of `every`
+  luaunit.assertEquals(counter.checked, 2 * every + 10)
+  luaunit.assertEquals(checked_at_yield, {every, 2 * every})
+
+  -- the slots are all still live, so they are kept, and their deadlines are
+  -- refreshed rather than re-checked on every round from now on
+  luaunit.assertEquals(self.key_index.index["yield_1"], 1)
+  luaunit.assertEquals(self.key_index.index["yield_" .. (2 * every + 10)],
+                       2 * every + 10)
+  counter.checked = 0
+  self.key_index:remove_expired_keys()
+  luaunit.assertEquals(counter.checked, 0)
+
+  self.dict.ttl = nil
+  luaunit.assertEquals(ngx.logs, nil)
+end
+
+-- A round yields, so the timer can fire again before it finishes. The second
+-- round must step aside instead of asking the dict about the same slots again.
+function TestKeyIndex:testRemoveExpiredKeysDoesNotOverlap()
+  local every = self.key_index.REMOVE_EXPIRED_KEYS_YIELD_EVERY
+  local counter = lapse_all_slots(self, "overlap_", every + 10)
+
+  local checked_by_reentry
+  ngx.sleep = function()
+    if checked_by_reentry == nil then
+      local before = counter.checked
+      self.key_index:remove_expired_keys()
+      checked_by_reentry = counter.checked - before
+    end
+  end
+  self.key_index:remove_expired_keys()
+  ngx.sleep = nil
+
+  luaunit.assertEquals(checked_by_reentry, 0)
+  luaunit.assertEquals(counter.checked, every + 10)
+
+  self.dict.ttl = nil
+  luaunit.assertEquals(ngx.logs, nil)
+end
+
+-- A worker that is shutting down stops scanning at the next yield, so it does
+-- not hold up a reload.
+function TestKeyIndex:testRemoveExpiredKeysStopsWhenExiting()
+  local every = self.key_index.REMOVE_EXPIRED_KEYS_YIELD_EVERY
+  local counter = lapse_all_slots(self, "exiting_", every + 10)
+
+  ngx.worker.exiting = function() return true end
+  self.key_index:remove_expired_keys()
+  ngx.worker.exiting = function() return false end
+
+  luaunit.assertEquals(counter.checked, every)
+
+  self.dict.ttl = nil
+  luaunit.assertEquals(ngx.logs, nil)
+end
+
+-- A throwing ngx.sleep -- a caller reaching remove_expired_keys() from a phase
+-- that does not allow yielding -- must not leave the guard set, or the worker
+-- would stop scanning for good.
+function TestKeyIndex:testRemoveExpiredKeysReleasesTheGuardAfterAnError()
+  local every = self.key_index.REMOVE_EXPIRED_KEYS_YIELD_EVERY
+  local counter = lapse_all_slots(self, "error_", every + 10)
+
+  ngx.sleep = function() error("API disabled in the current context") end
+  self.key_index:remove_expired_keys()
+  ngx.sleep = nil
+
+  luaunit.assertEquals(#ngx.logs, 1)
+  luaunit.assertStrContains(ngx.logs[1], "failed to remove expired keys")
+  ngx.logs = nil
+  luaunit.assertEquals(counter.checked, every)
+
+  -- the guard is released, so the next round asks about the rest
+  counter.checked = 0
+  self.key_index:remove_expired_keys()
+  luaunit.assertEquals(counter.checked, 10)
+
+  self.dict.ttl = nil
+end
+
+-- A non-positive interval is not a way to turn the cleanup off: it falls back
+-- to the default, as a missing one does.
+function TestKeyIndex:testNonPositiveIntervalFallsBackToDefault()
+  local at, every = ngx.timer.at, ngx.timer.every
+  -- init() also starts the counter sync timer; count only cleanup timers.
+  local intervals = {}
+  local function schedule(delay, _, context)
+    if type(context) == "table" and context.expire_keys then
+      intervals[#intervals + 1] = delay
+    end
+  end
+  ngx.timer.at, ngx.timer.every = schedule, schedule
+  require('prometheus_keys').new(self.dict, "_zero_", 0)
+  require('prometheus_keys').new(self.dict, "_negative_", -1)
+  local p = require('prometheus').init("metrics", {remove_expired_keys_interval=0})
+  ngx.timer.at, ngx.timer.every = at, every
+  luaunit.assertEquals(intervals, {600, 600, 3600})
+  luaunit.assertEquals(p.remove_expired_keys_interval, 3600)
+  luaunit.assertStrContains(ngx.logs[#ngx.logs],
+                            "remove_expired_keys_interval must be positive")
+end
+
+local function count_flushes(dict)
+  local flushes = {n = 0}
+  dict.flush_expired = function(d)
+    flushes.n = flushes.n + 1
+    return SimpleDict.flush_expired(d)
+  end
+  return flushes
+end
+
+-- flush_expired() reclaims for every worker, so one call per interval is
+-- enough: the first worker to finish its local cleanup runs it, the others
+-- skip it until the interval has passed.
+function TestKeyIndex:testOneFlushPerInterval()
+  local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1)
+  local flushes = count_flushes(self.dict)
+
+  self.key_index:remove_expired_keys()
+  worker2:remove_expired_keys()
+  self.key_index:remove_expired_keys()
+  luaunit.assertEquals(flushes.n, 1)
+
+  sleep(2)
+  worker2:remove_expired_keys()
+  self.key_index:remove_expired_keys()
+  luaunit.assertEquals(flushes.n, 2)
+  self.dict.flush_expired = nil
+end
+
+-- Storing the marker must not evict a live entry, and a dict too full to hold
+-- it is the one that most needs reclaiming.
+function TestKeyIndex:testFlushWhenTheMarkerDoesNotFit()
+  local flushes = count_flushes(self.dict)
+  self.dict.safe_add = function() return nil, "no memory" end
+
+  self.key_index:remove_expired_keys()
+  self.key_index:remove_expired_keys()
+  luaunit.assertEquals(flushes.n, 2)
+  self.dict.safe_add = nil
+  self.dict.flush_expired = nil
+end
+
+-- A worker on its way out must not take the interval: it could exit before
+-- reclaiming, and no other worker would reclaim until the interval passed.
+function TestKeyIndex:testExitingWorkerLeavesTheFlushToOthers()
+  local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1)
+  local flushes = count_flushes(self.dict)
+  local exiting = ngx.worker.exiting
+  ngx.worker.exiting = function() return true end
+  self.key_index:remove_expired_keys()
+  ngx.worker.exiting = exiting
+  luaunit.assertEquals(flushes.n, 0)
+  luaunit.assertNil(self.dict:get("_prefix_flush_expired"))
+
+  worker2:remove_expired_keys()
+  luaunit.assertEquals(flushes.n, 1)
+  self.dict.flush_expired = nil
+end
+
 -- The index slots of a churning metric must not accumulate: each round expires
 -- the previous slot, and the reclaim must keep the dict bounded rather than
 -- letting every new series step free space down for good.
@@ -1271,7 +1624,7 @@ end
 function TestKeyIndex:testReaderFindsSlotReissuedAfterKeyCountEviction()
   local reader = evict_key_count_and_slot(self, 5)
   -- the reader's own round finds slot 5 gone and forgets it
-  reader.expire_keys[5] = true
+  reader.expire_keys[5] = 0
   reader:remove_expired_keys()
   luaunit.assertNil(reader.index["key5"])
 
@@ -1282,7 +1635,7 @@ end
 
 function TestKeyIndex:testReaderFindsSlotReissuedAfterKeyCountClimbsBack()
   local reader = evict_key_count_and_slot(self, 5)
-  reader.expire_keys[5] = true
+  reader.expire_keys[5] = 0
   reader:remove_expired_keys()
 
   luaunit.assertEquals(self.key_index:add("key5", "eviction_err", 60), nil)
@@ -1311,7 +1664,7 @@ end
 
 function TestKeyIndex:testReaderFindsLowestSlotReissuedWithoutRepair()
   local reader = evict_key_count_and_slot(self, 1)
-  reader.expire_keys[1] = true
+  reader.expire_keys[1] = 0
   reader:remove_expired_keys()
 
   -- slot 1 is free, so this takes it without any key_count repair, and
@@ -1323,7 +1676,7 @@ end
 
 function TestKeyIndex:testReaderFindsSlotReissuedByFreshWorker()
   local reader = evict_key_count_and_slot(self, 1)
-  reader.expire_keys[1] = true
+  reader.expire_keys[1] = 0
   reader:remove_expired_keys()
 
   -- a respawned worker: its self.last is 0, so it cannot see key_count going
@@ -1350,7 +1703,7 @@ local function evict_key_count_and_all_slots(self)
   self.dict:delete("_prefix_key_count")
   for i = 1, 100 do
     self.dict:delete("_prefix_key_" .. i)
-    reader.expire_keys[i] = true
+    reader.expire_keys[i] = 0
   end
   reader:remove_expired_keys()
   luaunit.assertNil(reader.index["key5"])
@@ -1742,6 +2095,8 @@ function TestPrometheus:testKeyTimeout()
 
   sleep(1)
   self.p.key_index:sync()
+  -- the round above reclaimed for this interval; let the interval pass
+  self.dict:delete("__ngx_prom__flush_expired")
   -- the first round keeps the expired node and flushes it, the second one
   -- finds it reclaimed and drops the reference
   self.p.key_index:remove_expired_keys()