Asynchronous stats support

Author: mrlika

src/build-uws-tracker.ts

@@ -9,15 +9,15 @@ type BuildServerParams = {
   serverSettings: ServerItemSettings;
   websocketsAccess: Partial<WebSocketsAccessSettings> | undefined;
   indexHtml: Buffer | undefined;
-  servers: UWebSocketsTracker[];
+  getServersStats: () => Promise<unknown>;
 };
 
 export function buildUwsTracker({
   tracker,
   serverSettings,
   websocketsAccess,
   indexHtml,
-  servers,
+  getServersStats,
 }: BuildServerParams): UWebSocketsTracker {
   if (!(serverSettings instanceof Object)) {
     throw Error(
@@ -41,39 +41,56 @@ export function buildUwsTracker({
         response.end(indexHtml);
       }
     })
-    .get("/stats.json", (response: HttpResponse, request: HttpRequest) => {
-      debugRequest(server, request);
+    .get(
+      "/stats.json",
+      async (response: HttpResponse, request: HttpRequest) => {
+        debugRequest(server, request);
 
-      const { swarms } = tracker;
-      const peersCountPerInfoHash: Record<string, number> = {};
+        response.onAborted(() => {
+          response.aborted = true;
+        });
 
-      let peersCount = 0;
-      for (const [infoHash, swarm] of swarms) {
-        peersCount += swarm.peers.length;
+        const swarms = await tracker.getSwarms();
+        const serversStats = await getServersStats();
 
-        const infoHashHex = Buffer.from(infoHash, "binary").toString("hex");
-        peersCountPerInfoHash[infoHashHex] = swarm.peers.length;
-      }
+        if (!response.aborted) {
+          const peersCountPerInfoHashPerTracker: Record<string, number>[] = [];
+          let peersCount = 0;
 
-      const serversStats = [];
-      for (const serverForStats of servers) {
-        const { settings } = serverForStats;
-        serversStats.push({
-          server: `${settings.server.host}:${settings.server.port}`,
-          webSocketsCount: serverForStats.stats.webSocketsCount,
-        });
-      }
+          for (const trackerSwarms of swarms) {
+            const peersCountPerInfoHash: Record<string, number> = {
+              totalPeers: 0,
+            };
 
-      response.writeHeader("Content-Type", "application/json").end(
-        JSON.stringify({
-          torrentsCount: swarms.size,
-          peersCount,
-          servers: serversStats,
-          memory: process.memoryUsage(),
-          peersCountPerInfoHash,
-        }),
-      );
-    })
+            for (const swarm of trackerSwarms) {
+              peersCount += swarm.peersCount;
+
+              const infoHashHex = Buffer.from(
+                swarm.infoHash,
+                "binary",
+              ).toString("hex");
+
+              peersCountPerInfoHash[infoHashHex] = swarm.peersCount;
+              peersCountPerInfoHash.totalPeers += swarm.peersCount;
+            }
+
+            peersCountPerInfoHashPerTracker.push(peersCountPerInfoHash);
+          }
+
+          response.cork(() => {
+            response.writeHeader("Content-Type", "application/json").end(
+              JSON.stringify({
+                torrentsCount: swarms.length,
+                peersCount,
+                servers: serversStats,
+                memory: process.memoryUsage(),
+                peersCountPerInfoHashPerTracker,
+              }),
+            );
+          });
+        }
+      },
+    )
     .any("/*", (response: HttpResponse, request: HttpRequest) => {
       debugRequest(server, request);
 

src/fast-tracker.ts

@@ -14,10 +14,12 @@
  * limitations under the License.
  */
 
+import { threadId } from "node:worker_threads";
 import Debug from "debug";
 import { Tracker, TrackerError } from "./tracker.js";
 
-const debug = Debug("wt-tracker:fast-tracker");
+const debugSuffix = threadId ? `-${threadId}` : "";
+const debug = Debug(`wt-tracker:fast-tracker${debugSuffix}`);
 const debugEnabled = debug.enabled;
 
 interface Swarm<ConnectionContext extends Record<string, unknown>> {
@@ -50,6 +52,19 @@ export class FastTracker<ConnectionContext extends Record<string, unknown>>
     return this.#swarms;
   }
 
+  public getSwarms() {
+    const result = [];
+
+    for (const swarm of this.#swarms.values()) {
+      result.push({
+        infoHash: swarm.infoHash,
+        peersCount: swarm.peers.length,
+      });
+    }
+
+    return Promise.resolve([result]);
+  }
+
   readonly #peers = new Map<string, PeerContext<ConnectionContext>>();
   public get peers() {
     return this.#peers;
@@ -390,10 +405,12 @@ export class FastTracker<ConnectionContext extends Record<string, unknown>>
       }
     }
 
-    debug(
-      "announce: sent offers",
-      countOffersToSend < 0 ? 0 : countOffersToSend,
-    );
+    if (debugEnabled) {
+      debug(
+        "announce: sent offers",
+        countOffersToSend < 0 ? 0 : countOffersToSend,
+      );
+    }
   }
 
   private processAnswer(json: UnknownObject): void {

src/socket-app.ts

@@ -38,13 +38,22 @@ export async function runSocketApp(
 
   const servers: UWebSocketsTracker[] = [];
 
+  const getServersStats = async () => {
+    return Promise.resolve(
+      servers.map((server, index) => ({
+        server: `${settings.servers[index].server?.host}:${settings.servers[index].server?.port}`,
+        webSocketsCount: server.stats.webSocketsCount,
+      })),
+    );
+  };
+
   const serverPromises = settings.servers.map(async (serverSettings) => {
     const server = buildUwsTracker({
       tracker,
       serverSettings,
       websocketsAccess: settings.websocketsAccess,
       indexHtml,
-      servers,
+      getServersStats,
     });
     servers.push(server);
     await server.run();

src/tracker.ts

@@ -15,8 +15,7 @@
  */
 
 export interface Tracker<ConnectionContext> {
-  readonly swarms: ReadonlyMap<string, { peers: readonly unknown[] }>;
-
+  getSwarms: () => Promise<{ infoHash: string; peersCount: number }[][]>;
   processMessage: (
     json: Record<string, unknown>,
     connection: ConnectionContext,

src/uws-tracker.ts

@@ -31,14 +31,17 @@ import {
   WebSocketsAccessSettings,
   WebSocketsSettings,
 } from "./settings.js";
+import { threadId } from "node:worker_threads";
 
-const debugWebSockets = Debug("wt-tracker:uws-tracker");
+const debugSuffix = threadId ? `-${threadId}` : "";
+
+const debugWebSockets = Debug(`wt-tracker:uws-tracker${debugSuffix}`);
 const debugWebSocketsEnabled = debugWebSockets.enabled;
 
-const debugMessages = Debug("wt-tracker:uws-tracker-messages");
+const debugMessages = Debug(`wt-tracker:uws-tracker-messages${debugSuffix}`);
 const debugMessagesEnabled = debugMessages.enabled;
 
-const debugRequests = Debug("wt-tracker:uws-tracker-requests");
+const debugRequests = Debug(`wt-tracker:uws-tracker-requests${debugSuffix}`);
 const debugRequestsEnabled = debugRequests.enabled;
 
 const decoder = new StringDecoder();
@@ -61,7 +64,6 @@ export interface PartialUwsTrackerSettings {
 
 export class UWebSocketsTracker {
   public readonly settings: UwsTrackerSettings;
-  public readonly tracker: Readonly<Tracker<UwsConnectionContext>>;
 
   private webSocketsCount = 0;
   private validateOrigin = false;
@@ -70,10 +72,9 @@ export class UWebSocketsTracker {
   readonly #app: TemplatedApp;
 
   public constructor(
-    tracker: Readonly<Tracker<UwsConnectionContext>>,
+    public readonly tracker: Readonly<Tracker<UwsConnectionContext>>,
     settings: PartialUwsTrackerSettings,
   ) {
-    this.tracker = tracker;
     this.settings = {
       server: {
         port: 8000,
@@ -112,8 +113,9 @@ export class UWebSocketsTracker {
     return this.#app;
   }
 
-  public get stats(): { webSocketsCount: number } {
+  public get stats() {
     return {
+      threadId,
       webSocketsCount: this.webSocketsCount,
     };
   }