fix(tables): keep a batched write that already committed part of its work out of the retryable 503

Author: waleedlatif1

apps/sim/lib/table/api/write-contention.integration.ts

@@ -23,7 +23,9 @@ import {
   internalTableRowsErrorPolicy,
   v2TableRowsErrorPolicy,
 } from '@/lib/table/api/row-route-policies'
-import { insertRow, updateRowsByFilter } from '@/lib/table/rows/service'
+import { getDeleteSnapshotBatchSize, TABLE_LIMITS } from '@/lib/table/constants'
+import { TablePartialWriteError } from '@/lib/table/errors'
+import { deleteRowsByFilter, insertRow, updateRowsByFilter } from '@/lib/table/rows/service'
 import { getTableById } from '@/lib/table/service'
 import type { TableDefinition } from '@/lib/table/types'
 import { orchestrationErrorResponse } from '@/app/api/table/utils'
@@ -61,6 +63,26 @@ async function createTable(): Promise<TableDefinition> {
   return table
 }
 
+/** Seeds `count` rows whose id order and row order agree, created before any write's cutoff. */
+async function seedNotes(tableId: string, count: number): Promise<string[]> {
+  const ids = Array.from(
+    { length: count },
+    (_, index) => `${tableId}-${String(index).padStart(4, '0')}`
+  )
+  const createdAt = new Date(Date.now() - 60_000)
+  await db.insert(userTableRows).values(
+    ids.map((id, index) => ({
+      id,
+      tableId,
+      workspaceId,
+      data: { note: 'n' },
+      orderKey: `a${String(index).padStart(4, '0')}`,
+      createdAt,
+    }))
+  )
+  return ids
+}
+
 /** Runs `write` while a second session holds whatever `hold` locks, and returns what it threw. */
 async function failedUnderLock(
   hold: (holder: postgres.ReservedSql) => Promise<unknown>,
@@ -92,6 +114,13 @@ function postgresErrorContext(error: unknown): string | undefined {
   return undefined
 }
 
+/** Every surface leaves the failure to the route's generic 500 rather than inviting a retry. */
+async function expectNotRetryableOnAnySurface(error: unknown) {
+  expect(internalTableRowsErrorPolicy.project(error)).toBeNull()
+  expect(v2TableRowsErrorPolicy.render(error)).toBeNull()
+  expect(orchestrationErrorResponse(error)).toBeNull()
+}
+
 async function expectRetryableOnEverySurface(error: unknown) {
   const internal = internalTableRowsErrorPolicy.project(error)
   expect(internal?.status).toBe(503)
@@ -184,4 +213,57 @@ describe('table writes that lose a lock race', () => {
       expect(row.data).toEqual({ note: 'n' })
     }
   )
+
+  it('leaves a filtered update that loses a lock race after a committed page non-retryable', async () => {
+    const table = await createTable()
+    const ids = await seedNotes(table.id, TABLE_LIMITS.UPDATE_BATCH_SIZE + 1)
+    const lastId = ids[ids.length - 1]
+
+    const error = await failedUnderLock(
+      (holder) => holder`SELECT 1 FROM user_table_rows
+        WHERE id = ${lastId} AND table_id = ${table.id} AND workspace_id = ${workspaceId}
+        FOR UPDATE`,
+      () =>
+        updateRowsByFilter(
+          table,
+          {
+            filter: { note: 'n' },
+            data: { note: 'patched' },
+            secretProvenance: undefined,
+            capabilityGovernedUserId: null,
+          },
+          'contention-partial-update'
+        )
+    )
+
+    expect(error).toBeInstanceOf(TablePartialWriteError)
+    expect(error).toMatchObject({ committedCount: TABLE_LIMITS.UPDATE_BATCH_SIZE })
+    await expectNotRetryableOnAnySurface(error)
+  })
+
+  it('leaves a limited filtered delete that loses a lock race after a committed batch non-retryable', async () => {
+    const table = await createTable()
+    const batchSize = getDeleteSnapshotBatchSize()
+    const ids = await seedNotes(table.id, batchSize + 1)
+    const lastId = ids[ids.length - 1]
+
+    const error = await failedUnderLock(
+      (holder) => holder`SELECT 1 FROM user_table_rows
+        WHERE id = ${lastId} AND table_id = ${table.id} AND workspace_id = ${workspaceId}
+        FOR UPDATE`,
+      () =>
+        deleteRowsByFilter(
+          table,
+          { filter: { note: 'n' }, limit: batchSize + 1 },
+          'contention-partial-delete'
+        )
+    )
+
+    expect(error).toBeInstanceOf(TablePartialWriteError)
+    expect(error).toMatchObject({ committedCount: batchSize })
+    await expectNotRetryableOnAnySurface(error)
+    const [{ remaining }] = await control<{ remaining: number }[]>`SELECT count(*)::int AS remaining
+      FROM user_table_rows WHERE table_id = ${table.id} AND workspace_id = ${workspaceId}`
+    expect(remaining).toBe(1)
+  })
 })

