class

JoobQ::Delivery::Ledger

Inherits Reference < Object

Constants

MAX_BATCH = 128
MAX_BATCH_BYTES = (2 * 1024) * 1024
QUARANTINE = " local raw = redis.call('HGET', KEYS[1], ARGV[1])\n if raw ~= ARGV[2] then return 0 end\n redis.call('HSET', KEYS[4], ARGV[1], raw)\n redis.call('ZREM', KEYS[2], ARGV[1])\n redis.call('ZREM', KEYS[3], ARGV[1])\n return 1"
REPAIR = " if redis.call('HGET', KEYS[1], ARGV[1]) ~= ARGV[2] then return 0 end\n redis.call('ZREM', KEYS[2], ARGV[1])\n redis.call('ZREM', KEYS[3], ARGV[1])\n if ARGV[3] ~= '' then redis.call('ZADD', KEYS[2], ARGV[3], ARGV[1]) end\n if ARGV[4] ~= '' then redis.call('ZADD', KEYS[3], ARGV[4], ARGV[1]) end\n return 1"

Repair checks the exact observed canonical record, so a stale repair pass cannot overwrite indexes for a newer claim or terminal outcome.

WRITE = " local old = redis.call('HGET', KEYS[1], ARGV[1])\n if ARGV[2] == '' then\n if old then return 0 end\n if redis.call('HLEN', KEYS[1]) >= tonumber(ARGV[8]) then return -1 end\n elseif old ~= ARGV[2] then return 0 end\n local clock = redis.call('TIME')\n local now = tonumber(clock[1]) * 1000 + math.floor(tonumber(clock[2]) / 1000)\n -- Validate time inside the same atomic operation as the ownership CAS.\n if ARGV[6] ~= '' and now >= tonumber(ARGV[6]) then return 0 end\n if ARGV[7] ~= '' and now < tonumber(ARGV[7]) then return 0 end\n if ARGV[5] ~= '' and now >= tonumber(ARGV[5]) then return 0 end\n if ARGV[5] ~= '' and redis.call('EXISTS', KEYS[4]) == 1 then return 0 end\n redis.call('HSET', KEYS[1], ARGV[1], ARGV[3])\n redis.call('ZREM', KEYS[2], ARGV[1])\n redis.call('ZREM', KEYS[3], ARGV[1])\n if ARGV[4] ~= '' then redis.call('ZADD', KEYS[2], ARGV[4], ARGV[1]) end\n if ARGV[5] ~= '' then redis.call('ZADD', KEYS[3], ARGV[5], ARGV[1]) end\n return 1"

One authoritative write before derived indexes. If an index command fails, canonical intent survives and repair reconstructs it. Never assume rollback.

Constructors

new(redis : Redis::PooledClient, namespace : String, queue : String, max_payload_bytes : Int32 = 1048576, max_records : Int32 = 100000, optimized : Bool = true)
Source

Instance methods

admit(request_id : String, payload : String, delay : Time::Span = Time::Span.zero, traceparent : String | Nil = nil) : Admission
Source
admit_many(requests : Array(Tuple(String, String)), delay : Time::Span = Time::Span.zero, traceparent : String | Nil = nil) : Array(Admission)

Bounded transport batches, not all-or-nothing transactions. An interrupted script may have persisted any prefix; callers reconcile each stable ID.

Source
claim(owner : String, lease : Time::Span = 30.seconds) : Claim | Nil
Source
claim_many(owner : String, limit : Int32, lease : Time::Span = 30.seconds) : Array(Claim)
Source
gate_key
Source
import(request_id : String, payload : String, state : State, due_at : Int64, owner : String) : Admission

Offline migration only. The operator-owned gate prevents claims while the imported set is verified. Existing identities must agree exactly.

Source
leases_key
Source
quarantine_key
Source
ready_key
Source
records_key
Source
renew(claim : Claim, lease : Time::Span = 30.seconds) : Settlement
Source
repair(cursor : String = "0", count : Int32 = 100) : String

One HSCAN page (COUNT is a Redis work hint, not a strict result cap). Continue with the returned cursor until "0"; schedule repeated full cycles to repair missed indexes and expired leases.

Source
resolve(request_id : String) : Record | Nil
Source
resolve_many(request_ids : Array(String)) : Array(Record | Nil)

Batch reads still decode/checksum every canonical receipt, preserving the benchmark's per-identity conservation check without a round trip per ID.

Source
settle(claim : Claim, state : State = State::Completed, delay : Time::Span = Time::Span.zero) : Settlement
Source
settle_many(items : Array(Tuple(Claim, State, Time::Span))) : Array(Settlement)
Source