fixed traffic stats for http connection reuse

src/forward.ts

@@ -5,8 +5,9 @@
 import type { URL } from 'url';
 import util from 'util';
 
+import { Socket } from './socket';
 import { badGatewayStatusCodes, errorCodeToStatusCode } from './statuses';
-import { countTargetBytes } from './utils/count_target_bytes';
+import { countTargetBytes, SocketPreviousStats } from './utils/count_target_bytes';
 import { getBasicAuthorizationHeader } from './utils/get_basic';
 import { validHeadersOnly } from './utils/valid_headers_only';
 
@@ -114,8 +115,12 @@
         }
     });
 
-    client.once('socket', (socket) => {
-        countTargetBytes(request.socket, socket);
+    client.once('socket', (socket: Socket & SocketPreviousStats) => {
+        // socket can be re-used by multiple requests (HTTP keep alive)
+        // (even in multiple Server objects)
+        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,23 @@ const calculateTargetStats = Symbol('calculateTargetStats');
 
 type Stats = { bytesWritten: number | null, bytesRead: number | null };
 
+export type SocketPreviousStats = { previousBytesWritten?: number, previousBytesRead?: number };
+
 interface Extras {
     [targetBytesWritten]: number;
     [targetBytesRead]: number;
-    [targets]: Set<net.Socket>;
+    [targets]: Set<net.Socket & SocketPreviousStats>;
     [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: net.Socket & SocketPreviousStats,
+    finishRegister?: (handler: () => void) => void,
+): void => {
     typeSocket(source);
 
     source[targetBytesWritten] = source[targetBytesWritten] || 0;
@@ -26,21 +32,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 finishHandler = () => {
+        source[targetBytesWritten] += (target.bytesWritten - (target.previousBytesWritten || 0));
+        source[targetBytesRead] += (target.bytesRead - (target.previousBytesRead || 0));
         source[targets].delete(target);
-    });
+    };
+    if (!finishRegister) {
+        finishRegister = (handler: () => void) => target.once('close', handler);
+    }
+    finishRegister(finishHandler);
 
     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 {