fix(process): own bounded worker control framing
Share incremental replacement UTF-8 decoding across web and JavaScript workers, bound content before complete-line allocation and handle split CRLF consistently. Keep named schemas, request versus checkpoint failures, worker reuse and stream cleanup authority in their servers. Cover real server recovery and deterministic byte boundaries.
This commit is contained in:
1 parent
f91b7f8009
commit
34563fcc9b
6 files changed
+666
-65
No files matched your search
@@ -5,8 +5,11 @@ import type { JavaScriptSemanticProjectionCapacity } from "../../domain/javascri
|
||||
import { createHash } from "node:crypto";
|
||||
import { readFile, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { createInterface } from "node:readline";
|
||||
import { getHeapStatistics } from "node:v8";
|
||||
import {
|
||||
readBoundedControlLines,
|
||||
type BoundedControlLine,
|
||||
} from "../../process/BoundedControlLines.js";
|
||||
import { AnalysisResourceConstraintError } from "../../domain/analysisErrorCore.js";
|
||||
import { javaScriptRuntime } from "../../domain/runtimeVersion.js";
|
||||
import type { JavaScriptArtifactFile } from "../../domain/javascript/javascriptArtifactFiles.js";
|
||||
@@ -63,8 +66,10 @@ export const serveJavaScriptAnalysisWorker = async (
|
||||
analyze: JavaScriptSourceWorkerAnalyzer,
|
||||
): Promise<void> => {
|
||||
const root = process.cwd();
|
||||
const input = createInterface({ input: process.stdin, crlfDelay: Infinity });
|
||||
const requests = input[Symbol.asyncIterator]();
|
||||
const requests = readBoundedControlLines(
|
||||
process.stdin.iterator({ destroyOnReturn: false }),
|
||||
DEFAULTS.protocolBytes,
|
||||
);
|
||||
await send({
|
||||
kind: "ready",
|
||||
heap_limit_bytes:
|
||||
@@ -76,12 +81,11 @@ export const serveJavaScriptAnalysisWorker = async (
|
||||
for (;;) {
|
||||
const next = await requests.next();
|
||||
if (next.done) break;
|
||||
const line = next.value;
|
||||
if (Buffer.byteLength(line) > DEFAULTS.protocolBytes)
|
||||
if (!next.value.ok)
|
||||
throw new RangeError(
|
||||
"JavaScript worker request exceeds its byte budget",
|
||||
);
|
||||
const raw: unknown = JSON.parse(line);
|
||||
const raw: unknown = JSON.parse(next.value.value);
|
||||
const request = javaScriptWorkerRequestSchema.parse(raw);
|
||||
// Function scope releases source AST/IR/results before accepting another job.
|
||||
await analyzeJob(root, request, analyze, requests);
|
||||
@@ -92,7 +96,7 @@ export const serveJavaScriptAnalysisWorker = async (
|
||||
global.gc?.();
|
||||
}
|
||||
} finally {
|
||||
input.close();
|
||||
await requests.return(undefined);
|
||||
}
|
||||
};
|
||||
|
||||
@@ -100,7 +104,7 @@ const analyzeJob = async (
|
||||
root: string,
|
||||
request: ReturnType<typeof javaScriptWorkerRequestSchema.parse>,
|
||||
analyze: JavaScriptSourceWorkerAnalyzer,
|
||||
requests: AsyncIterator<string>,
|
||||
requests: AsyncIterator<BoundedControlLine>,
|
||||
): Promise<void> => {
|
||||
try {
|
||||
let resultBudgetBytes = 0;
|
||||
@@ -166,14 +170,11 @@ const analyzeJob = async (
|
||||
sha256: createHash("sha256").update(text).digest("hex"),
|
||||
});
|
||||
const next = await requests.next();
|
||||
if (
|
||||
next.done ||
|
||||
Buffer.byteLength(next.value) > DEFAULTS.protocolBytes
|
||||
)
|
||||
if (next.done || !next.value.ok)
|
||||
throw new TypeError(
|
||||
"JavaScript worker admission frame is missing or oversized",
|
||||
);
|
||||
const raw: unknown = JSON.parse(next.value);
|
||||
const raw: unknown = JSON.parse(next.value.value);
|
||||
const admission = javaScriptWorkerAdmissionSchema.parse(raw);
|
||||
if (admission.id !== request.id)
|
||||
throw new TypeError(
|
||||
|
||||
Reference in new issue
Block a user