| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341 |
- const content = `--[[
- Completely obliterates a queue and all of its contents
- This command completely destroys a queue including all of its jobs, current or past
- leaving no trace of its existence. Since this script needs to iterate to find all the job
- keys, consider that this call may be slow for very large queues.
- The queue needs to be "paused" or it will return an error
- If the queue has currently active jobs then the script by default will return error,
- however this behaviour can be overrided using the 'force' option.
- Input:
- KEYS[1] meta
- KEYS[2] base
- ARGV[1] count
- ARGV[2] force
- ]]
- local maxCount = tonumber(ARGV[1])
- local baseKey = KEYS[2]
- local rcall = redis.call
- -- Includes
- --[[
- Functions to remove jobs.
- ]]
- -- Includes
- --[[
- Function to remove job.
- ]]
- -- Includes
- --[[
- Function to remove deduplication key if needed
- when a job is being removed.
- ]]
- local function removeDeduplicationKeyIfNeededOnRemoval(prefixKey,
- jobId, deduplicationId)
- if deduplicationId then
- local deduplicationKey = prefixKey .. "de:" .. deduplicationId
- local currentJobId = rcall('GET', deduplicationKey)
- if currentJobId and currentJobId == jobId then
- rcall("DEL", deduplicationKey)
- -- Also clean up any pending dedup-next data for this dedup ID
- rcall("DEL", prefixKey .. "dn:" .. deduplicationId)
- return 1
- end
- end
- end
- --[[
- Function to remove job keys.
- ]]
- local function removeJobKeys(jobKey)
- return rcall("DEL", jobKey, jobKey .. ':logs', jobKey .. ':dependencies',
- jobKey .. ':processed', jobKey .. ':failed', jobKey .. ':unsuccessful')
- end
- --[[
- Check if this job has a parent. If so we will just remove it from
- the parent child list, but if it is the last child we should move the parent to "wait/paused"
- which requires code from "moveToFinished"
- ]]
- -- Includes
- --[[
- Function to add job in target list and add marker if needed.
- ]]
- -- Includes
- --[[
- Add marker if needed when a job is available.
- ]]
- local function addBaseMarkerIfNeeded(markerKey, isPausedOrMaxed)
- if not isPausedOrMaxed then
- rcall("ZADD", markerKey, 0, "0")
- end
- end
- local function addJobInTargetList(targetKey, markerKey, pushCmd, isPausedOrMaxed, jobId)
- rcall(pushCmd, targetKey, jobId)
- addBaseMarkerIfNeeded(markerKey, isPausedOrMaxed)
- end
- --[[
- Functions to destructure job key.
- Just a bit of warning, these functions may be a bit slow and affect performance significantly.
- ]]
- local getJobIdFromKey = function (jobKey)
- return string.match(jobKey, ".*:(.*)")
- end
- local getJobKeyPrefix = function (jobKey, jobId)
- return string.sub(jobKey, 0, #jobKey - #jobId)
- end
- --[[
- Function to check for the meta.paused key to decide if we are paused or not
- (since an empty list and !EXISTS are not really the same).
- ]]
- local function getTargetQueueList(queueMetaKey, activeKey, waitKey, pausedKey)
- local queueAttributes = rcall("HMGET", queueMetaKey, "paused", "concurrency", "max", "duration")
- if queueAttributes[1] then
- return pausedKey, true, queueAttributes[3], queueAttributes[4]
- else
- if queueAttributes[2] then
- local activeCount = rcall("LLEN", activeKey)
- if activeCount >= tonumber(queueAttributes[2]) then
- return waitKey, true, queueAttributes[3], queueAttributes[4]
- else
- return waitKey, false, queueAttributes[3], queueAttributes[4]
- end
- end
- end
- return waitKey, false, queueAttributes[3], queueAttributes[4]
- end
- local function _moveParentToWait(parentPrefix, parentId, emitEvent)
- local parentTarget, isPausedOrMaxed = getTargetQueueList(parentPrefix .. "meta", parentPrefix .. "active",
- parentPrefix .. "wait", parentPrefix .. "paused")
- addJobInTargetList(parentTarget, parentPrefix .. "marker", "RPUSH", isPausedOrMaxed, parentId)
- if emitEvent then
- local parentEventStream = parentPrefix .. "events"
- rcall("XADD", parentEventStream, "*", "event", "waiting", "jobId", parentId, "prev", "waiting-children")
- end
- end
- local function removeParentDependencyKey(jobKey, hard, parentKey, baseKey, debounceId)
- if parentKey then
- local parentDependenciesKey = parentKey .. ":dependencies"
- local result = rcall("SREM", parentDependenciesKey, jobKey)
- if result > 0 then
- local pendingDependencies = rcall("SCARD", parentDependenciesKey)
- if pendingDependencies == 0 then
- local parentId = getJobIdFromKey(parentKey)
- local parentPrefix = getJobKeyPrefix(parentKey, parentId)
- local numRemovedElements = rcall("ZREM", parentPrefix .. "waiting-children", parentId)
- if numRemovedElements == 1 then
- if hard then -- remove parent in same queue
- if parentPrefix == baseKey then
- removeParentDependencyKey(parentKey, hard, nil, baseKey, nil)
- removeJobKeys(parentKey)
- if debounceId then
- rcall("DEL", parentPrefix .. "de:" .. debounceId)
- end
- else
- _moveParentToWait(parentPrefix, parentId)
- end
- else
- _moveParentToWait(parentPrefix, parentId, true)
- end
- end
- end
- return true
- end
- else
- local parentAttributes = rcall("HMGET", jobKey, "parentKey", "deid")
- local missedParentKey = parentAttributes[1]
- if( (type(missedParentKey) == "string") and missedParentKey ~= ""
- and (rcall("EXISTS", missedParentKey) == 1)) then
- local parentDependenciesKey = missedParentKey .. ":dependencies"
- local result = rcall("SREM", parentDependenciesKey, jobKey)
- if result > 0 then
- local pendingDependencies = rcall("SCARD", parentDependenciesKey)
- if pendingDependencies == 0 then
- local parentId = getJobIdFromKey(missedParentKey)
- local parentPrefix = getJobKeyPrefix(missedParentKey, parentId)
- local numRemovedElements = rcall("ZREM", parentPrefix .. "waiting-children", parentId)
- if numRemovedElements == 1 then
- if hard then
- if parentPrefix == baseKey then
- removeParentDependencyKey(missedParentKey, hard, nil, baseKey, nil)
- removeJobKeys(missedParentKey)
- if parentAttributes[2] then
- rcall("DEL", parentPrefix .. "de:" .. parentAttributes[2])
- end
- else
- _moveParentToWait(parentPrefix, parentId)
- end
- else
- _moveParentToWait(parentPrefix, parentId, true)
- end
- end
- end
- return true
- end
- end
- end
- return false
- end
- local function removeJob(jobId, hard, baseKey, shouldRemoveDeduplicationKey)
- local jobKey = baseKey .. jobId
- removeParentDependencyKey(jobKey, hard, nil, baseKey)
- if shouldRemoveDeduplicationKey then
- local deduplicationId = rcall("HGET", jobKey, "deid")
- removeDeduplicationKeyIfNeededOnRemoval(baseKey, jobId, deduplicationId)
- end
- removeJobKeys(jobKey)
- end
- local function removeJobs(keys, hard, baseKey, max)
- for i, key in ipairs(keys) do
- removeJob(key, hard, baseKey, true --[[remove debounce key]])
- end
- return max - #keys
- end
- --[[
- Functions to remove jobs.
- ]]
- -- Includes
- --[[
- Function to filter out jobs to ignore from a table.
- ]]
- local function filterOutJobsToIgnore(jobs, jobsToIgnore)
- local filteredJobs = {}
- for i = 1, #jobs do
- if not jobsToIgnore[jobs[i]] then
- table.insert(filteredJobs, jobs[i])
- end
- end
- return filteredJobs
- end
- local function getListItems(keyName, max)
- return rcall('LRANGE', keyName, 0, max - 1)
- end
- local function removeListJobs(keyName, hard, baseKey, max, jobsToIgnore)
- local jobs = getListItems(keyName, max)
- if jobsToIgnore then
- jobs = filterOutJobsToIgnore(jobs, jobsToIgnore)
- end
- local count = removeJobs(jobs, hard, baseKey, max)
- rcall("LTRIM", keyName, #jobs, -1)
- return count
- end
- -- Includes
- --[[
- Function to loop in batches.
- Just a bit of warning, some commands as ZREM
- could receive a maximum of 7000 parameters per call.
- ]]
- local function batches(n, batchSize)
- local i = 0
- return function()
- local from = i * batchSize + 1
- i = i + 1
- if (from <= n) then
- local to = math.min(from + batchSize - 1, n)
- return from, to
- end
- end
- end
- --[[
- Function to get ZSet items.
- ]]
- local function getZSetItems(keyName, max)
- return rcall('ZRANGE', keyName, 0, max - 1)
- end
- local function removeZSetJobs(keyName, hard, baseKey, max, jobsToIgnore)
- local jobs = getZSetItems(keyName, max)
- if jobsToIgnore then
- jobs = filterOutJobsToIgnore(jobs, jobsToIgnore)
- end
- local count = removeJobs(jobs, hard, baseKey, max)
- if(#jobs > 0) then
- for from, to in batches(#jobs, 7000) do
- rcall("ZREM", keyName, unpack(jobs, from, to))
- end
- end
- return count
- end
- local function removeLockKeys(keys)
- for i, key in ipairs(keys) do
- rcall("DEL", baseKey .. key .. ':lock')
- end
- end
- -- 1) Check if paused, if not return with error.
- if rcall("HEXISTS", KEYS[1], "paused") ~= 1 then
- return -1 -- Error, NotPaused
- end
- -- 2) Check if there are active jobs, if there are and not "force" return error.
- local activeKey = baseKey .. 'active'
- local activeJobs = getListItems(activeKey, maxCount)
- if (#activeJobs > 0) then
- if(ARGV[2] == "") then
- return -2 -- Error, ExistActiveJobs
- end
- end
- removeLockKeys(activeJobs)
- maxCount = removeJobs(activeJobs, true, baseKey, maxCount)
- rcall("LTRIM", activeKey, #activeJobs, -1)
- if(maxCount <= 0) then
- return 1
- end
- local delayedKey = baseKey .. 'delayed'
- maxCount = removeZSetJobs(delayedKey, true, baseKey, maxCount)
- if(maxCount <= 0) then
- return 1
- end
- local repeatKey = baseKey .. 'repeat'
- local repeatJobsIds = getZSetItems(repeatKey, maxCount)
- for i, key in ipairs(repeatJobsIds) do
- local jobKey = repeatKey .. ":" .. key
- rcall("DEL", jobKey)
- end
- if(#repeatJobsIds > 0) then
- for from, to in batches(#repeatJobsIds, 7000) do
- rcall("ZREM", repeatKey, unpack(repeatJobsIds, from, to))
- end
- end
- maxCount = maxCount - #repeatJobsIds
- if(maxCount <= 0) then
- return 1
- end
- local completedKey = baseKey .. 'completed'
- maxCount = removeZSetJobs(completedKey, true, baseKey, maxCount)
- if(maxCount <= 0) then
- return 1
- end
- local waitKey = baseKey .. 'paused'
- maxCount = removeListJobs(waitKey, true, baseKey, maxCount)
- if(maxCount <= 0) then
- return 1
- end
- local prioritizedKey = baseKey .. 'prioritized'
- maxCount = removeZSetJobs(prioritizedKey, true, baseKey, maxCount)
- if(maxCount <= 0) then
- return 1
- end
- local failedKey = baseKey .. 'failed'
- maxCount = removeZSetJobs(failedKey, true, baseKey, maxCount)
- if(maxCount <= 0) then
- return 1
- end
- if(maxCount > 0) then
- rcall("DEL",
- baseKey .. 'events',
- baseKey .. 'delay',
- baseKey .. 'stalled-check',
- baseKey .. 'stalled',
- baseKey .. 'id',
- baseKey .. 'pc',
- baseKey .. 'marker',
- baseKey .. 'meta',
- baseKey .. 'metrics:completed',
- baseKey .. 'metrics:completed:data',
- baseKey .. 'metrics:failed',
- baseKey .. 'metrics:failed:data')
- return 0
- else
- return 1
- end
- `;
- export const obliterate = {
- name: 'obliterate',
- content,
- keys: 2,
- };
- //# sourceMappingURL=obliterate-2.js.map
|