diff --git a/src/modules/aggregator/aggregator.data.handler.spec.ts b/src/modules/aggregator/aggregator.data.handler.spec.ts new file mode 100644 index 00000000..9fcbd19b --- /dev/null +++ b/src/modules/aggregator/aggregator.data.handler.spec.ts @@ -0,0 +1,74 @@ +import AggregatorDataHandler from './aggregator.data.handler'; +import {UpsertError} from '../../services/mongo/upsert.error'; + +const contextFor = (redelivered: boolean): {channel: {ack: jest.Mock, nack: jest.Mock}, message: unknown, context: never} => { + const channel = {ack: jest.fn(), nack: jest.fn()}; + const message = {fields: {redelivered}}; + + return {channel, message, context: {getChannelRef: () => channel, getMessage: () => message} as never}; +}; + +describe('AggregatorDataHandler.upsert', () => { + const data = {docs: [{$inc: {kills: 1}}], conditionals: [{id: 1}]} as never; + + it('acks once the write succeeds', async () => { + const handler = new AggregatorDataHandler({upsert: jest.fn().mockResolvedValue(true)} as never); + const {channel, message, context} = contextFor(false); + + await handler.upsert(data, context, 'Entity'); + + expect(channel.ack).toHaveBeenCalledWith(message); + expect(channel.nack).not.toHaveBeenCalled(); + }); + + it('requeues a write that provably did nothing, the first time', async () => { + const handler = new AggregatorDataHandler({upsert: jest.fn().mockRejectedValue(new UpsertError('down', true))} as never); + const {channel, message, context} = contextFor(false); + + await handler.upsert(data, context, 'Entity'); + + expect(channel.nack).toHaveBeenCalledWith(message, false, true); + expect(channel.ack).not.toHaveBeenCalled(); + }); + + it('drops a write that fails again on redelivery, so a bad message cannot loop', async () => { + const handler = new AggregatorDataHandler({upsert: jest.fn().mockRejectedValue(new UpsertError('down', true))} as never); + const {channel, message, context} = contextFor(true); + + await handler.upsert(data, context, 'Entity'); + + expect(channel.ack).toHaveBeenCalledWith(message); + expect(channel.nack).not.toHaveBeenCalled(); + }); + + it('does not replay a write that may have half-applied, since that would double count', async () => { + const handler = new AggregatorDataHandler({upsert: jest.fn().mockRejectedValue(new UpsertError('timeout', false))} as never); + const {channel, message, context} = contextFor(false); + + await handler.upsert(data, context, 'Entity'); + + expect(channel.ack).toHaveBeenCalledWith(message); + expect(channel.nack).not.toHaveBeenCalled(); + }); +}); + +describe('AggregatorDataHandler.settleGlobalFailure', () => { + const handler = new AggregatorDataHandler({} as never); + + it('drops a message whose alert was deleted', async () => { + const {channel, message, context} = contextFor(false); + + await handler.settleGlobalFailure(context, new Error('Instance 10-1 does not exist.'), 'pattern', '10-1'); + + expect(channel.ack).toHaveBeenCalledWith(message); + }); + + it('requeues a message whose alert lookup failed, since nothing was written yet', async () => { + const {channel, message, context} = contextFor(false); + + await handler.settleGlobalFailure(context, new Error('connection reset'), 'pattern', '10-1'); + + expect(channel.nack).toHaveBeenCalledWith(message, false, true); + expect(channel.ack).not.toHaveBeenCalled(); + }); +}); diff --git a/src/modules/aggregator/aggregator.data.handler.ts b/src/modules/aggregator/aggregator.data.handler.ts index 6c6fee65..fab434f9 100644 --- a/src/modules/aggregator/aggregator.data.handler.ts +++ b/src/modules/aggregator/aggregator.data.handler.ts @@ -3,6 +3,7 @@ import AggregatorMessageInterface from '../../modules/aggregator/interfaces/aggr import {RmqContext} from '@nestjs/microservices'; import {Inject, Injectable, Logger} from '@nestjs/common'; import MongoOperationsService from '../../services/mongo/mongo.operations.service'; +import {UpsertError} from '../../services/mongo/upsert.error'; import GlobalAggregatorMessageInterface from './interfaces/global.aggregator.message.interface'; import InstanceMetagameTerritoryEntity from '../data/entities/instance/instance.metagame.territory.entity'; import {Ps2AlertsEventState} from '../data/ps2alerts-constants/ps2AlertsEventState'; @@ -22,12 +23,10 @@ export default class AggregatorDataHandler { entity, data.docs, ); - } catch (err: any) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access,@typescript-eslint/no-unsafe-call - if (err.message && !err.message.includes('E11000')) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access,@typescript-eslint/restrict-template-expressions - this.logger.error(`Unable to create data for Aggregation! E: ${err.message}`); - } + } catch (err) { + // Every collection written here has a unique index, so a replayed insert cannot duplicate rows + this.retryOnce(context, `Unable to create data for Aggregation! E: ${err instanceof Error ? err.message : String(err)}`); + return; } // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access @@ -41,12 +40,9 @@ export default class AggregatorDataHandler { data.docs, data.conditionals, ); - } catch (err: any) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - if (err.message && !err.message.includes('E11000')) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access,@typescript-eslint/restrict-template-expressions - this.logger.error(`Unable to upsert data for Aggregation! E: ${err.message}`); - } + } catch (err) { + this.settleFailedUpsert(context, err, 'Unable to upsert data for Aggregation!'); + return; } // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access @@ -70,12 +66,9 @@ export default class AggregatorDataHandler { data.docs, data.conditionals, ); - } catch (err: any) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access,@typescript-eslint/no-unsafe-call - if (err.message && !err.message.includes('E11000')) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access,@typescript-eslint/restrict-template-expressions - throw new Error(`Unable to upsert data for Global Aggregation! E: ${err.message}`); - } + } catch (err) { + this.settleFailedUpsert(context, err, `Unable to upsert data for Global Aggregation of instance ${data.instance}!`); + return; } // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access @@ -105,7 +98,11 @@ export default class AggregatorDataHandler { {instanceId: data.instance}, ); } catch (e) { - throw new Error(`Instance ${data.instance} does not exist.`); + if (e instanceof Error && e.name === 'EntityNotFoundError') { + throw new Error(`Instance ${data.instance} does not exist.`); + } + + throw e; } if (instance.state === Ps2AlertsEventState.ENDED) { @@ -124,6 +121,24 @@ export default class AggregatorDataHandler { return data; } + // Called by global controllers when the message failed before anything was written + public async settleGlobalFailure(context: RmqContext, err: unknown, pattern: string, instance: string): Promise { + const reason = err instanceof Error ? err.message : String(err); + + // A deleted alert, or a bracket message for an alert not yet ended, cannot succeed on a retry + if (reason.includes('does not exist') || reason.includes('unfinished instance')) { + if (!reason.includes('does not exist')) { + this.logger.error(`Unable to process ${pattern} message for instance ${instance}! Error: ${reason}`); + } + + // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access + await context.getChannelRef().ack(context.getMessage()); + return; + } + + this.retryOnce(context, `Unable to process ${pattern} message for instance ${instance}! Error: ${reason}`); + } + private transformDateConditional(conditional: any): any { // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access if (conditional.date) { @@ -137,4 +152,37 @@ export default class AggregatorDataHandler { // eslint-disable-next-line @typescript-eslint/no-unsafe-return return conditional; } + + // A replay after a partial write would count its increments twice, so only a write that provably did nothing is retried + private settleFailedUpsert(context: RmqContext, err: unknown, reason: string): void { + const message = `${reason} E: ${err instanceof Error ? err.message : String(err)}`; + + if (err instanceof UpsertError && err.nothingWritten) { + this.retryOnce(context, message); + return; + } + + this.logger.error(`${message}. Part of it may have been written, so it is not retried.`); + // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access + context.getChannelRef().ack(context.getMessage()); + } + + // A failed write is requeued once, since most failures are a Mongo blip; a second failure is logged and dropped + private retryOnce(context: RmqContext, reason: string): void { + const message = context.getMessage(); + // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment + const channel = context.getChannelRef(); + + // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access + if (!message.fields.redelivered) { + this.logger.warn(`${reason}. Requeueing once.`); + // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access + channel.nack(message, false, true); + return; + } + + this.logger.error(`${reason}. Failed on redelivery too, dropping the message.`); + // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access + channel.ack(message); + } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.character.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.character.aggregate.controller.ts index 783f7d81..7386c321 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.character.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.character.aggregate.controller.ts @@ -20,12 +20,7 @@ export default class AggregatorGlobalCharacterAggregateController { GlobalCharacterAggregateEntity, ); } catch (e) { - if (e instanceof Error && !e.message.includes('does not exist')) { - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_CHARACTER_AGGREGATE} message for instance ${data.instance}! Error: ${e.message}`); - } - - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + await this.aggregatorDataHandler.settleGlobalFailure(context, e, MqAcceptedPatterns.GLOBAL_CHARACTER_AGGREGATE, data.instance); } } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.facility.control.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.facility.control.aggregate.controller.ts index b3d8819d..0113cd3f 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.facility.control.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.facility.control.aggregate.controller.ts @@ -20,12 +20,7 @@ export default class AggregatorGlobalFacilityControlAggregateController { GlobalFacilityControlAggregateEntity, ); } catch (e) { - if (e instanceof Error && !e.message.includes('does not exist')) { - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_FACILITY_CONTROL_AGGREGATE} message for instance ${data.instance}! Error: ${e.message}`); - } - - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + await this.aggregatorDataHandler.settleGlobalFailure(context, e, MqAcceptedPatterns.GLOBAL_FACILITY_CONTROL_AGGREGATE, data.instance); } } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.faction.combat.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.faction.combat.aggregate.controller.ts index 45e76f75..fded2e7c 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.faction.combat.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.faction.combat.aggregate.controller.ts @@ -20,12 +20,7 @@ export default class AggregatorGlobalFactionCombatAggregateController { GlobalFactionCombatAggregateEntity, ); } catch (e) { - if (e instanceof Error && !e.message.includes('does not exist')) { - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_FACTION_COMBAT_AGGREGATE} message for instance ${data.instance}! Error: ${e.message}`); - } - - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + await this.aggregatorDataHandler.settleGlobalFailure(context, e, MqAcceptedPatterns.GLOBAL_FACTION_COMBAT_AGGREGATE, data.instance); } } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.loadout.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.loadout.aggregate.controller.ts index 9e6df619..a4fabf8e 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.loadout.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.loadout.aggregate.controller.ts @@ -20,12 +20,7 @@ export default class AggregatorGlobalLoadoutAggregateController { GlobalLoadoutAggregateEntity, ); } catch (e) { - if (e instanceof Error && !e.message.includes('does not exist')) { - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_LOADOUT_AGGREGATE} message for instance ${data.instance}! Error: ${e.message}`); - } - - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + await this.aggregatorDataHandler.settleGlobalFailure(context, e, MqAcceptedPatterns.GLOBAL_LOADOUT_AGGREGATE, data.instance); } } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.outfit.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.outfit.aggregate.controller.ts index 190b6e67..387abc54 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.outfit.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.outfit.aggregate.controller.ts @@ -20,12 +20,7 @@ export default class AggregatorGlobalOutfitAggregateController { GlobalOutfitAggregateEntity, ); } catch (e) { - if (e instanceof Error && !e.message.includes('does not exist')) { - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_OUTFIT_AGGREGATE} message for instance ${data.instance}! Error: ${e.message}`); - } - - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + await this.aggregatorDataHandler.settleGlobalFailure(context, e, MqAcceptedPatterns.GLOBAL_OUTFIT_AGGREGATE, data.instance); } } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.vehicle.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.vehicle.aggregate.controller.ts index d4ab9aa7..bf1528e4 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.vehicle.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.vehicle.aggregate.controller.ts @@ -20,12 +20,7 @@ export default class AggregatorGlobalVehicleAggregateController { GlobalVehicleAggregateEntity, ); } catch (e) { - if (e instanceof Error && !e.message.includes('does not exist')) { - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_VEHICLE_AGGREGATE} message for instance ${data.instance}! Error: ${e.message}`); - } - - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + await this.aggregatorDataHandler.settleGlobalFailure(context, e, MqAcceptedPatterns.GLOBAL_VEHICLE_AGGREGATE, data.instance); } } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.vehicle.character.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.vehicle.character.aggregate.controller.ts index f63987a5..217e8a2c 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.vehicle.character.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.vehicle.character.aggregate.controller.ts @@ -20,12 +20,7 @@ export default class AggregatorGlobalVehicleCharacterAggregateController { GlobalVehicleCharacterAggregateEntity, ); } catch (e) { - if (e instanceof Error && !e.message.includes('does not exist')) { - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_VEHICLE_CHARACTER_AGGREGATE} message for instance ${data.instance}! Error: ${e.message}`); - } - - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + await this.aggregatorDataHandler.settleGlobalFailure(context, e, MqAcceptedPatterns.GLOBAL_VEHICLE_CHARACTER_AGGREGATE, data.instance); } } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.victory.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.victory.aggregate.controller.ts index bdd9f757..807a22cf 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.victory.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.victory.aggregate.controller.ts @@ -23,14 +23,16 @@ export default class AggregatorGlobalVictoryAggregateController { context, GlobalVictoryAggregateEntity, ); - await this.cacheService.set(GLOBAL_VICTORIES_GENERATION_KEY, Date.now(), 2592000); - // eslint-disable-next-line @typescript-eslint/no-explicit-any - } catch (err: any) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access,@typescript-eslint/restrict-template-expressions - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_VICTORY_AGGREGATE} message for instance ${data.instance}! Error: ${err.message}`); + } catch (err) { + await this.aggregatorDataHandler.settleGlobalFailure(context, err, MqAcceptedPatterns.GLOBAL_VICTORY_AGGREGATE, data.instance); + return; + } - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + // Outside the try: the message is already settled, so a cache failure must not ack it a second time + try { + await this.cacheService.set(GLOBAL_VICTORIES_GENERATION_KEY, Date.now(), 2592000); + } catch (err) { + this.logger.error(`Could not bump the victories cache generation: ${String(err)}`); } } } diff --git a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.weapon.aggregate.controller.ts b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.weapon.aggregate.controller.ts index baa288e3..f60569f3 100644 --- a/src/modules/aggregator/controllers/aggregates/global/aggregator.global.weapon.aggregate.controller.ts +++ b/src/modules/aggregator/controllers/aggregates/global/aggregator.global.weapon.aggregate.controller.ts @@ -20,12 +20,7 @@ export default class AggregatorGlobalWeaponAggregateController { GlobalWeaponAggregateEntity, ); } catch (e) { - if (e instanceof Error && !e.message.includes('does not exist')) { - this.logger.error(`Unable to process ${MqAcceptedPatterns.GLOBAL_WEAPON_AGGREGATE} message for instance ${data.instance}! Error: ${e.message}`); - } - - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - await context.getChannelRef().ack(context.getMessage()); + await this.aggregatorDataHandler.settleGlobalFailure(context, e, MqAcceptedPatterns.GLOBAL_WEAPON_AGGREGATE, data.instance); } } } diff --git a/src/services/mongo/mongo.operations.service.spec.ts b/src/services/mongo/mongo.operations.service.spec.ts new file mode 100644 index 00000000..feaab289 --- /dev/null +++ b/src/services/mongo/mongo.operations.service.spec.ts @@ -0,0 +1,77 @@ +/* eslint-disable @typescript-eslint/no-unsafe-member-access */ +import MongoOperationsService from './mongo.operations.service'; +import {UpsertError} from './upsert.error'; + +const duplicateAt = (index: number): Error => Object.assign(new Error('E11000 duplicate key error'), {writeErrors: [{code: 11000, index}]}); +const ok = {upsertedCount: 0, modifiedCount: 1}; + +describe('MongoOperationsService.upsert', () => { + const docs = (): Array> => [{$setOnInsert: {a: 1}}, {$inc: {kills: 1}}, {$set: {name: 'x'}}]; + + it('writes every doc in order in one bulk write', async () => { + const bulkWrite = jest.fn().mockResolvedValue(ok); + const service = new MongoOperationsService({bulkWrite} as never); + + await expect(service.upsert('Entity', docs(), [{id: 1}])).resolves.toBe(true); + expect(bulkWrite).toHaveBeenCalledTimes(1); + expect(bulkWrite.mock.calls[0][1]).toHaveLength(3); + expect(bulkWrite.mock.calls[0][1][0]).toEqual({updateOne: {filter: {id: 1}, update: {$setOnInsert: {a: 1}}, upsert: true}}); + }); + + it('retries from the failed doc after a duplicate-key race, keeping the docs after it', async () => { + const bulkWrite = jest.fn().mockRejectedValueOnce(duplicateAt(0)).mockResolvedValue(ok); + const service = new MongoOperationsService({bulkWrite} as never); + + await service.upsert('Entity', docs(), [{id: 1}]); + + expect(bulkWrite).toHaveBeenCalledTimes(2); + expect(bulkWrite.mock.calls[1][1]).toHaveLength(3); + }); + + it('resumes at the failing doc rather than the start of the message', async () => { + const bulkWrite = jest.fn().mockRejectedValueOnce(duplicateAt(1)).mockResolvedValue(ok); + const service = new MongoOperationsService({bulkWrite} as never); + + await service.upsert('Entity', docs(), [{id: 1}]); + + expect(bulkWrite.mock.calls[1][1]).toHaveLength(2); + expect(bulkWrite.mock.calls[1][1][0].updateOne.update).toEqual({$inc: {kills: 1}}); + }); + + it('throws when the same doc hits a duplicate key twice', async () => { + const bulkWrite = jest.fn().mockRejectedValue(duplicateAt(0)); + const service = new MongoOperationsService({bulkWrite} as never); + + await expect(service.upsert('Entity', docs(), [{id: 1}])).rejects.toThrow('Upsert failed'); + expect(bulkWrite).toHaveBeenCalledTimes(2); + }); + + it('throws any other error, marked as possibly written when it could have half-applied', async () => { + const bulkWrite = jest.fn().mockRejectedValue(new Error('connection reset')); + const service = new MongoOperationsService({bulkWrite} as never); + + const error = await service.upsert('Entity', docs(), [{id: 1}]).catch((e: UpsertError) => e); + + expect(error).toBeInstanceOf(UpsertError); + expect((error as UpsertError).nothingWritten).toBe(false); + expect(bulkWrite).toHaveBeenCalledTimes(1); + }); + + it('marks a failure on the very first operation as nothing written', async () => { + const failedFirst = Object.assign(new Error('validation'), {writeErrors: [{code: 121, index: 0}]}); + const service = new MongoOperationsService({bulkWrite: jest.fn().mockRejectedValue(failedFirst)} as never); + + const error = await service.upsert('Entity', docs(), [{id: 1}]).catch((e: UpsertError) => e); + + expect((error as UpsertError).nothingWritten).toBe(true); + }); + + it('marks a failure after earlier operations applied as possibly written', async () => { + const failedLater = Object.assign(new Error('validation'), {writeErrors: [{code: 121, index: 2}]}); + const service = new MongoOperationsService({bulkWrite: jest.fn().mockRejectedValue(failedLater)} as never); + + const error = await service.upsert('Entity', docs(), [{id: 1}]).catch((e: UpsertError) => e); + + expect((error as UpsertError).nothingWritten).toBe(false); + }); +}); diff --git a/src/services/mongo/mongo.operations.service.ts b/src/services/mongo/mongo.operations.service.ts index 9c75d3ec..4cb1ceaa 100644 --- a/src/services/mongo/mongo.operations.service.ts +++ b/src/services/mongo/mongo.operations.service.ts @@ -4,6 +4,7 @@ import {AggregateOptions} from 'typeorm/driver/mongodb/typings'; import {InjectEntityManager} from '@nestjs/typeorm'; import {Injectable} from '@nestjs/common'; import Pagination from './pagination'; +import {UpsertError} from './upsert.error'; @Injectable() export default class MongoOperationsService { @@ -64,7 +65,8 @@ export default class MongoOperationsService { docs = this.transform(docs); try { - const result = await this.em.insertMany(entity, docs); + // Unordered, so on a replay the rows already inserted do not stop the rest + const result = await this.em.insertMany(entity, docs, {ordered: false}); return Object.values(result.insertedIds); } catch (error: any) { @@ -79,38 +81,10 @@ export default class MongoOperationsService { } public async upsert(entity: any, docs: any[], conditionals: any[]): Promise { - docs = this.transform(docs); - const operations: any[] = []; - - // Gather operations, setOnInserts etc should be first and will create the record correctly to then subsequently update. - docs.forEach((doc) => { - operations.push({ - updateMany: { - filter: conditionals[0], - update: doc, - upsert: true, - }, - }); - }); - - try { - const result = await this.em.bulkWrite(entity, operations, {ordered: true}); - - return result.upsertedCount - ? result.upsertedCount > 0 - : result.modifiedCount - ? result.modifiedCount > 0 - : false; - - } catch (error: any) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - if (!error.message.includes('E11000')) { - // eslint-disable-next-line @typescript-eslint/restrict-template-expressions,@typescript-eslint/no-unsafe-member-access - throw new Error(`Upsert failed! E: ${error.message}`); - } - - return true; - } + // One update per doc, in order: a message's $setOnInsert must create the row before its $inc and $set apply + return await this.runUpserts(entity, (this.transform(docs) as any[]).map((doc: any) => ({ + updateOne: {filter: conditionals[0], update: doc, upsert: true}, + }))); } /** @@ -125,32 +99,9 @@ export default class MongoOperationsService { throw new Error('UpsertMany requires equal lengths of documents and conditionals!'); } - docs = this.transform(docs); - const operations: any[] = []; - - // Gather operations, setOnInserts etc should be first and will create the record correctly to then subsequently update. - docs.forEach((doc, index) => { - operations.push({ - updateMany: { - filter: conditionals[index], - update: doc, - upsert: true, - }, - }); - }); - - try { - const result = await this.em.bulkWrite(entity, operations, {ordered: true}); - return result.upsertedCount ? result.upsertedCount > 0 : result.modifiedCount ? result.modifiedCount > 0 : false; - } catch (error: any) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-call,@typescript-eslint/no-unsafe-member-access - if (!error.message.includes('E11000')) { - // eslint-disable-next-line @typescript-eslint/restrict-template-expressions,@typescript-eslint/no-unsafe-member-access - throw new Error(`UpsertMany failed! E: ${error.message}`); - } - - return true; - } + return await this.runUpserts(entity, (this.transform(docs) as any[]).map((doc: any, index: number) => ({ + updateOne: {filter: conditionals[index], update: doc, upsert: true}, + }))); } public async deleteOne(entity: any, conditional: any): Promise { @@ -191,6 +142,39 @@ export default class MongoOperationsService { } /* eslint-disable */ + /** + * Two consumers can insert the same new row at once; the loser gets a duplicate-key error. Its operation is retried + * once, now matching the winner's row, and the operations after it still run. Anything else is thrown. + */ + private async runUpserts(entity: any, operations: any[]): Promise { + let start = 0; + let retried = -1; + let changed = false; + + while (start < operations.length) { + try { + const result = await this.em.bulkWrite(entity, operations.slice(start), {ordered: true}); + changed = changed || result.upsertedCount > 0 || result.modifiedCount > 0; + break; + } catch (error: any) { + const writeErrors = error.writeErrors === undefined ? [] : [error.writeErrors].flat(); + const failed = writeErrors[0]; + + if (!failed || failed.code !== 11000 || start + Number(failed.index) === retried) { + // Only provable when the first operation of the first attempt failed, or no server was reached at all + const nothingWritten = error.name === 'MongoServerSelectionError' || (start === 0 && failed !== undefined && Number(failed.index) === 0); + throw new UpsertError(`Upsert failed! E: ${String(error.message)}`, nothingWritten); + } + + changed = changed || Number(failed.index) > 0; + retried = start + Number(failed.index); + start = retried; + } + } + + return changed; + } + private transform(docs: any): any { if (docs.constructor === Array) { docs.map((doc: any) => { diff --git a/src/services/mongo/upsert.error.ts b/src/services/mongo/upsert.error.ts new file mode 100644 index 00000000..e32b02fd --- /dev/null +++ b/src/services/mongo/upsert.error.ts @@ -0,0 +1,9 @@ +/** + * A failed upsert, marked with whether any part of it may already be in the database. Replaying a message that + * wrote nothing is safe; replaying one that half-applied its increments counts them twice. + */ +export class UpsertError extends Error { + constructor(message: string, public readonly nothingWritten: boolean) { + super(message); + } +}