fix(locking-redis): order multi-key acquires, atomic releaseAll, same-owner reentrancy under awaitQueue

.changeset/locking-redis-order-release-reentrancy.md

@@ -0,0 +1,5 @@
+---
+"@medusajs/locking-redis": patch
+---
+
+fix(locking-redis): acquire multi-key locks in a stable deduplicated order, make releaseAll an atomic compare-and-delete, and let a named owner re-enter its own lock under awaitQueue

packages/modules/providers/locking-redis/integration-tests/__tests__/index.spec.ts

@@ -2,6 +2,8 @@ import { ILockingModule } from "@medusajs/framework/types"
 import { Modules, promiseAll } from "@medusajs/framework/utils"
 import { moduleIntegrationTestRunner } from "@medusajs/test-utils"
 import { setTimeout } from "node:timers/promises"
+import { Redis } from "ioredis"
+import { RedisLockingProvider } from "../../src/services/redis-lock"
 
 jest.setTimeout(5000)
 
@@ -159,6 +161,149 @@ moduleIntegrationTestRunner<ILockingModule>({
         const release = await service.release(keyToLock)
         expect(release).toBe(true)
       })
+
+      it("should re-enter a lock held by the same owner when awaitQueue is set", async () => {
+        await service.acquire("reentrant_key", {
+          ownerId: "owner_reentry",
+          expire: 10,
+        })
+
+        // Against the unfixed provider the awaitQueue branch never runs the
+        // same-owner check, so this call backs off against its own lock and the
+        // test dies on the jest timeout instead of resolving. That asymmetry
+        // with the awaitQueue: false path above is the proof.
+        await expect(
+          service.acquire("reentrant_key", {
+            ownerId: "owner_reentry",
+            expire: 10,
+            awaitQueue: true,
+          })
+        ).resolves.toBeUndefined()
+
+        expect(
+          await service.release("reentrant_key", { ownerId: "owner_reentry" })
+        ).toBe(true)
+      })
+
+      it("should scope releaseAll to the given owner", async () => {
+        await service.acquire("ra_mine", { ownerId: "owner_a", expire: 10 })
+        await service.acquire("ra_theirs", { ownerId: "owner_b", expire: 10 })
+
+        await service.releaseAll({ ownerId: "owner_a" })
+
+        await expect(
+          service.acquire("ra_mine", { ownerId: "owner_c", expire: 10 })
+        ).resolves.toBeUndefined()
+
+        await expect(
+          service.acquire("ra_theirs", { ownerId: "owner_c", expire: 10 })
+        ).rejects.toThrow(`Failed to acquire lock for key "ra_theirs"`)
+
+        expect(await service.release("ra_mine", { ownerId: "owner_c" })).toBe(
+          true
+        )
+        expect(await service.release("ra_theirs", { ownerId: "owner_b" })).toBe(
+          true
+        )
+      })
+
+      it("should not deadlock when two callers request the same keys in opposite orders", async () => {
+        // The suite's module service serves every caller from a single ioredis
+        // connection, so two callers started with Promise.all never interleave:
+        // the first one queues both of its commands before the second one is
+        // scheduled, takes both keys, and an argument-order implementation
+        // looks safe. Each caller here therefore gets its own connection, and a
+        // delay is injected before its second acquisition so both callers hold
+        // their first key before either asks for its second.
+        //
+        // Under argument order that is the ABBA interleave: caller one holds
+        // "abba_a", caller two holds "abba_b", and both then spin on the key
+        // the other holds. Under the sorted order both callers target "abba_a"
+        // first, so only one of them is ever in the critical section and the
+        // injected delay is inconsequential.
+        const redisUrl = process.env.REDIS_URL ?? "redis://localhost:6379"
+        const clients: Redis[] = []
+
+        const makeCaller = async (keys: string[], ownerId: string) => {
+          // defineCommand attaches acquireLock to the instance at runtime and
+          // the provider's own client type is not exported, so restate the one
+          // command this test wraps.
+          const client = new Redis(redisUrl, {
+            lazyConnect: true,
+          }) as Redis & {
+            acquireLock: (
+              key: string,
+              ownerId: string,
+              ttl: number
+            ) => Promise<number>
+          }
+          clients.push(client)
+          await client.connect()
+
+          const provider = new RedisLockingProvider(
+            { redisClient: client, prefix: "medusa_lock:" },
+            {}
+          )
+
+          // Fault injection lives in the test; the provider is untouched.
+          const acquireLock = client.acquireLock.bind(client)
+          let calls = 0
+          client.acquireLock = async (key, owner, ttl) => {
+            if (++calls === 2) {
+              await setTimeout(150)
+            }
+            return acquireLock(key, owner, ttl)
+          }
+
+          return async () => {
+            await provider.acquire(keys, {
+              ownerId,
+              expire: 10,
+              awaitQueue: true,
+            })
+            expect(await provider.release(keys, { ownerId })).toBe(true)
+          }
+        }
+
+        try {
+          const callerOne = await makeCaller(
+            ["abba_a", "abba_b"],
+            "owner_abba_one"
+          )
+          const callerTwo = await makeCaller(
+            ["abba_b", "abba_a"],
+            "owner_abba_two"
+          )
+
+          const settled = Promise.all([callerOne(), callerTwo()])
+          // The teardown below aborts the in-flight command of a caller that is
+          // still retrying; swallow that follow-up rejection.
+          settled.catch(() => {})
+
+          // awaitQueue has no overall deadline, so a deadlock here would hang
+          // until the jest timeout killed the whole file. Racing a deadline
+          // turns it into a readable assertion failure instead.
+          const deadlocked =
+            "deadlocked: both callers are still waiting on each other"
+          const outcome = await Promise.race([
+            settled,
+            setTimeout(2000, deadlocked, { ref: false }),
+          ])
+
+          expect(outcome).not.toBe(deadlocked)
+        } finally {
+          // Disconnect first: it stops a still-spinning retry loop before the
+          // cleanup runs, so a deadlocked run cannot leave the two keys held
+          // for their whole TTL and fail the next run for the wrong reason.
+          clients.forEach((client) => client.disconnect())
+          await service.release(["abba_a", "abba_b"], {
+            ownerId: "owner_abba_one",
+          })
+          await service.release(["abba_a", "abba_b"], {
+            ownerId: "owner_abba_two",
+          })
+        }
+      })
     })
 
     it("should release lock in case of failure", async () => {

packages/modules/providers/locking-redis/src/services/__tests__/redis-lock.spec.ts

@@ -66,3 +66,108 @@ describe("RedisLockingProvider Jitter", () => {
     expect(secondDelay).toBeLessThanOrEqual(200)
   })
 })
