feat(files): reap stale pending direct-upload files via hourly cron (#22531)

## Context Follow-up to #22449 (direct-to-storage upload endpoints). That PR introduced the `PENDING` → `UPLOADED` file lifecycle: `createFileUpload` inserts a file record in `PENDING`, the client uploads the bytes directly to storage, then `completeFileUpload` flips it to `UPLOADED`. A client that initiates an upload but never confirms — a crash, a closed tab, an expired presigned URL — leaves a `PENDING` file record and a possibly-partial storage object behind forever. This PR reaps them. ## What this does Adds an hourly cron that hard-deletes `PENDING` files older than 24h together with their storage objects, in bounded batches. - **`PendingFileCleanupService`** — finds `PENDING` files with `createdAt` older than `PENDING_FILE_MAX_AGE_MS` (24h), capped at `PENDING_FILE_CLEANUP_BATCH_SIZE` (200) per run, and deletes each via `FileStorageService.deleteByFileId` (which tolerates a missing object). A failure on one file is logged and skipped so the rest of the batch still gets cleaned. - **`PendingFileCleanupCronJob`** — `@Processor(cronQueue)` job that runs the service and reports exceptions. - **`PendingFileCleanupCronCommand`** — registers the job on the hourly pattern (`0 * * * *`). - Wired into `FileUploadModule` (providers + export) and registered in `cron:register:all`. ### Why 24h The reaper threshold sits well past the presigned URL expiry, so a `PENDING` file only becomes reapable long after any legitimate in-flight upload could still complete — the cleanup can never race a real upload. A file that was never confirmed is referenced by nothing; the client recovery path is simply re-uploading under a fresh `fileId`, so we never promote to `UPLOADED`. ## Tests `pending-file-cleanup.service.spec.ts` covers: the query shape (status + age threshold + batch cap), deleting each stale file and returning the count, continuing past a per-file deletion failure, and the empty-batch no-op. ## Scope Server-only, non-breaking, no user-facing change. Part of the incremental direct-upload rollout being split into small PRs. https://claude.ai/code/session_015UH8KWmsB9zdYaog8MFG1d --- _Generated by [Claude Code](https://claude.ai/code/session_015UH8KWmsB9zdYaog8MFG1d)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22531?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-07-06 10:39:32 +02:00
parent d6b6962604
commit e5e3fadbbb
7 changed files with 371 additions and 1 deletions
@@ -9,6 +9,7 @@ import { ApplicationVersionCheckCronCommand } from 'src/engine/core-modules/appl
 import { BillingReminderCronCommand } from 'src/engine/core-modules/billing/reminders/crons/commands/billing-reminder.cron.command';
 import { EnterpriseKeyValidationCronCommand } from 'src/engine/core-modules/enterprise/cron/command/enterprise-key-validation.cron.command';
 import { EventLogCleanupCronCommand } from 'src/engine/core-modules/event-logs/cleanup/commands/event-log-cleanup.cron.command';
+import { PendingFileCleanupCronCommand } from 'src/engine/core-modules/file/file-upload/crons/commands/pending-file-cleanup.cron.command';
 import { RotateSigningKeysCronCommand } from 'src/engine/core-modules/jwt/crons/commands/rotate-signing-keys.cron.command';
 import { CronTriggerCronCommand } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/cron/cron-trigger.cron.command';
 import { CheckPublicDomainsValidRecordsCronCommand } from 'src/engine/core-modules/public-domain/crons/commands/check-public-domains-valid-records.cron.command';
@@ -65,6 +66,7 @@ export class CronRegisterAllCommand extends CommandRunner {
    private readonly marketplaceCatalogSyncCronCommand: MarketplaceCatalogSyncCronCommand,
    private readonly applicationVersionCheckCronCommand: ApplicationVersionCheckCronCommand,
    private readonly staleRegistrationCleanupCronCommand: StaleRegistrationCleanupCronCommand,
+    private readonly pendingFileCleanupCronCommand: PendingFileCleanupCronCommand,
    private readonly billingReminderCronCommand: BillingReminderCronCommand,
    private readonly twentyConfigService: TwentyConfigService,
  ) {
@@ -183,6 +185,10 @@ export class CronRegisterAllCommand extends CommandRunner {
        name: 'StaleRegistrationCleanup',
        command: this.staleRegistrationCleanupCronCommand,
      },
+      {
+        name: 'PendingFileCleanup',
+        command: this.pendingFileCleanupCronCommand,
+      },
      {
        name: 'BillingReminder',
        command: this.billingReminderCronCommand,
@@ -0,0 +1,33 @@
+import { Command, CommandRunner } from 'nest-commander';
+
+import { PENDING_FILE_CLEANUP_CRON_PATTERN } from 'src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants';
+import { PendingFileCleanupCronJob } from 'src/engine/core-modules/file/file-upload/crons/jobs/pending-file-cleanup.cron.job';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+
+@Command({
+  name: 'cron:pending-file-cleanup',
+  description:
+    'Starts a cron job to clean up stale pending direct file uploads',
+})
+export class PendingFileCleanupCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>({
+      jobName: PendingFileCleanupCronJob.name,
+      data: undefined,
+      options: {
+        repeat: {
+          pattern: PENDING_FILE_CLEANUP_CRON_PATTERN,
+        },
+      },
+    });
+  }
+}
@@ -0,0 +1,7 @@
+export const PENDING_FILE_CLEANUP_CRON_PATTERN = '0 * * * *';
+
+// A pending file only becomes reapable long after its upload URL has expired,
+// so the cleanup can never race a legitimate in-flight upload.
+export const PENDING_FILE_MAX_AGE_MS = 24 * 60 * 60 * 1000;
+
+export const PENDING_FILE_CLEANUP_BATCH_SIZE = 200;
@@ -0,0 +1,41 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
+import { PENDING_FILE_CLEANUP_CRON_PATTERN } from 'src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants';
+import { PendingFileCleanupService } from 'src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service';
+import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+
+@Injectable()
+@Processor(MessageQueue.cronQueue)
+export class PendingFileCleanupCronJob {
+  private readonly logger = new Logger(PendingFileCleanupCronJob.name);
+
+  constructor(
+    private readonly pendingFileCleanupService: PendingFileCleanupService,
+    private readonly exceptionHandlerService: ExceptionHandlerService,
+  ) {}
+
+  @Process(PendingFileCleanupCronJob.name)
+  @SentryCronMonitor(
+    PendingFileCleanupCronJob.name,
+    PENDING_FILE_CLEANUP_CRON_PATTERN,
+  )
+  async handle(): Promise<void> {
+    try {
+      const deletedCount =
+        await this.pendingFileCleanupService.cleanupStalePendingFiles();
+
+      if (deletedCount > 0) {
+        this.logger.log(
+          `Pending file cleanup completed: ${deletedCount} stale file(s) deleted`,
+        );
+      }
+    } catch (error) {
+      this.exceptionHandlerService.captureExceptions([error]);
+      throw error;
+    }
+  }
+}
@@ -6,9 +6,12 @@ import { ApplicationModule } from 'src/engine/core-modules/application/applicati
 import { FileStorageModule } from 'src/engine/core-modules/file-storage/file-storage.module';
 import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity';
 import { FileUploadController } from 'src/engine/core-modules/file/file-upload/controllers/file-upload.controller';
+import { PendingFileCleanupCronCommand } from 'src/engine/core-modules/file/file-upload/crons/commands/pending-file-cleanup.cron.command';
+import { PendingFileCleanupCronJob } from 'src/engine/core-modules/file/file-upload/crons/jobs/pending-file-cleanup.cron.job';
 import { FileUploadTokenGuard } from 'src/engine/core-modules/file/file-upload/guards/file-upload-token.guard';
 import { FileUploadResolver } from 'src/engine/core-modules/file/file-upload/resolvers/file-upload.resolver';
 import { FileUploadService } from 'src/engine/core-modules/file/file-upload/services/file-upload.service';
+import { PendingFileCleanupService } from 'src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service';
 import { FileUrlModule } from 'src/engine/core-modules/file/file-url/file-url.module';
 import { JwtModule } from 'src/engine/core-modules/jwt/jwt.module';
 import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity';
@@ -32,9 +35,12 @@ import { provideWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspac
    FileUploadService,
    FileUploadResolver,
    FileUploadTokenGuard,
+    PendingFileCleanupService,
+    PendingFileCleanupCronJob,
+    PendingFileCleanupCronCommand,
    provideWorkspaceScopedRepository(FileEntity),
  ],
-  exports: [FileUploadService],
+  exports: [FileUploadService, PendingFileCleanupCronCommand],
  controllers: [FileUploadController],
 })
 export class FileUploadModule {}
