diff --git a/apps/api/tests/api-http.integration.ts b/apps/api/tests/api-http.integration.ts index 990a55c..7dfd7b4 100644 --- a/apps/api/tests/api-http.integration.ts +++ b/apps/api/tests/api-http.integration.ts @@ -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; + 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((resolve) => { + signalEntered = resolve; + }); + const released = new Promise((resolve) => { + signalReleased = resolve; + }); + (globalThis as unknown as Record)[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 { + 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 | undefined; try { const responseBytes: Buffer[] = []; const responseClosed = new Promise((resolve) => { @@ -2802,44 +2903,38 @@ describe.sequential("live API — bounded HTTP drain and shutdown", () => { await new Promise((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((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);