test: gate the graceful drain proof on handler entry
The Pass 4 drain test waited for `server.getConnections() > 0` before starting shutdown, and a comment described that as the request having "demonstrably entered". It had not: the connection count proves the server accepted a TCP connection, nothing about whether Hono had dispatched to a handler. Run five times against the real integration harness it was not flaky, but the claim was stronger than the evidence, so the evidence is now stronger instead. A fixture module's route handler announces entry through a `globalThis` gate — the same channel the existing lifecycle recorders use — and then blocks until the test releases it. Shutdown therefore begins at a point where the handler is provably inside user code. The assertions are now about recorded ordering rather than timing: the handler finished, the module stopped, and `handler-finished` precedes `stop` in the recorded events, with the full 200 response and a clean transport close. No sleeps; the gate is released in `finally` so a failed assertion cannot hang the suite. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
ekklesia
co-authored by
Claude Opus 5
Claude Opus 5
parent
07a3c61083
commit
cda841f8c7
@@ -236,6 +236,97 @@ function createEventRecorder(): { events: string[]; expression: string } {
|
||||
* module also registers a restricted `GET /events/:eventId` route and a restricted, state-changing
|
||||
* `POST /events` route against that same permission, so the full HTTP mounting/enforcement chain
|
||||
* can be exercised live. */
|
||||
/** A handler gate shared with a dynamically imported fixture module through `globalThis`, the same
|
||||
* technique `createEventRecorder` uses. The handler resolves `entered` when it is genuinely inside
|
||||
* user code and then awaits `released`, so a test can position shutdown precisely instead of
|
||||
* inferring handler entry from a connection count. `release` is idempotent and safe in `finally`. */
|
||||
function createHandlerGate(): {
|
||||
readonly entered: Promise<void>;
|
||||
readonly expression: string;
|
||||
release: () => void;
|
||||
} {
|
||||
const key = `ekklesia.api.integration.gate.${eventSymbols.length}.${Date.now()}.${Math.random()}`;
|
||||
const symbol = Symbol.for(key);
|
||||
let signalEntered: () => void = () => undefined;
|
||||
let signalReleased: () => void = () => undefined;
|
||||
const entered = new Promise<void>((resolve) => {
|
||||
signalEntered = resolve;
|
||||
});
|
||||
const released = new Promise<void>((resolve) => {
|
||||
signalReleased = resolve;
|
||||
});
|
||||
(globalThis as unknown as Record<PropertyKey, unknown>)[symbol] = {
|
||||
enter: () => signalEntered(),
|
||||
released,
|
||||
};
|
||||
eventSymbols.push(symbol);
|
||||
return {
|
||||
entered,
|
||||
expression: `globalThis[Symbol.for(${JSON.stringify(key)})]`,
|
||||
release: () => signalReleased(),
|
||||
};
|
||||
}
|
||||
|
||||
/** A fixture module with one public route whose handler blocks on the supplied gate, plus a stop
|
||||
* hook that records when it ran — everything the graceful-drain ordering assertion needs. */
|
||||
async function createGatedRouteModule(
|
||||
modulesDirectory: string,
|
||||
id: string,
|
||||
recorder: { expression: string },
|
||||
gate: { expression: string },
|
||||
): Promise<void> {
|
||||
const permissionId = `${id}.gated.read`;
|
||||
const moduleDirectory = path.join(modulesDirectory, id);
|
||||
await mkdir(path.join(moduleDirectory, "dist/server"), { recursive: true });
|
||||
await Promise.all([
|
||||
writeFile(
|
||||
path.join(moduleDirectory, "package.json"),
|
||||
JSON.stringify({
|
||||
ekklesia: { layer: "module" },
|
||||
name: `@test/module-${id}`,
|
||||
type: "module",
|
||||
version: "0.1.0",
|
||||
}),
|
||||
),
|
||||
writeFile(
|
||||
path.join(moduleDirectory, "ekklesia.module.json"),
|
||||
JSON.stringify({
|
||||
compatibility: { moduleSdk: "^0.1.0", platform: "^0.1.0" },
|
||||
entrypoints: { server: "./dist/server/index.js" },
|
||||
id,
|
||||
manifestVersion: 3,
|
||||
name: `${id} module`,
|
||||
permissions: [{ baseline: "public", description: "Gated drain probe.", id: permissionId }],
|
||||
version: "0.1.0",
|
||||
}),
|
||||
),
|
||||
writeFile(
|
||||
path.join(moduleDirectory, "dist/server/index.js"),
|
||||
`export default {
|
||||
id: ${JSON.stringify(id)},
|
||||
version: "0.1.0",
|
||||
register({ http }) {
|
||||
http.route({
|
||||
method: "GET",
|
||||
path: "/gated",
|
||||
permission: ${JSON.stringify(permissionId)},
|
||||
maxBodyBytes: 0,
|
||||
async handler() {
|
||||
${recorder.expression}.push(${JSON.stringify(`${id}:handler-entered`)});
|
||||
${gate.expression}.enter();
|
||||
await ${gate.expression}.released;
|
||||
${recorder.expression}.push(${JSON.stringify(`${id}:handler-finished`)});
|
||||
return Response.json({ gated: true });
|
||||
},
|
||||
});
|
||||
},
|
||||
start() {},
|
||||
stop() { ${recorder.expression}.push(${JSON.stringify(`${id}:stop`)}); },
|
||||
};`,
|
||||
),
|
||||
]);
|
||||
}
|
||||
|
||||
async function createFixtureModule(
|
||||
modulesDirectory: string,
|
||||
id: string,
|
||||
@@ -2779,18 +2870,28 @@ describe.sequential("live API — bounded HTTP drain and shutdown", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("lets an already-entered request finish and close cleanly rather than force-closing it", async () => {
|
||||
// Deterministic rather than racing an unresolved `fetch`: a raw socket sends a complete request
|
||||
// with `Connection: close`, and shutdown only begins once the server provably holds that
|
||||
// connection — observed by polling the real server's own connection count, never by sleeping.
|
||||
it("lets a handler that has already entered finish, and closes its transport before module stop", async () => {
|
||||
// A real handler gate, not an inference from the connection count: the fixture module's route
|
||||
// handler announces that it has *entered*, then blocks until this test releases it. Shutdown
|
||||
// therefore begins at a point where the handler is provably inside user code, and the ordering
|
||||
// claim below is an assertion about recorded events rather than about timing.
|
||||
const modulesDirectory = await createFixtureModulesDirectory();
|
||||
const recorder = createEventRecorder();
|
||||
const gate = createHandlerGate();
|
||||
await createGatedRouteModule(modulesDirectory, "drainorder", recorder, gate);
|
||||
|
||||
const database = await createDisposableDatabase("drain_order");
|
||||
await applyCoreMigrations(database.url);
|
||||
const running = await startConfiguredApi(database.url, { drainTimeoutMsOverride: 20_000 });
|
||||
const running = await startConfiguredApi(database.url, {
|
||||
drainTimeoutMsOverride: 20_000,
|
||||
modulesDirectoryOverride: modulesDirectory,
|
||||
});
|
||||
const address = running.server.address();
|
||||
if (!address || typeof address !== "object") {
|
||||
throw harnessError("The API integration server has no bound address.");
|
||||
}
|
||||
const socket = net.connect({ host: "127.0.0.1", port: address.port });
|
||||
let shutdown: Promise<void> | undefined;
|
||||
try {
|
||||
const responseBytes: Buffer[] = [];
|
||||
const responseClosed = new Promise<void>((resolve) => {
|
||||
@@ -2802,44 +2903,38 @@ describe.sequential("live API — bounded HTTP drain and shutdown", () => {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
socket.once("error", reject);
|
||||
socket.write(
|
||||
"GET /api/v1/principal HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n",
|
||||
`GET /api/v1/modules/drainorder/gated HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n`,
|
||||
() => resolve(),
|
||||
);
|
||||
});
|
||||
|
||||
// Bounded, condition-based: wait until the server actually holds the connection, so the
|
||||
// request has demonstrably entered before shutdown starts.
|
||||
const connectionCounter = running.server as unknown as {
|
||||
getConnections?: (callback: (error: Error | null, count: number) => void) => void;
|
||||
};
|
||||
const deadline = Date.now() + 15_000;
|
||||
for (;;) {
|
||||
const connections = await new Promise<number>((resolve) => {
|
||||
if (typeof connectionCounter.getConnections !== "function") {
|
||||
resolve(1);
|
||||
return;
|
||||
}
|
||||
connectionCounter.getConnections((_error, count) => resolve(count ?? 0));
|
||||
});
|
||||
if (connections > 0) {
|
||||
break;
|
||||
}
|
||||
if (Date.now() > deadline) {
|
||||
throw harnessError("The request never reached the server.");
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
// Bounded wait on the handler's own signal — no sleep, no proxy for "entered".
|
||||
await withDeadline(gate.entered, 20_000);
|
||||
expect(recorder.events).toContain("drainorder:handler-entered");
|
||||
|
||||
// The graceful path: nothing is stalled, so shutdown completes without reaching the drain
|
||||
// deadline and without force-closing anything.
|
||||
await expect(withDeadline(running.shutdown(), 30_000)).resolves.toBeUndefined();
|
||||
// Shutdown starts while the handler is still inside user code.
|
||||
shutdown = running.shutdown();
|
||||
// Release it, then require the graceful path to complete.
|
||||
gate.release();
|
||||
await expect(withDeadline(shutdown, 30_000)).resolves.toBeUndefined();
|
||||
|
||||
// The already-entered request completed normally — full response, clean transport close.
|
||||
// The response really was served, in full, rather than force-closed mid-flight.
|
||||
await withDeadline(responseClosed, 10_000);
|
||||
const raw = Buffer.concat(responseBytes).toString("utf8");
|
||||
expect(raw).toContain("HTTP/1.1 200");
|
||||
expect(raw).toContain('{"kind":"anonymous"}');
|
||||
expect(raw).toContain('"gated":true');
|
||||
|
||||
// Explicit ordering: the handler finished and the module stopped, in that order.
|
||||
const finished = recorder.events.indexOf("drainorder:handler-finished");
|
||||
const stopped = recorder.events.indexOf("drainorder:stop");
|
||||
expect(finished).toBeGreaterThanOrEqual(0);
|
||||
expect(stopped).toBeGreaterThanOrEqual(0);
|
||||
expect(finished).toBeLessThan(stopped);
|
||||
} finally {
|
||||
gate.release();
|
||||
if (shutdown) {
|
||||
await shutdown.catch(() => undefined);
|
||||
}
|
||||
socket.destroy();
|
||||
await running.shutdown().catch(() => undefined);
|
||||
await dropDatabase(database.name);
|
||||
|
||||
Reference in New Issue
Block a user