@@ -0,0 +1,179 @@
+import { Test, type TestingModule } from '@nestjs/testing';
+import { getRepositoryToken } from '@nestjs/typeorm';
+
+import { FileFolder } from 'twenty-shared/types';
+
+import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity';
+import { FileStorageService } from 'src/engine/core-modules/file-storage/file-storage.service';
+import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity';
+import {
+  PENDING_FILE_CLEANUP_BATCH_SIZE,
+  PENDING_FILE_MAX_AGE_MS,
+} from 'src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants';
+import { PendingFileCleanupService } from 'src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service';
+import { FILE_STATUS } from 'src/engine/core-modules/file/types/file-status.types';
+
+describe('PendingFileCleanupService', () => {
+  let service: PendingFileCleanupService;
+
+  const fileRepository = {
+    find: jest.fn(),
+    delete: jest.fn(),
+  };
+
+  const applicationRepository = {
+    findOne: jest.fn(),
+  };
+
+  const fileStorageService = {
+    deleteFile: jest.fn(),
+  };
+
+  beforeEach(async () => {
+    jest.clearAllMocks();
+
+    applicationRepository.findOne.mockResolvedValue({
+      id: 'application-id',
+      universalIdentifier: 'application-uid',
+    });
+    fileRepository.delete.mockResolvedValue({ affected: 1 });
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        PendingFileCleanupService,
+        {
+          provide: getRepositoryToken(FileEntity),
+          useValue: fileRepository,
+        },
+        {
+          provide: getRepositoryToken(ApplicationEntity),
+          useValue: applicationRepository,
+        },
+        {
+          provide: FileStorageService,
+          useValue: fileStorageService,
+        },
+      ],
+    }).compile();
+
+    service = module.get<PendingFileCleanupService>(PendingFileCleanupService);
+  });
+
+  it('should query for stale pending files with the batch size cap', async () => {
+    fileRepository.find.mockResolvedValue([]);
+
+    const before = Date.now() - PENDING_FILE_MAX_AGE_MS;
+
+    await service.cleanupStalePendingFiles();
+
+    const after = Date.now() - PENDING_FILE_MAX_AGE_MS;
+
+    expect(fileRepository.find).toHaveBeenCalledTimes(1);
+
+    const where = fileRepository.find.mock.calls[0][0].where;
+
+    expect(where.status).toBe(FILE_STATUS.PENDING);
+    // LessThan wraps the threshold in _value
+    const threshold = where.createdAt._value.getTime();
+
+    expect(threshold).toBeGreaterThanOrEqual(before);
+    expect(threshold).toBeLessThanOrEqual(after);
+    expect(fileRepository.find.mock.calls[0][0].take).toBe(
+      PENDING_FILE_CLEANUP_BATCH_SIZE,
+    );
+  });
+
+  it('should delete each stale file guarded by status and return the deleted count', async () => {
+    fileRepository.find.mockResolvedValue([
+      {
+        id: 'file-1',
+        workspaceId: 'workspace-1',
+        applicationId: 'application-id',
+        path: `${FileFolder.FilesField}/some/file-1.png`,
+      },
+      {
+        id: 'file-2',
+        workspaceId: 'workspace-2',
+        applicationId: 'application-id',
+        path: `${FileFolder.Workflow}/file-2.pdf`,
+      },
+    ]);
+
+    const deletedCount = await service.cleanupStalePendingFiles();
+
+    expect(deletedCount).toBe(2);
+    expect(fileRepository.delete).toHaveBeenNthCalledWith(1, {
+      id: 'file-1',
+      status: FILE_STATUS.PENDING,
+    });
+    expect(fileRepository.delete).toHaveBeenNthCalledWith(2, {
+      id: 'file-2',
+      status: FILE_STATUS.PENDING,
+    });
+    expect(fileStorageService.deleteFile).toHaveBeenNthCalledWith(1, {
+      workspaceId: 'workspace-1',
+      applicationUniversalIdentifier: 'application-uid',
+      fileFolder: FileFolder.FilesField,
+      resourcePath: 'some/file-1.png',
+    });
+    expect(fileStorageService.deleteFile).toHaveBeenNthCalledWith(2, {
+      workspaceId: 'workspace-2',
+      applicationUniversalIdentifier: 'application-uid',
+      fileFolder: FileFolder.Workflow,
+      resourcePath: 'file-2.pdf',
+    });
+  });
+
+  it('should skip storage deletion when the file was confirmed concurrently', async () => {
+    fileRepository.find.mockResolvedValue([
+      {
+        id: 'file-1',
+        workspaceId: 'workspace-1',
+        applicationId: 'application-id',
+        path: `${FileFolder.FilesField}/file-1.png`,
+      },
+    ]);
+    // A racing completeFileUpload promoted the row, so the guarded delete is a no-op.
+    fileRepository.delete.mockResolvedValueOnce({ affected: 0 });
+
+    const deletedCount = await service.cleanupStalePendingFiles();
+
+    expect(deletedCount).toBe(0);
+    expect(fileStorageService.deleteFile).not.toHaveBeenCalled();
+  });
+
+  it('should keep cleaning remaining files when one deletion fails', async () => {
+    fileRepository.find.mockResolvedValue([
+      {
+        id: 'file-1',
+        workspaceId: 'workspace-1',
+        applicationId: 'application-id',
+        path: `${FileFolder.FilesField}/file-1.png`,
+      },
+      {
+        id: 'file-2',
+        workspaceId: 'workspace-2',
+        applicationId: 'application-id',
+        path: `${FileFolder.FilesField}/file-2.png`,
+      },
+    ]);
+    fileStorageService.deleteFile
+      .mockRejectedValueOnce(new Error('storage exploded'))
+      .mockResolvedValueOnce(undefined);
+
+    const deletedCount = await service.cleanupStalePendingFiles();
+
+    expect(deletedCount).toBe(1);
+    expect(fileStorageService.deleteFile).toHaveBeenCalledTimes(2);
+  });
+
+  it('should not delete anything when there are no stale files', async () => {
+    fileRepository.find.mockResolvedValue([]);
+
+    const deletedCount = await service.cleanupStalePendingFiles();
+
+    expect(deletedCount).toBe(0);
+    expect(fileRepository.delete).not.toHaveBeenCalled();
+    expect(fileStorageService.deleteFile).not.toHaveBeenCalled();
+  });
+});
@@ -0,0 +1,98 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { FileFolder } from 'twenty-shared/types';
+import { isDefined } from 'twenty-shared/utils';
+import { LessThan, Repository } from 'typeorm';
+
+import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity';
+import { FileStorageService } from 'src/engine/core-modules/file-storage/file-storage.service';
+import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity';
+import {
+  PENDING_FILE_CLEANUP_BATCH_SIZE,
+  PENDING_FILE_MAX_AGE_MS,
+} from 'src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants';
+import { FILE_STATUS } from 'src/engine/core-modules/file/types/file-status.types';
+import { removeFileFolderFromFileEntityPath } from 'src/engine/core-modules/file/utils/remove-file-folder-from-file-entity-path.utils';
+
+@Injectable()
+export class PendingFileCleanupService {
+  private readonly logger = new Logger(PendingFileCleanupService.name);
+
+  constructor(
+    // eslint-disable-next-line twenty/prefer-workspace-scoped-repository -- the reaper runs in a cron with no workspace context and must sweep stale PENDING files across every workspace
+    @InjectRepository(FileEntity)
+    private readonly fileRepository: Repository<FileEntity>,
+    // eslint-disable-next-line twenty/prefer-workspace-scoped-repository -- resolves the application universalIdentifier of a cross-workspace file while reaping outside any workspace context
+    @InjectRepository(ApplicationEntity)
+    private readonly applicationRepository: Repository<ApplicationEntity>,
+    private readonly fileStorageService: FileStorageService,
+  ) {}
+
+  // Deletes file records stuck in PENDING (direct uploads that were initiated
+  // but never completed) together with any partially uploaded object. Never
+  // promotes to UPLOADED: a file that was never confirmed is referenced by
+  // nothing, the client recovery path is re-uploading under a fresh fileId.
+  async cleanupStalePendingFiles(): Promise<number> {
+    const staleThreshold = new Date(Date.now() - PENDING_FILE_MAX_AGE_MS);
+
+    const staleFiles = await this.fileRepository.find({
+      where: {
+        status: FILE_STATUS.PENDING,
+        createdAt: LessThan(staleThreshold),
+      },
+      take: PENDING_FILE_CLEANUP_BATCH_SIZE,
+    });
+
+    let deletedCount = 0;
+
+    for (const file of staleFiles) {
+      try {
+        // Claim the row atomically: delete it only while it is still PENDING.
+        // If completeFileUpload promoted it to UPLOADED between the fetch above
+        // and here, the delete affects no rows and the now-live file (and its
+        // object) are left untouched.
+        const { affected } = await this.fileRepository.delete({
+          id: file.id,
+          status: FILE_STATUS.PENDING,
+        });
+
+        if (!isDefined(affected) || affected === 0) {
+          continue;
+        }
+
+        await this.deleteStorageObject(file);
+
+        deletedCount++;
+      } catch (error) {
+        this.logger.warn(
+          `Failed to clean up stale pending file ${file.id} in workspace ${file.workspaceId}: ${error.message}`,
+        );
+      }
+    }
+
+    return deletedCount;
+  }
+
+  // The row has already been removed, so this only tidies the (possibly
+  // partial, possibly absent) storage object. A failure here leaks bytes but
+  // never data, so it is logged rather than retried.
+  private async deleteStorageObject(file: FileEntity): Promise<void> {
+    const [fileFolder] = file.path.split('/');
+
+    const application = await this.applicationRepository.findOne({
+      where: { id: file.applicationId, workspaceId: file.workspaceId },
+    });
+
+    if (!isDefined(application)) {
+      return;
+    }
+
+    await this.fileStorageService.deleteFile({
+      workspaceId: file.workspaceId,
+      applicationUniversalIdentifier: application.universalIdentifier,
+      fileFolder: fileFolder as FileFolder,
+      resourcePath: removeFileFolderFromFileEntityPath(file.path),
+    });
+  }
+}