57 lines
2.5 KiB
TypeScript
57 lines
2.5 KiB
TypeScript
// The upstream tee allows the faster hash reader to queue ahead of disk writes.
|
|
// Keep upload memory bounded by acknowledging each write before hashing/pulling
|
|
// the next chunk. No new dependencies, workers or subprocess permissions.
|
|
export function patchUploadWorker(source: string): string {
|
|
const before = ` // Split the stream into two independent branches:
|
|
// s1 → digest() — consumed by the @std/crypto WASM DigestContext
|
|
// s2 → pipeTo() — written to the temp file on disk
|
|
//
|
|
// digest("SHA-256", s1) uses the AsyncIterable branch of @std/crypto:
|
|
// for await (const chunk of s1) { context.update(chunk) }
|
|
// Hash state is constant ~104 bytes; no chunk accumulation occurs.
|
|
//
|
|
// The size counter runs as a TransformStream on s2 so it never touches s1.
|
|
// Both branches are driven concurrently by Promise.all(). The event loop
|
|
// interleaves them cooperatively at every chunk boundary. Under disk
|
|
// backpressure, the tee internal queue rate-limits s1 to match disk speed —
|
|
// correct behaviour, not a deadlock.
|
|
const [s1, s2] = stream.tee();
|
|
|
|
let totalSize = 0;
|
|
const countingTransform = new TransformStream<Uint8Array, Uint8Array>({
|
|
transform(chunk, controller) {
|
|
totalSize += chunk.byteLength;
|
|
_bytesThisWindow += chunk.byteLength;
|
|
controller.enqueue(chunk);
|
|
},
|
|
});
|
|
|
|
const [hashBuffer] = await Promise.all([
|
|
stdCrypto.subtle.digest(
|
|
"SHA-256",
|
|
s1 as ReadableStream<Uint8Array<ArrayBuffer>>,
|
|
),
|
|
s2.pipeThrough(countingTransform).pipeTo(file.writable),
|
|
]);
|
|
file = null; // writable closed by pipeTo`;
|
|
if (source.split(before).length !== 2) throw new Error('Upstream upload worker integration changed');
|
|
return source.replace(before, ` let totalSize = 0;
|
|
const output = file;
|
|
async function* persistChunks(): AsyncGenerator<Uint8Array<ArrayBuffer>> {
|
|
for await (const chunk of stream) {
|
|
let offset = 0;
|
|
while (offset < chunk.byteLength) {
|
|
const written = await output.write(chunk.subarray(offset));
|
|
if (written <= 0) throw new Error("Upload write made no progress");
|
|
offset += written;
|
|
}
|
|
totalSize += chunk.byteLength;
|
|
_bytesThisWindow += chunk.byteLength;
|
|
yield chunk as Uint8Array<ArrayBuffer>;
|
|
}
|
|
}
|
|
const hashBuffer = await stdCrypto.subtle.digest("SHA-256", persistChunks());
|
|
output.close();
|
|
file = null;`);
|
|
}
|