perf(outbox): resume each retention batch where the previous run's batch ended

apps/sim/lib/core/outbox/retention.integration.ts

@@ -1,18 +1,21 @@
 /** Real PostgreSQL retention removes only finished events that nothing reads again. */
 
 import { outboxEvent } from '@sim/db/schema'
-import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
+import { readTestDatabaseUrl, readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
 import { withUtcTimestamps } from '@sim/db/timestamps'
+import { redisConfigMock, redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
 import { storageServiceMock, storageServiceMockFns } from '@sim/testing/mocks/storage-service.mock'
 import { generateId } from '@sim/utils/id'
 import { sql } from 'drizzle-orm'
 import { drizzle, type PostgresJsDatabase } from 'drizzle-orm/postgres-js'
+import Redis from 'ioredis'
 import postgres from 'postgres'
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'
 
 const database = vi.hoisted(() => ({ current: undefined as PostgresJsDatabase | undefined }))
 
 vi.mock('@/lib/uploads/core/storage-service', () => storageServiceMock)
+vi.mock('@/lib/core/config/redis', () => redisConfigMock)
 
 vi.mock('@sim/db', () => ({
   get db() {
@@ -24,6 +27,7 @@ vi.mock('@sim/db', () => ({
 import {
   COMPLETED_OUTBOX_RETENTION_MS,
   OUTBOX_PRUNE_BATCH_SIZE,
+  OUTBOX_PRUNE_PASS_MAX_MS,
   pruneCompletedOutboxEvents,
 } from '@/lib/core/outbox/retention'
 import { writeFramedCheckpoint } from '@/lib/knowledge/documents/processing-checkpoint-store'
@@ -34,6 +38,7 @@ const ADMIN_OPERATION = 'admin.organization-member-operation'
 const OCR_CHECKPOINT_EXPIRY = 'knowledge.document.ocr-checkpoint.expire'
 const EMBEDDING_CHECKPOINT_EXPIRY = 'knowledge.document.embedding-checkpoint.expire'
 const DEPLOYMENT_EFFECT = 'workflow.deployment.sync-active-side-effects'
+const redisUrl = readTestRedisUrl()
 
 describe('completed outbox retention in PostgreSQL', () => {
   const schemaName = `outbox_retention_${generateId().replaceAll('-', '')}`
@@ -49,14 +54,25 @@ describe('completed outbox retention in PostgreSQL', () => {
   )
   const testDb = drizzle(connection)
   const expired = () => new Date(Date.now() - COMPLETED_OUTBOX_RETENTION_MS - 60_000)
+  const cursorKeys = [RECOVER, EMBEDDING_CHECKPOINT_EXPIRY].map(
+    (eventType) => `outbox-prune-cursor:v1:${eventType}`
+  )
+  let redis: Redis | undefined
 
   beforeAll(async () => {
     await connection`CREATE SCHEMA ${connection(schemaName)}`
     await connection`CREATE TABLE outbox_event (LIKE public.outbox_event INCLUDING ALL)`
     database.current = testDb
+    if (redisUrl) {
+      redis = new Redis(redisUrl, { lazyConnect: true, maxRetriesPerRequest: 0 })
+      await redis.connect()
+      await redis.del(...cursorKeys)
+    }
   })
 
   afterEach(async () => {
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(null)
+    await redis?.del(...cursorKeys)
     await connection`TRUNCATE outbox_event`
   })
 
@@ -65,10 +81,18 @@ describe('completed outbox retention in PostgreSQL', () => {
       await connection`DROP SCHEMA ${connection(schemaName)} CASCADE`
     } finally {
       await connection.end()
+      await redis?.quit()
       database.current = undefined
     }
   })
 
+  /** Runs the rest of the test with each type's pass resuming through real Redis. */
+  function resumeThroughRedis(): Redis {
+    if (!redis) throw new Error('TEST_REDIS_URL is required for the resume cases')
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(redis)
+    return redis
+  }
+
   async function seed(eventType: string, status: string, createdAt: Date) {
     const id = generateId()
     await testDb.insert(outboxEvent).values({
@@ -154,4 +178,113 @@ describe('completed outbox retention in PostgreSQL', () => {
     const rows = await testDb.select().from(outboxEvent)
     expect(rows.map((row) => [row.id, row.status])).toEqual([[gate.id, 'pending']])
   })
+
+  describe.runIf(Boolean(redisUrl))("resuming each type's pass across runs", () => {
+    /** One full batch plus one newer row, all completed, a millisecond apart, oldest first. */
+    async function seedFullBatchPlusOne(eventType: string) {
+      const newest = expired().toISOString()
+      await testDb.execute(sql`
+        INSERT INTO outbox_event (id, event_type, payload, status, available_at, created_at)
+        SELECT 'row:' || n, ${eventType}, '{}'::json, 'completed',
+          ${newest}::timestamp, ${newest}::timestamp - n * interval '1 millisecond'
+        FROM generate_series(0, ${OUTBOX_PRUNE_BATCH_SIZE}::integer) AS n
+      `)
+    }
+
+    /** Pending before any seeded row, then completed once the first batch has gone past it. */
+    async function seedLateCompletion(eventType: string) {
+      const id = await seed(
+        eventType,
+        'pending',
+        new Date(expired().getTime() - OUTBOX_PRUNE_BATCH_SIZE - 60_000)
+      )
+      return {
+        id,
+        complete: () =>
+          testDb
+            .update(outboxEvent)
+            .set({ status: 'completed' })
+            .where(sql`${outboxEvent.id} = ${id}`),
+      }
+    }
+
+    it('resumes after the last row the previous batch deleted, and the next pass covers a row it went by', async () => {
+      resumeThroughRedis()
+      await seedFullBatchPlusOne(EMBEDDING_CHECKPOINT_EXPIRY)
+      const late = await seedLateCompletion(EMBEDDING_CHECKPOINT_EXPIRY)
+
+      expect(await pruneCompletedOutboxEvents()).toBe(OUTBOX_PRUNE_BATCH_SIZE)
+      expect(await remainingIds()).toEqual(new Set(['row:0', late.id]))
+      await late.complete()
+
+      /** Starting again from the oldest row would take the late completion in this batch too. */
+      expect(await pruneCompletedOutboxEvents()).toBe(1)
+      expect(await remainingIds()).toEqual(new Set([late.id]))
+
+      expect(await pruneCompletedOutboxEvents()).toBe(1)
+      expect(await remainingIds()).toEqual(new Set())
+    })
+
+    it('deletes exactly the eligible rows across batches that split one creation time', async () => {
+      resumeThroughRedis()
+      const createdAt = `${expired().toISOString().slice(0, -1)}123Z`
+      await testDb.execute(sql`
+        INSERT INTO outbox_event (id, event_type, payload, status, available_at, created_at)
+        SELECT 'tie:' || lpad(n::text, 5, '0'), ${RECOVER}, '{}'::json,
+          CASE WHEN n % 100 = 7 THEN 'pending' ELSE 'completed' END,
+          ${createdAt}::timestamp, ${createdAt}::timestamp
+        FROM generate_series(1, ${OUTBOX_PRUNE_BATCH_SIZE + 500}::integer) AS n
+      `)
+      const kept = [
+        await seed(ADMIN_OPERATION, 'completed', expired()),
+        await seed(RECOVER, 'completed', new Date(Date.now() - 60_000)),
+      ]
+      const pendingTies = await testDb
+        .select({ id: outboxEvent.id })
+        .from(outboxEvent)
+        .where(sql`${outboxEvent.status} = 'pending'`)
+      const completedTies = OUTBOX_PRUNE_BATCH_SIZE + 500 - pendingTies.length
+
+      expect(await pruneCompletedOutboxEvents()).toBe(OUTBOX_PRUNE_BATCH_SIZE)
+      expect(await pruneCompletedOutboxEvents()).toBe(completedTies - OUTBOX_PRUNE_BATCH_SIZE)
+      expect(await pruneCompletedOutboxEvents()).toBe(0)
+      expect(await remainingIds()).toEqual(new Set([...kept, ...pendingTies.map((row) => row.id)]))
+    })
+
+    it.each([
+      ['the pass is over its time limit', async () => {}, OUTBOX_PRUNE_PASS_MAX_MS],
+      [
+        'the cursor is unreadable',
+        async (client: Redis) => {
+          await client.set(`outbox-prune-cursor:v1:${RECOVER}`, '{"createdAt":')
+        },
+        0,
+      ],
+      [
+        'the cursor names no real creation time',
+        async (client: Redis) => {
+          await client.set(
+            `outbox-prune-cursor:v1:${RECOVER}`,
+            JSON.stringify({
+              createdAt: '2026-02-30 00:00:00',
+              id: 'row:1',
+              passStartedAt: Date.now(),
+            })
+          )
+        },
+        0,
+      ],
+    ])('starts again from the oldest row when %s', async (_, corrupt, laterMs) => {
+      const client = resumeThroughRedis()
+      await seedFullBatchPlusOne(RECOVER)
+      const late = await seedLateCompletion(RECOVER)
+
+      expect(await pruneCompletedOutboxEvents()).toBe(OUTBOX_PRUNE_BATCH_SIZE)
+      await late.complete()
+      await corrupt(client)
+
+      expect(await pruneCompletedOutboxEvents(new Date(Date.now() + laterMs))).toBe(2)
+      expect(await remainingIds()).toEqual(new Set())
+    })
+  })
 })

apps/sim/lib/core/outbox/retention.ts

@@ -1,11 +1,14 @@
 import { db } from '@sim/db'
 import { outboxEvent } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
 import { and, asc, eq, inArray, lt, sql } from 'drizzle-orm'
+import { z } from 'zod'
 import {
   CHAT_BACKEND_CLEANUP_EVENT,
   CHAT_INLINE_IMAGE_CLEANUP_EVENT,
 } from '@/lib/cleanup/chat-backend-cleanup-event'
+import { getRedisClient } from '@/lib/core/config/redis'
 import {
   EMBEDDING_CHECKPOINT_CLEANUP_EVENT,
   OCR_CHECKPOINT_CLEANUP_OUTBOX_EVENT,
@@ -57,24 +60,134 @@ const PRUNABLE_OUTBOX_EVENT_TYPES = [
   CHAT_INLINE_IMAGE_CLEANUP_EVENT,
 ] as const
 
-async function pruneCompletedBatch(eventType: string, cutoff: Date): Promise<number> {
+/**
+ * How long one oldest-to-newest pass over a type keeps resuming where the previous run's batch
+ * ended. The run after that starts again from the oldest row, so a row the pass went by before it
+ * completed, or while an overlapping run held its lock, is still pruned.
+ */
+export const OUTBOX_PRUNE_PASS_MAX_MS = 60 * 60_000
+
+/**
+ * `created_at::text` under PostgreSQL's ISO date style, to the microsecond. A stored value that is
+ * not a real calendar time fails the round trip here instead of the batch's `::timestamp` cast.
+ */
+const postgresTimestampSchema = z.string().refine((value) => {
+  if (!/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(\.\d{1,6})?$/.test(value)) return false
+  const seconds = value.slice(0, 19).replace(' ', 'T')
+  const parsed = new Date(`${seconds}Z`)
+  return !Number.isNaN(parsed.getTime()) && parsed.toISOString().startsWith(seconds)
+})
+
+/**
+ * Where a type's pass resumes: the creation time, as PostgreSQL prints it so no microseconds are
+ * lost, and id of the last row the previous batch deleted.
+ */
+const resumePointSchema = z.object({ createdAt: postgresTimestampSchema, id: z.string() })
+const pruneCursorSchema = resumePointSchema.extend({ passStartedAt: z.number() })
+
+type ResumePoint = z.infer<typeof resumePointSchema>
+type PruneCursor = z.infer<typeof pruneCursorSchema>
+
+const pruneCursorKey = (eventType: string) => `outbox-prune-cursor:v1:${eventType}`
+
+/**
+ * Each type's cursor from the pass in progress, or null when Redis is unavailable. A missing,
+ * unreadable or expired cursor starts that type's pass from the oldest row.
+ */
+async function readPruneCursors(now: Date): Promise<Map<string, PruneCursor> | null> {
+  const redis = getRedisClient()
+  if (!redis) return null
+  try {
+    const stored = await redis.mget(...PRUNABLE_OUTBOX_EVENT_TYPES.map(pruneCursorKey))
+    const cursors = new Map<string, PruneCursor>()
+    PRUNABLE_OUTBOX_EVENT_TYPES.forEach((eventType, index) => {
+      const cursor = parsePruneCursor(stored[index])
+      if (cursor && now.getTime() - cursor.passStartedAt < OUTBOX_PRUNE_PASS_MAX_MS) {
+        cursors.set(eventType, cursor)
+      }
+    })
+    return cursors
+  } catch (error) {
+    logger.warn('Failed to read outbox prune cursors; pruning from the oldest rows', {
+      error: getErrorMessage(error),
+    })
+    return null
+  }
+}
+
+function parsePruneCursor(stored: string | null): PruneCursor | null {
+  if (!stored) return null
+  try {
+    const parsed = pruneCursorSchema.safeParse(JSON.parse(stored))
+    return parsed.success ? parsed.data : null
+  } catch {
+    return null
+  }
+}
+
+/** Stores each type's next cursor, or clears it when that type's pass is over. */
+async function savePruneCursors(next: ReadonlyMap<string, PruneCursor | null>): Promise<void> {
+  const redis = getRedisClient()
+  if (!redis || next.size === 0) return
+  try {
+    const pipeline = redis.pipeline()
+    for (const [eventType, cursor] of next) {
+      if (cursor) {
+        pipeline.set(
+          pruneCursorKey(eventType),
+          JSON.stringify(cursor),
+          'EX',
+          OUTBOX_PRUNE_PASS_MAX_MS / 1000
+        )
+      } else {
+        pipeline.del(pruneCursorKey(eventType))
+      }
+    }
+    await pipeline.exec()
+  } catch (error) {
+    logger.warn('Failed to save outbox prune cursors', { error: getErrorMessage(error) })
+  }
+}
+
+/**
+ * Deletes the oldest batch after `resumeAfter` in (creation time, id) order. Returns the last
+ * deleted row when the batch was full, and null when it reached the cutoff and the pass is over.
+ */
+async function pruneCompletedBatch(
+  eventType: string,
+  cutoff: Date,
+  resumeAfter: ResumePoint | undefined
+): Promise<{ deleted: number; resumeAfter: ResumePoint | null }> {
   return db.transaction(async (tx) => {
     await tx.execute(sql`SELECT set_config('statement_timeout', '10000', true)`)
-    const expired = tx
-      .select({ id: outboxEvent.id })
+    const expired = await tx
+      .select({ id: outboxEvent.id, createdAt: sql<string>`${outboxEvent.createdAt}::text` })
       .from(outboxEvent)
       .where(
         and(
           eq(outboxEvent.eventType, eventType),
           lt(outboxEvent.createdAt, cutoff),
-          eq(outboxEvent.status, 'completed')
+          eq(outboxEvent.status, 'completed'),
+          resumeAfter
+            ? sql`(${outboxEvent.createdAt}, ${outboxEvent.id}) > (${resumeAfter.createdAt}::timestamp, ${resumeAfter.id})`
+            : undefined
         )
       )
-      .orderBy(asc(outboxEvent.createdAt))
+      .orderBy(asc(outboxEvent.createdAt), asc(outboxEvent.id))
       .limit(OUTBOX_PRUNE_BATCH_SIZE)
       .for('update', { skipLocked: true })
-    const deleted = await tx.delete(outboxEvent).where(inArray(outboxEvent.id, expired))
-    return deleted.count
+    const last = expired.at(-1)
+    if (!last) return { deleted: 0, resumeAfter: null }
+    const deleted = await tx.delete(outboxEvent).where(
+      inArray(
+        outboxEvent.id,
+        expired.map((row) => row.id)
+      )
+    )
+    return {
+      deleted: deleted.count,
+      resumeAfter: expired.length === OUTBOX_PRUNE_BATCH_SIZE ? last : null,
+    }
   })
 }
 
@@ -83,12 +196,33 @@ async function pruneCompletedBatch(eventType: string, cutoff: Date): Promise<num
  * retention window, through the type/creation index. One bounded batch per run keeps a large
  * backlog from turning into a burst of deletes; overlapping runs skip each other's locked rows.
  * Pending, processing and dead-letter rows are never deleted.
+ *
+ * Each batch resumes after the last row the previous run's batch deleted, kept in Redis. Starting
+ * again from the oldest row would walk every index entry the earlier batches deleted that vacuum
+ * has not yet removed, which grows with the backlog being drained. A pass starts over from the
+ * oldest row once a batch comes up short of the cutoff or the pass is older than an hour; without
+ * Redis every batch starts from the oldest row.
  */
 export async function pruneCompletedOutboxEvents(now = new Date()): Promise<number> {
   const cutoff = new Date(now.getTime() - COMPLETED_OUTBOX_RETENTION_MS)
+  const cursors = await readPruneCursors(now)
+  const nextCursors = new Map<string, PruneCursor | null>()
   let pruned = 0
-  for (const eventType of PRUNABLE_OUTBOX_EVENT_TYPES) {
-    pruned += await pruneCompletedBatch(eventType, cutoff)
+  try {
+    for (const eventType of PRUNABLE_OUTBOX_EVENT_TYPES) {
+      const cursor = cursors?.get(eventType)
+      const batch = await pruneCompletedBatch(eventType, cutoff, cursor)
+      pruned += batch.deleted
+      nextCursors.set(
+        eventType,
+        batch.resumeAfter && {
+          ...batch.resumeAfter,
+          passStartedAt: cursor?.passStartedAt ?? now.getTime(),
+        }
+      )
+    }
+  } finally {
+    if (cursors) await savePruneCursors(nextCursors)
   }
   if (pruned > 0) logger.info('Pruned completed outbox events', { pruned })
   return pruned