apps/sim/lib/table/api/write-contention.ts

@@ -1,36 +1,33 @@
 import { createLogger } from '@sim/logger'
 import { getPostgresCancellationReason, getPostgresErrorCode } from '@sim/utils/errors'
 import { ADMISSION_RETRY_AFTER_SECONDS } from '@/lib/core/admission/transient-failure'
+import { isTableLockRace, TablePartialWriteError } from '@/lib/table/errors'
 
 const logger = createLogger('TableWriteContention')
 
-/** `lock_not_available` (a `lock_timeout` fired) and `deadlock_detected`. */
-const WRITE_CONTENTION_SQLSTATES = new Set(['55P03', '40P01'])
-
 export const TABLE_WRITE_CONTENTION_MESSAGE =
   'Table is busy with other writes. Try again in a few seconds.'
 
 export const TABLE_WRITE_CONTENTION_RETRY_AFTER_SECONDS = ADMISSION_RETRY_AFTER_SECONDS
 
 /**
  * Whether a table operation failed because its transaction lost a lock race, which callers should
- * answer with a retryable 503 instead of a generic 500.
+ * answer with a retryable 503 rather than a generic 500.
  *
  * Every row write locks its table's `user_table_definitions` row (the `row_count` and
  * `rows_version` triggers, the latter at COMMIT), so a transaction that holds it longer than the
  * table's 3 s `lock_timeout` — typically one stuck waiting on a synchronous standby, which keeps its
- * locks — fails the writers queued behind it. The failed transaction rolled back. A bulk write that
- * commits page by page may have applied earlier pages, but every such write re-applies the same
- * patch or filter, so retrying it is safe.
+ * locks — fails the writers queued behind it. Those writers rolled back, so retrying them is safe.
+ * A batched write that had already committed part of its work is a {@link TablePartialWriteError}
+ * instead, and is not retryable as a whole.
  *
  * Logs the failure, because route builders do not log errors their policy projects. Only the
  * SQLSTATE is logged: the driver's message carries the query's bound parameters.
  */
 export function isTableWriteContention(error: unknown): boolean {
-  const code = getPostgresErrorCode(error)
-  if (!code || !WRITE_CONTENTION_SQLSTATES.has(code)) return false
+  if (error instanceof TablePartialWriteError || !isTableLockRace(error)) return false
   logger.warn('Table operation lost a lock race', {
-    sqlstate: code,
+    sqlstate: getPostgresErrorCode(error),
     reason: getPostgresCancellationReason(error) ?? null,
   })
   return true

apps/sim/lib/table/errors.ts

@@ -1,5 +1,30 @@
+import { getPostgresErrorCode } from '@sim/utils/errors'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 
+/** `lock_not_available` (a `lock_timeout` fired) and `deadlock_detected`. */
+const LOCK_RACE_SQLSTATES = new Set(['55P03', '40P01'])
+
+/** Whether a table write's transaction failed because it lost a lock race, and so rolled back. */
+export function isTableLockRace(error: unknown): boolean {
+  const code = getPostgresErrorCode(error)
+  return code !== undefined && LOCK_RACE_SQLSTATES.has(code)
+}
+
+/**
+ * A write that commits batch by batch lost a lock race after at least one batch committed. Unlike
+ * a single rolled-back transaction it is not retryable as a whole: a re-run selects its rows again,
+ * so a limited filtered delete would remove more than its limit.
+ */
+export class TablePartialWriteError extends Error {
+  constructor(
+    readonly committedCount: number,
+    cause: unknown
+  ) {
+    super(`Table write failed after ${committedCount} rows were committed`, { cause })
+    this.name = 'TablePartialWriteError'
+  }
+}
+
 /** A disabled TTL feature, distinct from malformed column input. */
 export class TableRowTtlDisabledError extends OrchestrationError {
   readonly detailCode = 'TABLE_ROW_TTL_DISABLED'