fix(tables): bound delete trigger snapshots

Author: j15z

apps/sim/background/cleanup-table-row-ttl.test.ts

@@ -32,6 +32,7 @@ vi.mock('@sim/db', () => ({
 
 vi.mock('@trigger.dev/sdk', () => ({ task: mockTask }))
 vi.mock('@/lib/table/events', () => ({ signalTableRowsChanged: mockSignalTableRowsChanged }))
+vi.mock('@/lib/table/constants', () => ({ getDeleteSnapshotBatchSize: () => 500 }))
 vi.mock('@/lib/table/service', () => ({ withLockedTable: mockWithLockedTable }))
 vi.mock('@/lib/table/ttl-availability', () => ({
   isTableRowTtlEnabled: mockIsTableRowTtlEnabled,

apps/sim/background/cleanup-table-row-ttl.ts

@@ -5,6 +5,7 @@ import { task } from '@trigger.dev/sdk'
 import { sql } from 'drizzle-orm'
 import { asOrchestrationError } from '@/lib/core/orchestration/types'
 import { getColumnId } from '@/lib/table/column-keys'
+import { getDeleteSnapshotBatchSize } from '@/lib/table/constants'
 import { signalTableRowsChanged } from '@/lib/table/events'
 import { assertRowDelete, TableLockedError } from '@/lib/table/mutation-locks'
 import type { DbTransaction } from '@/lib/table/planner'
@@ -17,7 +18,6 @@ import type { RowData, TableSchema } from '@/lib/table/types'
 const logger = createLogger('CleanupTableRowTtl')
 const cleanupDb = dbFor('cleanup')
 
-const TTL_CLEANUP_BATCH_SIZE = 500
 const TTL_CLEANUP_MAX_BATCHES = 100
 
 interface ExpiredTtlTableRef {
@@ -96,12 +96,12 @@ async function listExpiredTtlTables(nowEpochSeconds: number): Promise<ExpiredTtl
   return Array.isArray(rows) ? rows : []
 }
 
-function parseDeletedBatch(rows: unknown): DeletedTtlRows {
+function parseDeletedBatch(rows: unknown, batchSize: number): DeletedTtlRows {
   if (!Array.isArray(rows)) {
     throw new Error('Table row TTL cleanup did not return deleted rows')
   }
   const deletedRows = rows as Array<{ id?: unknown; data?: unknown; createdAt?: unknown }>
-  if (deletedRows.length > TTL_CLEANUP_BATCH_SIZE) {
+  if (deletedRows.length > batchSize) {
     throw new Error('Table row TTL cleanup returned an invalid deleted count')
   }
   const parsed = deletedRows.map((row) => {
@@ -129,6 +129,7 @@ async function deleteExpiredTableRowBatch(
   workspaceId: string,
   columnKey: string,
   nowEpochSeconds: number,
+  batchSize: number,
   after?: TtlCleanupCursor
 ): Promise<DeletedTtlRows> {
   const rows = await trx.execute<{ id: string; data: RowData; createdAt: string }>(sql`
@@ -145,7 +146,7 @@ async function deleteExpiredTableRowBatch(
         AND jsonb_typeof(table_row.data->${columnKey}) = 'number'
         AND (table_row.data->>${columnKey})::numeric <= ${nowEpochSeconds}
       ORDER BY table_row.created_at, table_row.id
-      LIMIT ${TTL_CLEANUP_BATCH_SIZE}
+      LIMIT ${batchSize}
       FOR UPDATE OF table_row SKIP LOCKED
     ), deleted AS (
       DELETE FROM ${userTableRows} AS table_row
@@ -160,12 +161,13 @@ async function deleteExpiredTableRowBatch(
     FROM deleted
     ORDER BY "createdAt", id
   `)
-  return parseDeletedBatch(rows)
+  return parseDeletedBatch(rows, batchSize)
 }
 
 async function deleteExpiredRowsForTable(
   ref: ExpiredTtlTableRef,
   nowEpochSeconds: number,
+  batchSize: number,
   after?: TtlCleanupCursor
 ): Promise<DeletedTtlBatch> {
   try {
@@ -190,6 +192,7 @@ async function deleteExpiredRowsForTable(
           table.workspaceId,
           getColumnId(ttlColumn),
           nowEpochSeconds,
+          batchSize,
           after
         )
         return {
@@ -202,7 +205,7 @@ async function deleteExpiredRowsForTable(
       { expectedWorkspaceId: ref.workspaceId }
     )
     if (batch.attempted && batch.rows.length > 0) {
-      void fireTableTrigger(
+      await fireTableTrigger(
         ref.id,
         batch.tableName,
         'delete',
@@ -232,6 +235,7 @@ export async function runCleanupTableRowTtl(
   }
 
   const nowEpochSeconds = Math.floor(Date.now() / 1000)
+  const batchSize = getDeleteSnapshotBatchSize()
   const tableRefs = await listExpiredTtlTables(nowEpochSeconds)
   const tableStates: TtlTableCleanupState[] = tableRefs.map((ref) => ({
     ref,
@@ -250,7 +254,12 @@ export async function runCleanupTableRowTtl(
       if (state.complete) continue
       if (batches === TTL_CLEANUP_MAX_BATCHES || signal?.aborted) break
 
-      const batch = await deleteExpiredRowsForTable(state.ref, nowEpochSeconds, state.after)
+      const batch = await deleteExpiredRowsForTable(
+        state.ref,
+        nowEpochSeconds,
+        batchSize,
+        state.after
+      )
       if (!batch.attempted) {
         state.complete = true
         continue
@@ -260,7 +269,7 @@ export async function runCleanupTableRowTtl(
       deleted += batch.deleted
       state.deleted += batch.deleted
       state.after = batch.cursor ?? undefined
-      if (batch.deleted < TTL_CLEANUP_BATCH_SIZE) state.complete = true
+      if (batch.deleted < batchSize) state.complete = true
     }
   }
 

apps/sim/lib/table/__tests__/service-filter-threading.test.ts

@@ -234,6 +234,31 @@ describe('delete trigger dispatch', () => {
       'req-delete-many'
     )
   })
+
+  it('dispatches byte-bounded ID-delete snapshots before loading the next batch', async () => {
+    setEnv({
+      TABLE_MAX_ROW_SIZE_BYTES: TABLE_LIMITS.DELETE_SNAPSHOT_BATCH_MAX_BYTES * 2,
+    })
+    dbChainMockFns.returning
+      .mockResolvedValueOnce([{ id: 'row-1', data: { name: 'Ada' } }])
+      .mockResolvedValueOnce([{ id: 'row-2', data: { name: 'Grace' } }])
+
+    try {
+      await deleteRowsByIds(
+        TABLE,
+        { tableId: TABLE.id, workspaceId: TABLE.workspaceId, rowIds: ['row-1', 'row-2'] },
+        'req-delete-bounded'
+      )
+    } finally {
+      setEnv({ TABLE_MAX_ROW_SIZE_BYTES: undefined })
+    }
+
+    expect(mockFireTableTrigger).toHaveBeenCalledTimes(2)
+    expect(mockFireTableTrigger.mock.calls[0][3]).toEqual([{ id: 'row-1', data: { name: 'Ada' } }])
+    expect(mockFireTableTrigger.mock.calls[1][3]).toEqual([
+      { id: 'row-2', data: { name: 'Grace' } },
+    ])
+  })
 })
 
 describe('bulk update/delete limited-subset ordering', () => {

apps/sim/lib/table/constants.test.ts

@@ -39,6 +39,7 @@ declare module '@/lib/table/constants?constants-test' {
 
 import {
   getBillingDisabledTableLimits,
+  getDeleteSnapshotBatchSize,
   getMaxPageBytes,
   TABLE_LIMITS,
 } from '@/lib/table/constants?constants-test'
@@ -86,3 +87,23 @@ describe('getMaxPageBytes', () => {
     expect(getMaxPageBytes()).toBe(2 * 1024 * 1024)
   })
 })
+
+describe('getDeleteSnapshotBatchSize', () => {
+  beforeEach(() => {
+    for (const key of Object.keys(mockEnv)) delete mockEnv[key]
+  })
+
+  it('derives a worst-case row cap from the delete snapshot byte budget', () => {
+    expect(getDeleteSnapshotBatchSize()).toBe(
+      Math.floor(TABLE_LIMITS.DELETE_SNAPSHOT_BATCH_MAX_BYTES / TABLE_LIMITS.MAX_ROW_SIZE_BYTES)
+    )
+  })
+
+  it('always processes one row and never exceeds the delete row-count cap', () => {
+    mockEnv.TABLE_MAX_ROW_SIZE_BYTES = String(TABLE_LIMITS.DELETE_SNAPSHOT_BATCH_MAX_BYTES * 2)
+    expect(getDeleteSnapshotBatchSize()).toBe(1)
+
+    mockEnv.TABLE_MAX_ROW_SIZE_BYTES = '1'
+    expect(getDeleteSnapshotBatchSize()).toBe(TABLE_LIMITS.DELETE_BATCH_SIZE)
+  })
+})

apps/sim/lib/table/constants.ts

@@ -39,6 +39,12 @@ export const TABLE_LIMITS = {
   UPDATE_BATCH_SIZE: 100,
   /** Batch size for bulk delete operations */
   DELETE_BATCH_SIZE: 1000,
+  /**
+   * Maximum serialized row-data bytes returned from one committed delete for
+   * post-commit trigger dispatch. The row-count cap is derived from this budget
+   * and the configured maximum row size before a DELETE materializes snapshots.
+   */
+  DELETE_SNAPSHOT_BATCH_MAX_BYTES: 32 * 1024 * 1024, // 32MB
   /** Maximum rows per batch insert */
   MAX_BATCH_INSERT_SIZE: 1000,
   /** Maximum rows per bulk update/delete operation */
@@ -149,6 +155,21 @@ export function getMaxRowSizeBytes(): number {
   })
 }
 
+/**
+ * Maximum rows one delete may materialize with their JSON data for trigger
+ * dispatch. Uses the worst-case configured row size so every batch has an
+ * explicit byte bound before PostgreSQL returns it to the app process.
+ */
+export function getDeleteSnapshotBatchSize(): number {
+  return Math.max(
+    1,
+    Math.min(
+      TABLE_LIMITS.DELETE_BATCH_SIZE,
+      Math.floor(TABLE_LIMITS.DELETE_SNAPSHOT_BATCH_MAX_BYTES / getMaxRowSizeBytes())
+    )
+  )
+}
+
 export type PlanName = keyof typeof DEFAULT_TABLE_PLAN_LIMITS
 
 export interface TablePlanLimits {

apps/sim/lib/table/delete-runner.test.ts

@@ -87,17 +87,20 @@ describe('runTableDelete', () => {
     mockMarkJobReady.mockResolvedValue(true)
     mockMarkJobFailed.mockResolvedValue(undefined)
     mockDeletePageByIds.mockImplementation(
-      (
+      async (
         _t,
         _w,
         ids: string[],
         _proof,
         _revalidate,
-        onDeleted?: (rows: Array<{ id: string; data: Record<string, unknown> }>) => void
+        onDeleted?: (
+          rows: Array<{ id: string; data: Record<string, unknown> }>,
+          table?: typeof table
+        ) => void | Promise<void>
       ) => {
         const rows = ids.map((id) => ({ id, data: { title: id } }))
-        onDeleted?.(rows)
-        return Promise.resolve(rows.length)
+        await onDeleted?.(rows)
+        return rows.length
       }
     )
     mockBuildFilterClause.mockReturnValue({})
@@ -208,6 +211,34 @@ describe('runTableDelete', () => {
     expect(mockSignalTableRowsChanged).toHaveBeenCalledWith('tbl_1')
   })
 
+  it('uses the table definition revalidated with each committed delete batch', async () => {
+    const renamedTable = {
+      ...table,
+      name: 'Renamed issues',
+      schema: { columns: [{ id: 'col-title', name: 'Renamed title', type: 'string' }] },
+    }
+    mockSelectRowIdPage.mockResolvedValueOnce(['a']).mockResolvedValueOnce([])
+    mockDeletePageByIds.mockImplementationOnce(
+      async (_t, _w, ids: string[], _proof, _revalidate, onDeleted) => {
+        const rows = ids.map((id) => ({ id, data: { 'col-title': id } }))
+        await onDeleted?.(rows, renamedTable)
+        return rows.length
+      }
+    )
+
+    await runTableDelete(basePayload())
+
+    expect(mockFireTableTrigger).toHaveBeenCalledWith(
+      renamedTable.id,
+      renamedTable.name,
+      'delete',
+      [{ id: 'a', data: { 'col-title': 'a' } }],
+      null,
+      renamedTable.schema,
+      expect.any(String)
+    )
+  })
+
   it('stops once maxRows is reached and caps the final page fetch to the remaining budget', async () => {
     // budget 3 with page size 2: first page fills 2, the second is capped to the remaining 1.
     mockSelectRowIdPage.mockResolvedValueOnce(['a', 'b']).mockResolvedValueOnce(['c'])

apps/sim/lib/table/delete-runner.ts

@@ -123,8 +123,20 @@ export async function runTableDelete(payload: TableDeletePayload): Promise<void>
     // an absent filter is still legitimate (delete-all is an explicit caller mode).
     if (filter && !filterClause) throw new Error('Filter is required for bulk delete')
     const excluded = new Set(excludeRowIds ?? [])
-    const dispatchDeleteTriggers = (rows: DeletedTableRow[]) => {
-      void fireTableTrigger(table.id, table.name, 'delete', rows, null, table.schema, requestId)
+    const dispatchDeleteTriggers = async (
+      rows: DeletedTableRow[],
+      committedTable?: TableDefinition
+    ) => {
+      const triggerTable = committedTable ?? table
+      await fireTableTrigger(
+        triggerTable.id,
+        triggerTable.name,
+        'delete',
+        rows,
+        null,
+        triggerTable.schema,
+        requestId
+      )
     }
 
     // Resume the persisted count: a retried attempt's earlier batches are already committed,