fix(upload): stop buffering a chunk body before anything checks its size (#1406)

* fix(upload): stop buffering a chunk body before anything checks its size The chunk route drained the whole request into an array and concatenated it before calling uploadChunk — which is where every check lives. So a 300MB body sent against an unknown upload id was read in full, cost ~300MB of heap, and was only then answered with an error. The per-file cap was real but applied after the damage, and nothing looked at Content-Length at all. uploadChunk now takes the request stream itself and consumes nothing until the upload id, the chunk index and the declared Content-Length have all been checked against the remaining allowance. A body that clears those is streamed straight to the chunk file under a hard byte cap, so a sender that lies about its length — or sends none, the Transfer-Encoding: chunked case — is cut off at the allowance instead of being read to the end. A Buffer is still accepted, so the existing callers and chunkedUploadSizeCap tests are untouched. Worth noting the old code silently did nothing when handed a stream: fs.promises.writeFile accepts an async iterable, so the body was written while `chunkData.length` was undefined and the cap comparison was NaN > max, i.e. always false. Reachable only for an authenticated admin holding photos.upload on an event they own, and only once the CSRF gate lets application/octet-stream through. Relates to issue 1403 * 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 * 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 * fix(upload): revalidate the per-file cap before publishing a chunk `allowance` is computed before the body arrives, so a chunk that completed while this one was still streaming was not counted in it. Two overlapping 0.75MB chunks under a 1MB cap were therefore both accepted, leaving 1.5MB on disk; enough concurrent streams could go well past the cap before anyone asked to complete the upload. The buffered version got this right for free, because it only ever checked after reading the whole body. The streaming version keeps its pre-read check — that is what makes a too-large request cheap — and asks again with the current aggregate before renaming the staging file into place. Relates to issue 1403 * fix(upload): answer client-caused chunk states with their own status code An unknown, finished or expired upload id, and a complete call with chunks missing, were plain Errors with no statusCode, so both routes fell through to the blanket 500 and logged at error level. All four are the client's mistake: they read as a backend fault in monitoring and invite a retry that can never succeed. They now carry 404, 409, 410 and 400 respectively, and both routes pass a tagged status through instead of matching on the two they happened to know about. Only genuinely unexpected errors reach the 500 and the error log. No client is affected: uploadLargeFile, the only caller of this endpoint family, still has no callers of its own. Relates to issue 1403 * fix(upload): retire the connection after an early refusal, clean up a failed publish Two more from review. Refusing a body before reading it is the point of the streaming cap, but the unread bytes are still in flight on a connection the response advertises as keep-alive. Node does not drain them, so the NEXT request on that socket hangs until it times out — reproducible with an 8MB body against a 1MB cap. The error response now sets Connection: close whenever the request was not read to the end. A failed rename — ENOSPC, a vanished directory — left the fully written staging file behind. Staging names are per-attempt, so a client that retries instead of aborting accumulates one per try until the upload expires. The partial is removed on that path too. Relates to issue 1403 --------- Co-authored-by: Paul Nothaft <paul@MacStudio-von-Paul.local>
2026-09-11 10:41:50 +02:00
parent f92d4bb2d9
commit 4622478e44
3 changed files with 480 additions and 23 deletions
@@ -0,0 +1,294 @@
+/**
+ * A rejected chunk must not cost its own size in memory (#1403).
+ *
+ * The route used to drain the whole request into an array and `Buffer.concat`
+ * it before calling uploadChunk, which is where every check lives — the
+ * per-file cap, the chunk index, and even "does this upload id exist". So a
+ * 300MB body against an unknown upload id was read in full, added ~300MB to
+ * RSS, and was then answered with an error. The size cap was real but only
+ * applied after the damage.
+ *
+ * The contract these tests pin: uploadChunk consumes NOTHING until every check
+ * has passed, and once it does start reading it stops at the remaining
+ * allowance rather than trusting the sender.
+ */
+const path = require('path');
+const os = require('os');
+const fs = require('fs').promises;
+const { Readable } = require('stream');
+
+process.env.STORAGE_PATH = path.join(os.tmpdir(), `picpeak-chunk-stream-test-${process.pid}`);
+
+const chunkedUpload = require('../../src/services/chunkedUploadService');
+
+const MB = 1024 * 1024;
+
+const init = (overrides = {}) => chunkedUpload.initializeUpload({
+  filename: 'clip.mp4',
+  fileSize: 1,
+  mimeType: 'video/mp4',
+  eventId: 1,
+  totalChunks: 2,
+  maxFileSizeBytes: 1 * MB,
+  ...overrides,
+});
+
+/**
+ * A readable that reports how much of it was actually pulled. Bytes are
+ * generated lazily, so "never read" really means the body never materialized.
+ */
+function countingSource(totalBytes, sliceSize = 64 * 1024) {
+  let remaining = totalBytes;
+  const source = new Readable({
+    read() {
+      if (remaining <= 0) return this.push(null);
+      const n = Math.min(sliceSize, remaining);
+      remaining -= n;
+      source.bytesRead += n;
+      this.push(Buffer.alloc(n));
+    },
+  });
+  source.bytesRead = 0;
+  return source;
+}
+
+describe('chunked upload streams the body under a cap (#1403)', () => {
+  afterAll(async () => {
+    await fs.rm(process.env.STORAGE_PATH, { recursive: true, force: true }).catch(() => {});
+  });
+
+  describe('refused before the body is read', () => {
+    it('reads nothing for an unknown upload id', async () => {
+      const source = countingSource(8 * MB);
+      await expect(chunkedUpload.uploadChunk('does-not-exist', 0, source, { declaredBytes: 8 * MB }))
+        .rejects.toThrow('Upload not found or expired');
+      expect(source.bytesRead).toBe(0);
+    });
+
+    it('reads nothing for an out-of-range chunk index', async () => {
+      const { uploadId } = await init();
+      const source = countingSource(8 * MB);
+      await expect(chunkedUpload.uploadChunk(uploadId, 99, source, { declaredBytes: 8 * MB }))
+        .rejects.toMatchObject({ code: 'INVALID_CHUNK', statusCode: 400 });
+      expect(source.bytesRead).toBe(0);
+    });
+
+    it('reads nothing when Content-Length already exceeds the cap', async () => {
+      const { uploadId } = await init();
+      const source = countingSource(8 * MB);
+      await expect(chunkedUpload.uploadChunk(uploadId, 0, source, { declaredBytes: 8 * MB }))
+        .rejects.toMatchObject({ code: 'FILE_TOO_LARGE', statusCode: 413 });
+      expect(source.bytesRead).toBe(0);
+      expect(chunkedUpload.getUploadStatus(uploadId)).toBeNull();
+    });
+
+    it('counts what earlier chunks already banked when checking Content-Length', async () => {
+      const { uploadId } = await init();
+      await chunkedUpload.uploadChunk(uploadId, 0, Buffer.alloc(0.75 * MB));
+      const source = countingSource(0.5 * MB);
+      // 0.75MB banked + 0.5MB declared > the 1MB cap.
+      await expect(chunkedUpload.uploadChunk(uploadId, 1, source, { declaredBytes: 0.5 * MB }))
+        .rejects.toMatchObject({ code: 'FILE_TOO_LARGE', statusCode: 413 });
+      expect(source.bytesRead).toBe(0);
+    });
+  });
+
+  describe('a sender that lies, or says nothing', () => {
+    it('stops at the allowance instead of reading the whole body', async () => {
+      const { uploadId } = await init();
+      // No declaredBytes at all — the Transfer-Encoding: chunked case.
+      const source = countingSource(8 * MB);
+      await expect(chunkedUpload.uploadChunk(uploadId, 0, source))
+        .rejects.toMatchObject({ code: 'FILE_TOO_LARGE', statusCode: 413 });
+      // The overshoot is whatever the readable had already buffered ahead when
+      // the cap tripped — a small constant tied to highWaterMark, NOT a
+      // function of the body size. That is the whole claim: 8MB offered, ~1MB
+      // read. The slack is deliberately loose so this doesn't turn into a
+      // Node-version canary.
+      expect(source.bytesRead).toBeLessThan(2 * MB);
+    });
+
+    it('leaves no partial chunk file behind when it cuts a body off', async () => {
+      const { uploadId } = await init();
+      const meta = chunkedUpload.getUploadStatus(uploadId);
+      await expect(chunkedUpload.uploadChunk(uploadId, 0, countingSource(8 * MB)))
+        .rejects.toMatchObject({ statusCode: 413 });
+      // abortUpload removes the whole directory; assert nothing survived it.
+      await expect(fs.readdir(path.join(process.env.STORAGE_PATH, 'chunks', uploadId)))
+        .rejects.toMatchObject({ code: 'ENOENT' });
+      expect(meta).not.toBeNull();
+    });
+  });
+
+  // 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('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('counts chunks that landed while another was still streaming', async () => {
+      const { uploadId } = await init({ totalChunks: 3 });
+
+      // Start a slow 0.75MB chunk. Its allowance is computed now, when nothing
+      // else is banked.
+      const slow = new Readable({ read() {} });
+      const slowDone = chunkedUpload.uploadChunk(uploadId, 0, slow);
+
+      // A second 0.75MB chunk completes in the meantime.
+      await chunkedUpload.uploadChunk(uploadId, 1, Buffer.alloc(0.75 * MB));
+
+      // Finishing the first must not publish: 1.5MB against a 1MB cap.
+      slow.push(Buffer.alloc(0.75 * MB));
+      slow.push(null);
+      await expect(slowDone).rejects.toMatchObject({ code: 'FILE_TOO_LARGE', statusCode: 413 });
+      expect(chunkedUpload.getUploadStatus(uploadId)).toBeNull();
+    });
+
+    it('removes the staging file when publishing it fails', async () => {
+      const { uploadId } = await init();
+      const dir = path.join(process.env.STORAGE_PATH, 'chunks', uploadId);
+      // Make the rename fail by putting a directory where the chunk goes.
+      await fs.mkdir(path.join(dir, 'chunk_000000'), { recursive: true });
+
+      await expect(chunkedUpload.uploadChunk(uploadId, 0, countingSource(1024)))
+        .rejects.toThrow();
+
+      // The fully written .part must not survive a failed publish — its name is
+      // per-attempt, so retries would otherwise pile them up until expiry.
+      const leftovers = (await fs.readdir(dir)).filter((f) => f.endsWith('.part'));
+      expect(leftovers).toEqual([]);
+    });
+
+    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);
+    });
+  });
+
+  // These were plain Errors, so the routes answered 500 for what are plainly
+  // client mistakes — a backend fault in monitoring, and an invitation to
+  // retry something that can never succeed.
+  describe('client-caused states carry their own status code', () => {
+    it('404s an unknown upload id rather than 500', async () => {
+      await expect(chunkedUpload.uploadChunk('does-not-exist', 0, Buffer.alloc(10)))
+        .rejects.toMatchObject({ statusCode: 404 });
+    });
+
+    // 409 (wrong status) and 410 (expired) share uploadStateError with the two
+    // covered here. Reaching them from the public surface needs either a clock
+    // or a setter the service does not expose, and a test that pretends to
+    // exercise them while actually hitting the 404 path is worse than none.
+
+    it('404s completing an unknown upload rather than 500', async () => {
+      await expect(chunkedUpload.completeUpload('does-not-exist'))
+        .rejects.toMatchObject({ statusCode: 404 });
+    });
+
+    it('400s completing an upload that is missing chunks', async () => {
+      const { uploadId } = await init();
+      await chunkedUpload.uploadChunk(uploadId, 0, Buffer.alloc(10));
+      await expect(chunkedUpload.completeUpload(uploadId))
+        .rejects.toMatchObject({ statusCode: 400 });
+    });
+  });
+
+  describe('the happy path still works', () => {
+    it('writes a streamed chunk and reports progress', async () => {
+      const { uploadId } = await init();
+      const result = await chunkedUpload.uploadChunk(uploadId, 0, countingSource(0.25 * MB), {
+        declaredBytes: 0.25 * MB,
+      });
+      expect(result).toMatchObject({ chunkIndex: 0, received: 1, expected: 2, complete: false });
+
+      const chunkPath = path.join(process.env.STORAGE_PATH, 'chunks', uploadId, 'chunk_000000');
+      expect((await fs.stat(chunkPath)).size).toBe(0.25 * MB);
+    });
+
+    it('still accepts a Buffer, the shape the service was written for', async () => {
+      const { uploadId } = await init();
+      const result = await chunkedUpload.uploadChunk(uploadId, 0, Buffer.alloc(0.25 * MB));
+      expect(result).toMatchObject({ chunkIndex: 0, received: 1 });
+    });
+
+    it('lets a re-sent chunk replace itself without double-counting', async () => {
+      const { uploadId } = await init();
+      await chunkedUpload.uploadChunk(uploadId, 0, countingSource(0.6 * MB), { declaredBytes: 0.6 * MB });
+      // Same index again: the first copy's 0.6MB must not count toward the cap.
+      await expect(
+        chunkedUpload.uploadChunk(uploadId, 0, countingSource(0.6 * MB), { declaredBytes: 0.6 * MB }),
+      ).resolves.toBeTruthy();
+    });
+  });
+});
@@ -1706,18 +1706,30 @@ router.post('/:eventId/chunked-upload/:uploadId/chunk/:chunkIndex', adminAuth, r
  try {
    const { uploadId, chunkIndex } = req.params;

-    // Get chunk data from request body
-    const chunks = [];
-    for await (const chunk of req) {
-      chunks.push(chunk);
-    }
-    const chunkData = Buffer.concat(chunks);
-
-    const result = await chunkedUpload.uploadChunk(uploadId, parseInt(chunkIndex), chunkData);
+    // The request stream is handed over unread (#1403). Every check — unknown
+    // upload id, bad index, the per-file cap against Content-Length — runs
+    // inside uploadChunk before a byte is consumed, and the body is then
+    // streamed to the chunk file under a hard cap rather than concatenated in
+    // memory. Buffering it first meant a rejected 300MB request still cost
+    // 300MB of heap.
+    const declaredBytes = Number(req.headers['content-length']);
+    const result = await chunkedUpload.uploadChunk(uploadId, parseInt(chunkIndex), req, {
+      declaredBytes: Number.isFinite(declaredBytes) ? declaredBytes : undefined,
+    });

    res.json(result);
  } catch (error) {
-    if (error.statusCode === 413 || error.statusCode === 400) {
+    // Client-caused states (unknown/finished/expired upload, bad index, too
+    // large) carry their own status. Only a genuinely unexpected error should
+    // reach the 500 below and the error log with it.
+    if (error.statusCode) {
+      // Refusing the body early is the point — but it leaves unread bytes in
+      // flight on a connection this response still advertises as keep-alive.
+      // Node does not drain them, so the NEXT request on that socket hangs
+      // until it times out. Retire the connection instead.
+      if (!req.readableEnded) {
+        res.set('Connection', 'close');
+      }
      return res.status(error.statusCode).json({ error: error.message });
    }
    logger.error('Error uploading chunk:', error);
@@ -1767,8 +1779,10 @@ router.post('/:eventId/chunked-upload/:uploadId/complete', adminAuth, requirePer
      photos: uploadedPhotos
    });
  } catch (error) {
-    if (error.statusCode === 413) {
-      return res.status(413).json({ error: error.message });
+    // Same rule as the chunk route: a tagged status is a client-caused state
+    // (unknown/expired upload, missing chunks), not a server fault.
+    if (error.statusCode) {
+      return res.status(error.statusCode).json({ error: error.message });
    }
    logger.error('Error completing chunked upload:', error);
    res.status(500).json({ error: error.message || 'Failed to complete upload' });
@@ -30,6 +30,28 @@ 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 });
+}
+
+// A client-supplied upload id that is unknown, finished or expired is the
+// client's mistake, not the server's. These used to be plain Errors, so the
+// routes answered 500 — which reads as a backend fault in monitoring and
+// invites the client to retry something that will never succeed.
+function uploadStateError(message, statusCode) {
+  const err = new Error(message);
+  err.code = 'UPLOAD_STATE';
+  err.statusCode = statusCode;
+  return err;
+}
+
 function invalidChunkError(message) {
  const err = new Error(message);
  err.code = 'INVALID_CHUNK';
@@ -115,28 +137,100 @@ async function initializeUpload(options) {
  };
 }

