Module: Wurk::Lua

Defined in:

lib/wurk/lua.rb,
lib/wurk/lua/loader.rb,
lib/wurk/lua/fast_delete.rb

Overview

EVALSHA-cached Lua scripts. Loaded once per pool, never re-uploaded. Multi-pop, atomic schedule promotion, batch ops.

Source strings are intentionally bare — the SHA1 of each is computed at load time and is the same value Redis reports from SCRIPT LOAD. Whitespace edits change the SHA, which forces a re-upload at runtime.

:zpopbyscore implements the pop-one-due-entry step in sidekiq-free.md §1.8. It is the minimal expression of that step — range by score, take one, remove it, return it — and there is no room to write it differently without changing what it does.

Defined Under Namespace

Modules: FastDelete Classes: Loader

Constant Summary

ZPOPBYSCORE =

<<~LUA
  local key, now = KEYS[1], ARGV[1]
  local jobs = redis.call("zrange", key, "-inf", now, "byscore", "limit", 0, 1)
  if jobs[1] then
    redis.call("zrem", key, jobs[1])
    return jobs[1]
  end
LUA

RELIABLE_SCHEDULE_PROMOTE =

Pro reliable scheduler: atomically promote all due jobs in a sorted set to their target queues. Pure-Ruby promotion does ZRANGE → ZREM → LPUSH non-atomically and can lose jobs on a mid-step crash.

