perf: keep api-queue-production in memory, off the disk

.github/workflows/test.yml

@@ -0,0 +1,31 @@
+---
+name: Tests
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    name: Tests
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          submodules: true
+
+      - name: Install node v22
+        uses: actions/setup-node@v7
+        with:
+          node-version: 22
+
+      - name: Install modules
+        run: yarn install
+
+      - name: Run tests
+        run: yarn test

package.json

@@ -13,7 +13,8 @@
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
     "prepare": "husky install",
     "lint:check": "eslint ./src --ext .ts",
-    "lint:fix": "eslint ./src --ext .ts --fix"
+    "lint:fix": "eslint ./src --ext .ts --fix",
+    "test": "jest"
   },
   "repository": {
     "type": "git",
@@ -49,20 +50,36 @@
     "@nestjs/cli": "^9.0.0",
     "@nestjs/schematics": "^9.0.0",
     "@types/amqplib": "^0.10.1",
+    "@types/jest": "^29.5.14",
     "@types/lodash": "^4.14.192",
     "@types/node": "^18.15.10",
     "@types/ws": "^8.5.4",
     "@typescript-eslint/eslint-plugin": "^5.15.0",
     "@typescript-eslint/parser": "^5.32.0",
     "eslint": "^8.21.0",
     "husky": "^8.0.1",
+    "jest": "^29.7.0",
     "lint-staged": "^13.0.3",
+    "ts-jest": "^29.2.5",
     "ts-loader": "^9.2.3",
     "ts-node": "^10.9.1",
     "ts-node-dev": "^2.0.0",
     "tsconfig-paths": "^4.0.0",
     "typescript": "^4.6.2"
   },
