| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339 |
- const content = `--[[
- Remove a job from all the statuses it may be in as well as all its data.
- In order to be able to remove a job, it cannot be active.
- Input:
- KEYS[1] jobKey
- KEYS[2] meta key
- ARGV[1] prefix
- ARGV[2] jobId
- Events:
- 'removed' for every children removed
- ]]
- -- Includes
- --[[
- Remove a job from all the statuses it may be in as well as all its data,
- including its children. Active children can be ignored.
- Events:
- 'removed'
- ]]
- local rcall = redis.call
- -- Includes
- --[[
- 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 get max events value or set by default 10000.
- ]]
- local function getOrSetMaxEvents(metaKey)
- local maxEvents = rcall("HGET", metaKey, "opts.maxLenEvents")
- if not maxEvents then
- maxEvents = 10000
- rcall("HSET", metaKey, "opts.maxLenEvents", maxEvents)
- end
- return maxEvents
- end
- --[[
- Function to check if the job belongs to a job scheduler and
- current delayed job matches with jobId
- ]]
- local function isJobSchedulerJob(jobId, jobKey, jobSchedulersKey)
- local repeatJobKey = rcall("HGET", jobKey, "rjk")
- if repeatJobKey then
- local prevMillis = rcall("ZSCORE", jobSchedulersKey, repeatJobKey)
- if prevMillis then
- local currentDelayedJobId = "repeat:" .. repeatJobKey .. ":" .. prevMillis
- return jobId == currentDelayedJobId
- end
- end
- return false
- end
- --[[
- 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 from any state.
- returns:
- prev state
- ]]
- local function removeJobFromAnyState( prefix, jobId)
- -- We start with the ZSCORE checks, since they have O(1) complexity
- if rcall("ZSCORE", prefix .. "completed", jobId) then
- rcall("ZREM", prefix .. "completed", jobId)
- return "completed"
- elseif rcall("ZSCORE", prefix .. "waiting-children", jobId) then
- rcall("ZREM", prefix .. "waiting-children", jobId)
- return "waiting-children"
- elseif rcall("ZSCORE", prefix .. "delayed", jobId) then
- rcall("ZREM", prefix .. "delayed", jobId)
- return "delayed"
- elseif rcall("ZSCORE", prefix .. "failed", jobId) then
- rcall("ZREM", prefix .. "failed", jobId)
- return "failed"
- elseif rcall("ZSCORE", prefix .. "prioritized", jobId) then
- rcall("ZREM", prefix .. "prioritized", jobId)
- return "prioritized"
- -- We remove only 1 element from the list, since we assume they are not added multiple times
- elseif rcall("LREM", prefix .. "wait", 1, jobId) == 1 then
- return "wait"
- elseif rcall("LREM", prefix .. "paused", 1, jobId) == 1 then
- return "paused"
- elseif rcall("LREM", prefix .. "active", 1, jobId) == 1 then
- return "active"
- end
- return "unknown"
- 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
- --[[
- 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
- --[[
- Function to recursively check if there are no locks
- on the jobs to be removed.
- returns:
- boolean
- ]]
- local function isLocked( prefix, jobId, removeChildren)
- local jobKey = prefix .. jobId;
- -- Check if this job is locked
- local lockKey = jobKey .. ':lock'
- local lock = rcall("GET", lockKey)
- if not lock then
- if removeChildren == "1" then
- local dependencies = rcall("SMEMBERS", jobKey .. ":dependencies")
- if (#dependencies > 0) then
- for i, childJobKey in ipairs(dependencies) do
- -- We need to get the jobId for this job.
- local childJobId = getJobIdFromKey(childJobKey)
- local childJobPrefix = getJobKeyPrefix(childJobKey, childJobId)
- local result = isLocked( childJobPrefix, childJobId, removeChildren )
- if result then
- return true
- end
- end
- end
- end
- return false
- end
- return true
- end
- local removeJobChildren
- local removeJobWithChildren
- removeJobChildren = function(prefix, jobKey, options)
- -- Check if this job has children
- -- If so, we are going to try to remove the children recursively in a depth-first way
- -- because if some job is locked, we must exit with an error.
- if not options.ignoreProcessed then
- local processed = rcall("HGETALL", jobKey .. ":processed")
- if #processed > 0 then
- for i = 1, #processed, 2 do
- local childJobId = getJobIdFromKey(processed[i])
- local childJobPrefix = getJobKeyPrefix(processed[i], childJobId)
- removeJobWithChildren(childJobPrefix, childJobId, jobKey, options)
- end
- end
- local failed = rcall("HGETALL", jobKey .. ":failed")
- if #failed > 0 then
- for i = 1, #failed, 2 do
- local childJobId = getJobIdFromKey(failed[i])
- local childJobPrefix = getJobKeyPrefix(failed[i], childJobId)
- removeJobWithChildren(childJobPrefix, childJobId, jobKey, options)
- end
- end
- local unsuccessful = rcall("ZRANGE", jobKey .. ":unsuccessful", 0, -1)
- if #unsuccessful > 0 then
- for i = 1, #unsuccessful, 1 do
- local childJobId = getJobIdFromKey(unsuccessful[i])
- local childJobPrefix = getJobKeyPrefix(unsuccessful[i], childJobId)
- removeJobWithChildren(childJobPrefix, childJobId, jobKey, options)
- end
- end
- end
- local dependencies = rcall("SMEMBERS", jobKey .. ":dependencies")
- if #dependencies > 0 then
- for i, childJobKey in ipairs(dependencies) do
- local childJobId = getJobIdFromKey(childJobKey)
- local childJobPrefix = getJobKeyPrefix(childJobKey, childJobId)
- removeJobWithChildren(childJobPrefix, childJobId, jobKey, options)
- end
- end
- end
- removeJobWithChildren = function(prefix, jobId, parentKey, options)
- local jobKey = prefix .. jobId
- if options.ignoreLocked then
- if isLocked(prefix, jobId) then
- return
- end
- end
- -- Check if job is in the failed zset
- local failedSet = prefix .. "failed"
- if not (options.ignoreProcessed and rcall("ZSCORE", failedSet, jobId)) then
- removeParentDependencyKey(jobKey, false, parentKey, nil)
- if options.removeChildren then
- removeJobChildren(prefix, jobKey, options)
- end
- local prev = removeJobFromAnyState(prefix, jobId)
- local deduplicationId = rcall("HGET", jobKey, "deid")
- removeDeduplicationKeyIfNeededOnRemoval(prefix, jobId, deduplicationId)
- if removeJobKeys(jobKey) > 0 then
- local metaKey = prefix .. "meta"
- local maxEvents = getOrSetMaxEvents(metaKey)
- rcall("XADD", prefix .. "events", "MAXLEN", "~", maxEvents, "*", "event", "removed",
- "jobId", jobId, "prev", prev)
- end
- end
- end
- local prefix = ARGV[1]
- local jobId = ARGV[2]
- local jobKey = KEYS[1]
- local metaKey = KEYS[2]
- local options = {
- removeChildren = "1",
- ignoreProcessed = true,
- ignoreLocked = true
- }
- removeJobChildren(prefix, jobKey, options)
- `;
- export const removeUnprocessedChildren = {
- name: 'removeUnprocessedChildren',
- content,
- keys: 2,
- };
- //# sourceMappingURL=removeUnprocessedChildren-2.js.map
|