From d2839797b77f88b976a447dcc0a4b04e52b21fa0 Mon Sep 17 00:00:00 2001 From: Thibault Le Ouay Ducasse Date: Thu, 10 Sep 2026 15:35:35 +0200 Subject: [PATCH] probes: endpoint --- apps/server/src/env.ts | 4 + apps/server/src/index.ts | 20 ++- apps/server/src/libs/health.test.ts | 156 ++++++++++++++++++++ apps/server/src/libs/health.ts | 113 ++++++++++++++ apps/server/src/libs/machine.ts | 151 +++++++++++++++++++ apps/server/src/routes/health/index.test.ts | 124 ++++++++++++++++ apps/server/src/routes/health/index.ts | 86 +++++++++++ apps/server/src/routes/health/probes.ts | 66 +++++++++ 8 files changed, 719 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/libs/health.test.ts create mode 100644 apps/server/src/libs/health.ts create mode 100644 apps/server/src/libs/machine.ts create mode 100644 apps/server/src/routes/health/index.test.ts create mode 100644 apps/server/src/routes/health/index.ts create mode 100644 apps/server/src/routes/health/probes.ts diff --git a/apps/server/src/env.ts b/apps/server/src/env.ts index 300f03e7..2ffcc084 100644 --- a/apps/server/src/env.ts +++ b/apps/server/src/env.ts @@ -12,6 +12,10 @@ export const env = createEnv({ UPSTASH_REDIS_REST_URL: z.string().min(1), UPSTASH_REDIS_REST_TOKEN: z.string().min(1), FLY_REGION: z.enum(monitorRegions), + // Injected by Fly at boot, reported by `/ping`. Absent locally and in tests. + FLY_MACHINE_ID: z.string().optional(), + FLY_APP_NAME: z.string().optional(), + FLY_MACHINE_VERSION: z.string().optional(), CRON_SECRET: z.string(), SCREENSHOT_SERVICE_URL: z.string(), QSTASH_TOKEN: z.string(), diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 58811627..ab7f58ff 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -27,8 +27,11 @@ import { requestId } from "hono/request-id"; import { env } from "./env"; import { handleError } from "./libs/errors"; import { concurrencyGuard } from "./libs/middlewares/concurrency"; +import { limits } from "./libs/middlewares/limits"; import { rateLimit } from "./libs/middlewares/rate-limit"; import { shouldSample } from "./libs/sampling"; +import { createHealthRoute } from "./routes/health"; +import { probesFromEnv } from "./routes/health/probes"; import { mcpRoute } from "./routes/mcp"; import { createOAuthRoutes } from "./routes/oauth"; import { oauthConfigFromEnv } from "./routes/oauth/config"; @@ -215,7 +218,9 @@ app.route("/", createOAuthRoutes(oauthConfigFromEnv())); app.route("/public", publicRoute); /** - * Ping Pong + * Ping Pong — liveness only, and deliberately so: this is the Fly HTTP check + * and the container healthcheck, so it must not fail over a dependency + * someone else operates. Dependency status lives on `/health`. */ app.get("/ping", (c) => { return c.json( @@ -224,6 +229,19 @@ app.get("/ping", (c) => { ); }); +/** + * Readiness — Turso, Upstash, Tinybird and Unkey, plus the vitals of the + * machine that answered. Rate limited and shed like any other route. + */ +app.route( + "/", + createHealthRoute({ + probes: probesFromEnv(), + inFlight: () => concurrencyGuard.inFlight(), + maxInFlight: () => limits.maxInFlight, + }), +); + app.route("/", openapiRoute); app.get( diff --git a/apps/server/src/libs/health.test.ts b/apps/server/src/libs/health.test.ts new file mode 100644 index 00000000..ad2ef6ae --- /dev/null +++ b/apps/server/src/libs/health.test.ts @@ -0,0 +1,156 @@ +import { expect } from "@std/expect"; +import { describe, test } from "@std/testing/bdd"; + +import { type Probe, runProbes } from "@/libs/health"; + +function probe(over: Partial & Pick): Probe { + return { + critical: false, + timeoutMs: 50, + run: () => Promise.resolve(), + ...over, + }; +} + +describe("runProbes", () => { + test("is ok when every dependency answers", async () => { + const report = await runProbes([ + probe({ name: "database", critical: true }), + probe({ name: "redis" }), + ]); + + expect(report.status).toBe("ok"); + expect(report.checks.map((c) => c.status)).toEqual(["ok", "ok"]); + expect(report.checks[0].latencyMs).toBeGreaterThanOrEqual(0); + expect(report.checkedAt).toMatch(/^\d{4}-/); + }); + + test("a non-critical dependency going down is degraded, not unhealthy", async () => { + const report = await runProbes([ + probe({ name: "database", critical: true }), + probe({ + name: "tinybird", + run: () => Promise.reject(new Error("HTTP 503")), + }), + ]); + + expect(report.status).toBe("degraded"); + expect(report.checks[1]).toMatchObject({ + name: "tinybird", + status: "down", + error: "HTTP 503", + }); + }); + + test("turso going down is unhealthy", async () => { + const report = await runProbes([ + probe({ + name: "database", + critical: true, + run: () => Promise.reject(new Error("SQLITE_UNKNOWN")), + }), + probe({ name: "redis" }), + ]); + + expect(report.status).toBe("unhealthy"); + }); + + test("a hung dependency is down at its own deadline, not the report's", async () => { + const report = await runProbes([ + probe({ name: "redis", timeoutMs: 20, run: () => new Promise(() => {}) }), + probe({ name: "database", critical: true }), + ]); + + expect(report.status).toBe("degraded"); + expect(report.checks[0].error).toBe("timed out after 20ms"); + // Concurrent: the whole report waits for the slowest probe, not their sum. + expect(report.latencyMs).toBeLessThan(200); + }); + + test("aborts the signal it hands a probe when the deadline passes", async () => { + let aborted = false; + await runProbes([ + probe({ + name: "unkey", + timeoutMs: 20, + run: (signal) => + new Promise((_, reject) => { + signal.addEventListener("abort", () => { + aborted = true; + reject(new Error("aborted")); + }); + }), + }), + ]); + + expect(aborted).toBe(true); + }); + + test("a probe that throws synchronously is down, not a crash", async () => { + // `fetch` throws rather than rejects on a malformed URL, which is what a + // misconfigured base URL produces. + const report = await runProbes([ + probe({ + name: "tinybird", + run: () => { + throw new TypeError("Invalid URL: '/v0/health'"); + }, + }), + ]); + + expect(report.status).toBe("degraded"); + expect(report.checks[0]).toMatchObject({ + status: "down", + error: "Invalid URL: '/v0/health'", + }); + }); + + test("that synchronous throw leaves no stray rejection behind", async () => { + // It never reaches `Promise.race`, so nothing is watching the deadline; + // rejecting it from the abort in `runProbe`'s `finally` would take the + // process down. `deno test` swallows that, so assert on a bare process. + const module = new URL("./health.ts", import.meta.url).href; + const { code, stderr } = await new Deno.Command(Deno.execPath(), { + args: [ + "eval", + "--ext=ts", + `import { runProbes } from ${JSON.stringify(module)}; + await runProbes([{ + name: "tinybird", + critical: false, + timeoutMs: 2000, + run: () => { throw new TypeError("Invalid URL"); }, + }]); + await new Promise((r) => setTimeout(r, 20));`, + ], + }).output(); + + expect(new TextDecoder().decode(stderr)).not.toContain("Uncaught"); + expect(code).toBe(0); + }); + + test("skips a dependency that is switched off", async () => { + const report = await runProbes([ + probe({ + name: "tinybird", + skip: () => true, + run: () => Promise.reject(new Error("must not run")), + }), + ]); + + expect(report.status).toBe("ok"); + expect(report.checks[0]).toMatchObject({ status: "skipped", latencyMs: 0 }); + }); + + test("truncates probe errors, which carry connection strings", async () => { + const report = await runProbes([ + probe({ + name: "database", + run: () => Promise.reject(new Error("x".repeat(500))), + }), + ]); + + expect(report.checks[0].error).toHaveLength(201); + expect(report.checks[0].error?.endsWith("…")).toBe(true); + }); +}); diff --git a/apps/server/src/libs/health.ts b/apps/server/src/libs/health.ts new file mode 100644 index 00000000..23c17374 --- /dev/null +++ b/apps/server/src/libs/health.ts @@ -0,0 +1,113 @@ +/** + * Dependency probes behind `GET /health`. + * + * A probe answers one question: can this machine still reach the thing it + * needs? It resolves when healthy and throws otherwise. Probes run + * concurrently, each under its own deadline, so one slow dependency cannot + * hold the whole report — an unreachable service is exactly the case where a + * call hangs until the socket times out. + */ + +export type ProbeStatus = "ok" | "down" | "skipped"; + +export type Probe = { + name: string; + /** A critical dependency that is down makes `/health` answer 503. */ + critical: boolean; + /** Deadline for this probe alone. */ + timeoutMs: number; + /** Opt out at runtime — e.g. Tinybird in noop mode, which talks to nothing. */ + skip?: () => boolean; + /** `signal` aborts at the deadline; probes that cannot take one are raced instead. */ + run: (signal: AbortSignal) => Promise; +}; + +export type ProbeResult = { + name: string; + critical: boolean; + status: ProbeStatus; + latencyMs: number; + /** Only on `down`, truncated: probe errors carry connection strings. */ + error?: string; +}; + +export type HealthReport = { + status: "ok" | "degraded" | "unhealthy"; + checkedAt: string; + latencyMs: number; + checks: ProbeResult[]; +}; + +/** Long enough to name the failure, short enough not to paste a credential into a public response. */ +const MAX_ERROR_LENGTH = 200; + +function describe(error: unknown): string { + const message = error instanceof Error ? error.message : String(error); + const collapsed = message.replace(/\s+/g, " ").trim(); + return collapsed.length > MAX_ERROR_LENGTH + ? `${collapsed.slice(0, MAX_ERROR_LENGTH)}…` + : collapsed; +} + +async function runProbe(probe: Probe): Promise { + const base = { name: probe.name, critical: probe.critical }; + if (probe.skip?.()) { + return { ...base, status: "skipped", latencyMs: 0 }; + } + + const startedAt = performance.now(); + const controller = new AbortController(); + + // `race` is what actually enforces the deadline: the libSQL client takes no + // signal, so aborting only helps the probes built on fetch. The timer owns + // the rejection — hanging it off `abort` instead would fire again when the + // `finally` below aborts a probe that already answered, rejecting a promise + // nothing is watching any more. + let expire!: (reason: Error) => void; + const expired = new Promise((_, reject) => { + expire = reject; + }); + const timer = setTimeout(() => { + controller.abort(); + expire(new Error(`timed out after ${probe.timeoutMs}ms`)); + }, probe.timeoutMs); + + try { + await Promise.race([probe.run(controller.signal), expired]); + return { + ...base, + status: "ok", + latencyMs: Math.round(performance.now() - startedAt), + }; + } catch (error) { + return { + ...base, + status: "down", + latencyMs: Math.round(performance.now() - startedAt), + error: describe(error), + }; + } finally { + clearTimeout(timer); + // Release a probe still waiting on the network once we have our answer. + controller.abort(); + } +} + +export async function runProbes(probes: Probe[]): Promise { + const startedAt = performance.now(); + const checks = await Promise.all(probes.map(runProbe)); + + const down = checks.filter((check) => check.status === "down"); + const status = down.some((check) => check.critical) + ? "unhealthy" + : down.length + ? "degraded" + : "ok"; + + return { + status, + checkedAt: new Date().toISOString(), + latencyMs: Math.round(performance.now() - startedAt), + checks, + }; +} diff --git a/apps/server/src/libs/machine.ts b/apps/server/src/libs/machine.ts new file mode 100644 index 00000000..c12d304c --- /dev/null +++ b/apps/server/src/libs/machine.ts @@ -0,0 +1,151 @@ +import { env } from "@/env"; + +/** + * Machine vitals for `/health`. + * + * Everything here is read from this process and the host it runs on. It says + * nothing about whether Turso or Tinybird are reachable — that is what the + * probes in `libs/health.ts` are for. `pressure` reports what is wrong with + * *this* machine, so a report can distinguish "the dependency is down" from + * "this machine is out of memory". + */ + +/** Resolved once: Fly injects these at boot and they never change. */ +const identity = { + id: env.FLY_MACHINE_ID ?? null, + app: env.FLY_APP_NAME ?? null, + version: env.FLY_MACHINE_VERSION ?? null, + region: env.FLY_REGION ?? null, + environment: env.NODE_ENV, + runtime: `deno/${Deno.version.deno}`, +}; + +const bootedAt = Date.now(); + +/** Above these, the machine reports pressure; the caller decides what that means. */ +const MEMORY_DEGRADED_PERCENT = 90; +const SATURATION_DEGRADED_PERCENT = 80; + +/** Sampled at most once a second so a polled endpoint never hammers /proc. */ +const SAMPLE_TTL_MS = 1_000; + +/** `loadavg`, `systemMemoryInfo` and `hostname` need `--allow-sys` and are not implemented on every target. */ +function readSys(read: () => T): T | null { + try { + return read(); + } catch { + return null; + } +} + +const hostname = readSys(() => Deno.hostname()); + +type HostSample = { + process: Deno.MemoryUsage; + system: Deno.SystemMemoryInfo | null; + loadAverage: number[] | null; +}; + +let cached: { at: number; sample: HostSample } | null = null; + +function sampleHost(now: number): HostSample { + if (cached && now - cached.at < SAMPLE_TTL_MS) return cached.sample; + const sample: HostSample = { + process: Deno.memoryUsage(), + system: readSys(() => Deno.systemMemoryInfo()), + loadAverage: readSys(() => Deno.loadavg()), + }; + cached = { at: now, sample }; + return sample; +} + +function percent(part: number, whole: number): number | null { + if (!whole) return null; + return Math.round((part / whole) * 1000) / 10; +} + +export type MachineVitals = { + /** Empty when the machine is healthy; one human-readable reason per breached threshold. */ + pressure: string[]; + machine: typeof identity & { host: string | null; uptimeSeconds: number }; + cpu: { + cores: number; + /** 1/5/15 minute load, `null` where the OS does not expose it. */ + loadAverage: number[] | null; + loadPerCore: number | null; + }; + memory: { + rssBytes: number; + heapUsedBytes: number; + heapTotalBytes: number; + totalBytes: number | null; + availableBytes: number | null; + usedPercent: number | null; + }; + requests: { + inFlight: number; + maxInFlight: number; + saturationPercent: number | null; + }; +}; + +export type MachineVitalsOptions = { + inFlight: number; + maxInFlight: number; +}; + +export function machineVitals(options: MachineVitalsOptions): MachineVitals { + const now = Date.now(); + const { process, system, loadAverage } = sampleHost(now); + const cores = navigator.hardwareConcurrency; + + const usedPercent = system + ? percent(system.total - system.available, system.total) + : null; + const saturationPercent = percent(options.inFlight, options.maxInFlight); + + const pressure: string[] = []; + if (usedPercent !== null && usedPercent >= MEMORY_DEGRADED_PERCENT) { + pressure.push(`memory at ${usedPercent}% of the machine`); + } + if ( + saturationPercent !== null && + saturationPercent >= SATURATION_DEGRADED_PERCENT + ) { + pressure.push( + `${options.inFlight} of ${options.maxInFlight} in-flight slots taken`, + ); + } + // Load average is reported but never counts as pressure: Fly's shared CPUs + // make it spike on neighbours we do not control, and no action follows. + + return { + pressure, + machine: { + ...identity, + host: hostname, + uptimeSeconds: Math.round((now - bootedAt) / 1000), + }, + cpu: { + cores, + loadAverage, + loadPerCore: + loadAverage && cores + ? Math.round((loadAverage[0] / cores) * 100) / 100 + : null, + }, + memory: { + rssBytes: process.rss, + heapUsedBytes: process.heapUsed, + heapTotalBytes: process.heapTotal, + totalBytes: system?.total ?? null, + availableBytes: system?.available ?? null, + usedPercent, + }, + requests: { + inFlight: options.inFlight, + maxInFlight: options.maxInFlight, + saturationPercent, + }, + }; +} diff --git a/apps/server/src/routes/health/index.test.ts b/apps/server/src/routes/health/index.test.ts new file mode 100644 index 00000000..8ccfe469 --- /dev/null +++ b/apps/server/src/routes/health/index.test.ts @@ -0,0 +1,124 @@ +import { expect } from "@std/expect"; +import { describe, test } from "@std/testing/bdd"; + +import type { Probe } from "@/libs/health"; +import { createHealthRoute } from "@/routes/health"; + +function route( + probes: Probe[], + over: { cacheMs?: number; inFlight?: number } = {}, +) { + return createHealthRoute({ + probes, + inFlight: () => over.inFlight ?? 0, + maxInFlight: () => 128, + cacheMs: over.cacheMs ?? 0, + }); +} + +const ok = (name: string, critical = false): Probe => ({ + name, + critical, + timeoutMs: 50, + run: () => Promise.resolve(), +}); + +const down = (name: string, critical = false): Probe => ({ + ...ok(name, critical), + run: () => Promise.reject(new Error("unreachable")), +}); + +describe("GET /health", () => { + test("reports every dependency it probed", async () => { + const res = await route([ok("database", true), ok("redis")]).request( + "/health", + ); + expect(res.status).toBe(200); + expect(res.headers.get("cache-control")).toBe("no-store"); + + const body = await res.json(); + expect(body.status).toBe("ok"); + expect(body.checks).toMatchObject([ + { name: "database", critical: true, status: "ok" }, + { name: "redis", critical: false, status: "ok" }, + ]); + expect(body.checks[0].latencyMs).toBeGreaterThanOrEqual(0); + expect(body.machine.runtime).toBe(`deno/${Deno.version.deno}`); + expect(body.memory.rssBytes).toBeGreaterThan(0); + expect(body.pressure).toEqual([]); + }); + + test("stays 200 when a dependency we can live without is down", async () => { + const res = await route([ok("database", true), down("tinybird")]).request( + "/health", + ); + + expect(res.status).toBe(200); + expect((await res.json()).status).toBe("degraded"); + }); + + test("answers 503 when turso is unreachable", async () => { + const res = await route([down("database", true), ok("redis")]).request( + "/health", + ); + + expect(res.status).toBe(503); + const body = await res.json(); + expect(body.status).toBe("unhealthy"); + expect(body.checks[0]).toMatchObject({ + status: "down", + error: "unreachable", + }); + }); + + test("a saturated machine is degraded even with every dependency up", async () => { + const res = await route([ok("database", true)], { inFlight: 120 }).request( + "/health", + ); + + expect(res.status).toBe(200); + const body = await res.json(); + expect(body.status).toBe("degraded"); + expect(body.pressure).toEqual(["120 of 128 in-flight slots taken"]); + }); + + test("reuses one round of probes inside the cache window", async () => { + let runs = 0; + const counted: Probe = { + ...ok("database", true), + run: () => { + runs++; + return Promise.resolve(); + }, + }; + const health = route([counted], { cacheMs: 10_000 }); + + await health.request("/health"); + await health.request("/health"); + expect(runs).toBe(1); + }); + + test("concurrent callers share one round rather than each starting their own", async () => { + let runs = 0; + let release: (() => void) | undefined; + const held: Probe = { + ...ok("database", true), + run: () => { + runs++; + return new Promise((resolve) => { + release = resolve; + }); + }, + }; + const health = route([held], { cacheMs: 0 }); + + const first = health.request("/health"); + const second = health.request("/health"); + await new Promise((r) => setTimeout(r, 0)); + release?.(); + + expect((await first).status).toBe(200); + expect((await second).status).toBe(200); + expect(runs).toBe(1); + }); +}); diff --git a/apps/server/src/routes/health/index.ts b/apps/server/src/routes/health/index.ts new file mode 100644 index 00000000..12c0bf5c --- /dev/null +++ b/apps/server/src/routes/health/index.ts @@ -0,0 +1,86 @@ +import { Hono } from "hono"; + +import { type HealthReport, type Probe, runProbes } from "@/libs/health"; +import { machineVitals } from "@/libs/machine"; + +export type HealthRouteConfig = { + probes: Probe[]; + /** Live readers, not values: the guard and its limits change while the process runs. */ + inFlight: () => number; + maxInFlight: () => number; + /** How long one round of probes is reused. `0` probes on every request. */ + cacheMs?: number; +}; + +/** Two Fly check intervals: pollers see fresh data, a hot loop still probes once. */ +const DEFAULT_CACHE_MS = 5_000; + +/** + * `GET /health` — readiness. Can this machine still reach Turso, Upstash, + * Tinybird and Unkey? + * + * 503 only when a *critical* dependency is down. A degraded report stays 200: + * the API can still serve most of its surface without Tinybird, and answering + * 503 would tell every caller to fail over for a metrics outage. + * + * This is deliberately not the Fly check (`/ping` is). Wiring Fly here would + * mean a Turso blip restarts every machine at once — losing the capacity that + * has to absorb the recovery, and none of it would bring Turso back. + */ +export function createHealthRoute(config: HealthRouteConfig) { + const cacheMs = config.cacheMs ?? DEFAULT_CACHE_MS; + let cached: { at: number; report: HealthReport } | undefined; + /** Concurrent callers share one round of probes instead of each starting their own. */ + let pending: Promise | undefined; + + function report(): Promise { + const now = Date.now(); + if (cached && now - cached.at < cacheMs) + return Promise.resolve(cached.report); + if (pending) return pending; + + pending = runProbes(config.probes) + .then((fresh) => { + cached = { at: Date.now(), report: fresh }; + return fresh; + }) + .finally(() => { + pending = undefined; + }); + + return pending; + } + + const health = new Hono({ strict: false }); + + health.get("/health", async (c) => { + const dependencies = await report(); + const vitals = machineVitals({ + inFlight: config.inFlight(), + maxInFlight: config.maxInFlight(), + }); + + // A dependency outage outranks local pressure: it is the one a caller + // can neither retry around nor wait out. + const status = + dependencies.status === "ok" && vitals.pressure.length + ? "degraded" + : dependencies.status; + + return c.json( + { + status, + region: vitals.machine.region, + requestId: c.get("requestId" as never) as string | undefined, + checkedAt: dependencies.checkedAt, + latencyMs: dependencies.latencyMs, + checks: dependencies.checks, + ...vitals, + }, + status === "unhealthy" ? 503 : 200, + { "Cache-Control": "no-store" }, + ); + }); + + return health; +} diff --git a/apps/server/src/routes/health/probes.ts b/apps/server/src/routes/health/probes.ts new file mode 100644 index 00000000..76df214b --- /dev/null +++ b/apps/server/src/routes/health/probes.ts @@ -0,0 +1,66 @@ +import { db, sql } from "@openstatus/db"; + +import { env } from "@/env"; +import { redis } from "@/libs/clients"; +import type { Probe } from "@/libs/health"; + +/** + * The services this API cannot do its job without. + * + * Only Turso is critical: every route reads or writes it, so a machine that + * cannot reach it has nothing to serve. Redis backs caches and the Slack + * confirmation store, Tinybird backs the metrics endpoints and Unkey only the + * legacy key path — losing any of those costs features, not the API. + */ + +/** Health checks queue behind real traffic, so they get a short leash. */ +const TIMEOUT_MS = 2_000; + +/** `skipValidation` never runs the schemas, so the env default never applies — it lands here. */ +const TINYBIRD_URL = env.TINYBIRD_URL || "https://api.tinybird.co"; + +export function probesFromEnv(): Probe[] { + return [ + { + name: "database", + critical: true, + timeoutMs: TIMEOUT_MS, + // Turso over HTTP: no pool to exhaust, and `select 1` never touches a table. + run: () => db.run(sql`select 1`), + }, + { + name: "redis", + critical: false, + timeoutMs: TIMEOUT_MS, + // Unset locally; without a URL the client just retries into the deadline. + skip: () => !env.UPSTASH_REDIS_REST_URL, + run: () => redis.ping(), + }, + { + name: "tinybird", + critical: false, + timeoutMs: TIMEOUT_MS, + // `noop` mode answers from nothing, so there is no dependency to probe. + skip: () => env.TINYBIRD_NOOP, + // Reachability only. Our token is workspace-scoped and cannot list pipes + // (`/v0/pipes` answers 403), so an authenticated probe would report a + // healthy Tinybird as down. A bad token surfaces on the metrics routes. + run: (signal) => expectOk(fetch(`${TINYBIRD_URL}/v0/health`, { signal })), + }, + { + name: "unkey", + critical: false, + timeoutMs: TIMEOUT_MS, + run: (signal) => + expectOk(fetch("https://api.unkey.com/v2/liveness", { signal })), + }, + ]; +} + +async function expectOk(pending: Promise): Promise { + const res = await pending; + // Drain the body or Deno holds the connection open and the test runner + // reports a leaked resource. + await res.body?.cancel(); + if (!res.ok) throw new Error(`HTTP ${res.status}`); +} -- 2.51.2