Each promoted payload is restamped with a fresh enqueued_at (ARGV, epoch ms): that field marks arrival on an immediate queue, so a job leaving schedule/retry must get a new one at promotion rather than keep its stale scheduled-origin value (or none). This matches the default Ruby scheduler, whose push path restamps enqueued_at too (Client#push_plain / #push_batched) — so both schedulers emit wire-identical promoted payloads (spec §7.1).

The restamp is a surgical string patch, NOT a cjson.decode -> cjson.encode round-trip: cjson maps every JSON number to a Lua double, so re-encoding would silently corrupt integer args past 2^53 (snowflake IDs, 64-bit counters) and reformat 15+ digit numbers into scientific notation -- breaking wire-compat AND diverging from the loss-free Ruby scheduler, whose Ruby-side JSON keeps big integers exact. So the stored member is preserved byte-for-byte and only its top-level enqueued_at is rewritten: replaced in place when present (retry members keep theirs), inserted after the opening brace when absent (schedule members are stored stripped of it, via Client#push_scheduled). cjson.decode is still called -- but only to read the queue string and test for a top-level enqueued_at, never to re-serialize, so a lossy decode never reaches the payload. (Residual: a retry member whose user args embed a numeric key literally named enqueued_at ahead of the top-level one patches the arg instead -- vanishingly rare, and still strictly safer than round-tripping every arg.) A member that is not a JSON object naming a string queue (garbage bytes, a bare scalar, a nil/numeric queue) can never be promoted. Raising on it would abort the script at the same lowest-scored member on every sweep — and since one Ruby call drains retry before schedule, a single poison retry would starve the whole cluster's scheduled jobs. It is moved to the dead set (scored now, like a kill) so it stays inspectable, and the sweep continues. pcall keeps a failed decode from raising out of the script. KEYS = [sorted_set, queues_set, dead_set] ARGV = [now, queue_prefix, now_ms, batch] Returns the number of members handled (promoted or moved to dead). Order matters: decode + push BEFORE zrem. Redis Lua has no rollback, so a failed cjson.decode after a zrem would lose the job. Decode first; push first; only then remove from the sorted set — and zrem the ORIGINAL member, not the restamped copy. Worst case is a crash between lpush and zrem → at-least-once redelivery, never loss. ARGV caps members per call: Lua is atomic and single-threaded in Redis, so an unbatched promote of a post-outage backlog (100k+ due members) would block every client for the whole sweep. The Ruby caller loops until a short batch comes back. The first line is a Lua comment to Redis and Valkey. Dragonfly reads it as a per-script grant, because this script alone derives its target key from the payload (ARGV[2] .. q) and that server refuses undeclared keys by default. It spares operators the server-wide allow-undeclared-keys flag, which serializes every script and cost about 3x script throughput.

<<~LUA
  --!df flags=allow-undeclared-keys
  local jobs = redis.call("zrangebyscore", KEYS[1], "-inf", ARGV[1], "LIMIT", 0, tonumber(ARGV[4]))
  for i = 1, #jobs do
    local job = jobs[i]
    local ok, decoded = pcall(cjson.decode, job)
    local q = ok and type(decoded) == "table" and decoded["queue"]
    if type(q) == "string" then
      local pat, rep = '"enqueued_at":%-?%d[%d.eE+-]*', '"enqueued_at":' .. ARGV[3]
      if decoded["enqueued_at"] == nil then pat, rep = "^{", "{" .. rep .. "," end
      redis.call("sadd", KEYS[2], q)
      redis.call("lpush", ARGV[2] .. q, (string.gsub(job, pat, rep, 1)))
    else
      redis.call("zadd", KEYS[3], ARGV[1], job)
    end
    redis.call("zrem", KEYS[1], job)
  end
  return #jobs
LUA

SCHEDULED_CLAIM =

Default scheduler claim: lease up to ARGV due members instead of popping them. Each stays in its set, rescored past any due window, so a lost reply, a failed push or a crash costs a later re-promotion and never the job. The caller ZREMs a member only once it has been pushed.

Redis's clock decides both ends, so a host with a skewed clock can neither cut a lease short nor see another host's lease as due: the window is the earlier of the caller's and the server's time, and the expiry counts from the server's. tostring() would round a 10-digit epoch, hence the format. KEYS = [sorted_set] ARGV = [now, lease_seconds, limit] Returns the claimed members, lowest score first.

<<~LUA
  local t = redis.call("TIME")
  local server_now = tonumber(t[1]) + (tonumber(t[2]) / 1000000)
  local window = math.min(tonumber(ARGV[1]), server_now)
  local jobs = redis.call("zrange", KEYS[1], "-inf", string.format("%.6f", window), "byscore", "limit", 0, tonumber(ARGV[3]))
  local expiry = string.format("%.6f", server_now + tonumber(ARGV[2]))
  for i = 1, #jobs do
    redis.call("zadd", KEYS[1], "XX", expiry, jobs[i])
  end
  return jobs
LUA

RELIABLE_REQUEUE =

Reliable fetch (Pro super_fetch §3) shutdown requeue: atomically move one in-flight job from a per-process private list back to its public queue. The LPOS guard is the whole point — RPUSH runs only when the job is still in the private list. A job the Processor ACKed in the window between hard_shutdown's cross-thread job read and this move is NOT re-pushed, so the job lands in exactly one place and can't double-execute. RPUSH (public tail) not LPUSH so the reclaimed job is fetched next — LMOVE pops the tail — ahead of fresh LPUSH'd enqueues.

Push before remove, in this script, the one below and lua/reclaim_orphan.lua. Nothing rolls a script back, and a server can refuse the write mid-script (Dragonfly at maxmemory refuses the RPUSH after applying an LREM): the raise has to leave the job where it was, never in neither place. KEYS = [private_list, public_queue] ARGV = [job_json] Returns 1 when the job was moved, 0 when it was already acked.

<<~LUA
  if redis.call("lpos", KEYS[1], ARGV[1]) then
    redis.call("rpush", KEYS[2], ARGV[1])
    redis.call("lrem", KEYS[1], 1, ARGV[1])
    return 1
  end
  return 0
LUA

REQUEUE_UNIT =

UnitOfWork#requeue: unguarded, as the caller owns the unit and a job whose ACK already went out must still go back. Not replay-safe, so it is only ever sent once (Loader.eval_once). KEYS = [private_list, public_queue] ARGV = [job_json]

<<~LUA
  redis.call("rpush", KEYS[2], ARGV[1])
  return redis.call("lrem", KEYS[1], 1, ARGV[1])
LUA

RELEASE_IF_OWNER =

Ent Unique (§3): atomic compare-and-delete of a lock key. Replaces the two-command GET-then-DEL — between those calls the key can expire and a fresh owner can grab it, and the bare DEL would then drop the new owner's lock. Shared by Unique::ServerMiddleware#release (normal success/start release), Unique::DEATH_HANDLER (automatic-death release) and Leader#release (stepping down from the cluster lock) so those paths cannot drift. KEYS = [the lock key — unique: | dear-leader] ARGV = [the owner that must still hold it — jid | ::] Returns 1 when the key was deleted, 0 otherwise.

<<~LUA
  if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
  end
  return 0
LUA

CRON_CLAIM_FIRE =

Ent Periodic (§2): compare-and-swap the fire marks of loops:<lid>. The cron poller's leader gate is a cached read (Component#leader?, LEADER_CACHE_TTL_MS), so for a few seconds after a handover two processes both believe they lead and both reach the same due loop — HMGET → decide → enqueue → HSET then fires it twice. Claiming the slot atomically is what makes a tick fire exactly once; the loser gets 0 and enqueues nothing. Values written are the same decimal-epoch strings the Ruby path wrote, byte for byte — the dashboard reads these fields.

nf present: the caller must still be looking at the mark it read (byte compare, so no parse can drift the token). nf absent: the caller derived the slot from lf, so refuse once lf has reached it — that is the exhausted-schedule case, where the winner cleared nf and a loser would otherwise re-fire the very same slot. KEYS = [loops:] ARGV = [expected nf (or the derived slot), new lf, new nf ('' → HDEL)] Returns 1 when this caller claimed the slot, 0 otherwise.

<<~LUA
  local key = KEYS[1]
  local cur = redis.call("hget", key, "nf")
  if cur and cur ~= "" then
    if cur ~= ARGV[1] then return 0 end
  else
    local lf = tonumber(redis.call("hget", key, "lf") or "")
    if lf and lf >= tonumber(ARGV[1]) then return 0 end
  end
  redis.call("hset", key, "lf", ARGV[2])
  if ARGV[3] == "" then
    redis.call("hdel", key, "nf")
  else
    redis.call("hset", key, "nf", ARGV[3])
  end
  return 1
LUA

LUA_DIR =

The rest — batch (batch_*), limiter (limiter_*), flow and the other per-feature scripts — live in lib/wurk/lua/*.lua, one script per file. Loaded at boot, the file's basename (minus .lua) becomes the SCRIPTS key as a symbol. Separate files keep each script diffable on its own and self-contained for the redis-cli --eval debug workflow.

File.expand_path('lua', __dir__)

FILE_SCRIPTS =

Dir.glob(File.join(LUA_DIR, '*.lua')).to_h do |path|
  [File.basename(path, '.lua').to_sym, File.read(path)]
end.freeze

SCRIPTS =

{
  zpopbyscore: ZPOPBYSCORE,
  reliable_schedule_promote: RELIABLE_SCHEDULE_PROMOTE,
  scheduled_claim: SCHEDULED_CLAIM,
  reliable_requeue: RELIABLE_REQUEUE,
  requeue_unit: REQUEUE_UNIT,
  release_if_owner: RELEASE_IF_OWNER,
  cron_claim_fire: CRON_CLAIM_FIRE
}.merge(FILE_SCRIPTS).freeze

SHAS =

SHA1 of each script source — matches what SCRIPT LOAD returns. Precomputing keeps eval_cached allocation-free in the hot path.

SCRIPTS.transform_values { |src| Digest::SHA1.hexdigest(src) }.freeze