moveJobsToWait-8.js 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115
  1. const content = `--[[
  2. Move completed, failed or delayed jobs to wait.
  3. Note: Does not support jobs with priorities.
  4. Input:
  5. KEYS[1] base key
  6. KEYS[2] events stream
  7. KEYS[3] state key (failed, completed, delayed)
  8. KEYS[4] 'wait'
  9. KEYS[5] 'paused'
  10. KEYS[6] 'meta'
  11. KEYS[7] 'active'
  12. KEYS[8] 'marker'
  13. ARGV[1] count
  14. ARGV[2] timestamp
  15. ARGV[3] prev state
  16. Output:
  17. 1 means the operation is not completed
  18. 0 means the operation is completed
  19. ]]
  20. local maxCount = tonumber(ARGV[1])
  21. local timestamp = tonumber(ARGV[2])
  22. local rcall = redis.call;
  23. -- Includes
  24. --[[
  25. Add marker if needed when a job is available.
  26. ]]
  27. local function addBaseMarkerIfNeeded(markerKey, isPausedOrMaxed)
  28. if not isPausedOrMaxed then
  29. rcall("ZADD", markerKey, 0, "0")
  30. end
  31. end
  32. --[[
  33. Function to loop in batches.
  34. Just a bit of warning, some commands as ZREM
  35. could receive a maximum of 7000 parameters per call.
  36. ]]
  37. local function batches(n, batchSize)
  38. local i = 0
  39. return function()
  40. local from = i * batchSize + 1
  41. i = i + 1
  42. if (from <= n) then
  43. local to = math.min(from + batchSize - 1, n)
  44. return from, to
  45. end
  46. end
  47. end
  48. --[[
  49. Function to get max events value or set by default 10000.
  50. ]]
  51. local function getOrSetMaxEvents(metaKey)
  52. local maxEvents = rcall("HGET", metaKey, "opts.maxLenEvents")
  53. if not maxEvents then
  54. maxEvents = 10000
  55. rcall("HSET", metaKey, "opts.maxLenEvents", maxEvents)
  56. end
  57. return maxEvents
  58. end
  59. --[[
  60. Function to check for the meta.paused key to decide if we are paused or not
  61. (since an empty list and !EXISTS are not really the same).
  62. ]]
  63. local function getTargetQueueList(queueMetaKey, activeKey, waitKey, pausedKey)
  64. local queueAttributes = rcall("HMGET", queueMetaKey, "paused", "concurrency", "max", "duration")
  65. if queueAttributes[1] then
  66. return pausedKey, true, queueAttributes[3], queueAttributes[4]
  67. else
  68. if queueAttributes[2] then
  69. local activeCount = rcall("LLEN", activeKey)
  70. if activeCount >= tonumber(queueAttributes[2]) then
  71. return waitKey, true, queueAttributes[3], queueAttributes[4]
  72. else
  73. return waitKey, false, queueAttributes[3], queueAttributes[4]
  74. end
  75. end
  76. end
  77. return waitKey, false, queueAttributes[3], queueAttributes[4]
  78. end
  79. local metaKey = KEYS[6]
  80. local target, isPausedOrMaxed = getTargetQueueList(metaKey, KEYS[7], KEYS[4], KEYS[5])
  81. local jobs = rcall('ZRANGEBYSCORE', KEYS[3], 0, timestamp, 'LIMIT', 0, maxCount)
  82. if (#jobs > 0) then
  83. if ARGV[3] == "failed" then
  84. for i, key in ipairs(jobs) do
  85. local jobKey = KEYS[1] .. key
  86. rcall("HDEL", jobKey, "finishedOn", "processedOn", "failedReason")
  87. end
  88. elseif ARGV[3] == "completed" then
  89. for i, key in ipairs(jobs) do
  90. local jobKey = KEYS[1] .. key
  91. rcall("HDEL", jobKey, "finishedOn", "processedOn", "returnvalue")
  92. end
  93. end
  94. local maxEvents = getOrSetMaxEvents(metaKey)
  95. for i, key in ipairs(jobs) do
  96. -- Emit waiting event
  97. rcall("XADD", KEYS[2], "MAXLEN", "~", maxEvents, "*", "event",
  98. "waiting", "jobId", key, "prev", ARGV[3]);
  99. end
  100. for from, to in batches(#jobs, 7000) do
  101. rcall("ZREM", KEYS[3], unpack(jobs, from, to))
  102. rcall("LPUSH", target, unpack(jobs, from, to))
  103. end
  104. addBaseMarkerIfNeeded(KEYS[8], isPausedOrMaxed)
  105. end
  106. maxCount = maxCount - #jobs
  107. if (maxCount <= 0) then return 1 end
  108. return 0
  109. `;
  110. export const moveJobsToWait = {
  111. name: 'moveJobsToWait',
  112. content,
  113. keys: 8,
  114. };
  115. //# sourceMappingURL=moveJobsToWait-8.js.map