+
+describe("RedisLockingProvider acquire ordering and release", () => {
+  let provider: RedisLockingProvider
+  const redisClientMock = {
+    defineCommand: jest.fn(),
+    acquireLock: jest.fn(),
+    releaseLock: jest.fn(),
+    scan: jest.fn(),
+    pipeline: jest.fn(),
+  }
+
+  beforeEach(() => {
+    provider = new RedisLockingProvider(
+      {
+        redisClient: redisClientMock as any,
+        prefix: "test:",
+      },
+      {
+        defaultRetryInterval: 10,
+        backoffFactor: 2,
+      } as any
+    )
+    jest.clearAllMocks()
+  })
+
+  it("should deduplicate and sort keys and acquire them sequentially", async () => {
+    redisClientMock.acquireLock.mockResolvedValue(1)
+
+    await provider.acquire(["b", "a", "b"], { ownerId: "owner_1" })
+
+    // A stable, duplicate-free order is what prevents two callers requesting
+    // the same keys in different orders from deadlocking against each other.
+    // The third argument is the ttl: the script no longer takes awaitQueue.
+    expect(redisClientMock.acquireLock.mock.calls).toEqual([
+      ["test:a", "owner_1", 0],
+      ["test:b", "owner_1", 0],
+    ])
+  })
+
+  it("should stop at the first key it cannot take", async () => {
+    redisClientMock.acquireLock.mockImplementation((key: string) =>
+      Promise.resolve(key === "test:b" ? 0 : 1)
+    )
+
+    await expect(
+      provider.acquire(["c", "a", "b"], { ownerId: "owner_1" })
+    ).rejects.toThrow('Failed to acquire lock for key "b"')
+
+    // Sorted order means "a" then "b"; "c" is never attempted because the
+    // keys are taken one after the other rather than in parallel.
+    expect(redisClientMock.acquireLock.mock.calls).toEqual([
+      ["test:a", "owner_1", 0],
+      ["test:b", "owner_1", 0],
+    ])
+  })
+
+  it("should not back off when the first attempt succeeds under awaitQueue", async () => {
+    redisClientMock.acquireLock.mockResolvedValue(1)
+
+    await expect(
+      provider.acquire("k", { ownerId: "owner_1", awaitQueue: true })
+    ).resolves.toBeUndefined()
+
+    // The awaitQueue branch has to return on the first success instead of
+    // always sleeping once. Whether an owner re-entering its own lock is
+    // given that success is decided by the Lua script, so the re-entrancy
+    // fix itself is covered by the integration suite against a real Redis.
+    expect(redisClientMock.acquireLock).toHaveBeenCalledTimes(1)
+    expect(setTimeout).not.toHaveBeenCalled()
+  })
+
+  it("should release scanned keys through a single atomic pipeline", async () => {
+    redisClientMock.scan.mockResolvedValueOnce(["0", ["test:x", "test:y"]])
+
+    // Neither mock exposes get or unlink: the previous read-then-delete
+    // shape needed both, so it cannot pass this test.
+    const pipelineMock = {
+      releaseLock: jest.fn().mockReturnThis(),
+      exec: jest.fn().mockResolvedValue([]),
+    }
+    redisClientMock.pipeline.mockReturnValue(pipelineMock)
+
+    await provider.releaseAll({ ownerId: "owner_1" })
+
+    expect(redisClientMock.scan).toHaveBeenCalledWith(
+      "0",
+      "MATCH",
+      "test:*",
+      "COUNT",
+      100
+    )
+    // Scanned keys are already prefixed and go to the script as-is, and the
+    // owner check now happens inside the delete instead of before it.
+    expect(pipelineMock.releaseLock.mock.calls).toEqual([
+      ["test:x", "owner_1"],
+      ["test:y", "owner_1"],
+    ])
+    expect(redisClientMock.pipeline).toHaveBeenCalledTimes(1)
+    expect(pipelineMock.exec).toHaveBeenCalledTimes(1)
+    expect(redisClientMock).not.toHaveProperty("get")
+    expect(redisClientMock).not.toHaveProperty("unlink")
+    expect(pipelineMock).not.toHaveProperty("get")
+    expect(pipelineMock).not.toHaveProperty("unlink")
+  })
+})