From 3c39d5d5c8ee396d6c72ef8d416927a098396cd2 Mon Sep 17 00:00:00 2001 From: James Acres Date: Fri, 30 Oct 2026 12:38:13 +0000 Subject: [PATCH 1/2] feat: support typeorm slave option --- .../01-transactional/05-typeorm-adapter.md | 34 +++++ .../src/lib/transactional-adapter-typeorm.ts | 56 +++++++ .../transactional-adapter-typeorm.spec.ts | 139 +++++++++++++++++- 3 files changed, 226 insertions(+), 3 deletions(-) diff --git a/docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md b/docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md index 21b323ea..a1f4d9ac 100644 --- a/docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md +++ b/docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md @@ -75,6 +75,40 @@ new TransactionalAdapterTypeOrm({ The `tx` property on the `TransactionHost` is typed as [`EntityManager`](https://typeorm.io/working-with-entity-manager). +## Running on a read replica + +When the `DataSource` is configured with [replication](https://typeorm.io/data-source-options#replication), you can route a whole transaction to a read replica by passing `mode: 'slave'` in the transaction options. + +This is stronger than TypeORM's default behaviour: TypeORM only sends plain `find`-style reads to a replica, whereas `mode: 'slave'` runs the transaction on a replica query runner so that **every** read in the callback - including custom query-builder reads - goes to the replica. + +```ts +@Injectable() +class UserService { + constructor(private readonly userRepository: UserRepository) {} + + // highlight-next-line + @Transactional({ mode: 'slave' }) + async getReportData() { + // every read here runs against a read replica + return await this.userRepository.getActiveUsers(); + } +} +``` + +The same option works with the function wrapper: + +```ts +await this.txHost.withTransaction({ mode: 'slave' }, async () => { + // ... +}); +``` + +:::note + +`mode` defaults to `'master'`. Nested transactions are not supported in `'slave'` mode - propagating a nested transaction with `mode: 'slave'` throws an error. + +::: + ## Example ```ts title="user.service.ts" diff --git a/packages/transactional-adapters/transactional-adapter-typeorm/src/lib/transactional-adapter-typeorm.ts b/packages/transactional-adapters/transactional-adapter-typeorm/src/lib/transactional-adapter-typeorm.ts index 8a357f8d..81b74d3f 100644 --- a/packages/transactional-adapters/transactional-adapter-typeorm/src/lib/transactional-adapter-typeorm.ts +++ b/packages/transactional-adapters/transactional-adapter-typeorm/src/lib/transactional-adapter-typeorm.ts @@ -1,6 +1,7 @@ import { TransactionalAdapter } from '@nestjs-cls/transactional'; import { DataSource, EntityManager } from 'typeorm'; import type { IsolationLevel } from 'typeorm/driver/types/IsolationLevel.js'; +import type { ReplicationMode } from 'typeorm/driver/types/ReplicationMode.js'; export interface TypeOrmTransactionalAdapterOptions { /** @@ -17,6 +18,21 @@ export interface TypeOrmTransactionalAdapterOptions { export interface TypeOrmTransactionOptions { isolationLevel: IsolationLevel; + + /** + * Which replica to run the transaction against, using TypeORM's + * {@link ReplicationMode}. Only has an effect when the `DataSource` is + * configured with replication. Defaults to `'master'`. + * + * - `'master'` runs on the primary/write connection. + * - `'slave'` runs on a read replica. Unlike TypeORM's default behaviour - + * which only routes plain `find`-style reads to a replica - this forces + * the whole transaction (including custom query-builder reads) onto the + * replica. + * + * Note: nested transactions are not supported in `'slave'` mode. + */ + mode?: ReplicationMode; } export class TransactionalAdapterTypeOrm implements TransactionalAdapter< @@ -39,6 +55,14 @@ export class TransactionalAdapterTypeOrm implements TransactionalAdapter< fn: (...args: any[]) => Promise, setClient: (client?: EntityManager) => void, ) => { + if (options?.mode === 'slave') { + return this.wrapWithReplicaTransaction( + dataSource, + options, + fn, + setClient, + ); + } return dataSource.transaction(options?.isolationLevel, (trx) => { setClient(trx); return fn(); @@ -51,6 +75,11 @@ export class TransactionalAdapterTypeOrm implements TransactionalAdapter< setClient: (client?: EntityManager) => void, client: EntityManager, ) => { + if (options?.mode === 'slave') { + throw new Error( + 'Nested transactions are not supported in slave mode', + ); + } return client.transaction(options?.isolationLevel, (trx) => { setClient(trx); return fn(); @@ -58,4 +87,31 @@ export class TransactionalAdapterTypeOrm implements TransactionalAdapter< }, getFallbackInstance: () => dataSource.manager, }); + + /** + * Runs a transaction on a replica (read) query runner so that every read in + * the callback - including custom query-builder reads, which TypeORM would + * otherwise send to the master - is routed to the replica. + */ + private async wrapWithReplicaTransaction( + dataSource: DataSource, + options: TypeOrmTransactionOptions, + fn: (...args: any[]) => Promise, + setClient: (client?: EntityManager) => void, + ) { + const queryRunner = dataSource.createQueryRunner('slave'); + await queryRunner.connect(); + await queryRunner.startTransaction(options?.isolationLevel); + setClient(queryRunner.manager); + try { + const result = await fn(); + await queryRunner.commitTransaction(); + return result; + } catch (error) { + await queryRunner.rollbackTransaction(); + throw error; + } finally { + await queryRunner.release(); + } + } } diff --git a/packages/transactional-adapters/transactional-adapter-typeorm/test/transactional-adapter-typeorm.spec.ts b/packages/transactional-adapters/transactional-adapter-typeorm/test/transactional-adapter-typeorm.spec.ts index 1705d658..cfc5922a 100644 --- a/packages/transactional-adapters/transactional-adapter-typeorm/test/transactional-adapter-typeorm.spec.ts +++ b/packages/transactional-adapters/transactional-adapter-typeorm/test/transactional-adapter-typeorm.spec.ts @@ -1,8 +1,9 @@ +import { jest } from '@jest/globals'; import { ClsPluginTransactional, InjectTransaction, Propagation, - Transaction, + type Transaction, Transactional, TransactionHost, } from '@nestjs-cls/transactional'; @@ -25,13 +26,23 @@ class User { email?: string; } -const dataSource = new DataSource({ - type: 'postgres', +const connectionOptions = { host: 'localhost', port: 5446, username: 'postgres', password: 'postgres', database: 'postgres', +}; + +const dataSource = new DataSource({ + type: 'postgres', + // Configured with replication so the `slave` transaction mode has a replica + // to route to. Both master and replicas point at the same database (there + // is a single test instance), which is enough to exercise the routing. + replication: { + master: connectionOptions, + slaves: [connectionOptions], + }, entities: [User], synchronize: true, }); @@ -112,6 +123,48 @@ class UserService { throw new Error('Rollback'); } + @Transactional({ + mode: 'slave', + isolationLevel: 'SERIALIZABLE', + }) + async readInSlaveModeWithIsolationLevel(id: number) { + return await this.userRepository.getUserById(id); + } + + @Transactional({ mode: 'slave' }) + async readInSlaveMode(id: number) { + return await this.dataSource + .getRepository(User) + .createQueryBuilder('user') + .where('user.id = :id', { id }) + .getOne(); + } + + async readInSlaveModeWithFunctionWrapper(id: number) { + return this.transactionHost.withTransaction( + { mode: 'slave' }, + async () => this.userRepository.getUserById(id), + ); + } + + @Transactional({ mode: 'slave' }) + async slaveModeError() { + await this.userRepository.getUserById(1); + throw new Error('Rollback'); + } + + @Transactional({ mode: 'slave' }) + async slaveModeHasNested() { + await this.nestedSlaveTransaction(); + } + + @Transactional(Propagation.Nested, { + mode: 'slave', + }) + async nestedSlaveTransaction() { + await this.userRepository.getUserById(1); + } + @Transactional() async transactionalHasNested(name?: string) { await this.nestedTransaction(name); @@ -253,6 +306,86 @@ describe('Transactional', () => { ); }); }); + + describe('Slave mode', () => { + let createQueryRunnerSpy: ReturnType; + let startTransactionSpy: ReturnType; + + beforeEach(() => { + const original = dataSource.createQueryRunner.bind(dataSource); + createQueryRunnerSpy = jest + .spyOn(dataSource, 'createQueryRunner') + .mockImplementation((mode) => { + const queryRunner = original(mode); + startTransactionSpy = jest.spyOn( + queryRunner, + 'startTransaction', + ); + return queryRunner; + }); + }); + + afterEach(() => { + startTransactionSpy?.mockRestore(); + createQueryRunnerSpy.mockRestore(); + }); + + it('should route the transaction to a replica query runner', async () => { + const user = await callingService.withoutTransaction(); + + const found = await callingService.readInSlaveMode(user.r1.id); + + expect(found).toEqual(user.r1); + expect(createQueryRunnerSpy).toHaveBeenCalledWith('slave'); + }); + + it('should route to a replica with the function wrapper', async () => { + const user = await callingService.withoutTransaction(); + + const found = + await callingService.readInSlaveModeWithFunctionWrapper( + user.r1.id, + ); + + expect(found).toEqual(user.r1); + expect(createQueryRunnerSpy).toHaveBeenCalledWith('slave'); + }); + + it('should start the replica transaction with the given isolation level', async () => { + const user = await callingService.withoutTransaction(); + + const found = + await callingService.readInSlaveModeWithIsolationLevel( + user.r1.id, + ); + + expect(found).toEqual(user.r1); + expect(createQueryRunnerSpy).toHaveBeenCalledWith('slave'); + + const queryRunner = createQueryRunnerSpy.mock.results[0] + .value as ReturnType; + expect(startTransactionSpy).toHaveBeenCalledWith('SERIALIZABLE'); + expect(queryRunner.isReleased).toBe(true); + }); + + it('should release the query runner on error', async () => { + await expect(callingService.slaveModeError()).rejects.toThrow( + new Error('Rollback'), + ); + + const queryRunner = createQueryRunnerSpy.mock.results[0] + .value as ReturnType; + expect(queryRunner.isReleased).toBe(true); + }); + + it('should not support nested transactions', async () => { + await expect(callingService.slaveModeHasNested()).rejects.toThrow( + new Error( + 'Nested transactions are not supported in slave mode', + ), + ); + }); + }); }); describe('Default options', () => { From c969528726ec84d1d29ea222de6e22ae3b624bfe Mon Sep 17 00:00:00 2001 From: James Acres Date: Fri, 30 Oct 2026 13:05:32 +0000 Subject: [PATCH 2/2] docs: improve documentation, fix link --- .../01-transactional/05-typeorm-adapter.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md b/docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md index a1f4d9ac..9fcd42ec 100644 --- a/docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md +++ b/docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md @@ -77,9 +77,9 @@ The `tx` property on the `TransactionHost` is typed ## Running on a read replica -When the `DataSource` is configured with [replication](https://typeorm.io/data-source-options#replication), you can route a whole transaction to a read replica by passing `mode: 'slave'` in the transaction options. +When the `DataSource` is configured with [replication](https://typeorm.io/docs/data-source/multiple-data-sources/#replication), you can route a whole transaction to a read replica by passing `mode: 'slave'` in the transaction options. -This is stronger than TypeORM's default behaviour: TypeORM only sends plain `find`-style reads to a replica, whereas `mode: 'slave'` runs the transaction on a replica query runner so that **every** read in the callback - including custom query-builder reads - goes to the replica. +This is needed because TypeORM only routes reads to a replica when they run on a query runner that was not already bound to the master. Inside a transaction every read is bound to the transaction's query runner, which TypeORM opens on the master by default - so reads in a `@Transactional()` method would otherwise all hit the master. Passing `mode: 'slave'` opens the transaction's query runner on a replica instead, so **every** read in the callback goes to the replica. ```ts @Injectable()