fix(tables): address review feedback on cell retry resilience

- retryTransient: re-check the abort signal after the backoff sleep so a
  cancellation during sleep stops the next attempt (don't run/return work for
  an already-cancelled task).
- isRetryableRedisError: walk the .cause chain (mirroring the infra
  classifier) so wrapped Redis timeouts are recognized; drop "Connection is in
  subscriber mode" — that's a connection-state programming error, not a
  transient drop, and would just fail identically every retry.
- cascade-lock: stop wrapping acquireLock in retryTransient. acquireLock is a
  non-idempotent SET NX, so retrying after a timed-out-but-applied first SET
  returns false (key already ours) and yields a false `contended` that skips
  the cascade. A transient Redis blip here just fails the run before pickup
  (no stranded cell); the dispatcher re-drives it.
- Tests: cause-chain Redis match, subscriber-mode exclusion, abort-during-sleep.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: TheodoreSpeaks

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

@@ -1,7 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { acquireLock, extendLock, releaseLock } from '@/lib/core/config/redis'
-import { retryTransient } from '@/lib/table/retry-transient'
 
 const logger = createLogger('TableCascadeLock')
 
@@ -41,11 +40,12 @@ export async function withCascadeLock<T>(
   fn: () => Promise<T>
 ): Promise<{ status: 'acquired'; result: T } | { status: 'contended' }> {
   const key = cascadeLockKey(tableId, rowId)
-  // A timed-out/dropped Redis command here throws before the cell is picked up;
-  // retry so a transient Redis blip doesn't fail the run outright.
-  const acquired = await retryTransient('cascade acquireLock', () =>
-    acquireLock(key, ownerId, LOCK_TTL_SECONDS)
-  )
+  // NOT wrapped in retryTransient: acquireLock is a non-idempotent `SET NX`, so
+  // a retry after a timed-out-but-applied first SET would see the key already
+  // present and return false — a false `contended` that skips the cascade. A
+  // transient Redis blip here just fails the run before pickup (no stranded
+  // cell); the dispatcher/re-drive recovers it.
+  const acquired = await acquireLock(key, ownerId, LOCK_TTL_SECONDS)
   if (!acquired) return { status: 'contended' }
 
   const heartbeat = setInterval(() => {

apps/sim/lib/table/retry-transient.test.ts

@@ -27,6 +27,14 @@ describe('isRetryableCellError', () => {
     expect(isRetryableCellError(new Error('Connection is closed'))).toBe(true)
   })
 
+  it('matches a Redis timeout wrapped in a cause chain', () => {
+    expect(isRetryableCellError(new Error('outer wrapper', { cause: redisTimeout() }))).toBe(true)
+  })
+
+  it('does not retry Redis connection-state programming errors', () => {
+    expect(isRetryableCellError(new Error('Connection is in subscriber mode'))).toBe(false)
+  })
+
   it('does not retry application/logic errors', () => {
     expect(isRetryableCellError(new Error('row not found'))).toBe(false)
   })
@@ -80,4 +88,19 @@ describe('retryTransient', () => {
     expect(fn).toHaveBeenCalledTimes(1)
     expect(mockSleep).not.toHaveBeenCalled()
   })
+
+  it('stops retrying if the signal aborts during backoff sleep', async () => {
+    const controller = new AbortController()
+    // Cancellation fires mid-backoff, after the first failure but before the
+    // next attempt would run.
+    mockSleep.mockImplementationOnce(async () => {
+      controller.abort()
+    })
+    const fn = vi.fn().mockRejectedValue(connError())
+    await expect(retryTransient('t', fn, { signal: controller.signal })).rejects.toThrow(
+      'Failed query'
+    )
+    expect(fn).toHaveBeenCalledTimes(1)
+    expect(mockSleep).toHaveBeenCalledTimes(1)
+  })
 })

apps/sim/lib/table/retry-transient.ts

@@ -11,16 +11,26 @@ const logger = createLogger('TableRetryTransient')
  *  connection drop without risking duplicate side effects. */
 const DEFAULT_MAX_ATTEMPTS = 4
 
+/** Redis-side equivalents of a dropped socket. Excludes connection-STATE
+ *  programming errors (e.g. "Connection is in subscriber mode"), which are
+ *  misconfigurations that would just fail identically on every retry. */
+const RETRYABLE_REDIS_MESSAGE = /Command timed out|Connection is closed|Stream isn't writeable/i
+
 /**
  * ioredis surfaces command timeouts and severed connections as plain `Error`s
  * with no `code`/`errno`, so the SQLSTATE/errno-based
  * {@link isRetryableInfrastructureError} classifier misses them. Match those by
- * message instead — these are the Redis-side equivalents of a dropped socket.
+ * message instead, walking the `.cause` chain (depth-bounded) so a wrapped
+ * Redis failure is still recognized — mirroring how the infra classifier walks
+ * causes.
  */
 function isRetryableRedisError(error: unknown): boolean {
-  return /Command timed out|Connection is closed|Stream isn't writeable|Connection is in subscriber mode/i.test(
-    getErrorMessage(error)
-  )
+  let current: unknown = error
+  for (let depth = 0; depth < 10 && current instanceof Error; depth++) {
+    if (RETRYABLE_REDIS_MESSAGE.test(current.message)) return true
+    current = current.cause
+  }
+  return false
 }
 
 /**
@@ -68,6 +78,10 @@ export async function retryTransient<T>(
         { error: getErrorMessage(error) }
       )
       await sleep(waitMs)
+      // Re-check after backoff: if cancellation fired during the sleep, don't
+      // run another attempt (and don't return a success from work that started
+      // after the task was already cancelled).
+      if (options.signal?.aborted) throw error
     }
   }
 }