fix(table): reclaim a cascade lock a timed-out acquire may have taken (#7680)

A client-side timeout does not mean Redis declined the SET. The command can
still be parked in the offline queue and take the lock once the connection
completes, leaving the row's cascade held for the full 30s TTL by an owner that
already threw — no heartbeat, no release. Every other cell task for that row
then reads `contended` and bails on the silent path, so one stalled connection
quietly drops later cells rather than just failing the one run.

Releasing after a failed acquire is what the Redlock algorithm prescribes:
a client that fails to acquire unlocks the instances anyway, including ones it
believed it had not locked. Both preconditions the option documents hold here —
`ownerId` is the cell task's unique `executionId`, and a throw means `fn` never
runs, so the reclaim cannot cut under a caller still doing work.

Adds the cascade lock's first tests, covering acquire, contention, reclaim,
release on throw, and heartbeat teardown.

Author: waleedlatif1

apps/sim/lib/table/cascade-lock.test.ts

@@ -0,0 +1,89 @@
+/**
+ * @vitest-environment node
+ */
+import { redisConfigMockFns } from '@sim/testing'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { cascadeLockKey, withCascadeLock } from '@/lib/table/cascade-lock'
+
+const TABLE_ID = 'tbl_1'
+const ROW_ID = 'row_1'
+const OWNER_ID = 'exec-1'
+
+describe('withCascadeLock', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    redisConfigMockFns.mockAcquireLock.mockResolvedValue(true)
+    redisConfigMockFns.mockReleaseLock.mockResolvedValue(true)
+    redisConfigMockFns.mockExtendLock.mockResolvedValue(true)
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('runs the work and releases under the owner that took the lock', async () => {
+    const fn = vi.fn().mockResolvedValue('done')
+
+    await expect(withCascadeLock(TABLE_ID, ROW_ID, OWNER_ID, fn)).resolves.toEqual({
+      status: 'acquired',
+      result: 'done',
+    })
+    expect(fn).toHaveBeenCalledOnce()
+    expect(redisConfigMockFns.mockReleaseLock).toHaveBeenCalledWith(
+      cascadeLockKey(TABLE_ID, ROW_ID),
+      OWNER_ID
+    )
+  })
+
+  it('skips the work when another task holds the row', async () => {
+    redisConfigMockFns.mockAcquireLock.mockResolvedValue(false)
+    const fn = vi.fn()
+
+    await expect(withCascadeLock(TABLE_ID, ROW_ID, OWNER_ID, fn)).resolves.toEqual({
+      status: 'contended',
+    })
+    expect(fn).not.toHaveBeenCalled()
+    // Nothing was taken, so nothing may be deleted — the holder still owns it.
+    expect(redisConfigMockFns.mockReleaseLock).not.toHaveBeenCalled()
+  })
+
+  it('reclaims a lock a timed-out acquire may have taken', async () => {
+    // A client-side timeout does not mean Redis declined the SET: the command
+    // can still land and hold the row for the full TTL under an owner that
+    // already threw, silently starving every later cell task for that row.
+    redisConfigMockFns.mockAcquireLock.mockRejectedValue(new Error('Command timed out'))
+    const fn = vi.fn()
+
+    await expect(withCascadeLock(TABLE_ID, ROW_ID, OWNER_ID, fn)).rejects.toThrow(
+      'Command timed out'
+    )
+    expect(fn).not.toHaveBeenCalled()
+    expect(redisConfigMockFns.mockAcquireLock).toHaveBeenCalledWith(
+      cascadeLockKey(TABLE_ID, ROW_ID),
+      OWNER_ID,
+      expect.any(Number),
+      { reclaimOnFailure: true }
+    )
+  })
+
+  it('releases the lock when the work throws', async () => {
+    const fn = vi.fn().mockRejectedValue(new Error('boom'))
+
+    await expect(withCascadeLock(TABLE_ID, ROW_ID, OWNER_ID, fn)).rejects.toThrow('boom')
+    expect(redisConfigMockFns.mockReleaseLock).toHaveBeenCalledWith(
+      cascadeLockKey(TABLE_ID, ROW_ID),
+      OWNER_ID
+    )
+  })
+
+  it('stops the heartbeat once the work settles', async () => {
+    vi.useFakeTimers()
+    const fn = vi.fn().mockResolvedValue(undefined)
+
+    await withCascadeLock(TABLE_ID, ROW_ID, OWNER_ID, fn)
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    // A heartbeat outliving the work would keep extending a lock nobody holds.
+    expect(redisConfigMockFns.mockExtendLock).not.toHaveBeenCalled()
+  })
+})

apps/sim/lib/table/cascade-lock.ts

@@ -40,7 +40,21 @@ export async function withCascadeLock<T>(
   fn: () => Promise<T>
 ): Promise<{ status: 'acquired'; result: T } | { status: 'contended' }> {
   const key = cascadeLockKey(tableId, rowId)
-  const acquired = await acquireLock(key, ownerId, LOCK_TTL_SECONDS)
+  const acquired = await acquireLock(key, ownerId, LOCK_TTL_SECONDS, {
+    /**
+     * A client-side timeout does not mean Redis declined the SET — the command
+     * can still be sitting in the offline queue and take the lock once the
+     * connection completes, leaving the row's cascade held for the full TTL by
+     * an owner that already threw, with no heartbeat and no release. Every other
+     * cell task for that row then reads `contended` and bails on the silent
+     * path, so one stalled connection quietly drops later cells too.
+     *
+     * Both preconditions hold here: `ownerId` is the cell task's `executionId`,
+     * unique to this holder, and a throw means `fn` never runs, so freeing a
+     * lock this call may have taken cannot cut under a caller still working.
+     */
+    reclaimOnFailure: true,
+  })
   if (!acquired) return { status: 'contended' }
 
   const heartbeat = setInterval(() => {