fix(tables): prioritize oldest TTL cleanup rows

Page cleanup by created_at and id, with a supporting index, so capped runs make progress on the oldest rows first.

Author: j15z

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

@@ -1,8 +1,13 @@
 /**
  * @vitest-environment node
  */
+import type { SQL } from 'drizzle-orm'
+import { PgDialect } from 'drizzle-orm/pg-core'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
+vi.unmock('@sim/db/schema')
+vi.unmock('drizzle-orm')
+
 const {
   mockDeleteExecute,
   mockListExecute,
@@ -27,6 +32,8 @@ vi.mock('@/lib/table/service', () => ({ withLockedTable: mockWithLockedTable }))
 
 import { cleanupTableRowTtlTask, runCleanupTableRowTtl } from '@/background/cleanup-table-row-ttl'
 
+const dialect = new PgDialect()
+
 const table = {
   id: 'table-1',
   workspaceId: 'workspace-1',
@@ -49,10 +56,14 @@ describe('table row TTL cleanup', () => {
     )
   })
 
-  it('deletes expired rows in locked, keyset batches and signals the table', async () => {
+  it('deletes expired rows in locked, created-at keyset batches and signals the table', async () => {
     mockDeleteExecute
-      .mockResolvedValueOnce([{ count: 500, lastId: 'row-500' }])
-      .mockResolvedValueOnce([{ count: 12, lastId: 'row-512' }])
+      .mockResolvedValueOnce([
+        { count: 500, createdAt: '2026-01-01T00:00:00.123456', lastId: 'row-500' },
+      ])
+      .mockResolvedValueOnce([
+        { count: 12, createdAt: '2026-01-02T00:00:00.000000', lastId: 'row-512' },
+      ])
 
     await expect(runCleanupTableRowTtl()).resolves.toEqual({
       batches: 2,
@@ -61,27 +72,58 @@ describe('table row TTL cleanup', () => {
     })
     expect(mockWithLockedTable).toHaveBeenCalledTimes(2)
     expect(mockDeleteExecute).toHaveBeenCalledTimes(2)
+    const secondQuery = dialect.sqlToQuery(mockDeleteExecute.mock.calls[1][0] as SQL)
+    expect(secondQuery.sql.replace(/\$\d+/g, '?').replace(/\s+/g, ' ')).toContain(
+      'AND (table_row.created_at, table_row.id) > (?::timestamp, ?)'
+    )
+    expect(secondQuery.params).toEqual(
+      expect.arrayContaining(['2026-01-01T00:00:00.123456', 'row-500'])
+    )
     expect(mockSignalTableRowsChanged).toHaveBeenCalledWith(table.id)
   })
 
   it('compares TTL values with whole Date.now epoch seconds', async () => {
     const nowEpochMilliseconds = 1_700_000_000_123
     const nowEpochSeconds = 1_700_000_000
     const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(nowEpochMilliseconds)
-    mockDeleteExecute.mockResolvedValue([{ count: 0, lastId: null }])
+    mockDeleteExecute.mockResolvedValue([{ count: 0, createdAt: null, lastId: null }])
 
     try {
       await runCleanupTableRowTtl()
     } finally {
       nowSpy.mockRestore()
     }
 
-    expect(mockListExecute.mock.calls[0][0]).toMatchObject({
-      values: expect.arrayContaining([nowEpochSeconds]),
-    })
-    expect(mockDeleteExecute.mock.calls[0][0]).toMatchObject({
-      values: expect.arrayContaining([nowEpochSeconds]),
-    })
+    expect(dialect.sqlToQuery(mockListExecute.mock.calls[0][0] as SQL).params).toContain(
+      nowEpochSeconds
+    )
+    expect(dialect.sqlToQuery(mockDeleteExecute.mock.calls[0][0] as SQL).params).toContain(
+      nowEpochSeconds
+    )
+  })
+
+  it('checks the oldest expired rows first without using creation time as an expiry rule', async () => {
+    mockDeleteExecute.mockResolvedValue([{ count: 0, createdAt: null, lastId: null }])
+
+    await runCleanupTableRowTtl()
+
+    const query = dialect
+      .sqlToQuery(mockDeleteExecute.mock.calls[0][0] as SQL)
+      .sql.replace(/\s+/g, ' ')
+      .replace(/\$\d+/g, '?')
+      .trim()
+    expect(query).toContain('AND (table_row.data->>?)::numeric <= ?')
+    expect(query).toContain('ORDER BY table_row.created_at, table_row.id')
+    expect(query).toContain(`to_char(created_at, 'YYYY-MM-DD"T"HH24:MI:SS.US')`)
+    expect(query).not.toContain('table_row.created_by')
+  })
+
+  it('rejects a batch without a creation-time cursor', async () => {
+    mockDeleteExecute.mockResolvedValue([{ count: 1, lastId: 'row-1' }])
+
+    await expect(runCleanupTableRowTtl()).rejects.toThrow(
+      'Table row TTL cleanup did not return a creation-time cursor'
+    )
   })
 
   it('does no work when already aborted', async () => {
@@ -114,7 +156,9 @@ describe('table row TTL cleanup', () => {
   })
 
   it('stops after one hundred full batches', async () => {
-    mockDeleteExecute.mockResolvedValue([{ count: 500, lastId: 'row-cursor' }])
+    mockDeleteExecute.mockResolvedValue([
+      { count: 500, createdAt: '2026-01-01T00:00:00.000000', lastId: 'row-cursor' },
+    ])
 
     await expect(runCleanupTableRowTtl()).resolves.toEqual({
       batches: 100,
@@ -144,12 +188,12 @@ describe('table row TTL cleanup', () => {
           const attempt = (tableAttempts.get(tableId) ?? 0) + 1
           tableAttempts.set(tableId, attempt)
           if (tableId === table.id && attempt === 1) {
-            return [{ count: 500, lastId: 'row-500' }]
+            return [{ count: 500, createdAt: '2026-01-01T00:00:00.000000', lastId: 'row-500' }]
           }
           if (tableId === secondTable.id) {
-            return [{ count: 1, lastId: 'row-1' }]
+            return [{ count: 1, createdAt: '2026-01-01T00:00:00.000000', lastId: 'row-1' }]
           }
-          return [{ count: 0, lastId: null }]
+          return [{ count: 0, createdAt: null, lastId: null }]
         }),
       })
     })

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

@@ -25,12 +25,17 @@ interface ExpiredTtlTableRef {
 interface DeletedTtlBatch {
   attempted: boolean
   deleted: number
-  lastId: string | null
+  cursor: TtlCleanupCursor | null
+}
+
+interface TtlCleanupCursor {
+  createdAt: string
+  id: string
 }
 
 interface TtlTableCleanupState {
   ref: ExpiredTtlTableRef
-  afterId?: string
+  after?: TtlCleanupCursor
   deleted: number
   complete: boolean
 }
@@ -81,18 +86,33 @@ async function listExpiredTtlTables(nowEpochSeconds: number): Promise<ExpiredTtl
 
 function parseDeletedBatch(rows: unknown): Omit<DeletedTtlBatch, 'attempted'> {
   const [row] = Array.isArray(rows)
-    ? (rows as Array<{ count?: number | string; lastId?: string | null }>)
+    ? (rows as Array<{
+        count?: number | string
+        createdAt?: string | null
+        lastId?: string | null
+      }>)
     : []
   if (!row) throw new Error('Table row TTL cleanup did not return a deleted count')
 
   const deleted = Number(row.count)
   if (!Number.isSafeInteger(deleted) || deleted < 0 || deleted > TTL_CLEANUP_BATCH_SIZE) {
     throw new Error('Table row TTL cleanup returned an invalid deleted count')
   }
-  if (deleted > 0 && typeof row.lastId !== 'string') {
-    throw new Error('Table row TTL cleanup did not return a row cursor')
+  if (deleted > 0) {
+    if (typeof row.lastId !== 'string') {
+      throw new Error('Table row TTL cleanup did not return a row cursor')
+    }
+    if (typeof row.createdAt !== 'string') {
+      throw new Error('Table row TTL cleanup did not return a creation-time cursor')
+    }
+  }
+  return {
+    deleted,
+    cursor:
+      typeof row.createdAt === 'string' && typeof row.lastId === 'string'
+        ? { createdAt: row.createdAt, id: row.lastId }
+        : null,
   }
-  return { deleted, lastId: row.lastId ?? null }
 }
 
 async function deleteExpiredTableRowBatch(
@@ -101,29 +121,43 @@ async function deleteExpiredTableRowBatch(
   workspaceId: string,
   columnKey: string,
   nowEpochSeconds: number,
-  afterId?: string
+  after?: TtlCleanupCursor
 ): Promise<Omit<DeletedTtlBatch, 'attempted'>> {
-  const rows = await trx.execute<{ count: number | string; lastId: string | null }>(sql`
+  const rows = await trx.execute<{
+    count: number | string
+    createdAt: string | null
+    lastId: string | null
+  }>(sql`
     WITH candidates AS MATERIALIZED (
       SELECT table_row.id
       FROM ${userTableRows} AS table_row
       WHERE table_row.table_id = ${tableId}
         AND table_row.workspace_id = ${workspaceId}
-        ${afterId ? sql`AND table_row.id > ${afterId}` : sql``}
+        ${
+          after
+            ? sql`AND (table_row.created_at, table_row.id) > (${after.createdAt}::timestamp, ${after.id})`
+            : sql``
+        }
         AND jsonb_typeof(table_row.data->${columnKey}) = 'number'
         AND (table_row.data->>${columnKey})::numeric <= ${nowEpochSeconds}
-      ORDER BY table_row.id
+      ORDER BY table_row.created_at, table_row.id
       LIMIT ${TTL_CLEANUP_BATCH_SIZE}
       FOR UPDATE OF table_row SKIP LOCKED
     ), deleted AS (
       DELETE FROM ${userTableRows} AS table_row
       USING candidates
       WHERE table_row.id = candidates.id
-      RETURNING table_row.id
+      RETURNING table_row.id, table_row.created_at
     )
     SELECT
       count(*)::integer AS count,
-      max(id) AS "lastId"
+      (array_agg(id ORDER BY created_at DESC, id DESC))[1] AS "lastId",
+      (
+        array_agg(
+          to_char(created_at, 'YYYY-MM-DD"T"HH24:MI:SS.US')
+          ORDER BY created_at DESC, id DESC
+        )
+      )[1] AS "createdAt"
     FROM deleted
   `)
   return parseDeletedBatch(rows)
@@ -132,7 +166,7 @@ async function deleteExpiredTableRowBatch(
 async function deleteExpiredRowsForTable(
   ref: ExpiredTtlTableRef,
   nowEpochSeconds: number,
-  afterId?: string
+  after?: TtlCleanupCursor
 ): Promise<DeletedTtlBatch> {
   try {
     return await withLockedTable(
@@ -142,29 +176,29 @@ async function deleteExpiredRowsForTable(
           assertRowDelete(table)
         } catch (error) {
           if (error instanceof TableLockedError) {
-            return { attempted: false, deleted: 0, lastId: null }
+            return { attempted: false, deleted: 0, cursor: null }
           }
           throw error
         }
 
         const ttlColumn = table.schema.columns.find((column) => column.type === 'ttl')
-        if (!ttlColumn) return { attempted: false, deleted: 0, lastId: null }
+        if (!ttlColumn) return { attempted: false, deleted: 0, cursor: null }
 
         const batch = await deleteExpiredTableRowBatch(
           trx,
           table.id,
           table.workspaceId,
           getColumnId(ttlColumn),
           nowEpochSeconds,
-          afterId
+          after
         )
         return { attempted: true, ...batch }
       },
       { expectedWorkspaceId: ref.workspaceId }
     )
   } catch (error) {
     if (asOrchestrationError(error)?.code === 'not_found') {
-      return { attempted: false, deleted: 0, lastId: null }
+      return { attempted: false, deleted: 0, cursor: null }
     }
     throw error
   }
@@ -195,7 +229,7 @@ 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.afterId)
+      const batch = await deleteExpiredRowsForTable(state.ref, nowEpochSeconds, state.after)
       if (!batch.attempted) {
         state.complete = true
         continue
@@ -204,7 +238,7 @@ export async function runCleanupTableRowTtl(
       batches++
       deleted += batch.deleted
       state.deleted += batch.deleted
-      state.afterId = batch.lastId ?? undefined
+      state.after = batch.cursor ?? undefined
       if (batch.deleted < TTL_CLEANUP_BATCH_SIZE) state.complete = true
     }
   }

packages/db/migrations/0309_cooing_ghost_rider.sql

@@ -0,0 +1,7 @@
+-- Concurrent index operations cannot run inside the migration runner's transaction.
+COMMIT;--> statement-breakpoint
+SET lock_timeout = 0;--> statement-breakpoint
+-- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve row writes.
+DROP INDEX CONCURRENTLY IF EXISTS "user_table_rows_table_created_id_idx";--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "user_table_rows_table_created_id_idx" ON "user_table_rows" USING btree ("table_id","created_at","id");--> statement-breakpoint
+SET lock_timeout = '5s';