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 da3a55a..e643fef 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 @@ -236,6 +236,15 @@ export class SystemMessageFanoutService { }); if (!event || !this.ownsClaim(event, token)) return 'stale_claim'; + if (await this.hasStrictlyNewerEvent(manager, event)) { + await this.supersedeCurrentEventDeliveries( + manager, + event, + subscriptionId, + ); + return 'handled'; + } + const subscriptions = manager.getRepository(QqbotMessageSubscription); const subscription = await subscriptions.findOne({ where: { id: subscriptionId }, @@ -258,6 +267,64 @@ export class SystemMessageFanoutService { return 'handled'; } + /** + * Uses a locking current read to detect committed newer facts before any subscription work. + * @param manager - Current subscription transaction manager that already owns the event lock. + * @param event - Locked claimed event that must not create work after a later fact exists. + * @returns Whether an exact source/resource event is strictly newer by occurrence then BIGINT ID. + */ + private async hasStrictlyNewerEvent( + manager: EntityManager, + event: QqbotMessageEvent, + ): Promise { + const newerEvent = await manager + .getRepository(QqbotMessageEvent) + .createQueryBuilder('newerEvent') + .setLock('pessimistic_read') + .where( + 'newerEvent.sourceKey = :sourceKey AND newerEvent.resourceKey = :resourceKey', + { resourceKey: event.resourceKey, sourceKey: event.sourceKey }, + ) + .andWhere( + new Brackets((where) => { + where + .where('newerEvent.occurredAt > :occurredAt', { + occurredAt: event.occurredAt, + }) + .orWhere( + 'newerEvent.occurredAt = :occurredAt AND newerEvent.id > :eventId', + { eventId: event.id, occurredAt: event.occurredAt }, + ); + }), + ) + .take(1) + .getOne(); + return !!newerEvent; + } + + /** + * Supersedes only the current old event's mutable rows after a newer fact is committed. + * @param manager - Current subscription transaction manager. + * @param event - Older locked event whose existing unfinished rows are being retired. + * @param subscriptionId - Exact subscription scope that excludes newer and unrelated work. + */ + private async supersedeCurrentEventDeliveries( + manager: EntityManager, + event: QqbotMessageEvent, + subscriptionId: string, + ): Promise { + const deliveries = manager.getRepository(QqbotMessageDelivery); + const candidates = await deliveries.find({ + where: { messageEventId: event.id, subscriptionId }, + }); + for (const delivery of candidates) { + if (SUPERSEDED_STATUSES.has(delivery.status)) { + delivery.status = 'superseded'; + await deliveries.save(delivery); + } + } + } + /** * Supersedes only unfinished deliveries of events strictly earlier than the current event. * @param manager - Current subscription transaction manager. 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 e099264..248d23a 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 @@ -219,6 +219,12 @@ function setup(seed: Partial = {}) { predicates: [] as RecordedPredicate[], take: 0, }; + const newerEventReads: Array<{ + lock: string; + order: string[]; + predicates: RecordedPredicate[]; + take: number; + }> = []; const transactions: string[] = []; const locked = new Set(); const activeClaimLocks = new Set(); @@ -262,9 +268,16 @@ function setup(seed: Partial = {}) { return { create: (input: object) => Object.assign(new (entity as new () => object)(), input), - createQueryBuilder: () => { + createQueryBuilder: (alias = 'event') => { let builderLock = ''; let queryNow: Date | null = null; + const newerEventRead = { + lock: '', + order: [] as string[], + predicates: [] as RecordedPredicate[], + take: 0, + }; + const newerEventParameters: Record = {}; /** Records the nested TypeORM predicate structure and its bound values. */ const recordBrackets = (value: { whereFactory?: (where: { @@ -281,6 +294,9 @@ function setup(seed: Partial = {}) { if (typeof expression === 'string') { const now = (parameters as { now?: unknown }).now; if (now instanceof Date) queryNow = now; + if (alias === 'newerEvent') { + Object.assign(newerEventParameters, parameters); + } predicates.push({ expression, parameters }); return recorder; } @@ -303,10 +319,42 @@ function setup(seed: Partial = {}) { }; const builder = { addOrderBy: (field: string) => { - query.order.push(field); + if (alias === 'newerEvent') newerEventRead.order.push(field); + else query.order.push(field); return builder; }, getOne: async () => { + if (alias === 'newerEvent') { + newerEventReads.push(structuredClone(newerEventRead)); + const sourceKey = newerEventParameters.sourceKey; + const resourceKey = newerEventParameters.resourceKey; + const occurredAt = newerEventParameters.occurredAt; + const eventId = newerEventParameters.eventId; + if ( + typeof sourceKey !== 'string' || + typeof resourceKey !== 'string' || + !occurredAt || + typeof (occurredAt as { getTime?: unknown }).getTime !== + 'function' || + typeof eventId !== 'string' + ) { + return null; + } + const currentOccurredAt = ( + occurredAt as { getTime: () => number } + ).getTime(); + return ( + store.events.find((candidate) => { + return ( + candidate.sourceKey === sourceKey && + candidate.resourceKey === resourceKey && + (candidate.occurredAt.getTime() > currentOccurredAt || + (candidate.occurredAt.getTime() === currentOccurredAt && + BigInt(candidate.id) > BigInt(eventId))) + ); + }) ?? null + ); + } const claimed = rows .filter((row) => { @@ -357,11 +405,13 @@ function setup(seed: Partial = {}) { return claimed; }, orderBy: (field: string) => { - query.order.push(field); + if (alias === 'newerEvent') newerEventRead.order.push(field); + else query.order.push(field); return builder; }, setLock: (value: string) => { - query.lock = value; + if (alias === 'newerEvent') newerEventRead.lock = value; + else query.lock = value; builderLock = value; return builder; }, @@ -370,12 +420,27 @@ function setup(seed: Partial = {}) { return builder; }, take: (value: number) => { - query.take = value; + if (alias === 'newerEvent') newerEventRead.take = value; + else query.take = value; return builder; }, - where: (value: unknown) => { - query.brackets = value.constructor.name === 'Brackets'; - query.predicates.push( + where: (value: unknown, parameters?: object) => { + const predicates = + alias === 'newerEvent' + ? newerEventRead.predicates + : query.predicates; + if (typeof value === 'string') { + predicates.push({ + expression: value, + parameters: parameters ?? {}, + }); + Object.assign(newerEventParameters, parameters); + return builder; + } + if (alias !== 'newerEvent') { + query.brackets = value.constructor.name === 'Brackets'; + } + predicates.push( recordBrackets( value as { whereFactory?: (where: { @@ -387,6 +452,9 @@ function setup(seed: Partial = {}) { ); return builder; }, + andWhere: (value: unknown, parameters?: object) => { + return builder.where(value, parameters); + }, }; return builder; }, @@ -536,6 +604,7 @@ function setup(seed: Partial = {}) { failDeliveryTarget = id; }, lock: (id: string) => locked.add(id), + newerEventReads, query, savedDeliveryTargets: () => savedDeliveryTargets, pauseNextClaim: () => { @@ -695,11 +764,6 @@ describe('SystemMessageFanoutService', () => { id: '100', occurredAt: new KtDateTime(NOW.getTime() - 1), }); - const newer = event({ - id: '201', - occurredAt: new KtDateTime(NOW.getTime() + 1), - nextFanoutAt: new KtDateTime(NOW.getTime() + 1), - }); const makeDelivery = ( id: string, messageEventId: string, @@ -713,7 +777,7 @@ describe('SystemMessageFanoutService', () => { subscriptionId: '300', }); const fixture = setup({ - events: [older, event(), newer], + events: [older, event()], deliveries: [ makeDelivery('1', '100', 'pending'), makeDelivery('2', '100', 'retry'), @@ -729,7 +793,7 @@ describe('SystemMessageFanoutService', () => { .deliveries() .filter((item) => item.messageEventId === '100') .map((item) => item.status), - ).toEqual(['superseded', 'superseded', 'processing', 'superseded']); + ).toEqual(['superseded', 'superseded', 'processing']); expect( fixture.deliveries().find((item) => item.messageEventId === '201') ?.status, @@ -740,6 +804,208 @@ describe('SystemMessageFanoutService', () => { ).toBe('pending'); }); + it('does not recreate old A work after committed A-to-B-to-A events return to A', async () => { + const oldLease = new KtDateTime(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS); + const oldEvent = event({ + fanoutAttemptCount: 1, + fanoutLeaseUntil: oldLease, + fanoutStatus: 'processing', + id: '9', + occurredAt: new KtDateTime(NOW.getTime() - 2), + payload: { endpoint: 'endpoint-a', portForwardId: RESOURCE_KEY }, + }); + const middleEvent = event({ + fanoutStatus: 'completed', + id: '10', + nextFanoutAt: null, + occurredAt: new KtDateTime(NOW.getTime() - 1), + payload: { endpoint: 'endpoint-b', portForwardId: RESOURCE_KEY }, + }); + const newestEvent = event({ + fanoutStatus: 'completed', + id: '11', + nextFanoutAt: null, + payload: { endpoint: 'endpoint-a', portForwardId: RESOURCE_KEY }, + }); + const unrelatedSourceEvent = event({ + fanoutStatus: 'completed', + id: '12', + nextFanoutAt: null, + resourceKey: 'other-resource', + sourceKey: 'other.source', + }); + const oldRows = [ + 'waiting_ddns', + 'pending', + 'retry', + 'processing', + 'success', + ].map((status, index) => + Object.assign(new QqbotMessageDelivery(), { + id: `old-${index}`, + messageEventId: oldEvent.id, + publishTargetId: `old-target-${index}`, + status, + subscriptionId: '300', + }), + ); + const newestDelivery = Object.assign(new QqbotMessageDelivery(), { + id: 'newest', + messageEventId: newestEvent.id, + publishTargetId: '700', + renderedMessage: 'endpoint=endpoint-a', + status: 'pending', + subscriptionId: '300', + }); + const unrelated = Object.assign(new QqbotMessageDelivery(), { + id: 'unrelated', + messageEventId: unrelatedSourceEvent.id, + publishTargetId: 'other-target', + status: 'pending', + subscriptionId: '301', + }); + const unrelatedSourceResource = Object.assign(new QqbotMessageDelivery(), { + id: 'unrelated-source-resource', + messageEventId: unrelatedSourceEvent.id, + publishTargetId: 'other-source-resource-target', + status: 'pending', + subscriptionId: '300', + }); + const fixture = setup({ + deliveries: [ + ...oldRows, + newestDelivery, + unrelated, + unrelatedSourceResource, + ], + events: [oldEvent, middleEvent, newestEvent, unrelatedSourceEvent], + }); + const processClaim = ( + fixture.service as unknown as { + processClaim: (token: object, now: Date) => Promise; + } + ).processClaim; + const newestSnapshot = structuredClone(newestDelivery); + + await processClaim.call( + fixture.service, + { attempt: 1, event: oldEvent, leaseUntil: oldLease }, + NOW, + ); + + expect(fixture.newerEventReads).toHaveLength(1); + expect(fixture.adapter.resolveDelivery).not.toHaveBeenCalled(); + expect( + fixture + .deliveries() + .filter((item) => item.messageEventId === oldEvent.id), + ).toHaveLength(5); + expect( + fixture + .deliveries() + .filter((item) => item.messageEventId === oldEvent.id) + .map((item) => item.status), + ).toEqual([ + 'superseded', + 'superseded', + 'superseded', + 'processing', + 'success', + ]); + expect( + fixture.deliveries().find((item) => item.id === newestDelivery.id), + ).toEqual(newestSnapshot); + expect( + fixture.deliveries().find((item) => item.id === unrelated.id), + ).toEqual(unrelated); + expect( + fixture + .deliveries() + .find((item) => item.id === unrelatedSourceResource.id), + ).toEqual(unrelatedSourceResource); + expect( + fixture.events().find((item) => item.id === oldEvent.id), + ).toMatchObject({ + fanoutAttemptCount: 1, + fanoutLeaseUntil: null, + fanoutStatus: 'completed', + }); + expect(fixture.newerEventReads).toEqual([ + { + lock: 'pessimistic_read', + order: [], + predicates: [ + { + expression: + 'newerEvent.sourceKey = :sourceKey AND newerEvent.resourceKey = :resourceKey', + parameters: { sourceKey: SOURCE_KEY, resourceKey: RESOURCE_KEY }, + }, + { + brackets: [ + { + expression: 'newerEvent.occurredAt > :occurredAt', + parameters: { occurredAt: oldEvent.occurredAt }, + }, + { + expression: + 'newerEvent.occurredAt = :occurredAt AND newerEvent.id > :eventId', + parameters: { + occurredAt: oldEvent.occurredAt, + eventId: oldEvent.id, + }, + }, + ], + }, + ], + take: 1, + }, + ]); + }); + + it('treats same-timestamp BIGINT id 10 as newer than lexical-trap id 9', async () => { + const oldLease = new KtDateTime(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS); + const oldEvent = event({ + fanoutAttemptCount: 1, + fanoutLeaseUntil: oldLease, + fanoutStatus: 'processing', + id: '9', + payload: { endpoint: 'endpoint-a', portForwardId: RESOURCE_KEY }, + }); + const newestEvent = event({ + fanoutStatus: 'completed', + id: '10', + nextFanoutAt: null, + payload: { endpoint: 'endpoint-a', portForwardId: RESOURCE_KEY }, + }); + const oldDelivery = Object.assign(new QqbotMessageDelivery(), { + id: 'old', + messageEventId: oldEvent.id, + publishTargetId: 'old-target', + status: 'pending', + subscriptionId: '300', + }); + const fixture = setup({ + deliveries: [oldDelivery], + events: [oldEvent, newestEvent], + }); + const processClaim = ( + fixture.service as unknown as { + processClaim: (token: object, now: Date) => Promise; + } + ).processClaim; + + await processClaim.call( + fixture.service, + { attempt: 1, event: oldEvent, leaseUntil: oldLease }, + NOW, + ); + + expect(fixture.adapter.resolveDelivery).not.toHaveBeenCalled(); + expect(fixture.deliveries()).toEqual([ + expect.objectContaining({ id: 'old', status: 'superseded' }), + ]); + }); + it('accepts exact event-target replay and verifies only an exact duplicate-key race', async () => { const existing = Object.assign(new QqbotMessageDelivery(), { id: 'old',