From 45a3a7a17c2cfa1f7f9b7f3ba07ff150da738466 Mon Sep 17 00:00:00 2001 From: sunlei Date: Fri, 24 Jul 2026 06:38:53 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=81=BF=E5=85=8D=E8=AE=A2=E9=98=85?= =?UTF-8?q?=E6=9B=B4=E6=96=B0=E9=94=81=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../qqbot-message-subscription.service.ts | 3 +- ...qqbot-message-subscription.service.spec.ts | 115 ++++++++++++++++++ 2 files changed, 116 insertions(+), 2 deletions(-) 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 9d93adc..9259d38 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 @@ -146,12 +146,11 @@ export class QqbotMessageSubscriptionService { const normalized = await this.normalizeInput(input); try { const saved = await this.subscriptionRepository.manager.transaction( - /** Holds both target and prospective active-key rows through the update save. */ + /** Holds the target row through save; the unique index resolves prospective-key races. */ async (manager) => { const repository = manager.getRepository(QqbotMessageSubscription); const current = await this.findActiveForWrite(repository, id); const conflict = await repository.findOne({ - lock: { mode: 'pessimistic_write' }, where: { activeKey: normalized.activeKey, isDeleted: false }, }); if (conflict && conflict.id !== current.id) { 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 234d623..937766e 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 @@ -419,6 +419,121 @@ describe('QqbotMessageSubscriptionService', () => { ).rejects.toBe(infrastructureFailure); }); + it('rejects opposite active-key updates as conflicts without taking a second write lock', async () => { + const secondConfig = { + ddnsRecordId: '2041700000000000004', + portForwardId: '2041700000000000003', + }; + const first = subscription({ id: '100' }); + const second = subscription({ + activeKey: `${SOURCE_KEY}:${digestFor(secondConfig)}`, + id: '101', + sourceConfig: secondConfig, + sourceConfigDigest: digestFor(secondConfig), + }); + const rows = [first, second]; + const bothTargetsLocked = deferred(); + let targetLocks = 0; + const repositories = ['100', '101'].map( + () => + ({ + findOne: jest.fn(async (options) => { + const where = + options.where as FindOptionsWhere; + if (where.id) { + if (options.lock?.mode === 'pessimistic_write') { + targetLocks += 1; + if (targetLocks === 2) bothTargetsLocked.resolve(); + await bothTargetsLocked.promise; + } + return rows.find((row) => row.id === where.id) ?? null; + } + if (options.lock?.mode === 'pessimistic_write') { + throw { code: 'ER_LOCK_DEADLOCK', errno: 1213 }; + } + return ( + rows.find((row) => row.activeKey === where.activeKey) ?? null + ); + }), + save: jest.fn(), + }) as unknown as jest.Mocked>, + ); + const managers = repositories.map( + (repository) => + ({ + getRepository: jest.fn((entity) => { + if (entity === QqbotMessageSubscription) return repository; + throw new Error('unexpected repository'); + }), + }) as unknown as EntityManager, + ); + const source = adapter(); + source.normalizeSubscriptionConfig.mockImplementation(async (input) => { + const config = input as typeof CONFIG; + const canonicalConfig = + config.portForwardId === secondConfig.portForwardId + ? secondConfig + : CONFIG; + return { + canonicalConfig, + resourceKey: canonicalConfig.portForwardId, + sourceSummary: '交叉更新', + }; + }); + const transaction = jest.fn((callback) => { + const manager = managers.shift(); + if (!manager) throw new Error('unexpected transaction'); + return callback(manager); + }); + const service = new QqbotMessageSubscriptionService( + { + manager: { transaction }, + } as unknown as Repository, + registry(source), + ); + + const [firstError, secondError] = await Promise.all([ + service + .update('100', { + enabled: true, + name: '改为第二条配置', + sourceConfig: secondConfig, + sourceKey: SOURCE_KEY, + }) + .catch((error: unknown) => error), + service + .update('101', { + enabled: true, + name: '改为第一条配置', + sourceConfig: CONFIG, + sourceKey: SOURCE_KEY, + }) + .catch((error: unknown) => error), + ]); + + expect(firstError).toMatchObject({ status: HttpStatus.CONFLICT }); + expect(secondError).toMatchObject({ status: HttpStatus.CONFLICT }); + for (const repository of repositories) { + expect(repository.findOne).toHaveBeenCalledWith( + expect.objectContaining({ + lock: { mode: 'pessimistic_write' }, + where: expect.objectContaining({ id: expect.any(String) }), + }), + ); + expect(repository.findOne).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ activeKey: expect.any(String) }), + }), + ); + const prospectiveRead = repository.findOne.mock.calls.find( + ([options]) => + (options.where as FindOptionsWhere) + .activeKey !== undefined, + )?.[0]; + expect(prospectiveRead?.lock).toBeUndefined(); + } + }); + it('pages only undeleted records with real filters and returns detached, real-time source status', async () => { const source = adapter({ invalidReasonCode: 'ddns_mapping_mismatch',