server: rTorrent: manage request queue with p-queue - fix memory leak (…

…#650)

Our self-baked request queue doesn't work well in some cases, leading to memory leak and hung. 

Bug: #648

package.json

@@ -57,7 +57,7 @@
     "prepack": "rm -rf dist && npm run build",
     "start": "node --use_strict dist/index.js",
     "start:development:client": "node client/scripts/development.js",
-    "start:development:server": "NODE_ENV=development TS_NODE_PROJECT=server/tsconfig.json ts-node-dev --transpile-only -r tsconfig-paths/register server/bin/start.ts",
+    "start:development:server": "NODE_ENV=development tsx watch server/bin/start.ts",
     "start:test:rtorrent": "node scripts/testsetup.js",
     "test": "jest --forceExit",
     "test:watch": "jest --watchAll --forceExit",
@@ -137,6 +137,7 @@
     "d3-selection": "^3.0.0",
     "d3-shape": "^3.2.0",
     "debug": "^4.3.4",
+    "esbuild-jest": "^0.5.0",
     "eslint": "^8.42.0",
     "eslint-config-airbnb": "^19.0.4",
     "eslint-config-airbnb-typescript": "^17.0.0",
@@ -167,6 +168,7 @@
     "morgan": "^1.10.0",
     "overlayscrollbars": "^2.2.0",
     "overlayscrollbars-react": "^0.5.0",
+    "p-queue": "^7.3.4",
     "parse-torrent": "^9.1.5",
     "passport": "^0.6.0",
     "passport-jwt": "^4.0.1",
@@ -199,8 +201,8 @@
     "tldts": "^5.7.112",
     "ts-jest": "^28.0.8",
     "ts-loader": "^9.4.3",
-    "ts-node-dev": "^2.0.0",
     "tsconfig-paths": "^4.2.0",
+    "tsx": "^3.12.7",
     "typed-emitter": "^2.1.0",
     "typescript": "~4.9.5",
     "webpack": "^5.86.0",

server/.jest/auth.config.js

@@ -1,7 +1,9 @@
 const {compilerOptions} = require('../tsconfig.json');
 const {pathsToModuleNameMapper} = require('ts-jest');
+const common = require('./common');
 
 module.exports = {
+  ...common,
   displayName: 'auth',
   preset: 'ts-jest/presets/js-with-babel',
   rootDir: './../',
@@ -11,9 +13,4 @@ module.exports = {
   testEnvironment: 'node',
   testMatch: ['<rootDir>/routes/api/auth.test.ts'],
   setupFilesAfterEnv: ['<rootDir>/.jest/auth.setup.js'],
-  globals: {
-    'ts-jest': {
-      isolatedModules: true,
-    },
-  },
 };

server/.jest/common.js

@@ -0,0 +1,12 @@
+module.exports = {
+  transformIgnorePatterns: ['node_modules/(?!(p-queue|p-timeout)/)'],
+  transform: {
+    // transform ESM only package to CommonJS
+    '^.+\\.(t|j)sx?$': [
+      'esbuild-jest',
+      {
+        format: 'cjs',
+      },
+    ],
+  },
+};

server/.jest/qbittorrent.config.js

@@ -1,7 +1,9 @@
 const {compilerOptions} = require('../tsconfig.json');
 const {pathsToModuleNameMapper} = require('ts-jest');
+const common = require('./common');
 
 module.exports = {
+  ...common,
   displayName: 'qbittorrent',
   preset: 'ts-jest/presets/js-with-babel',
   rootDir: './../',
@@ -11,9 +13,4 @@ module.exports = {
   testEnvironment: 'node',
   testPathIgnorePatterns: ['auth.test.ts'],
   setupFilesAfterEnv: ['<rootDir>/.jest/qbittorrent.setup.js'],
-  globals: {
-    'ts-jest': {
-      isolatedModules: true,
-    },
-  },
 };

server/.jest/rtorrent.config.js

@@ -1,7 +1,9 @@
 const {compilerOptions} = require('../tsconfig.json');
 const {pathsToModuleNameMapper} = require('ts-jest');
+const common = require('./common');
 
 module.exports = {
+  ...common,
   displayName: 'rtorrent',
   preset: 'ts-jest/presets/js-with-babel',
   rootDir: './../',
@@ -11,9 +13,4 @@ module.exports = {
   testEnvironment: 'node',
   testPathIgnorePatterns: ['auth.test.ts'],
   setupFilesAfterEnv: ['<rootDir>/.jest/rtorrent.setup.js'],
-  globals: {
-    'ts-jest': {
-      isolatedModules: true,
-    },
-  },
 };

server/.jest/transmission.config.js

@@ -1,7 +1,9 @@
 const {compilerOptions} = require('../tsconfig.json');
 const {pathsToModuleNameMapper} = require('ts-jest');
+const common = require('./common');
 
 module.exports = {
+  ...common,
   displayName: 'transmission',
   preset: 'ts-jest/presets/js-with-babel',
   rootDir: './../',
@@ -11,9 +13,4 @@ module.exports = {
   testEnvironment: 'node',
   testPathIgnorePatterns: ['auth.test.ts'],
   setupFilesAfterEnv: ['<rootDir>/.jest/transmission.setup.js'],
-  globals: {
-    'ts-jest': {
-      isolatedModules: true,
-    },
-  },
 };

server/services/rTorrent/clientRequestManager.ts

@@ -6,20 +6,14 @@ import {methodCallJSON, methodCallXML} from './util/scgiUtil';
 import {sanitizePath} from '../../util/fileUtil';
 
 import type {MultiMethodCalls} from './util/rTorrentMethodCallUtil';
+import PQueue from 'p-queue';
 
 type MethodCallParameters = Array<string | Buffer | MultiMethodCalls>;
 
 class ClientRequestManager {
+  queue: PQueue;
   connectionSettings: RTorrentConnectionSettings;
   isJSONCapable = false;
-  isRequestPending = false;
-  lastResponseTimestamp = 0;
-  pendingRequests: Array<{
-    methodName: string;
-    parameters: MethodCallParameters;
-    resolve: (value?: Record<string, string>) => void;
-    reject: (error?: NodeJS.ErrnoException) => void;
-  }> = [];
 
   constructor(connectionSettings: RTorrentConnectionSettings) {
     if (connectionSettings.type === 'socket') {
@@ -30,36 +24,10 @@ class ClientRequestManager {
     } else {
       this.connectionSettings = connectionSettings;
     }
-  }
-
-  handleRequestEnd() {
-    this.isRequestPending = false;
-
-    // We avoid initiating any deferred requests until at least 250ms have
-    // since the previous response.
-    const currentTimestamp = Date.now();
-    const timeSinceLastResponse = currentTimestamp - this.lastResponseTimestamp;
 
-    if (timeSinceLastResponse <= 250) {
-      const delay = 250 - timeSinceLastResponse;
-      setTimeout(this.sendDeferredMethodCall, delay);
-      this.lastResponseTimestamp = currentTimestamp + delay;
-    } else {
-      this.sendDeferredMethodCall();
-      this.lastResponseTimestamp = currentTimestamp;
-    }
+    this.queue = new PQueue({autoStart: true, concurrency: 1});
   }
 
-  sendDeferredMethodCall = () => {
-    const nextRequest = this.pendingRequests.shift();
-    if (nextRequest == null) {
-      return;
-    }
-
-    this.isRequestPending = true;
-    this.sendMethodCall(nextRequest.methodName, nextRequest.parameters).then(nextRequest.resolve, nextRequest.reject);
-  };
-
   sendMethodCall = (methodName: string, parameters: MethodCallParameters) => {
     const connectionOptions: NetConnectOpts =
       this.connectionSettings.type === 'socket'
@@ -71,38 +39,14 @@ class ClientRequestManager {
             port: this.connectionSettings.port,
           };
 
-    const methodCall = this.isJSONCapable
+    return this.isJSONCapable
       ? methodCallJSON(connectionOptions, methodName, parameters)
       : methodCallXML(connectionOptions, methodName, parameters);
-
-    return methodCall.then(
-      (response) => {
-        this.handleRequestEnd();
-        return response;
-      },
-      (error) => {
-        this.handleRequestEnd();
-        throw error;
-      },
-    );
   };
 
   methodCall = (methodName: string, parameters: MethodCallParameters) => {
     // We only allow one request at a time.
-    if (this.isRequestPending) {
-      return new Promise(
-        (resolve: (value?: Record<string, string>) => void, reject: (error?: NodeJS.ErrnoException) => void) => {
-          this.pendingRequests.push({
-            methodName,
-            parameters,
-            resolve,
-            reject,
-          });
-        },
-      );
-    }
-    this.isRequestPending = true;
-    return this.sendMethodCall(methodName, parameters);
+    return this.queue.add(() => this.sendMethodCall(methodName, parameters));
   };
 }