fix(outbox): prevent bulk cleanup from starving delivery (#7654)

* fix(outbox): prevent bulk cleanup from starving delivery

* test(outbox): isolate scheduler PostgreSQL fixtures

Author: icecrasher321

.github/workflows/test-build.yml

@@ -179,14 +179,16 @@ jobs:
           lib/table/rows/secret-provenance.postgres.test.ts
           lib/memory/message-provenance.postgres.test.ts
 
-      - name: Verify Search progress and pagination in PostgreSQL
+      - name: Verify Search progress, pagination, and outbox scheduling in PostgreSQL
         working-directory: apps/sim
         env:
           KNOWLEDGE_ACL_TEST_DATABASE_URL: postgresql://postgres:postgres@127.0.0.1:5432/sim_auth_scim
         run: >-
           bunx vitest run --mode integration
           lib/knowledge/__integration__/search-source-progress.integration.ts
           lib/knowledge/__integration__/search-source-pagination.integration.ts
+          lib/core/outbox/service.integration.ts
+          lib/knowledge/__integration__/connector-upload.integration.ts
 
   test-build:
     name: Lint and Test

apps/sim/app/api/webhooks/outbox/process/route.ts

@@ -54,7 +54,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     }
 
     const result = await processOutboxEvents(handlers, {
-      batchSize: 20,
+      batchSize: 500,
       maxRuntimeMs: 790_000,
       minRemainingMs: 95_000,
     })

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

