Skip to content

Commit f786543

Browse files
committed
fix(retention): charge committed Project version batches to the shared run budget
Project version cleanup commits one transaction per batch. An unbudgeted run isolates a failing owner, which reported zero rows even when batches before the failure had committed, so later owners received the full remaining limit again. Cleanup now charges each committed batch to one RowBudget shared across the run, matching how archive cleanup decrements its budgets in place.
1 parent b99acdd commit f786543

3 files changed

Lines changed: 61 additions & 8 deletions

File tree

‎apps/sim/lib/file-retention/service.ts‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import type { CleanupJobPayload } from '@/lib/billing/cleanup-dispatcher'
22
import {
33
DEFAULT_DELETE_CHUNK_SIZE,
44
DEFAULT_MAX_BATCHES_PER_TABLE,
5+
type RowBudget,
56
} from '@/lib/cleanup/batch-delete'
67
import { isolateOwnerFailure } from '@/lib/file-retention/failure'
78
import { prepareLegacyFileArchiveCleanup } from '@/lib/file-retention/legacy-archive'
@@ -43,13 +44,14 @@ const OWNER_ADAPTERS: FileOwnerAdapters<RetentionAdapter> = {
4344
},
4445
project: {
4546
async versions(ids, { budgets, label }, limit) {
46-
let deleted = 0
47+
const budget: RowBudget = { remaining: limit }
4748
for (const id of ids) {
48-
if (deleted >= limit) break
49-
deleted += await isolateOwnerFailure(Boolean(budgets), { label, projectId: id }, () =>
50-
cleanupProjectFileVersions(id, limit - deleted)
49+
if (budget.remaining <= 0) break
50+
await isolateOwnerFailure(Boolean(budgets), { label, projectId: id }, () =>
51+
cleanupProjectFileVersions(id, budget)
5152
)
5253
}
54+
const deleted = limit - budget.remaining
5355
return { deleted, attempted: deleted }
5456
},
5557
async archive(ids, { budgets, label }) {

‎apps/sim/lib/projects/files/__integration__/retention.integration.ts‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import { featureFlagsMock, featureFlagsMockFns } from '@sim/testing/mocks/featur
2323
import { getErrorMessage } from '@sim/utils/errors'
2424
import { sleep } from '@sim/utils/helpers'
2525
import { generateId } from '@sim/utils/id'
26+
import { compareStrings } from '@sim/utils/string'
2627
import { and, eq, inArray, sql } from 'drizzle-orm'
2728
import postgres from 'postgres'
2829
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
@@ -35,6 +36,7 @@ import { type CleanupJobPayload, runCleanupWithLimits } from '@/lib/billing/clea
3536
import { changeWorkspaceStoragePayerInTx } from '@/lib/billing/storage/payer-transfer'
3637
import { createCleanupBudgets } from '@/lib/cleanup/limits'
3738
import { beginFileArchiveCleanup, cleanupFileVersions } from '@/lib/file-retention'
39+
import { FILES_PER_QUERY } from '@/lib/file-retention/versions'
3840
import { authorizeProject } from '@/lib/projects/application/authorization'
3941
import { projectOperations } from '@/lib/projects/application/operations'
4042
import { lockWorkspaceProject } from '@/lib/projects/membership'
@@ -345,6 +347,50 @@ describe('Project file retention follows the current payer in PostgreSQL', () =>
345347
}
346348
)
347349

350+
check(
351+
'a Project that fails after committing a version batch still spends the shared run limit',
352+
async () => {
353+
const [first, second] = [await fixture(), await fixture()].sort((a, b) =>
354+
a.projectId < b.projectId ? -1 : 1
355+
)
356+
// One more file than a batch reads, so the first batch commits and the second one fails.
357+
const files = await Promise.all(
358+
Array.from({ length: FILES_PER_QUERY + 1 }, () => seedFile(first, 60 * 24))
359+
)
360+
const failingFileId = files
361+
.map(({ fileId }) => fileId)
362+
.sort(compareStrings)
363+
.at(-1)
364+
const survivor = await seedFile(second, 60 * 24)
365+
const trigger = `version_retention_failure_${generateId().replaceAll('-', '')}`
366+
await db.execute(
367+
sql.raw(`CREATE FUNCTION ${trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN
368+
IF OLD.file_id = '${failingFileId}' THEN RAISE EXCEPTION 'version retention failure fixture'; END IF; RETURN OLD; END $$`)
369+
)
370+
await db.execute(
371+
sql.raw(
372+
`CREATE TRIGGER ${trigger} BEFORE DELETE ON workspace_file_version FOR EACH ROW EXECUTE FUNCTION ${trigger}()`
373+
)
374+
)
375+
const committedByFirst = FILES_PER_QUERY * 2
376+
try {
377+
const deleted = await cleanupFileVersions(
378+
[
379+
{ entityType: 'project', entityId: first.projectId },
380+
{ entityType: 'project', entityId: second.projectId },
381+
],
382+
{ plan: 'free', cutoff: new Date(), label: 'failed-owner-spends-limit' },
383+
committedByFirst + 1
384+
)
385+
expect(deleted).toBe(committedByFirst + 1)
386+
expect(await versions(survivor.fileId)).toHaveLength(11)
387+
} finally {
388+
await db.execute(sql.raw(`DROP TRIGGER ${trigger} ON workspace_file_version`))
389+
await db.execute(sql.raw(`DROP FUNCTION ${trigger}()`))
390+
}
391+
}
392+
)
393+
348394
check(
349395
'omits empty Projects and batches retained Projects with the same cleanup policy',
350396
async () => {

‎apps/sim/lib/projects/files/retention.ts‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -64,14 +64,18 @@ function projectFiles(projectId: string) {
6464
return fileOwnerCondition({ entityType: 'project', entityId: projectId })
6565
}
6666

67-
/** Releases only superseded history, retaining the current head and the newest nine predecessors. */
67+
/**
68+
* Releases only superseded history, retaining the current head and the newest nine predecessors.
69+
* Each committed batch is charged to `budget` as it commits, so a later failure cannot erase
70+
* progress the shared run limit has to count.
71+
*/
6872
export async function cleanupProjectFileVersions(
6973
projectId: string,
70-
limit: number
74+
budget: RowBudget
7175
): Promise<number> {
7276
let released = 0
7377
let afterId = ''
74-
while (released < limit) {
78+
while (budget.remaining > 0) {
7579
const batch = await cleanupDb.transaction(async (tx) => {
7680
const policy = await prepareRetention(tx, projectId, 'cleanup-file-versions')
7781
if (!policy) return null
@@ -115,7 +119,7 @@ export async function cleanupProjectFileVersions(
115119
)
116120
.orderBy(asc(workspaceFiles.id))
117121
.for('update')
118-
const versionLimit = Math.min(DEFAULT_DELETE_CHUNK_SIZE, limit - released)
122+
const versionLimit = Math.min(DEFAULT_DELETE_CHUNK_SIZE, budget.remaining)
119123
const expired = await selectExpiredFileVersions(
120124
tx,
121125
candidates.map(({ id }) => id),
@@ -130,6 +134,7 @@ export async function cleanupProjectFileVersions(
130134
}
131135
})
132136
if (!batch) break
137+
consumeRowBudget(budget, batch.removed)
133138
afterId = batch.lastId
134139
released += batch.removed
135140
}

0 commit comments

Comments
 (0)