diff --git a/services/document-updater/app/coffee/DeleteQueueManager.coffee b/services/document-updater/app/coffee/DeleteQueueManager.coffee index 0a122369c3..cad65ae2f5 100644 --- a/services/document-updater/app/coffee/DeleteQueueManager.coffee +++ b/services/document-updater/app/coffee/DeleteQueueManager.coffee @@ -35,9 +35,11 @@ module.exports = DeleteQueueManager = logger.log "hit count limit on flushing old projects" return callback() cutoffTime = now - options.min_delete_age - RedisManager.getNextProjectToFlushAndDelete cutoffTime, (err, project_id, flushTimestamp) -> + RedisManager.getNextProjectToFlushAndDelete cutoffTime, (err, project_id, flushTimestamp, queueLength) -> return callback(err) if err? return callback() if !project_id? + logger.log {project_id, queueLength: queueLength}, "flushing queued project" + metrics.globalGauge "queued-flush-backlog", queueLength flushProjectIfNotModified project_id, flushTimestamp, (err, flushed) -> count++ if flushed flushNextProject() diff --git a/services/document-updater/app/coffee/RedisManager.coffee b/services/document-updater/app/coffee/RedisManager.coffee index 799d5a7fc1..f5530b99fd 100644 --- a/services/document-updater/app/coffee/RedisManager.coffee +++ b/services/document-updater/app/coffee/RedisManager.coffee @@ -298,18 +298,20 @@ module.exports = RedisManager = rclient.zadd keys.flushAndDeleteQueue(), Date.now(), project_id, callback getNextProjectToFlushAndDelete: (cutoffTime, callback = (error, key, timestamp)->) -> - # find the oldest queued flsus + # find the oldest queued flush rclient.zrangebyscore keys.flushAndDeleteQueue(), 0, cutoffTime, "WITHSCORES", "LIMIT", 0, 1, (err, reply) -> return callback(err) if err? return callback() if !reply?.length multi = rclient.multi() multi.zrange keys.flushAndDeleteQueue(), 0, 0, "WITHSCORES" multi.zremrangebyrank keys.flushAndDeleteQueue(), 0, 0 + multi.zcard keys.flushAndDeleteQueue() multi.exec (err, reply) -> return callback(err) if err? return callback() if !reply?.length [key, timestamp] = reply[0] - callback(null, key, timestamp) + queueLength = reply[2] + callback(null, key, timestamp, queueLength) _serializeRanges: (ranges, callback = (error, serializedRanges) ->) -> jsonRanges = JSON.stringify(ranges)