fix: stop aggregate writes losing messages silently

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();
+    });
+});

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<void> {
+        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);
+    }
 }

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);
         }
     }
 }

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);
         }
     }
 }

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);
         }
     }
 }

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);
         }
     }
 }

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);
         }
     }
 }

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);
         }
     }
 }

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);
         }
     }
 }

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)}`);
         }
     }
 }

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);
         }
     }
 }