@@ -0,0 +1,17 @@
+# Redis Queues Survive Payloads Lua Cannot Decode
+
+## Bug Fix
+
+Redis Lua scripts decode job JSON with cjson, which rejects some JSON that `JSON.parse` accepts: a
+lone surrogate such as `'\ud800'`, or more than 1000 levels of nesting. One such payload poisoned
+its queue:
+
+- `popFrom()` failed every time once the job was in the queue, or due as a delayed job, so no job of
+  that queue ran;
+- `retryJob()` removed the job from the active jobs before failing, so it ended in no state and
+  never ran again;
+- `recoverStalledJobs()` failed, so stalled jobs of that queue were never recovered.
+
+The scripts no longer need to decode the payload: they read the priority, attempts, and stalled
+count with a guarded decode, and fall back to the defaults (priority 5) for such a job. Scripts
+read everything they need before their first write.
@@ -36,6 +36,18 @@ export function encodeRedisJobPayloadOverlay(payload: JobData['payload']): [stri
 }
 
 export const REDIS_JOB_STORAGE_LUA = `
+  -- Read the fields a script needs from the original job JSON. cjson rejects
+  -- some JSON that JSON.parse accepts, such as a lone surrogate or more than
+  -- 1000 levels of nesting in the payload: such a job gets the defaults
+  -- (priority 5, attempts 0) instead of failing every script that touches it.
+  local function read_job_fields(job_data)
+    local ok, job = pcall(cjson.decode, job_data)
+    if ok and type(job) == 'table' then
+      return job
+    end
+    return {}
+  end
+
   local function read_job_overlay(overlay_key, job_id)
     local overlay_json = redis.call('HGET', overlay_key, job_id)
     if overlay_json then
 
@@ -128,8 +128,7 @@ ${REDIS_JOB_STORAGE_LUA}
       local job_id = ready_job_ids[i]
       local job_data = redis.call('HGET', data_key, job_id)
       if job_data then
-        local job = cjson.decode(job_data)
-        local priority = job.priority or 5
+        local priority = read_job_fields(job_data).priority or 5
         local score = priority * 10000000000000 + now
         redis.call('ZADD', pending_key, score, job_id)
         redis.call('ZREM', delayed_key, job_id)
@@ -198,8 +197,8 @@ ${REDIS_LEASE_LUA}
   -- Read job data to extract dedup.id before deleting
   local job_data = redis.call('HGET', data_key, job_id)
   if job_data then
-    local ok, job = pcall(cjson.decode, job_data)
-    if ok and job and job.dedup and job.dedup.id then
+    local job = read_job_fields(job_data)
+    if job.dedup and job.dedup.id then
       local dkey = dedup_prefix .. job.dedup.id
       if redis.call('GET', dkey) == job_id then
         redis.call('DEL', dkey)
@@ -258,8 +257,8 @@ ${REDIS_LEASE_LUA}
       local id = ids[i]
       local d = redis.call('HGET', data_key, id)
       if d then
-        local ok, job = pcall(cjson.decode, d)
-        if ok and job and job.dedup and job.dedup.id then
+        local job = read_job_fields(d)
+        if job.dedup and job.dedup.id then
           local dkey = dedup_prefix .. job.dedup.id
           if redis.call('GET', dkey) == id then
             redis.call('DEL', dkey)
@@ -331,12 +330,15 @@ ${REDIS_LEASE_LUA}
     return 0
   end
 
+  -- Read everything before the first write: an error past it would leave
+  -- the job in no state, since a script does not roll back its writes.
+  local job = read_job_fields(job_data)
+  local overlay = read_job_overlay(overlay_key, job_id)
+
   -- Remove from active
   redis.call('HDEL', active_key, job_id)
 
   -- Increment attempts without rewriting opaque job JSON.
-  local job = cjson.decode(job_data)
-  local overlay = read_job_overlay(overlay_key, job_id)
   overlay.attempts = (overlay.attempts or job.attempts or 0) + 1
   write_job_overlay(overlay_key, job_id, overlay)
 
@@ -395,7 +397,7 @@ ${REDIS_JOB_STORAGE_LUA}
     if active.acquiredAt < stalled_cutoff then
       local job_data = redis.call('HGET', data_key, job_id)
       if job_data then
-        local job = cjson.decode(job_data)
+        local job = read_job_fields(job_data)
         local overlay = read_job_overlay(overlay_key, job_id)
         local current_stalled_count = overlay.stalledCount or job.stalledCount or 0
 
 
@@ -1393,6 +1393,48 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
     assert.equal(retried!.attempts, 1)
   })
 
+  test('payloads that Redis Lua cannot decode do not block the queue', async ({ assert }) => {
+    const adapter = await options.createAdapter()
+    adapter.setWorkerId('worker-1')
+
+    // JSON.parse accepts both, Redis cjson rejects them.
+    let deep: unknown = 'leaf'
+    for (let depth = 0; depth < 1100; depth++) deep = [deep]
+    const payloads: Record<string, unknown> = { 'lone-surrogate': '\ud800', 'deep-nesting': deep }
+
+    for (const [id, payload] of Object.entries(payloads)) {
+      await adapter.pushLaterOn('test-queue', { id, name: 'TestJob', payload, attempts: 0 }, 1)
+    }
+    await adapter.pushOn('test-queue', { id: 'healthy', name: 'TestJob', payload: {}, attempts: 0 })
+    await new Promise((resolve) => setTimeout(resolve, 10))
+
+    const acquired: AcquiredJob[] = []
+    for (let i = 0; i < 3; i++) {
+      const job = await adapter.popFrom('test-queue')
+      if (job) acquired.push(job)
+    }
+    assert.sameMembers(
+      acquired.map((job) => job.id),
+      ['lone-surrogate', 'deep-nesting', 'healthy']
+    )
+
+    for (const job of acquired.filter((acquiredJob) => acquiredJob.id !== 'healthy')) {
+      assert.deepEqual(job.payload, payloads[job.id])
+      assert.isTrue(await adapter.retryJob(job, 'test-queue'))
+
+      const record = await adapter.getJob(job.id, 'test-queue')
+      assert.equal(record?.status, 'pending')
+      assert.equal(record?.data.attempts, 1)
+    }
+
+    assert.isNotNull(await adapter.popFrom('test-queue'))
+    assert.isNotNull(await adapter.popFrom('test-queue'))
+    await new Promise((resolve) => setTimeout(resolve, 30))
+
+    const { recovered } = await adapter.recoverStalledJobs('test-queue', 10, 1, 100)
+    assert.equal(recovered, 3)
+  })
+
   test('delayed job becomes available after delay', async ({ assert }) => {
     const adapter = await options.createAdapter()
     adapter.setWorkerId('worker-1')