+/**
+ * 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 read stops and the partial file is removed, so an oversized
+ * body costs the allowance rather than its own size.
+ */
+function writeChunkStream(source, partPath, allowance) {
+  const fsSync = require('fs');
+  return new Promise((resolve, reject) => {
+    // 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 out = fsSync.createWriteStream(partPath);
+    let written = 0;
+    let settled = false;
+
+    const settle = (err, value) => {
+      if (settled) return;
+      settled = true;
+      source.unpipe(out);
+      if (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);
+      }
+    };
+
+    source.on('data', (buf) => {
+      written += buf.length;
+      if (written > allowance) {
+        // 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', 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);
+  });
+}
+
 /**
 * Upload a single chunk
 * @param {string} uploadId - Upload ID
 * @param {number} chunkIndex - Chunk index (0-based)
- * @param {Buffer} chunkData - Chunk data
+ * @param {Buffer|import('stream').Readable} source - Chunk bytes, or a stream
+ *   of them (the request). A stream is never read until every check below has
+ *   passed, so a rejected request costs nothing (#1403).
+ * @param {Object} [options]
+ * @param {number} [options.declaredBytes] - Content-Length, when the caller
+ *   has one. Checked against the remaining allowance before the body is read.
 * @returns {Promise<Object>} - Chunk upload result
 */
-async function uploadChunk(uploadId, chunkIndex, chunkData) {
+async function uploadChunk(uploadId, chunkIndex, source, { declaredBytes } = {}) {
  const uploadMeta = activeUploads.get(uploadId);

  if (!uploadMeta) {
-    throw new Error('Upload not found or expired');
+    throw uploadStateError('Upload not found or expired', 404);
  }

  if (uploadMeta.status !== 'in_progress') {
-    throw new Error(`Upload is ${uploadMeta.status}`);
+    throw uploadStateError(`Upload is ${uploadMeta.status}`, 409);
  }

  // Check expiration
  if (Date.now() > uploadMeta.expiresAt) {
    await abortUpload(uploadId);
-    throw new Error('Upload expired');
+    throw uploadStateError('Upload expired', 410);
  }

  // Only the announced chunk indices are valid — anything else would merge
@@ -148,19 +242,73 @@ async function uploadChunk(uploadId, chunkIndex, chunkData) {
  // Enforce the per-file cap on the running byte total. The upload is
  // aborted, not just rejected: the chunks on disk are already over the
  // limit and the client can't complete the file any more.
-  const receivedBytes = totalReceivedBytes(uploadMeta) - (uploadMeta.chunkSizes.get(chunkIndex) || 0) + chunkData.length;
-  if (receivedBytes > uploadMeta.maxFileSizeBytes) {
+  //
+  // What this chunk may still contribute — everything already banked, minus a
+  // re-sent copy of this same index. Computed before the body is touched so a
+  // Content-Length that already blows the budget is refused having read zero
+  // bytes (#1403).
+  const bankedBytes = totalReceivedBytes(uploadMeta) - (uploadMeta.chunkSizes.get(chunkIndex) || 0);
+  const allowance = uploadMeta.maxFileSizeBytes - bankedBytes;
+
+  if (Number.isFinite(declaredBytes) && declaredBytes > allowance) {
    await abortUpload(uploadId);
    throw fileTooLargeError(uploadMeta.maxFileSizeBytes);
  }

-  // Write chunk to disk
  const chunkPath = path.join(uploadMeta.uploadDir, `chunk_${String(chunkIndex).padStart(6, '0')}`);
-  await fs.writeFile(chunkPath, chunkData);
+  let chunkLength;
+
+  if (Buffer.isBuffer(source)) {
+    if (bankedBytes + source.length > uploadMeta.maxFileSizeBytes) {
+      await abortUpload(uploadId);
+      throw fileTooLargeError(uploadMeta.maxFileSizeBytes);
+    }
+    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.
+    //
+    // 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);
+      // Re-check the aggregate before publishing. `allowance` was computed
+      // before the body arrived, so a chunk that completed while this one was
+      // still streaming is not counted in it — two overlapping 0.75MB chunks
+      // under a 1MB cap would otherwise both be accepted. The buffered version
+      // got this right for free by checking after the read; streaming has to
+      // ask again.
+      const bankedNow = totalReceivedBytes(uploadMeta) - (uploadMeta.chunkSizes.get(chunkIndex) || 0);
+      if (bankedNow + chunkLength > uploadMeta.maxFileSizeBytes) {
+        await fs.rm(partPath, { force: true }).catch(() => {});
+        await abortUpload(uploadId);
+        throw fileTooLargeError(uploadMeta.maxFileSizeBytes);
+      }
+      await fs.rename(partPath, chunkPath).catch(async (renameErr) => {
+        // A failed publish (ENOSPC, a vanished directory) left the fully
+        // written staging file behind. Its name is per-attempt, so a client
+        // that retries instead of aborting just accumulates more of them until
+        // the upload expires.
+        await fs.rm(partPath, { force: true }).catch(() => {});
+        throw renameErr;
+      });
+    } catch (err) {
+      if (err.overAllowance) {
+        await abortUpload(uploadId);
+        throw fileTooLargeError(uploadMeta.maxFileSizeBytes);
+      }
+      throw err;
+    }
+  }

  // Mark chunk as received
  uploadMeta.receivedChunks.add(chunkIndex);
-  uploadMeta.chunkSizes.set(chunkIndex, chunkData.length);
+  uploadMeta.chunkSizes.set(chunkIndex, chunkLength);

  const progress = (uploadMeta.receivedChunks.size / uploadMeta.expectedChunks) * 100;

@@ -190,12 +338,13 @@ async function completeUpload(uploadId) {
  const uploadMeta = activeUploads.get(uploadId);

  if (!uploadMeta) {
-    throw new Error('Upload not found or expired');
+    throw uploadStateError('Upload not found or expired', 404);
  }

  // Verify all chunks received
  if (uploadMeta.receivedChunks.size !== uploadMeta.expectedChunks) {
-    throw new Error(`Missing chunks: received ${uploadMeta.receivedChunks.size} of ${uploadMeta.expectedChunks}`);
+    throw uploadStateError(
+      `Missing chunks: received ${uploadMeta.receivedChunks.size} of ${uploadMeta.expectedChunks}`, 400);
  }

  uploadMeta.status = 'merging';