fix(upload): harden the chunk stream against abort, retry and cap failures

Three failure paths the streaming rewrite introduced, all found by external
review. None existed in the buffered version: the async iterator it replaced
rejected a dead request on its own, and never opened the chunk file until it
already held the whole body.

- An already-destroyed request hung the call forever. If the client hangs up
  while auth and ownership are awaiting the database, pipe() emits neither
  `end` nor `error`, so the promise never settled and the write descriptor
  stayed open. Checked up front now, alongside `aborted` and a `close` without
  `readableEnded` for a body cut short mid-flight.

- A failed re-send destroyed the chunk it was replacing. createWriteStream
  truncates on open, so re-sending an index and then failing left
  receivedChunks and chunkSizes still claiming the old copy: status reported
  100% and completeUpload died on ENOENT. Chunks are staged through a sibling
  .part file and renamed only on success.

- Tripping the cap stopped the 413 from reaching the client. `source` is the
  IncomingMessage, so destroying it destroyed the socket under the response
  and the client saw a connection reset instead of the size-limit JSON. The
  read is paused instead, which is all the cap needs.

Relates to issue 1403

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();

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);