326 lines
9.7 KiB
TypeScript
326 lines
9.7 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
|
|
import {
|
|
ByteCursor,
|
|
HelperLimitError,
|
|
WORKER_JOB_PROTOCOL,
|
|
checkedByteRange,
|
|
checkedOffsetAdd,
|
|
checksumByteSource,
|
|
createIncrementalChecksum,
|
|
createWorkerJobMessageHandler,
|
|
digestByteSourceHex,
|
|
encodeText,
|
|
iterateByteChunks,
|
|
ownedBytes,
|
|
startWorkerJob,
|
|
WorkerJobTimeoutError,
|
|
type WorkerJobCommand,
|
|
type WorkerJobEndpoint,
|
|
type WorkerJobResponse,
|
|
} from "../../src/helpers";
|
|
|
|
describe("bounded byte access", () => {
|
|
it("checks offset arithmetic before slicing", () => {
|
|
expect(checkedOffsetAdd(4, 5, 9)).toBe(9);
|
|
expect(checkedByteRange(9, 4, 5)).toEqual({
|
|
offset: 4,
|
|
length: 5,
|
|
end: 9,
|
|
});
|
|
expect(() => checkedOffsetAdd(Number.MAX_SAFE_INTEGER, 1)).toThrow(
|
|
HelperLimitError,
|
|
);
|
|
expect(() => checkedByteRange(8, 7, 2)).toThrow(HelperLimitError);
|
|
expect(() => checkedByteRange(8, -1, 1)).toThrow(/non-negative/u);
|
|
});
|
|
|
|
it("reads typed values without moving after a failed read", () => {
|
|
const cursor = new ByteCursor(
|
|
new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x41, 0x42]),
|
|
);
|
|
expect(cursor.readUint16()).toBe(0x0102);
|
|
expect(cursor.readUint16(true)).toBe(0x0403);
|
|
expect(cursor.readAscii(2)).toBe("AB");
|
|
expect(cursor.done).toBe(true);
|
|
expect(() => cursor.readUint8()).toThrow(HelperLimitError);
|
|
expect(cursor.offset).toBe(6);
|
|
});
|
|
|
|
it("supports bounded subcursors and 24-bit integers", () => {
|
|
const cursor = new ByteCursor(new Uint8Array([1, 2, 3, 4]), {
|
|
littleEndian: true,
|
|
maximumBytes: 4,
|
|
});
|
|
expect(cursor.readUint24()).toBe(0x030201);
|
|
expect(cursor.subcursor(1).readUint8()).toBe(4);
|
|
expect(
|
|
() => new ByteCursor(new Uint8Array(5), { maximumBytes: 4 }),
|
|
).toThrow(HelperLimitError);
|
|
});
|
|
|
|
it("does not advance after invalid ASCII and makes bounded owned copies", () => {
|
|
const cursor = new ByteCursor(new Uint8Array([0x41, 0xff]));
|
|
expect(() => cursor.readAscii(2)).toThrow(/ASCII/u);
|
|
expect(cursor.offset).toBe(0);
|
|
const source = new Uint8Array([1, 2]);
|
|
const copy = ownedBytes(source, 2);
|
|
source[0] = 9;
|
|
expect(copy).toEqual(new Uint8Array([1, 2]));
|
|
expect(() => ownedBytes(source, 1)).toThrow(HelperLimitError);
|
|
});
|
|
});
|
|
|
|
describe("incremental byte sources", () => {
|
|
it("produces the same checksums across arbitrary chunk boundaries", async () => {
|
|
async function* chunks() {
|
|
yield encodeText("123");
|
|
yield encodeText("456");
|
|
yield encodeText("789");
|
|
}
|
|
await expect(
|
|
checksumByteSource(chunks(), "CRC-32", { maximumBytes: 9 }),
|
|
).resolves.toBe(0xcbf43926);
|
|
const checksum = createIncrementalChecksum("Adler-32", 9);
|
|
checksum.update(encodeText("1234")).update(encodeText("56789"));
|
|
expect(checksum.digestHex()).toBe("091e01de");
|
|
expect(checksum.bytesProcessed).toBe(9);
|
|
expect(() => checksum.update(new Uint8Array([0]))).toThrow(
|
|
HelperLimitError,
|
|
);
|
|
expect(checksum.reset().bytesProcessed).toBe(0);
|
|
expect(() => createIncrementalChecksum("unknown" as "CRC-32")).toThrow(
|
|
/Unsupported checksum/u,
|
|
);
|
|
});
|
|
|
|
it("hashes bounded Blob chunks and reports progress", async () => {
|
|
const progress: number[] = [];
|
|
await expect(
|
|
digestByteSourceHex(new Blob(["abc"]), "SHA-256", {
|
|
chunkBytes: 2,
|
|
maximumBytes: 3,
|
|
onProgress: (event) => progress.push(event.processedBytes),
|
|
}),
|
|
).resolves.toBe(
|
|
"ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
|
|
);
|
|
expect(progress).toEqual([2, 3]);
|
|
});
|
|
|
|
it("cancels streams and enforces a total ceiling", async () => {
|
|
const stream = new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(new Uint8Array(3));
|
|
controller.enqueue(new Uint8Array(3));
|
|
controller.close();
|
|
},
|
|
});
|
|
const consume = async () => {
|
|
for await (const chunk of iterateByteChunks(stream, {
|
|
maximumBytes: 5,
|
|
})) {
|
|
// Consume until the shared total limit rejects the second chunk.
|
|
void chunk;
|
|
}
|
|
};
|
|
await expect(consume()).rejects.toThrow(HelperLimitError);
|
|
|
|
async function* shortSource() {
|
|
yield new Uint8Array(2);
|
|
}
|
|
const short = async () => {
|
|
for await (const chunk of iterateByteChunks(shortSource(), {
|
|
maximumBytes: 4,
|
|
knownTotalBytes: 3,
|
|
}))
|
|
void chunk;
|
|
};
|
|
await expect(short()).rejects.toThrow(/expected 3/u);
|
|
|
|
const controller = new AbortController();
|
|
controller.abort();
|
|
const aborted = async () => {
|
|
for await (const chunk of iterateByteChunks(new Blob(["x"]), {
|
|
signal: controller.signal,
|
|
})) {
|
|
// No chunk may escape after cancellation.
|
|
void chunk;
|
|
}
|
|
};
|
|
await expect(aborted()).rejects.toMatchObject({ name: "AbortError" });
|
|
});
|
|
});
|
|
|
|
class FakeWorker implements WorkerJobEndpoint {
|
|
readonly posted: unknown[] = [];
|
|
terminated = 0;
|
|
readonly #messages = new Set<(event: MessageEvent<unknown>) => void>();
|
|
readonly #errors = new Set<(event: ErrorEvent) => void>();
|
|
|
|
postMessage(message: unknown): void {
|
|
this.posted.push(message);
|
|
}
|
|
|
|
terminate(): void {
|
|
this.terminated += 1;
|
|
}
|
|
|
|
addEventListener(
|
|
type: "message" | "error",
|
|
listener:
|
|
((event: MessageEvent<unknown>) => void) | ((event: ErrorEvent) => void),
|
|
): void {
|
|
if (type === "message")
|
|
this.#messages.add(listener as (event: MessageEvent<unknown>) => void);
|
|
else this.#errors.add(listener as (event: ErrorEvent) => void);
|
|
}
|
|
|
|
removeEventListener(
|
|
type: "message" | "error",
|
|
listener:
|
|
((event: MessageEvent<unknown>) => void) | ((event: ErrorEvent) => void),
|
|
): void {
|
|
if (type === "message")
|
|
this.#messages.delete(listener as (event: MessageEvent<unknown>) => void);
|
|
else this.#errors.delete(listener as (event: ErrorEvent) => void);
|
|
}
|
|
|
|
respond(message: unknown): void {
|
|
const event = new MessageEvent("message", { data: message });
|
|
for (const listener of this.#messages) listener(event);
|
|
}
|
|
}
|
|
|
|
describe("disposable worker job protocol", () => {
|
|
it("routes matching progress and result messages and disposes the worker", async () => {
|
|
const worker = new FakeWorker();
|
|
const progress = vi.fn();
|
|
const task = startWorkerJob<{ value: number }, number, string>(
|
|
worker,
|
|
{ value: 2 },
|
|
{ jobId: "job-a", onProgress: progress },
|
|
);
|
|
expect(worker.posted[0]).toMatchObject({
|
|
protocol: WORKER_JOB_PROTOCOL,
|
|
type: "run",
|
|
jobId: "job-a",
|
|
});
|
|
worker.respond({
|
|
protocol: WORKER_JOB_PROTOCOL,
|
|
type: "progress",
|
|
jobId: "someone-else",
|
|
progress: "ignored",
|
|
});
|
|
worker.respond({
|
|
protocol: WORKER_JOB_PROTOCOL,
|
|
type: "progress",
|
|
jobId: "job-a",
|
|
progress: "half",
|
|
});
|
|
worker.respond({
|
|
protocol: WORKER_JOB_PROTOCOL,
|
|
type: "result",
|
|
jobId: "job-a",
|
|
result: 4,
|
|
});
|
|
await expect(task.promise).resolves.toBe(4);
|
|
expect(progress).toHaveBeenCalledWith("half");
|
|
expect(worker.terminated).toBe(1);
|
|
});
|
|
|
|
it("hard-stops a worker at its deadline", async () => {
|
|
vi.useFakeTimers();
|
|
try {
|
|
const worker = new FakeWorker();
|
|
const task = startWorkerJob(worker, "work", {
|
|
jobId: "slow",
|
|
timeoutMs: 25,
|
|
});
|
|
const rejected = expect(task.promise).rejects.toBeInstanceOf(
|
|
WorkerJobTimeoutError,
|
|
);
|
|
await vi.advanceTimersByTimeAsync(25);
|
|
await rejected;
|
|
expect(worker.posted.at(-1)).toMatchObject({
|
|
type: "cancel",
|
|
jobId: "slow",
|
|
});
|
|
expect(worker.terminated).toBe(1);
|
|
} finally {
|
|
vi.useRealTimers();
|
|
}
|
|
});
|
|
|
|
it("settles and disposes when posting or progress handling fails", async () => {
|
|
const postingWorker = new FakeWorker();
|
|
postingWorker.postMessage = () => {
|
|
throw new DOMException("Not cloneable", "DataCloneError");
|
|
};
|
|
const postingTask = startWorkerJob(postingWorker, { invalid: true });
|
|
await expect(postingTask.promise).rejects.toMatchObject({
|
|
name: "DataCloneError",
|
|
});
|
|
expect(postingWorker.terminated).toBe(1);
|
|
|
|
const progressWorker = new FakeWorker();
|
|
const progressTask = startWorkerJob<null, number, number>(
|
|
progressWorker,
|
|
null,
|
|
{
|
|
jobId: "progress-error",
|
|
onProgress: () => {
|
|
throw new Error("Progress consumer failed");
|
|
},
|
|
},
|
|
);
|
|
progressWorker.respond({
|
|
protocol: WORKER_JOB_PROTOCOL,
|
|
type: "progress",
|
|
jobId: "progress-error",
|
|
progress: 1,
|
|
});
|
|
await expect(progressTask.promise).rejects.toThrow(
|
|
"Progress consumer failed",
|
|
);
|
|
expect(progressWorker.terminated).toBe(1);
|
|
});
|
|
|
|
it("provides a worker-side progress/result/error adapter", async () => {
|
|
type Response = WorkerJobResponse<number, string, { message: string }>;
|
|
const responses: Response[] = [];
|
|
const handler = createWorkerJobMessageHandler<
|
|
number,
|
|
number,
|
|
string,
|
|
{ message: string }
|
|
>(
|
|
(payload: number, context) => {
|
|
context.report("working");
|
|
return payload * 2;
|
|
},
|
|
(response) => responses.push(response),
|
|
{
|
|
serializeError: (error) => ({
|
|
message: error instanceof Error ? error.message : String(error),
|
|
}),
|
|
},
|
|
);
|
|
handler(
|
|
new MessageEvent<WorkerJobCommand<number>>("message", {
|
|
data: {
|
|
protocol: WORKER_JOB_PROTOCOL,
|
|
type: "run",
|
|
jobId: "worker-side",
|
|
payload: 3,
|
|
},
|
|
}),
|
|
);
|
|
await vi.waitFor(() => expect(responses).toHaveLength(2));
|
|
expect(responses).toEqual([
|
|
expect.objectContaining({ type: "progress", progress: "working" }),
|
|
expect.objectContaining({ type: "result", result: 6 }),
|
|
]);
|
|
});
|
|
});
|