fix(upload): isolate chunk staging per attempt and close before cleanup
Two races found by a second review round, both in the staging logic added by the previous commit. - Two in-flight sends of the same chunk index shared one `.part` path, so whichever renamed first published bytes the other had already truncated. An acknowledged 10-byte chunk could end up 2 bytes. The staging suffix is now per-attempt rather than per-index. - Unlinking the partial file raced the write stream's pending open(). destroy() does not await it, so the unlink failed with ENOENT and the open then recreated the `.part` file after cleanup had supposedly finished — reported reproducible in 121 of 300 immediately-failing streams. Cleanup now waits for the stream to close. Relates to issue 1403
This commit is contained in:
@@ -157,6 +157,41 @@ describe('chunked upload streams the body under a cap (#1403)', () => {
|
||||
expect(chunkedUpload.getUploadStatus(uploadId).receivedChunks).toBe(1);
|
||||
});
|
||||
|
||||
it('keeps two in-flight sends of the same chunk off each other\'s staging file', async () => {
|
||||
const { uploadId } = await init();
|
||||
const chunkPath = path.join(process.env.STORAGE_PATH, 'chunks', uploadId, 'chunk_000000');
|
||||
|
||||
// Two requests for the same index, overlapping. A shared .part path let
|
||||
// whichever renamed first publish bytes the other had already truncated.
|
||||
const slow = new Readable({ read() {} });
|
||||
const doomed = new Readable({ read() {} });
|
||||
const slowDone = chunkedUpload.uploadChunk(uploadId, 0, slow);
|
||||
const doomedDone = chunkedUpload.uploadChunk(uploadId, 0, doomed);
|
||||
|
||||
doomed.push(Buffer.alloc(2));
|
||||
doomed.destroy(new Error('retry gave up'));
|
||||
await expect(doomedDone).rejects.toThrow();
|
||||
|
||||
slow.push(Buffer.alloc(10));
|
||||
slow.push(null);
|
||||
await expect(slowDone).resolves.toBeTruthy();
|
||||
|
||||
// The surviving attempt's 10 bytes, not the failed one's 2.
|
||||
expect((await fs.stat(chunkPath)).size).toBe(10);
|
||||
});
|
||||
|
||||
it('leaves no staging files behind after a failure', async () => {
|
||||
const { uploadId } = await init();
|
||||
await expect(chunkedUpload.uploadChunk(uploadId, 0, countingSource(8 * MB)))
|
||||
.rejects.toMatchObject({ statusCode: 413 });
|
||||
// The cap path aborts the whole upload, so the directory is gone; what
|
||||
// must not happen is a .part file reappearing after cleanup because the
|
||||
// write stream's open() was still pending when the unlink ran.
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
await expect(fs.readdir(path.join(process.env.STORAGE_PATH, 'chunks', uploadId)))
|
||||
.rejects.toMatchObject({ code: 'ENOENT' });
|
||||
});
|
||||
|
||||
it('does not destroy the request stream when it trips the cap', async () => {
|
||||
const { uploadId } = await init();
|
||||
const source = countingSource(8 * MB);
|
||||
|
||||
@@ -154,8 +154,17 @@ function writeChunkStream(source, partPath, allowance) {
|
||||
settled = true;
|
||||
source.unpipe(out);
|
||||
if (err) {
|
||||
out.destroy();
|
||||
fsSync.unlink(partPath, () => reject(err));
|
||||
// Wait for the descriptor to actually close before unlinking. destroy()
|
||||
// does not await a pending open(), so unlinking straight away races it:
|
||||
// the unlink fails with ENOENT and the open then recreates the .part
|
||||
// file after cleanup was supposed to be done.
|
||||
const removePart = () => fsSync.unlink(partPath, () => reject(err));
|
||||
if (out.destroyed) {
|
||||
removePart();
|
||||
} else {
|
||||
out.once('close', removePart);
|
||||
out.destroy();
|
||||
}
|
||||
} else {
|
||||
resolve(value);
|
||||
}
|
||||
@@ -250,7 +259,11 @@ async function uploadChunk(uploadId, chunkIndex, source, { declaredBytes } = {})
|
||||
// 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`;
|
||||
//
|
||||
// The suffix is per-attempt, not per-index: two in-flight requests for the
|
||||
// same chunk would otherwise share one staging file, and whichever renamed
|
||||
// first would publish bytes the other had already truncated.
|
||||
const partPath = `${chunkPath}.${crypto.randomBytes(6).toString('hex')}.part`;
|
||||
try {
|
||||
chunkLength = await writeChunkStream(source, partPath, allowance);
|
||||
await fs.rename(partPath, chunkPath);
|
||||
|
||||
Reference in New Issue
Block a user