apalis-redis 1.0.0

Background task processing for rust using apalis and redis
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

-- KEYS[1]: this worker's inflight set
-- KEYS[2]: the active workers sorted set
-- KEYS[3]: the task metadata prefix
-- KEYS[4]: the active jobs list (jobs must currently be queued, not already inflight elsewhere)
-- ARGV[1]: worker name
-- ARGV[2]: current timestamp
-- ARGV[3]: emit pubsub events
-- ARGV[4..]: task_ids to acquire
-- Returns: number of leases acquired (== #task_ids, or the call errors)
local worker_inflight_set = KEYS[1]
local worker_name = ARGV[1]
local now = ARGV[2]
local emit_events = ARGV[3] == "1"

-- Confirm the worker itself is registered/alive
local registered = redis.call("zscore", KEYS[2], worker_name)
if not registered then
    error("Cant acquire lease: worker not registered")
end

local task_ids = {}
for i = 4, #ARGV do
    table.insert(task_ids, ARGV[i])
end

if #task_ids == 0 then
    return 0
end

-- Validate first: every task must already be in this worker's inflight set
for _, task_id in ipairs(task_ids) do
    local is_member = redis.call("sismember", worker_inflight_set, task_id)
    if is_member == 0 then
        error("task not held by this worker: " .. task_id)
    end
end

-- All validated — now mutate
for _, task_id in ipairs(task_ids) do
    redis.call("lrem", KEYS[4], 1, task_id)
    redis.call("sadd", KEYS[1], task_id)

    local meta_key = KEYS[3] .. ":" .. task_id
    redis.call("hset", meta_key, "locked_at", now, "locked_by", worker_name, "status", "Running")
    if emit_events then
        redis.call("publish", "tasks:" .. KEYS[4] .. ':lock', task_id)
    end
end

return #task_ids