185 lines
6.4 KiB
JavaScript
185 lines
6.4 KiB
JavaScript
"use strict";
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
|
exports.moveStalledJobsToWait = void 0;
|
|
const content = `--[[
|
|
Move stalled jobs to wait.
|
|
Input:
|
|
KEYS[1] 'stalled' (SET)
|
|
KEYS[2] 'wait', (LIST)
|
|
KEYS[3] 'active', (LIST)
|
|
KEYS[4] 'stalled-check', (KEY)
|
|
KEYS[5] 'meta', (KEY)
|
|
KEYS[6] 'paused', (LIST)
|
|
KEYS[7] 'marker'
|
|
KEYS[8] 'event stream' (STREAM)
|
|
ARGV[1] Max stalled job count
|
|
ARGV[2] queue.toKey('')
|
|
ARGV[3] timestamp
|
|
ARGV[4] max check time
|
|
Events:
|
|
'stalled' with stalled job id.
|
|
]]
|
|
local rcall = redis.call
|
|
-- 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 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 move job to wait to be picked up by a waiting worker.
|
|
]]
|
|
-- Includes
|
|
--[[
|
|
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 moveJobToWait(metaKey, activeKey, waitKey, pausedKey, markerKey, eventStreamKey,
|
|
jobId, pushCmd)
|
|
local target, isPausedOrMaxed = getTargetQueueList(metaKey, activeKey, waitKey, pausedKey)
|
|
addJobInTargetList(target, markerKey, pushCmd, isPausedOrMaxed, jobId)
|
|
rcall("XADD", eventStreamKey, "*", "event", "waiting", "jobId", jobId, 'prev', 'active')
|
|
end
|
|
--[[
|
|
Function to trim events, default 10000.
|
|
]]
|
|
-- Includes
|
|
--[[
|
|
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
|
|
local function trimEvents(metaKey, eventStreamKey)
|
|
local maxEvents = getOrSetMaxEvents(metaKey)
|
|
if maxEvents then
|
|
rcall("XTRIM", eventStreamKey, "MAXLEN", "~", maxEvents)
|
|
else
|
|
rcall("XTRIM", eventStreamKey, "MAXLEN", "~", 10000)
|
|
end
|
|
end
|
|
local stalledKey = KEYS[1]
|
|
local waitKey = KEYS[2]
|
|
local activeKey = KEYS[3]
|
|
local stalledCheckKey = KEYS[4]
|
|
local metaKey = KEYS[5]
|
|
local pausedKey = KEYS[6]
|
|
local markerKey = KEYS[7]
|
|
local eventStreamKey = KEYS[8]
|
|
local maxStalledJobCount = tonumber(ARGV[1])
|
|
local queueKeyPrefix = ARGV[2]
|
|
local timestamp = ARGV[3]
|
|
local maxCheckTime = ARGV[4]
|
|
if rcall("EXISTS", stalledCheckKey) == 1 then
|
|
return {}
|
|
end
|
|
rcall("SET", stalledCheckKey, timestamp, "PX", maxCheckTime)
|
|
-- Trim events before emiting them to avoid trimming events emitted in this script
|
|
trimEvents(metaKey, eventStreamKey)
|
|
-- Move all stalled jobs to wait
|
|
local stalling = rcall('SMEMBERS', stalledKey)
|
|
local stalled = {}
|
|
if (#stalling > 0) then
|
|
rcall('DEL', stalledKey)
|
|
-- Remove from active list
|
|
for i, jobId in ipairs(stalling) do
|
|
-- Markers in waitlist DEPRECATED in v5: Remove in v6.
|
|
if string.sub(jobId, 1, 2) == "0:" then
|
|
-- If the jobId is a delay marker ID we just remove it.
|
|
rcall("LREM", activeKey, 1, jobId)
|
|
else
|
|
local jobKey = queueKeyPrefix .. jobId
|
|
-- Check that the lock is also missing, then we can handle this job as really stalled.
|
|
if (rcall("EXISTS", jobKey .. ":lock") == 0) then
|
|
-- Remove from the active queue.
|
|
local removed = rcall("LREM", activeKey, 1, jobId)
|
|
if (removed > 0) then
|
|
-- If this job has been stalled too many times, such as if it crashes the worker, then fail it.
|
|
local stalledCount = rcall("HINCRBY", jobKey, "stc", 1)
|
|
-- Check if this is a repeatable job by looking at job options
|
|
local jobOpts = rcall("HGET", jobKey, "opts")
|
|
local isRepeatableJob = false
|
|
if jobOpts then
|
|
local opts = cjson.decode(jobOpts)
|
|
if opts and opts["repeat"] then
|
|
isRepeatableJob = true
|
|
end
|
|
end
|
|
-- Only fail job if it exceeds stall limit AND is not a repeatable job
|
|
if stalledCount > maxStalledJobCount and not isRepeatableJob then
|
|
local failedReason = "job stalled more than allowable limit"
|
|
rcall("HSET", jobKey, "defa", failedReason)
|
|
end
|
|
moveJobToWait(metaKey, activeKey, waitKey, pausedKey, markerKey, eventStreamKey, jobId,
|
|
"RPUSH")
|
|
-- Emit the stalled event
|
|
rcall("XADD", eventStreamKey, "*", "event", "stalled", "jobId", jobId)
|
|
table.insert(stalled, jobId)
|
|
end
|
|
end
|
|
end
|
|
end
|
|
end
|
|
-- Mark potentially stalled jobs
|
|
local active = rcall('LRANGE', activeKey, 0, -1)
|
|
if (#active > 0) then
|
|
for from, to in batches(#active, 7000) do
|
|
rcall('SADD', stalledKey, unpack(active, from, to))
|
|
end
|
|
end
|
|
return stalled
|
|
`;
|
|
exports.moveStalledJobsToWait = {
|
|
name: 'moveStalledJobsToWait',
|
|
content,
|
|
keys: 8,
|
|
};
|
|
//# sourceMappingURL=moveStalledJobsToWait-8.js.map
|