moveJobsToWait-8.lua 1.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273
  1. --[[
  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. --- @include "includes/addBaseMarkerIfNeeded"
  25. --- @include "includes/batches"
  26. --- @include "includes/getOrSetMaxEvents"
  27. --- @include "includes/getTargetQueueList"
  28. local metaKey = KEYS[6]
  29. local target, isPausedOrMaxed = getTargetQueueList(metaKey, KEYS[7], KEYS[4], KEYS[5])
  30. local jobs = rcall('ZRANGEBYSCORE', KEYS[3], 0, timestamp, 'LIMIT', 0, maxCount)
  31. if (#jobs > 0) then
  32. if ARGV[3] == "failed" then
  33. for i, key in ipairs(jobs) do
  34. local jobKey = KEYS[1] .. key
  35. rcall("HDEL", jobKey, "finishedOn", "processedOn", "failedReason")
  36. end
  37. elseif ARGV[3] == "completed" then
  38. for i, key in ipairs(jobs) do
  39. local jobKey = KEYS[1] .. key
  40. rcall("HDEL", jobKey, "finishedOn", "processedOn", "returnvalue")
  41. end
  42. end
  43. local maxEvents = getOrSetMaxEvents(metaKey)
  44. for i, key in ipairs(jobs) do
  45. -- Emit waiting event
  46. rcall("XADD", KEYS[2], "MAXLEN", "~", maxEvents, "*", "event",
  47. "waiting", "jobId", key, "prev", ARGV[3]);
  48. end
  49. for from, to in batches(#jobs, 7000) do
  50. rcall("ZREM", KEYS[3], unpack(jobs, from, to))
  51. rcall("LPUSH", target, unpack(jobs, from, to))
  52. end
  53. addBaseMarkerIfNeeded(KEYS[8], isPausedOrMaxed)
  54. end
  55. maxCount = maxCount - #jobs
  56. if (maxCount <= 0) then return 1 end
  57. return 0