class

JoobQ::DeadLetterQueue

Inherits Reference < Object

Operator view of the dead-letter set: paged reads, redrive and purge. Each state change is one atomic Lua step, so two operators can never redrive or purge the same entry twice.

Constants

ADD = QUEUE_OF + " local added = 0\n for i = 2, #ARGV do\n if redis.call('ZADD', KEYS[1], ARGV[1], ARGV[i]) == 1 then\n redis.call('HINCRBY', KEYS[2], queue_of(ARGV[i]), 1)\n added = added + 1\n end\n end\n return added"

Adds members (ARGV[2..], scored ARGV[1]) to the set. A member that is already there is not counted again.

CHUNK = 200
COUNTS = "joobq:dead_letter:counts"
KEY = "joobq:dead_letter"
QUEUE_OF = " local function queue_of(member)\n local ok, job = pcall(cjson.decode, member)\n if ok and type(job) == 'table' and type(job.queue) == 'string' then return job.queue end\n return ''\n end"

The queue an entry belongs to, read the way parse reads it: from the entry itself, never from the caller.

READ = " return {redis.call('ZCARD', KEYS[1]), redis.call('HGETALL', KEYS[2])}"

The set's size and the counts, read together; and the counts rebuilt from the set (for data written before they existed). The rebuild blocks Redis for as long as the scan takes, so it runs only when the two disagree.

RECOUNT = QUEUE_OF + " redis.call('DEL', KEYS[2])\n for _, member in ipairs(redis.call('ZRANGE', KEYS[1], 0, -1)) do\n redis.call('HINCRBY', KEYS[2], queue_of(member), 1)\n end\n return 1"
REDRIVE = " local kind = redis.call('TYPE', KEYS[2]).ok\n if kind ~= 'list' and kind ~= 'none' then\n return -1\n end\n if redis.call('ZREM', KEYS[1], ARGV[1]) == 1 then\n redis.call('RPUSH', KEYS[2], ARGV[2])\n redis.call('DEL', KEYS[3])\n redis.call('HINCRBY', KEYS[4], ARGV[3], -1) -- ARGV[3]: the entry's own queue\n return 1\n end\n return 0"

Moves one entry back onto its queue; 0 when someone else got there first, -1 when the queue's key holds something that is not a list (nothing is touched then: a failure after the ZREM would lose the entry).

REMOVE = QUEUE_OF + " local removed = 0\n for _, member in ipairs(ARGV) do\n if redis.call('ZREM', KEYS[1], member) == 1 then\n redis.call('HINCRBY', KEYS[2], queue_of(member), -1)\n removed = removed + 1\n end\n end\n return removed"

Removes members (ARGV) from the set and from the counts; returns how many were there.

UNKNOWN = ""

Entries whose JSON has no queue are counted under this name, so the counts always add up to the set.

Constructors

new(redis : Redis::PooledClient = RedisStore.instance.redis)
Source

Class methods

add(redis : Redis::PooledClient, score : Int64, members : Array(String)) : Int64

Every writer to the set goes through here, so the counts stay exact. Members already present are not counted twice; returns how many were new.

Source

Instance methods

count(queue : String | Nil = nil) : Int64
Source
counts

Entries per queue without reading them. If the counts do not add up to the set (entries written by code that predates them), they are rebuilt once.

Source
find(jid : String) : Entry | Nil
Source
page(offset : Int32 = 0, limit : Int32 = 50, queue : String | Nil = nil) : Array(Entry)

Newest first. queue filters; offset/limit page through the matches.

Source
purge(jids : Array(String)) : Outcome
Source
purge_all(queue : String | Nil = nil) : Int32
Source
redrive(jids : Array(String)) : Outcome
Source
redrive_reviewed(jid : String, expected_digest : String) : Outcome

Replay precisely the member an operator reviewed. The Lua transition consumes that member, so a replacement between lookup and EVAL conflicts.

Source

Nested types