From 6d3d57f97f5299ca37a0c0e719a6d6a804aa2d27 Mon Sep 17 00:00:00 2001 From: jubnl Date: Wed, 19 Aug 2026 21:26:23 +0200 Subject: [PATCH] =?UTF-8?q?fix(server):=20S3=20driver=20Upload=20monitorin?= =?UTF-8?q?g=20=E2=80=94=20Transform=20interposition,=20stream=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/src/nest/storage/drivers/s3.driver.ts | 65 ++++++++++++++++--- .../tests/unit/nest/storage/s3.driver.test.ts | 55 +++++++++++++++- 2 files changed, 111 insertions(+), 9 deletions(-) diff --git a/server/src/nest/storage/drivers/s3.driver.ts b/server/src/nest/storage/drivers/s3.driver.ts index fa97ab07c..804da8864 100644 --- a/server/src/nest/storage/drivers/s3.driver.ts +++ b/server/src/nest/storage/drivers/s3.driver.ts @@ -1,6 +1,6 @@ /// 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); } } diff --git a/server/tests/unit/nest/storage/s3.driver.test.ts b/server/tests/unit/nest/storage/s3.driver.test.ts index d3202a9fd..52aee13a5 100644 --- a/server/tests/unit/nest/storage/s3.driver.test.ts +++ b/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);