From 6696eca0765809abf47fbdb997c15978a2962acd Mon Sep 17 00:00:00 2001 From: assetcorp Date: Wed, 21 Oct 2026 17:35:54 +0000 Subject: [PATCH] perf: speed up middleware routes and whole-body uploads Routes behind middleware without a body reader now take the lazy fast path, and a plain 200 with one folded block of default headers goes out with one head write. bytes() collects the body inside the reader instead of through an async generator, the exchange calls the body reader directly instead of through listeners, and the framing scan allocates less. --- .../src/__tests__/e2e/request-bodies.test.ts | 89 +++++++++++++++++- .../src/__tests__/e2e/request-context.test.ts | 91 ++++++++++++------- packages/ts/src/body/reader.ts | 54 +++++++++-- packages/ts/src/body/request-body.ts | 26 +++--- packages/ts/src/chain.ts | 3 +- packages/ts/src/exchange/state.ts | 15 ++- packages/ts/src/headers.ts | 10 +- packages/ts/src/native/body.ts | 35 ++++--- packages/ts/src/native/direct.ts | 62 +++++++++++-- packages/ts/src/native/framing.ts | 19 ++-- packages/ts/src/native/handler.ts | 2 +- packages/ts/src/native/head.ts | 21 +++++ packages/ts/src/native/write.ts | 15 ++- packages/ts/src/runtime.ts | 13 +++ 14 files changed, 354 insertions(+), 101 deletions(-) diff --git a/packages/ts/src/__tests__/e2e/request-bodies.test.ts b/packages/ts/src/__tests__/e2e/request-bodies.test.ts index 55b0459..d367b48 100644 --- a/packages/ts/src/__tests__/e2e/request-bodies.test.ts +++ b/packages/ts/src/__tests__/e2e/request-bodies.test.ts @@ -1,12 +1,21 @@ +import { createHash, randomBytes } from 'node:crypto' import { describe, expect, it } from 'vitest' -import { createApp } from '../../index.js' +import { createApp, UhttpError } from '../../index.js' import { bodyApp, KIB, MIB } from '../helpers/apps.js' -import { chunkedRequest, openConnection, requestHead, requestWithBody, send } from '../helpers/http-client.js' +import { + chunkedRequest, + chunkFrame, + openConnection, + requestHead, + requestWithBody, + send, +} from '../helpers/http-client.js' import { expectClosedAfterReply, expectHello, piecesOf, SMUGGLED_REQUEST, + tooLarge, withEveryConnectionChoice, } from '../helpers/replies.js' import { pause, start } from '../helpers/server.js' @@ -40,6 +49,82 @@ describe('request bodies over the network', () => { await expectHello(connection) }) + it('reads a 512 KiB body that arrives in pieces while the handler waits on bytes(), and keeps the connection open', async () => { + const server = await start( + createApp() + .post('/digest', async c => { + const bytes = await c.req.body.bytes() + return c.json({ length: bytes.byteLength, sha256: createHash('sha256').update(bytes).digest('hex') }) + }) + .get('/hello', c => c.text('Hello')), + ) + const payload = randomBytes(512 * KIB) + const connection = await openConnection(server.port) + + void connection.write(requestHead('POST', '/digest', [`Content-Length: ${payload.byteLength}`])) + for (const piece of piecesOf(payload, 16)) { + await connection.write(piece) + await pause(5) + } + const response = await connection.response() + + expect(response.status).toBe(200) + expect(JSON.parse(response.body.toString())).toEqual({ + length: payload.byteLength, + sha256: createHash('sha256').update(payload).digest('hex'), + }) + await expectHello(connection) + }) + + it('fails bytes() with REQUEST_ABORTED when the client goes in the middle of the body, and serves the next client', async () => { + const failure = Promise.withResolvers() + const server = await start( + createApp() + .post('/upload', async c => { + try { + await c.req.body.bytes() + } catch (error) { + failure.resolve(error instanceof UhttpError ? error.code : error) + throw error + } + return c.text('read') + }) + .get('/hello', c => c.text('Hello')), + ) + const leaving = await openConnection(server.port) + + await leaving.write(requestHead('POST', '/upload', [`Content-Length: ${256 * KIB}`])) + await leaving.write(Buffer.alloc(64 * KIB, 97)) + await pause(20) + leaving.destroy() + + expect(await failure.promise).toBe('REQUEST_ABORTED') + await expectHello(await openConnection(server.port)) + }) + + it('answers 413 to a chunked body that passes the limit while the handler waits on bytes(), and closes the connection', async () => { + const handled: string[] = [] + const server = await start( + createApp({ maxBodyBytes: 100 }).post('/upload', async c => { + const bytes = await c.req.body.bytes() + handled.push(String(bytes.byteLength)) + return c.text('read') + }), + ) + const connection = await openConnection(server.port) + + void connection.write(requestHead('POST', '/upload', ['Transfer-Encoding: chunked'])) + for (let piece = 0; piece < 3; piece++) { + await connection.write(chunkFrame(Buffer.alloc(40, 97))) + await pause(10) + } + const response = await connection.response() + + expect([response.status, response.body.toString()]).toEqual([413, tooLarge(100)]) + expect(handled).toEqual([]) + await expectClosedAfterReply(connection, response) + }) + it('reads an empty body from a POST with no length and no chunked encoding', async () => { const server = await start( createApp() diff --git a/packages/ts/src/__tests__/e2e/request-context.test.ts b/packages/ts/src/__tests__/e2e/request-context.test.ts index dcc4cb6..dda5832 100644 --- a/packages/ts/src/__tests__/e2e/request-context.test.ts +++ b/packages/ts/src/__tests__/e2e/request-context.test.ts @@ -1,11 +1,24 @@ import { describe, expect, it, vi } from 'vitest' -import { createApp } from '../../index.js' +import { createApp, type Middleware } from '../../index.js' import { KIB, MIB } from '../helpers/apps.js' import { openConnection, requestHead, requestWithBody, send } from '../helpers/http-client.js' import { errorBody, expectClosedAfterReply, expectHello } from '../helpers/replies.js' import { memoryLogger, pause, start } from '../helpers/server.js' const INTERNAL_ERROR = errorBody('INTERNAL_ERROR', 'The request failed.') +const MIDDLEWARE_CHOICES: { route: string; middleware: Middleware[] }[] = [ + { route: 'on a route without middleware', middleware: [] }, + { route: 'behind middleware that passes it on at once', middleware: [(_c, next) => next()] }, + { + route: 'behind middleware that awaits before it passes it on', + middleware: [ + async (_c, next) => { + await pause(5) + await next() + }, + ], + }, +] describe('the copy from uWebSockets.js', () => { it('copies the method, the path, the query, the parameters, the content type, and the length before any await', async () => { @@ -184,31 +197,35 @@ describe('the query string and the content type, which the package reads when co await expectHello(connection) }) - it('copies the request for a handler that first reads it after an await, on a route without middleware', async () => { - const app = createApp() - .get('/later', async c => { - await pause(5) - const { method, path, route, query, contentType, contentLength } = c.req - return c.json({ method, path, route, query, contentType, contentLength: contentLength ?? null }) - }) - .get('/hello', c => c.text('Hello')) - const server = await start(app) + it.each(MIDDLEWARE_CHOICES)( + 'copies the request for a handler that first reads it after an await, $route', + async ({ middleware }) => { + const app = createApp() + .use(...middleware) + .get('/later', async c => { + await pause(5) + const { method, path, route, query, contentType, contentLength } = c.req + return c.json({ method, path, route, query, contentType, contentLength: contentLength ?? null }) + }) + .get('/hello', c => c.text('Hello')) + const server = await start(app) - const { connection, response } = await send( - server.port, - requestHead('GET', '/later?page=3', ['Content-Type: text/csv']), - ) + const { connection, response } = await send( + server.port, + requestHead('GET', '/later?page=3', ['Content-Type: text/csv']), + ) - expect(JSON.parse(response.body.toString())).toEqual({ - method: 'GET', - path: '/later', - route: '/later', - query: 'page=3', - contentType: 'text/csv', - contentLength: null, - }) - await expectHello(connection) - }) + expect(JSON.parse(response.body.toString())).toEqual({ + method: 'GET', + path: '/later', + route: '/later', + query: 'page=3', + contentType: 'text/csv', + contentLength: null, + }) + await expectHello(connection) + }, + ) it('gives a task that first reads the request after the reply its method, path, and route, and REQUEST_RELEASED for the rest', async () => { const outcome = Promise.withResolvers() @@ -241,17 +258,21 @@ describe('the query string and the content type, which the package reads when co await expectHello(connection) }) - it('reads the length of a body that a client sends to a GET route without middleware, and keeps the connection open', async () => { - const app = createApp() - .get('/sized', c => c.json({ contentLength: c.req.contentLength ?? null, query: c.req.query })) - .get('/hello', c => c.text('Hello')) - const server = await start(app) - - const { connection, response } = await send(server.port, requestWithBody('GET', '/sized?x=1', [], 'hello')) - - expect(JSON.parse(response.body.toString())).toEqual({ contentLength: 5, query: 'x=1' }) - await expectHello(connection) - }) + it.each(MIDDLEWARE_CHOICES)( + 'reads the length of a body that a client sends to a GET route $route, and keeps the connection open', + async ({ middleware }) => { + const app = createApp() + .use(...middleware) + .get('/sized', c => c.json({ contentLength: c.req.contentLength ?? null, query: c.req.query })) + .get('/hello', c => c.text('Hello')) + const server = await start(app) + + const { connection, response } = await send(server.port, requestWithBody('GET', '/sized?x=1', [], 'hello')) + + expect(JSON.parse(response.body.toString())).toEqual({ contentLength: 5, query: 'x=1' }) + await expectHello(connection) + }, + ) it('copies both for a stream producer that reads them after the handler returned', async () => { const app = createApp() diff --git a/packages/ts/src/body/reader.ts b/packages/ts/src/body/reader.ts index 931a427..5a06a15 100644 --- a/packages/ts/src/body/reader.ts +++ b/packages/ts/src/body/reader.ts @@ -6,8 +6,8 @@ import type { BodyState } from '../runtime.js' const IDENTITY = 'identity' -interface Waiter { - readonly resolve: (chunk: Uint8Array | undefined) => void +interface Waiter { + readonly resolve: (value: T) => void readonly reject: (error: unknown) => void } @@ -113,7 +113,8 @@ export class BodyReader implements BodyState { #paused = false #flowing = false #error: UhttpError | undefined - #waiter: Waiter | undefined + #waiter: Waiter | undefined + #collector: Waiter | undefined #pauseTimer: ReturnType | undefined /** @@ -130,8 +131,6 @@ export class BodyReader implements BodyState { this.#source = source this.#ended = empty exchange.body = this - exchange.onRelease(error => this.fail(error)) - exchange.onClose(() => this.#closed()) } /** This is `true` while the reader queues the chunks that the source delivers. */ @@ -153,6 +152,7 @@ export class BodyReader implements BodyState { if (!last) return this.#ended = true this.#wake() + this.#deliver() this.#exchange.arrival.ended() } @@ -167,8 +167,11 @@ export class BodyReader implements BodyState { this.#queue.length = 0 this.#queuedBytes = 0 const waiter = this.#waiter + const collector = this.#collector this.#waiter = undefined + this.#collector = undefined waiter?.reject(error) + collector?.reject(error) } /** @@ -193,6 +196,24 @@ export class BodyReader implements BodyState { }) } + /** + * Returns the whole body once its last chunk arrives. While the reader collects the body, it + * never pauses the source, because it keeps every chunk until the end of the body anyway, and + * the body limit of the route still applies. The caller reads the body either through this + * method once or through {@link BodyReader.read}, never both. + * + * @returns The promise resolves to the bytes of the body. It rejects with the error of a failed + * reader. + */ + collect(): Promise { + if (this.#error !== undefined) return Promise.reject(this.#error) + this.keepFlowing() + if (this.#ended) return Promise.resolve(this.#drained()) + return new Promise((resolve, reject) => { + this.#collector = { resolve, reject } + }) + } + /** * Discards the chunks that the handler hasn't read and every chunk that arrives later, and * resumes a paused source, so that the rest of the body keeps arriving after the response starts. @@ -200,6 +221,7 @@ export class BodyReader implements BodyState { settle(): void { if (this.#paused) this.#resume() this.#discarding = true + if (this.#collector !== undefined) return this.#queue.length = 0 this.#queuedBytes = 0 } @@ -240,6 +262,22 @@ export class BodyReader implements BodyState { waiter.resolve(chunk) } + #deliver(): void { + const collector = this.#collector + if (collector === undefined) return + this.#collector = undefined + collector.resolve(this.#drained()) + } + + #drained(): Uint8Array { + const queue = this.#queue + const first = queue[0] + const whole = queue.length === 1 && first !== undefined ? first : Buffer.concat(queue, this.#queuedBytes) + queue.length = 0 + this.#queuedBytes = 0 + return whole + } + #pause(): void { if (this.#paused || this.#flowing) return this.#paused = true @@ -256,7 +294,11 @@ export class BodyReader implements BodyState { this.#source.resume() } - #closed(): void { + /** + * Stops the pause timer once the client goes or the response ends, and fails a reader whose body + * has not ended with `REQUEST_ABORTED`. + */ + close(): void { this.#stopTimer() if (!this.#ended) this.fail(new UhttpError('REQUEST_ABORTED', 'The response ended before the request body.')) } diff --git a/packages/ts/src/body/request-body.ts b/packages/ts/src/body/request-body.ts index 09067f8..5696669 100644 --- a/packages/ts/src/body/request-body.ts +++ b/packages/ts/src/body/request-body.ts @@ -119,16 +119,6 @@ async function* chunksOf(reader: BodyReader): AsyncGenerator { } } -async function collect(reader: BodyReader): Promise { - const chunks: Uint8Array[] = [] - let total = 0 - for await (const chunk of chunksOf(reader)) { - chunks.push(chunk) - total += chunk.byteLength - } - return chunks.length === 1 && chunks[0] !== undefined ? chunks[0] : Buffer.concat(chunks, total) -} - /** * The package creates this body for each request on the first read of `c.req.body`. * @@ -162,12 +152,18 @@ export class ReaderBody implements RequestBody { return chunksOf(reader) } - async bytes(): Promise { - const reader = this.#claimed() - if (this.#bytes === undefined) { - if (this.#streamed) throw new UhttpError('BODY_ALREADY_READ', 'Your code has already streamed this body.') - this.#bytes = collect(reader) + bytes(): Promise { + if (this.#bytes !== undefined) return this.#bytes + let reader: BodyReader + try { + reader = this.#claimed() + } catch (error) { + return Promise.reject(error) + } + if (this.#streamed) { + return Promise.reject(new UhttpError('BODY_ALREADY_READ', 'Your code has already streamed this body.')) } + this.#bytes = reader.collect() return this.#bytes } diff --git a/packages/ts/src/chain.ts b/packages/ts/src/chain.ts index 356a0da..a7f1a77 100644 --- a/packages/ts/src/chain.ts +++ b/packages/ts/src/chain.ts @@ -254,6 +254,7 @@ export function runChain( ): undefined | Promise { const result = dispatch(context, exchange, layers, handler, 0) if (result === undefined) return ensureReply(context, exchange) + if (layers.length === 0) return result return ensureReplyLater(context, exchange, result) } @@ -275,7 +276,7 @@ export function settleResult(context: Context, exchange: Exchange, result: unkno return undefined } if (!(result instanceof Promise)) return settled(context, exchange, fail(context, exchange, missingReply())) - return ensureReplyLater(context, exchange, settleHandler(context, exchange, result)) + return settleHandler(context, exchange, result) } /** diff --git a/packages/ts/src/exchange/state.ts b/packages/ts/src/exchange/state.ts index 8e37447..c2562e6 100644 --- a/packages/ts/src/exchange/state.ts +++ b/packages/ts/src/exchange/state.ts @@ -207,8 +207,11 @@ export class Exchange implements ResponseParts { if (this.closed) return this.mark(ABORTED | CLOSED, true) const events = this.ensureEvents() + const error = new UhttpError('REQUEST_ABORTED', 'The client disconnected before the request body ended.') events.fireSignal(undefined) - events.release(new UhttpError('REQUEST_ABORTED', 'The client disconnected before the request body ended.')) + this.body?.fail(error) + events.release(error) + this.body?.close() events.fireClose() } @@ -235,13 +238,14 @@ export class Exchange implements ResponseParts { this.mark(TIMED_OUT | CLOSE_AFTER_REPLY, true) const events = this.ensureEvents() events.fireSignal(error) + this.body?.fail(error) events.release(error) } /** - * Returns `true` when only the reply shapes the response: the route has no body reader, no body - * is on its way, your code set no header, no middleware attached default headers, and the - * connection stays open after the response. + * Returns `true` when only the reply and the default headers of middleware shape the response: + * the route has no body reader, no body is on its way, your code set no header, and the + * connection stays open after the response. The caller checks the default headers itself. * * @returns The method returns `false` when the package writes the reply through its full path. */ @@ -251,7 +255,6 @@ export class Exchange implements ResponseParts { this.body === undefined && this.arrivalState === undefined && this.headerStore === undefined && - this.defaultBlocks === undefined && (this.drain === undefined || this.drain.mode === SERVING_MODE) ) } @@ -272,6 +275,7 @@ export class Exchange implements ResponseParts { * @param error - The package passes the error that the body reader and the uploads fail with. */ release(error: UhttpError): void { + this.body?.fail(error) this.ensureEvents().release(error) } @@ -291,6 +295,7 @@ export class Exchange implements ResponseParts { if ((flags & CLOSED) !== 0) return this.flagBits = flags | CLOSED if ((flags & (LINGERING | COUNTED)) === COUNTED) this.uncount() + if (this.body !== undefined) this.body.close() if (this.events !== undefined) this.events.fireClose() } diff --git a/packages/ts/src/headers.ts b/packages/ts/src/headers.ts index 635fc21..26289b4 100644 --- a/packages/ts/src/headers.ts +++ b/packages/ts/src/headers.ts @@ -2,7 +2,8 @@ import { COOKIE_HEADER_SEPARATOR, FRAMING_HEADER_NAMES, LIST_HEADER_SEPARATOR } import { UhttpError } from './errors.js' const TOKEN = /^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$/ -const DECIMAL_DIGITS = /^\d+$/ +const DIGIT_ZERO_CODE = 0x30 +const DIGIT_NINE_CODE = 0x39 const SPACE_CODE = 0x20 const TAB_CODE = 0x09 const DELETE_CODE = 0x7f @@ -45,7 +46,12 @@ export function isToken(text: string): boolean { * @internal */ export function isDecimalDigits(text: string): boolean { - return DECIMAL_DIGITS.test(text) + if (text.length === 0) return false + for (let index = 0; index < text.length; index++) { + const code = text.charCodeAt(index) + if (code < DIGIT_ZERO_CODE || code > DIGIT_NINE_CODE) return false + } + return true } /** diff --git a/packages/ts/src/native/body.ts b/packages/ts/src/native/body.ts index 3622389..f7bca7a 100644 --- a/packages/ts/src/native/body.ts +++ b/packages/ts/src/native/body.ts @@ -7,15 +7,25 @@ import type { RequestFraming } from './framing.js' const EMPTY = new Uint8Array(0) -function sourceOf(res: HttpResponse, exchange: Exchange): BodySource { - return { - pause: () => { - if (!exchange.closed) res.pause() - }, - resume: () => { - if (!exchange.closed) res.resume() - }, - close: () => closeConnection(res, exchange), +class NativeSource implements BodySource { + readonly #res: HttpResponse + readonly #exchange: Exchange + + constructor(res: HttpResponse, exchange: Exchange) { + this.#res = res + this.#exchange = exchange + } + + pause(): void { + if (!this.#exchange.closed) this.#res.pause() + } + + resume(): void { + if (!this.#exchange.closed) this.#res.resume() + } + + close(): void { + closeConnection(this.#res, this.#exchange) } } @@ -59,10 +69,13 @@ export function attachBody( } function attachReader(res: HttpResponse, exchange: Exchange, chain: RequestChain, framing: RequestFraming): OpenedBody { - const opened = openBody(exchange, chain, framing, sourceOf(res, exchange)) + const opened = openBody(exchange, chain, framing, new NativeSource(res, exchange)) const { reader } = opened if (opened.flowing && reader !== undefined) { - listenForBody(res, exchange, framing, (chunk, isLast) => { + const { arrival } = exchange + arrival.expect(declaredLength(framing)) + res.onData((chunk, isLast) => { + arrival.received(chunk.byteLength) reader.push(reader.accepting && chunk.byteLength > 0 ? new Uint8Array(chunk.slice(0)) : EMPTY, isLast) }) } else if (framing.declaresBody) { diff --git a/packages/ts/src/native/direct.ts b/packages/ts/src/native/direct.ts index 54a9bc5..889be8b 100644 --- a/packages/ts/src/native/direct.ts +++ b/packages/ts/src/native/direct.ts @@ -1,5 +1,5 @@ import type { HttpRequest, HttpResponse } from 'uWebSockets.js' -import { settleError, settleResult } from '../chain.js' +import { runChain, settleError, settleResult } from '../chain.js' import type { CompiledApp, CompiledRoute } from '../compile.js' import { type Context, createContext, createLazyContext, releaseRequest } from '../context.js' import { Exchange } from '../exchange/index.js' @@ -55,7 +55,29 @@ function crashed(res: HttpResponse, exchange: Exchange, error: unknown, method: if (!crash(res, exchange, error, method, path)) watchAbort(res, exchange) } +function writePlainly( + res: HttpResponse, + exchange: Exchange, + context: Context, + okHead: ArrayBuffer, + reply: Reply, +): void { + try { + res.writeStatus(okHead) + res.end(reply.body ?? '') + } catch (error) { + abandon(res, exchange, error, context) + releaseRequest(context, false) + return + } + exchange.end() + releaseRequest(context, false) +} + function answer(res: HttpResponse, exchange: Exchange, context: Context, route: CompiledRoute, head: boolean): boolean { + if (route.layers.length > 0) { + return answerLater(res, exchange, context, route, runChain(context, exchange, route.layers, route.handler)) + } let result: unknown try { result = route.handler(context) @@ -96,13 +118,13 @@ function copiesLazily(route: CompiledRoute): boolean { } /** - * Returns the uWebSockets.js handler of a route without middleware, without path parameters, and - * without a body reader. The handler makes the same framing checks as the handler of any other route, - * and sends the request straight to the handler of the route. When the route declares no header and no - * client address, and the request declares no body, the package copies the request only when your - * code first reads `c.req`, and a plain 200 with a text or a JSON body goes out with one call into - * uWebSockets.js for its head. V8 compiles that path into one function, so the package keeps every - * step of it inside the handler. + * Returns the uWebSockets.js handler of a route without path parameters and without a body reader. + * The handler makes the same framing checks as the handler of any other route, and sends the + * request straight to the handler of the route, or through its middleware when it has any. When the + * route and its middleware declare no header and no client address, and the request declares no + * body, the package copies the request only when your code first reads `c.req`, and a plain 200 + * with a text or a JSON body goes out with one call into uWebSockets.js for its head. V8 compiles + * that path into one function, so the package keeps every step of it inside the handler. * * @param app - The package passes the compiled app. * @param route - The package passes the compiled route. @@ -157,6 +179,28 @@ export function directHandler( if (!ended) watchAbort(res, exchange) } if (!lazy) return (res, req) => checked(res, req, undefined) + if (route.layers.length > 0) { + const layers = route.layers + return (res, req) => { + if (drain.mode === CLOSING || refusesAnyConnection()) return checked(res, req, undefined) + let framing: RequestFraming + try { + framing = readFraming(req) + } catch (error) { + return crashed(res, new Exchange(app.runtime, drain), error, method, route.path) + } + if (framing !== NO_BODY) return checked(res, req, framing) + const exchange = new Exchange(app.runtime, drain) + const context = createLazyContext(copy, exchange, req) + const pending = runChain(context, exchange, layers, route.handler) + const reply = exchange.reply + const okHead = pending !== undefined || head || reply === undefined ? undefined : plainHeadOf(exchange, reply) + if (okHead === undefined || reply === undefined || exchange.closed || !exchange.writesReplyAlone()) { + return settle(res, exchange, context, route, pending) + } + writePlainly(res, exchange, context, okHead, reply) + } + } return (res, req) => { if (drain.mode === CLOSING || refusesAnyConnection()) return checked(res, req, undefined) let framing: RequestFraming @@ -177,7 +221,7 @@ export function directHandler( if (!(result instanceof Reply)) return settle(res, exchange, context, route, settleResult(context, exchange, result)) exchange.assign(result) - const okHead = head ? undefined : plainHeadOf(result) + const okHead = head ? undefined : plainHeadOf(exchange, result) if (okHead === undefined || exchange.closed || !exchange.writesReplyAlone()) { return settle(res, exchange, context, route, undefined) } diff --git a/packages/ts/src/native/framing.ts b/packages/ts/src/native/framing.ts index 7814bfd..a92b208 100644 --- a/packages/ts/src/native/framing.ts +++ b/packages/ts/src/native/framing.ts @@ -67,21 +67,22 @@ export function readFraming(req: HttpRequest): RequestFraming { function readBodyFraming(req: HttpRequest, contentLength: string, transferEncoding: string): RequestFraming { const lengths: string[] = [] const encodings: string[] = [] - const codings: string[] = [] + let contentEncoding: string | undefined req.forEach((name, value) => { if (name === 'content-length') lengths.push(value) else if (name === 'transfer-encoding') encodings.push(value) - else if (name === 'content-encoding' && value !== '') codings.push(value) + else if (name === 'content-encoding' && value !== '') { + contentEncoding = contentEncoding === undefined ? value : `${contentEncoding}, ${value}` + } }) const parserRejects = parserRejectsFraming(lengths, encodings) - const lines = { - contentLength: contentLength === '' ? undefined : contentLength, - transferEncoding: transferEncoding === '' ? undefined : transferEncoding, - contentEncoding: codings.length === 0 ? undefined : codings.join(', '), - } + const length = contentLength === '' ? undefined : contentLength + const encoding = transferEncoding === '' ? undefined : transferEncoding return { - ...lines, - declaresBody: !declaresNoBody(lines), + contentLength: length, + transferEncoding: encoding, + contentEncoding, + declaresBody: !declaresNoBody({ contentLength: length, transferEncoding: encoding }), clientCloses: req.getHeader(CONNECTION).length === CONNECTION_CLOSE_VALUE_LENGTH, parserRejects, rejection: parserRejects ? undefined : transferCodingError(encodings), diff --git a/packages/ts/src/native/handler.ts b/packages/ts/src/native/handler.ts index 35a6ac5..86a6495 100644 --- a/packages/ts/src/native/handler.ts +++ b/packages/ts/src/native/handler.ts @@ -82,7 +82,7 @@ export function routeHandler(app: CompiledApp, route: CompiledRoute, method: str limits: route.limits, } const withoutParameters: DecodedChain = { chain: base, parameters: NO_PARAMETERS } - if (route.layers.length === 0 && !route.limits.readsBody && count === 0) { + if (!route.limits.readsBody && count === 0) { return directHandler(app, route, method, drain, withoutParameters) } const answer = (res: HttpResponse, req: HttpRequest, exchange: Exchange): boolean => { diff --git a/packages/ts/src/native/head.ts b/packages/ts/src/native/head.ts index cd14950..abcec59 100644 --- a/packages/ts/src/native/head.ts +++ b/packages/ts/src/native/head.ts @@ -63,6 +63,27 @@ function foldedHeadOf(block: DefaultHeaders, heads: Map, co return built } +/** + * Returns the head of a plain 200 that also holds one block of default headers, which a + * middleware built once at startup. The package builds the bytes of that head once for each + * content type and keeps them on the block. + * + * @param blocks - The package passes the blocks of default headers of the request. + * @param contentType - The package passes the content type of the reply. + * @returns The function returns the head, or `undefined` when the request has more than one block, + * when the block changes for each request, or when the content type is neither text nor JSON. + * @internal + */ +export function foldedPlainHeadOf( + blocks: readonly DefaultHeaders[], + contentType: string | undefined, +): ArrayBuffer | undefined { + const block = blocks[0] + if (blocks.length !== 1 || block?.heads === undefined) return undefined + if (contentType === undefined || okHeadOf(contentType) === undefined) return undefined + return foldedHeadOf(block, block.heads, contentType) +} + /** * Calls a function with the content type and then with each line of every other header of the * current reply. diff --git a/packages/ts/src/native/write.ts b/packages/ts/src/native/write.ts index 985255f..ccc71d1 100644 --- a/packages/ts/src/native/write.ts +++ b/packages/ts/src/native/write.ts @@ -7,7 +7,7 @@ import { bodyByteLength, errorReply, type Reply, statusHasBody } from '../reply. import { followBody } from './body.js' import { closeConnection, watchAbort } from './close.js' import type { RequestFraming } from './framing.js' -import { okHeadOf, writeHead } from './head.js' +import { foldedPlainHeadOf, okHeadOf, writeHead } from './head.js' import { writeLingeringReply } from './linger.js' import { refuseLaterRequests } from './refuse.js' import { NativeStream } from './stream.js' @@ -49,15 +49,20 @@ export function writeReply(res: HttpResponse, exchange: Exchange, context: Conte /** * Returns the head that uWebSockets.js writes in one call for a plain 200, which has no stream, no - * header of its own, and a text or a JSON content type. + * header of its own, and a text or a JSON content type. When a middleware added one block of + * default headers that it built at startup, such as the block of `secureHeaders()`, the head holds + * that block too. * + * @param exchange - The package passes the state of the request, which holds the default headers. * @param reply - The package passes the reply to write. * @returns The function returns the head of the reply, or `undefined` when the reply is not plain. * @internal */ -export function plainHeadOf(reply: Reply): ArrayBuffer | undefined { +export function plainHeadOf(exchange: Exchange, reply: Reply): ArrayBuffer | undefined { if (reply.status !== OK_STATUS || reply.stream !== undefined || reply.headers !== undefined) return undefined - return okHeadOf(reply.contentType) + const blocks = exchange.defaultHeaders + if (blocks === undefined) return okHeadOf(reply.contentType) + return foldedPlainHeadOf(blocks, reply.contentType) } /** @@ -73,7 +78,7 @@ export function plainHeadOf(reply: Reply): ArrayBuffer | undefined { * @internal */ export function writeReplyAlone(res: HttpResponse, exchange: Exchange, reply: Reply): boolean { - const okHead = plainHeadOf(reply) + const okHead = plainHeadOf(exchange, reply) if (okHead === undefined || !exchange.writesReplyAlone()) return false res.writeStatus(okHead) res.end(reply.body ?? '') diff --git a/packages/ts/src/runtime.ts b/packages/ts/src/runtime.ts index 3391566..8ddf582 100644 --- a/packages/ts/src/runtime.ts +++ b/packages/ts/src/runtime.ts @@ -1,3 +1,4 @@ +import type { UhttpError } from './errors.js' import type { ErrorHandler, Handler, Logger } from './types.js' import type { PubSubAdapter } from './websocket/types.js' @@ -40,4 +41,16 @@ export interface BodyState { * of the route. */ keepFlowing(): void + /** + * Fails every read from now on with an error. The package calls this method when the client + * disconnects or the handler passes its time limit. + * + * @param error - The package passes the error that every read rejects with. + */ + fail(error: UhttpError): void + /** + * Stops the reader once the client goes or the response ends, and fails it when its body has + * not ended. + */ + close(): void }