mirror of
https://github.com/morluto/rea.git
synced 2026-11-15 03:02:56 +00:00
fix(ghidra): retain scoped failures through process finalization
Drain output within caller bounds at asynchronous completion owners and classify JVM exhaustion from the failing operation only. Preserve raw-byte scope, credential redaction, configured heap provenance, cancellation and exact cleanup failures. Validation uses controlled processes and provider fixtures; native Ghidra OOM execution was not repeated. Refs #4241, #4275 Co-authored-by: Syed Rishad <rishadzzzz@gmail.com>
This commit is contained in:
1 parent
43bc6e1d23
commit
d52a488476
16 files changed
+1187
-75
No files matched your search
@@ -126,6 +126,28 @@ export interface ProviderProcessSnapshot {
|
||||
readonly outputIncomplete?: true;
|
||||
}
|
||||
|
||||
declare const processOutputCheckpointBrand: unique symbol;
|
||||
/** Opaque cursor into one supervisor's raw, byte-counted output history. */
|
||||
export interface ProviderProcessOutputCheckpoint {
|
||||
readonly [processOutputCheckpointBrand]: true;
|
||||
}
|
||||
|
||||
/** Output produced after a checkpoint; incomplete when bytes were not retained or drained. */
|
||||
export interface ProviderProcessOutputSince {
|
||||
readonly snapshot: ProviderProcessSnapshot;
|
||||
readonly complete: boolean;
|
||||
}
|
||||
|
||||
const outputCheckpoints = new WeakMap<
|
||||
ProviderProcessOutputCheckpoint,
|
||||
{
|
||||
readonly owner: ProviderProcessSupervisor;
|
||||
readonly epoch: number;
|
||||
readonly stdout: number;
|
||||
readonly stderr: number;
|
||||
}
|
||||
>();
|
||||
|
||||
/** Resource-safe events emitted without imposing provider protocol semantics. */
|
||||
export type ProviderProcessDiagnostic =
|
||||
| {
|
||||
@@ -313,6 +335,7 @@ export class ProviderProcessSupervisor {
|
||||
#stopPromise: Promise<ProviderProcessStopResult> | undefined;
|
||||
#disposed = false;
|
||||
#maxDiagnosticBytes: number | undefined;
|
||||
#outputEpoch = 0;
|
||||
|
||||
constructor(
|
||||
readonly launch: ProviderProcessLaunch,
|
||||
@@ -358,8 +381,75 @@ export class ProviderProcessSupervisor {
|
||||
};
|
||||
}
|
||||
|
||||
/** Mark the current raw-byte position without copying or retaining output. */
|
||||
checkpointOutput(): ProviderProcessOutputCheckpoint {
|
||||
const checkpoint = Object.freeze({}) as ProviderProcessOutputCheckpoint;
|
||||
outputCheckpoints.set(checkpoint, {
|
||||
owner: this,
|
||||
epoch: this.#outputEpoch,
|
||||
stdout: this.#stdoutObservedBytes,
|
||||
stderr: this.#stderrObservedBytes,
|
||||
});
|
||||
return checkpoint;
|
||||
}
|
||||
|
||||
/** Snapshot only bytes produced after a valid checkpoint on this supervisor. */
|
||||
snapshotOutputSince(
|
||||
checkpoint: ProviderProcessOutputCheckpoint,
|
||||
): ProviderProcessOutputSince {
|
||||
const position =
|
||||
typeof checkpoint === "object" && checkpoint !== null
|
||||
? outputCheckpoints.get(checkpoint)
|
||||
: undefined;
|
||||
if (
|
||||
position === undefined ||
|
||||
position.owner !== this ||
|
||||
position.epoch !== this.#outputEpoch
|
||||
)
|
||||
return {
|
||||
snapshot: {
|
||||
stdout: { text: "", bytes: 0, observedBytes: 0 },
|
||||
stderr: { text: "", bytes: 0, observedBytes: 0 },
|
||||
exitCode: this.#exitObservation?.code,
|
||||
signal: this.#exitObservation?.signal,
|
||||
outputIncomplete: true,
|
||||
diagnosticTruncated: true,
|
||||
},
|
||||
complete: false,
|
||||
};
|
||||
const stdout = this.#stdout.snapshotRange(
|
||||
position.stdout,
|
||||
this.#stdoutObservedBytes,
|
||||
);
|
||||
const stderr = this.#stderr.snapshotRange(
|
||||
position.stderr,
|
||||
this.#stderrObservedBytes,
|
||||
);
|
||||
const incomplete =
|
||||
stdout.complete === false ||
|
||||
stderr.complete === false ||
|
||||
this.launch.process.outputIncomplete === true;
|
||||
return {
|
||||
snapshot: {
|
||||
...(incomplete || this.#diagnosticTruncated
|
||||
? { diagnosticTruncated: true }
|
||||
: {}),
|
||||
stdout: stdout.stream,
|
||||
stderr: stderr.stream,
|
||||
exitCode: this.#exitObservation?.code,
|
||||
signal: this.#exitObservation?.signal,
|
||||
...(this.launch.process.memoryLimitReached === true
|
||||
? { memoryLimitReached: true }
|
||||
: {}),
|
||||
...(incomplete ? { outputIncomplete: true } : {}),
|
||||
},
|
||||
complete: !incomplete && !this.#diagnosticTruncated,
|
||||
};
|
||||
}
|
||||
|
||||
/** Begin a new observation on a retained process without retaining earlier output. */
|
||||
resetOutput(): void {
|
||||
this.#outputEpoch += 1;
|
||||
this.#stdout.reset();
|
||||
this.#stderr.reset();
|
||||
this.#stdoutObservedBytes = 0;
|
||||
@@ -377,11 +467,15 @@ export class ProviderProcessSupervisor {
|
||||
}
|
||||
|
||||
/** Wait until output streams close; WindowsOwnedProcess closes only after its whole Job Object settles. */
|
||||
async waitForOutputClose(timeoutMs: number): Promise<boolean> {
|
||||
async waitForOutputClose(
|
||||
timeoutMs: number,
|
||||
signal?: AbortSignal,
|
||||
): Promise<boolean> {
|
||||
return waitForProcessEvent(
|
||||
this.#outputClosed,
|
||||
this.#outputClose.promise,
|
||||
timeoutMs,
|
||||
signal,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -627,7 +721,39 @@ class ProcessOutputCapture {
|
||||
}
|
||||
|
||||
snapshot(observedBytes: number): ProviderProcessSnapshot["stdout"] {
|
||||
const buffer = Buffer.concat(this.#chunks, this.#bytes);
|
||||
return this.#snapshotBuffer(
|
||||
Buffer.concat(this.#chunks, this.#bytes),
|
||||
observedBytes,
|
||||
);
|
||||
}
|
||||
|
||||
snapshotRange(
|
||||
start: number,
|
||||
end: number,
|
||||
): { stream: ProviderProcessSnapshot["stdout"]; complete: boolean } {
|
||||
const availableEnd = Math.min(end, this.#bytes);
|
||||
const availableStart = Math.min(start, availableEnd);
|
||||
const chunks: Buffer[] = [];
|
||||
let offset = 0;
|
||||
for (const chunk of this.#chunks) {
|
||||
const chunkEnd = offset + chunk.byteLength;
|
||||
const from = Math.max(availableStart, offset);
|
||||
const to = Math.min(availableEnd, chunkEnd);
|
||||
if (to > from) chunks.push(chunk.subarray(from - offset, to - offset));
|
||||
offset = chunkEnd;
|
||||
if (offset >= availableEnd) break;
|
||||
}
|
||||
const buffer = Buffer.concat(chunks, availableEnd - availableStart);
|
||||
return {
|
||||
stream: this.#snapshotBuffer(buffer, Math.max(0, end - start)),
|
||||
complete: start <= this.#bytes && end <= this.#bytes,
|
||||
};
|
||||
}
|
||||
|
||||
#snapshotBuffer(
|
||||
buffer: Buffer,
|
||||
observedBytes: number,
|
||||
): ProviderProcessSnapshot["stdout"] {
|
||||
const text = buffer.toString("utf8");
|
||||
let invalidUtf8 = false;
|
||||
try {
|
||||
@@ -637,7 +763,7 @@ class ProcessOutputCapture {
|
||||
}
|
||||
return {
|
||||
text,
|
||||
bytes: this.#bytes,
|
||||
bytes: buffer.byteLength,
|
||||
observedBytes,
|
||||
...(invalidUtf8 ? { invalidUtf8: true } : {}),
|
||||
};
|
||||
@@ -680,17 +806,25 @@ const waitForProcessEvent = async (
|
||||
observed: boolean,
|
||||
event: Promise<unknown>,
|
||||
timeoutMs: number,
|
||||
signal?: AbortSignal,
|
||||
): Promise<boolean> => {
|
||||
if (observed) return true;
|
||||
if (signal?.aborted === true) return false;
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
let onAbort: (() => void) | undefined;
|
||||
try {
|
||||
return await Promise.race([
|
||||
event.then(() => true),
|
||||
new Promise<boolean>((resolve) => {
|
||||
timer = setTimeout(() => resolve(false), Math.max(0, timeoutMs));
|
||||
if (signal !== undefined) {
|
||||
onAbort = () => resolve(false);
|
||||
signal.addEventListener("abort", onAbort, { once: true });
|
||||
}
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timer !== undefined) clearTimeout(timer);
|
||||
if (onAbort !== undefined) signal?.removeEventListener("abort", onAbort);
|
||||
}
|
||||
};
|
||||
Reference in new issue
Block a user