perf: avoid needless downloads and track request latency (#876)

* perf: avoid needless downloads and track request latency

* fix: correct request timing and monitoring revalidation

src/core/anonymization-pool.ts

@@ -1,6 +1,8 @@
 import { Worker } from "worker_threads";
 import { existsSync } from "fs";
 import { join } from "path";
+import { startStage } from "./request-monitoring";
+import { AsyncResource } from "async_hooks";
 
 const MEMORY_BUDGET = Math.max(128, Number(process.env.ANONYMIZATION_MEMORY_MB) || 512) * 1024 * 1024;
 const MAX_WORKERS = Math.min(4, Math.max(1, Number(process.env.ANONYMIZATION_WORKERS) || 1));
@@ -12,6 +14,8 @@ interface Job {
   reject: (error: Error) => void;
   timer?: NodeJS.Timeout;
   abort?: () => void;
+  waitingDone: ReturnType<typeof startStage>;
+  transformDone: () => ReturnType<typeof startStage>;
 }
 const waiting: Job[] = [];
 const idle: Worker[] = [];
@@ -23,16 +27,20 @@ export function anonymizeOnWorker(message: Job["message"], size: number, signal:
   if (weight > MEMORY_BUDGET) return Promise.reject(new Error("anonymization_memory_budget_exceeded"));
   if (waiting.length >= 16) return Promise.reject(new Error("anonymization_queue_full"));
   return new Promise((resolve, reject) => {
-    const job: Job = { message, weight, signal, resolve, reject };
+    const waitingDone = startStage("worker_wait");
+    // Bind the stage factory to the submitting request, not the preceding job.
+    const transformDone = AsyncResource.bind(() => startStage("anonymize"));
+    const job: Job = { message, weight, signal, resolve, reject, waitingDone, transformDone };
     job.abort = () => {
       const index = waiting.indexOf(job);
       if (index < 0) return;
       waiting.splice(index, 1);
       clearTimeout(job.timer);
       signal.removeEventListener("abort", job.abort!);
+      waitingDone(true);
       reject(new Error("anonymization_cancelled"));
     };
-    if (signal.aborted) return reject(new Error("anonymization_cancelled"));
+    if (signal.aborted) { waitingDone(true); return reject(new Error("anonymization_cancelled")); }
     signal.addEventListener("abort", job.abort, { once: true });
     job.timer = setTimeout(job.abort, 30_000);
     waiting.push(job);
@@ -42,6 +50,8 @@ export function anonymizeOnWorker(message: Job["message"], size: number, signal:
 function pump() {
   while (waiting.length && running < MAX_WORKERS && bytes + waiting[0].weight <= MEMORY_BUDGET) {
     const job = waiting.shift()!;
+    job.waitingDone();
+    const transformDone = job.transformDone();
     clearTimeout(job.timer);
     job.signal.removeEventListener("abort", job.abort!);
     running++;
@@ -51,12 +61,13 @@ function pump() {
     try { worker = idle.pop() || new Worker(existsSync(compiled) ? compiled : join(__dirname, "anonymization.worker.ts"), {
       execArgv: existsSync(compiled) ? [] : ["-r", require.resolve("ts-node/register/transpile-only")],
       resourceLimits: { maxOldGenerationSizeMb: Math.ceil(MEMORY_BUDGET / 1024 / 1024) },
-    }); } catch (error) { running--; bytes -= job.weight; job.reject(error as Error); continue; }
+    }); } catch (error) { running--; bytes -= job.weight; transformDone(true); job.reject(error as Error); continue; }
     worker.ref();
     let finished = false;
     const finish = (error?: Error, changed = false) => {
       if (finished) return;
       finished = true;
+      transformDone(!!error);
       clearTimeout(deadline);
       job.signal.removeEventListener("abort", cancel);
       worker.removeListener("message", message);
@@ -89,3 +100,7 @@ export function reserveSpool(bytes: number) {
   spoolBytes += bytes;
 }
 export function releaseSpool(bytes: number) { spoolBytes -= bytes; }
+export function getAnonymizationPoolStats() {
+  return { running, waiting: waiting.length, idle: idle.length, reservedBytes: bytes, spoolBytes,
+    maxWorkers: MAX_WORKERS, maxWaiting: 16, memoryBudgetBytes: MEMORY_BUDGET };
+}