smile-nestjs-sqs – entwickelt, um NestJS SQS in Production sicherer zu nutzen

Hallo.

Ich entwickle Services mit der Kombination aus NestJS + SQS und habe dabei andere SQS-Pakete aus dem NestJS-Ökosystem verwendet.
Dabei hatte ich einige Unannehmlichkeiten, die ich in echter Production erlebt habe. Für ein bestimmtes Paket habe ich auch einen PR eingereicht, aber manche werden nicht mehr aktualisiert, und anderswo wird das nicht einmal als Problem erkannt. Deshalb habe ich ein Modul erstellt, um diese Punkte selbst zu lösen. Tatsächlich habe ich das bisher verwendete SQS-Modul bereits dadurch ersetzt und nutze es nun stabil.
Ich teile es hier, weil es für alle hilfreich sein könnte, die in NestJS nach einem SQS-Modul suchen oder ähnliche Überlegungen wie ich hatten.

Dieses Paket hat die folgenden Eigenschaften:

NestJS-freundlich

In NestJS können Module dynamisch erzeugt und importiert werden, und durch das Injizieren von Handlern mit Dekoratoren ist eine einfache Implementierung möglich.
Die Signatur ist ähnlich wie bei den meisten anderen SQS-Paketen.

@Module({  
  imports: [  
    ConfigModule.forRoot(),  
    SqsModule.registerAsync({  
      inject: [ConfigService],  
      useFactory: (config: ConfigService) => {  
        const region = config.getOrThrow<string>('AWS_REGION');  
        const queueUrl = config.getOrThrow<string>('ORDERS_QUEUE_URL');  
        const defaultSqsClient = new SQSClient({ region });  
  
        return {  
          defaultSqsClient,  
          consumers: [{ name: 'orders', queueUrl }],  
          producers: [{ name: 'orders', queueUrl }],  
        };  
      },  
    }),  
  ],  
})  
export class AppModule {}  
  
@Injectable()  
export class OrderQueueHandler {  
  @SqsMessageHandler('orders')  
  public async onMessage(message: Message): Promise<Message> {  
    // return message to ack/delete  
    return message;  
  }  
  
  @SqsConsumerEventHandler('orders', 'processing_error')  
  public onProcessingError(error: Error, message: Message) {  
    // report error  
  }  
}  

Verwendet aktuelle BBC-Pakete

Dieses Paket basiert auf den neuesten Versionen der beiden untenstehenden Bibliotheken, die mit mehr als 1,3 Millionen Downloads pro Woche zu den bekanntesten SQS-Bibliotheken im Node.js-Umfeld gehören.

https://github.com/bbc/sqs-producer
https://github.com/bbc/sqs-consumer

In sqs-consumer gibt es ab v14 eine bedeutende Breaking Change.
https://github.com/bbc/sqs-consumer/discussions/584

Bis einschließlich Version 13 wurde es als Ack-Verhalten betrachtet, selbst wenn der SQS-Handler void zurückgab.
Ab Version 14 gilt jedoch ausdrücklich die vom SQS-Handler zurückgegebene Message als Ack.
Diese Änderung ist sinnvoll, weil sie expliziter und besser vorhersehbar ist und Nutzern bei der Implementierung von Ack-Verhalten weniger Verwirrung bereitet.
Die meisten der derzeit als NestJS-SQS-Module angebotenen npm-Pakete verwenden jedoch weiterhin sqs-consumer in Version 13 oder älter.

Validierung zur Boot-Zeit

Fehlkonfigurationen werden direkt beim Start der App erkannt, sodass das Paket sicherer genutzt werden kann.

• Erkennung doppelter Consumer-/Producer-Namen
• Sofortiger Fehler, wenn ein Dekorator für einen nicht existierenden Consumer verwendet wird
• Erkennung von Tippfehlern in Event-Namen
• Erkennung von Typabweichungen bei Parametern von Batch-/Single-Handlern
• Erkennung von Typabweichungen bei Rückgabewerten von Batch-/Single-Handlern

Graceful Shutdown

bbc/sqs-consumer bietet für Graceful Shutdown die Option pollingCompleteWaitTimeMs.
Und in NestJS gibt es den NestJS-Lifecycle.
In der Phase onModuleDestroy von NestJS wird intern gewartet, bis jeder Consumer das Event stopped veröffentlicht.
Da dieses Event erst ausgelöst wird, wenn die gerade verarbeitete Message abgeschlossen ist, wird die Nachrichtenverarbeitung beendet und der Prozess sicher heruntergefahren.
Weil dies automatisch mit dem Lifecycle von NestJS verbunden ist,
reicht es, app.enableShutdownHooks() zu aktivieren und
pollingCompleteWaitTimeMs sowie shutdownTimeoutMs passend zu konfigurieren, damit es auch in einer k8s-Umgebung bei plötzlichem Beenden sicherer verwendet werden kann.

Während ich SQS in Production verwendet habe, gab es immer wieder Dinge, bei denen ich dachte: Wann wird das eigentlich aktualisiert? Da diese Punkte weiterhin nicht adressiert wurden, war das ein Hindernis beim Aufbau einer robusten Entwicklungskultur, und dieses Ergebnis ist meine Lösung dafür.
Wenn Sie in einer ähnlichen Umgebung mit ähnlichen Problemen arbeiten, würde ich mich freuen, wenn Sie es einmal ausprobieren und Feedback geben.