fix(server): S3 driver Upload monitoring — Transform interposition, stream cleanup

server/src/nest/storage/drivers/s3.driver.ts

@@ -1,6 +1,6 @@
 /// <reference types="@aws-lite/s3-types" />
 import awsLite from '@aws-lite/client';
-import { PassThrough, type Readable } from 'node:stream';
+import { PassThrough, Transform, type Readable } from 'node:stream';
 import fs from 'node:fs';
 import { assertValidKey, assertValidPrefix, isValidKey } from '../storage-keys';
 import {
@@ -20,8 +20,9 @@ import {
  * is a driver-internal detail — nothing outside this file imports aws-lite.
  * aws-lite exposes no request timeout or abort signal, so the driver owns the
  * deadline: per client call for single-request operations, inactivity-based
- * for multipart Upload (destroying the interposed PassThrough drives Upload's
- * own error path — in-flight parts drain, then AbortMultipartUpload fires).
+ * for multipart Upload (destroying the interposed monitoring Transform drives
+ * Upload's own error path — in-flight parts drain, then AbortMultipartUpload
+ * fires).
  */
 
 /** One multipart chunk (aws-lite Upload default) — also the peek threshold. */
@@ -186,8 +187,23 @@ export class S3Driver implements StorageDriver {
   /**
    * Multipart Upload under an INACTIVITY deadline: a fixed whole-operation
    * deadline would break large backups, so the timer resets on data movement.
-   * Expiry destroys the monitored stream, driving Upload's own error path —
-   * in-flight parts drain, then AbortMultipartUpload fires (verified upstream).
+   *
+   * The reset signal comes from an interposed Transform, NOT a `'data'`
+   * listener on `body` itself: `Readable.on('data', ...)` calls `resume()`
+   * on the next tick, so `body` would start flowing before aws-lite's Upload
+   * ever attaches its own consumer (it awaits a full CreateMultipartUpload
+   * round trip first) — anything emitted in that window has nowhere to land
+   * and is lost, silently truncating the object. A Transform's readable side
+   * only flows once something actually reads it, so bytes queue in its
+   * internal buffer (bounded by backpressure) until Upload's real consumer
+   * attaches; nothing is dropped, and "inactivity" tracks real data
+   * movement through the transform.
+   *
+   * Expiry destroys the Transform, the piped `body`, and (for the streaming
+   * put path) the original source — driving Upload's own error path
+   * (in-flight parts drain, then AbortMultipartUpload fires, verified
+   * upstream). Any OTHER Upload failure gets the same cleanup so a failed
+   * large upload never leaks an open fd/socket.
    */
   private async uploadMonitored(
     s3: S3Api,
@@ -201,6 +217,7 @@ export class S3Driver implements StorageDriver {
       const err = new StorageBackendError(
         `put '${key}' stalled for ${this.timeoutMs}ms on '${this.id}' — multipart upload aborted`,
       );
+      monitored.destroy(err);
       body.destroy(err);
       sourceToDestroy?.destroy(err);
     };
@@ -209,13 +226,45 @@ export class S3Driver implements StorageDriver {
       timer = setTimeout(expire, this.timeoutMs);
       timer.unref?.();
     };
-    body.on('data', reset); // body is consumed by Upload; extra listener observes only
+    const monitored = new Transform({
+      transform(chunk, _encoding, callback) {
+        reset();
+        callback(null, chunk);
+      },
+    });
+    // Attached BEFORE pipe()/destroy() can fire: an 'error' emitted on a
+    // Readable with no listener crashes the process, and both a genuine
+    // upstream read failure and our own expire()/catch cleanup emit one.
+    // `monitored` gets its own last-resort no-op: real aws-lite Upload
+    // attaches its own listener to `monitored` (the `Body` it receives)
+    // only after its CreateMultipartUpload round trip, so an expiry that
+    // fires in that window would otherwise destroy an unlistened stream —
+    // Upload's later read still observes the already-errored/destroyed
+    // state, so this doesn't swallow anything Upload would have seen.
+    //
+    // `onBodyError` is intentionally never `.off()`'d: `expire()` calls
+    // `monitored.destroy(err)` and `body.destroy(err)` back to back, each
+    // scheduling its own deferred 'error' emission. The FIRST (monitored's)
+    // rejects `s3.Upload(...)`, whose `catch`/`finally` runs as an
+    // interleaved microtask — before the SECOND (body's) emission gets its
+    // turn. Removing the listener there would leave that second emission
+    // unlistened and crash the process (verified with a standalone repro).
+    // Leaving it attached is harmless: `body` is single-use and discarded
+    // with `uploadMonitored`, so nothing keeps accumulating listeners.
+    const onBodyError = (err: Error) => monitored.destroy(err);
+    body.on('error', onBodyError);
+    monitored.on('error', () => {});
     reset();
+    body.pipe(monitored);
     try {
-      await s3.Upload({ ...base, Body: body, ChunkSize: MULTIPART_THRESHOLD });
+      await s3.Upload({ ...base, Body: monitored, ChunkSize: MULTIPART_THRESHOLD });
+    } catch (err) {
+      monitored.destroy();
+      body.destroy();
+      sourceToDestroy?.destroy();
+      throw err;
     } finally {
       clearTimeout(timer);
-      body.off('data', reset);
     }
   }
 

server/tests/unit/nest/storage/s3.driver.test.ts

@@ -273,6 +273,27 @@ describe('S3Driver put — bounded peek routing', () => {
       ChunkSize: MULTIPART_THRESHOLD,
     });
   });
+  it('does not drop bytes buffered ahead of a deferred Upload consumer (Transform interposition, not body.on("data"))', async () => {
+    // Real aws-lite Upload awaits a full CreateMultipartUpload round trip
+    // before attaching its own consumer. A `body.on('data', ...)` listener
+    // added directly to the Body stream calls resume() immediately, so any
+    // bytes emitted during that window are delivered only to the listener
+    // and lost forever. Simulate the round trip with two deferred ticks
+    // before the mock ever reads Body.
+    let uploaded: Buffer | null = null;
+    const api = makeMockApi({
+      Upload: vi.fn().mockImplementation(async ({ Body }: { Body: Readable }) => {
+        await new Promise((resolve) => setImmediate(resolve));
+        await new Promise((resolve) => setImmediate(resolve));
+        uploaded = await drain(Body);
+      }),
+    });
+    const chunk = Buffer.alloc(6 * 1024 * 1024, 7);
+    const big = Buffer.concat([chunk, chunk, chunk]);
+    await makeDriver(api).put('a/deferred.bin', Readable.from([chunk, chunk, chunk]));
+    expect(uploaded!.length).toBe(big.length);
+    expect(uploaded!.equals(big)).toBe(true);
+  });
   it('routes a single-chunk oversized stream that ends at the peek boundary to PutObject (readableEnded guard)', async () => {
     // One huge chunk: 'end' can fire between the peek resolving and pipe()
     // attaching — the driver must detect readableEnded and PutObject the
@@ -305,7 +326,7 @@ describe('S3Driver put — bounded peek routing', () => {
     expect(api.PutObject).not.toHaveBeenCalled();
     expect(api.Upload).not.toHaveBeenCalled();
   });
-  it('expires a stalled Upload via the inactivity deadline (PassThrough destroyed)', async () => {
+  it('expires a stalled Upload via the inactivity deadline (monitoring Transform destroyed)', async () => {
     const api = makeMockApi({
       Upload: vi.fn().mockImplementation(
         ({ Body }: { Body: Readable }) =>
@@ -320,6 +341,18 @@ describe('S3Driver put — bounded peek routing', () => {
       StorageBackendError,
     );
   });
+  it('destroys the original source when Upload rejects for a non-timeout reason (no fd/socket leak)', async () => {
+    const api = makeMockApi({
+      Upload: vi.fn().mockImplementation(async ({ Body }: { Body: Readable }) => {
+        await drain(Body); // full data transfer succeeds...
+        throw awsError(500); // ...but the finalize step (e.g. CompleteMultipartUpload) fails
+      }),
+    });
+    const chunk = Buffer.alloc(6 * 1024 * 1024, 7);
+    const source = Readable.from([chunk, chunk, chunk]);
+    await expect(makeDriver(api).put('a/fail-after-drain.bin', source)).rejects.toBeInstanceOf(StorageBackendError);
+    expect(source.destroyed).toBe(true);
+  });
 });
 
 describe('S3Driver put — LocalTempFile ownership', () => {
@@ -345,6 +378,26 @@ describe('S3Driver put — LocalTempFile ownership', () => {
     expect(api.Upload).toHaveBeenCalled();
     expect(fs.existsSync(tmp)).toBe(false);
   });
+  it('rejects a stalled threshold-sized temp-file Upload with StorageBackendError rather than crashing', async () => {
+    // The Upload mock never reads/drains Body — only listens for its error,
+    // matching aws-lite's real Upload contract — so the fs.createReadStream
+    // source AND the interposed Transform must each have an 'error' handler
+    // attached before expire() destroys them; otherwise an unlistened
+    // 'error' anywhere in that chain crashes the process instead of
+    // rejecting this promise.
+    const api = makeMockApi({
+      Upload: vi.fn().mockImplementation(
+        ({ Body }: { Body: Readable }) =>
+          new Promise((_, reject) => {
+            Body.on('error', reject);
+          }),
+      ),
+    });
+    const tmp = await makeTmp(MULTIPART_THRESHOLD);
+    await expect(
+      makeDriver(api, { timeoutMs: 60 }).put('a/stall-tmp.bin', { tmpPath: tmp }),
+    ).rejects.toBeInstanceOf(StorageBackendError);
+  });
   it('consumes the temp file on the failure path too (MirrorDriver precedent)', async () => {
     const api = makeMockApi({ PutObject: vi.fn().mockRejectedValue(awsError(500)) });
     const tmp = await makeTmp(10);