+  "jest": {
+    "moduleFileExtensions": [
+      "js",
+      "json",
+      "ts"
+    ],
+    "rootDir": "src",
+    "testRegex": ".spec.ts$",
+    "transform": {
+      "^.+\\.(t|j)s$": "ts-jest"
+    },
+    "testEnvironment": "node"
+  },
   "lint-staged": {
     "src/**/*.ts": [
       "eslint --fix"

src/modules/rabbitmq/factories/RabbitMQQueueFactory.ts

@@ -50,6 +50,7 @@ export default class RabbitMQQueueFactory {
     public createApiQueue(
         queueName: string,
         ttl: number,
+        durable: boolean,
         deadLetterExchange?: string,
         deadLetterRoutingKey?: string,
     ): ApiQueue {
@@ -59,6 +60,7 @@ export default class RabbitMQQueueFactory {
             this.metricsHandler,
             this.config.get('rabbitmq.exchange'),
             ttl,
+            durable,
             deadLetterExchange,
             deadLetterRoutingKey,
         );

src/modules/rabbitmq/publishers/ApiMQDelayPublisher.ts

@@ -25,13 +25,15 @@ export default class ApiMQDelayPublisher implements RabbitMQQueueWrapperInterfac
         this.longQueue = this.queueFactory.createApiQueue(
             `${this.apiDelayQueueName}-91min`,
             91 * 60 * 1000,
+            true,
             '',
             this.apiQueueName,
         );
 
         this.shortQueue = this.queueFactory.createApiQueue(
             `${this.apiDelayQueueName}-46min`,
             46 * 60 * 1000,
+            true,
             '',
             this.apiQueueName,
         );

src/modules/rabbitmq/publishers/ApiMQPublisher.ts

@@ -18,9 +18,11 @@ export default class ApiMQPublisher implements RabbitMQQueueWrapperInterface {
     }
 
     public async connect(): Promise<void> {
+        // Memory-only. Must match the api's declaration of this queue, or whichever declares second fails.
         this.queue = this.queueFactory.createApiQueue(
             this.apiQueueName,
             180 * 60 * 1000,
+            false,
         );
 
         await this.queue.connect();

src/modules/rabbitmq/queues/ApiQueue.spec.ts

@@ -0,0 +1,63 @@
+import {AmqpConnectionManager} from 'amqp-connection-manager';
+import {ConfigService} from '@nestjs/config';
+import {ApiQueue} from './ApiQueue';
+import MetricsHandler from '../../../handlers/MetricsHandler';
+import ApiMQMessage from '../../../data/ApiMQMessage';
+import ApiMQPublisher from '../publishers/ApiMQPublisher';
+import RabbitMQQueueFactory from '../factories/RabbitMQQueueFactory';
+
+const declareAndSend = async (durable: boolean, deadLetterExchange?: string, deadLetterRoutingKey?: string) => {
+    const assertQueue = jest.fn<Promise<unknown>, [string, Record<string, unknown>]>().mockResolvedValue({});
+    const sendToQueue = jest.fn<Promise<boolean>, [string, unknown, Record<string, unknown>]>().mockResolvedValue(true);
+    const manager = {
+        createChannel: jest.fn((opts: {setup(channel: unknown): Promise<void>}) => {
+            void opts.setup({checkExchange: jest.fn(), assertQueue, bindQueue: jest.fn()});
+            return {on: jest.fn(), waitForConnect: jest.fn().mockResolvedValue(undefined), sendToQueue};
+        }),
+    } as unknown as AmqpConnectionManager;
+
+    const queue = new ApiQueue(manager, 'api-queue-test', {} as MetricsHandler, 'ps2alerts', 1000, durable, deadLetterExchange, deadLetterRoutingKey);
+    await queue.connect();
+    await queue.send({pattern: 'test', data: {docs: [{}]}} as unknown as ApiMQMessage);
+
+    return {
+        options: assertQueue.mock.calls[0][1],
+        publishOptions: sendToQueue.mock.calls[0][2],
+    };
+};
+
+describe('ApiQueue', () => {
+    it('declares a non-durable queue memory-only and publishes to it transient', async () => {
+        const {options, publishOptions} = await declareAndSend(false);
+
+        expect(options).toMatchObject({durable: false, messageTtl: 1000, arguments: {}});
+        expect(publishOptions).toEqual({persistent: false});
+    });
+
+    it('keeps a durable delay queue lazy and persistent, dead-lettering to the API queue', async () => {
+        const {options, publishOptions} = await declareAndSend(true, '', 'api-queue-test');
+
+        expect(options).toMatchObject({
+            durable: true,
+            deadLetterExchange: '',
+            deadLetterRoutingKey: 'api-queue-test',
+            arguments: {'x-queue-mode': 'lazy'},
+        });
+        expect(publishOptions).toEqual({persistent: true});
+    });
+});
+
+describe('ApiMQPublisher', () => {
+    // RabbitMQ refuses a redeclare with different arguments, so this must equal the api's rabbitmq.queueOptions
+    it('declares the API queue memory-only with the api\'s 3 hour TTL', async () => {
+        const connect = jest.fn().mockResolvedValue(undefined);
+        const createApiQueue = jest.fn().mockReturnValue({connect});
+        const factory = {createApiQueue} as unknown as RabbitMQQueueFactory;
+        const config = {get: () => 'api-queue-test'} as unknown as ConfigService;
+
+        await new ApiMQPublisher(factory, config).connect();
+
+        expect(createApiQueue).toHaveBeenCalledWith('api-queue-test', 10800000, false);
+        expect(connect).toHaveBeenCalled();
+    });
+});

src/modules/rabbitmq/queues/ApiQueue.ts

@@ -22,21 +22,21 @@ export class ApiQueue extends RabbitMQQueue implements PS2AlertsQueueInterface {
         metricsHandler: MetricsHandler,
         private readonly exchange: string,
         private readonly ttl: number,
+        private readonly durable: boolean,
         private readonly deadLetterExchange: string | undefined,
         private readonly deadLetterRoutingKey: string | undefined,
     ) {
         super(connectionManager, queueName, metricsHandler);
     }
 
     public async connect(): Promise<void> {
+        // A non-durable queue is memory-only: lazy mode would write every message to disk regardless
         const queueOptions = {
-            durable: true,
-            messageTtl: this.ttl, // 46 minutes
+            durable: this.durable,
+            messageTtl: this.ttl,
             deadLetterExchange: this.deadLetterExchange ?? undefined,
             deadLetterRoutingKey: this.deadLetterRoutingKey ?? undefined,
-            arguments: {
-                'x-queue-mode': 'lazy',
-            },
+            arguments: this.durable ? {'x-queue-mode': 'lazy'} : {},
         };
 
         await this.createChannel({
@@ -62,7 +62,7 @@ export class ApiQueue extends RabbitMQQueue implements PS2AlertsQueueInterface {
             await this.getChannel().sendToQueue(
                 this.queueName,
                 msg,
-                {persistent: true},
+                {persistent: this.durable},
             );
             return true;
         } catch (err) {