fix: correct target stats when sockets are reused (#576)

This PR attempts to fix incorrect stats due to the reuse of target sockets for HTTP(S) protocols. 

Based on #572

Note: I was forced to upgrade `actions/cache` as `v2` was deprecated and it wouldn't run with it.
I also had to edit eslint config to run with `.` instead of `src` and excluded `tests`, because with `src` the CI was failing (no idea why).

Author: jirimoravcik

.github/scripts/before-beta-release.js

@@ -1,9 +1,10 @@
-const path = require('path');
-const fs = require('fs');
 const { execSync } = require('child_process');
+const fs = require('fs');
+const path = require('path');
 
 const PKG_JSON_PATH = path.join(__dirname, '..', '..', 'package.json');
 
+// eslint-disable-next-line import/no-dynamic-require
 const pkgJson = require(PKG_JSON_PATH);
 
 const PACKAGE_NAME = pkgJson.name;
@@ -13,20 +14,20 @@ const nextVersion = getNextVersion(VERSION);
 console.log(`before-deploy: Setting version to ${nextVersion}`);
 pkgJson.version = nextVersion;
 
-fs.writeFileSync(PKG_JSON_PATH, JSON.stringify(pkgJson, null, 2) + '\n');
+fs.writeFileSync(PKG_JSON_PATH, `${JSON.stringify(pkgJson, null, 2)}\n`);
 
 function getNextVersion(version) {
-    const versionString = execSync(`npm show ${PACKAGE_NAME} versions --json`, { encoding: 'utf8'});
+    const versionString = execSync(`npm show ${PACKAGE_NAME} versions --json`, { encoding: 'utf8' });
     const versions = JSON.parse(versionString);
 
-    if (versions.some(v => v === VERSION)) {
+    if (versions.some((v) => v === VERSION)) {
         console.error(`before-deploy: A release with version ${VERSION} already exists. Please increment version accordingly.`);
         process.exit(1);
     }
 
     const prereleaseNumbers = versions
-        .filter(v => (v.startsWith(VERSION) && v.includes('-')))
-        .map(v => Number(v.match(/\.(\d+)$/)[1]));
+        .filter((v) => (v.startsWith(VERSION) && v.includes('-')))
+        .map((v) => Number(v.match(/\.(\d+)$/)[1]));
     const lastPrereleaseNumber = Math.max(-1, ...prereleaseNumbers);
-    return `${version}-beta.${lastPrereleaseNumber + 1}`
+    return `${version}-beta.${lastPrereleaseNumber + 1}`;
 }

.github/workflows/check.yaml

@@ -28,7 +28,7 @@ jobs:
             -
                 name: Cache Node Modules
                 if: ${{ matrix.node-version == 18 }}
-                uses: actions/cache@v2
+                uses: actions/cache@v4
                 with:
                     path: |
                         node_modules
@@ -64,7 +64,7 @@ jobs:
                     node-version: 18
             -
                 name: Load Cache
-                uses: actions/cache@v2
+                uses: actions/cache@v4
                 with:
                     path: |
                         node_modules

.github/workflows/release.yaml

@@ -32,7 +32,7 @@ jobs:
       -
         name: Cache Node Modules
         if: ${{ matrix.node-version == 18 }}
-        uses: actions/cache@v2
+        uses: actions/cache@v4
         with:
           path: |
             node_modules
@@ -67,7 +67,7 @@ jobs:
           node-version: 18
       -
         name: Load Cache
-        uses: actions/cache@v2
+        uses: actions/cache@v4
         with:
           path: |
             node_modules
@@ -93,7 +93,7 @@ jobs:
           registry-url: https://registry.npmjs.org/
       -
         name: Load Cache
-        uses: actions/cache@v2
+        uses: actions/cache@v4
         with:
           path: |
             node_modules

eslint.config.mjs

@@ -2,7 +2,7 @@ import apify from '@apify/eslint-config';
 
 // eslint-disable-next-line import/no-default-export
 export default [
-    { ignores: ['**/dist'] }, // Ignores need to happen first
+    { ignores: ['**/dist', 'test'] }, // Ignores need to happen first
     ...apify,
     {
         languageOptions: {

jest.config.ts

@@ -1,5 +1,6 @@
 import type { Config } from '@jest/types';
 
+// eslint-disable-next-line import/no-default-export
 export default (): Config.InitialOptions => ({
     verbose: true,
     preset: 'ts-jest',

package.json

@@ -37,8 +37,8 @@
     "prepublishOnly": "npm run build",
     "local-proxy": "node ./dist/run_locally.js",
     "test": "nyc cross-env NODE_OPTIONS=--insecure-http-parser mocha --bail",
-    "lint": "eslint src",
-    "lint:fix": "eslint src --fix"
+    "lint": "eslint .",
+    "lint:fix": "eslint . --fix"
   },
   "engines": {
     "node": ">=14"

src/chain.ts

@@ -7,6 +7,7 @@ import type { URL } from 'url';
 
 import type { Socket } from './socket';
 import { badGatewayStatusCodes, createCustomStatusHttpResponse, errorCodeToStatusCode } from './statuses';
+import type { SocketWithPreviousStats } from './utils/count_target_bytes';
 import { countTargetBytes } from './utils/count_target_bytes';
 import { getBasicAuthorizationHeader } from './utils/get_basic';
 
@@ -85,9 +86,15 @@ export const chain = (
     const fn = proxy.protocol === 'https:' ? https.request : http.request;
     const client = fn(proxy.origin, options as unknown as http.ClientRequestArgs);
 
-    client.on('connect', (response, targetSocket, clientHead) => {
+    client.once('socket', (targetSocket: SocketWithPreviousStats) => {
+        // Socket can be re-used by multiple requests.
+        // That's why we need to track the previous stats.
+        targetSocket.previousBytesRead = targetSocket.bytesRead;
+        targetSocket.previousBytesWritten = targetSocket.bytesWritten;
         countTargetBytes(sourceSocket, targetSocket);
+    });
 
+    client.on('connect', (response, targetSocket, clientHead) => {
         if (sourceSocket.readyState !== 'open') {
             // Sanity check, should never reach.
             targetSocket.destroy();

src/forward.ts

@@ -6,6 +6,7 @@ import type { URL } from 'url';
 import util from 'util';
 
 import { badGatewayStatusCodes, errorCodeToStatusCode } from './statuses';
+import type { SocketWithPreviousStats } from './utils/count_target_bytes';
 import { countTargetBytes } from './utils/count_target_bytes';
 import { getBasicAuthorizationHeader } from './utils/get_basic';
 import { validHeadersOnly } from './utils/valid_headers_only';
@@ -114,8 +115,12 @@ export const forward = async (
         }
     });
 
-    client.once('socket', (socket) => {
-        countTargetBytes(request.socket, socket);
+    client.once('socket', (socket: SocketWithPreviousStats) => {
+        // Socket can be re-used by multiple requests.
+        // That's why we need to track the previous stats.
+        socket.previousBytesRead = socket.bytesRead;
+        socket.previousBytesWritten = socket.bytesWritten;
+        countTargetBytes(request.socket, socket, (handler) => response.once('close', handler));
     });
 
     // Can't use pipeline here as it automatically destroys the streams

src/utils/count_target_bytes.ts

@@ -7,17 +7,27 @@ const calculateTargetStats = Symbol('calculateTargetStats');
 
 type Stats = { bytesWritten: number | null, bytesRead: number | null };
 
+/**
+ * Socket object extended with previous read and written bytes.
+ * Necessary due to target socket re-use.
+ */
+export type SocketWithPreviousStats = net.Socket & { previousBytesWritten?: number, previousBytesRead?: number };
+
 interface Extras {
     [targetBytesWritten]: number;
     [targetBytesRead]: number;
-    [targets]: Set<net.Socket>;
+    [targets]: Set<SocketWithPreviousStats>;
     [calculateTargetStats]: () => Stats;
 }
 
 // @ts-expect-error TS is not aware that `source` is used in the assertion.
 function typeSocket(source: unknown): asserts source is net.Socket & Extras {}
 
-export const countTargetBytes = (source: net.Socket, target: net.Socket): void => {
+export const countTargetBytes = (
+    source: net.Socket,
+    target: SocketWithPreviousStats,
+    registerCloseHandler?: (handler: () => void) => void,
+): void => {
     typeSocket(source);
 
     source[targetBytesWritten] = source[targetBytesWritten] || 0;
@@ -26,21 +36,24 @@ export const countTargetBytes = (source: net.Socket, target: net.Socket): void =
 
     source[targets].add(target);
 
-    target.once('close', () => {
-        source[targetBytesWritten] += target.bytesWritten;
-        source[targetBytesRead] += target.bytesRead;
-
+    const closeHandler = () => {
+        source[targetBytesWritten] += (target.bytesWritten - (target.previousBytesWritten || 0));
+        source[targetBytesRead] += (target.bytesRead - (target.previousBytesRead || 0));
         source[targets].delete(target);
-    });
+    };
+    if (!registerCloseHandler) {
+        registerCloseHandler = (handler: () => void) => target.once('close', handler);
+    }
+    registerCloseHandler(closeHandler);
 
     if (!source[calculateTargetStats]) {
         source[calculateTargetStats] = () => {
             let bytesWritten = source[targetBytesWritten];
             let bytesRead = source[targetBytesRead];
 
             for (const socket of source[targets]) {
-                bytesWritten += socket.bytesWritten;
-                bytesRead += socket.bytesRead;
+                bytesWritten += (socket.bytesWritten - (socket.previousBytesWritten || 0));
+                bytesRead += (socket.bytesRead - (socket.previousBytesRead || 0));
             }
 
             return {