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. <!-- This is an auto-generated description by cubic. --> <a href="https://cubic.dev/pr/twentyhq/twenty/pull/21924?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> <!-- End of auto-generated description by cubic. --> --------- Co-authored-by: Ali Bildir <[alibildir@gmail.com]> Co-authored-by: Charles Bochet <charles@twenty.com>
This commit is contained in:
+8
-4
@@ -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'],
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
@@ -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(
|
||||
|
||||
+43
@@ -29,6 +29,7 @@ const createMockRepository = (): jest.Mocked<Repository<FakeEntity>> =>
|
||||
softDelete: jest.fn(),
|
||||
insert: jest.fn(),
|
||||
upsert: jest.fn(),
|
||||
create: jest.fn(),
|
||||
save: jest.fn(),
|
||||
createQueryBuilder: jest.fn(),
|
||||
}) as unknown as jest.Mocked<Repository<FakeEntity>>;
|
||||
@@ -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' });
|
||||
|
||||
+26
@@ -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<T extends WorkspaceScopedEntity> {
|
||||
);
|
||||
}
|
||||
|
||||
upsertAndReturnOne(
|
||||
workspaceId: string,
|
||||
entity: QueryDeepPartialEntity<T>,
|
||||
conflictPaths: string[],
|
||||
): Promise<T> {
|
||||
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<T>);
|
||||
});
|
||||
}
|
||||
|
||||
save<E extends DeepPartial<T>>(
|
||||
workspaceId: string,
|
||||
entity: E,
|
||||
|
||||
Reference in New Issue
Block a user