@@ -0,0 +1,277 @@
+/** Real PostgreSQL claims verify scheduling fairness and concurrent delivery. */
+import { db } from '@sim/db'
+import { outboxEvent } from '@sim/db/schema'
+import { withUtcTimestamps } from '@sim/db/timestamps'
+import { generateId } from '@sim/utils/id'
+import { eq, inArray, sql } from 'drizzle-orm'
+import { drizzle, type PostgresJsDatabase } from 'drizzle-orm/postgres-js'
+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('@sim/db', () => ({
+  get db() {
+    if (!database.current) throw new Error('Outbox PostgreSQL test database is not initialized')
+    return database.current
+  },
+}))
+
+import {
+  type OutboxHandler,
+  processOutboxEvents,
+  withOutboxHandlerTimeout,
+} from '@/lib/core/outbox/service'
+
+interface QueryPlan {
+  'Node Type': string
+  'Index Name'?: string
+  Plans?: QueryPlan[]
+}
+
+function planNodes(plan: QueryPlan): QueryPlan[] {
+  return [plan, ...(plan.Plans ?? []).flatMap(planNodes)]
+}
+
+describe('outbox scheduling in PostgreSQL', () => {
+  const eventTypes = new Set<string>()
+  const schemaName = `outbox_test_${generateId().replaceAll('-', '')}`
+  const databaseUrl = process.env.KNOWLEDGE_ACL_TEST_DATABASE_URL
+  if (!databaseUrl) throw new Error('Outbox tests require a disposable local database')
+  const connection = postgres(
+    databaseUrl,
+    withUtcTimestamps({
+      max: 4,
+      prepare: false,
+      fetch_types: false,
+      connection: { search_path: schemaName },
+      onnotice: () => {},
+    })
+  )
+
+  beforeAll(async () => {
+    await connection`CREATE SCHEMA ${connection(schemaName)}`
+    /** Copy the provisioned table and indexes without consuming another suite's pending events. */
+    await connection`CREATE TABLE outbox_event (LIKE public.outbox_event INCLUDING ALL)`
+    database.current = drizzle(connection)
+  })
+
+  afterEach(async () => {
+    if (eventTypes.size) {
+      await db.delete(outboxEvent).where(inArray(outboxEvent.eventType, [...eventTypes]))
+    }
+    eventTypes.clear()
+  })
+
+  afterAll(async () => {
+    try {
+      await connection`DROP SCHEMA ${connection(schemaName)} CASCADE`
+    } finally {
+      await connection.end()
+      database.current = undefined
+    }
+  })
+
+  async function enqueue(eventType: string, count: number, ageMs = 10_000) {
+    if (count > 1_000) throw new Error('Use bounded SQL batches for large fixtures')
+    const now = Date.now()
+    const rows = Array.from({ length: count }, (_, index) => ({
+      id: generateId(),
+      eventType,
+      payload: {},
+      createdAt: new Date(now - ageMs + index),
+      availableAt: new Date(now - 1),
+    }))
+    eventTypes.add(eventType)
+    await db.insert(outboxEvent).values(rows)
+    return rows
+  }
+
+  async function seedBacklog(
+    eventType: string,
+    count: number,
+    status: 'pending' | 'completed' | 'processing'
+  ) {
+    const prefix = generateId()
+    const availableAt = new Date(Date.now() - 60_000)
+    const createdAt = new Date(Date.now() - 24 * 60 * 60_000)
+    const lockedAt = status === 'processing' ? new Date(Date.now() - 11 * 60_000) : null
+    eventTypes.add(eventType)
+    for (let offset = 0; offset < count; offset += 1_000) {
+      await db.execute(sql`
+        INSERT INTO outbox_event
+          (id, event_type, payload, status, available_at, created_at, locked_at)
+        SELECT ${prefix} || ':' || n, ${eventType}, '{}'::json, ${status},
+          ${availableAt.toISOString()}::timestamp,
+          ${createdAt.toISOString()}::timestamp + n * interval '1 millisecond',
+          ${lockedAt?.toISOString() ?? null}::timestamp
+        FROM generate_series(${offset}::integer, ${Math.min(offset + 999, count - 1)}::integer) AS n
+      `)
+    }
+  }
+
+  it('serves newer event types before exhausting an older cleanup backlog', async () => {
+    await enqueue('test.outbox.cleanup', 1_000)
+    const [dispatch] = await enqueue('test.outbox.dispatch', 1, 2_000)
+    const [billing] = await enqueue('test.outbox.billing', 1, 1_000)
+    const delivered: string[] = []
+    const handler: OutboxHandler = async (_payload, context) => {
+      delivered.push(context.eventId)
+    }
+
+    const result = await processOutboxEvents(
+      {
+        'test.outbox.cleanup': handler,
+        'test.outbox.dispatch': handler,
+        'test.outbox.billing': handler,
+      },
+      { batchSize: 20 }
+    )
+
+    expect(result.processed).toBe(20)
+    expect(delivered.slice(0, 3)).toContain(dispatch.id)
+    expect(delivered.slice(0, 3)).toContain(billing.id)
+  })
+
+  it('keeps draining a single event type up to the batch limit in creation order', async () => {
+    const rows = await enqueue('test.outbox.cleanup', 30)
+    const delivered: string[] = []
+    const handler: OutboxHandler = async (_payload, context) => {
+      delivered.push(context.eventId)
+    }
+
+    const result = await processOutboxEvents({ 'test.outbox.cleanup': handler }, { batchSize: 25 })
+
+    expect(result.processed).toBe(25)
+    expect(delivered).toEqual(rows.slice(0, 25).map((row) => row.id))
+  })
+
+  it('keeps future events pending while serving other ready event types', async () => {
+    const [future] = await enqueue('test.outbox.future', 1)
+    await db
+      .update(outboxEvent)
+      .set({ availableAt: new Date(Date.now() + 60_000) })
+      .where(eq(outboxEvent.id, future.id))
+    await enqueue('test.outbox.cleanup', 5)
+    const delivered: string[] = []
+    const handler: OutboxHandler = async (_payload, context) => {
+      delivered.push(context.eventId)
+    }
+
+    const result = await processOutboxEvents({
+      'test.outbox.future': handler,
+      'test.outbox.cleanup': handler,
+    })
+
+    expect(result.processed).toBe(5)
+    expect(delivered).not.toContain(future.id)
+  })
+
+  it('does not deliver the same event twice when cron invocations overlap', async () => {
+    await enqueue('test.outbox.cleanup', 50)
+    await enqueue('test.outbox.dispatch', 2, 1_000)
+    const delivered: string[] = []
+    const handler: OutboxHandler = async (_payload, context) => {
+      delivered.push(context.eventId)
+    }
+    const handlers = { 'test.outbox.cleanup': handler, 'test.outbox.dispatch': handler }
+
+    const results = await Promise.all([
+      processOutboxEvents(handlers, { batchSize: 40 }),
+      processOutboxEvents(handlers, { batchSize: 40 }),
+    ])
+
+    expect(results.reduce((sum, result) => sum + result.processed, 0)).toBe(52)
+    expect(delivered).toHaveLength(52)
+    expect(new Set(delivered).size).toBe(52)
+  })
+
+  it('serves rare types using indexed heads beside 100,000 pending and 100,000 completed events', async () => {
+    await seedBacklog('test.outbox.cleanup', 100_000, 'completed')
+    await seedBacklog('test.outbox.cleanup', 100_000, 'pending')
+    const [dispatch] = await enqueue('test.outbox.dispatch', 1)
+    const [billing] = await enqueue('test.outbox.billing', 1)
+    await db.execute(sql`ANALYZE outbox_event`)
+
+    const plans = await db.execute<{ 'QUERY PLAN': { Plan: QueryPlan }[] }>(sql`
+      EXPLAIN (FORMAT JSON)
+      SELECT * FROM outbox_event
+      WHERE status = 'pending' AND event_type = 'test.outbox.cleanup'
+        AND available_at <= now()
+      ORDER BY available_at, created_at, id
+      LIMIT 1 FOR UPDATE SKIP LOCKED
+      `)
+    const nodes = planNodes(plans[0]['QUERY PLAN'][0].Plan)
+    const indexScan = nodes.find((node) => node['Node Type'] === 'Index Scan')
+    expect(indexScan).toBeDefined()
+    const [index] = await connection<{ indexdef: string }[]>`
+        SELECT indexdef FROM pg_indexes
+        WHERE schemaname = ${schemaName} AND indexname = ${indexScan?.['Index Name'] ?? ''}
+        LIMIT 1
+      `
+    expect(index.indexdef).toContain('(event_type, available_at, created_at, id)')
+    expect(index.indexdef).toContain("WHERE (status = 'pending'::text)")
+    expect(nodes.some((node) => node['Node Type'] === 'Sort')).toBe(false)
+
+    const delivered: string[] = []
+    const handler: OutboxHandler = async (_payload, context) => {
+      delivered.push(context.eventId)
+    }
+    const result = await processOutboxEvents(
+      {
+        'test.outbox.cleanup': handler,
+        'test.outbox.dispatch': handler,
+        'test.outbox.billing': handler,
+      },
+      { batchSize: 20 }
+    )
+
+    expect(result.processed).toBe(20)
+    expect(delivered.slice(0, 3)).toContain(dispatch.id)
+    expect(delivered.slice(0, 3)).toContain(billing.id)
+  }, 60_000)
+
+  it('runs eligible short handlers without claiming a type that exceeds the deadline', async () => {
+    const [long] = await enqueue('test.outbox.long', 1)
+    const [short] = await enqueue('test.outbox.short', 1)
+    const delivered: string[] = []
+    const handler: OutboxHandler = async (_payload, context) => {
+      delivered.push(context.eventId)
+    }
+    const result = await processOutboxEvents(
+      {
+        'test.outbox.long': withOutboxHandlerTimeout(async () => {
+          throw new Error('Long handler must remain pending')
+        }, 550_000),
+        'test.outbox.short': handler,
+      },
+      { maxRuntimeMs: 110_000 }
+    )
+
+    expect(result.processed).toBe(1)
+    expect(delivered).toEqual([short.id])
+    const [pending] = await db.select().from(outboxEvent).where(eq(outboxEvent.id, long.id))
+    expect(pending).toMatchObject({ status: 'pending', attempts: 0, lockedAt: null })
+  })
+
+  it('bounds stale-lease recovery and leaves excess rows for the next invocation', async () => {
+    await seedBacklog('test.outbox.stale', 1_005, 'processing')
+
+    const first = await processOutboxEvents({}, { batchSize: 0 })
+    expect(first.reaped).toBe(1_000)
+    const counts = await db
+      .select({ status: outboxEvent.status, count: sql<number>`count(*)::int` })
+      .from(outboxEvent)
+      .where(eq(outboxEvent.eventType, 'test.outbox.stale'))
+      .groupBy(outboxEvent.status)
+    expect(counts).toEqual(
+      expect.arrayContaining([
+        { status: 'pending', count: 1_000 },
+        { status: 'processing', count: 5 },
+      ])
+    )
+
+    const second = await processOutboxEvents({}, { batchSize: 0 })
+    expect(second.reaped).toBe(5)
+  })
+})