Skip to content

refactor: extract payload ingestion and validation module from HTTP handler (#139) #140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jonbaldie merged 1 commit into from
Sep 25, 2026
+473 −156
Merged

refactor: extract payload ingestion and validation module from HTTP handler (#139) #140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions mutation/mutasaurus_ci.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ const mutasaurus = new Mutasaurus({
"./tests/e2e_test.ts",
"./tests/handler_test.ts",
"./tests/manager_test.ts",
"./tests/payload_test.ts",
"./tests/persist_test.ts",
"./tests/rate_limiter_test.ts",
"./tests/router_test.ts",
Expand Down
1 change: 1 addition & 0 deletions scripts/messcript.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ const productionUnits = new Map([
["rate-limiter", ["src/rate_limiter.ts"]],
["queue-manager", ["src/manager.ts"]],
["persist-engine", ["src/persist.ts"]],
["payload", ["src/payload.ts"]],
["http-handler", ["src/handler.ts"]],
["entrypoint", ["main.ts"]],
]);
Expand Down
173 changes: 17 additions & 156 deletions src/handler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,149 +2,10 @@ import QueueManager, { QueueNameTooLongError } from "./manager.ts";
import { RateLimiter } from "./rate_limiter.ts";
import { withAuth, withRateLimit } from "./middleware.ts";
import { RouteHandler, Router } from "./router.ts";
import * as Payload from "./payload.ts";

const MAX_BODY_SIZE = 1024 * 1024; // 1 MB
const LOG_ENCODER = Reflect.construct(TextEncoder, []);

class UnsupportedNumberError extends Error {
constructor() {
super("Payload contains an unsupported number");
}
}

class JsonNestingTooDeepError extends Error {}

function canonicalJsonNumber(source: string): string {
const match = /^(-?)(\d+)(?:\.(\d+))?(?:[eE]([+-]?\d+))?$/.exec(source);
if (match === null) {
return source;
}

const sign = match[1] === "-" ? "-" : "";
let digits = `${match[2]}${match[3] ?? ""}`.replace(/^0+/, "");
if (digits === "") {
return "0";
}

let exponent = Number(match[4] ?? "0") - (match[3]?.length ?? 0);
const digitsWithoutTrailingZeros = digits.replace(/0+$/, "");
exponent += digits.length - digitsWithoutTrailingZeros.length;
digits = digitsWithoutTrailingZeros;
return `${sign}${digits}e${exponent}`;
}

function isUnsupportedNumber(value: number, source: string): boolean {
if (!Number.isFinite(value)) {
return true;
}

const serializedValue = JSON.stringify(value)!;
return canonicalJsonNumber(source) !== canonicalJsonNumber(serializedValue);
}

// Matches strings, containers, and number literals in valid JSON. Scanning
// the source avoids invoking a JSON.parse reviver once per JSON value.
const JSON_TOKEN = /"[^"\\]*(?:\\.[^"\\]*)*"|[[{]|[\]}]|-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?/g;
const EXACT_INTEGER = /^-?\d{1,15}$/;
const MAX_JSON_DEPTH = 3000;

function rejectUnsupportedNumber(source: string): void {
// Every integer with at most 15 digits is below Number.MAX_SAFE_INTEGER.
if (EXACT_INTEGER.test(source)) {
return;
}
if (isUnsupportedNumber(Number(source), source)) {
throw new UnsupportedNumberError();
}
}

function validateJsonSource(source: string): void {
let depth = 0;
for (const match of source.matchAll(JSON_TOKEN)) {
const token = match[0];
if (token === "[" || token === "{") {
depth++;
if (depth > MAX_JSON_DEPTH) {
throw new JsonNestingTooDeepError();
}
} else if (token === "]" || token === "}") {
depth--;
} else if (token[0] !== '"') {
rejectUnsupportedNumber(token);
}
}
}

function parseJsonBody(body: string) {
try {
const json = JSON.parse(body);
validateJsonSource(body);
return json;
} catch (error) {
// Deeply nested JSON can overflow the native parser's stack; treat it
// as unparseable input rather than a server fault.
if (error instanceof RangeError) {
throw new JsonNestingTooDeepError();
}
throw error;
}
}

async function readRequestBody(request: Request): Promise<string | Response> {
const contentLength = request.headers.get("content-length");
if (contentLength && parseInt(contentLength) > MAX_BODY_SIZE) {
return new Response("Payload too large", { status: 413 });
}

if (request.body === null) {
return "";
}

const reader = request.body.getReader();
const chunks: Uint8Array[] = [];
let bodySize = 0;
try {
while (true) {
const { done, value } = await reader.read();
if (done) {
break;
}
bodySize += value.byteLength;
if (bodySize > MAX_BODY_SIZE) {
await reader.cancel();
return new Response("Payload too large", { status: 413 });
}
chunks.push(value);
}
} catch {
try {
await reader.cancel();
} catch (error) {
// The stream may already be closed or errored.
void error;
}
return new Response("Payload too large", { status: 413 });
} finally {
reader.releaseLock();
}

const body = new Uint8Array(bodySize);
let offset = 0;
for (const chunk of chunks) {
body.set(chunk, offset);
offset += chunk.byteLength;
}
return decodeUtf8Body(body);
}

function decodeUtf8Body(body: Uint8Array): string | Response {
try {
return new TextDecoder("utf-8", { fatal: true }).decode(body);
} catch {
return new Response("Invalid JSON", { status: 400 });
}
}

function extractQueueName(match: Parameters<RouteHandler>[1]): { name: string } | { error: Response } {
const raw = match.pathname.groups.queue;
if (raw === undefined) {
Expand All @@ -167,25 +28,19 @@ function enqueueHandler(mgr: QueueManager<string>): RouteHandler {
return queueResult.error;
}
const queueName = queueResult.name;
const body = await readRequestBody(request);
if (body instanceof Response) {
return body;
}
try {
const json = parseJsonBody(body);
if (json === null || typeof json !== "object") {
return new Response("Missing payload key", { status: 400 });
}
if (!("payload" in json)) {
return new Response("Missing payload key", { status: 400 });
}
if (json.payload === null) {
return new Response("Null payload not allowed", { status: 400 });
const contentLength = request.headers.get("content-length");
if (contentLength && parseInt(contentLength) > Payload.DEFAULT_MAX_PAYLOAD_SIZE) {
return new Response("Payload too large", { status: 413 });
}
const payload = await Payload.readAndValidatePayload<string>(
request.body,
Payload.DEFAULT_MAX_PAYLOAD_SIZE,
);
if (!mgr.canEnqueue(queueName)) {
return new Response("Queue full or too many queues", { status: 507 });
}
mgr.enqueue(queueName, json.payload);
mgr.enqueue(queueName, payload);
return new Response(`Payload successfully queued onto ${queueName}.`);
} catch (error) {
return enqueueErrorResponse(error);
Expand All @@ -194,10 +49,16 @@ function enqueueHandler(mgr: QueueManager<string>): RouteHandler {
}

function enqueueErrorResponse(error: unknown): Response {
if (error instanceof SyntaxError || error instanceof JsonNestingTooDeepError) {
if (error instanceof Payload.PayloadTooLargeError) {
return new Response(error.message, { status: 413 });
}
if (error instanceof Payload.UnsupportedNumberError) {
return new Response(error.message, { status: 400 });
}
if (error instanceof Payload.JsonNestingTooDeepError) {
return new Response("Invalid JSON", { status: 400 });
}
if (error instanceof UnsupportedNumberError) {
if (error instanceof Payload.InvalidPayloadError) {
return new Response(error.message, { status: 400 });
}
return queueNameErrorResponse(error);
Expand Down
Loading
Loading