fix: 收敛 WebProbe 物理内存与资源上限

2026-07-13 08:42:44 +02:00
parent 0c22a05cee
commit 2e38d690e7
31 changed files with 3882 additions and 1772 deletions
@@ -2,13 +2,221 @@
 // Responsibility: Runner low-level IO, URL, YAML-derived policy parsing, redaction, and timeout utility source fragment.

 export function nodeWebObserveRunnerUtilitySource(): string {
-  return String.raw`async function appendJsonl(file, value) {
-  await appendFile(file, JSON.stringify(sanitize(value)) + "\n", { mode: 0o600 });
+  return String.raw`function appendJsonl(file, value) {
+  let line;
+  try {
+    line = JSON.stringify(sanitize(value)) + "\n";
+  } catch {
+    recordJsonlDrop(file, 0);
+    return Promise.resolve({ ok: false, dropped: true, reason: "json-encode-failed", valuesRedacted: true });
+  }
+  const bytes = Buffer.byteLength(line);
+  if (bytes > maxJsonlLineBytes) {
+    recordJsonlDrop(file, bytes);
+    return Promise.resolve({ ok: false, dropped: true, reason: "entry-too-large", bytes, maxJsonlLineBytes, valuesRedacted: true });
+  }
+  if (jsonlWriterState.queue.length >= maxJsonlQueueEntries || jsonlWriterState.queuedBytes + bytes > maxJsonlQueueBytes) {
+    recordJsonlDrop(file, bytes);
+    return Promise.resolve({ ok: false, dropped: true, reason: "queue-saturated", bytes, valuesRedacted: true });
+  }
+  jsonlWriterState.acceptedEntries += 1;
+  jsonlWriterState.acceptedBytes += bytes;
+  jsonlWriterState.queuedBytes += bytes;
+  const completion = new Promise((resolve) => jsonlWriterState.queue.push({ file, line, bytes, resolve }));
+  startJsonlPump();
+  return completion;
+}
+
+function recordJsonlDrop(file, bytes) {
+  jsonlWriterState.droppedEntries += 1;
+  jsonlWriterState.droppedBytes += Math.max(0, Number(bytes) || 0);
+  const key = path.basename(String(file || "unknown")).slice(0, 80);
+  jsonlWriterState.dropsByFile[key] = (jsonlWriterState.dropsByFile[key] || 0) + 1;
+}
+
+function startJsonlPump() {
+  if (jsonlWriterState.pump) return jsonlWriterState.pump;
+  jsonlWriterState.pump = (async () => {
+    while (jsonlWriterState.queue.length > 0) {
+      const entry = jsonlWriterState.queue.shift();
+      jsonlWriterState.queuedBytes = Math.max(0, jsonlWriterState.queuedBytes - entry.bytes);
+      jsonlWriterState.inFlight = 1;
+      try {
+        await appendFile(entry.file, entry.line, { mode: 0o600 });
+        jsonlWriterState.writtenEntries += 1;
+        jsonlWriterState.writtenBytes += entry.bytes;
+        entry.resolve({ ok: true, dropped: false, valuesRedacted: true });
+      } catch (error) {
+        jsonlWriterState.failedEntries += 1;
+        entry.resolve({ ok: false, dropped: false, error: errorSummary(error), valuesRedacted: true });
+      } finally {
+        jsonlWriterState.inFlight = 0;
+      }
+    }
+  })().finally(() => {
+    jsonlWriterState.pump = null;
+    if (jsonlWriterState.queue.length > 0) startJsonlPump();
+  });
+  return jsonlWriterState.pump;
+}
+
+function jsonlWriterStats() {
+  return {
+    queueEntries: jsonlWriterState.queue.length,
+    queueBytes: jsonlWriterState.queuedBytes,
+    inFlight: jsonlWriterState.inFlight,
+    maxQueueEntries: maxJsonlQueueEntries,
+    maxQueueBytes: maxJsonlQueueBytes,
+    maxLineBytes: maxJsonlLineBytes,
+    acceptedEntries: jsonlWriterState.acceptedEntries,
+    acceptedBytes: jsonlWriterState.acceptedBytes,
+    writtenEntries: jsonlWriterState.writtenEntries,
+    writtenBytes: jsonlWriterState.writtenBytes,
+    droppedEntries: jsonlWriterState.droppedEntries,
+    droppedBytes: jsonlWriterState.droppedBytes,
+    failedEntries: jsonlWriterState.failedEntries,
+    dropsByFile: { ...jsonlWriterState.dropsByFile },
+    valuesRedacted: true,
+  };
+}
+
+async function flushJsonlQueue() {
+  try {
+    await withHardTimeout((async () => {
+      while (jsonlWriterState.queue.length > 0 || jsonlWriterState.inFlight > 0 || jsonlWriterState.pump) {
+        if (jsonlWriterState.pump) await jsonlWriterState.pump;
+        else if (jsonlWriterState.queue.length > 0) await startJsonlPump();
+        else break;
+      }
+    })(), jsonlFlushTimeoutMs, "jsonl terminal flush exceeded " + jsonlFlushTimeoutMs + "ms");
+    return { ok: true, timedOut: false, stats: jsonlWriterStats(), valuesRedacted: true };
+  } catch (error) {
+    return { ok: false, timedOut: true, error: errorSummary(error), stats: jsonlWriterStats(), valuesRedacted: true };
+  }
 }

 async function fileMeta(file) {
-  const [buffer, stats] = await Promise.all([readFile(file), stat(file)]);
-  return { byteCount: stats.size, sha256: "sha256:" + createHash("sha256").update(buffer).digest("hex") };
+  const digestPromise = new Promise((resolve, reject) => {
+    const digest = createHash("sha256");
+    const stream = createReadStream(file);
+    stream.on("data", (chunk) => digest.update(chunk));
+    stream.once("error", reject);
+    stream.once("end", () => resolve("sha256:" + digest.digest("hex")));
+  });
+  const [sha256, stats] = await Promise.all([digestPromise, stat(file)]);
+  return { byteCount: stats.size, sha256 };
+}
+
+function responseBodyReadContract(response, maxBytes = responseBodyMaxBytes) {
+  let headers = {};
+  try { headers = response?.headers?.() || {}; } catch {}
+  const contentEncoding = String(headers["content-encoding"] || headers["Content-Encoding"] || "").trim().toLowerCase();
+  if (contentEncoding && contentEncoding !== "identity") {
+    return { ok: false, reason: "encoded-body-size-unbounded", contentEncoding: truncate(contentEncoding, 80), contentLength: null, maxBytes, valuesRedacted: true };
+  }
+  const rawContentLength = headers["content-length"] || headers["Content-Length"];
+  const contentLength = Number(rawContentLength);
+  if (!Number.isFinite(contentLength) || contentLength < 0) {
+    return { ok: false, reason: "content-length-unavailable", contentLength: null, maxBytes, valuesRedacted: true };
+  }
+  if (contentLength > maxBytes) {
+    return { ok: false, reason: "content-length-too-large", contentLength, maxBytes, valuesRedacted: true };
+  }
+  return { ok: true, reason: null, contentLength, maxBytes, valuesRedacted: true };
+}
+
+async function boundedResponseJson(response, maxBytes = responseBodyMaxBytes) {
+  const bodyContract = responseBodyReadContract(response, maxBytes);
+  if (bodyContract.ok !== true) return { ok: false, value: null, error: bodyContract, bodyContract, valuesRedacted: true };
+  const permit = tryAcquireResponseBodyPermit();
+  if (!permit) return { ok: false, value: null, error: { reason: "response-body-inflight-saturated", valuesRedacted: true }, bodyContract, valuesRedacted: true };
+  try {
+    const value = await response.json();
+    responseBodyState.completed += 1;
+    return { ok: true, value, error: null, bodyContract, valuesRedacted: true };
+  } catch (error) {
+    responseBodyState.failed += 1;
+    return { ok: false, value: null, error: errorSummary(error), bodyContract, valuesRedacted: true };
+  } finally {
+    permit.release();
+  }
+}
+
+async function boundedResponseText(response, maxBytes = responseBodyMaxBytes) {
+  const bodyContract = responseBodyReadContract(response, maxBytes);
+  if (bodyContract.ok !== true) return { ok: false, value: null, error: bodyContract, bodyContract, valuesRedacted: true };
+  const permit = tryAcquireResponseBodyPermit();
+  if (!permit) return { ok: false, value: null, error: { reason: "response-body-inflight-saturated", valuesRedacted: true }, bodyContract, valuesRedacted: true };
+  try {
+    const value = await response.text();
+    responseBodyState.completed += 1;
+    return { ok: true, value, error: null, bodyContract, valuesRedacted: true };
+  } catch (error) {
+    responseBodyState.failed += 1;
+    return { ok: false, value: null, error: errorSummary(error), bodyContract, valuesRedacted: true };
+  } finally {
+    permit.release();
+  }
+}
+
+function tryAcquireResponseBodyPermit() {
+  if (responseBodyState.inFlight >= maxResponseBodyInFlight) {
+    responseBodyState.skippedSaturated += 1;
+    return null;
+  }
+  responseBodyState.inFlight += 1;
+  responseBodyState.accepted += 1;
+  let released = false;
+  return { release() { if (released) return; released = true; responseBodyState.inFlight = Math.max(0, responseBodyState.inFlight - 1); } };
+}
+
+function responseBodyStats() {
+  return { ...responseBodyState, maxInFlight: maxResponseBodyInFlight, valuesRedacted: true };
+}
+
+function trackPassiveTask(factory, label) {
+  if (stopping || passiveTaskState.tasks.size >= maxPassiveListenerTasks) {
+    passiveTaskState.dropped += 1;
+    return false;
+  }
+  passiveTaskState.accepted += 1;
+  let task;
+  task = Promise.resolve()
+    .then(factory)
+    .then(() => { passiveTaskState.completed += 1; })
+    .catch(async (error) => {
+      passiveTaskState.failed += 1;
+      await appendJsonl(files.errors, eventRecord("passive-task-error", { label, error: errorSummary(error), valuesRedacted: true })).catch(() => {});
+    })
+    .finally(() => passiveTaskState.tasks.delete(task));
+  passiveTaskState.tasks.add(task);
+  task.catch(() => {});
+  return true;
+}
+
+function passiveTaskStats() {
+  return {
+    inFlight: passiveTaskState.tasks.size,
+    maxInFlight: maxPassiveListenerTasks,
+    accepted: passiveTaskState.accepted,
+    dropped: passiveTaskState.dropped,
+    completed: passiveTaskState.completed,
+    failed: passiveTaskState.failed,
+    drainTimedOut: passiveTaskState.drainTimedOut,
+    valuesRedacted: true,
+  };
+}
+
+async function drainPassiveTasks() {
+  try {
+    await withHardTimeout((async () => {
+      while (passiveTaskState.tasks.size > 0) await Promise.allSettled([...passiveTaskState.tasks]);
+    })(), passiveTaskFlushTimeoutMs, "passive task drain exceeded " + passiveTaskFlushTimeoutMs + "ms");
+    return { ok: true, timedOut: false, stats: passiveTaskStats(), valuesRedacted: true };
+  } catch (error) {
+    passiveTaskState.drainTimedOut = true;
+    return { ok: false, timedOut: true, error: errorSummary(error), stats: passiveTaskStats(), valuesRedacted: true };
+  }
 }

 function currentPageUrl() {
@@ -77,6 +285,14 @@ function boundedInteger(value, fallback, min, max) {
  return integer;
 }

+function requiredBoundedInteger(value, name, min, max) {
+  const parsed = Number(value);
+  if (!Number.isInteger(parsed) || parsed < min || parsed > max) {
+    throw new Error(name + " must come from the owning YAML and be an integer between " + min + " and " + max);
+  }
+  return parsed;
+}
+
 function positiveNumber(value, fallback) {
  const parsed = Number(value);
  return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
@@ -306,9 +522,33 @@ function errorSummary(error) {
 }

 function sleep(ms) {
+  return interruptibleRunnerSleep(ms);
+}
+
+function boundedCleanupSleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, Math.max(0, ms)));
 }

+function interruptibleRunnerSleep(ms) {
+  if (stopping) return Promise.resolve();
+  return new Promise((resolve) => {
+    const timer = setTimeout(() => {
+      runnerSleepWakes.delete(wake);
+      resolve();
+    }, Math.max(0, ms));
+    const wake = () => {
+      clearTimeout(timer);
+      runnerSleepWakes.delete(wake);
+      resolve();
+    };
+    runnerSleepWakes.add(wake);
+  });
+}
+
+function interruptRunnerSleep() {
+  for (const wake of [...runnerSleepWakes]) wake();
+}
+
 function withHardTimeout(promise, timeoutMs, message) {
  const guarded = Promise.resolve(promise);
  let timer = null;