From b26748a2a75c515a40a916158261914e9790ecdb Mon Sep 17 00:00:00 2001 From: Hampus Date: Sun, 30 Aug 2026 22:53:00 +0200 Subject: [PATCH] fix(api): bound http server limits and shed on overload (#2170) --- fluxer_api/src/api/App.ts | 1 + fluxer_api/src/api/Config.ts | 1 + fluxer_api/src/api/app/MiddlewarePipeline.ts | 5 +- fluxer_api/src/api/config/APIConfig.ts | 1 + .../middleware/ConcurrencyLimitMiddleware.ts | 34 +++++++ .../tests/ConcurrencyLimitMiddleware.test.ts | 97 +++++++++++++++++++ packages/config/src/ConfigLoader.ts | 2 + packages/config/src/MasterConfig.ts | 1 + .../src/config_loader/EnvironmentOverrides.ts | 1 + packages/hono/src/Server.ts | 26 ++++- packages/hono/src/tests/ServerLimits.test.ts | 64 ++++++++++++ 11 files changed, 230 insertions(+), 3 deletions(-) create mode 100644 fluxer_api/src/api/middleware/ConcurrencyLimitMiddleware.ts create mode 100644 fluxer_api/src/api/middleware/tests/ConcurrencyLimitMiddleware.test.ts create mode 100644 packages/hono/src/tests/ServerLimits.test.ts diff --git a/fluxer_api/src/api/App.ts b/fluxer_api/src/api/App.ts index 9d982b505..284961fe8 100644 --- a/fluxer_api/src/api/App.ts +++ b/fluxer_api/src/api/App.ts @@ -48,6 +48,7 @@ export async function createAPIApp(options: CreateAPIAppOptions): Promise; trustClientIpHeader: boolean; clientIpHeaderName?: string; + maxInflightRequests: number; } export function configureMiddleware(routes: HonoApp, options: MiddlewarePipelineOptions): void { - const {logger, nodeEnv, corsOrigins, trustClientIpHeader, clientIpHeaderName} = options; + const {logger, nodeEnv, corsOrigins, trustClientIpHeader, clientIpHeaderName, maxInflightRequests} = options; const resolvedHeader = resolveClientIpHeaderName(clientIpHeaderName); routes.use('/webhooks/:webhook_id/:token', cors({origins: '*'})); routes.use('/webhooks/:webhook_id/:token/messages/:message_id', cors({origins: '*'})); @@ -40,6 +42,7 @@ export function configureMiddleware(routes: HonoApp, options: MiddlewarePipeline skipLogger: true, skipErrorHandler: true, }); + routes.use(ConcurrencyLimitMiddleware({maxInflightRequests})); routes.get('/_health', async (ctx) => ctx.text('OK')); routes.use(IpBanMiddleware); routes.use( diff --git a/fluxer_api/src/api/config/APIConfig.ts b/fluxer_api/src/api/config/APIConfig.ts index 26e3d8b18..8d498b992 100644 --- a/fluxer_api/src/api/config/APIConfig.ts +++ b/fluxer_api/src/api/config/APIConfig.ts @@ -34,6 +34,7 @@ export type APIGeoipConfig = APIGeoipFilesystemConfig | APIGeoipS3Config; export interface APIConfig { nodeEnv: 'development' | 'production'; port: number; + maxInflightRequests: number; ipBanExemptIps: Array; desktopGitHubRedirectCountries: ReadonlySet; cassandra: { diff --git a/fluxer_api/src/api/middleware/ConcurrencyLimitMiddleware.ts b/fluxer_api/src/api/middleware/ConcurrencyLimitMiddleware.ts new file mode 100644 index 000000000..d8200b2fd --- /dev/null +++ b/fluxer_api/src/api/middleware/ConcurrencyLimitMiddleware.ts @@ -0,0 +1,34 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {ServiceUnavailableError} from '@fluxer/errors/src/domains/core/ServiceUnavailableError'; +import {createMiddleware} from 'hono/factory'; +import type {HonoEnv} from '../types/HonoEnv'; +import {normalizeRequestPath} from '../utils/RequestPathUtils'; + +const PROBE_PATHS = new Set(['/_health', '/_healthz', '/_metrics']); +const OVERLOAD_RETRY_AFTER_SECONDS = 1; + +interface ConcurrencyLimitOptions { + maxInflightRequests: number; +} + +export function ConcurrencyLimitMiddleware({maxInflightRequests}: ConcurrencyLimitOptions) { + let inflight = 0; + return createMiddleware(async (ctx, next) => { + if (PROBE_PATHS.has(normalizeRequestPath(ctx.req.path))) { + await next(); + return; + } + inflight += 1; + try { + if (inflight > maxInflightRequests) { + throw new ServiceUnavailableError({ + headers: {'Retry-After': String(OVERLOAD_RETRY_AFTER_SECONDS)}, + }); + } + await next(); + } finally { + inflight -= 1; + } + }); +} diff --git a/fluxer_api/src/api/middleware/tests/ConcurrencyLimitMiddleware.test.ts b/fluxer_api/src/api/middleware/tests/ConcurrencyLimitMiddleware.test.ts new file mode 100644 index 000000000..e1a24e092 --- /dev/null +++ b/fluxer_api/src/api/middleware/tests/ConcurrencyLimitMiddleware.test.ts @@ -0,0 +1,97 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {AppErrorHandler} from '@fluxer/errors/src/domains/core/ErrorHandlers'; +import {Hono} from 'hono'; +import {describe, expect, test} from 'vitest'; +import type {HonoEnv} from '../../types/HonoEnv'; +import {ConcurrencyLimitMiddleware} from '../ConcurrencyLimitMiddleware'; + +interface Deferred { + promise: Promise; + resolve: () => void; +} + +function createDeferred(): Deferred { + let resolve: () => void = () => undefined; + const promise = new Promise((resolveFn) => { + resolve = resolveFn; + }); + return {promise, resolve}; +} + +function buildApp(maxInflightRequests: number) { + const release = createDeferred(); + const entered = createDeferred(); + const routes = new Hono({strict: true}); + routes.use(ConcurrencyLimitMiddleware({maxInflightRequests})); + routes.get('/_health', (ctx) => ctx.text('OK')); + routes.get('/_metrics', (ctx) => ctx.text('metrics')); + routes.get('/fast', (ctx) => ctx.text('fast')); + routes.get('/slow', async (ctx) => { + entered.resolve(); + await release.promise; + return ctx.text('slow'); + }); + routes.get('/boom', () => { + throw new Error('boom'); + }); + routes.onError(AppErrorHandler); + const app = new Hono({strict: true}); + app.route('/v1', routes); + app.route('/', routes); + app.onError(AppErrorHandler); + return {app, release, entered}; +} + +describe('ConcurrencyLimitMiddleware', () => { + test('sheds with 503 once the in-flight ceiling is reached', async () => { + const {app, release, entered} = buildApp(1); + const inflight = app.request('/slow'); + await entered.promise; + + const shed = await app.request('/fast'); + + expect(shed.status).toBe(503); + expect(shed.headers.get('Retry-After')).toBe('1'); + expect(await shed.json()).toMatchObject({code: 'SERVICE_UNAVAILABLE'}); + + release.resolve(); + expect((await inflight).status).toBe(200); + }); + + test('admits probe paths while shedding everything else', async () => { + const {app, release, entered} = buildApp(1); + const inflight = app.request('/slow'); + await entered.promise; + + expect((await app.request('/_health')).status).toBe(200); + expect((await app.request('/_metrics')).status).toBe(200); + expect((await app.request('/v1/_health')).status).toBe(200); + expect((await app.request('/fast')).status).toBe(503); + + release.resolve(); + await inflight; + }); + + test('releases the slot when the handler throws', async () => { + const {app} = buildApp(1); + for (let attempt = 0; attempt < 3; attempt += 1) { + expect((await app.request('/boom')).status).toBe(500); + } + expect((await app.request('/fast')).status).toBe(200); + }); + + test('releases the slot taken by a shed request', async () => { + const {app, release, entered} = buildApp(1); + const inflight = app.request('/slow'); + await entered.promise; + + expect((await app.request('/fast')).status).toBe(503); + expect((await app.request('/fast')).status).toBe(503); + + release.resolve(); + await inflight; + + expect((await app.request('/fast')).status).toBe(200); + }); +}); diff --git a/packages/config/src/ConfigLoader.ts b/packages/config/src/ConfigLoader.ts index 447f934dd..6fe38c39b 100644 --- a/packages/config/src/ConfigLoader.ts +++ b/packages/config/src/ConfigLoader.ts @@ -91,6 +91,7 @@ function defaultConfig(): MasterConfig { services: { api: { port: 8080, + max_inflight_requests: 512, ip_ban_exempt_ips: [], desktop_github_redirect_countries: [], presigned_attachment_uploads_enabled: false, @@ -408,6 +409,7 @@ function normalizeConfig(config: MasterConfig): MasterConfig { ); validatePostgresConfig(config); validateApiWorkerConfig(config); + assertIntegerInRange(config.services.api.max_inflight_requests, 'FLUXER_API_MAX_INFLIGHT_REQUESTS', 1, 100_000); requireString(config.domain.base_domain, 'FLUXER_BASE_DOMAIN'); requireString(config.auth.sudo_mode_secret, 'FLUXER_SUDO_MODE_SECRET'); requireString(config.auth.connection_initiation_secret, 'FLUXER_CONNECTION_INITIATION_SECRET'); diff --git a/packages/config/src/MasterConfig.ts b/packages/config/src/MasterConfig.ts index 031a94f8f..c29cfbd66 100644 --- a/packages/config/src/MasterConfig.ts +++ b/packages/config/src/MasterConfig.ts @@ -90,6 +90,7 @@ export interface MasterConfig { services: { api: { port: number; + max_inflight_requests: number; ip_ban_exempt_ips: Array; desktop_github_redirect_countries: Array; presigned_attachment_uploads_enabled: boolean; diff --git a/packages/config/src/config_loader/EnvironmentOverrides.ts b/packages/config/src/config_loader/EnvironmentOverrides.ts index a3f4ccc60..8b33e579a 100644 --- a/packages/config/src/config_loader/EnvironmentOverrides.ts +++ b/packages/config/src/config_loader/EnvironmentOverrides.ts @@ -76,6 +76,7 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record = { FLUXER_NATS_JETSTREAM_URL: {path: ['services', 'nats', 'jetstream_url']}, FLUXER_NATS_AUTH_TOKEN: {path: ['services', 'nats', 'auth_token']}, FLUXER_API_PORT: {path: ['services', 'api', 'port'], parse: parseEnvValue}, + FLUXER_API_MAX_INFLIGHT_REQUESTS: {path: ['services', 'api', 'max_inflight_requests'], parse: parseEnvValue}, FLUXER_API_IP_BAN_EXEMPT_IPS: {path: ['services', 'api', 'ip_ban_exempt_ips'], parse: parseCsv}, FLUXER_API_DESKTOP_GITHUB_REDIRECT_COUNTRIES: { path: ['services', 'api', 'desktop_github_redirect_countries'], diff --git a/packages/hono/src/Server.ts b/packages/hono/src/Server.ts index 0feae9d1e..8f59ec150 100644 --- a/packages/hono/src/Server.ts +++ b/packages/hono/src/Server.ts @@ -4,10 +4,17 @@ import {applyFluxerVersionHeader} from '@fluxer/hono/src/middleware/VersionHeade import {type Http2Bindings, type HttpBindings, type ServerType, serve} from '@hono/node-server'; import type {Env, Hono} from 'hono'; +const DEFAULT_REQUEST_TIMEOUT_MS = 30_000; +const DEFAULT_KEEP_ALIVE_TIMEOUT_MS = 125_000; +const DEFAULT_MAX_REQUESTS_PER_SOCKET = 1_000; + interface ServerOptions { port: number; hostname?: string; onListen?: (info: {address: string; port: number}) => void; + requestTimeoutMs?: number; + keepAliveTimeoutMs?: number; + maxRequestsPerSocket?: number; } type NodeFetchCallback = (request: Request, env: HttpBindings | Http2Bindings) => Promise | unknown; @@ -17,15 +24,30 @@ function createVersionedFetch(app: Hono): NodeFetchCallback { } export function createServer(app: Hono, options: ServerOptions): ServerType { - const {port, hostname, onListen} = options; - return serve( + const { + port, + hostname, + onListen, + requestTimeoutMs = DEFAULT_REQUEST_TIMEOUT_MS, + keepAliveTimeoutMs = DEFAULT_KEEP_ALIVE_TIMEOUT_MS, + maxRequestsPerSocket = DEFAULT_MAX_REQUESTS_PER_SOCKET, + } = options; + const server = serve( { fetch: createVersionedFetch(app), port, ...(hostname !== undefined && {hostname}), + serverOptions: { + keepAliveTimeout: keepAliveTimeoutMs, + requestTimeout: requestTimeoutMs, + }, }, onListen, ); + if ('maxRequestsPerSocket' in server) { + server.maxRequestsPerSocket = maxRequestsPerSocket; + } + return server; } type CleanupFunction = () => void | Promise; diff --git a/packages/hono/src/tests/ServerLimits.test.ts b/packages/hono/src/tests/ServerLimits.test.ts new file mode 100644 index 000000000..2e8a8e937 --- /dev/null +++ b/packages/hono/src/tests/ServerLimits.test.ts @@ -0,0 +1,64 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import type {Server} from 'node:http'; +import {createServer} from '@fluxer/hono/src/Server'; +import type {ServerType} from '@hono/node-server'; +import {Hono} from 'hono'; +import {afterEach, describe, expect, test} from 'vitest'; + +function closeServer(server: ServerType): Promise { + return new Promise((resolve, reject) => { + server.close((error?: Error) => { + if (error) { + reject(error); + return; + } + resolve(); + }); + }); +} + +function listen(options: Parameters[1]): Promise<{server: ServerType; port: number}> { + return new Promise((resolve) => { + const app = new Hono(); + app.get('/', (ctx) => ctx.text('OK')); + const server = createServer(app, { + ...options, + onListen: ({port}) => resolve({server, port}), + }); + }); +} + +describe('Server limits', () => { + let server: ServerType | null = null; + afterEach(async () => { + if (server) { + await closeServer(server); + server = null; + } + }); + + test('applies bounded defaults instead of stock node timeouts', async () => { + const listening = await listen({port: 0}); + server = listening.server; + const httpServer = server as Server; + expect(httpServer.requestTimeout).toBe(30_000); + expect(httpServer.keepAliveTimeout).toBe(125_000); + expect(httpServer.maxRequestsPerSocket).toBe(1_000); + expect(httpServer.headersTimeout).toBeLessThanOrEqual(httpServer.requestTimeout); + }); + + test('honours explicit limit overrides', async () => { + const listening = await listen({ + port: 0, + requestTimeoutMs: 15_000, + keepAliveTimeoutMs: 61_000, + maxRequestsPerSocket: 250, + }); + server = listening.server; + const httpServer = server as Server; + expect(httpServer.requestTimeout).toBe(15_000); + expect(httpServer.keepAliveTimeout).toBe(61_000); + expect(httpServer.maxRequestsPerSocket).toBe(250); + }); +});