fix(entropy): retain completed file outcomes through cleanup

This commit is contained in:
morluto committed 2026-10-23 00:06:00 +08:00
1 parent 6d977db1b5
commit 5bda0a7dc3
2 files changed
+261 -130

No files matched your search

@@ -102,9 +102,14 @@ it("honors cancellation before and during streaming", async () => {
const path = join(root, "bytes");
await writeFile(path, Buffer.alloc(1024 * 1024));
const service = new ByteEntropyService();
expect(
await service.inspect({ path }, { signal: AbortSignal.abort() }),
).toMatchObject({ ok: false, error: { _tag: "AnalysisCancelledError" } });
const reason = { request: "opaque cancellation" };
const before = await service.inspect(
{ path },
{ signal: AbortSignal.abort(reason) },
);
if (before.ok) throw new Error("Expected cancellation before admission");
expect(before.error.cause).toBe(reason);
expect(before.error.partialObservation).toBeUndefined();
const controller = new AbortController();
let reads = 0;
const streaming = new ByteEntropyService({
@@ -115,7 +120,7 @@ it("honors cancellation before and during streaming", async () => {
vi.spyOn(file, "read").mockImplementation(async (...readArgs) => {
const result = await read(...readArgs);
reads += 1;
controller.abort();
controller.abort(reason);
return result;
});
return file;
@@ -125,10 +130,11 @@ it("honors cancellation before and during streaming", async () => {
{ path, block_size: 256 },
{ signal: controller.signal },
);
expect(await pending).toMatchObject({
ok: false,
error: { _tag: "AnalysisCancelledError" },
});
const during = await pending;
if (during.ok) throw new Error("Expected cancellation during read");
expect(during.error).toMatchObject({ _tag: "AnalysisCancelledError" });
expect(during.error.cause).toBe(reason);
expect(during.error.partialObservation).toBeUndefined();
expect(reads).toBe(1);
});
@@ -161,58 +167,84 @@ it("maps cancellation during open to the cancelled algebra with its original cau
await service.close();
});
it("keeps finalizer cancellation primary when close also fails, then resolves its cleanup owner", async () => {
const root = await createTestTempDirectory("rea-entropy-finalizer-cancel-");
const path = join(root, "bytes");
await writeFile(path, Buffer.alloc(512));
const controller = new AbortController();
const reason = new Error("cancelled during close");
const closeFailure = new Error("injected close failure");
let failClose = true;
const service = new ByteEntropyService({
stat,
open: async (...args) => {
const file = await open(...args);
const close = file.close.bind(file);
vi.spyOn(file, "close").mockImplementation(async () => {
if (failClose) {
it.each([false, true])(
"retains completed Evidence when final close is cancelled (close fails: %s)",
async (closeFails) => {
const root = await createTestTempDirectory("rea-entropy-finalizer-cancel-");
const path = join(root, "bytes");
await writeFile(path, Buffer.alloc(512));
const controller = new AbortController();
const reason = new Error("cancelled during close");
const closeFailure = new Error("injected close failure");
let failClose = closeFails;
let closes = 0;
const service = new ByteEntropyService({
stat,
open: async (...args) => {
const file = await open(...args);
const close = file.close.bind(file);
vi.spyOn(file, "close").mockImplementation(async () => {
closes += 1;
controller.abort(reason);
throw closeFailure;
}
await close();
if (failClose) throw closeFailure;
await close();
});
return file;
},
});
const result = await service.inspect(
{ path, block_size: 256 },
{ signal: controller.signal },
);
if (result.ok) throw new Error("Expected cancellation during finalization");
expect(result.error).toMatchObject({
_tag: "AnalysisCancelledError",
cleanupIncomplete: closeFails,
});
expect(result.error.cause).toBe(reason);
const evidence = result.error.partialObservation;
expect(evidence).toMatchObject({
operation: "inspect_byte_entropy",
confidence: "observed",
normalized_result: {
artifact: {
path,
bytes: 512,
sha256: createHash("sha256").update(Buffer.alloc(512)).digest("hex"),
},
window: { offset: 0, length: 512 },
file_entropy: 0,
blocks: [
{ offset: 0, length: 256, entropy: 0 },
{ offset: 256, length: 256, entropy: 0 },
],
},
});
expect(closes).toBe(1);
if (closeFails) {
expect(result.error.secondaryFailures).toMatchObject([
{ _tag: "AnalysisOutputError", cleanupIncomplete: true },
]);
expect(projectAnalysisError(result.error)).toMatchObject({
code: "cleanup_incomplete",
retryable: false,
details: { cleanup: "incomplete", execution_failure: "cancelled" },
});
return file;
},
});
const result = await service.inspect(
{ path, block_size: 256 },
{ signal: controller.signal },
);
if (result.ok) throw new Error("Expected cancellation during finalization");
expect(result.error).toMatchObject({
_tag: "AnalysisCancelledError",
cleanupIncomplete: true,
secondaryFailures: [
{ _tag: "AnalysisOutputError", cleanupIncomplete: true },
],
});
expect(result.error.cause).toBe(reason);
expect(result.error.secondaryFailures[0]?.cause).toBeDefined();
expect(projectAnalysisError(result.error)).toMatchObject({
code: "cleanup_incomplete",
retryable: false,
details: {
cleanup: "incomplete",
execution_failure: "cancelled",
},
});
failClose = false;
await service.close();
expect(result.error.cleanupIncomplete).toBe(false);
});
} else {
expect(result.error.secondaryFailures).toEqual([]);
expect(projectAnalysisError(result.error)).toMatchObject({
code: "cancelled",
});
}
failClose = false;
await service.close();
expect(result.error.cleanupIncomplete).toBe(false);
expect(result.error.partialObservation).toBe(evidence);
expect(closes).toBe(closeFails ? 2 : 1);
},
);
it("reports a changed artifact when its parent directory moves during streaming", async () => {
const root = await createTestTempDirectory("rea-entropy-moved-");
@@ -313,6 +345,11 @@ it("retains a failed descriptor close and retries it before opening another file
_tag: "AnalysisOutputError",
cleanupIncomplete: true,
});
const observation = first.error.partialObservation;
expect(observation).toMatchObject({
operation: "inspect_byte_entropy",
normalized_result: { artifact: { path, bytes: 512 }, file_entropy: 0 },
});
expect(opens).toBe(1);
const retryDenied = await service.inspect({ path, block_size: 256 });
@@ -326,6 +363,7 @@ it("retains a failed descriptor close and retries it before opening another file
const recovered = await service.inspect({ path, block_size: 256 });
expect(recovered.ok).toBe(true);
expect(first.error.cleanupIncomplete).toBe(false);
expect(first.error.partialObservation).toBe(observation);
expect(opens).toBe(2);
await service.close();
});
@@ -371,3 +409,76 @@ it("preserves a primary read failure through failed close and resolves it on ret
expect(first.error.cleanupIncomplete).toBe(false);
await service.close();
});
it.each(["stat", "open"] as const)(
"preserves %s access failure when the signal aborts before rejection is consumed",
async (stage) => {
const root = await createTestTempDirectory("rea-entropy-physical-failure-");
const path = join(root, "bytes");
await writeFile(path, Buffer.alloc(512));
const controller = new AbortController();
const failure = Object.assign(new Error("access denied"), {
code: "EACCES",
});
const reject = async () => {
controller.abort({ request: "late cancellation" });
throw failure;
};
const service = new ByteEntropyService({
stat: stage === "stat" ? reject : stat,
open: stage === "open" ? reject : open,
});
const result = await service.inspect(
{ path },
{ signal: controller.signal },
);
if (result.ok) throw new Error("Expected access failure");
expect(result.error).toMatchObject({ _tag: "AnalysisAccessDeniedError" });
expect(result.error.cause).toBe(failure);
expect(result.error.partialObservation).toBeUndefined();
await service.close();
},
);
it.each([false, true])(
"preserves read access failure through late cancellation (close fails: %s)",
async (closeFails) => {
const root = await createTestTempDirectory("rea-entropy-read-access-");
const path = join(root, "bytes");
await writeFile(path, Buffer.alloc(512));
const controller = new AbortController();
const failure = Object.assign(new Error("read access denied"), {
code: "EACCES",
});
let failClose = closeFails;
const service = new ByteEntropyService({
stat,
open: async (...args) => {
const file = await open(...args);
const close = file.close.bind(file);
vi.spyOn(file, "read").mockImplementation(async () => {
controller.abort({ request: "late cancellation" });
throw failure;
});
vi.spyOn(file, "close").mockImplementation(async () => {
if (failClose) throw new Error("close denied");
await close();
});
return file;
},
});
const result = await service.inspect(
{ path },
{ signal: controller.signal },
);
if (result.ok) throw new Error("Expected read failure");
const primary = result.error.cleanupPrimary ?? result.error;
expect(primary).toMatchObject({ _tag: "AnalysisAccessDeniedError" });
expect(primary.cause).toBe(failure);
expect(result.error.cleanupIncomplete).toBe(closeFails);
expect(result.error.partialObservation).toBeUndefined();
failClose = false;
await service.close();
expect(result.error.cleanupIncomplete).toBe(false);
},
);
@@ -16,22 +16,26 @@ import {
import { createEvidence, type Evidence } from "../../domain/evidence.js";
import { analysisInputErrorFromIssues } from "../../domain/inputIssueProjection.js";
import {
byteEntropySchema,
computeBlockEntropy,
entropyFromCounts,
inspectByteEntropyInputSchema,
mergeEntropyRuns,
type ByteEntropy,
} from "../../domain/native/byteEntropy.js";
import { err, ok, type Result } from "../../domain/result.js";
import { err, type Result } from "../../domain/result.js";
import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
import {
withArtifactFileHandle,
bindArtifactFileCleanup,
ArtifactFileCleanupFailure,
ArtifactFileCancelled,
assertArtifactReadActive,
artifactFileCancellationCause,
} from "../../artifacts/ArtifactFileOperation.js";
import { analysisErrorWithCleanupFailure } from "../../domain/analysisErrorCleanup.js";
import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
import { completeProviderEvidence } from "../ProviderOperationExecution.js";
/** Bound retained result rows before constructing the Evidence representation. */
export const MAX_BYTE_ENTROPY_BLOCKS = 32_768;
@@ -92,7 +96,9 @@ export class ByteEntropyService {
options?: ExecutionOptions,
): Promise<Result<Evidence, AnalysisError>> {
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(OPERATION));
return err(
new AnalysisCancelledError(OPERATION, { cause: options.signal.reason }),
);
const parsed = inspectByteEntropyInputSchema.safeParse(rawInput);
if (!parsed.success)
return err(
@@ -105,7 +111,62 @@ export class ByteEntropyService {
return err(
invalid("path", "Expected an absolute filesystem path on this host."),
);
const evidenceFromReport = (report: ByteEntropy): Evidence =>
createEvidence(
{ path: input.path, format: "file", sha256: report.artifact.sha256 },
BYTE_ENTROPY_PROVIDER,
{
operation: OPERATION,
parameters: {
path: input.path,
block_size: input.block_size,
offset: input.offset,
...(input.length === undefined ? {} : { length: input.length }),
},
result: report,
rawResult: null,
confidence: "observed",
limitations: report.limitations,
locations: [{ kind: "artifact-path", path: input.path }],
},
);
// Cleanup errors erase the callback's generic type. Validate their completed
// value before publishing it as this operation's observation.
const completedObservation = (value: unknown): Evidence | undefined => {
const report = byteEntropySchema.safeParse(value);
return report.success && report.data.artifact.path === input.path
? evidenceFromReport(report.data)
: undefined;
};
let admitted = false;
const readFailure = (cause: unknown): AnalysisError => {
if (cause instanceof AnalysisError) return cause;
if (cause instanceof Error && "code" in cause) {
if (cause.code === "EACCES" || cause.code === "EPERM")
return new AnalysisAccessDeniedError(
OPERATION,
input.path,
cause.code,
{ cause },
);
if (cause.code === "ENOENT" || cause.code === "ENOTDIR")
return admitted
? new AnalysisArtifactChangedError(
OPERATION,
input.path,
"Selected file path changed during entropy inspection.",
{ cause },
)
: invalid("path", "Selected file could not be found.");
}
return new AnalysisOutputError(
OPERATION,
cause instanceof ArtifactReaderFailure
? cause.message
: "Could not read complete byte entropy evidence.",
{ cause },
);
};
try {
const before = await this.files.stat(input.path, { bigint: true });
if (!before.isFile()) throw invalid("path", "Expected a regular file.");
@@ -172,14 +233,14 @@ export class ByteEntropyService {
const hash = createHash("sha256");
let position = 0;
while (position < size) {
options?.signal?.throwIfAborted();
assertArtifactReadActive(options?.signal);
const selected = position >= input.offset && position < end;
const boundary =
position < input.offset ? input.offset : selected ? end : size;
const wanted = Math.min(input.block_size, boundary - position);
let filled = 0;
while (filled < wanted) {
options?.signal?.throwIfAborted();
assertArtifactReadActive(options?.signal);
const { bytesRead } = await file.read(
buffer,
filled,
@@ -205,7 +266,7 @@ export class ByteEntropyService {
!same(await this.files.stat(input.path, { bigint: true }))
)
throw changed();
options?.signal?.throwIfAborted();
assertArtifactReadActive(options?.signal);
const thresholds = { high: 7.2, low: 1.0 };
return {
artifact: {
@@ -228,70 +289,49 @@ export class ByteEntropyService {
};
},
);
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(OPERATION));
return ok(
createEvidence(
{ path: input.path, format: "file", sha256: report.artifact.sha256 },
BYTE_ENTROPY_PROVIDER,
{
operation: OPERATION,
parameters: {
path: input.path,
block_size: input.block_size,
offset: input.offset,
...(input.length === undefined ? {} : { length: input.length }),
},
result: report,
rawResult: null,
confidence: "observed",
limitations: report.limitations,
locations: [{ kind: "artifact-path", path: input.path }],
},
),
return completeProviderEvidence(
evidenceFromReport(report),
options?.signal,
);
} catch (cause) {
if (cause instanceof ArtifactFileCleanupFailure) {
const outcome = cause.fileCleanupFailure.outcome;
const cleanup = new AnalysisOutputError(
OPERATION,
"Could not confirm closing the selected file.",
{ cause, cleanup: cause.cleanup },
{
cause,
cleanup: cause.cleanup,
...(outcome.kind === "completed"
? { partialObservation: completedObservation(outcome.value) }
: {}),
},
);
bindArtifactFileCleanup(cause, cleanup);
const outcome = cause.fileCleanupFailure.outcome;
if (
outcome.kind === "failed" &&
outcome.cause instanceof ArtifactReaderFailure &&
outcome.cause.reason === "cancelled"
) {
const cancelled = new AnalysisCancelledError(OPERATION, {
cause: outcome.cause.cancellation?.reason,
cause: artifactFileCancellationCause(outcome.cause),
cleanup: cause.cleanup,
}).retainSecondaryFailure(cleanup);
bindArtifactFileCleanup(cause, cancelled);
return err(cancelled);
}
if (options?.signal?.aborted && outcome.kind === "completed") {
const cancellation = cause.fileCleanupFailure.completionCancellation;
if (cancellation !== undefined && outcome.kind === "completed") {
const cancelled = new AnalysisCancelledError(OPERATION, {
cause: options.signal.reason,
cause: cancellation.reason,
partialObservation: cleanup.partialObservation,
cleanup: cause.cleanup,
}).retainSecondaryFailure(cleanup);
bindArtifactFileCleanup(cause, cancelled);
return err(cancelled);
}
if (outcome.kind === "failed") {
const primary =
outcome.cause instanceof AnalysisError
? outcome.cause
: outcome.cause instanceof ArtifactReaderFailure
? new AnalysisOutputError(OPERATION, outcome.cause.message, {
cause: outcome.cause,
})
: new AnalysisOutputError(
OPERATION,
"Could not read complete byte entropy evidence.",
{ cause: outcome.cause },
);
const primary = readFailure(outcome.cause);
return err(
analysisErrorWithCleanupFailure(
primary,
@@ -306,6 +346,14 @@ export class ByteEntropyService {
if (cause.reason === "cancelled") {
const cancelled = new AnalysisCancelledError(OPERATION, {
cause: artifactFileCancellationCause(cause),
...(cause instanceof ArtifactFileCancelled &&
cause.fileCancellation.outcome.kind === "completed"
? {
partialObservation: completedObservation(
cause.fileCancellation.outcome.value,
),
}
: {}),
});
bindArtifactFileCleanup(cause, cancelled);
return err(cancelled);
@@ -321,35 +369,7 @@ export class ByteEntropyService {
bindArtifactFileCleanup(cause, retry);
return err(retry);
}
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(OPERATION, { cause }));
if (cause instanceof AnalysisError) return err(cause);
if (cause instanceof Error && "code" in cause) {
if (cause.code === "EACCES" || cause.code === "EPERM")
return err(
new AnalysisAccessDeniedError(OPERATION, input.path, cause.code, {
cause,
}),
);
if (cause.code === "ENOENT" || cause.code === "ENOTDIR")
return err(
admitted
? new AnalysisArtifactChangedError(
OPERATION,
input.path,
"Selected file path changed during entropy inspection.",
{ cause },
)
: invalid("path", "Selected file could not be found."),
);
}
return err(
new AnalysisOutputError(
OPERATION,
"Could not read complete byte entropy evidence.",
{ cause },
),
);
return err(readFailure(cause));
}
}
}