diff --git a/backend/__tests__/services/chunkedUploadStreaming.test.js b/backend/__tests__/services/chunkedUploadStreaming.test.js index 8051ccc9..0f120bdd 100644 --- a/backend/__tests__/services/chunkedUploadStreaming.test.js +++ b/backend/__tests__/services/chunkedUploadStreaming.test.js @@ -120,6 +120,55 @@ describe('chunked upload streams the body under a cap (#1403)', () => { }); }); + // Every case here was found by an external review of the first cut of this + // fix. All three are regressions the buffered version did not have: the + // async iterator it replaced rejected a dead request on its own, and never + // opened the chunk file at all until it had the whole body in hand. + describe('failure paths the streaming rewrite introduced', () => { + it('rejects an already-destroyed request instead of hanging forever', async () => { + const { uploadId } = await init(); + const source = countingSource(1024); + source.destroy(); + // pipe() on a dead stream emits neither `end` nor `error`, so without an + // explicit check this promise never settles and the write fd leaks. + await expect(chunkedUpload.uploadChunk(uploadId, 0, source)) + .rejects.toMatchObject({ code: 'CHUNK_PREMATURE_CLOSE', statusCode: 400 }); + }); + + it('leaves a previously banked chunk intact when a re-send fails', async () => { + const { uploadId } = await init(); + await chunkedUpload.uploadChunk(uploadId, 0, Buffer.alloc(1000)); + const chunkPath = path.join(process.env.STORAGE_PATH, 'chunks', uploadId, 'chunk_000000'); + expect((await fs.stat(chunkPath)).size).toBe(1000); + + // Re-send the same index, then fail it mid-flight. + const source = new Readable({ read() {} }); + const pending = chunkedUpload.uploadChunk(uploadId, 0, source); + source.push(Buffer.alloc(10)); + source.destroy(new Error('client went away')); + await expect(pending).rejects.toThrow(); + + // The banked copy must still be there: receivedChunks/chunkSizes still + // count it, so a truncated file here means status reports 100% and + // completeUpload dies on ENOENT. + expect((await fs.stat(chunkPath)).size).toBe(1000); + // Still counted as received — which is exactly why the file has to still + // be there and be the full 1000 bytes. + expect(chunkedUpload.getUploadStatus(uploadId).receivedChunks).toBe(1); + }); + + it('does not destroy the request stream when it trips the cap', async () => { + const { uploadId } = await init(); + const source = countingSource(8 * MB); + await expect(chunkedUpload.uploadChunk(uploadId, 0, source)) + .rejects.toMatchObject({ statusCode: 413 }); + // `source` stands in for the IncomingMessage. Destroying it would take + // the socket down before the route could send its 413 JSON, so the client + // would see a connection reset instead of the error. + expect(source.destroyed).toBe(false); + }); + }); + describe('the happy path still works', () => { it('writes a streamed chunk and reports progress', async () => { const { uploadId } = await init(); diff --git a/backend/src/services/chunkedUploadService.js b/backend/src/services/chunkedUploadService.js index b67eb48e..fe1fe766 100644 --- a/backend/src/services/chunkedUploadService.js +++ b/backend/src/services/chunkedUploadService.js @@ -30,6 +30,17 @@ function fileTooLargeError(maxFileSizeBytes) { return err; } +function prematureCloseError() { + const err = new Error('Request body closed before the chunk was fully received'); + err.code = 'CHUNK_PREMATURE_CLOSE'; + err.statusCode = 400; + return err; +} + +function overAllowanceError() { + return Object.assign(new Error('CHUNK_OVER_ALLOWANCE'), { overAllowance: true }); +} + function invalidChunkError(message) { const err = new Error(message); err.code = 'INVALID_CHUNK'; @@ -116,36 +127,59 @@ async function initializeUpload(options) { } /** - * Stream `source` into `chunkPath`, refusing to write more than `allowance` + * Stream `source` into `partPath`, refusing to write more than `allowance` * bytes (#1403). The cap is the backstop for a request that lies about its * Content-Length or omits it: the moment the running total passes the - * allowance the source is destroyed and the partial file removed, so an - * oversized body costs the allowance rather than its own size. + * allowance the read stops and the partial file is removed, so an oversized + * body costs the allowance rather than its own size. */ -function writeChunkStream(source, chunkPath, allowance) { +function writeChunkStream(source, partPath, allowance) { const fsSync = require('fs'); return new Promise((resolve, reject) => { - const out = fsSync.createWriteStream(chunkPath); - let written = 0; - let failed = null; + // A client that hung up while auth and ownership were awaiting the database + // hands us an already-dead stream. pipe() would then emit neither `end` nor + // `error`, leaving this promise pending forever with the write descriptor + // open. The async-iterator version this replaced rejected that case, so it + // has to be checked explicitly rather than inferred from an event. + if (source.destroyed || source.aborted) { + return reject(prematureCloseError()); + } - const fail = (err) => { - if (failed) return; - failed = err; - source.destroy(); - out.destroy(); - fsSync.unlink(chunkPath, () => reject(err)); + const out = fsSync.createWriteStream(partPath); + let written = 0; + let settled = false; + + const settle = (err, value) => { + if (settled) return; + settled = true; + source.unpipe(out); + if (err) { + out.destroy(); + fsSync.unlink(partPath, () => reject(err)); + } else { + resolve(value); + } }; source.on('data', (buf) => { written += buf.length; if (written > allowance) { - fail(Object.assign(new Error('CHUNK_OVER_ALLOWANCE'), { overAllowance: true })); + // Deliberately NOT source.destroy(). `source` is the IncomingMessage, + // and destroying it destroys the socket under it — the 413 the route is + // about to send would never reach the client, who would see a connection + // reset instead of the size-limit JSON. Pausing stops the read, which is + // the whole point of the cap. + source.pause(); + settle(overAllowanceError()); } }); - source.on('error', fail); - out.on('error', fail); - out.on('finish', () => { if (!failed) resolve(written); }); + source.on('error', settle); + source.on('aborted', () => settle(prematureCloseError())); + source.on('close', () => { + if (!source.readableEnded) settle(prematureCloseError()); + }); + out.on('error', settle); + out.on('finish', () => settle(null, written)); source.pipe(out); }); } @@ -212,8 +246,14 @@ async function uploadChunk(uploadId, chunkIndex, source, { declaredBytes } = {}) await fs.writeFile(chunkPath, source); chunkLength = source.length; } else { + // Staged through a sibling .part file, then renamed. Writing the canonical + // path directly truncates it the moment the stream opens, so a re-sent + // chunk that then failed left receivedChunks/chunkSizes still claiming the + // old copy: status reported 100% and completeUpload died on ENOENT. + const partPath = `${chunkPath}.part`; try { - chunkLength = await writeChunkStream(source, chunkPath, allowance); + chunkLength = await writeChunkStream(source, partPath, allowance); + await fs.rename(partPath, chunkPath); } catch (err) { if (err.overAllowance) { await abortUpload(uploadId);