From 513cf79c601bcccefa8ab8583aaed500a5c07639 Mon Sep 17 00:00:00 2001 From: sunlei Date: Fri, 24 Jul 2026 10:45:02 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=94=B6=E7=B4=A7=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E6=89=87=E5=87=BA=E7=A7=9F=E7=BA=A6=E4=B8=8E=E5=B9=B6=E5=8F=91?= =?UTF-8?q?=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../system-message-fanout.service.ts | 53 ++- .../core/qqbot-core-module-contract.spec.ts | 5 + .../system-message-fanout.service.spec.ts | 377 ++++++++++++++++-- 3 files changed, 384 insertions(+), 51 deletions(-) diff --git a/src/modules/qqbot/core/application/message-push/system-message-fanout.service.ts b/src/modules/qqbot/core/application/message-push/system-message-fanout.service.ts index 4983efb..08082ee 100644 --- a/src/modules/qqbot/core/application/message-push/system-message-fanout.service.ts +++ b/src/modules/qqbot/core/application/message-push/system-message-fanout.service.ts @@ -37,6 +37,8 @@ interface ClaimToken { leaseUntil: KtDateTime; } +type SubscriptionFanOutOutcome = 'handled' | 'stale_claim'; + /** * Creates frozen, idempotent delivery work from committed system-message Outbox facts. * @@ -141,16 +143,16 @@ export class SystemMessageFanoutService { for (const subscription of subscriptions) { try { - await this.dataSource.transaction((manager) => + const outcome = await this.dataSource.transaction((manager) => this.fanOutSubscription( manager, - token.event, + token, subscription.id, adapter, - payload, now, ), ); + if (outcome === 'stale_claim') return; } catch { transientFailure = true; } @@ -211,35 +213,45 @@ export class SystemMessageFanoutService { /** * Executes one subscription's isolated persistence unit. * @param manager - Transaction manager whose mutations commit only for this subscription. - * @param event - Frozen Outbox event currently owned by the caller. + * @param token - Exact claim token that must still own the locked Outbox row. * @param subscriptionId - Primary key of the subscription to lock and recheck. * @param adapter - Source adapter already used to validate the frozen payload. - * @param payload - Validated scalar payload passed to the adapter exactly once. * @param now - Stable scheduling instant for newly created delivery rows. */ private async fanOutSubscription( manager: EntityManager, - event: QqbotMessageEvent, + token: ClaimToken, subscriptionId: string, adapter: SystemMessageSourceAdapter, - payload: Record, now: Date, - ): Promise { + ): Promise { + const events = manager.getRepository(QqbotMessageEvent); + const event = await events.findOne({ + where: { id: token.event.id }, + lock: { mode: 'pessimistic_write' }, + }); + if (!event || !this.ownsClaim(event, token)) return 'stale_claim'; + const subscriptions = manager.getRepository(QqbotMessageSubscription); const subscription = await subscriptions.findOne({ where: { id: subscriptionId }, lock: { mode: 'pessimistic_write' }, }); - if (!subscription || !this.matchesSubscription(subscription, event)) return; + if (!subscription || !this.matchesSubscription(subscription, event)) { + return 'handled'; + } + const payload = adapter.validateEventPayload(event.payload); + this.assertResourceIdentity(event, payload); const readiness = await adapter.resolveDelivery({ eventPayload: payload, subscriptionConfig: subscription.sourceConfig, }); await this.supersedeEarlierDeliveries(manager, event, subscription.id); - if (!this.hasRenderableVariables(readiness)) return; + if (!this.hasRenderableVariables(readiness)) return 'handled'; await this.createDeliveries(manager, event, subscription, readiness, now); + return 'handled'; } /** @@ -419,7 +431,10 @@ export class SystemMessageFanoutService { await deliveries.save(delivery); } catch (error) { if (!this.isDuplicateKeyError(error)) throw error; - const existing = await deliveries.findOne({ where: key }); + const existing = await deliveries.findOne({ + where: key, + lock: { mode: 'pessimistic_read' }, + }); if (!existing) throw error; } } @@ -569,6 +584,22 @@ export class SystemMessageFanoutService { ); } + /** + * Verifies that a locked event row still belongs to the exact claim attempt. + * @param event - Event row freshly locked in the subscription transaction. + * @param token - Attempt and lease values returned by the original claim. + * @returns Whether this transaction may mutate deliveries for the claim. + */ + private ownsClaim(event: QqbotMessageEvent, token: ClaimToken): boolean { + return ( + event.id === token.event.id && + event.fanoutStatus === 'processing' && + event.fanoutAttemptCount === token.attempt && + !!event.fanoutLeaseUntil && + event.fanoutLeaseUntil.getTime() === token.leaseUntil.getTime() + ); + } + /** * Narrows the source union to outcomes that provide variables required by delivery rows. * @param readiness - Source adapter result for the locked subscription. diff --git a/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts b/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts index ac9a64d..73af88d 100644 --- a/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts +++ b/test/modules/qqbot/core/qqbot-core-module-contract.spec.ts @@ -179,6 +179,11 @@ describe('QQBot core module contract', () => { }, ]), ); + expect( + QQBOT_CORE_PROVIDERS.filter( + (provider) => provider === SystemMessageFanoutService, + ), + ).toHaveLength(1); expect(QQBOT_CORE_EXPORTS).toEqual( expect.arrayContaining([SYSTEM_MESSAGE_EVENT_STAGER]), ); diff --git a/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts b/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts index 885fa9e..1ee4398 100644 --- a/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts +++ b/test/modules/qqbot/message-push/system-message-fanout.service.spec.ts @@ -212,13 +212,27 @@ function setup(seed: Partial = {}) { lock: '', onLocked: '', order: [] as string[], + predicates: [] as Array<{ expression: string; parameters: object }>, take: 0, }; const transactions: string[] = []; const locked = new Set(); + const activeClaimLocks = new Set(); + const deliveryReadLocks: Array = []; let deliverySequence = 1000; let duplicateRace: 'exact' | 'wrong' | null = null; + const concurrentDeliveries: QqbotMessageDelivery[] = []; let failSubscription: null | string = null; + let failAfterDeliverySaves: null | number = null; + let deliverySaveCount = 0; + let pauseClaim: null | { + entered: () => void; + resume: Promise; + } = null; + let pauseEventLock: null | { + entered: () => void; + resume: Promise; + } = null; /** Returns a repository facade bound to exactly one authoritative or transaction-local store. */ const repository = (entity: unknown, store: Store) => { @@ -241,46 +255,67 @@ function setup(seed: Partial = {}) { create: (input: object) => Object.assign(new (entity as new () => object)(), input), createQueryBuilder: () => { + let builderLock = ''; const builder = { addOrderBy: (field: string) => { query.order.push(field); return builder; }, - getOne: async () => - rows - .filter((row) => { - const item = row as unknown as QqbotMessageEvent; - return ( - ((item.fanoutStatus === 'accepted' || - item.fanoutStatus === 'retry') && - !!item.nextFanoutAt && - item.nextFanoutAt.getTime() <= NOW.getTime()) || - (item.fanoutStatus === 'processing' && - !!item.fanoutLeaseUntil && - item.fanoutLeaseUntil.getTime() <= NOW.getTime()) - ); - }) - .filter( - (row) => !locked.has((row as unknown as QqbotMessageEvent).id), - ) - .sort((left, right) => { - const a = left as unknown as QqbotMessageEvent; - const b = right as unknown as QqbotMessageEvent; - return ( - a.occurredAt.getTime() - b.occurredAt.getTime() || - (BigInt(a.id) < BigInt(b.id) - ? -1 - : BigInt(a.id) > BigInt(b.id) - ? 1 - : 0) - ); - })[0] ?? null, + getOne: async () => { + const claimed = + rows + .filter((row) => { + const item = row as unknown as QqbotMessageEvent; + return ( + ((item.fanoutStatus === 'accepted' || + item.fanoutStatus === 'retry') && + !!item.nextFanoutAt && + item.nextFanoutAt.getTime() <= NOW.getTime()) || + (item.fanoutStatus === 'processing' && + !!item.fanoutLeaseUntil && + item.fanoutLeaseUntil.getTime() <= NOW.getTime()) + ); + }) + .filter( + (row) => + !locked.has((row as unknown as QqbotMessageEvent).id) && + !activeClaimLocks.has( + (row as unknown as QqbotMessageEvent).id, + ), + ) + .sort((left, right) => { + const a = left as unknown as QqbotMessageEvent; + const b = right as unknown as QqbotMessageEvent; + return ( + a.occurredAt.getTime() - b.occurredAt.getTime() || + (BigInt(a.id) < BigInt(b.id) + ? -1 + : BigInt(a.id) > BigInt(b.id) + ? 1 + : 0) + ); + })[0] ?? null; + if ( + !claimed || + !pauseClaim || + builderLock !== 'pessimistic_write' + ) { + return claimed; + } + const id = (claimed as unknown as QqbotMessageEvent).id; + activeClaimLocks.add(id); + pauseClaim.entered(); + await pauseClaim.resume; + activeClaimLocks.delete(id); + return claimed; + }, orderBy: (field: string) => { query.order.push(field); return builder; }, setLock: (value: string) => { query.lock = value; + builderLock = value; return builder; }, setOnLocked: (value: string) => { @@ -293,6 +328,23 @@ function setup(seed: Partial = {}) { }, where: (value: unknown) => { query.brackets = value.constructor.name === 'Brackets'; + const brackets = value as { + whereFactory?: (where: { + orWhere: (expression: string, parameters: object) => void; + where: (expression: string, parameters: object) => void; + }) => void; + }; + const expressionRecorder = { + orWhere: (expression: string, parameters: object) => { + query.predicates.push({ expression, parameters }); + return expressionRecorder; + }, + where: (expression: string, parameters: object) => { + query.predicates.push({ expression, parameters }); + return expressionRecorder; + }, + }; + brackets.whereFactory?.(expressionRecorder); return builder; }, }; @@ -311,8 +363,38 @@ function setup(seed: Partial = {}) { ) : result; }, - findOne: async ({ where }: { where: Record }) => - rows.find((row) => matches(row, where)) ?? null, + findOne: async ({ + lock, + where, + }: { + lock?: { mode: string }; + where: Record; + }) => { + if ( + entity === QqbotMessageEvent && + lock?.mode === 'pessimistic_write' && + pauseEventLock + ) { + const event = rows.find((row) => matches(row, where)); + if (!event) return null; + const id = (event as unknown as QqbotMessageEvent).id; + activeClaimLocks.add(id); + pauseEventLock.entered(); + await pauseEventLock.resume; + activeClaimLocks.delete(id); + return event; + } + if (entity === QqbotMessageDelivery) { + deliveryReadLocks.push(lock?.mode ?? null); + const visible = lock ? [...rows, ...concurrentDeliveries] : rows; + return ( + visible.find((row) => + matches(row as Record, where), + ) ?? null + ); + } + return rows.find((row) => matches(row, where)) ?? null; + }, save: async (item: Record) => { if ( entity === QqbotMessageDelivery && @@ -320,6 +402,7 @@ function setup(seed: Partial = {}) { ) throw new Error('repository offline'); if (entity === QqbotMessageDelivery) { + deliverySaveCount += 1; const pair = rows.find((row) => matches(row, { messageEventId: item.messageEventId, @@ -327,16 +410,23 @@ function setup(seed: Partial = {}) { }), ); if (!item.id && duplicateRace) { - if (duplicateRace === 'exact') - rows.push( + if (duplicateRace === 'exact') { + concurrentDeliveries.push( Object.assign(new QqbotMessageDelivery(), item, { id: 'race', - }) as unknown as Record, + }), ); + } duplicateRace = null; throw { errno: 1062 }; } if (pair && pair !== item) throw { errno: 1062 }; + if ( + failAfterDeliverySaves !== null && + deliverySaveCount > failAfterDeliverySaves + ) { + throw new Error('repository offline after delivery mutation'); + } if (!item.id) item.id = `${deliverySequence++}`; } const index = rows.findIndex( @@ -370,6 +460,11 @@ function setup(seed: Partial = {}) { const result = await callback({ getRepository: (entity) => repository(entity, draft), }); + for (const delivery of concurrentDeliveries) { + if (!draft.deliveries.some((item) => item.id === delivery.id)) { + draft.deliveries.push(structuredClone(delivery)); + } + } state = draft; transactions.push('commit'); return result; @@ -386,17 +481,48 @@ function setup(seed: Partial = {}) { ); return { adapter, + bindings: () => state.bindings, events: () => state.events, deliveries: () => state.deliveries, + deliveryReadLocks, failSubscription: (id: null | string) => { failSubscription = id; }, + failAfterDeliverySaves: (count: null | number) => { + failAfterDeliverySaves = count; + }, lock: (id: string) => locked.add(id), query, + pauseNextClaim: () => { + let entered!: () => void; + const reached = new Promise((resolve) => { + entered = resolve; + }); + let resume!: () => void; + const wait = new Promise((resolve) => { + resume = resolve; + }); + pauseClaim = { entered, resume: wait }; + return { reached, resume }; + }, + pauseNextEventLock: () => { + let entered!: () => void; + const reached = new Promise((resolve) => { + entered = resolve; + }); + let resume!: () => void; + const wait = new Promise((resolve) => { + resume = resolve; + }); + pauseEventLock = { entered, resume: wait }; + return { reached, resume }; + }, setDuplicateRace: (value: 'exact' | 'wrong' | null) => { duplicateRace = value; }, service, + targets: () => state.targets, + templates: () => state.templates, transactions, }; } @@ -584,6 +710,7 @@ describe('SystemMessageFanoutService', () => { race.setDuplicateRace('exact'); await race.service.runOnce(NOW); expect(race.deliveries()).toHaveLength(1); + expect(race.deliveryReadLocks).toEqual([null, 'pessimistic_read']); const wrong = setup(); wrong.setDuplicateRace('wrong'); @@ -604,6 +731,47 @@ describe('SystemMessageFanoutService', () => { take: 1, }); expect(fixture.query.order).toEqual(['event.occurredAt', 'event.id']); + expect(fixture.query.predicates).toEqual([ + { + expression: + 'event.fanoutStatus IN (:...due) AND event.nextFanoutAt <= :now', + parameters: { due: ['accepted', 'retry'], now: NOW }, + }, + { + expression: + 'event.fanoutStatus = :processing AND event.fanoutLeaseUntil <= :now', + parameters: { processing: 'processing', now: NOW }, + }, + ]); + }); + + it('gives overlapping claim transactions one owner only through write lock and skip-locked selection', async () => { + const fixture = setup(); + const gate = fixture.pauseNextClaim(); + const first = fixture.service.runOnce(NOW); + await gate.reached; + await expect(fixture.service.runOnce(NOW)).resolves.toBe(0); + gate.resume(); + await expect(first).resolves.toBe(1); + expect(fixture.events()[0].fanoutAttemptCount).toBe(1); + }); + + it('lets the current event-lock holder finish atomically while an expired-lease reclaim waits', async () => { + const fixture = setup(); + const gate = fixture.pauseNextEventLock(); + const currentOwner = fixture.service.runOnce(NOW); + await gate.reached; + + await expect( + fixture.service.runOnce( + new Date(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS), + ), + ).resolves.toBe(0); + + gate.resume(); + await expect(currentOwner).resolves.toBe(1); + expect(fixture.events()[0].fanoutStatus).toBe('completed'); + expect(fixture.deliveries()).toHaveLength(1); }); it('skips a locked oldest due event and reclaims expired processing with an exact new lease', async () => { @@ -632,15 +800,20 @@ describe('SystemMessageFanoutService', () => { }); }); - it('uses a lease-and-attempt ownership fence so stale completion cannot clobber a new owner', async () => { + it('uses independent attempt and lease ownership fences so stale completion cannot clobber a new owner', async () => { const fixture = setup({ events: [] }); const staleEvent = event({ fanoutAttemptCount: 1, fanoutLeaseUntil: new KtDateTime(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS), fanoutStatus: 'processing', }); - const current = event({ + const attemptMismatch = event({ fanoutAttemptCount: 2, + fanoutLeaseUntil: staleEvent.fanoutLeaseUntil, + fanoutStatus: 'processing', + }); + const leaseMismatch = event({ + fanoutAttemptCount: 1, fanoutLeaseUntil: new KtDateTime( NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS + 1, ), @@ -658,7 +831,7 @@ describe('SystemMessageFanoutService', () => { ).finish; (fixture as unknown as { events: () => QqbotMessageEvent[] }) .events() - .push(current); + .push(attemptMismatch); await finish.call( fixture.service, @@ -672,6 +845,73 @@ describe('SystemMessageFanoutService', () => { null, ); expect(fixture.events()[0].fanoutStatus).toBe('processing'); + + fixture.events().splice(0, 1, leaseMismatch); + await finish.call( + fixture.service, + { + attempt: 1, + event: staleEvent, + leaseUntil: staleEvent.fanoutLeaseUntil, + }, + 'completed', + null, + null, + ); + expect(fixture.events()[0].fanoutStatus).toBe('processing'); + }); + + it('stops a stale owner before supersession or inserts after a newer owner completes', async () => { + const staleEvent = event({ + fanoutAttemptCount: 1, + fanoutLeaseUntil: new KtDateTime(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS), + fanoutStatus: 'processing', + }); + const winningEvent = event({ + fanoutAttemptCount: 2, + fanoutLeaseUntil: null, + fanoutStatus: 'completed', + }); + const prior = Object.assign(new QqbotMessageDelivery(), { + id: 'prior', + messageEventId: '100', + publishTargetId: 'old-target', + status: 'pending', + subscriptionId: '300', + }); + const fixture = setup({ + deliveries: [prior], + events: [ + winningEvent, + event({ + id: '100', + occurredAt: new KtDateTime(NOW.getTime() - 1), + nextFanoutAt: new KtDateTime(NOW.getTime() + 1), + }), + ], + }); + const processClaim = ( + fixture.service as unknown as { + processClaim: (token: object, now: Date) => Promise; + } + ).processClaim; + + await processClaim.call( + fixture.service, + { + attempt: 1, + event: staleEvent, + leaseUntil: staleEvent.fanoutLeaseUntil, + }, + NOW, + ); + + expect(fixture.deliveries()).toEqual([prior]); + expect(fixture.events()[0]).toMatchObject({ + fanoutAttemptCount: 2, + fanoutStatus: 'completed', + }); + expect(fixture.adapter.resolveDelivery).not.toHaveBeenCalled(); }); it('rolls back one failing subscription, retains valid work, then retries and recovers idempotently', async () => { @@ -701,6 +941,44 @@ describe('SystemMessageFanoutService', () => { expect(fixture.transactions).toContain('rollback'); }); + it('rolls back supersession and an earlier target insert when a subscription later fails', async () => { + const older = event({ + id: '100', + occurredAt: new KtDateTime(NOW.getTime() - 1), + nextFanoutAt: new KtDateTime(NOW.getTime() + 1), + }); + const priorFailed = Object.assign(new QqbotMessageDelivery(), { + id: 'old-failed', + messageEventId: '100', + publishTargetId: 'old-failed-target', + status: 'pending', + subscriptionId: '301', + }); + const fixture = setup({ + deliveries: [priorFailed], + events: [older, event()], + subscriptions: [subscription(), subscription({ id: '301' })], + bindings: [binding(), binding({ id: '501', subscriptionId: '301' })], + targets: [ + target(), + target({ bindingId: '501', id: '701' }), + target({ bindingId: '501', id: '702', targetId: 'group-3' }), + ], + }); + fixture.failAfterDeliverySaves(2); + + await fixture.service.runOnce(NOW); + + expect( + fixture.deliveries().find((item) => item.id === 'old-failed'), + ).toMatchObject({ status: 'pending' }); + expect(fixture.deliveries().map((item) => item.publishTargetId)).toEqual([ + 'old-failed-target', + '700', + ]); + expect(fixture.events()[1].fanoutStatus).toBe('retry'); + }); + it('fails at the occurrence deadline and permanently rejects malformed source facts', async () => { const deadline = setup({ events: [ @@ -946,15 +1224,34 @@ describe('SystemMessageFanoutService', () => { ).toEqual(statuses); }); - it('preserves snapshots after later in-memory configuration edits', async () => { + it('preserves every delivery snapshot after source configuration rows later change', async () => { const fixture = setup(); await fixture.service.runOnce(NOW); - const frozen = fixture.deliveries()[0]; - frozen.templateContent = 'endpoint=${{endpoint}}'; + const frozen = structuredClone(fixture.deliveries()[0]); + fixture.templates()[0].content = 'changed=${{endpoint}}'; + fixture.bindings()[0].selfId = 'changed-bot'; + fixture.targets()[0].targetId = 'changed-target'; expect(frozen).toMatchObject({ + attemptCount: 0, + bindingId: '500', + lastErrorCode: null, + lastErrorMessage: null, + messageEventId: '200', + processingLeaseUntil: null, renderedMessage: 'endpoint=pal.example.com:38213', + sendLogId: null, selfId: 'bot-a', + status: 'pending', + subscriptionId: '300', + templateContent: 'endpoint=${{endpoint}}', + templateId: '600', targetId: 'group-1', + targetType: 'group', + variableSnapshot: { endpoint: 'pal.example.com:38213' }, }); + expect(frozen.nextAttemptAt.getTime()).toBe(NOW.getTime()); + expect(frozen.expiresAt.getTime()).toBe( + NOW.getTime() + SYSTEM_MESSAGE_RETRY_WINDOW_MS, + ); }); });