fix: preflight cyclic trash cleanup components

This commit is contained in:
Philipinho
2026-09-02 12:36:40 +01:00
parent 9dad90c142
commit 857c219f2b
2 changed files with 140 additions and 29 deletions
@@ -5,7 +5,10 @@ import { KyselyDB } from '@docmost/db/types/kysely.types';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
import { QueueJob, QueueName } from '../../../integrations/queue/constants';
import { assertAcyclicPageTraversal } from '../../../database/helpers/page-hierarchy-cycle';
import {
assertAcyclicPageTraversal,
PageHierarchyCycleError,
} from '../../../database/helpers/page-hierarchy-cycle';
import { sql } from 'kysely';
const DEFAULT_RETENTION_DAYS = 30;
@@ -30,8 +33,15 @@ export class TrashCleanupService {
.where('deletedAt', 'is', null)
.execute();
let totalCleaned = 0;
const cleanupCandidates: Array<{
pageId: string;
pageIds: string[];
}> = [];
const excludedPageIds = new Set<string>();
// Preflight every candidate before any cleanup mutates the snapshot.
// This lets a corrupt root exclude its whole reachable component even
// when one of its descendants is also an independently expired page.
for (const workspace of workspaces) {
const retentionDays =
workspace.trashRetentionDays ?? DEFAULT_RETENTION_DAYS;
@@ -47,15 +57,47 @@ export class TrashCleanupService {
.execute();
for (const page of oldDeletedPages) {
let descendants: Array<{ id: string; isCycle: boolean }> | undefined;
try {
await this.cleanupPage(page.id);
descendants = await this.getPageDescendants(page.id);
assertAcyclicPageTraversal(descendants, page.id);
cleanupCandidates.push({
pageId: page.id,
pageIds: descendants.map((descendant) => descendant.id),
});
} catch (error) {
if (error instanceof PageHierarchyCycleError) {
for (const descendant of descendants ?? []) {
excludedPageIds.add(descendant.id);
}
}
this.logCleanupError(page.id, error);
}
}
}
let totalCleaned = 0;
const cleanedPageIds = new Set<string>();
for (const candidate of cleanupCandidates) {
const pageIds = candidate.pageIds.filter(
(pageId) =>
!excludedPageIds.has(pageId) && !cleanedPageIds.has(pageId),
);
if (pageIds.length === 0) {
continue;
}
try {
await this.cleanupPage(candidate.pageId, pageIds);
for (const pageId of pageIds) {
cleanedPageIds.add(pageId);
}
totalCleaned++;
} catch (error) {
this.logger.error(
`Failed to cleanup page ${page.id}: ${error instanceof Error ? error.message : 'Unknown error'}`,
error instanceof Error ? error.stack : undefined,
);
}
this.logCleanupError(candidate.pageId, error);
}
}
@@ -72,9 +114,9 @@ export class TrashCleanupService {
}
}
private async cleanupPage(pageId: string) {
private async getPageDescendants(pageId: string) {
// Get all descendants using recursive CTE (including the page itself)
const descendants = await this.db
return this.db
.withRecursive('page_descendants', (db) =>
db
.selectFrom('pages')
@@ -99,10 +141,9 @@ export class TrashCleanupService {
.selectFrom('page_descendants')
.select(['id', 'isCycle'])
.execute();
}
assertAcyclicPageTraversal(descendants, pageId);
const pageIds = descendants.map((d) => d.id);
private async cleanupPage(pageId: string, pageIds: string[]) {
this.logger.debug(
`Cleaning up page ${pageId} with ${pageIds.length - 1} descendants`,
);
@@ -136,4 +177,11 @@ export class TrashCleanupService {
);
}
}
private logCleanupError(pageId: string, error: unknown) {
this.logger.error(
`Failed to cleanup page ${pageId}: ${error instanceof Error ? error.message : 'Unknown error'}`,
error instanceof Error ? error.stack : undefined,
);
}
}
@@ -384,12 +384,53 @@ describe('cycle-safe page hierarchy reads', () => {
it('trash cleanup logs and skips a corrupt root before continuing with acyclic trash', async () => {
const { self } = await seedSelfCycle();
const { root, child, grandchild } = await seedAcyclicPageChain();
const context = await db
.selectFrom('pages')
.select(['spaceId', 'workspaceId'])
.where('id', '=', self.id)
.executeTakeFirstOrThrow();
const corruptSideChild = await db
.insertInto('pages')
.values({
parentPageId: self.id,
slugId: randomUUID(),
spaceId: context.spaceId,
title: 'Corrupt side child',
workspaceId: context.workspaceId,
})
.returning(['id', 'parentPageId'])
.executeTakeFirstOrThrow();
const healthyRoot = await db
.insertInto('pages')
.values({
slugId: randomUUID(),
spaceId: context.spaceId,
title: 'Healthy cleanup root',
workspaceId: context.workspaceId,
})
.returning('id')
.executeTakeFirstOrThrow();
const healthyChild = await db
.insertInto('pages')
.values({
parentPageId: healthyRoot.id,
slugId: randomUUID(),
spaceId: context.spaceId,
title: 'Healthy cleanup child',
workspaceId: context.workspaceId,
})
.returning('id')
.executeTakeFirstOrThrow();
const expiredAt = new Date('2024-01-02T03:04:05.000Z');
await db
.updateTable('pages')
.set({ deletedAt: expiredAt })
.where('id', 'in', [self.id, root.id])
.where('id', 'in', [
self.id,
corruptSideChild.id,
healthyRoot.id,
healthyChild.id,
])
.execute();
const attachmentQueue = { add: jest.fn() };
const loggerError = jest
@@ -404,26 +445,48 @@ describe('cycle-safe page hierarchy reads', () => {
await timeoutCleanupService.cleanupOldTrash();
});
const corruptPage = await db
const corruptPages = await db
.selectFrom('pages')
.select(['id', 'deletedAt'])
.where('id', '=', self.id)
.executeTakeFirst();
const cleanedPageIds = await db
.select(['id', 'parentPageId', 'deletedAt'])
.where('id', 'in', [self.id, corruptSideChild.id])
.orderBy('id')
.execute();
const healthyPages = await db
.selectFrom('pages')
.select('id')
.where('id', 'in', [root.id, child.id, grandchild.id])
.where('id', 'in', [healthyRoot.id, healthyChild.id])
.execute();
expect(corruptPage).toEqual({ id: self.id, deletedAt: expiredAt });
expect(cleanedPageIds).toEqual([]);
expect(attachmentQueue.add).toHaveBeenCalledTimes(3);
expect(
attachmentQueue.add.mock.calls.map(([, payload]) => payload.pageId),
).toEqual(expect.arrayContaining([root.id, child.id, grandchild.id]));
expect(corruptPages).toEqual(
[
{ id: self.id, parentPageId: self.id, deletedAt: expiredAt },
{
id: corruptSideChild.id,
parentPageId: self.id,
deletedAt: expiredAt,
},
].sort((left, right) => left.id.localeCompare(right.id)),
);
expect(healthyPages).toEqual([]);
expect(attachmentQueue.add).toHaveBeenCalledTimes(2);
const queuedPageIds = attachmentQueue.add.mock.calls.map(
([, payload]) => payload.pageId,
);
expect(queuedPageIds).toEqual(
expect.arrayContaining([healthyRoot.id, healthyChild.id]),
);
expect(queuedPageIds).not.toEqual(
expect.arrayContaining([self.id, corruptSideChild.id]),
);
expect(loggerError).toHaveBeenCalledTimes(1);
expect(loggerError).toHaveBeenCalledWith(
expect.stringContaining(`Failed to cleanup page ${self.id}`),
expect.any(String),
`Failed to cleanup page ${self.id}: Cyclic page hierarchy detected`,
expect.stringContaining(
'PageHierarchyCycleError: Cyclic page hierarchy detected',
),
);
expect(loggerError.mock.invocationCallOrder[0]).toBeLessThan(
attachmentQueue.add.mock.invocationCallOrder[0],
);
});
});