One timer, not N

EchoMQ.LockManager is a GenServer. At init it reads lock_duration (default 30000 ms — the lock TTL) and lock_renew_time (default div(lock_duration, 2) = 15000 ms), then arms one timer for the whole worker:

defp schedule_renewal(lock_renew_time) do
  interval = div(lock_renew_time, 2)            -- 7500 ms by default
  Process.send_after(self(), :extend_locks, interval)
end

Each worker has exactly one of these timers, regardless of how many jobs it holds. When the worker picks a job up it calls track_job(manager, job_id, token) (arity 3), storing the job's token and a millisecond timestamp; on complete or fail it calls untrack_job(manager, job_id) (arity 2). The public surface is start_link/1, track_job/3, untrack_job/2, get_active_job_count/1, get_tracked_job_ids/1, is_tracked?/2, and stop/1.

When the timer fires, handle_info(:extend_locks, state) scans all tracked_jobs and selects those whose info.ts + threshold < now, where threshold = div(lock_renew_time, 2). Step the scan below over a fixed set of tracked jobs, each with its own elapsed-since-renewal time, and read which jobs go into the one batched call.

The due jobs are renewed together via EchoMQ.Scripts.extend_locks(connection, keys, job_ids, tokens, lock_duration) — arity 5, the EchoMQ.LockManager.extend_locks/5 path. The script side packs all of it into one call with exactly one key, so the script is extendLocks-1.lua:

def extend_locks(conn, ctx, job_ids, tokens, duration) do
  keys = [Keys.stalled(ctx)]               -- exactly ONE key
  args = [
    Keys.key(ctx),                         -- the baseKey
    Msgpax.pack!(tokens, iodata: false),   -- all tokens in one ARGV
    Msgpax.pack!(job_ids, iodata: false),  -- all ids in one ARGV
    duration
  ]
  execute(conn, :extend_locks, keys, args)
end

Inside extendLocks-1.lua the script walks the packed ids and tokens. For each job it reads the lock; if the stored token matches it re-sets the PX TTL and removes the job from the stalled set; otherwise it records the id as failed:

-- extendLocks-1.lua · KEYS[1] stalled · ARGV baseKey, tokens, jobIds, lockDuration
local currentToken = rcall("GET", baseKey .. jobIds[i] .. ':lock')
if currentToken == token then
  rcall("SET", lockKey, token, "PX", lockDuration)   -- re-set the TTL
  rcall("SREM", stalledKey, jobId)                   -- clear from stalled set
else
  table.insert(failedJobs, jobId)                    -- record a lost lock
end
return failedJobs   -- the FAILED ids; empty means all renewed

The script returns the list of failed job ids — an empty list means every lock was renewed. The LockManager untracks any job whose lock was lost and fires its on_lock_renewal_failed callback; that job has stalled, and the stalled checker recovers it.

Three numbers set the rhythm, with these defaults.

A lock lives 30000 ms; the one timer fires every ~7500 ms and renews any tracked job past its half-window threshold (div(lock_renew_time, 2) = 7500 ms elapsed). So every held lock is refreshed well before its 30000 ms TTL — the timeline below lays the three windows against each other: the TTL, the renew window inside it, and the timer ticks inside that.

The lock key emq:{queue}:{jobId}:lock (a UUID token under a PX TTL, L1) and the renew script extendLocks-1.lua (L2) are frozen and shared. The single-timer batch design — the threshold scan in handle_info, one batched extend_locks/5 — is the Elixir runtime's own OTP GenServer choice (L3/L4). Go and Node.js renew the same locks with the same extend script on their own schedule.

The lock-renewal protocol and the Redis commands the batched extension issues.