mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-01 14:59:19 +08:00
fix(large-refs): cleanup based on table read (#4716)
* fix(large-refs): cleanup based on table read * address comments * address comments * bubble up storage ref errors * cleanup code * do not attempt blob deletion for infra outage * cleanup dup helper
This commit is contained in:
Vikhyath Mondreti
committed by
GitHub
GitHub
parent
786c6f0607
commit
209ca5f121
@@ -15,8 +15,16 @@ import { EXECUTION_RESOURCE_LIMIT_CODE } from '@/lib/execution/resource-errors'
|
||||
import type { ExecutionResult } from '@/lib/workflows/types'
|
||||
import { createHttpResponseFromBlock, workflowHasResponseBlock } from '@/lib/workflows/utils'
|
||||
|
||||
const { mockDownloadFile, mockUploadFile, uploadedFiles } = vi.hoisted(() => ({
|
||||
const {
|
||||
mockAddLargeValueReference,
|
||||
mockDownloadFile,
|
||||
mockRegisterLargeValueOwner,
|
||||
mockUploadFile,
|
||||
uploadedFiles,
|
||||
} = vi.hoisted(() => ({
|
||||
mockAddLargeValueReference: vi.fn(),
|
||||
mockDownloadFile: vi.fn(),
|
||||
mockRegisterLargeValueOwner: vi.fn(),
|
||||
mockUploadFile: vi.fn(),
|
||||
uploadedFiles: new Map<string, Buffer>(),
|
||||
}))
|
||||
@@ -35,6 +43,11 @@ vi.mock('@/lib/uploads', () => ({
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
|
||||
addLargeValueReference: mockAddLargeValueReference,
|
||||
registerLargeValueOwner: mockRegisterLargeValueOwner,
|
||||
}))
|
||||
|
||||
function buildExecutionResult(overrides: Partial<ExecutionResult> = {}): ExecutionResult {
|
||||
return {
|
||||
success: true,
|
||||
@@ -66,6 +79,8 @@ describe('Response block gating by auth type', () => {
|
||||
vi.clearAllMocks()
|
||||
clearLargeValueCacheForTests()
|
||||
uploadedFiles.clear()
|
||||
mockAddLargeValueReference.mockResolvedValue(undefined)
|
||||
mockRegisterLargeValueOwner.mockResolvedValue(true)
|
||||
mockUploadFile.mockImplementation(async ({ customKey, file }) => {
|
||||
uploadedFiles.set(customKey, file)
|
||||
return { key: customKey }
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
interface CleanupRow {
|
||||
id: string
|
||||
files: unknown
|
||||
}
|
||||
|
||||
interface CapturedBatchDeleteOptions {
|
||||
selectChunk: (chunkIds: string[], limit: number) => Promise<unknown>
|
||||
onBatch?: (rows: CleanupRow[]) => Promise<void>
|
||||
batchSize?: number
|
||||
maxBatches?: number
|
||||
totalRowLimit?: number
|
||||
}
|
||||
|
||||
const {
|
||||
mockAnd,
|
||||
mockBatchDeleteByWorkspaceAndTimestamp,
|
||||
mockChunkedBatchDelete,
|
||||
mockDeleteFileMetadata,
|
||||
mockDeleteFiles,
|
||||
mockEq,
|
||||
mockExecute,
|
||||
mockFrom,
|
||||
mockInArray,
|
||||
mockIsNull,
|
||||
mockLeftJoin,
|
||||
mockLimit,
|
||||
mockLt,
|
||||
mockMarkLargeValuesDeleted,
|
||||
mockNotInArray,
|
||||
mockOr,
|
||||
mockOrderBy,
|
||||
mockPruneLargeValueMetadata,
|
||||
mockSelect,
|
||||
mockTask,
|
||||
mockWhere,
|
||||
} = vi.hoisted(() => {
|
||||
const mockLimit = vi.fn(async () => [])
|
||||
const mockOrderBy = vi.fn(() => ({ limit: mockLimit }))
|
||||
const mockWhere = vi.fn(() => ({ limit: mockLimit, orderBy: mockOrderBy }))
|
||||
const mockLeftJoin = vi.fn(() => ({ where: mockWhere }))
|
||||
const mockFrom = vi.fn(() => ({ leftJoin: mockLeftJoin, where: mockWhere }))
|
||||
const mockSelect = vi.fn(() => ({ from: mockFrom }))
|
||||
|
||||
return {
|
||||
mockAnd: vi.fn((...args: unknown[]) => ({ op: 'and', args })),
|
||||
mockBatchDeleteByWorkspaceAndTimestamp: vi.fn(async () => ({
|
||||
table: 'job',
|
||||
deleted: 0,
|
||||
failed: 0,
|
||||
})),
|
||||
mockChunkedBatchDelete: vi.fn(),
|
||||
mockDeleteFileMetadata: vi.fn(async () => true),
|
||||
mockDeleteFiles: vi.fn(async () => ({ deleted: 2, failed: [] })),
|
||||
mockEq: vi.fn((...args: unknown[]) => ({ op: 'eq', args })),
|
||||
mockExecute: vi.fn(),
|
||||
mockFrom,
|
||||
mockInArray: vi.fn((...args: unknown[]) => ({ op: 'inArray', args })),
|
||||
mockIsNull: vi.fn((...args: unknown[]) => ({ op: 'isNull', args })),
|
||||
mockLeftJoin,
|
||||
mockLimit,
|
||||
mockLt: vi.fn((...args: unknown[]) => ({ op: 'lt', args })),
|
||||
mockMarkLargeValuesDeleted: vi.fn(async () => undefined),
|
||||
mockNotInArray: vi.fn((...args: unknown[]) => ({ op: 'notInArray', args })),
|
||||
mockOr: vi.fn((...args: unknown[]) => ({ op: 'or', args })),
|
||||
mockOrderBy,
|
||||
mockPruneLargeValueMetadata: vi.fn(async () => ({
|
||||
referencesDeleted: 0,
|
||||
dependenciesDeleted: 0,
|
||||
tombstonesDeleted: 0,
|
||||
})),
|
||||
mockSelect,
|
||||
mockTask: vi.fn((config: unknown) => config),
|
||||
mockWhere,
|
||||
}
|
||||
})
|
||||
|
||||
vi.mock('@sim/db', () => ({
|
||||
db: {
|
||||
execute: mockExecute,
|
||||
select: mockSelect,
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@sim/db/schema', () => ({
|
||||
executionLargeValueDependencies: {
|
||||
childKey: 'executionLargeValueDependencies.childKey',
|
||||
parentKey: 'executionLargeValueDependencies.parentKey',
|
||||
workspaceId: 'executionLargeValueDependencies.workspaceId',
|
||||
},
|
||||
executionLargeValueReferences: {
|
||||
executionId: 'executionLargeValueReferences.executionId',
|
||||
key: 'executionLargeValueReferences.key',
|
||||
source: 'executionLargeValueReferences.source',
|
||||
},
|
||||
executionLargeValues: {
|
||||
createdAt: 'executionLargeValues.createdAt',
|
||||
deletedAt: 'executionLargeValues.deletedAt',
|
||||
key: 'executionLargeValues.key',
|
||||
workspaceId: 'executionLargeValues.workspaceId',
|
||||
},
|
||||
jobExecutionLogs: {
|
||||
startedAt: 'jobExecutionLogs.startedAt',
|
||||
workspaceId: 'jobExecutionLogs.workspaceId',
|
||||
},
|
||||
pausedExecutions: {
|
||||
executionId: 'pausedExecutions.executionId',
|
||||
status: 'pausedExecutions.status',
|
||||
},
|
||||
workspaceFiles: {
|
||||
context: 'workspaceFiles.context',
|
||||
deletedAt: 'workspaceFiles.deletedAt',
|
||||
key: 'workspaceFiles.key',
|
||||
uploadedAt: 'workspaceFiles.uploadedAt',
|
||||
workspaceId: 'workspaceFiles.workspaceId',
|
||||
},
|
||||
workflowExecutionLogs: {
|
||||
executionData: 'workflowExecutionLogs.executionData',
|
||||
executionId: 'workflowExecutionLogs.executionId',
|
||||
files: 'workflowExecutionLogs.files',
|
||||
id: 'workflowExecutionLogs.id',
|
||||
startedAt: 'workflowExecutionLogs.startedAt',
|
||||
workspaceId: 'workflowExecutionLogs.workspaceId',
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@sim/logger', () => ({
|
||||
createLogger: vi.fn(() => ({
|
||||
error: vi.fn(),
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
})),
|
||||
}))
|
||||
|
||||
vi.mock('@trigger.dev/sdk', () => ({ task: mockTask }))
|
||||
|
||||
vi.mock('drizzle-orm', () => ({
|
||||
and: mockAnd,
|
||||
asc: vi.fn((column: unknown) => ({ op: 'asc', column })),
|
||||
eq: mockEq,
|
||||
inArray: mockInArray,
|
||||
isNull: mockIsNull,
|
||||
lt: mockLt,
|
||||
notInArray: mockNotInArray,
|
||||
or: mockOr,
|
||||
sql: vi.fn((strings: TemplateStringsArray, ...values: unknown[]) => ({ strings, values })),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/cleanup/batch-delete', () => ({
|
||||
batchDeleteByWorkspaceAndTimestamp: mockBatchDeleteByWorkspaceAndTimestamp,
|
||||
chunkArray: (items: string[], size: number) => {
|
||||
const chunks: string[][] = []
|
||||
for (let index = 0; index < items.length; index += size) {
|
||||
chunks.push(items.slice(index, index + size))
|
||||
}
|
||||
return chunks
|
||||
},
|
||||
chunkedBatchDelete: mockChunkedBatchDelete,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
|
||||
LIVE_PAUSED_REFERENCE_STATUSES: ['paused', 'partially_resumed', 'cancelling'],
|
||||
markLargeValuesDeleted: mockMarkLargeValuesDeleted,
|
||||
pruneLargeValueMetadata: mockPruneLargeValueMetadata,
|
||||
unreferencedLargeValuePredicate: vi.fn(() => ({ op: 'unreferencedLargeValuePredicate' })),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/logs/execution/snapshot/service', () => ({
|
||||
snapshotService: {
|
||||
cleanupOrphanedSnapshots: vi.fn(async () => 0),
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/uploads', () => ({
|
||||
isUsingCloudStorage: vi.fn(() => true),
|
||||
StorageService: {
|
||||
deleteFiles: mockDeleteFiles,
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/uploads/server/metadata', () => ({
|
||||
deleteFileMetadata: mockDeleteFileMetadata,
|
||||
}))
|
||||
|
||||
import { cleanupLogsTask, runCleanupLogs } from '@/background/cleanup-logs'
|
||||
|
||||
describe('cleanup logs worker', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
mockChunkedBatchDelete.mockImplementation(async (options: CapturedBatchDeleteOptions) => {
|
||||
await options.selectChunk(['workspace-1'], 500)
|
||||
await options.onBatch?.([
|
||||
{
|
||||
id: 'log-1',
|
||||
files: [
|
||||
{ key: 'execution-file-a' },
|
||||
{ key: 'execution-file-a' },
|
||||
{ key: 'execution-file-b' },
|
||||
],
|
||||
},
|
||||
])
|
||||
return { table: 'workflow_execution_logs', deleted: 1, failed: 0 }
|
||||
})
|
||||
})
|
||||
|
||||
it('cleans logs without selecting execution_data or scanning refs', async () => {
|
||||
await runCleanupLogs({
|
||||
label: 'free/1',
|
||||
plan: 'free',
|
||||
retentionHours: 720,
|
||||
workspaceIds: ['workspace-1'],
|
||||
})
|
||||
|
||||
expect(mockChunkedBatchDelete).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
batchSize: 500,
|
||||
maxBatches: 50,
|
||||
totalRowLimit: 25_000,
|
||||
})
|
||||
)
|
||||
expect(mockSelect).toHaveBeenCalledWith({
|
||||
id: 'workflowExecutionLogs.id',
|
||||
files: 'workflowExecutionLogs.files',
|
||||
})
|
||||
expect(mockExecute).not.toHaveBeenCalled()
|
||||
expect(mockDeleteFiles).toHaveBeenCalledWith(
|
||||
['execution-file-a', 'execution-file-b'],
|
||||
'execution'
|
||||
)
|
||||
expect(mockDeleteFileMetadata).toHaveBeenCalledTimes(2)
|
||||
expect(mockPruneLargeValueMetadata).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ workspaceIds: ['workspace-1'] })
|
||||
)
|
||||
expect(mockBatchDeleteByWorkspaceAndTimestamp).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('does not count large values as deleted when deleted_at marking fails', async () => {
|
||||
const largeValueKey =
|
||||
'execution/workspace-1/workflow-1/execution-1/large-value-lv_abcdefghijkl.json'
|
||||
mockLimit.mockResolvedValueOnce([]).mockResolvedValueOnce([{ key: largeValueKey }])
|
||||
mockDeleteFiles
|
||||
.mockResolvedValueOnce({ deleted: 2, failed: [] })
|
||||
.mockResolvedValueOnce({ deleted: 1, failed: [] })
|
||||
mockMarkLargeValuesDeleted.mockRejectedValueOnce(new Error('db unavailable'))
|
||||
|
||||
await runCleanupLogs({
|
||||
label: 'free/1',
|
||||
plan: 'free',
|
||||
retentionHours: 720,
|
||||
workspaceIds: ['workspace-1'],
|
||||
})
|
||||
|
||||
expect(mockMarkLargeValuesDeleted).toHaveBeenCalledWith([largeValueKey])
|
||||
expect(mockDeleteFileMetadata).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('cleans legacy large values from file metadata without selecting execution_data', async () => {
|
||||
const legacyKey =
|
||||
'execution/workspace-1/workflow-1/execution-1/large-value-lv_abcdefghijkl.json'
|
||||
mockLimit
|
||||
.mockResolvedValueOnce([])
|
||||
.mockResolvedValueOnce([])
|
||||
.mockResolvedValueOnce([{ key: legacyKey }])
|
||||
mockDeleteFiles
|
||||
.mockResolvedValueOnce({ deleted: 2, failed: [] })
|
||||
.mockResolvedValueOnce({ deleted: 1, failed: [] })
|
||||
|
||||
await runCleanupLogs({
|
||||
label: 'free/1',
|
||||
plan: 'free',
|
||||
retentionHours: 720,
|
||||
workspaceIds: ['workspace-1'],
|
||||
})
|
||||
|
||||
expect(mockSelect).toHaveBeenCalledWith({
|
||||
id: 'workflowExecutionLogs.id',
|
||||
files: 'workflowExecutionLogs.files',
|
||||
})
|
||||
expect(mockSelect).not.toHaveBeenCalledWith(
|
||||
expect.objectContaining({ executionData: expect.anything() })
|
||||
)
|
||||
const legacyWhereArgs = mockAnd.mock.calls
|
||||
.flat()
|
||||
.filter((arg): arg is { strings: string[] } => {
|
||||
return (
|
||||
typeof arg === 'object' &&
|
||||
arg !== null &&
|
||||
Array.isArray((arg as { strings?: unknown }).strings)
|
||||
)
|
||||
})
|
||||
.map((arg) => arg.strings.join(' '))
|
||||
.join(' ')
|
||||
expect(legacyWhereArgs).toContain('FROM ')
|
||||
expect(legacyWhereArgs).toContain("ref.source = 'execution_log'")
|
||||
expect(legacyWhereArgs).toContain("ref.source = 'paused_snapshot'")
|
||||
expect(legacyWhereArgs).toContain('dependency.child_key')
|
||||
expect(mockDeleteFiles).toHaveBeenLastCalledWith([legacyKey], 'execution')
|
||||
expect(mockDeleteFileMetadata).toHaveBeenCalledWith(legacyKey)
|
||||
})
|
||||
|
||||
it('caps Trigger.dev concurrency for log cleanup tasks', () => {
|
||||
expect(cleanupLogsTask).toMatchObject({
|
||||
queue: { concurrencyLimit: 2 },
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -1,8 +1,16 @@
|
||||
import { db } from '@sim/db'
|
||||
import { jobExecutionLogs, pausedExecutions, workflowExecutionLogs } from '@sim/db/schema'
|
||||
import {
|
||||
executionLargeValueDependencies,
|
||||
executionLargeValueReferences,
|
||||
executionLargeValues,
|
||||
jobExecutionLogs,
|
||||
pausedExecutions,
|
||||
workflowExecutionLogs,
|
||||
workspaceFiles,
|
||||
} from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { task } from '@trigger.dev/sdk'
|
||||
import { and, eq, inArray, isNull, lt, notInArray, or, sql } from 'drizzle-orm'
|
||||
import { and, asc, eq, inArray, isNull, lt, notInArray, or, sql } from 'drizzle-orm'
|
||||
import type { CleanupJobPayload } from '@/lib/billing/cleanup-dispatcher'
|
||||
import {
|
||||
batchDeleteByWorkspaceAndTimestamp,
|
||||
@@ -10,7 +18,12 @@ import {
|
||||
chunkedBatchDelete,
|
||||
type TableCleanupResult,
|
||||
} from '@/lib/cleanup/batch-delete'
|
||||
import { collectLargeValueKeys } from '@/lib/execution/payloads/large-execution-value'
|
||||
import {
|
||||
LIVE_PAUSED_REFERENCE_STATUSES,
|
||||
markLargeValuesDeleted,
|
||||
pruneLargeValueMetadata,
|
||||
unreferencedLargeValuePredicate,
|
||||
} from '@/lib/execution/payloads/large-value-metadata'
|
||||
import { snapshotService } from '@/lib/logs/execution/snapshot/service'
|
||||
import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
|
||||
import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
|
||||
@@ -21,90 +34,332 @@ interface FileDeleteStats {
|
||||
filesTotal: number
|
||||
filesDeleted: number
|
||||
filesDeleteFailed: number
|
||||
}
|
||||
|
||||
const WORKFLOW_LOG_CLEANUP_BATCH_SIZE = 500
|
||||
const WORKFLOW_LOG_CLEANUP_MAX_BATCHES = 50
|
||||
const WORKFLOW_LOG_CLEANUP_ROW_LIMIT =
|
||||
WORKFLOW_LOG_CLEANUP_BATCH_SIZE * WORKFLOW_LOG_CLEANUP_MAX_BATCHES
|
||||
const LOG_CLEANUP_CONCURRENCY_LIMIT = 2
|
||||
const LARGE_VALUE_CLEANUP_BATCH_SIZE = 500
|
||||
const LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT = 5_000
|
||||
const LARGE_VALUE_CLEANUP_GRACE_HOURS = 7 * 24
|
||||
const LEGACY_LARGE_VALUE_CLEANUP_GRACE_HOURS = 30 * 24
|
||||
const LARGE_VALUE_TOMBSTONE_RETENTION_HOURS = 30 * 24
|
||||
|
||||
async function deleteExecutionFiles(files: unknown, stats: FileDeleteStats): Promise<void> {
|
||||
if (!isUsingCloudStorage() || !files || !Array.isArray(files)) return
|
||||
|
||||
const keys = Array.from(
|
||||
new Set(files.filter((f) => f && typeof f === 'object' && f.key).map((f) => f.key as string))
|
||||
)
|
||||
stats.filesTotal += keys.length
|
||||
if (keys.length === 0) return
|
||||
|
||||
let result: Awaited<ReturnType<typeof StorageService.deleteFiles>>
|
||||
try {
|
||||
result = await StorageService.deleteFiles(keys, 'execution')
|
||||
} catch (error) {
|
||||
stats.filesDeleteFailed += keys.length
|
||||
logger.error('Failed to bulk delete execution files:', { error })
|
||||
return
|
||||
}
|
||||
|
||||
const failedKeys = new Set(result.failed.map(({ key }) => key))
|
||||
stats.filesDeleted += result.deleted
|
||||
stats.filesDeleteFailed += result.failed.length
|
||||
|
||||
for (const { key, error } of result.failed) {
|
||||
logger.error(`Failed to delete file ${key}:`, { error })
|
||||
}
|
||||
for (const key of keys) {
|
||||
if (failedKeys.has(key)) continue
|
||||
try {
|
||||
await deleteFileMetadata(key)
|
||||
} catch (metadataError) {
|
||||
stats.filesDeleteFailed++
|
||||
logger.error(`Failed to delete file metadata ${key}:`, { metadataError })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
interface LargeValueCleanupStats {
|
||||
largeValuesTotal: number
|
||||
largeValuesDeleted: number
|
||||
largeValuesDeleteFailed: number
|
||||
}
|
||||
|
||||
const RESUMABLE_PAUSED_STATUSES = ['paused', 'partially_resumed', 'cancelling']
|
||||
|
||||
/** Caps the per-row predicate cost: keys-per-row is `O(chunk)` not `O(uniqueKeys)`. */
|
||||
const REFERENCE_CHECK_KEY_CHUNK_SIZE = 200
|
||||
|
||||
/**
|
||||
* One `LATERAL unnest` scan per chunk replaces N per-key sequential scans
|
||||
* (each detoasting the entire JSONB column). Substring semantics identical.
|
||||
*/
|
||||
async function filterLargeValueKeysWithoutRetainedReferences(
|
||||
keys: string[],
|
||||
deletedLogIds: string[]
|
||||
): Promise<string[]> {
|
||||
if (keys.length === 0 || deletedLogIds.length === 0) return []
|
||||
|
||||
const uniqueKeys = Array.from(new Set(keys))
|
||||
const workspaceIds = Array.from(
|
||||
new Set(
|
||||
uniqueKeys
|
||||
.map((key) => key.split('/')[1])
|
||||
.filter((workspaceId): workspaceId is string => Boolean(workspaceId))
|
||||
)
|
||||
)
|
||||
if (workspaceIds.length === 0) return []
|
||||
const referencedKeys = new Set<string>()
|
||||
|
||||
for (const keyChunk of chunkArray(uniqueKeys, REFERENCE_CHECK_KEY_CHUNK_SIZE)) {
|
||||
const rows = await db.execute<{ key: string }>(sql`
|
||||
SELECT DISTINCT k.key AS key
|
||||
FROM ${workflowExecutionLogs} AS wel,
|
||||
unnest(${keyChunk}::text[]) AS k(key)
|
||||
WHERE wel.workspace_id = ANY(${workspaceIds}::text[])
|
||||
AND wel.id <> ALL(${deletedLogIds}::text[])
|
||||
AND position(k.key in wel.execution_data::text) > 0
|
||||
`)
|
||||
for (const row of rows) referencedKeys.add(row.key)
|
||||
async function deleteLargeValueKeys(keys: string[]): Promise<{ deleted: number; failed: number }> {
|
||||
if (!isUsingCloudStorage() || keys.length === 0) {
|
||||
return { deleted: 0, failed: 0 }
|
||||
}
|
||||
|
||||
return uniqueKeys.filter((key) => !referencedKeys.has(key))
|
||||
let result: Awaited<ReturnType<typeof StorageService.deleteFiles>>
|
||||
try {
|
||||
result = await StorageService.deleteFiles(keys, 'execution')
|
||||
} catch (error) {
|
||||
logger.error('Failed to bulk delete large execution values:', { error })
|
||||
return { deleted: 0, failed: keys.length }
|
||||
}
|
||||
|
||||
const failedKeys = new Set(result.failed.map(({ key }) => key))
|
||||
const deletedKeys = keys.filter((key) => !failedKeys.has(key))
|
||||
|
||||
if (deletedKeys.length > 0) {
|
||||
try {
|
||||
await markLargeValuesDeleted(deletedKeys)
|
||||
} catch (error) {
|
||||
logger.error('Failed to mark large execution values as deleted:', { error })
|
||||
return { deleted: 0, failed: result.failed.length + deletedKeys.length }
|
||||
}
|
||||
}
|
||||
|
||||
for (const { key, error } of result.failed) {
|
||||
logger.error(`Failed to delete large execution value ${key}:`, { error })
|
||||
}
|
||||
|
||||
for (const key of deletedKeys) {
|
||||
try {
|
||||
await deleteFileMetadata(key)
|
||||
} catch (metadataError) {
|
||||
logger.error(`Failed to delete large execution value metadata ${key}:`, { metadataError })
|
||||
}
|
||||
}
|
||||
|
||||
return { deleted: deletedKeys.length, failed: result.failed.length }
|
||||
}
|
||||
|
||||
async function deleteExecutionFiles(files: unknown, stats: FileDeleteStats): Promise<void> {
|
||||
if (!isUsingCloudStorage() || !files || !Array.isArray(files)) return
|
||||
async function cleanupLargeExecutionValues(
|
||||
workspaceIds: string[],
|
||||
retentionDate: Date,
|
||||
label: string
|
||||
): Promise<LargeValueCleanupStats> {
|
||||
const stats: LargeValueCleanupStats = {
|
||||
largeValuesTotal: 0,
|
||||
largeValuesDeleted: 0,
|
||||
largeValuesDeleteFailed: 0,
|
||||
}
|
||||
if (workspaceIds.length === 0) return stats
|
||||
|
||||
const keys = files.filter((f) => f && typeof f === 'object' && f.key).map((f) => f.key as string)
|
||||
stats.filesTotal += keys.length
|
||||
|
||||
await Promise.all(
|
||||
keys.map(async (key) => {
|
||||
try {
|
||||
await StorageService.deleteFile({ key, context: 'execution' })
|
||||
await deleteFileMetadata(key)
|
||||
stats.filesDeleted++
|
||||
} catch (fileError) {
|
||||
stats.filesDeleteFailed++
|
||||
logger.error(`Failed to delete file ${key}:`, { fileError })
|
||||
}
|
||||
})
|
||||
const largeValueRetentionDate = new Date(
|
||||
retentionDate.getTime() - LARGE_VALUE_CLEANUP_GRACE_HOURS * 60 * 60 * 1000
|
||||
)
|
||||
const workspaceChunks = chunkArray(workspaceIds, 50)
|
||||
let attempted = 0
|
||||
|
||||
for (const chunkIds of workspaceChunks) {
|
||||
while (attempted < LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT) {
|
||||
const limit = Math.min(
|
||||
LARGE_VALUE_CLEANUP_BATCH_SIZE,
|
||||
LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT - attempted
|
||||
)
|
||||
const rows = await db
|
||||
.select({ key: executionLargeValues.key })
|
||||
.from(executionLargeValues)
|
||||
.where(
|
||||
and(
|
||||
inArray(executionLargeValues.workspaceId, chunkIds),
|
||||
isNull(executionLargeValues.deletedAt),
|
||||
lt(executionLargeValues.createdAt, largeValueRetentionDate),
|
||||
unreferencedLargeValuePredicate()
|
||||
)
|
||||
)
|
||||
.orderBy(
|
||||
asc(executionLargeValues.workspaceId),
|
||||
asc(executionLargeValues.createdAt),
|
||||
asc(executionLargeValues.key)
|
||||
)
|
||||
.limit(limit)
|
||||
|
||||
if (rows.length === 0) break
|
||||
|
||||
const keys = rows.map((row) => row.key)
|
||||
stats.largeValuesTotal += keys.length
|
||||
attempted += keys.length
|
||||
const result = await deleteLargeValueKeys(keys)
|
||||
stats.largeValuesDeleted += result.deleted
|
||||
stats.largeValuesDeleteFailed += result.failed
|
||||
|
||||
if (result.deleted === 0) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if (attempted >= LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT) break
|
||||
}
|
||||
|
||||
logger.info(
|
||||
`[${label}/execution_large_values] Complete: ${stats.largeValuesDeleted}/${stats.largeValuesTotal} deleted, ${stats.largeValuesDeleteFailed} failed`
|
||||
)
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
async function deleteLargeValueStorageKeys(keys: string[], stats: FileDeleteStats): Promise<void> {
|
||||
if (!isUsingCloudStorage() || keys.length === 0) return
|
||||
async function cleanupLegacyLargeExecutionValues(
|
||||
workspaceIds: string[],
|
||||
retentionDate: Date,
|
||||
label: string
|
||||
): Promise<LargeValueCleanupStats> {
|
||||
const stats: LargeValueCleanupStats = {
|
||||
largeValuesTotal: 0,
|
||||
largeValuesDeleted: 0,
|
||||
largeValuesDeleteFailed: 0,
|
||||
}
|
||||
if (workspaceIds.length === 0) return stats
|
||||
|
||||
const uniqueKeys = Array.from(new Set(keys))
|
||||
stats.largeValuesTotal += uniqueKeys.length
|
||||
|
||||
await Promise.all(
|
||||
uniqueKeys.map(async (key) => {
|
||||
try {
|
||||
await StorageService.deleteFile({ key, context: 'execution' })
|
||||
await deleteFileMetadata(key)
|
||||
stats.largeValuesDeleted++
|
||||
} catch (error) {
|
||||
stats.largeValuesDeleteFailed++
|
||||
logger.error(`Failed to delete large execution value ${key}:`, { error })
|
||||
}
|
||||
})
|
||||
const legacyRetentionDate = new Date(
|
||||
retentionDate.getTime() - LEGACY_LARGE_VALUE_CLEANUP_GRACE_HOURS * 60 * 60 * 1000
|
||||
)
|
||||
const workspaceChunks = chunkArray(workspaceIds, 50)
|
||||
let attempted = 0
|
||||
|
||||
for (const chunkIds of workspaceChunks) {
|
||||
while (attempted < LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT) {
|
||||
const limit = Math.min(
|
||||
LARGE_VALUE_CLEANUP_BATCH_SIZE,
|
||||
LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT - attempted
|
||||
)
|
||||
const rows = await db
|
||||
.select({ key: workspaceFiles.key })
|
||||
.from(workspaceFiles)
|
||||
.where(
|
||||
and(
|
||||
inArray(workspaceFiles.workspaceId, chunkIds),
|
||||
eq(workspaceFiles.context, 'execution'),
|
||||
isNull(workspaceFiles.deletedAt),
|
||||
lt(workspaceFiles.uploadedAt, legacyRetentionDate),
|
||||
sql`${workspaceFiles.key} LIKE 'execution/%/%/%/large-value-lv_%.json'`,
|
||||
sql`NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValues} AS registered_value
|
||||
WHERE registered_value.key = ${workspaceFiles.key}
|
||||
)`,
|
||||
sql`NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValueReferences} AS ref
|
||||
WHERE ref.key = ${workspaceFiles.key}
|
||||
AND (
|
||||
(
|
||||
ref.source = 'execution_log'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS ref_wel
|
||||
WHERE ref_wel.execution_id = ref.execution_id
|
||||
)
|
||||
)
|
||||
OR (
|
||||
ref.source = 'paused_snapshot'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS ref_pe
|
||||
WHERE ref_pe.execution_id = ref.execution_id
|
||||
AND ref_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)
|
||||
)
|
||||
)
|
||||
)`,
|
||||
sql`NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValueDependencies} AS dependency
|
||||
INNER JOIN ${executionLargeValues} AS parent_value
|
||||
ON parent_value.key = dependency.parent_key
|
||||
AND parent_value.deleted_at IS NULL
|
||||
WHERE dependency.child_key = ${workspaceFiles.key}
|
||||
AND dependency.workspace_id = ${workspaceFiles.workspaceId}
|
||||
AND (
|
||||
EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS parent_owner_wel
|
||||
WHERE parent_owner_wel.execution_id = parent_value.owner_execution_id
|
||||
)
|
||||
OR EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS parent_owner_pe
|
||||
WHERE parent_owner_pe.execution_id = parent_value.owner_execution_id
|
||||
AND parent_owner_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)
|
||||
OR EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValueReferences} AS parent_ref
|
||||
WHERE parent_ref.key = parent_value.key
|
||||
AND (
|
||||
(
|
||||
parent_ref.source = 'execution_log'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS parent_ref_wel
|
||||
WHERE parent_ref_wel.execution_id = parent_ref.execution_id
|
||||
)
|
||||
)
|
||||
OR (
|
||||
parent_ref.source = 'paused_snapshot'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS parent_ref_pe
|
||||
WHERE parent_ref_pe.execution_id = parent_ref.execution_id
|
||||
AND parent_ref_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
)`,
|
||||
sql`NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS owner_wel
|
||||
WHERE owner_wel.execution_id = split_part(${workspaceFiles.key}, '/', 4)
|
||||
)`,
|
||||
sql`NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS pe
|
||||
WHERE pe.execution_id = split_part(${workspaceFiles.key}, '/', 4)
|
||||
AND pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)`
|
||||
)
|
||||
)
|
||||
.orderBy(
|
||||
asc(workspaceFiles.workspaceId),
|
||||
asc(workspaceFiles.uploadedAt),
|
||||
asc(workspaceFiles.key)
|
||||
)
|
||||
.limit(limit)
|
||||
|
||||
if (rows.length === 0) break
|
||||
|
||||
const keys = rows.map((row) => row.key)
|
||||
stats.largeValuesTotal += keys.length
|
||||
attempted += keys.length
|
||||
const result = await deleteLargeValueKeys(keys)
|
||||
stats.largeValuesDeleted += result.deleted
|
||||
stats.largeValuesDeleteFailed += result.failed
|
||||
|
||||
if (result.deleted === 0) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if (attempted >= LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT) break
|
||||
}
|
||||
|
||||
logger.info(
|
||||
`[${label}/legacy_execution_large_values] Complete: ${stats.largeValuesDeleted}/${stats.largeValuesTotal} deleted, ${stats.largeValuesDeleteFailed} failed`
|
||||
)
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
async function cleanupLargeValueMetadata(workspaceIds: string[], label: string): Promise<void> {
|
||||
try {
|
||||
const tombstonesDeletedBefore = new Date(
|
||||
Date.now() - LARGE_VALUE_TOMBSTONE_RETENTION_HOURS * 60 * 60 * 1000
|
||||
)
|
||||
const result = await pruneLargeValueMetadata({ workspaceIds, tombstonesDeletedBefore })
|
||||
logger.info(
|
||||
`[${label}/execution_large_value_metadata] Pruned ${result.referencesDeleted} stale references, ${result.dependenciesDeleted} dependencies, ${result.tombstonesDeleted} tombstones`
|
||||
)
|
||||
} catch (error) {
|
||||
logger.error(`[${label}/execution_large_value_metadata] Failed to prune metadata`, { error })
|
||||
}
|
||||
}
|
||||
|
||||
async function cleanupWorkflowExecutionLogs(
|
||||
@@ -116,9 +371,6 @@ async function cleanupWorkflowExecutionLogs(
|
||||
filesTotal: 0,
|
||||
filesDeleted: 0,
|
||||
filesDeleteFailed: 0,
|
||||
largeValuesTotal: 0,
|
||||
largeValuesDeleted: 0,
|
||||
largeValuesDeleteFailed: 0,
|
||||
}
|
||||
|
||||
const dbStats = await chunkedBatchDelete({
|
||||
@@ -129,9 +381,6 @@ async function cleanupWorkflowExecutionLogs(
|
||||
db
|
||||
.select({
|
||||
id: workflowExecutionLogs.id,
|
||||
workspaceId: workflowExecutionLogs.workspaceId,
|
||||
executionId: workflowExecutionLogs.executionId,
|
||||
executionData: workflowExecutionLogs.executionData,
|
||||
files: workflowExecutionLogs.files,
|
||||
})
|
||||
.from(workflowExecutionLogs)
|
||||
@@ -145,24 +394,19 @@ async function cleanupWorkflowExecutionLogs(
|
||||
lt(workflowExecutionLogs.startedAt, retentionDate),
|
||||
or(
|
||||
isNull(pausedExecutions.status),
|
||||
notInArray(pausedExecutions.status, RESUMABLE_PAUSED_STATUSES)
|
||||
notInArray(pausedExecutions.status, [...LIVE_PAUSED_REFERENCE_STATUSES])
|
||||
)
|
||||
)
|
||||
)
|
||||
.limit(limit),
|
||||
onBatch: async (rows) => {
|
||||
const deletedLogIds = rows.map((row) => row.id)
|
||||
const largeValueKeys = rows.flatMap((row) => collectLargeValueKeys(row.executionData))
|
||||
const unreferencedLargeValueKeys = await filterLargeValueKeysWithoutRetainedReferences(
|
||||
largeValueKeys,
|
||||
deletedLogIds
|
||||
)
|
||||
|
||||
for (const row of rows) {
|
||||
await deleteExecutionFiles(row.files, fileStats)
|
||||
}
|
||||
await deleteLargeValueStorageKeys(unreferencedLargeValueKeys, fileStats)
|
||||
},
|
||||
batchSize: WORKFLOW_LOG_CLEANUP_BATCH_SIZE,
|
||||
maxBatches: WORKFLOW_LOG_CLEANUP_MAX_BATCHES,
|
||||
totalRowLimit: WORKFLOW_LOG_CLEANUP_ROW_LIMIT,
|
||||
})
|
||||
|
||||
return { ...dbStats, ...fileStats }
|
||||
@@ -200,9 +444,19 @@ export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void>
|
||||
logger.info(
|
||||
`[${label}] workflow_execution_logs files: ${workflowResults.filesDeleted}/${workflowResults.filesTotal} deleted, ${workflowResults.filesDeleteFailed} failed`
|
||||
)
|
||||
const largeValueResults = await cleanupLargeExecutionValues(workspaceIds, retentionDate, label)
|
||||
logger.info(
|
||||
`[${label}] workflow_execution_logs large values: ${workflowResults.largeValuesDeleted}/${workflowResults.largeValuesTotal} deleted, ${workflowResults.largeValuesDeleteFailed} failed`
|
||||
`[${label}] execution_large_values: ${largeValueResults.largeValuesDeleted}/${largeValueResults.largeValuesTotal} deleted, ${largeValueResults.largeValuesDeleteFailed} failed`
|
||||
)
|
||||
const legacyLargeValueResults = await cleanupLegacyLargeExecutionValues(
|
||||
workspaceIds,
|
||||
retentionDate,
|
||||
label
|
||||
)
|
||||
logger.info(
|
||||
`[${label}] legacy_execution_large_values: ${legacyLargeValueResults.largeValuesDeleted}/${legacyLargeValueResults.largeValuesTotal} deleted, ${legacyLargeValueResults.largeValuesDeleteFailed} failed`
|
||||
)
|
||||
await cleanupLargeValueMetadata(workspaceIds, label)
|
||||
|
||||
await batchDeleteByWorkspaceAndTimestamp({
|
||||
tableDef: jobExecutionLogs,
|
||||
@@ -224,6 +478,6 @@ export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void>
|
||||
export const cleanupLogsTask = task({
|
||||
id: 'cleanup-logs',
|
||||
machine: 'large-1x',
|
||||
queue: { concurrencyLimit: 5 },
|
||||
queue: { concurrencyLimit: LOG_CLEANUP_CONCURRENCY_LIMIT },
|
||||
run: runCleanupLogs,
|
||||
})
|
||||
|
||||
@@ -117,9 +117,10 @@ export async function chunkedBatchDelete<TRow extends { id: string }>({
|
||||
|
||||
const chunks = chunkArray(workspaceIds, workspaceChunkSize)
|
||||
let stoppedEarly = false
|
||||
let attempted = 0
|
||||
|
||||
for (const [chunkIdx, chunkIds] of chunks.entries()) {
|
||||
if (result.deleted + result.failed >= totalRowLimit) {
|
||||
if (attempted >= totalRowLimit) {
|
||||
stoppedEarly = true
|
||||
break
|
||||
}
|
||||
@@ -127,20 +128,24 @@ export async function chunkedBatchDelete<TRow extends { id: string }>({
|
||||
let batchesProcessed = 0
|
||||
let hasMore = true
|
||||
|
||||
while (
|
||||
hasMore &&
|
||||
batchesProcessed < maxBatches &&
|
||||
result.deleted + result.failed < totalRowLimit
|
||||
) {
|
||||
while (hasMore && batchesProcessed < maxBatches && attempted < totalRowLimit) {
|
||||
let rows: TRow[] = []
|
||||
try {
|
||||
rows = await selectChunk(chunkIds, batchSize)
|
||||
const remainingLimit = totalRowLimit - attempted
|
||||
const effectiveBatchSize = Math.min(batchSize, remainingLimit)
|
||||
if (effectiveBatchSize <= 0) {
|
||||
hasMore = false
|
||||
break
|
||||
}
|
||||
|
||||
rows = await selectChunk(chunkIds, effectiveBatchSize)
|
||||
|
||||
if (rows.length === 0) {
|
||||
hasMore = false
|
||||
break
|
||||
}
|
||||
|
||||
attempted += rows.length
|
||||
if (onBatch) await onBatch(rows)
|
||||
|
||||
const ids = rows.map((r) => r.id)
|
||||
@@ -150,7 +155,8 @@ export async function chunkedBatchDelete<TRow extends { id: string }>({
|
||||
.returning({ id: sql`id` })
|
||||
|
||||
result.deleted += deleted.length
|
||||
hasMore = rows.length === batchSize
|
||||
result.failed += rows.length - deleted.length
|
||||
hasMore = rows.length === effectiveBatchSize && attempted < totalRowLimit
|
||||
batchesProcessed++
|
||||
} catch (error) {
|
||||
// Count rows we tried to delete; SELECT-stage errors leave rows=[].
|
||||
|
||||
@@ -0,0 +1,457 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const {
|
||||
mockAnd,
|
||||
mockDelete,
|
||||
mockEq,
|
||||
mockExecute,
|
||||
mockInsert,
|
||||
mockOnConflictDoNothing,
|
||||
mockSelect,
|
||||
mockSelectFrom,
|
||||
mockSelectLimit,
|
||||
mockSelectWhere,
|
||||
mockTransaction,
|
||||
mockTxDelete,
|
||||
mockTxInsert,
|
||||
mockTxSelect,
|
||||
mockTxSelectDistinct,
|
||||
mockTxSelectFrom,
|
||||
mockTxSelectLimit,
|
||||
mockTxSelectWhere,
|
||||
mockTxValues,
|
||||
mockValues,
|
||||
mockWhere,
|
||||
mockTxWhere,
|
||||
mockNotInArray,
|
||||
} = vi.hoisted(() => {
|
||||
const mockOnConflictDoNothing = vi.fn(async () => undefined)
|
||||
const mockValues = vi.fn(() => ({ onConflictDoNothing: mockOnConflictDoNothing }))
|
||||
const mockInsert = vi.fn(() => ({ values: mockValues }))
|
||||
const mockWhere = vi.fn(async () => undefined)
|
||||
const mockDelete = vi.fn(() => ({ where: mockWhere }))
|
||||
const mockSelectLimit = vi.fn(async () => [])
|
||||
const mockSelectWhere = vi.fn(() => ({ limit: mockSelectLimit }))
|
||||
const mockSelectFrom = vi.fn(() => ({ where: mockSelectWhere }))
|
||||
const mockSelect = vi.fn(() => ({ from: mockSelectFrom }))
|
||||
|
||||
const mockTxValues = vi.fn(() => ({ onConflictDoNothing: mockOnConflictDoNothing }))
|
||||
const mockTxInsert = vi.fn(() => ({ values: mockTxValues }))
|
||||
const mockTxWhere = vi.fn(async () => undefined)
|
||||
const mockTxDelete = vi.fn(() => ({ where: mockTxWhere }))
|
||||
const mockTxSelectLimit = vi.fn(async () => [])
|
||||
const mockTxSelectWhere = vi.fn(() => ({ limit: mockTxSelectLimit }))
|
||||
const mockTxSelectFrom = vi.fn(() => ({ where: mockTxSelectWhere }))
|
||||
const mockTxSelect = vi.fn(() => ({ from: mockTxSelectFrom }))
|
||||
const mockTxSelectDistinct = vi.fn(() => ({ from: mockTxSelectFrom }))
|
||||
|
||||
return {
|
||||
mockAnd: vi.fn((...args: unknown[]) => ({ op: 'and', args })),
|
||||
mockDelete,
|
||||
mockEq: vi.fn((...args: unknown[]) => ({ op: 'eq', args })),
|
||||
mockExecute: vi.fn(async () => [{ count: 0 }]),
|
||||
mockInsert,
|
||||
mockNotInArray: vi.fn((...args: unknown[]) => ({ op: 'notInArray', args })),
|
||||
mockOnConflictDoNothing,
|
||||
mockSelect,
|
||||
mockSelectFrom,
|
||||
mockSelectLimit,
|
||||
mockSelectWhere,
|
||||
mockTransaction: vi.fn(async (callback) =>
|
||||
callback({
|
||||
delete: mockTxDelete,
|
||||
insert: mockTxInsert,
|
||||
select: mockTxSelect,
|
||||
selectDistinct: mockTxSelectDistinct,
|
||||
})
|
||||
),
|
||||
mockTxDelete,
|
||||
mockTxInsert,
|
||||
mockTxSelect,
|
||||
mockTxSelectDistinct,
|
||||
mockTxSelectFrom,
|
||||
mockTxSelectLimit,
|
||||
mockTxSelectWhere,
|
||||
mockTxValues,
|
||||
mockValues,
|
||||
mockWhere,
|
||||
mockTxWhere,
|
||||
}
|
||||
})
|
||||
|
||||
vi.mock('@sim/db', () => ({
|
||||
db: {
|
||||
delete: mockDelete,
|
||||
execute: mockExecute,
|
||||
insert: mockInsert,
|
||||
select: mockSelect,
|
||||
transaction: mockTransaction,
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@sim/db/schema', () => ({
|
||||
executionLargeValueDependencies: {
|
||||
childKey: 'executionLargeValueDependencies.childKey',
|
||||
parentKey: 'executionLargeValueDependencies.parentKey',
|
||||
workspaceId: 'executionLargeValueDependencies.workspaceId',
|
||||
},
|
||||
executionLargeValueReferences: {
|
||||
executionId: 'executionLargeValueReferences.executionId',
|
||||
key: 'executionLargeValueReferences.key',
|
||||
source: 'executionLargeValueReferences.source',
|
||||
workspaceId: 'executionLargeValueReferences.workspaceId',
|
||||
},
|
||||
executionLargeValues: {
|
||||
key: 'executionLargeValues.key',
|
||||
ownerExecutionId: 'executionLargeValues.ownerExecutionId',
|
||||
workspaceId: 'executionLargeValues.workspaceId',
|
||||
},
|
||||
pausedExecutions: {
|
||||
executionId: 'pausedExecutions.executionId',
|
||||
status: 'pausedExecutions.status',
|
||||
},
|
||||
workflowExecutionLogs: {
|
||||
executionId: 'workflowExecutionLogs.executionId',
|
||||
},
|
||||
}))
|
||||
|
||||
vi.mock('@sim/logger', () => ({
|
||||
createLogger: vi.fn(() => ({
|
||||
warn: vi.fn(),
|
||||
})),
|
||||
}))
|
||||
|
||||
vi.mock('drizzle-orm', () => ({
|
||||
and: mockAnd,
|
||||
eq: mockEq,
|
||||
inArray: vi.fn((...args: unknown[]) => ({ op: 'inArray', args })),
|
||||
notInArray: mockNotInArray,
|
||||
sql: vi.fn((strings: TemplateStringsArray, ...values: unknown[]) => ({ strings, values })),
|
||||
}))
|
||||
|
||||
import {
|
||||
addLargeValueReference,
|
||||
MAX_LARGE_VALUE_REFERENCES_PER_SCOPE,
|
||||
pruneLargeValueMetadata,
|
||||
registerLargeValueOwner,
|
||||
replaceLargeValueReferences,
|
||||
} from '@/lib/execution/payloads/large-value-metadata'
|
||||
|
||||
function largeValueKey(id: string, executionId = 'source-execution'): string {
|
||||
return `execution/workspace-1/workflow-1/${executionId}/large-value-lv_${id}.json`
|
||||
}
|
||||
|
||||
describe('large value metadata', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
it('registers valid large value owner metadata', async () => {
|
||||
const registered = await registerLargeValueOwner({
|
||||
key: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_abcdefghijkl.json',
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
size: 123.4,
|
||||
})
|
||||
|
||||
expect(registered).toBe(true)
|
||||
expect(mockTxInsert).toHaveBeenCalledOnce()
|
||||
expect(mockTxValues).toHaveBeenCalledWith({
|
||||
key: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_abcdefghijkl.json',
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
ownerExecutionId: 'execution-1',
|
||||
size: 124,
|
||||
})
|
||||
expect(mockOnConflictDoNothing).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('skips malformed owner keys', async () => {
|
||||
const registered = await registerLargeValueOwner({
|
||||
key: 'execution/workspace-1/workflow-1/other-execution/large-value-lv_abcdefghijkl.json',
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
size: 123,
|
||||
})
|
||||
|
||||
expect(registered).toBe(false)
|
||||
expect(mockTxInsert).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('records dependency closure for nested large value refs', async () => {
|
||||
const directKey = largeValueKey('abcdefghijkl')
|
||||
const transitiveKey = largeValueKey('mnopqrstuvwx', 'root-execution')
|
||||
const deepTransitiveKey = largeValueKey('deepqrstuvwx', 'deep-execution')
|
||||
mockTxSelectLimit
|
||||
.mockResolvedValueOnce([{ childKey: transitiveKey }])
|
||||
.mockResolvedValueOnce([{ childKey: deepTransitiveKey }])
|
||||
.mockResolvedValueOnce([])
|
||||
|
||||
const registered = await registerLargeValueOwner(
|
||||
{
|
||||
key: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_zyxwvutsrqpo.json',
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
size: 123,
|
||||
},
|
||||
[directKey]
|
||||
)
|
||||
|
||||
expect(registered).toBe(true)
|
||||
expect(mockTxSelectDistinct).toHaveBeenCalledTimes(3)
|
||||
expect(mockTxValues).toHaveBeenLastCalledWith([
|
||||
{
|
||||
parentKey: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_zyxwvutsrqpo.json',
|
||||
childKey: directKey,
|
||||
workspaceId: 'workspace-1',
|
||||
},
|
||||
{
|
||||
parentKey: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_zyxwvutsrqpo.json',
|
||||
childKey: transitiveKey,
|
||||
workspaceId: 'workspace-1',
|
||||
},
|
||||
{
|
||||
parentKey: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_zyxwvutsrqpo.json',
|
||||
childKey: deepTransitiveKey,
|
||||
workspaceId: 'workspace-1',
|
||||
},
|
||||
])
|
||||
})
|
||||
|
||||
it('chunks dependency writes instead of emitting one oversized VALUES statement', async () => {
|
||||
const keys = Array.from({ length: 501 }, (_, index) =>
|
||||
largeValueKey(`a${index.toString(36).padStart(11, '0')}`)
|
||||
)
|
||||
|
||||
await registerLargeValueOwner(
|
||||
{
|
||||
key: largeValueKey('zyxwvutsrqpo', 'execution-1'),
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
size: 123,
|
||||
},
|
||||
keys
|
||||
)
|
||||
|
||||
expect(mockTxValues).toHaveBeenCalledTimes(3)
|
||||
expect(mockTxValues.mock.calls[1]?.[0]).toHaveLength(500)
|
||||
expect(mockTxValues.mock.calls[2]?.[0]).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('rejects reference sets over the metadata cardinality limit', async () => {
|
||||
const keys = Array.from({ length: MAX_LARGE_VALUE_REFERENCES_PER_SCOPE + 1 }, (_, index) =>
|
||||
largeValueKey(`b${index.toString(36).padStart(11, '0')}`)
|
||||
)
|
||||
|
||||
await expect(
|
||||
registerLargeValueOwner(
|
||||
{
|
||||
key: largeValueKey('zyxwvutsrqpo', 'execution-1'),
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
size: 123,
|
||||
},
|
||||
keys
|
||||
)
|
||||
).rejects.toThrow('exceeding the limit')
|
||||
})
|
||||
|
||||
it('limits dependency closure reads to the remaining reference budget', async () => {
|
||||
const directKey = largeValueKey('a00000000000')
|
||||
mockTxSelectLimit.mockResolvedValueOnce(
|
||||
Array.from({ length: MAX_LARGE_VALUE_REFERENCES_PER_SCOPE }, (_, index) => ({
|
||||
childKey: largeValueKey(`c${index.toString(36).padStart(11, '0')}`),
|
||||
}))
|
||||
)
|
||||
|
||||
await expect(
|
||||
registerLargeValueOwner(
|
||||
{
|
||||
key: largeValueKey('zyxwvutsrqpo', 'execution-1'),
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
size: 123,
|
||||
},
|
||||
[directKey]
|
||||
)
|
||||
).rejects.toThrow('Large value dependency closure exceeds the limit')
|
||||
|
||||
expect(mockTxSelectLimit).toHaveBeenCalledWith(MAX_LARGE_VALUE_REFERENCES_PER_SCOPE)
|
||||
})
|
||||
|
||||
it('filters known dependency children before applying the remaining reference budget', async () => {
|
||||
const directKeys = Array.from({ length: MAX_LARGE_VALUE_REFERENCES_PER_SCOPE }, (_, index) =>
|
||||
largeValueKey(`e${index.toString(36).padStart(11, '0')}`)
|
||||
)
|
||||
const knownChildKey = directKeys[1]
|
||||
const unseenChildKey = largeValueKey('unseenchild1', 'source-execution')
|
||||
mockTxSelectLimit.mockImplementationOnce(async () => {
|
||||
const filtersKnownChildren = mockNotInArray.mock.calls.some(
|
||||
([field, values]) =>
|
||||
field === 'executionLargeValueDependencies.childKey' &&
|
||||
Array.isArray(values) &&
|
||||
values.includes(knownChildKey)
|
||||
)
|
||||
return [{ childKey: filtersKnownChildren ? unseenChildKey : knownChildKey }]
|
||||
})
|
||||
|
||||
await expect(
|
||||
registerLargeValueOwner(
|
||||
{
|
||||
key: largeValueKey('zyxwvutsrqpo', 'execution-1'),
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
size: 123,
|
||||
},
|
||||
directKeys
|
||||
)
|
||||
).rejects.toThrow('Large value dependency closure exceeds the limit')
|
||||
|
||||
expect(mockTxSelectLimit).toHaveBeenCalledWith(1)
|
||||
})
|
||||
|
||||
it('replaces an execution reference set with same-workspace unique keys', async () => {
|
||||
const matchingKey = largeValueKey('abcdefghijkl')
|
||||
const otherWorkspaceKey =
|
||||
'execution/workspace-2/workflow-1/source-execution/large-value-lv_abcdefghijkl.json'
|
||||
|
||||
await replaceLargeValueReferences(
|
||||
{
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-2',
|
||||
source: 'execution_log',
|
||||
},
|
||||
{
|
||||
a: {
|
||||
__simLargeValueRef: true,
|
||||
version: 1,
|
||||
id: 'lv_abcdefghijkl',
|
||||
kind: 'json',
|
||||
size: 123,
|
||||
key: matchingKey,
|
||||
},
|
||||
duplicate: {
|
||||
__simLargeValueRef: true,
|
||||
version: 1,
|
||||
id: 'lv_abcdefghijkl',
|
||||
kind: 'json',
|
||||
size: 123,
|
||||
key: matchingKey,
|
||||
},
|
||||
ignored: {
|
||||
__simLargeValueRef: true,
|
||||
version: 1,
|
||||
id: 'lv_abcdefghijkl',
|
||||
kind: 'json',
|
||||
size: 123,
|
||||
key: otherWorkspaceKey,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
expect(mockTransaction).toHaveBeenCalledOnce()
|
||||
expect(mockTxDelete).toHaveBeenCalledOnce()
|
||||
expect(mockEq).toHaveBeenCalledWith('executionLargeValueReferences.source', 'execution_log')
|
||||
expect(mockTxValues).toHaveBeenCalledWith([
|
||||
{
|
||||
key: matchingKey,
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-2',
|
||||
source: 'execution_log',
|
||||
},
|
||||
])
|
||||
})
|
||||
|
||||
it('adds a materialized reference only when the scope is below the reference cap', async () => {
|
||||
const key = largeValueKey('abcdefghijkl')
|
||||
|
||||
await addLargeValueReference(
|
||||
{
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-2',
|
||||
source: 'execution_log',
|
||||
},
|
||||
key
|
||||
)
|
||||
|
||||
expect(mockSelectLimit).toHaveBeenCalledWith(1)
|
||||
expect(mockSelectLimit).toHaveBeenCalledWith(MAX_LARGE_VALUE_REFERENCES_PER_SCOPE + 1)
|
||||
expect(mockValues).toHaveBeenCalledWith({
|
||||
key,
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-2',
|
||||
source: 'execution_log',
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects materialized references once the scope reaches the reference cap', async () => {
|
||||
mockSelectLimit.mockResolvedValueOnce([]).mockResolvedValueOnce(
|
||||
Array.from({ length: MAX_LARGE_VALUE_REFERENCES_PER_SCOPE }, (_, index) => ({
|
||||
key: largeValueKey(`d${index.toString(36).padStart(11, '0')}`),
|
||||
}))
|
||||
)
|
||||
|
||||
await expect(
|
||||
addLargeValueReference(
|
||||
{
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-2',
|
||||
source: 'execution_log',
|
||||
},
|
||||
largeValueKey('zyxwvutsrqpo')
|
||||
)
|
||||
).rejects.toThrow('exceeding the limit')
|
||||
|
||||
expect(mockInsert).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('prunes large value metadata in bounded batches', async () => {
|
||||
mockExecute
|
||||
.mockResolvedValueOnce([{ count: 2 }])
|
||||
.mockResolvedValueOnce([{ count: 3 }])
|
||||
.mockResolvedValueOnce([{ count: 4 }])
|
||||
|
||||
await expect(
|
||||
pruneLargeValueMetadata({
|
||||
workspaceIds: ['workspace-1'],
|
||||
tombstonesDeletedBefore: new Date('2026-01-01T00:00:00Z'),
|
||||
batchSize: 10,
|
||||
maxRowsPerTable: 100,
|
||||
})
|
||||
).resolves.toEqual({
|
||||
referencesDeleted: 2,
|
||||
dependenciesDeleted: 3,
|
||||
tombstonesDeleted: 4,
|
||||
})
|
||||
})
|
||||
|
||||
it('uses source-specific liveness when pruning stale references', async () => {
|
||||
await pruneLargeValueMetadata({
|
||||
workspaceIds: ['workspace-1'],
|
||||
tombstonesDeletedBefore: new Date('2026-01-01T00:00:00Z'),
|
||||
batchSize: 10,
|
||||
maxRowsPerTable: 100,
|
||||
})
|
||||
|
||||
const [query] = mockExecute.mock.calls[0] ?? []
|
||||
const sqlText = Array.isArray(query?.strings) ? query.strings.join(' ') : ''
|
||||
expect(sqlText).toContain("ref.source = 'execution_log'")
|
||||
expect(sqlText).toContain("ref.source = 'paused_snapshot'")
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,618 @@
|
||||
import { db } from '@sim/db'
|
||||
import {
|
||||
executionLargeValueDependencies,
|
||||
executionLargeValueReferences,
|
||||
executionLargeValues,
|
||||
pausedExecutions,
|
||||
workflowExecutionLogs,
|
||||
} from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, eq, inArray, notInArray, sql } from 'drizzle-orm'
|
||||
import { chunkArray } from '@/lib/cleanup/batch-delete'
|
||||
import { collectLargeValueKeys } from '@/lib/execution/payloads/large-execution-value'
|
||||
|
||||
const logger = createLogger('LargeValueMetadata')
|
||||
|
||||
type LargeValueMetadataClient = typeof db | Parameters<Parameters<typeof db.transaction>[0]>[0]
|
||||
|
||||
export const MAX_LARGE_VALUE_REFERENCES_PER_SCOPE = 5_000
|
||||
const LARGE_VALUE_METADATA_WRITE_CHUNK_SIZE = 500
|
||||
const LARGE_VALUE_METADATA_WORKSPACE_CHUNK_SIZE = 50
|
||||
const LARGE_VALUE_METADATA_PRUNE_BATCH_SIZE = 1_000
|
||||
const LARGE_VALUE_METADATA_PRUNE_MAX_ROWS_PER_TABLE = 5_000
|
||||
export const LIVE_PAUSED_REFERENCE_STATUSES = ['paused', 'partially_resumed', 'cancelling'] as const
|
||||
|
||||
export interface LargeValueOwner {
|
||||
key: string
|
||||
workspaceId: string
|
||||
workflowId: string
|
||||
executionId: string
|
||||
size: number
|
||||
}
|
||||
|
||||
export interface LargeValueReferenceScope {
|
||||
workspaceId?: string
|
||||
workflowId?: string | null
|
||||
executionId?: string
|
||||
source: 'execution_log' | 'paused_snapshot'
|
||||
}
|
||||
|
||||
interface LargeValueStorageKeyParts {
|
||||
workspaceId: string
|
||||
workflowId: string
|
||||
executionId: string
|
||||
}
|
||||
|
||||
export interface LargeValueMetadataPruneResult {
|
||||
referencesDeleted: number
|
||||
dependenciesDeleted: number
|
||||
tombstonesDeleted: number
|
||||
}
|
||||
|
||||
interface PruneLargeValueMetadataOptions {
|
||||
workspaceIds: string[]
|
||||
tombstonesDeletedBefore: Date
|
||||
batchSize?: number
|
||||
maxRowsPerTable?: number
|
||||
}
|
||||
|
||||
function parseLargeValueStorageKey(key: string): LargeValueStorageKeyParts | null {
|
||||
const parts = key.split('/')
|
||||
if (
|
||||
parts.length !== 5 ||
|
||||
parts[0] !== 'execution' ||
|
||||
!parts[1] ||
|
||||
!parts[2] ||
|
||||
!parts[3] ||
|
||||
!/^large-value-lv_[A-Za-z0-9_-]{12}\.json$/.test(parts[4])
|
||||
) {
|
||||
return null
|
||||
}
|
||||
|
||||
return {
|
||||
workspaceId: parts[1],
|
||||
workflowId: parts[2],
|
||||
executionId: parts[3],
|
||||
}
|
||||
}
|
||||
|
||||
function getBoundedUniqueKeys(keys: string[], label: string): string[] {
|
||||
const uniqueKeys = Array.from(new Set(keys))
|
||||
if (uniqueKeys.length > MAX_LARGE_VALUE_REFERENCES_PER_SCOPE) {
|
||||
throw new Error(
|
||||
`${label} contains ${uniqueKeys.length} large value references, exceeding the limit of ${MAX_LARGE_VALUE_REFERENCES_PER_SCOPE}`
|
||||
)
|
||||
}
|
||||
return uniqueKeys
|
||||
}
|
||||
|
||||
function getCount(rows: unknown): number {
|
||||
const [row] = Array.isArray(rows) ? rows : []
|
||||
if (!row || typeof row !== 'object' || !('count' in row)) {
|
||||
return 0
|
||||
}
|
||||
return Number((row as { count: unknown }).count) || 0
|
||||
}
|
||||
|
||||
export function collectLargeValueReferenceKeys(value: unknown, workspaceId?: string): string[] {
|
||||
return getBoundedUniqueKeys(
|
||||
collectLargeValueKeys(value).filter((key) => {
|
||||
const parsed = parseLargeValueStorageKey(key)
|
||||
return workspaceId ? parsed?.workspaceId === workspaceId : Boolean(parsed)
|
||||
}),
|
||||
'Large value reference set'
|
||||
)
|
||||
}
|
||||
|
||||
async function getDependencyClosure(
|
||||
client: LargeValueMetadataClient,
|
||||
ownerKey: string,
|
||||
workspaceId: string,
|
||||
referencedKeys: string[]
|
||||
): Promise<string[]> {
|
||||
const directKeys = getBoundedUniqueKeys(
|
||||
referencedKeys.filter((key) => {
|
||||
const parsed = parseLargeValueStorageKey(key)
|
||||
return parsed?.workspaceId === workspaceId && key !== ownerKey
|
||||
}),
|
||||
'Large value dependency set'
|
||||
)
|
||||
|
||||
if (directKeys.length === 0) {
|
||||
return []
|
||||
}
|
||||
|
||||
const closureKeys = new Set(directKeys)
|
||||
let frontier = directKeys
|
||||
|
||||
while (frontier.length > 0) {
|
||||
const nextFrontier: string[] = []
|
||||
|
||||
for (const keyChunk of chunkArray(frontier, LARGE_VALUE_METADATA_WRITE_CHUNK_SIZE)) {
|
||||
const remainingBudget = MAX_LARGE_VALUE_REFERENCES_PER_SCOPE - closureKeys.size
|
||||
const rows = await client
|
||||
.selectDistinct({ childKey: executionLargeValueDependencies.childKey })
|
||||
.from(executionLargeValueDependencies)
|
||||
.where(
|
||||
and(
|
||||
eq(executionLargeValueDependencies.workspaceId, workspaceId),
|
||||
inArray(executionLargeValueDependencies.parentKey, keyChunk),
|
||||
notInArray(executionLargeValueDependencies.childKey, Array.from(closureKeys))
|
||||
)
|
||||
)
|
||||
.limit(remainingBudget + 1)
|
||||
|
||||
for (const row of rows) {
|
||||
if (closureKeys.has(row.childKey)) {
|
||||
continue
|
||||
}
|
||||
closureKeys.add(row.childKey)
|
||||
nextFrontier.push(row.childKey)
|
||||
if (closureKeys.size > MAX_LARGE_VALUE_REFERENCES_PER_SCOPE) {
|
||||
throw new Error(
|
||||
`Large value dependency closure exceeds the limit of ${MAX_LARGE_VALUE_REFERENCES_PER_SCOPE}`
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
frontier = nextFrontier
|
||||
}
|
||||
|
||||
return Array.from(closureKeys)
|
||||
}
|
||||
|
||||
export async function registerLargeValueOwner(
|
||||
owner: LargeValueOwner,
|
||||
referencedKeys: string[] = []
|
||||
): Promise<boolean> {
|
||||
if (!Number.isFinite(owner.size) || owner.size <= 0) {
|
||||
return false
|
||||
}
|
||||
|
||||
const parsed = parseLargeValueStorageKey(owner.key)
|
||||
if (
|
||||
!parsed ||
|
||||
parsed.workspaceId !== owner.workspaceId ||
|
||||
parsed.workflowId !== owner.workflowId ||
|
||||
parsed.executionId !== owner.executionId
|
||||
) {
|
||||
logger.warn('Skipping large value owner registration for malformed storage key', {
|
||||
key: owner.key,
|
||||
workspaceId: owner.workspaceId,
|
||||
workflowId: owner.workflowId,
|
||||
executionId: owner.executionId,
|
||||
})
|
||||
return false
|
||||
}
|
||||
|
||||
await db.transaction(async (tx) => {
|
||||
await tx
|
||||
.insert(executionLargeValues)
|
||||
.values({
|
||||
key: owner.key,
|
||||
workspaceId: owner.workspaceId,
|
||||
workflowId: owner.workflowId,
|
||||
ownerExecutionId: owner.executionId,
|
||||
size: Math.ceil(owner.size),
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
|
||||
const dependencyKeys = await getDependencyClosure(
|
||||
tx,
|
||||
owner.key,
|
||||
owner.workspaceId,
|
||||
referencedKeys
|
||||
)
|
||||
if (dependencyKeys.length === 0) {
|
||||
return
|
||||
}
|
||||
|
||||
for (const keyChunk of chunkArray(dependencyKeys, LARGE_VALUE_METADATA_WRITE_CHUNK_SIZE)) {
|
||||
await tx
|
||||
.insert(executionLargeValueDependencies)
|
||||
.values(
|
||||
keyChunk.map((childKey) => ({
|
||||
parentKey: owner.key,
|
||||
childKey,
|
||||
workspaceId: owner.workspaceId,
|
||||
}))
|
||||
)
|
||||
.onConflictDoNothing()
|
||||
}
|
||||
})
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
export async function replaceLargeValueReferencesWithClient(
|
||||
client: LargeValueMetadataClient,
|
||||
scope: LargeValueReferenceScope,
|
||||
value: unknown
|
||||
): Promise<void> {
|
||||
if (!scope.workspaceId || !scope.executionId) {
|
||||
return
|
||||
}
|
||||
|
||||
await replaceLargeValueReferenceKeysWithClient(
|
||||
client,
|
||||
scope,
|
||||
collectLargeValueReferenceKeys(value, scope.workspaceId)
|
||||
)
|
||||
}
|
||||
|
||||
export async function replaceLargeValueReferenceKeysWithClient(
|
||||
client: LargeValueMetadataClient,
|
||||
scope: LargeValueReferenceScope,
|
||||
referenceKeys: string[]
|
||||
): Promise<void> {
|
||||
const { workspaceId, workflowId, executionId, source } = scope
|
||||
if (!workspaceId || !executionId) {
|
||||
return
|
||||
}
|
||||
|
||||
const keys = getBoundedUniqueKeys(
|
||||
referenceKeys.filter((key) => {
|
||||
const parsed = parseLargeValueStorageKey(key)
|
||||
return parsed?.workspaceId === workspaceId
|
||||
}),
|
||||
'Large value reference set'
|
||||
)
|
||||
|
||||
await client
|
||||
.delete(executionLargeValueReferences)
|
||||
.where(
|
||||
and(
|
||||
eq(executionLargeValueReferences.workspaceId, workspaceId),
|
||||
eq(executionLargeValueReferences.executionId, executionId),
|
||||
eq(executionLargeValueReferences.source, source)
|
||||
)
|
||||
)
|
||||
|
||||
if (keys.length === 0) {
|
||||
return
|
||||
}
|
||||
|
||||
for (const keyChunk of chunkArray(keys, LARGE_VALUE_METADATA_WRITE_CHUNK_SIZE)) {
|
||||
await client
|
||||
.insert(executionLargeValueReferences)
|
||||
.values(
|
||||
keyChunk.map((key) => ({
|
||||
key,
|
||||
workspaceId,
|
||||
workflowId: workflowId ?? null,
|
||||
executionId,
|
||||
source,
|
||||
}))
|
||||
)
|
||||
.onConflictDoNothing()
|
||||
}
|
||||
}
|
||||
|
||||
export async function addLargeValueReference(
|
||||
scope: LargeValueReferenceScope,
|
||||
key: string
|
||||
): Promise<void> {
|
||||
const { workspaceId, workflowId, executionId, source } = scope
|
||||
if (!workspaceId || !executionId) {
|
||||
return
|
||||
}
|
||||
|
||||
const [boundedKey] = getBoundedUniqueKeys(
|
||||
[key].filter((candidate) => {
|
||||
const parsed = parseLargeValueStorageKey(candidate)
|
||||
return parsed?.workspaceId === workspaceId
|
||||
}),
|
||||
'Large value reference set'
|
||||
)
|
||||
if (!boundedKey) {
|
||||
return
|
||||
}
|
||||
|
||||
const [existingRef] = await db
|
||||
.select({ key: executionLargeValueReferences.key })
|
||||
.from(executionLargeValueReferences)
|
||||
.where(
|
||||
and(
|
||||
eq(executionLargeValueReferences.workspaceId, workspaceId),
|
||||
eq(executionLargeValueReferences.executionId, executionId),
|
||||
eq(executionLargeValueReferences.source, source),
|
||||
eq(executionLargeValueReferences.key, boundedKey)
|
||||
)
|
||||
)
|
||||
.limit(1)
|
||||
|
||||
if (existingRef) {
|
||||
return
|
||||
}
|
||||
|
||||
const existingRefs = await db
|
||||
.select({ key: executionLargeValueReferences.key })
|
||||
.from(executionLargeValueReferences)
|
||||
.where(
|
||||
and(
|
||||
eq(executionLargeValueReferences.workspaceId, workspaceId),
|
||||
eq(executionLargeValueReferences.executionId, executionId),
|
||||
eq(executionLargeValueReferences.source, source)
|
||||
)
|
||||
)
|
||||
.limit(MAX_LARGE_VALUE_REFERENCES_PER_SCOPE + 1)
|
||||
|
||||
if (existingRefs.length >= MAX_LARGE_VALUE_REFERENCES_PER_SCOPE) {
|
||||
throw new Error(
|
||||
`Large value reference set contains at least ${existingRefs.length} references, exceeding the limit of ${MAX_LARGE_VALUE_REFERENCES_PER_SCOPE}`
|
||||
)
|
||||
}
|
||||
|
||||
await db
|
||||
.insert(executionLargeValueReferences)
|
||||
.values({
|
||||
key: boundedKey,
|
||||
workspaceId,
|
||||
workflowId: workflowId ?? null,
|
||||
executionId,
|
||||
source,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
}
|
||||
|
||||
export async function replaceLargeValueReferences(
|
||||
scope: LargeValueReferenceScope,
|
||||
value: unknown
|
||||
): Promise<void> {
|
||||
const referenceKeys = scope.workspaceId
|
||||
? collectLargeValueReferenceKeys(value, scope.workspaceId)
|
||||
: []
|
||||
await db.transaction(async (tx) => {
|
||||
await replaceLargeValueReferenceKeysWithClient(tx, scope, referenceKeys)
|
||||
})
|
||||
}
|
||||
|
||||
export async function markLargeValuesDeleted(keys: string[]): Promise<void> {
|
||||
if (keys.length === 0) {
|
||||
return
|
||||
}
|
||||
|
||||
await db
|
||||
.update(executionLargeValues)
|
||||
.set({ deletedAt: new Date() })
|
||||
.where(inArray(executionLargeValues.key, keys))
|
||||
}
|
||||
|
||||
async function pruneStaleReferences(workspaceIds: string[], batchSize: number): Promise<number> {
|
||||
const rows = await db.execute<{ count: number }>(sql`
|
||||
WITH deleted AS (
|
||||
DELETE FROM ${executionLargeValueReferences} AS ref
|
||||
WHERE ref.ctid IN (
|
||||
SELECT ref.ctid
|
||||
FROM ${executionLargeValueReferences} AS ref
|
||||
WHERE ref.workspace_id = ANY(${workspaceIds}::text[])
|
||||
AND (
|
||||
(
|
||||
ref.source = 'execution_log'
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS wel
|
||||
WHERE wel.execution_id = ref.execution_id
|
||||
)
|
||||
)
|
||||
OR (
|
||||
ref.source = 'paused_snapshot'
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS pe
|
||||
WHERE pe.execution_id = ref.execution_id
|
||||
AND pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)
|
||||
)
|
||||
OR ref.source NOT IN ('execution_log', 'paused_snapshot')
|
||||
)
|
||||
LIMIT ${batchSize}
|
||||
)
|
||||
RETURNING ref.key
|
||||
)
|
||||
SELECT count(*)::int AS count FROM deleted
|
||||
`)
|
||||
return getCount(rows)
|
||||
}
|
||||
|
||||
async function pruneDeletedParentDependencies(
|
||||
workspaceIds: string[],
|
||||
batchSize: number
|
||||
): Promise<number> {
|
||||
const rows = await db.execute<{ count: number }>(sql`
|
||||
WITH deleted AS (
|
||||
DELETE FROM ${executionLargeValueDependencies} AS dependency
|
||||
WHERE dependency.ctid IN (
|
||||
SELECT dependency.ctid
|
||||
FROM ${executionLargeValueDependencies} AS dependency
|
||||
WHERE dependency.workspace_id = ANY(${workspaceIds}::text[])
|
||||
AND (
|
||||
EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValues} AS parent_value
|
||||
WHERE parent_value.key = dependency.parent_key
|
||||
AND parent_value.deleted_at IS NOT NULL
|
||||
)
|
||||
OR NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValues} AS parent_value
|
||||
WHERE parent_value.key = dependency.parent_key
|
||||
)
|
||||
)
|
||||
LIMIT ${batchSize}
|
||||
)
|
||||
RETURNING dependency.parent_key
|
||||
)
|
||||
SELECT count(*)::int AS count FROM deleted
|
||||
`)
|
||||
return getCount(rows)
|
||||
}
|
||||
|
||||
async function pruneDeletedLargeValueTombstones(
|
||||
workspaceIds: string[],
|
||||
deletedBefore: Date,
|
||||
batchSize: number
|
||||
): Promise<number> {
|
||||
const rows = await db.execute<{ count: number }>(sql`
|
||||
WITH deleted AS (
|
||||
DELETE FROM ${executionLargeValues} AS value
|
||||
WHERE value.ctid IN (
|
||||
SELECT value.ctid
|
||||
FROM ${executionLargeValues} AS value
|
||||
WHERE value.workspace_id = ANY(${workspaceIds}::text[])
|
||||
AND value.deleted_at IS NOT NULL
|
||||
AND value.deleted_at < ${deletedBefore}
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValueDependencies} AS dependency
|
||||
WHERE dependency.parent_key = value.key
|
||||
)
|
||||
LIMIT ${batchSize}
|
||||
)
|
||||
RETURNING value.key
|
||||
)
|
||||
SELECT count(*)::int AS count FROM deleted
|
||||
`)
|
||||
return getCount(rows)
|
||||
}
|
||||
|
||||
export async function pruneLargeValueMetadata({
|
||||
workspaceIds,
|
||||
tombstonesDeletedBefore,
|
||||
batchSize = LARGE_VALUE_METADATA_PRUNE_BATCH_SIZE,
|
||||
maxRowsPerTable = LARGE_VALUE_METADATA_PRUNE_MAX_ROWS_PER_TABLE,
|
||||
}: PruneLargeValueMetadataOptions): Promise<LargeValueMetadataPruneResult> {
|
||||
const result: LargeValueMetadataPruneResult = {
|
||||
referencesDeleted: 0,
|
||||
dependenciesDeleted: 0,
|
||||
tombstonesDeleted: 0,
|
||||
}
|
||||
if (workspaceIds.length === 0) return result
|
||||
|
||||
for (const workspaceChunk of chunkArray(
|
||||
workspaceIds,
|
||||
LARGE_VALUE_METADATA_WORKSPACE_CHUNK_SIZE
|
||||
)) {
|
||||
const referencesRemaining = maxRowsPerTable - result.referencesDeleted
|
||||
if (referencesRemaining > 0) {
|
||||
result.referencesDeleted += await pruneStaleReferences(
|
||||
workspaceChunk,
|
||||
Math.min(batchSize, referencesRemaining)
|
||||
)
|
||||
}
|
||||
|
||||
const dependenciesRemaining = maxRowsPerTable - result.dependenciesDeleted
|
||||
if (dependenciesRemaining > 0) {
|
||||
result.dependenciesDeleted += await pruneDeletedParentDependencies(
|
||||
workspaceChunk,
|
||||
Math.min(batchSize, dependenciesRemaining)
|
||||
)
|
||||
}
|
||||
|
||||
const tombstonesRemaining = maxRowsPerTable - result.tombstonesDeleted
|
||||
if (tombstonesRemaining > 0) {
|
||||
result.tombstonesDeleted += await pruneDeletedLargeValueTombstones(
|
||||
workspaceChunk,
|
||||
tombstonesDeletedBefore,
|
||||
Math.min(batchSize, tombstonesRemaining)
|
||||
)
|
||||
}
|
||||
|
||||
if (
|
||||
result.referencesDeleted >= maxRowsPerTable &&
|
||||
result.dependenciesDeleted >= maxRowsPerTable &&
|
||||
result.tombstonesDeleted >= maxRowsPerTable
|
||||
) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
export function unreferencedLargeValuePredicate() {
|
||||
return sql`
|
||||
NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValueReferences} AS elvr
|
||||
WHERE elvr.key = ${executionLargeValues.key}
|
||||
AND (
|
||||
(
|
||||
elvr.source = 'execution_log'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS wel
|
||||
WHERE wel.execution_id = elvr.execution_id
|
||||
)
|
||||
)
|
||||
OR (
|
||||
elvr.source = 'paused_snapshot'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS pe
|
||||
WHERE pe.execution_id = elvr.execution_id
|
||||
AND pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS owner_wel
|
||||
WHERE owner_wel.execution_id = ${executionLargeValues.ownerExecutionId}
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS owner_pe
|
||||
WHERE owner_pe.execution_id = ${executionLargeValues.ownerExecutionId}
|
||||
AND owner_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValueDependencies} AS dependency
|
||||
INNER JOIN ${executionLargeValues} AS parent_value
|
||||
ON parent_value.key = dependency.parent_key
|
||||
AND parent_value.deleted_at IS NULL
|
||||
WHERE dependency.workspace_id = ${executionLargeValues.workspaceId}
|
||||
AND dependency.child_key = ${executionLargeValues.key}
|
||||
AND (
|
||||
EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS parent_owner_wel
|
||||
WHERE parent_owner_wel.execution_id = parent_value.owner_execution_id
|
||||
)
|
||||
OR EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS parent_owner_pe
|
||||
WHERE parent_owner_pe.execution_id = parent_value.owner_execution_id
|
||||
AND parent_owner_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)
|
||||
OR EXISTS (
|
||||
SELECT 1
|
||||
FROM ${executionLargeValueReferences} AS parent_ref
|
||||
WHERE parent_ref.key = parent_value.key
|
||||
AND (
|
||||
(
|
||||
parent_ref.source = 'execution_log'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM ${workflowExecutionLogs} AS parent_ref_wel
|
||||
WHERE parent_ref_wel.execution_id = parent_ref.execution_id
|
||||
)
|
||||
)
|
||||
OR (
|
||||
parent_ref.source = 'paused_snapshot'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM ${pausedExecutions} AS parent_ref_pe
|
||||
WHERE parent_ref_pe.execution_id = parent_ref.execution_id
|
||||
AND parent_ref_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
`
|
||||
}
|
||||
@@ -16,16 +16,29 @@ import {
|
||||
import { materializeLargeValueRef, storeLargeValue } from '@/lib/execution/payloads/store'
|
||||
import { EXECUTION_RESOURCE_LIMIT_CODE } from '@/lib/execution/resource-errors'
|
||||
|
||||
const { mockDownloadFile, mockUploadFile, mockVerifyFileAccess } = vi.hoisted(() => ({
|
||||
const {
|
||||
mockAddLargeValueReference,
|
||||
mockDeleteFileMetadata,
|
||||
mockDeleteFiles,
|
||||
mockDownloadFile,
|
||||
mockRegisterLargeValueOwner,
|
||||
mockUploadFile,
|
||||
mockVerifyFileAccess,
|
||||
} = vi.hoisted(() => ({
|
||||
mockAddLargeValueReference: vi.fn(),
|
||||
mockDeleteFileMetadata: vi.fn(),
|
||||
mockDeleteFiles: vi.fn(),
|
||||
mockDownloadFile: vi.fn(),
|
||||
mockRegisterLargeValueOwner: vi.fn(),
|
||||
mockUploadFile: vi.fn(),
|
||||
mockVerifyFileAccess: vi.fn(),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/uploads', () => ({
|
||||
StorageService: {
|
||||
uploadFile: mockUploadFile,
|
||||
deleteFiles: mockDeleteFiles,
|
||||
downloadFile: mockDownloadFile,
|
||||
uploadFile: mockUploadFile,
|
||||
},
|
||||
}))
|
||||
|
||||
@@ -38,11 +51,24 @@ vi.mock('@/app/api/files/authorization', () => ({
|
||||
verifyFileAccess: mockVerifyFileAccess,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
|
||||
addLargeValueReference: mockAddLargeValueReference,
|
||||
registerLargeValueOwner: mockRegisterLargeValueOwner,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/uploads/server/metadata', () => ({
|
||||
deleteFileMetadata: mockDeleteFileMetadata,
|
||||
}))
|
||||
|
||||
describe('large execution payload store', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
clearLargeValueCacheForTests()
|
||||
mockUploadFile.mockImplementation(async ({ customKey }) => ({ key: customKey }))
|
||||
mockAddLargeValueReference.mockResolvedValue(undefined)
|
||||
mockRegisterLargeValueOwner.mockResolvedValue(true)
|
||||
mockDeleteFiles.mockResolvedValue({ deleted: 1, failed: [] })
|
||||
mockDeleteFileMetadata.mockResolvedValue(true)
|
||||
mockVerifyFileAccess.mockResolvedValue(true)
|
||||
})
|
||||
|
||||
@@ -74,6 +100,126 @@ describe('large execution payload store', () => {
|
||||
customKey: ref.key,
|
||||
})
|
||||
)
|
||||
expect(mockRegisterLargeValueOwner).toHaveBeenCalledWith(
|
||||
{
|
||||
key: ref.key,
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
size: Buffer.byteLength(json, 'utf8'),
|
||||
},
|
||||
[]
|
||||
)
|
||||
})
|
||||
|
||||
it('cleans up uploaded storage and fails durable writes when owner metadata is not recorded', async () => {
|
||||
const value = { payload: 'x'.repeat(2048) }
|
||||
const json = JSON.stringify(value)
|
||||
mockRegisterLargeValueOwner.mockResolvedValueOnce(false)
|
||||
|
||||
await expect(
|
||||
storeLargeValue(value, json, Buffer.byteLength(json, 'utf8'), {
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
userId: 'user-1',
|
||||
requireDurable: true,
|
||||
})
|
||||
).rejects.toThrow('Failed to persist large execution value metadata')
|
||||
|
||||
const key = 'execution/workspace-1/workflow-1/execution-1/large-value-lv_'
|
||||
expect(mockDeleteFiles.mock.calls[0]?.[0][0]).toContain(key)
|
||||
expect(mockDeleteFileMetadata).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('keeps file metadata when untracked storage deletion reports failure', async () => {
|
||||
const value = { payload: 'x'.repeat(2048) }
|
||||
const json = JSON.stringify(value)
|
||||
mockRegisterLargeValueOwner.mockResolvedValueOnce(false)
|
||||
mockDeleteFiles.mockImplementationOnce(async (keys: string[]) => ({
|
||||
deleted: 0,
|
||||
failed: [{ key: keys[0] }],
|
||||
}))
|
||||
|
||||
await expect(
|
||||
storeLargeValue(value, json, Buffer.byteLength(json, 'utf8'), {
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
userId: 'user-1',
|
||||
requireDurable: true,
|
||||
})
|
||||
).rejects.toThrow('Failed to persist large execution value metadata')
|
||||
|
||||
expect(mockDeleteFiles).toHaveBeenCalledOnce()
|
||||
expect(mockDeleteFileMetadata).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not delete uploaded storage when owner metadata registration throws', async () => {
|
||||
const value = { payload: 'x'.repeat(2048) }
|
||||
const json = JSON.stringify(value)
|
||||
mockRegisterLargeValueOwner.mockRejectedValueOnce(new Error('metadata db down'))
|
||||
|
||||
await expect(
|
||||
storeLargeValue(value, json, Buffer.byteLength(json, 'utf8'), {
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
userId: 'user-1',
|
||||
requireDurable: true,
|
||||
})
|
||||
).rejects.toThrow('metadata db down')
|
||||
|
||||
expect(mockDeleteFiles).not.toHaveBeenCalled()
|
||||
expect(mockDeleteFileMetadata).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('falls back to memory-only refs for non-durable writes when orphan cleanup fails', async () => {
|
||||
const value = { payload: 'x'.repeat(2048) }
|
||||
const json = JSON.stringify(value)
|
||||
mockRegisterLargeValueOwner.mockResolvedValueOnce(false)
|
||||
mockDeleteFiles.mockImplementationOnce(async ([key]) => ({
|
||||
deleted: 0,
|
||||
failed: [{ key }],
|
||||
}))
|
||||
|
||||
const ref = await storeLargeValue(value, json, Buffer.byteLength(json, 'utf8'), {
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
userId: 'user-1',
|
||||
})
|
||||
|
||||
expect(ref.key).toBeUndefined()
|
||||
expect(materializeLargeValueRefSync(ref, { executionId: 'execution-1' })).toEqual(value)
|
||||
expect(mockDeleteFileMetadata).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('passes nested large value refs to owner metadata registration', async () => {
|
||||
const nestedKey =
|
||||
'execution/workspace-1/workflow-1/source-execution/large-value-lv_abcdefghijkl.json'
|
||||
const value = {
|
||||
nested: {
|
||||
__simLargeValueRef: true,
|
||||
version: 1,
|
||||
id: 'lv_abcdefghijkl',
|
||||
kind: 'object',
|
||||
size: 123,
|
||||
key: nestedKey,
|
||||
},
|
||||
payload: 'x'.repeat(2048),
|
||||
}
|
||||
const json = JSON.stringify(value)
|
||||
|
||||
await storeLargeValue(value, json, Buffer.byteLength(json, 'utf8'), {
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
userId: 'user-1',
|
||||
requireDurable: true,
|
||||
})
|
||||
|
||||
expect(mockRegisterLargeValueOwner).toHaveBeenCalledWith(expect.any(Object), [nestedKey])
|
||||
})
|
||||
|
||||
it('fails durable writes before producing refs when execution context is missing', async () => {
|
||||
@@ -123,6 +269,60 @@ describe('large execution payload store', () => {
|
||||
}
|
||||
)
|
||||
).resolves.toEqual({ ok: true })
|
||||
expect(mockAddLargeValueReference).toHaveBeenCalledWith(
|
||||
{
|
||||
workspaceId: 'workflow-1',
|
||||
workflowId: 'workflow-2',
|
||||
executionId: 'execution-1',
|
||||
source: 'execution_log',
|
||||
},
|
||||
'execution/workflow-1/workflow-2/execution-1/large-value-lv_ABCDEFGHIJKL.json'
|
||||
)
|
||||
})
|
||||
|
||||
it('records a reference before returning a cached prior-execution value', async () => {
|
||||
cacheLargeValue(
|
||||
'lv_CACHEDREF123',
|
||||
{ cached: true },
|
||||
16,
|
||||
{
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'source-execution',
|
||||
},
|
||||
{ recoverable: true }
|
||||
)
|
||||
|
||||
await expect(
|
||||
materializeLargeValueRef(
|
||||
{
|
||||
__simLargeValueRef: true,
|
||||
version: 1,
|
||||
id: 'lv_CACHEDREF123',
|
||||
kind: 'object',
|
||||
size: 16,
|
||||
key: 'execution/workspace-1/workflow-1/source-execution/large-value-lv_CACHEDREF123.json',
|
||||
executionId: 'source-execution',
|
||||
},
|
||||
{
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'consumer-execution',
|
||||
allowLargeValueWorkflowScope: true,
|
||||
}
|
||||
)
|
||||
).resolves.toEqual({ cached: true })
|
||||
|
||||
expect(mockAddLargeValueReference).toHaveBeenCalledWith(
|
||||
{
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'consumer-execution',
|
||||
source: 'execution_log',
|
||||
},
|
||||
'execution/workspace-1/workflow-1/source-execution/large-value-lv_CACHEDREF123.json'
|
||||
)
|
||||
expect(mockDownloadFile).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('bounds durable large-value writes', async () => {
|
||||
@@ -312,6 +512,28 @@ describe('large execution payload store', () => {
|
||||
expect(mockDownloadFile).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('fails loudly when reference tracking fails before returning cached durable refs', async () => {
|
||||
const scope = {
|
||||
workspaceId: 'workspace-1',
|
||||
workflowId: 'workflow-1',
|
||||
executionId: 'execution-1',
|
||||
}
|
||||
const ref = {
|
||||
__simLargeValueRef: true,
|
||||
version: 1,
|
||||
id: 'lv_TRACKFAIL12',
|
||||
kind: 'object',
|
||||
size: 32,
|
||||
key: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_TRACKFAIL12.json',
|
||||
executionId: 'execution-1',
|
||||
} as const
|
||||
cacheLargeValue(ref.id, { retained: true }, ref.size, scope, { recoverable: true })
|
||||
mockAddLargeValueReference.mockRejectedValueOnce(new Error('reference cap exceeded'))
|
||||
|
||||
await expect(materializeLargeValueRef(ref, scope)).rejects.toThrow('reference cap exceeded')
|
||||
expect(mockDownloadFile).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('enforces maxBytes before returning cached refs', async () => {
|
||||
const scope = {
|
||||
workspaceId: 'workspace-1',
|
||||
|
||||
@@ -3,6 +3,7 @@ import { toError } from '@sim/utils/errors'
|
||||
import { generateShortId } from '@sim/utils/id'
|
||||
import { truncate } from '@sim/utils/string'
|
||||
import { cacheLargeValue, materializeLargeValueRefSync } from '@/lib/execution/payloads/cache'
|
||||
import { collectLargeValueKeys } from '@/lib/execution/payloads/large-execution-value'
|
||||
import {
|
||||
LARGE_VALUE_REF_VERSION,
|
||||
type LargeValueKind,
|
||||
@@ -102,6 +103,55 @@ async function persistValue(
|
||||
}
|
||||
}
|
||||
|
||||
async function registerPersistedValueOwner(
|
||||
key: string | undefined,
|
||||
size: number,
|
||||
referencedKeys: string[],
|
||||
context: LargeValueStoreContext
|
||||
): Promise<boolean> {
|
||||
const { workspaceId, workflowId, executionId } = context
|
||||
if (!key || !workspaceId || !workflowId || !executionId) {
|
||||
return false
|
||||
}
|
||||
|
||||
const { registerLargeValueOwner } = await import('@/lib/execution/payloads/large-value-metadata')
|
||||
return await registerLargeValueOwner(
|
||||
{
|
||||
key,
|
||||
workspaceId,
|
||||
workflowId,
|
||||
executionId,
|
||||
size,
|
||||
},
|
||||
referencedKeys
|
||||
)
|
||||
}
|
||||
|
||||
async function deleteUntrackedPersistedValue(key: string): Promise<boolean> {
|
||||
try {
|
||||
const [{ StorageService }, { deleteFileMetadata }] = await Promise.all([
|
||||
import('@/lib/uploads'),
|
||||
import('@/lib/uploads/server/metadata'),
|
||||
])
|
||||
const result = await StorageService.deleteFiles([key], 'execution')
|
||||
const deleteFailed = result.failed.some((failed) => failed.key === key)
|
||||
if (deleteFailed) {
|
||||
logger.warn('Failed to delete untracked large execution value from storage', {
|
||||
key,
|
||||
})
|
||||
return false
|
||||
}
|
||||
await deleteFileMetadata(key)
|
||||
return true
|
||||
} catch (error) {
|
||||
logger.warn('Failed to clean up untracked large execution value', {
|
||||
key,
|
||||
error: toError(error).message,
|
||||
})
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export async function storeLargeValue(
|
||||
value: unknown,
|
||||
json: string,
|
||||
@@ -109,8 +159,19 @@ export async function storeLargeValue(
|
||||
context: LargeValueStoreContext
|
||||
): Promise<LargeValueRef> {
|
||||
assertDurableLargeValueSize(size)
|
||||
const referencedKeys = collectLargeValueKeys(value)
|
||||
const id = `lv_${generateShortId(12)}`
|
||||
const key = await persistValue(id, json, context)
|
||||
let key = await persistValue(id, json, context)
|
||||
if (key) {
|
||||
const registered = await registerPersistedValueOwner(key, size, referencedKeys, context)
|
||||
if (!registered) {
|
||||
await deleteUntrackedPersistedValue(key)
|
||||
if (context.requireDurable) {
|
||||
throw new Error('Failed to persist large execution value metadata')
|
||||
}
|
||||
key = undefined
|
||||
}
|
||||
}
|
||||
const cached = cacheLargeValue(id, value, size, context, { recoverable: Boolean(key) })
|
||||
if (!key && !cached) {
|
||||
throw new Error('Cannot retain large execution value without durable storage')
|
||||
@@ -139,16 +200,27 @@ export async function materializeLargeValueRef(
|
||||
assertLargeValueRefAccess(ref, context)
|
||||
assertInlineMaterializationSize(ref.size, context.maxBytes)
|
||||
|
||||
const cached = materializeLargeValueRefSync(ref, context)
|
||||
if (cached !== undefined) {
|
||||
return cached
|
||||
if (!ref.key || !isValidLargeValueKey(ref)) {
|
||||
return materializeLargeValueRefSync(ref, context)
|
||||
}
|
||||
|
||||
if (!ref.key || !isValidLargeValueKey(ref)) {
|
||||
return undefined
|
||||
}
|
||||
const { addLargeValueReference } = await import('@/lib/execution/payloads/large-value-metadata')
|
||||
await addLargeValueReference(
|
||||
{
|
||||
workspaceId: context.workspaceId,
|
||||
workflowId: context.workflowId,
|
||||
executionId: context.executionId,
|
||||
source: 'execution_log',
|
||||
},
|
||||
ref.key
|
||||
)
|
||||
|
||||
try {
|
||||
const cached = materializeLargeValueRefSync(ref, context)
|
||||
if (cached !== undefined) {
|
||||
return cached
|
||||
}
|
||||
|
||||
const value = await readLargeValueRefFromStorage(ref, {
|
||||
workspaceId: context.workspaceId,
|
||||
workflowId: context.workflowId,
|
||||
|
||||
@@ -22,6 +22,10 @@ import { checkAndBillOverageThreshold } from '@/lib/billing/threshold-billing'
|
||||
import { isBillingEnabled } from '@/lib/core/config/feature-flags'
|
||||
import { redactApiKeys } from '@/lib/core/security/redaction'
|
||||
import { filterForDisplay } from '@/lib/core/utils/display-filters'
|
||||
import {
|
||||
collectLargeValueReferenceKeys,
|
||||
replaceLargeValueReferenceKeysWithClient,
|
||||
} from '@/lib/execution/payloads/large-value-metadata'
|
||||
import { emitWorkflowExecutionCompleted } from '@/lib/logs/events'
|
||||
import { snapshotService } from '@/lib/logs/execution/snapshot/service'
|
||||
import type {
|
||||
@@ -739,11 +743,12 @@ export class ExecutionLogger implements IExecutionLoggerService {
|
||||
},
|
||||
executionId
|
||||
)
|
||||
const completedExecutionLargeValueKeys = collectLargeValueReferenceKeys(completedExecutionData)
|
||||
|
||||
const [updatedLog] = await db.transaction(async (tx) => {
|
||||
const updatedLog = await db.transaction(async (tx) => {
|
||||
await setExecutionLogWriteTimeouts(tx)
|
||||
|
||||
return tx
|
||||
const [log] = await tx
|
||||
.update(workflowExecutionLogs)
|
||||
.set({
|
||||
level,
|
||||
@@ -756,11 +761,24 @@ export class ExecutionLogger implements IExecutionLoggerService {
|
||||
})
|
||||
.where(eq(workflowExecutionLogs.executionId, executionId))
|
||||
.returning()
|
||||
})
|
||||
|
||||
if (!updatedLog) {
|
||||
throw new Error(`Workflow log not found for execution ${executionId}`)
|
||||
}
|
||||
if (!log) {
|
||||
throw new Error(`Workflow log not found for execution ${executionId}`)
|
||||
}
|
||||
|
||||
await replaceLargeValueReferenceKeysWithClient(
|
||||
tx,
|
||||
{
|
||||
workspaceId: log.workspaceId,
|
||||
workflowId: log.workflowId,
|
||||
executionId,
|
||||
source: 'execution_log',
|
||||
},
|
||||
completedExecutionLargeValueKeys
|
||||
)
|
||||
|
||||
return log
|
||||
})
|
||||
|
||||
try {
|
||||
// Skip workflow lookup if workflow was deleted
|
||||
|
||||
@@ -9,6 +9,7 @@ const {
|
||||
mockUpload,
|
||||
mockDownload,
|
||||
mockDelete,
|
||||
mockDeleteIfExists,
|
||||
mockGetBlockBlobClient,
|
||||
mockGetContainerClient,
|
||||
mockFromConnectionString,
|
||||
@@ -19,6 +20,7 @@ const {
|
||||
mockUpload: vi.fn(),
|
||||
mockDownload: vi.fn(),
|
||||
mockDelete: vi.fn(),
|
||||
mockDeleteIfExists: vi.fn(),
|
||||
mockGetBlockBlobClient: vi.fn(),
|
||||
mockGetContainerClient: vi.fn(),
|
||||
mockFromConnectionString: vi.fn(),
|
||||
@@ -66,6 +68,7 @@ describe('Azure Blob Storage Client', () => {
|
||||
upload: mockUpload,
|
||||
download: mockDownload,
|
||||
delete: mockDelete,
|
||||
deleteIfExists: mockDeleteIfExists,
|
||||
url: 'https://test.blob.core.windows.net/container/test-file',
|
||||
})
|
||||
|
||||
@@ -181,12 +184,12 @@ describe('Azure Blob Storage Client', () => {
|
||||
it('should delete a file from Azure Blob Storage', async () => {
|
||||
const testKey = 'test-file-key'
|
||||
|
||||
mockDelete.mockResolvedValueOnce({})
|
||||
mockDeleteIfExists.mockResolvedValueOnce({})
|
||||
|
||||
await deleteFromBlob(testKey)
|
||||
|
||||
expect(mockGetBlockBlobClient).toHaveBeenCalledWith(testKey)
|
||||
expect(mockDelete).toHaveBeenCalled()
|
||||
expect(mockDeleteIfExists).toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -435,7 +435,7 @@ export async function deleteFromBlob(key: string, customConfig?: BlobConfig): Pr
|
||||
const containerClient = blobServiceClient.getContainerClient(containerName)
|
||||
const blockBlobClient = containerClient.getBlockBlobClient(key)
|
||||
|
||||
await blockBlobClient.delete()
|
||||
await blockBlobClient.deleteIfExists()
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -13,6 +13,10 @@ import {
|
||||
resetExecutionStreamBuffer,
|
||||
type TerminalExecutionStreamStatus,
|
||||
} from '@/lib/execution/event-buffer'
|
||||
import {
|
||||
collectLargeValueReferenceKeys,
|
||||
replaceLargeValueReferenceKeysWithClient,
|
||||
} from '@/lib/execution/payloads/large-value-metadata'
|
||||
import { compactBlockLogs, compactExecutionPayload } from '@/lib/execution/payloads/serializer'
|
||||
import { preprocessExecution } from '@/lib/execution/preprocessing'
|
||||
import { LoggingSession } from '@/lib/logs/execution/logging-session'
|
||||
@@ -48,6 +52,21 @@ function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return value !== null && typeof value === 'object' && !Array.isArray(value)
|
||||
}
|
||||
|
||||
function parseSnapshotForReferenceTracking(snapshotSeed: SerializedSnapshot): unknown {
|
||||
try {
|
||||
return { ...snapshotSeed, snapshot: JSON.parse(snapshotSeed.snapshot) }
|
||||
} catch {
|
||||
return snapshotSeed
|
||||
}
|
||||
}
|
||||
|
||||
function getSnapshotWorkspaceId(snapshotValue: unknown): string | undefined {
|
||||
if (!isRecord(snapshotValue)) return undefined
|
||||
const metadata = snapshotValue.metadata
|
||||
if (!isRecord(metadata)) return undefined
|
||||
return typeof metadata.workspaceId === 'string' ? metadata.workspaceId : undefined
|
||||
}
|
||||
|
||||
function isResumablePausedStatus(status: string): boolean {
|
||||
return RESUMABLE_PAUSED_STATUSES.includes(status as (typeof RESUMABLE_PAUSED_STATUSES)[number])
|
||||
}
|
||||
@@ -169,6 +188,13 @@ export function computeEarliestResumeAt(
|
||||
export class PauseResumeManager {
|
||||
static async persistPauseResult(args: PersistPauseResultArgs): Promise<void> {
|
||||
const { workflowId, executionId, pausePoints, snapshotSeed, executorUserId } = args
|
||||
const snapshotReferenceValue = parseSnapshotForReferenceTracking(snapshotSeed)
|
||||
const snapshotWorkspaceId = getSnapshotWorkspaceId(
|
||||
isRecord(snapshotReferenceValue) ? snapshotReferenceValue.snapshot : undefined
|
||||
)
|
||||
const snapshotReferenceKeys = snapshotWorkspaceId
|
||||
? collectLargeValueReferenceKeys(snapshotReferenceValue, snapshotWorkspaceId)
|
||||
: []
|
||||
|
||||
const pausePointsRecord = pausePoints.reduce<Record<string, any>>((acc, point) => {
|
||||
acc[point.contextId] = {
|
||||
@@ -220,6 +246,18 @@ export class PauseResumeManager {
|
||||
updatedAt: now,
|
||||
nextResumeAt,
|
||||
})
|
||||
if (snapshotWorkspaceId) {
|
||||
await replaceLargeValueReferenceKeysWithClient(
|
||||
tx,
|
||||
{
|
||||
workspaceId: snapshotWorkspaceId,
|
||||
workflowId,
|
||||
executionId,
|
||||
source: 'paused_snapshot',
|
||||
},
|
||||
snapshotReferenceKeys
|
||||
)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@@ -264,6 +302,19 @@ export class PauseResumeManager {
|
||||
nextResumeAt: mergedNextResumeAt,
|
||||
})
|
||||
.where(eq(pausedExecutions.id, existing.id))
|
||||
|
||||
if (snapshotWorkspaceId) {
|
||||
await replaceLargeValueReferenceKeysWithClient(
|
||||
tx,
|
||||
{
|
||||
workspaceId: snapshotWorkspaceId,
|
||||
workflowId,
|
||||
executionId,
|
||||
source: 'paused_snapshot',
|
||||
},
|
||||
snapshotReferenceKeys
|
||||
)
|
||||
}
|
||||
})
|
||||
|
||||
await PauseResumeManager.processQueuedResumes(executionId, workflowId)
|
||||
@@ -1568,14 +1619,34 @@ export class PauseResumeManager {
|
||||
snapshot: JSON.stringify(snapshotData),
|
||||
triggerIds: currentSnapshot.triggerIds,
|
||||
}
|
||||
const snapshotWorkspaceId = getSnapshotWorkspaceId(snapshotData)
|
||||
const snapshotReferenceValue = { ...updatedSnapshot, snapshot: snapshotData }
|
||||
const snapshotReferenceKeys = snapshotWorkspaceId
|
||||
? collectLargeValueReferenceKeys(snapshotReferenceValue, snapshotWorkspaceId)
|
||||
: []
|
||||
|
||||
await db
|
||||
.update(pausedExecutions)
|
||||
.set({
|
||||
executionSnapshot: updatedSnapshot,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(pausedExecutions.id, pausedExecutionId))
|
||||
await db.transaction(async (tx) => {
|
||||
await tx
|
||||
.update(pausedExecutions)
|
||||
.set({
|
||||
executionSnapshot: updatedSnapshot,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(pausedExecutions.id, pausedExecutionId))
|
||||
|
||||
if (snapshotWorkspaceId) {
|
||||
await replaceLargeValueReferenceKeysWithClient(
|
||||
tx,
|
||||
{
|
||||
workspaceId: snapshotWorkspaceId,
|
||||
workflowId: pausedExecution.workflowId,
|
||||
executionId: pausedExecution.executionId,
|
||||
source: 'paused_snapshot',
|
||||
},
|
||||
snapshotReferenceKeys
|
||||
)
|
||||
}
|
||||
})
|
||||
|
||||
logger.info('Updated snapshot after resume', {
|
||||
pausedExecutionId,
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
CREATE TYPE "public"."execution_large_value_reference_source" AS ENUM('execution_log', 'paused_snapshot');--> statement-breakpoint
|
||||
CREATE TABLE "execution_large_value_dependencies" (
|
||||
"parent_key" text NOT NULL,
|
||||
"child_key" text NOT NULL,
|
||||
"workspace_id" text NOT NULL,
|
||||
"created_at" timestamp DEFAULT now() NOT NULL,
|
||||
CONSTRAINT "execution_large_value_dependencies_parent_key_child_key_pk" PRIMARY KEY("parent_key","child_key")
|
||||
);
|
||||
--> statement-breakpoint
|
||||
CREATE TABLE "execution_large_value_references" (
|
||||
"key" text NOT NULL,
|
||||
"execution_id" text NOT NULL,
|
||||
"source" "execution_large_value_reference_source" NOT NULL,
|
||||
"workspace_id" text NOT NULL,
|
||||
"workflow_id" text,
|
||||
"created_at" timestamp DEFAULT now() NOT NULL,
|
||||
CONSTRAINT "execution_large_value_references_key_execution_id_source_pk" PRIMARY KEY("key","execution_id","source")
|
||||
);
|
||||
--> statement-breakpoint
|
||||
CREATE TABLE "execution_large_values" (
|
||||
"key" text PRIMARY KEY NOT NULL,
|
||||
"workspace_id" text NOT NULL,
|
||||
"workflow_id" text,
|
||||
"owner_execution_id" text NOT NULL,
|
||||
"size" integer NOT NULL,
|
||||
"created_at" timestamp DEFAULT now() NOT NULL,
|
||||
"deleted_at" timestamp
|
||||
);
|
||||
--> statement-breakpoint
|
||||
ALTER TABLE "execution_large_value_dependencies" ADD CONSTRAINT "execution_large_value_dependencies_workspace_id_workspace_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspace"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
|
||||
ALTER TABLE "execution_large_value_references" ADD CONSTRAINT "execution_large_value_references_workspace_id_workspace_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspace"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
|
||||
ALTER TABLE "execution_large_value_references" ADD CONSTRAINT "execution_large_value_references_workflow_id_workflow_id_fk" FOREIGN KEY ("workflow_id") REFERENCES "public"."workflow"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint
|
||||
ALTER TABLE "execution_large_values" ADD CONSTRAINT "execution_large_values_workspace_id_workspace_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspace"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
|
||||
ALTER TABLE "execution_large_values" ADD CONSTRAINT "execution_large_values_workflow_id_workflow_id_fk" FOREIGN KEY ("workflow_id") REFERENCES "public"."workflow"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint
|
||||
CREATE INDEX "execution_large_value_dependencies_workspace_parent_key_idx" ON "execution_large_value_dependencies" USING btree ("workspace_id","parent_key");--> statement-breakpoint
|
||||
CREATE INDEX "execution_large_value_dependencies_workspace_child_key_idx" ON "execution_large_value_dependencies" USING btree ("workspace_id","child_key");--> statement-breakpoint
|
||||
CREATE INDEX "execution_large_value_references_workspace_execution_source_idx" ON "execution_large_value_references" USING btree ("workspace_id","execution_id","source");--> statement-breakpoint
|
||||
CREATE INDEX "execution_large_values_owner_execution_id_idx" ON "execution_large_values" USING btree ("owner_execution_id");--> statement-breakpoint
|
||||
CREATE INDEX "execution_large_values_cleanup_idx" ON "execution_large_values" USING btree ("workspace_id","created_at","key") WHERE "execution_large_values"."deleted_at" IS NULL;--> statement-breakpoint
|
||||
CREATE INDEX "execution_large_values_tombstone_cleanup_idx" ON "execution_large_values" USING btree ("workspace_id","deleted_at","key") WHERE "execution_large_values"."deleted_at" IS NOT NULL;
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1478,6 +1478,13 @@
|
||||
"when": 1779398164637,
|
||||
"tag": "0211_breezy_cloak",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 212,
|
||||
"version": "7",
|
||||
"when": 1779472552512,
|
||||
"tag": "0212_sturdy_guardsmen",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -367,6 +367,80 @@ export const workflowExecutionLogs = pgTable(
|
||||
})
|
||||
)
|
||||
|
||||
export const executionLargeValueReferenceSourceEnum = pgEnum(
|
||||
'execution_large_value_reference_source',
|
||||
['execution_log', 'paused_snapshot']
|
||||
)
|
||||
|
||||
export const executionLargeValues = pgTable(
|
||||
'execution_large_values',
|
||||
{
|
||||
key: text('key').primaryKey(),
|
||||
workspaceId: text('workspace_id')
|
||||
.notNull()
|
||||
.references(() => workspace.id, { onDelete: 'cascade' }),
|
||||
workflowId: text('workflow_id').references(() => workflow.id, { onDelete: 'set null' }),
|
||||
ownerExecutionId: text('owner_execution_id').notNull(),
|
||||
size: integer('size').notNull(),
|
||||
createdAt: timestamp('created_at').notNull().defaultNow(),
|
||||
deletedAt: timestamp('deleted_at'),
|
||||
},
|
||||
(table) => ({
|
||||
ownerExecutionIdIdx: index('execution_large_values_owner_execution_id_idx').on(
|
||||
table.ownerExecutionId
|
||||
),
|
||||
cleanupIdx: index('execution_large_values_cleanup_idx')
|
||||
.on(table.workspaceId, table.createdAt, table.key)
|
||||
.where(sql`${table.deletedAt} IS NULL`),
|
||||
tombstoneCleanupIdx: index('execution_large_values_tombstone_cleanup_idx')
|
||||
.on(table.workspaceId, table.deletedAt, table.key)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
export const executionLargeValueReferences = pgTable(
|
||||
'execution_large_value_references',
|
||||
{
|
||||
key: text('key').notNull(),
|
||||
executionId: text('execution_id').notNull(),
|
||||
source: executionLargeValueReferenceSourceEnum('source').notNull(),
|
||||
workspaceId: text('workspace_id')
|
||||
.notNull()
|
||||
.references(() => workspace.id, { onDelete: 'cascade' }),
|
||||
workflowId: text('workflow_id').references(() => workflow.id, { onDelete: 'set null' }),
|
||||
createdAt: timestamp('created_at').notNull().defaultNow(),
|
||||
},
|
||||
(table) => ({
|
||||
pk: primaryKey({ columns: [table.key, table.executionId, table.source] }),
|
||||
workspaceExecutionSourceIdx: index(
|
||||
'execution_large_value_references_workspace_execution_source_idx'
|
||||
).on(table.workspaceId, table.executionId, table.source),
|
||||
})
|
||||
)
|
||||
|
||||
export const executionLargeValueDependencies = pgTable(
|
||||
'execution_large_value_dependencies',
|
||||
{
|
||||
parentKey: text('parent_key').notNull(),
|
||||
childKey: text('child_key').notNull(),
|
||||
workspaceId: text('workspace_id')
|
||||
.notNull()
|
||||
.references(() => workspace.id, { onDelete: 'cascade' }),
|
||||
createdAt: timestamp('created_at').notNull().defaultNow(),
|
||||
},
|
||||
(table) => ({
|
||||
pk: primaryKey({ columns: [table.parentKey, table.childKey] }),
|
||||
workspaceParentKeyIdx: index('execution_large_value_dependencies_workspace_parent_key_idx').on(
|
||||
table.workspaceId,
|
||||
table.parentKey
|
||||
),
|
||||
workspaceChildKeyIdx: index('execution_large_value_dependencies_workspace_child_key_idx').on(
|
||||
table.workspaceId,
|
||||
table.childKey
|
||||
),
|
||||
})
|
||||
)
|
||||
|
||||
export const pausedExecutions = pgTable(
|
||||
'paused_executions',
|
||||
{
|
||||
|
||||
@@ -234,20 +234,25 @@ export const dbChainMock = {
|
||||
* Creates a mock database connection.
|
||||
*/
|
||||
export function createMockDb() {
|
||||
const fromBuilder = () => ({
|
||||
where: vi.fn(() => ({
|
||||
limit: vi.fn(() => Promise.resolve([])),
|
||||
orderBy: vi.fn(() => Promise.resolve([])),
|
||||
})),
|
||||
leftJoin: vi.fn(() => ({
|
||||
where: vi.fn(() => Promise.resolve([])),
|
||||
})),
|
||||
innerJoin: vi.fn(() => ({
|
||||
where: vi.fn(() => Promise.resolve([])),
|
||||
})),
|
||||
})
|
||||
|
||||
return {
|
||||
select: vi.fn(() => ({
|
||||
from: vi.fn(() => ({
|
||||
where: vi.fn(() => ({
|
||||
limit: vi.fn(() => Promise.resolve([])),
|
||||
orderBy: vi.fn(() => Promise.resolve([])),
|
||||
})),
|
||||
leftJoin: vi.fn(() => ({
|
||||
where: vi.fn(() => Promise.resolve([])),
|
||||
})),
|
||||
innerJoin: vi.fn(() => ({
|
||||
where: vi.fn(() => Promise.resolve([])),
|
||||
})),
|
||||
})),
|
||||
from: vi.fn(fromBuilder),
|
||||
})),
|
||||
selectDistinct: vi.fn(() => ({
|
||||
from: vi.fn(fromBuilder),
|
||||
})),
|
||||
insert: vi.fn(() => ({
|
||||
values: vi.fn(() => ({
|
||||
|
||||
@@ -157,6 +157,29 @@ export const schemaMock = {
|
||||
files: 'files',
|
||||
createdAt: 'createdAt',
|
||||
},
|
||||
executionLargeValues: {
|
||||
key: 'key',
|
||||
workspaceId: 'workspaceId',
|
||||
workflowId: 'workflowId',
|
||||
ownerExecutionId: 'ownerExecutionId',
|
||||
size: 'size',
|
||||
createdAt: 'createdAt',
|
||||
deletedAt: 'deletedAt',
|
||||
},
|
||||
executionLargeValueReferences: {
|
||||
key: 'key',
|
||||
executionId: 'executionId',
|
||||
source: 'source',
|
||||
workspaceId: 'workspaceId',
|
||||
workflowId: 'workflowId',
|
||||
createdAt: 'createdAt',
|
||||
},
|
||||
executionLargeValueDependencies: {
|
||||
parentKey: 'parentKey',
|
||||
childKey: 'childKey',
|
||||
workspaceId: 'workspaceId',
|
||||
createdAt: 'createdAt',
|
||||
},
|
||||
pausedExecutions: {
|
||||
id: 'id',
|
||||
workflowId: 'workflowId',
|
||||
|
||||
Reference in New Issue
Block a user