diff --git a/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts b/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts index 9259d38..b8c7b79 100644 --- a/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts +++ b/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts @@ -98,17 +98,15 @@ export class QqbotMessageSubscriptionService { const normalized = await this.normalizeInput(input); try { const saved = await this.subscriptionRepository.manager.transaction( - /** Serializes active-key ownership and deleted-row revival in one transaction. */ + /** Lets the unique active key arbitrate new rows while locking only a selected history row. */ async (manager) => { const repository = manager.getRepository(QqbotMessageSubscription); const active = await repository.findOne({ - lock: { mode: 'pessimistic_write' }, where: { activeKey: normalized.activeKey, isDeleted: false }, }); if (active) this.throwNaturalKeyConflict(); - const historical = await repository.findOne({ - lock: { mode: 'pessimistic_write' }, + const historicalCandidate = await repository.findOne({ order: { updateTime: 'DESC', id: 'DESC' }, where: { isDeleted: true, @@ -116,9 +114,15 @@ export class QqbotMessageSubscriptionService { sourceKey: normalized.sourceKey, }, }); - if (historical) { - Object.assign(historical, normalized, { isDeleted: false }); - return repository.save(historical); + if (historicalCandidate) { + const historical = await repository.findOne({ + lock: { mode: 'pessimistic_write' }, + where: { id: historicalCandidate.id, isDeleted: true }, + }); + if (historical) { + Object.assign(historical, normalized, { isDeleted: false }); + return repository.save(historical); + } } return repository.save( repository.create({ ...normalized, isDeleted: false }), diff --git a/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts b/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts index 937766e..b1eb474 100644 --- a/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts +++ b/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts @@ -210,6 +210,134 @@ function setup( return { items, manager, service, source, subscriptionRepository }; } +/** + * Builds two controlled create transactions that reject forbidden range locks with MySQL 1213. + * @param historical - Optional deleted candidate both transactions must contend to revive. + * @returns Shared rows, the service, and repository call evidence for concurrent-create assertions. + */ +function setupConcurrentCreateRace(historical?: QqbotMessageSubscription) { + const items = historical ? [historical] : []; + const bothActiveChecks = deferred(); + const bothHistoricalLookups = deferred(); + const contenderRequested = deferred(); + const historicalRowReleased = deferred(); + let activeChecks = 0; + let historicalLookups = 0; + let historicalLockAttempts = 0; + let createSequence = 1000; + const subscriptionRepository = { + create: jest.fn((input) => + subscription({ id: String(createSequence++), ...input }), + ), + findOne: jest.fn(async (options) => { + const where = options.where as FindOptionsWhere; + if (where.activeKey !== undefined) { + if (options.lock?.mode === 'pessimistic_write') { + throw { code: 'ER_LOCK_DEADLOCK', errno: 1213 }; + } + activeChecks += 1; + if (activeChecks === 2) bothActiveChecks.resolve(); + await bothActiveChecks.promise; + return null; + } + if ( + where.sourceConfigDigest !== undefined && + where.sourceKey !== undefined + ) { + if (options.lock?.mode === 'pessimistic_write') { + throw { code: 'ER_LOCK_DEADLOCK', errno: 1213 }; + } + historicalLookups += 1; + if (historicalLookups === 2) bothHistoricalLookups.resolve(); + await bothHistoricalLookups.promise; + return ( + items + .filter( + (item) => + item.isDeleted && + item.sourceConfigDigest === where.sourceConfigDigest && + item.sourceKey === where.sourceKey, + ) + .sort((left, right) => { + const time = String(right.updateTime).localeCompare( + String(left.updateTime), + ); + return time || right.id.localeCompare(left.id); + })[0] ?? null + ); + } + if (where.id !== undefined && where.isDeleted === true) { + if (options.lock?.mode !== 'pessimistic_write') { + throw new Error('historical candidate must be locked by primary key'); + } + historicalLockAttempts += 1; + if (historicalLockAttempts === 1) { + await contenderRequested.promise; + } else { + contenderRequested.resolve(); + await historicalRowReleased.promise; + } + return ( + items.find( + (item) => item.id === where.id && item.isDeleted === true, + ) ?? null + ); + } + throw new Error('unexpected subscription lookup'); + }), + save: jest.fn(async (item: QqbotMessageSubscription) => { + const duplicate = items.find( + (candidate) => + candidate.id !== item.id && + !candidate.isDeleted && + candidate.activeKey === item.activeKey, + ); + if (duplicate) throw { code: 'ER_DUP_ENTRY', errno: 1062 }; + const existing = items.find((candidate) => candidate.id === item.id); + if (existing) { + Object.assign(existing, item); + historicalRowReleased.resolve(); + } else { + items.push(item); + } + return item; + }), + } as unknown as jest.Mocked>; + const manager = { + getRepository: jest.fn((entity) => { + if (entity === QqbotMessageSubscription) return subscriptionRepository; + throw new Error('unexpected repository'); + }), + } as unknown as jest.Mocked; + Object.assign(subscriptionRepository, { + manager: { transaction: jest.fn((callback) => callback(manager)) }, + }); + return { + items, + service: new QqbotMessageSubscriptionService( + subscriptionRepository, + registry(adapter()), + ), + subscriptionRepository, + }; +} + +/** Runs two same-key creates and returns their success or failure values without short-circuiting. */ +async function runConcurrentCreates( + service: QqbotMessageSubscriptionService, +): Promise { + const input = { + enabled: true, + name: '并发订阅', + sourceConfig: CONFIG, + sourceKey: SOURCE_KEY, + }; + return Promise.all([ + service.create(input).catch((error: unknown) => error), + service.create(input).catch((error: unknown) => error), + ]); +} + describe('QqbotMessageSubscriptionService', () => { it('uses sorted allowlisted config JSON for the active natural key', async () => { const source = adapter(); @@ -291,6 +419,61 @@ describe('QqbotMessageSubscriptionService', () => { ).rejects.toThrow('database unavailable'); }); + it('lets the unique key settle concurrent creates for a previously unseen natural key', async () => { + const { items, service, subscriptionRepository } = + setupConcurrentCreateRace(); + + const results = await runConcurrentCreates(service); + const failures = results.filter((result) => result instanceof Error); + const successes = results.filter((result) => !(result instanceof Error)); + + expect(successes).toHaveLength(1); + expect(failures).toHaveLength(1); + expect(failures[0]).toBeInstanceOf(HttpException); + expect(failures[0]).toMatchObject({ status: HttpStatus.CONFLICT }); + expect(failures[0]).not.toMatchObject({ errno: 1213 }); + expect(items.filter((item) => !item.isDeleted)).toHaveLength(1); + for (const [options] of subscriptionRepository.findOne.mock.calls) { + const where = options.where as FindOptionsWhere; + if ( + where.activeKey !== undefined || + where.sourceConfigDigest !== undefined + ) { + expect(options.lock).toBeUndefined(); + } + } + }); + + it('locks only the selected history row before one concurrent revival wins', async () => { + const historical = subscription({ + activeKey: null, + id: '102', + isDeleted: true, + }); + const { items, service, subscriptionRepository } = + setupConcurrentCreateRace(historical); + + const results = await runConcurrentCreates(service); + const failures = results.filter((result) => result instanceof Error); + const successes = results.filter((result) => !(result instanceof Error)); + + expect(successes).toHaveLength(1); + expect(failures).toHaveLength(1); + expect(failures[0]).toBeInstanceOf(HttpException); + expect(failures[0]).toMatchObject({ status: HttpStatus.CONFLICT }); + expect(failures[0]).not.toMatchObject({ errno: 1213 }); + expect(items).toEqual([ + expect.objectContaining({ id: '102', isDeleted: false }), + ]); + const candidateLocks = subscriptionRepository.findOne.mock.calls.filter( + ([options]) => options.lock?.mode === 'pessimistic_write', + ); + expect(candidateLocks).toHaveLength(2); + for (const [options] of candidateLocks) { + expect(options.where).toEqual({ id: '102', isDeleted: true }); + } + }); + it('soft deletes safely and revives the newest matching historical row with its original ID', async () => { const older = subscription({ activeKey: null,