From b186c33d61cebce3d46c8ef71c5a84aedbd12c29 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 29 Oct 2026 11:21:35 -0700 Subject: [PATCH 1/2] perf(outbox): resume each retention batch where the previous run's batch ended The completed-event prune deletes one oldest-first batch per type per run. Every batch started its scan of outbox_event_type_created_idx at the oldest row of its type, and the rows earlier batches deleted stay in that index until vacuum removes them, so while a large backlog drains each batch first walks everything deleted since the last vacuum. That walk grows with each run: on a read replica during a large drain it took about 0.6s before the first live row, and a batch can reach the 10s statement timeout. Each batch now resumes after the (created_at, id) of the last row the previous run's batch deleted, the same shape as the idempotency sweep's resume. Batches are ordered by (created_at, id) so ties on created_at split exactly, the cursor keeps PostgreSQL's microsecond text form, and the row comparison becomes a created_at >= bound on the index, with the full comparison as a filter. Resuming from a cursor, the same batch took about 20ms. The prune runs once per processor run, so the cursor has to outlive the run; it is kept in Redis as a hint, one MGET before the batches and one pipeline after. A row the pass went by before it completed, or while an overlapping run held its lock, is covered by the next pass: the cursor is cleared when a batch comes up short of the cutoff, and a pass older than an hour starts over from the oldest row. Without Redis, or with an unreadable cursor, a batch starts from the oldest row as before. Batch size, timeout, cadence and which rows are eligible are unchanged. Autopsy: the retention suite asserted which rows each run deleted, which a scan from the oldest row gets right, so nothing observed where a batch started. The new Redis cases complete a row behind the cursor between runs: resuming leaves it for the next pass and the old scan takes it at once, so the first case fails on the previous code. The others split one creation time across batches with pending rows mixed in, and restart a pass that is over an hour old or whose cursor is unreadable. --- .../lib/core/outbox/retention.integration.ts | 121 ++++++++++++++- apps/sim/lib/core/outbox/retention.ts | 141 ++++++++++++++++-- 2 files changed, 252 insertions(+), 10 deletions(-) diff --git a/apps/sim/lib/core/outbox/retention.integration.ts b/apps/sim/lib/core/outbox/retention.integration.ts index 0aff3a09b97..e97e7a9a9b3 100644 --- a/apps/sim/lib/core/outbox/retention.integration.ts +++ b/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,99 @@ 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, + ], + ])('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()) + }) + }) }) diff --git a/apps/sim/lib/core/outbox/retention.ts b/apps/sim/lib/core/outbox/retention.ts index 2bc40c8e28d..f2825836da9 100644 --- a/apps/sim/lib/core/outbox/retention.ts +++ b/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,123 @@ const PRUNABLE_OUTBOX_EVENT_TYPES = [ CHAT_INLINE_IMAGE_CLEANUP_EVENT, ] as const -async function pruneCompletedBatch(eventType: string, cutoff: Date): Promise { +/** + * 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 + +/** + * 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: z.string(), id: z.string() }) +const pruneCursorSchema = resumePointSchema.extend({ passStartedAt: z.number() }) + +type ResumePoint = z.infer +type PruneCursor = z.infer + +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 | null> { + const redis = getRedisClient() + if (!redis) return null + try { + const stored = await redis.mget(...PRUNABLE_OUTBOX_EVENT_TYPES.map(pruneCursorKey)) + const cursors = new Map() + 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): Promise { + 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`${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 +185,33 @@ async function pruneCompletedBatch(eventType: string, cutoff: Date): Promise { const cutoff = new Date(now.getTime() - COMPLETED_OUTBOX_RETENTION_MS) + const cursors = await readPruneCursors(now) + const nextCursors = new Map() 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 From d257370e54b6d5e810fd40e0f4d4c4ccee5695fc Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 29 Oct 2026 11:31:11 -0700 Subject: [PATCH 2/2] fix(outbox): start a pass from the oldest row when its stored cursor names no real time A cursor whose JSON parsed but whose creation time was not a calendar time reached the batch's ::timestamp cast and failed the run, which stopped every later type's prune until the cursor expired. The cursor's creation time is now checked as PostgreSQL's ISO text with a calendar round trip, so such a cursor is dropped like an unreadable one and the pass starts from the oldest row. --- apps/sim/lib/core/outbox/retention.integration.ts | 14 ++++++++++++++ apps/sim/lib/core/outbox/retention.ts | 13 ++++++++++++- 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/apps/sim/lib/core/outbox/retention.integration.ts b/apps/sim/lib/core/outbox/retention.integration.ts index e97e7a9a9b3..7df03b3c014 100644 --- a/apps/sim/lib/core/outbox/retention.integration.ts +++ b/apps/sim/lib/core/outbox/retention.integration.ts @@ -260,6 +260,20 @@ describe('completed outbox retention in PostgreSQL', () => { }, 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) diff --git a/apps/sim/lib/core/outbox/retention.ts b/apps/sim/lib/core/outbox/retention.ts index f2825836da9..1de774fcd50 100644 --- a/apps/sim/lib/core/outbox/retention.ts +++ b/apps/sim/lib/core/outbox/retention.ts @@ -67,11 +67,22 @@ const PRUNABLE_OUTBOX_EVENT_TYPES = [ */ 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: z.string(), id: z.string() }) +const resumePointSchema = z.object({ createdAt: postgresTimestampSchema, id: z.string() }) const pruneCursorSchema = resumePointSchema.extend({ passStartedAt: z.number() }) type ResumePoint = z.infer