                                   "purge_reports"))
                               "other"])]
             (interspaced gc-interval-millis
-                         #(try
-                            (clean-up write-db database what)
-                            (catch Exception ex
-                              (log/error ex)))
+                         (fn []
+                           (.lock clean-lock)
+                           (try
+                             (try
+                               (clean-up write-db clean-lock database what)
+                               (catch Exception ex
+                                 (log/error ex)))
+                             (finally
+                               (.unlock clean-lock))))
                          job-pool)))
         (when (pos? dlo-compression-interval-millis)
           (interspaced dlo-compression-interval-millis

        (assoc context
               :job-pool job-pool
               :broker broker
               :shared-globals globals)))))

(defprotocol PuppetDBServer
  (shared-globals [this])

  "Cleans up the resources specified by request, or everything if
  request is empty?."
  [db
   lock :- ReentrantLock
   {:keys [node-ttl node-purge-ttl report-ttl]} :- {:node-ttl Period
                                                    :node-purge-ttl Period
                                                    :report-ttl Period
                                                    s/Keyword s/Any}
   ;; Later, the values might be maps, i.e. {:limit 1000}
   request :- clean-request-schema]
  (when-not (.isHeldByCurrentThread lock)
    (throw (IllegalStateException. "cleanup lock is not already held")))
  (let [request (if (empty? request) clean-options (set request))]
    (when (request "expire_nodes")
      (auto-expire-nodes! node-ttl db))
