From 5915759a43ea1f10fda89c30d425c3d467ddb593 Mon Sep 17 00:00:00 2001 From: Lenny Date: Tue, 6 Oct 2026 12:01:03 -0400 Subject: [PATCH 1/5] fix: abort TUS connections that receive no bytes within the idle timeout --- src/config.ts | 5 + src/http/routes/tus/index.ts | 43 +++++++- src/test/tus.test.ts | 184 ++++++++++++++++++++++++++++++++++- 3 files changed, 227 insertions(+), 5 deletions(-) diff --git a/src/config.ts b/src/config.ts index 55a82256a..4c1949843 100644 --- a/src/config.ts +++ b/src/config.ts @@ -206,6 +206,7 @@ type StorageConfigType = { tusUseFileVersionSeparator: boolean tusAllowS3Tags: boolean tusLockType: 'postgres' | 's3' + tusBodyIdleTimeoutMs: number s3ProtocolEnabled: boolean s3ProtocolPrefix: string s3ProtocolAllowForwardedHeader: boolean @@ -480,6 +481,10 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType { getOptionalConfigFromEnv('TUS_USE_FILE_VERSION_SEPARATOR') === 'true', tusAllowS3Tags: getOptionalConfigFromEnv('TUS_ALLOW_S3_TAGS') !== 'false', tusLockType: getOptionalConfigFromEnv('TUS_LOCK_TYPE') || 'postgres', + tusBodyIdleTimeoutMs: envPositiveInteger( + getOptionalConfigFromEnv('TUS_BODY_IDLE_TIMEOUT_MS'), + 1000 * 60 + ), // S3 Protocol s3ProtocolEnabled: getOptionalConfigFromEnv('S3_PROTOCOL_ENABLED') !== 'false', diff --git a/src/http/routes/tus/index.ts b/src/http/routes/tus/index.ts index 7ef09353f..a95222f92 100644 --- a/src/http/routes/tus/index.ts +++ b/src/http/routes/tus/index.ts @@ -55,6 +55,7 @@ const { tusMaxConcurrentUploads, tusAllowS3Tags, tusLockType, + tusBodyIdleTimeoutMs, uploadFileSizeLimit, storageBackendType, storageFilePath, @@ -283,6 +284,40 @@ function setTusRequestContext( done() } +async function handleTusRequestWithIdleTimeout( + tusServer: Server, + req: FastifyRequest, + res: FastifyReply +) { + const socket = req.raw.socket + if (!socket || tusBodyIdleTimeoutMs <= 0) { + return tusServer.handle(req.raw, res.raw) + } + + const onIdleTimeout = () => { + const err = ERRORS.TusError('TUS request body idle timeout - no bytes received', 408) + req.raw.executionError = err + req.raw.destroy(err) + } + const disarm = () => { + socket.setTimeout(0) + socket.removeListener('timeout', onIdleTimeout) + req.raw.removeListener('end', disarm) + } + + socket.setTimeout(tusBodyIdleTimeoutMs, onIdleTimeout) + // Stop tracking idle time once the client has sent the full body, so + // slow lock acquisition or upload finalization afterward can't trip + // a "no bytes received" timeout. + req.raw.once('end', disarm) + + try { + return await tusServer.handle(req.raw, res.raw) + } finally { + disarm() + } +} + export const authenticatedRoutes = fastifyPlugin( async (fastify: FastifyInstance, options: { tusServer: Server; signed: boolean }) => { const operationSuffix = options.signed ? '_signed' : '' @@ -310,7 +345,7 @@ export const authenticatedRoutes = fastifyPlugin( }, }, async (req, res) => { - await options.tusServer.handle(req.raw, res.raw) + await handleTusRequestWithIdleTimeout(options.tusServer, req, res) } ) @@ -323,7 +358,7 @@ export const authenticatedRoutes = fastifyPlugin( }, }, async (req, res) => { - await options.tusServer.handle(req.raw, res.raw) + await handleTusRequestWithIdleTimeout(options.tusServer, req, res) } ) @@ -336,7 +371,7 @@ export const authenticatedRoutes = fastifyPlugin( }, }, async (req, res) => { - await options.tusServer.handle(req.raw, res.raw) + await handleTusRequestWithIdleTimeout(options.tusServer, req, res) } ) fastify.patch( @@ -351,7 +386,7 @@ export const authenticatedRoutes = fastifyPlugin( }, }, async (req, res) => { - await options.tusServer.handle(req.raw, res.raw) + await handleTusRequestWithIdleTimeout(options.tusServer, req, res) } ) fastify.head( diff --git a/src/test/tus.test.ts b/src/test/tus.test.ts index f87b8b838..86630e725 100644 --- a/src/test/tus.test.ts +++ b/src/test/tus.test.ts @@ -1,9 +1,12 @@ import { once } from 'node:events' import { mkdtemp } from 'node:fs/promises' +import * as http from 'node:http' import { request as httpRequest, type IncomingMessage } from 'node:http' import { tmpdir } from 'node:os' import path from 'node:path' import { text } from 'node:stream/consumers' +import { setTimeout as sleep } from 'node:timers/promises' + import { CreateBucketCommand, HeadObjectCommand, @@ -181,7 +184,7 @@ function expectTusErrorResponse(error: unknown) { async function createTusTestContext( backendType: 'file' | 's3', - options: { fileBackendPath?: string } = {} + options: { fileBackendPath?: string; tusBodyIdleTimeoutMs?: number } = {} ): Promise { vi.resetModules() @@ -192,10 +195,14 @@ async function createTusTestContext( const overrides: Partial<{ storageBackendType: 'file' | 's3' storageFilePath: string + tusBodyIdleTimeoutMs: number }> = { storageBackendType: backendType } if (backendType === 'file') { overrides.storageFilePath = options.fileBackendPath } + if (options.tusBodyIdleTimeoutMs !== undefined) { + overrides.tusBodyIdleTimeoutMs = options.tusBodyIdleTimeoutMs + } configModule.mergeConfig(overrides) const [appModule, backendModule, storageModule, databaseModule, locatorModule] = @@ -1328,3 +1335,178 @@ describe('File-backed TUS — TUS_USE_FILE_VERSION_SEPARATOR', () => { expect(await pathExists(storedObjectPath)).toBe(true) }) }) + +describe('TUS body idle timeout', () => { + const idleTimeoutMs = 200 + + let context: TusTestContext + let db: StoragePgDBType + let storage: StorageType + let connection: Awaited> + let bucketName: string + let fileBackendPath: string + + beforeAll(async () => { + fileBackendPath = await mkdtemp(path.join(tmpdir(), 'storage-tus-idle-')) + context = await createTusTestContext('file', { + fileBackendPath, + tusBodyIdleTimeoutMs: idleTimeoutMs, + }) + }) + + afterAll(async () => { + await context?.server?.close() + vi.resetModules() + await removePath(fileBackendPath) + }) + + beforeEach(async () => { + const superUser = await getServiceKeyUser(context.config.tenantId) + connection = await getPostgresConnection({ + superUser, + user: superUser, + tenantId: context.config.tenantId, + host: 'localhost', + disableHostCheck: true, + }) + + db = new context.StoragePgDB(connection, { + tenantId: context.config.tenantId, + host: 'localhost', + }) + + bucketName = randomUUID() + storage = new context.Storage( + context.backend, + db, + new context.TenantLocation(context.config.storageS3Bucket) + ) + await storage.createBucket({ id: bucketName, name: bucketName, public: true }) + }) + + afterEach(async () => { + connection?.dispose() + }) + + async function createRawUpload(totalSize: number) { + const authorization = `Bearer ${await context.config.serviceKeyAsync}` + const objectName = `${randomUUID()}-idle-timeout.bin` + + const location = await new Promise((resolve, reject) => { + const req = http.request( + `${context.baseUrl}${context.config.tusPath}`, + { + method: 'POST', + headers: { + authorization, + 'x-upsert': 'true', + 'Tus-Resumable': '1.0.0', + 'Upload-Length': String(totalSize), + 'Upload-Metadata': encodeTusMetadata({ + bucketName, + objectName, + contentType: 'application/octet-stream', + }), + }, + }, + (res) => { + res.resume() + res.on('end', () => { + if (res.statusCode !== 201 || !res.headers.location) { + reject(new Error(`creation failed with status ${res.statusCode}`)) + return + } + resolve(res.headers.location as string) + }) + } + ) + req.on('error', reject) + req.end() + }) + + return { uploadUrl: new URL(location, context.baseUrl), authorization } + } + + function openRawPatch( + uploadUrl: URL, + authorization: string, + offset: number, + declaredLength: number + ) { + const req = http.request(uploadUrl, { + method: 'PATCH', + headers: { + authorization, + 'Tus-Resumable': '1.0.0', + 'Upload-Offset': String(offset), + 'Content-Type': 'application/offset+octet-stream', + 'Content-Length': String(declaredLength), + }, + }) + req.on('error', () => {}) + return req + } + + async function headOffset(uploadUrl: URL, authorization: string): Promise { + return new Promise((resolve, reject) => { + const req = http.request( + uploadUrl, + { method: 'HEAD', headers: { authorization, 'Tus-Resumable': '1.0.0' } }, + (res) => { + res.resume() + res.on('end', () => resolve(Number(res.headers['upload-offset']))) + } + ) + req.on('error', reject) + req.end() + }) + } + + it( + 'destroys a connection that goes silent mid-chunk, preserving already-received bytes', + async () => { + const totalSize = 2 * 1024 * 1024 + const sentSize = 64 * 1024 + const { uploadUrl, authorization } = await createRawUpload(totalSize) + + const patch = openRawPatch(uploadUrl, authorization, 0, totalSize) + const destroyed = new Promise((resolve) => { + patch.on('error', () => resolve()) + }) + patch.write(Buffer.alloc(sentSize)) + // Deliberately never call .end() or .destroy() + // simulate a connection that goes silent mid-chunk with no closing signal + + await destroyed + + const offset = await headOffset(uploadUrl, authorization) + expect(offset).toBe(sentSize) + }, + idleTimeoutMs * 15 + ) + + it( + 'does not time out a slow-but-steady trickle whose gaps stay under the idle threshold', + async () => { + const chunkSize = 16 * 1024 + const chunkCount = 6 + const totalSize = chunkSize * chunkCount + const { uploadUrl, authorization } = await createRawUpload(totalSize) + + const patch = openRawPatch(uploadUrl, authorization, 0, totalSize) + const response = new Promise((resolve, reject) => { + patch.on('response', (res) => resolve(res.statusCode ?? 0)) + patch.on('error', reject) + }) + + for (let i = 0; i < chunkCount; i++) { + await sleep(idleTimeoutMs / 2) + patch.write(Buffer.alloc(chunkSize)) + } + patch.end() + + expect(await response).toBe(204) + }, + idleTimeoutMs * 20 + ) +}) From bcbe90e4e3ee9cec2e842a9cdd4acbbc5c9c07c8 Mon Sep 17 00:00:00 2001 From: Lenny Date: Tue, 6 Oct 2026 17:58:45 -0400 Subject: [PATCH 2/5] handle completed and missing body --- src/http/routes/tus/index.test.ts | 88 ++++++++++++++++++++++++++++++- src/http/routes/tus/index.ts | 23 +++++--- 2 files changed, 102 insertions(+), 9 deletions(-) diff --git a/src/http/routes/tus/index.test.ts b/src/http/routes/tus/index.test.ts index 112036eb7..ea8bc9377 100644 --- a/src/http/routes/tus/index.test.ts +++ b/src/http/routes/tus/index.test.ts @@ -1,13 +1,14 @@ +import { EventEmitter } from 'node:events' import * as http from 'node:http' import * as https from 'node:https' import type { AddressInfo } from 'node:net' import { NodeHttpHandler } from '@smithy/node-http-handler' import { HttpRequest } from '@smithy/protocol-http' import type { Server } from '@tus/server' -import Fastify, { FastifyInstance } from 'fastify' +import Fastify, { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify' import { getConfig } from '../../../config' import { requestContext } from '../../plugins/request-context' -import { createTusLockS3Client, publicRoutes } from './index' +import { createTusLockS3Client, handleTusRequestWithIdleTimeout, publicRoutes } from './index' import type { MultiPartRequest } from './lifecycle' describe('TUS S3 clients', () => { @@ -149,3 +150,86 @@ describe('public tus route request context', () => { expect(observedUpload?.db.dispose).toHaveBeenCalled() }) }) + +class FakeSocket extends EventEmitter { + setTimeout = vi.fn((_ms: number, cb?: () => void) => { + if (cb) this.on('timeout', cb) + return this + }) +} + +class FakeIncomingMessage extends EventEmitter { + socket: FakeSocket | undefined = new FakeSocket() + complete = false + readableEnded = false + headers: Record = {} + executionError?: Error + destroy = vi.fn() +} + +describe('handleTusRequestWithIdleTimeout', () => { + function createReqRes(headers: Record = {}) { + const raw = new FakeIncomingMessage() + raw.headers = headers + const req = { raw } as unknown as FastifyRequest + const res = { raw: {} } as unknown as FastifyReply + return { req, res, raw } + } + + test('skips arming the idle timer when the request declares no body', async () => { + const handle = vi.fn().mockResolvedValue(undefined) + const { req, res, raw } = createReqRes() + const tusServer = { handle } as unknown as Server + + await handleTusRequestWithIdleTimeout(tusServer, req, res) + + expect(handle).toHaveBeenCalledWith(raw, res.raw) + expect(raw.socket?.setTimeout).not.toHaveBeenCalled() + }) + + test('skips arming the idle timer when the body is already fully received', async () => { + const handle = vi.fn().mockResolvedValue(undefined) + const { req, res, raw } = createReqRes({ 'content-length': '10' }) + raw.complete = true + const tusServer = { handle } as unknown as Server + + await handleTusRequestWithIdleTimeout(tusServer, req, res) + + expect(handle).toHaveBeenCalledWith(raw, res.raw) + expect(raw.socket?.setTimeout).not.toHaveBeenCalled() + }) + + test('destroys the connection when the idle timer fires before the body completes', async () => { + const handle = vi.fn().mockResolvedValue(undefined) + const { req, res, raw } = createReqRes({ 'content-length': '10' }) + const tusServer = { handle } as unknown as Server + + const pending = handleTusRequestWithIdleTimeout(tusServer, req, res) + // Fire the timer before `handle` resolves, simulating a client that + // stopped sending bytes partway through the body. + raw.socket?.emit('timeout') + await pending + + expect(raw.destroy).toHaveBeenCalledTimes(1) + expect(raw.destroy.mock.calls[0][0]).toMatchObject({ + message: 'TUS request body idle timeout - no bytes received', + }) + expect(raw.executionError).toBe(raw.destroy.mock.calls[0][0]) + }) + + test('disarms without destroying the connection if the idle timer fires as the body finishes arriving', async () => { + const handle = vi.fn().mockResolvedValue(undefined) + const { req, res, raw } = createReqRes({ 'content-length': '10' }) + const tusServer = { handle } as unknown as Server + + const pending = handleTusRequestWithIdleTimeout(tusServer, req, res) + // The body finishes arriving in the same window the idle timer fires, + // racing the 'end' listener that would otherwise disarm it first. + raw.complete = true + raw.socket?.emit('timeout') + await pending + + expect(raw.destroy).not.toHaveBeenCalled() + expect(raw.executionError).toBeUndefined() + }) +}) diff --git a/src/http/routes/tus/index.ts b/src/http/routes/tus/index.ts index a95222f92..a3a8f1007 100644 --- a/src/http/routes/tus/index.ts +++ b/src/http/routes/tus/index.ts @@ -284,27 +284,36 @@ function setTusRequestContext( done() } -async function handleTusRequestWithIdleTimeout( +export async function handleTusRequestWithIdleTimeout( tusServer: Server, req: FastifyRequest, res: FastifyReply ) { const socket = req.raw.socket - if (!socket || tusBodyIdleTimeoutMs <= 0) { + const isBodyFullyReceived = () => req.raw.complete || req.raw.readableEnded + const hasDeclaredBody = + req.raw.headers['transfer-encoding'] !== undefined || + Number(req.raw.headers['content-length']) > 0 + if (!socket || tusBodyIdleTimeoutMs <= 0 || !hasDeclaredBody || isBodyFullyReceived()) { return tusServer.handle(req.raw, res.raw) } - const onIdleTimeout = () => { - const err = ERRORS.TusError('TUS request body idle timeout - no bytes received', 408) - req.raw.executionError = err - req.raw.destroy(err) - } const disarm = () => { socket.setTimeout(0) socket.removeListener('timeout', onIdleTimeout) req.raw.removeListener('end', disarm) } + const onIdleTimeout = () => { + if (isBodyFullyReceived()) { + disarm() + return + } + const err = ERRORS.TusError('TUS request body idle timeout - no bytes received', 408) + req.raw.executionError = err + req.raw.destroy(err) + } + socket.setTimeout(tusBodyIdleTimeoutMs, onIdleTimeout) // Stop tracking idle time once the client has sent the full body, so // slow lock acquisition or upload finalization afterward can't trip From 5a1c0b5e77ca9ea14d48e22098bb951c87008235 Mon Sep 17 00:00:00 2001 From: Lenny Date: Wed, 7 Oct 2026 12:28:16 -0400 Subject: [PATCH 3/5] poll socket.bytesRead instead of using onTimeout hook --- src/http/routes/tus/index.test.ts | 68 ++++++++++++++++++++++--------- src/http/routes/tus/index.ts | 19 +++++++-- 2 files changed, 63 insertions(+), 24 deletions(-) diff --git a/src/http/routes/tus/index.test.ts b/src/http/routes/tus/index.test.ts index ea8bc9377..19afe443e 100644 --- a/src/http/routes/tus/index.test.ts +++ b/src/http/routes/tus/index.test.ts @@ -151,11 +151,8 @@ describe('public tus route request context', () => { }) }) -class FakeSocket extends EventEmitter { - setTimeout = vi.fn((_ms: number, cb?: () => void) => { - if (cb) this.on('timeout', cb) - return this - }) +class FakeSocket { + bytesRead = 0 } class FakeIncomingMessage extends EventEmitter { @@ -168,6 +165,16 @@ class FakeIncomingMessage extends EventEmitter { } describe('handleTusRequestWithIdleTimeout', () => { + const { tusBodyIdleTimeoutMs } = getConfig() + + beforeEach(() => { + vi.useFakeTimers() + }) + + afterEach(() => { + vi.useRealTimers() + }) + function createReqRes(headers: Record = {}) { const raw = new FakeIncomingMessage() raw.headers = headers @@ -176,7 +183,14 @@ describe('handleTusRequestWithIdleTimeout', () => { return { req, res, raw } } + // A `handle` that never resolves on its own, so the idle-check timers + // get a chance to fire before the wrapper's own `finally` disarms them. + function pendingHandle() { + return vi.fn(() => new Promise(() => {})) + } + test('skips arming the idle timer when the request declares no body', async () => { + const setTimeoutSpy = vi.spyOn(global, 'setTimeout') const handle = vi.fn().mockResolvedValue(undefined) const { req, res, raw } = createReqRes() const tusServer = { handle } as unknown as Server @@ -184,10 +198,11 @@ describe('handleTusRequestWithIdleTimeout', () => { await handleTusRequestWithIdleTimeout(tusServer, req, res) expect(handle).toHaveBeenCalledWith(raw, res.raw) - expect(raw.socket?.setTimeout).not.toHaveBeenCalled() + expect(setTimeoutSpy).not.toHaveBeenCalled() }) test('skips arming the idle timer when the body is already fully received', async () => { + const setTimeoutSpy = vi.spyOn(global, 'setTimeout') const handle = vi.fn().mockResolvedValue(undefined) const { req, res, raw } = createReqRes({ 'content-length': '10' }) raw.complete = true @@ -196,19 +211,16 @@ describe('handleTusRequestWithIdleTimeout', () => { await handleTusRequestWithIdleTimeout(tusServer, req, res) expect(handle).toHaveBeenCalledWith(raw, res.raw) - expect(raw.socket?.setTimeout).not.toHaveBeenCalled() + expect(setTimeoutSpy).not.toHaveBeenCalled() }) - test('destroys the connection when the idle timer fires before the body completes', async () => { - const handle = vi.fn().mockResolvedValue(undefined) + test('destroys the connection when no bytes arrive before the idle check fires', async () => { + const handle = pendingHandle() const { req, res, raw } = createReqRes({ 'content-length': '10' }) const tusServer = { handle } as unknown as Server - const pending = handleTusRequestWithIdleTimeout(tusServer, req, res) - // Fire the timer before `handle` resolves, simulating a client that - // stopped sending bytes partway through the body. - raw.socket?.emit('timeout') - await pending + void handleTusRequestWithIdleTimeout(tusServer, req, res) + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) expect(raw.destroy).toHaveBeenCalledTimes(1) expect(raw.destroy.mock.calls[0][0]).toMatchObject({ @@ -217,17 +229,33 @@ describe('handleTusRequestWithIdleTimeout', () => { expect(raw.executionError).toBe(raw.destroy.mock.calls[0][0]) }) - test('disarms without destroying the connection if the idle timer fires as the body finishes arriving', async () => { - const handle = vi.fn().mockResolvedValue(undefined) + test('reschedules instead of timing out while bytesRead keeps increasing, then times out once it stalls', async () => { + const handle = pendingHandle() + const { req, res, raw } = createReqRes({ 'content-length': '10' }) + const tusServer = { handle } as unknown as Server + + void handleTusRequestWithIdleTimeout(tusServer, req, res) + + // Bytes arrive just before the first check - should reschedule, not time out. + raw.socket!.bytesRead = 5 + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) + expect(raw.destroy).not.toHaveBeenCalled() + + // No further bytes arrive during the second window - now it should time out. + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) + expect(raw.destroy).toHaveBeenCalledTimes(1) + }) + + test('disarms without destroying the connection if the idle check runs as the body finishes arriving', async () => { + const handle = pendingHandle() const { req, res, raw } = createReqRes({ 'content-length': '10' }) const tusServer = { handle } as unknown as Server - const pending = handleTusRequestWithIdleTimeout(tusServer, req, res) - // The body finishes arriving in the same window the idle timer fires, + void handleTusRequestWithIdleTimeout(tusServer, req, res) + // The body finishes arriving in the same window the idle check runs, // racing the 'end' listener that would otherwise disarm it first. raw.complete = true - raw.socket?.emit('timeout') - await pending + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) expect(raw.destroy).not.toHaveBeenCalled() expect(raw.executionError).toBeUndefined() diff --git a/src/http/routes/tus/index.ts b/src/http/routes/tus/index.ts index a3a8f1007..fafcf452a 100644 --- a/src/http/routes/tus/index.ts +++ b/src/http/routes/tus/index.ts @@ -298,23 +298,34 @@ export async function handleTusRequestWithIdleTimeout( return tusServer.handle(req.raw, res.raw) } + // We poll bytesRead on our own timer rather than socket.setTimeout(), + // which also triggers Fastify's onTimeout hook (disposing the db + // connection) on every check, or a 'data' listener, which would start + // flowing the stream and steal bytes from @tus/server's own consumer. + let lastBytesRead = socket.bytesRead + let idleTimer: NodeJS.Timeout + const disarm = () => { - socket.setTimeout(0) - socket.removeListener('timeout', onIdleTimeout) + clearTimeout(idleTimer) req.raw.removeListener('end', disarm) } - const onIdleTimeout = () => { + const checkIdle = () => { if (isBodyFullyReceived()) { disarm() return } + if (socket.bytesRead > lastBytesRead) { + lastBytesRead = socket.bytesRead + idleTimer = setTimeout(checkIdle, tusBodyIdleTimeoutMs) + return + } const err = ERRORS.TusError('TUS request body idle timeout - no bytes received', 408) req.raw.executionError = err req.raw.destroy(err) } - socket.setTimeout(tusBodyIdleTimeoutMs, onIdleTimeout) + idleTimer = setTimeout(checkIdle, tusBodyIdleTimeoutMs) // Stop tracking idle time once the client has sent the full body, so // slow lock acquisition or upload finalization afterward can't trip // a "no bytes received" timeout. From 2382fff39c44664ded633b53be9c65f855935e94 Mon Sep 17 00:00:00 2001 From: Lenny Date: Wed, 7 Oct 2026 15:48:54 -0400 Subject: [PATCH 4/5] also check readableLength --- src/config.ts | 5 +++-- src/http/routes/tus/index.test.ts | 21 +++++++++++++++++++++ src/http/routes/tus/index.ts | 10 ++++++++++ 3 files changed, 34 insertions(+), 2 deletions(-) diff --git a/src/config.ts b/src/config.ts index 4c1949843..be84ad0e5 100644 --- a/src/config.ts +++ b/src/config.ts @@ -481,9 +481,10 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType { getOptionalConfigFromEnv('TUS_USE_FILE_VERSION_SEPARATOR') === 'true', tusAllowS3Tags: getOptionalConfigFromEnv('TUS_ALLOW_S3_TAGS') !== 'false', tusLockType: getOptionalConfigFromEnv('TUS_LOCK_TYPE') || 'postgres', - tusBodyIdleTimeoutMs: envPositiveInteger( + tusBodyIdleTimeoutMs: envBoundedPositiveInteger( getOptionalConfigFromEnv('TUS_BODY_IDLE_TIMEOUT_MS'), - 1000 * 60 + 1000 * 60, + MAX_TIMER_DELAY_MS ), // S3 Protocol diff --git a/src/http/routes/tus/index.test.ts b/src/http/routes/tus/index.test.ts index 19afe443e..e45ef6900 100644 --- a/src/http/routes/tus/index.test.ts +++ b/src/http/routes/tus/index.test.ts @@ -159,6 +159,7 @@ class FakeIncomingMessage extends EventEmitter { socket: FakeSocket | undefined = new FakeSocket() complete = false readableEnded = false + readableLength = 0 headers: Record = {} executionError?: Error destroy = vi.fn() @@ -246,6 +247,26 @@ describe('handleTusRequestWithIdleTimeout', () => { expect(raw.destroy).toHaveBeenCalledTimes(1) }) + test('reschedules instead of timing out when bytesRead stalls but data is backed up unconsumed (write backpressure)', async () => { + const handle = pendingHandle() + const { req, res, raw } = createReqRes({ 'content-length': '10' }) + const tusServer = { handle } as unknown as Server + + void handleTusRequestWithIdleTimeout(tusServer, req, res) + + // bytesRead never advances (simulating the socket read stalling under backpressure from a slow destination write) + // the client did send something, so this should not be treated as a timeout. + raw.readableLength = 65536 + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) + expect(raw.destroy).not.toHaveBeenCalled() + + // Once the backlog actually drains to empty while bytesRead is still + // stalled, that's a genuine idle client - now it should time out. + raw.readableLength = 0 + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) + expect(raw.destroy).toHaveBeenCalledTimes(1) + }) + test('disarms without destroying the connection if the idle check runs as the body finishes arriving', async () => { const handle = pendingHandle() const { req, res, raw } = createReqRes({ 'content-length': '10' }) diff --git a/src/http/routes/tus/index.ts b/src/http/routes/tus/index.ts index fafcf452a..b3d44198e 100644 --- a/src/http/routes/tus/index.ts +++ b/src/http/routes/tus/index.ts @@ -320,6 +320,16 @@ export async function handleTusRequestWithIdleTimeout( idleTimer = setTimeout(checkIdle, tusBodyIdleTimeoutMs) return } + + // bytesRead also stalls when our own write to the destination is back-pressured + // since Node stops reading the socket until the backlog drains. + // A non-zero readableLength means there's still unconsumed data sitting in the + // buffer, so the client did send something + if (req.raw.readableLength > 0) { + idleTimer = setTimeout(checkIdle, tusBodyIdleTimeoutMs) + return + } + const err = ERRORS.TusError('TUS request body idle timeout - no bytes received', 408) req.raw.executionError = err req.raw.destroy(err) From 00d8be329cd8245a4ef8fa01544b85958ae4e0cd Mon Sep 17 00:00:00 2001 From: Lenny Date: Thu, 8 Oct 2026 15:07:47 -0400 Subject: [PATCH 5/5] disarm on close, only idle if no bytes are moving --- src/config.ts | 3 +- src/http/routes/tus/index.test.ts | 61 +++++++++++++++++++++++++++---- src/http/routes/tus/index.ts | 37 +++++++++---------- 3 files changed, 73 insertions(+), 28 deletions(-) diff --git a/src/config.ts b/src/config.ts index be84ad0e5..85dd03950 100644 --- a/src/config.ts +++ b/src/config.ts @@ -481,9 +481,10 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType { getOptionalConfigFromEnv('TUS_USE_FILE_VERSION_SEPARATOR') === 'true', tusAllowS3Tags: getOptionalConfigFromEnv('TUS_ALLOW_S3_TAGS') !== 'false', tusLockType: getOptionalConfigFromEnv('TUS_LOCK_TYPE') || 'postgres', - tusBodyIdleTimeoutMs: envBoundedPositiveInteger( + tusBodyIdleTimeoutMs: envIntegerInRange( getOptionalConfigFromEnv('TUS_BODY_IDLE_TIMEOUT_MS'), 1000 * 60, + 0, MAX_TIMER_DELAY_MS ), diff --git a/src/http/routes/tus/index.test.ts b/src/http/routes/tus/index.test.ts index e45ef6900..f9a1e80eb 100644 --- a/src/http/routes/tus/index.test.ts +++ b/src/http/routes/tus/index.test.ts @@ -160,9 +160,12 @@ class FakeIncomingMessage extends EventEmitter { complete = false readableEnded = false readableLength = 0 + destroyed = false headers: Record = {} executionError?: Error - destroy = vi.fn() + destroy = vi.fn((_err?: Error) => { + this.destroyed = true + }) } describe('handleTusRequestWithIdleTimeout', () => { @@ -247,26 +250,68 @@ describe('handleTusRequestWithIdleTimeout', () => { expect(raw.destroy).toHaveBeenCalledTimes(1) }) - test('reschedules instead of timing out when bytesRead stalls but data is backed up unconsumed (write backpressure)', async () => { + test('reschedules while the destination write is draining a backpressure backlog, then times out once nothing moves at all', async () => { const handle = pendingHandle() const { req, res, raw } = createReqRes({ 'content-length': '10' }) const tusServer = { handle } as unknown as Server void handleTusRequestWithIdleTimeout(tusServer, req, res) - // bytesRead never advances (simulating the socket read stalling under backpressure from a slow destination write) - // the client did send something, so this should not be treated as a timeout. - raw.readableLength = 65536 + // Client burst fills the buffer faster than the destination write can + // drain it: bytesRead and readableLength climb together. + raw.socket!.bytesRead = 100_000 + raw.readableLength = 65_536 await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) expect(raw.destroy).not.toHaveBeenCalled() - // Once the backlog actually drains to empty while bytesRead is still - // stalled, that's a genuine idle client - now it should time out. - raw.readableLength = 0 + // bytesRead stops growing, but the destination write is still actively draining the backlog + raw.readableLength = 32_768 + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) + expect(raw.destroy).not.toHaveBeenCalled() + + // No further progress is genuine idleness and must time out, even with a backlog + // sitting in the buffer, or a permanently stuck write would hold the lock forever. await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) expect(raw.destroy).toHaveBeenCalledTimes(1) }) + test('skips arming the idle timer when the request is already destroyed', async () => { + const setTimeoutSpy = vi.spyOn(global, 'setTimeout') + const handle = vi.fn().mockResolvedValue(undefined) + const { req, res, raw } = createReqRes({ 'content-length': '10' }) + raw.destroyed = true + const tusServer = { handle } as unknown as Server + + await handleTusRequestWithIdleTimeout(tusServer, req, res) + + expect(handle).toHaveBeenCalledWith(raw, res.raw) + expect(setTimeoutSpy).not.toHaveBeenCalled() + }) + + test('disarms without re-destroying if the connection is destroyed by something else before the idle check runs', async () => { + const handle = pendingHandle() + const { req, res, raw } = createReqRes({ 'content-length': '10' }) + const tusServer = { handle } as unknown as Server + + void handleTusRequestWithIdleTimeout(tusServer, req, res) + raw.destroyed = true + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs) + + expect(raw.destroy).not.toHaveBeenCalled() + }) + + test('disarms the idle timer on an abrupt disconnect, which closes without ever emitting end', async () => { + const handle = pendingHandle() + const { req, res, raw } = createReqRes({ 'content-length': '10' }) + const tusServer = { handle } as unknown as Server + + void handleTusRequestWithIdleTimeout(tusServer, req, res) + raw.emit('close') + + await vi.advanceTimersByTimeAsync(tusBodyIdleTimeoutMs * 2) + expect(raw.destroy).not.toHaveBeenCalled() + }) + test('disarms without destroying the connection if the idle check runs as the body finishes arriving', async () => { const handle = pendingHandle() const { req, res, raw } = createReqRes({ 'content-length': '10' }) diff --git a/src/http/routes/tus/index.ts b/src/http/routes/tus/index.ts index b3d44198e..f3c723b25 100644 --- a/src/http/routes/tus/index.ts +++ b/src/http/routes/tus/index.ts @@ -290,42 +290,39 @@ export async function handleTusRequestWithIdleTimeout( res: FastifyReply ) { const socket = req.raw.socket - const isBodyFullyReceived = () => req.raw.complete || req.raw.readableEnded + const isDone = () => req.raw.complete || req.raw.readableEnded || req.raw.destroyed const hasDeclaredBody = req.raw.headers['transfer-encoding'] !== undefined || Number(req.raw.headers['content-length']) > 0 - if (!socket || tusBodyIdleTimeoutMs <= 0 || !hasDeclaredBody || isBodyFullyReceived()) { + if (!socket || tusBodyIdleTimeoutMs <= 0 || !hasDeclaredBody || isDone()) { return tusServer.handle(req.raw, res.raw) } - // We poll bytesRead on our own timer rather than socket.setTimeout(), - // which also triggers Fastify's onTimeout hook (disposing the db - // connection) on every check, or a 'data' listener, which would start - // flowing the stream and steal bytes from @tus/server's own consumer. + // We use our own timer instead of socket or stream events, so we don't interfere with how the body gets read + // bytesRead alone is not enough, because it also stalls when our own write to disk or S3 is slow + // bytesConsumed tracks the write side, so we can tell those two cases apart + // If either one is moving, the upload is alive let lastBytesRead = socket.bytesRead + let lastBytesConsumed = lastBytesRead - req.raw.readableLength let idleTimer: NodeJS.Timeout const disarm = () => { clearTimeout(idleTimer) req.raw.removeListener('end', disarm) + req.raw.removeListener('close', disarm) } const checkIdle = () => { - if (isBodyFullyReceived()) { + if (isDone()) { disarm() return } - if (socket.bytesRead > lastBytesRead) { - lastBytesRead = socket.bytesRead - idleTimer = setTimeout(checkIdle, tusBodyIdleTimeoutMs) - return - } - // bytesRead also stalls when our own write to the destination is back-pressured - // since Node stops reading the socket until the backlog drains. - // A non-zero readableLength means there's still unconsumed data sitting in the - // buffer, so the client did send something - if (req.raw.readableLength > 0) { + const currentBytesRead = socket.bytesRead + const currentBytesConsumed = currentBytesRead - req.raw.readableLength + if (currentBytesRead > lastBytesRead || currentBytesConsumed > lastBytesConsumed) { + lastBytesRead = currentBytesRead + lastBytesConsumed = currentBytesConsumed idleTimer = setTimeout(checkIdle, tusBodyIdleTimeoutMs) return } @@ -337,9 +334,11 @@ export async function handleTusRequestWithIdleTimeout( idleTimer = setTimeout(checkIdle, tusBodyIdleTimeoutMs) // Stop tracking idle time once the client has sent the full body, so - // slow lock acquisition or upload finalization afterward can't trip - // a "no bytes received" timeout. + // slow lock acquisition or upload finalization afterward can't trip a + // "no bytes received" timeout req.raw.once('end', disarm) + // A disconnect never fires 'end', so also disarm on 'close' + req.raw.once('close', disarm) try { return await tusServer.handle(req.raw, res.raw)