Files
Andrei 58f8093689 Rebrand from 'Redirect Intelligence v2' to 'URL Tracker Tool V2' throughout UI
- Updated all component headers and documentation
- Changed navbar and footer branding
- Updated homepage hero badge
- Modified page title in index.html
- Simplified footer text to 'Built with ❤️'
- Consistent V2 capitalization across all references
2025-08-19 19:12:23 +00:00

268 lines
9.0 KiB
Lua

--[[
Move job from active to a finished status (completed o failed)
A job can only be moved to completed if it was active.
The job must be locked before it can be moved to a finished status,
and the lock must be released in this script.
Input:
KEYS[1] wait key
KEYS[2] active key
KEYS[3] prioritized key
KEYS[4] event stream key
KEYS[5] stalled key
-- Rate limiting
KEYS[6] rate limiter key
KEYS[7] delayed key
KEYS[8] paused key
KEYS[9] meta key
KEYS[10] pc priority counter
KEYS[11] completed/failed key
KEYS[12] jobId key
KEYS[13] metrics key
ARGV[1] jobId
ARGV[2] timestamp
ARGV[3] msg property returnvalue / failedReason
ARGV[4] return value / failed reason
ARGV[5] target (completed/failed)
ARGV[6] event data (? maybe just send jobid).
ARGV[7] fetch next?
ARGV[8] keys prefix
ARGV[9] opts
opts - token - lock token
opts - keepJobs
opts - lockDuration - lock duration in milliseconds
opts - attempts max attempts
opts - attemptsMade
opts - maxMetricsSize
opts - fpof - fail parent on fail
opts - rdof - remove dependency on fail
Output:
0 OK
-1 Missing key.
-2 Missing lock.
-3 Job not in active set
-4 Job has pending dependencies
-6 Lock is not owned by this client
Events:
'completed/failed'
]]
local rcall = redis.call
--- Includes
--- @include "includes/collectMetrics"
--- @include "includes/getNextDelayedTimestamp"
--- @include "includes/moveJobFromPriorityToActive"
--- @include "includes/prepareJobForProcessing"
--- @include "includes/moveParentFromWaitingChildrenToFailed"
--- @include "includes/moveParentToWaitIfNeeded"
--- @include "includes/promoteDelayedJobs"
--- @include "includes/removeJobsByMaxAge"
--- @include "includes/removeJobsByMaxCount"
--- @include "includes/removeParentDependencyKey"
--- @include "includes/trimEvents"
--- @include "includes/updateParentDepsIfNeeded"
--- @include "includes/getRateLimitTTL"
local jobIdKey = KEYS[12]
if rcall("EXISTS", jobIdKey) == 1 then -- // Make sure job exists
local opts = cmsgpack.unpack(ARGV[9])
local token = opts['token']
local attempts = opts['attempts']
local attemptsMade = opts['attemptsMade']
local maxMetricsSize = opts['maxMetricsSize']
local maxCount = opts['keepJobs']['count']
local maxAge = opts['keepJobs']['age']
if token ~= "0" then
local lockKey = jobIdKey .. ':lock'
local lockToken = rcall("GET", lockKey)
if lockToken == token then
rcall("DEL", lockKey)
rcall("SREM", KEYS[5], ARGV[1])
else
if lockToken then
-- Lock exists but token does not match
return -6
else
-- Lock is missing completely
return -2
end
end
end
if rcall("SCARD", jobIdKey .. ":dependencies") ~= 0 then -- // Make sure it does not have pending dependencies
return -4
end
local parentReferences = rcall("HMGET", jobIdKey, "parentKey", "parent")
local parentKey = parentReferences[1] or ""
local parentId = ""
local parentQueueKey = ""
if parentReferences[2] ~= false then
local jsonDecodedParent = cjson.decode(parentReferences[2])
parentId = jsonDecodedParent['id']
parentQueueKey = jsonDecodedParent['queueKey']
end
local jobId = ARGV[1]
local timestamp = ARGV[2]
-- Remove from active list (if not active we shall return error)
local numRemovedElements = rcall("LREM", KEYS[2], -1, jobId)
if (numRemovedElements < 1) then return -3 end
-- Trim events before emiting them to avoid trimming events emitted in this script
trimEvents(KEYS[9], KEYS[4])
-- If job has a parent we need to
-- 1) remove this job id from parents dependencies
-- 2) move the job Id to parent "processed" set
-- 3) push the results into parent "results" list
-- 4) if parent's dependencies is empty, then move parent to "wait/paused". Note it may be a different queue!.
if parentId == "" and parentKey ~= "" then
parentId = getJobIdFromKey(parentKey)
parentQueueKey = getJobKeyPrefix(parentKey, ":" .. parentId)
end
if parentId ~= "" then
if ARGV[5] == "completed" then
local dependenciesSet = parentKey .. ":dependencies"
if rcall("SREM", dependenciesSet, jobIdKey) == 1 then
updateParentDepsIfNeeded(parentKey, parentQueueKey,
dependenciesSet, parentId, jobIdKey,
ARGV[4], timestamp)
end
else
if opts['fpof'] then
moveParentFromWaitingChildrenToFailed(parentQueueKey, parentKey,
parentId, jobIdKey, timestamp)
elseif opts['rdof'] then
local dependenciesSet = parentKey .. ":dependencies"
if rcall("SREM", dependenciesSet, jobIdKey) == 1 then
moveParentToWaitIfNeeded(parentQueueKey, dependenciesSet,
parentKey, parentId, timestamp)
end
end
end
end
-- Remove job?
if maxCount ~= 0 then
local targetSet = KEYS[11]
-- Add to complete/failed set
rcall("ZADD", targetSet, timestamp, jobId)
rcall("HMSET", jobIdKey, ARGV[3], ARGV[4], "finishedOn", timestamp)
-- "returnvalue" / "failedReason" and "finishedOn"
-- Remove old jobs?
local prefix = ARGV[8]
if maxAge ~= nil then
removeJobsByMaxAge(timestamp, maxAge, targetSet, prefix)
end
if maxCount ~= nil and maxCount > 0 then
removeJobsByMaxCount(maxCount, targetSet, prefix)
end
else
rcall("DEL", jobIdKey, jobIdKey .. ':logs', jobIdKey .. ':processed')
if parentKey ~= "" then
removeParentDependencyKey(jobIdKey, false, parentKey)
end
end
rcall("XADD", KEYS[4], "*", "event", ARGV[5], "jobId", jobId, ARGV[3],
ARGV[4])
if ARGV[5] == "failed" then
if tonumber(attemptsMade) >= tonumber(attempts) then
rcall("XADD", KEYS[4], "*", "event", "retries-exhausted", "jobId",
jobId, "attemptsMade", attemptsMade)
end
end
-- Collect metrics
if maxMetricsSize ~= "" then
collectMetrics(KEYS[13], KEYS[13] .. ':data', maxMetricsSize, timestamp)
end
-- Try to get next job to avoid an extra roundtrip if the queue is not closing,
-- and not rate limited.
if (ARGV[7] == "1") then
local target, paused = getTargetQueueList(KEYS[9], KEYS[1], KEYS[8])
-- Check if there are delayed jobs that can be promoted
promoteDelayedJobs(KEYS[7], KEYS[1], target, KEYS[3],
KEYS[4], ARGV[8], timestamp, paused, KEYS[10])
local maxJobs = tonumber(opts['limiter'] and opts['limiter']['max'])
-- Check if we are rate limited first.
local expireTime = getRateLimitTTL(maxJobs, KEYS[6])
if expireTime > 0 then return {0, 0, expireTime, 0} end
-- paused queue
if paused then return {0, 0, 0, 0} end
jobId = rcall("RPOPLPUSH", KEYS[1], KEYS[2])
if jobId then
if string.sub(jobId, 1, 2) == "0:" then
rcall("LREM", KEYS[2], 1, jobId)
-- If jobId is special ID 0:delay (delay greater than 0), then there is no job to process
-- but if ID is 0:0, then there is at least 1 prioritized job to process
if jobId == "0:0" then
jobId = moveJobFromPriorityToActive(KEYS[3], KEYS[2], KEYS[10])
return prepareJobForProcessing(KEYS, ARGV[8], target, jobId, timestamp,
maxJobs, expireTime, opts)
end
else
return prepareJobForProcessing(KEYS, ARGV[8], target, jobId, timestamp, maxJobs,
expireTime, opts)
end
else
jobId = moveJobFromPriorityToActive(KEYS[3], KEYS[2], KEYS[10])
if jobId then
return prepareJobForProcessing(KEYS, ARGV[8], target, jobId, timestamp, maxJobs,
expireTime, opts)
end
end
-- Return the timestamp for the next delayed job if any.
local nextTimestamp = getNextDelayedTimestamp(KEYS[7])
if nextTimestamp ~= nil then
-- The result is guaranteed to be positive, since the
-- ZRANGEBYSCORE command would have return a job otherwise.
return {0, 0, 0, nextTimestamp}
end
end
local waitLen = rcall("LLEN", KEYS[1])
if waitLen == 0 then
local activeLen = rcall("LLEN", KEYS[2])
if activeLen == 0 then
local prioritizedLen = rcall("ZCARD", KEYS[3])
if prioritizedLen == 0 then
rcall("XADD", KEYS[4], "*", "event", "drained")
end
end
end
return 0
else
return -1
end