fix(api): bound http server limits and shed on overload (#2170)

fluxer_api/src/api/App.ts

@@ -48,6 +48,7 @@ export async function createAPIApp(options: CreateAPIAppOptions): Promise<APIApp
 		corsOrigins: [config.endpoints.webApp, config.endpoints.marketing],
 		trustClientIpHeader: config.proxy.trust_client_ip_header,
 		clientIpHeaderName: config.proxy.client_ip_header,
+		maxInflightRequests: config.maxInflightRequests,
 	});
 	routes.onError(AbuseAwareAppErrorHandler);
 	routes.notFound(AppNotFoundHandler);

fluxer_api/src/api/Config.ts

@@ -160,6 +160,7 @@ export function buildAPIConfigFromMaster(master: MasterConfig): APIConfig {
 	return {
 		nodeEnv: master.env === 'test' ? 'development' : master.env,
 		port: master.services.api.port,
+		maxInflightRequests: master.services.api.max_inflight_requests,
 		ipBanExemptIps: normalizeIpBanExemptIps(master.services.api.ip_ban_exempt_ips),
 		desktopGitHubRedirectCountries: normalizeCountryCodes(
 			master.services.api.desktop_github_redirect_countries,

fluxer_api/src/api/app/MiddlewarePipeline.ts

@@ -9,6 +9,7 @@ import {resolveClientIpHeaderName} from '@fluxer/ip_utils/src/ClientIp';
 import type {ILogger} from '../ILogger';
 import {ClientErrorAbuseSignalMiddleware} from '../middleware/AbusiveIpAutoBanner';
 import {AuditLogMiddleware} from '../middleware/AuditLogMiddleware';
+import {ConcurrencyLimitMiddleware} from '../middleware/ConcurrencyLimitMiddleware';
 import ContentFilterMiddleware from '../middleware/ContentFilterMiddleware';
 import {GuildAvailabilityMiddleware} from '../middleware/GuildAvailabilityMiddleware';
 import {IpBanMiddleware} from '../middleware/IpBanMiddleware';
@@ -27,10 +28,11 @@ interface MiddlewarePipelineOptions {
 	corsOrigins: Array<string>;
 	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(

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<string>;
 	desktopGitHubRedirectCountries: ReadonlySet<string>;
 	cassandra: {

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<HonoEnv>(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;
+		}
+	});
+}

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<void>;
+	resolve: () => void;
+}
+
+function createDeferred(): Deferred {
+	let resolve: () => void = () => undefined;
+	const promise = new Promise<void>((resolveFn) => {
+		resolve = resolveFn;
+	});
+	return {promise, resolve};
+}
+
+function buildApp(maxInflightRequests: number) {
+	const release = createDeferred();
+	const entered = createDeferred();
+	const routes = new Hono<HonoEnv>({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<HonoEnv>({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);
+	});
+});

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');

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<string>;
 			desktop_github_redirect_countries: Array<string>;
 			presigned_attachment_uploads_enabled: boolean;

packages/config/src/config_loader/EnvironmentOverrides.ts

@@ -76,6 +76,7 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record<string, NamedEnvOverride> = {
 	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'],

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> | unknown;
@@ -17,15 +24,30 @@ function createVersionedFetch<E extends Env>(app: Hono<E>): NodeFetchCallback {
 }
 
 export function createServer<E extends Env = Env>(app: Hono<E>, 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<void>;

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<void> {
+	return new Promise((resolve, reject) => {
+		server.close((error?: Error) => {
+			if (error) {
+				reject(error);
+				return;
+			}
+			resolve();
+		});
+	});
+}
+
+function listen(options: Parameters<typeof createServer>[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);
+	});
+});