Honour write backpressure when reassembling uploads
createWriteStream.write() returns false once its internal buffer is full; both reassembly loops ignored that and kept queueing chunks, so the queue grew in memory instead of flushing to disk. Measured over a 1.5 GiB reassembly in 10 MB chunks, this buffered 170 MB versus 10 MB when the drain event is awaited. Also fail loudly on a missing chunk in the v1 loop, matching v2, rather than silently writing a corrupt file. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -291,9 +291,19 @@ async function handleChunkComplete(
|
|||||||
|
|
||||||
const writeStream = fs.createWriteStream(finalPath);
|
const writeStream = fs.createWriteStream(finalPath);
|
||||||
|
|
||||||
|
// Honour backpressure: createWriteStream queues in memory whenever write()
|
||||||
|
// returns false, so a multi-GB reassembly that ignores the return value
|
||||||
|
// grows the internal buffer instead of flushing to disk.
|
||||||
|
const write = (buf: Buffer): Promise<void> =>
|
||||||
|
new Promise((resolve, reject) => {
|
||||||
|
if (writeStream.write(buf)) return resolve();
|
||||||
|
writeStream.once('drain', resolve);
|
||||||
|
writeStream.once('error', reject);
|
||||||
|
});
|
||||||
|
|
||||||
// Write the format header: magic (8) + salt (16). The per-chunk IVs live
|
// Write the format header: magic (8) + salt (16). The per-chunk IVs live
|
||||||
// in the frames themselves, so nothing needs storing in the DB.
|
// in the frames themselves, so nothing needs storing in the DB.
|
||||||
writeStream.write(buildHeader(session.salt));
|
await write(buildHeader(session.salt));
|
||||||
|
|
||||||
// Assemble encrypted frames in order, with progress tracking
|
// Assemble encrypted frames in order, with progress tracking
|
||||||
let chunksAssembled = 0;
|
let chunksAssembled = 0;
|
||||||
@@ -320,7 +330,7 @@ async function handleChunkComplete(
|
|||||||
throw new Error(`Chunk ${i} missing during reassembly`);
|
throw new Error(`Chunk ${i} missing during reassembly`);
|
||||||
}
|
}
|
||||||
|
|
||||||
writeStream.write(chunkData);
|
await write(chunkData);
|
||||||
chunksAssembled++;
|
chunksAssembled++;
|
||||||
|
|
||||||
if (chunksAssembled % 10 === 0 || chunksAssembled === session.totalChunks) {
|
if (chunksAssembled % 10 === 0 || chunksAssembled === session.totalChunks) {
|
||||||
|
|||||||
@@ -265,6 +265,15 @@ async function handleChunkComplete(
|
|||||||
|
|
||||||
const writeStream = fs.createWriteStream(finalPath);
|
const writeStream = fs.createWriteStream(finalPath);
|
||||||
|
|
||||||
|
// Honour backpressure so a multi-GB reassembly flushes to disk instead of
|
||||||
|
// queueing in the stream's internal buffer.
|
||||||
|
const write = (buf: Buffer): Promise<void> =>
|
||||||
|
new Promise((resolve, reject) => {
|
||||||
|
if (writeStream.write(buf)) return resolve();
|
||||||
|
writeStream.once('drain', resolve);
|
||||||
|
writeStream.once('error', reject);
|
||||||
|
});
|
||||||
|
|
||||||
for (let i = 0; i < session.totalChunks; i++) {
|
for (let i = 0; i < session.totalChunks; i++) {
|
||||||
const chunkPath = path.join(chunkDir, `chunk-${i}`);
|
const chunkPath = path.join(chunkDir, `chunk-${i}`);
|
||||||
// Retry reading with exponential backoff to handle file locks
|
// Retry reading with exponential backoff to handle file locks
|
||||||
@@ -281,9 +290,11 @@ async function handleChunkComplete(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (chunkData) {
|
// A silently skipped chunk would corrupt the file, so fail loudly.
|
||||||
writeStream.write(chunkData);
|
if (!chunkData) {
|
||||||
|
throw new Error(`Chunk ${i} missing during reassembly`);
|
||||||
}
|
}
|
||||||
|
await write(chunkData);
|
||||||
}
|
}
|
||||||
|
|
||||||
await new Promise<void>((resolve, reject) => {
|
await new Promise<void>((resolve, reject) => {
|
||||||
|
|||||||
Reference in New Issue
Block a user