From 35d64ac7f216d46b0d39b199e6c7a1e675f84b0d Mon Sep 17 00:00:00 2001 From: alibildir <84903907+alibildir@users.noreply.github.com> Date: Thu, 25 Jun 2026 19:10:07 +0300 Subject: [PATCH] fix(server): wrap file storage upsert and read in transaction (#21924) ## Summary This PR fixes a race condition occurring during file and image uploads in environments that use Pgpool-II or similar connection poolers with read-replica scaling. Currently, in `FileStorageService.writeFile`, `fileRepository.upsert` writes the file record to the primary database node. However, the immediate subsequent call to `fileRepository.findOneOrFail` is executed outside of a transaction. Consequently, connection poolers like Pgpool can route this `SELECT` query to a read-replica. Due to replication lag, the replica may not yet reflect the newly inserted record, throwing an `EntityNotFoundError` and failing the upload process (even though the file is successfully saved in S3 and the database). This PR wraps both operations in a TypeORM database transaction when `queryRunner` is not provided. This ensures that the `SELECT` query correctly targets the primary node, guaranteeing immediate read-after-write consistency. ## Affected version - Twenty Self-hosted (e.g. `v2.14.x`) configured with Pgpool/read-replicas. ## Changes Made - **`file-storage.service.ts`**: Wrapped `transactionalFileRepo.upsert` and `transactionalFileRepo.findOneOrFail` within `this.applicationRepository.manager.transaction` to ensure read-after-write consistency. ## How to Test 1. Set up Twenty in a self-hosted environment using Pgpool configured with load balancing / read-replicas. 2. Attempt to upload a file to a `Files` custom field or upload an image as an organization logo. 3. Observe that the upload completes successfully without throwing an `EntityNotFoundError` in the server logs. Review in cubic --------- Co-authored-by: Ali Bildir <[alibildir@gmail.com]> Co-authored-by: Charles Bochet --- .../__tests__/file-storage.service.spec.ts | 12 ++++-- .../file-storage/file-storage.service.ts | 9 +--- .../workspace-scoped-repository.spec.ts | 43 +++++++++++++++++++ .../workspace-scoped-repository.ts | 26 +++++++++++ 4 files changed, 78 insertions(+), 12 deletions(-) diff --git a/packages/twenty-server/src/engine/core-modules/file-storage/__tests__/file-storage.service.spec.ts b/packages/twenty-server/src/engine/core-modules/file-storage/__tests__/file-storage.service.spec.ts index b1c1a4ab33..88b7e31c00 100644 --- a/packages/twenty-server/src/engine/core-modules/file-storage/__tests__/file-storage.service.spec.ts +++ b/packages/twenty-server/src/engine/core-modules/file-storage/__tests__/file-storage.service.spec.ts @@ -23,7 +23,7 @@ describe('FileStorageService', () => { const mockFileRepository = { save: jest.fn(), - upsert: jest.fn(), + upsertAndReturnOne: jest.fn(), findOneOrFail: jest.fn(), delete: jest.fn(), }; @@ -179,7 +179,11 @@ describe('FileStorageService', () => { id: 'app-id', universalIdentifier: 'app-uid', }); - mockFileRepository.upsert.mockResolvedValue(undefined); + mockFileRepository.upsertAndReturnOne.mockResolvedValue({ + id: 'file-id', + path: 'BuiltFrontComponent/file.mjs', + mimeType: 'application/javascript', + }); mockFileRepository.findOneOrFail.mockResolvedValue({ id: 'file-id', path: 'BuiltFrontComponent/file.mjs', @@ -442,10 +446,10 @@ describe('FileStorageService', () => { expect(mockDriver.writeFile).toHaveBeenCalledWith( expect.objectContaining({ mimeType: 'image/png' }), ); - expect(mockFileRepository.upsert).toHaveBeenCalledWith( + expect(mockFileRepository.upsertAndReturnOne).toHaveBeenCalledWith( 'workspace-123', expect.objectContaining({ mimeType: 'image/png' }), - expect.anything(), + ['path', 'workspaceId', 'applicationId'], ); }); diff --git a/packages/twenty-server/src/engine/core-modules/file-storage/file-storage.service.ts b/packages/twenty-server/src/engine/core-modules/file-storage/file-storage.service.ts index f564e2cda9..d42c2c2867 100644 --- a/packages/twenty-server/src/engine/core-modules/file-storage/file-storage.service.ts +++ b/packages/twenty-server/src/engine/core-modules/file-storage/file-storage.service.ts @@ -170,7 +170,7 @@ export class FileStorageService { sourceFile: persistedSourceFile, }); - await fileRepository.upsert( + return fileRepository.upsertAndReturnOne( workspaceId, { path: filePath, @@ -185,13 +185,6 @@ export class FileStorageService { }, ['path', 'workspaceId', 'applicationId'], ); - - return fileRepository.findOneOrFail(workspaceId, { - where: { - path: filePath, - applicationId: application.id, - }, - }); } async getPresignedUrl( diff --git a/packages/twenty-server/src/engine/twenty-orm/workspace-scoped-repository/__tests__/workspace-scoped-repository.spec.ts b/packages/twenty-server/src/engine/twenty-orm/workspace-scoped-repository/__tests__/workspace-scoped-repository.spec.ts index f58b0eb35d..9efb71ab81 100644 --- a/packages/twenty-server/src/engine/twenty-orm/workspace-scoped-repository/__tests__/workspace-scoped-repository.spec.ts +++ b/packages/twenty-server/src/engine/twenty-orm/workspace-scoped-repository/__tests__/workspace-scoped-repository.spec.ts @@ -29,6 +29,7 @@ const createMockRepository = (): jest.Mocked> => softDelete: jest.fn(), insert: jest.fn(), upsert: jest.fn(), + create: jest.fn(), save: jest.fn(), createQueryBuilder: jest.fn(), }) as unknown as jest.Mocked>; @@ -65,6 +66,10 @@ describe('WorkspaceScopedRepository', () => { ['softDelete', () => scoped.softDelete(undefined as never, {})], ['insert', () => scoped.insert(undefined as never, {})], ['upsert', () => scoped.upsert(undefined as never, {}, ['id'])], + [ + 'upsertAndReturnOne', + () => scoped.upsertAndReturnOne(undefined as never, {}, ['id']), + ], ['save', () => scoped.save(undefined as never, {})], ['saveMany', () => scoped.saveMany(undefined as never, [{}])], ['maximum', () => scoped.maximum(undefined as never, 'id')], @@ -381,6 +386,44 @@ describe('WorkspaceScopedRepository', () => { }); }); + describe('upsertAndReturnOne', () => { + it('upserts with RETURNING and hydrates the row from generatedMaps', async () => { + const persistedRow = { + id: 'a', + status: 'queued', + workspaceId: WORKSPACE_ID, + }; + + (repository.upsert as jest.Mock).mockResolvedValue({ + generatedMaps: [persistedRow], + }); + (repository.create as jest.Mock).mockReturnValue(persistedRow); + + const result = await scoped.upsertAndReturnOne( + WORKSPACE_ID, + { id: 'a', status: 'queued' }, + ['id'], + ); + + expect(repository.upsert).toHaveBeenCalledWith( + { id: 'a', status: 'queued', workspaceId: WORKSPACE_ID }, + { conflictPaths: ['id'], returning: '*' }, + ); + expect(repository.create).toHaveBeenCalledWith(persistedRow); + expect(result).toBe(persistedRow); + }); + + it('throws instead of returning a hollow entity when no row is returned', async () => { + (repository.upsert as jest.Mock).mockResolvedValue({ generatedMaps: [] }); + + await expect( + scoped.upsertAndReturnOne(WORKSPACE_ID, { id: 'a' }, ['id']), + ).rejects.toThrow(/upsert returned no row/); + + expect(repository.create).not.toHaveBeenCalled(); + }); + }); + describe('save', () => { it('stamps workspaceId on the entity passed to save', async () => { await scoped.save(WORKSPACE_ID, { id: 'a', status: 'queued' }); diff --git a/packages/twenty-server/src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository.ts b/packages/twenty-server/src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository.ts index 715db9c507..11e5be86a8 100644 --- a/packages/twenty-server/src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository.ts +++ b/packages/twenty-server/src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository.ts @@ -11,6 +11,7 @@ import { type SelectQueryBuilder, type UpdateResult, } from 'typeorm'; +import { isDefined } from 'twenty-shared/utils'; import { type QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity'; import { type UpsertOptions } from 'typeorm/repository/UpsertOptions'; @@ -210,6 +211,31 @@ export class WorkspaceScopedRepository { ); } + upsertAndReturnOne( + workspaceId: string, + entity: QueryDeepPartialEntity, + conflictPaths: string[], + ): Promise { + this.assertWorkspaceId(workspaceId); + + return this.repository + .upsert(this.stampWorkspaceIdOnEntities(workspaceId, entity), { + conflictPaths, + returning: '*', + }) + .then(({ generatedMaps }) => { + const [persistedRow] = generatedMaps; + + if (!isDefined(persistedRow)) { + throw new Error( + 'WorkspaceScopedRepository.upsertAndReturnOne: upsert returned no row.', + ); + } + + return this.repository.create(persistedRow as DeepPartial); + }); + } + save>( workspaceId: string, entity: E,