diff --git a/apps/sim/lib/core/outbox/retention.integration.ts b/apps/sim/lib/core/outbox/retention.integration.ts index 0aff3a09b97..7df03b3c014 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,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()) + }) + }) }) diff --git a/apps/sim/lib/core/outbox/retention.ts b/apps/sim/lib/core/outbox/retention.ts index 2bc40c8e28d..1de774fcd50 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,134 @@ 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 + +/** + * `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 +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 +196,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