From 9459372126e96ba182a7a5797977df866a646905 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 13 Nov 2026 22:53:11 +0000 Subject: [PATCH 1/3] fix(gateway): degrade when a provider is down A cache miss that times out against FrankerFaceZ or IVR was returned as 502, and the next request during backoff as 503. Serve any copy still held, keep provider 404s as negative cache, and log upstream failures apart from Worker faults. Add /health for a check that does not call a provider. Co-authored-by: Fabian Zimber --- docs/ARCHITECTURE.md | 14 +++- docs/DEPLOYMENT.md | 4 +- src/lib/chat/gateway.ts | 35 +++++---- src/worker/api.ts | 32 +++++++- src/worker/entry.ts | 4 +- src/worker/gateway/gateway.ts | 110 ++++++++++++++++++++++------ test/chat/gateway.test.ts | 15 ++++ test/worker/api.test.ts | 56 +++++++++++++- test/worker/gateway/gateway.test.ts | 83 +++++++++++++++++++++ 9 files changed, 308 insertions(+), 45 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8c4fc73..e829973 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -250,6 +250,9 @@ prerendered pages, the fonts, the images, `robots.txt` and the files for languag other request goes to `src/worker/entry.ts`, the entry of the Worker. It exports the Durable Object class and looks at the path: +- `/health` is answered by `src/worker/api.ts` before anything else. It reads the version + binding and returns. It does not touch KV, a hub or a provider, so a third-party outage cannot + fail it. - A path that starts with `/api/` is answered by `src/worker/api.ts` and never reaches Nitro. The prefix is exact and case-sensitive. - Everything else is passed to the Worker that Nitro builds, unchanged. There, the middleware @@ -264,6 +267,7 @@ the route rules, and the response of a WebSocket upgrade does not survive that. | Endpoint | Method | Answer | | -------------------------------- | --------- | ---------------------------------------------------- | +| `/health` | GET, HEAD | `{ ok: true, version }` — the running deployment id, or null | | `/api/irc?channel=` | GET with a WebSocket upgrade | The chat of one channel, in the form of Twitch IRC | | `/api/data//` | GET; POST for 7TV paints | Provider data as JSON, with the header `x-petal-cache` | | `/api/status` | GET | Counts as JSON | @@ -531,7 +535,8 @@ not a proxy for arbitrary URLs. The three windows follow each other. While an answer is fresh it is served without asking the provider. In the window after that, lists that every overlay shares are served at once and refreshed in the background. Channel data has no such window: a reload of the overlay must show an emote that -was added a minute ago. In the last window the stored answer is only served when the provider fails. +was added a minute ago. In the last window the stored answer is only served when the provider fails. A copy that is +still held after that window is served the same way when the provider fails, instead of a 502. A 404 of a provider means that the channel has no account there. It is stored for the time in the table, and served for up to an hour when the provider fails. Lists that always exist have no such @@ -606,8 +611,9 @@ twice. A replayed line is shown like any other. **Provider data.** Every REST request goes to the gateway first. After a network error, a timeout of 8 seconds, an answer without the header `x-petal-cache`, or a status that is neither a success nor -404, such as 429 or 503, the same request is sent to the provider directly. A 404 of the gateway is -the provider's answer and is not retried. +404, such as 429 or 503, the same request is sent to the provider directly, with the same 8 second +limit. A 404 of the gateway is the provider's answer and is not retried. A provider 404 is cached +by the gateway and is not an error: it means the channel has no account there. **Forced direct mode.** `?direct=1` on the overlay URL switches the relay and the gateway off for that overlay. The same holds for a page that was not loaded over http(s), such as a local file. @@ -630,7 +636,7 @@ log or to Analytics Engine, and they never appear in an error message or in `/ap | Overlay connection attachment | Channel, anonymous nick, phase of the handshake, time of acceptance | | KV | Answers of the providers: lists of emotes, badges and paints, under keys made of the route id and the id of a channel, an emote set or a paint | | Analytics Engine | Counters: the event (`irc-connect`, `irc-refused`, `data`, `hub-tick`, `hub-paused`, `hub-resumed`), the reason of a refusal or a pause, the route id, cache status and layer, the status code, the shard number, per alarm of a hub, how many overlays, channels, connections to Twitch, relayed lines, accepted and refused connections and failed connections to Twitch it counted and, per pause, the count that exceeded a threshold and the threshold. No channel, no address, no name, no text | -| Workers Logs | What Cloudflare writes per request and per hub event, kept 7 days: among it the URL, which names the channel in `/chat/` and `/api/irc?channel=` and carries the options of an overlay link, and the client address. The code adds error messages, which name a path, a route id or a shard number and never the query, and one line when a hub pauses itself, with the reason, the shard, a count and its threshold | +| Workers Logs | What Cloudflare writes per request and per hub event, kept 7 days: among it the URL, which names the channel in `/chat/` and `/api/irc?channel=` and carries the options of an overlay link, and the client address. Alarm invocations are logged by the platform as the scheduled time (`Date.toString()`); the hub's `alarm()` does not log. The code adds error messages, which name a path, a route id or a shard number and never the query; one line `upstream_unavailable` per provider per minute with the provider, the route id and a reason (`timeout`, `network`, `status`, `response`, `abandoned`) and never a channel; and one line when a hub pauses itself, with the reason, the shard, a count and its threshold | | Rate limiter | Counts of requests per client address and endpoint, over one minute, in the memory of Cloudflare's rate limiter | | `/api/status` | Counts per hub, the pause of a hub and the version of the deployment | | Browser, local storage | The theme choice under `theme`, once the button in the header was used. Not on overlay pages | diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 2f4219b..5c6b02a 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -15,7 +15,8 @@ The Worker and everything that belongs to it carry the product's name: the Worke | `/setup` (redirects to `/#setup`), unknown paths (404) | The Worker | One Worker request | | `/api/irc` (chat of one channel, WebSocket) | The Worker passes the connection to a `ChatHub` Durable Object, which reads the channel from Twitch | One Worker request and one Durable Object request per connection, plus the time the hub spends in memory | | `/api/data/...` (lists of emotes, badges and name paints) | The Worker's data gateway: from memory or KV, otherwise from the provider | One Worker request, usually one KV read, one KV write per refresh | -| `/api/status` (health) | The Worker, which asks every hub | One Worker request and one Durable Object request per hub | +| `/health` | The Worker, from the version binding only | One Worker request | +| `/api/status` (relay status) | The Worker, which asks every hub | One Worker request and one Durable Object request per hub | | Emote and badge images, live updates of 7TV, BetterTTV and FrankerFaceZ | The providers, directly from the browser | Never reaches Cloudflare | An overlay that loads costs one Worker request for the page, one for the chat connection and about a dozen for provider data. While it runs, a chat line costs nothing: messages from a hub to an overlay are not billed, and Cloudflare answers the overlay's keep-alive without running code. What is billed while overlays are connected is the time the hubs spend in memory. @@ -116,6 +117,7 @@ Open `https://petal.shiftbloom.studio` and check these in a browser; command lin | `/chat/?size=3&names=0` | The overlay with large text and without user names | | `/chat/?demo=1` | The overlay with sample messages, without a chat connection | | `/no-such-page` | 404 page | +| `/health` | JSON `{ ok: true, version }` with the deployment id; no hub and no provider is asked | | `/api/status` | JSON; after the overlay of a channel was opened, one hub reports a client and a joined channel | | `/api/data/bttv/3/cached/emotes/global` | JSON with the response header `x-petal-cache` | diff --git a/src/lib/chat/gateway.ts b/src/lib/chat/gateway.ts index 2adca40..4f7940c 100644 --- a/src/lib/chat/gateway.ts +++ b/src/lib/chat/gateway.ts @@ -14,7 +14,11 @@ export const GATEWAY_PREFIXES: Readonly> = { "https://api.ivr.fi/": "/api/data/ivr/", }; -/** The gateway itself gives a provider six seconds; longer than that and it is stuck. */ +/** + * The gateway itself gives a provider six seconds and answers before this. The direct attempt + * gets the same limit: a provider that hangs must fail the load, so it can be tried again later + * instead of holding it open for the rest of the stream. + */ const GATEWAY_TIMEOUT_MS = 8000; /** On every answer of the gateway, and on nothing else that may answer under its path. */ @@ -39,22 +43,23 @@ async function readJson(response: Response, url: string): Promise(url: string, request?: JsonRequest): Promise { +/** Gives up after {@link GATEWAY_TIMEOUT_MS}, including while the body is still downloading. */ +function fetchWithinDeadline(url: string, request?: JsonRequest): Promise { const deadline = new AbortController(); const timer = setTimeout(() => deadline.abort(), GATEWAY_TIMEOUT_MS); - try { - const response = await fetch(url, { ...request, signal: deadline.signal }); - // A 404 from the gateway is the provider's answer. A 404 from whatever else serves - // this origin, such as a deployment without the gateway, says nothing about the data. - const answer = response.ok || response.status === 404; - if (answer && !response.headers.has(GATEWAY_HEADER)) { - throw new Error(`${url} is not the gateway`); - } - // Awaited here, so the deadline covers the body as well. - return await readJson(response, url); - } finally { - clearTimeout(timer); + return fetch(url, { ...request, signal: deadline.signal }).finally(() => clearTimeout(timer)); +} + +async function fetchFromGateway(url: string, request?: JsonRequest): Promise { + const response = await fetchWithinDeadline(url, request); + // A 404 from the gateway is the provider's answer. A 404 from whatever else serves + // this origin, such as a deployment without the gateway, says nothing about the data. + const answer = response.ok || response.status === 404; + if (answer && !response.headers.has(GATEWAY_HEADER)) { + throw new Error(`${url} is not the gateway`); } + // Awaited here, so the deadline covers the body as well. + return readJson(response, url); } /** @@ -71,5 +76,5 @@ export async function fetchJson(url: string, request?: JsonRequest): Promise< console.warn("[gateway] no answer, asking the provider directly", error); } } - return readJson(await fetch(url, request), url); + return readJson(await fetchWithinDeadline(url, request), url); } diff --git a/src/worker/api.ts b/src/worker/api.ts index 2a53aaa..cb6d72e 100644 --- a/src/worker/api.ts +++ b/src/worker/api.ts @@ -84,6 +84,20 @@ function relayUnavailable(env: Env, shard: number): Response { return refuse("relay_unavailable", 503); } +/** + * A short reason for the log. Drops a URL or a query: both can name a channel. The name of the + * error remains, so a failure with nothing else to say is still distinct. + */ +function hubFailure(error: unknown): string { + const name = error instanceof Error ? error.name : "Error"; + const message = error instanceof Error ? error.message : ""; + const detail = message + .replace(/https?:\/\/\S+/g, "") + .replace(/\?[^ \n]*/g, "") + .trim(); + return detail ? `${name}: ${detail}` : name; +} + /** Rejects when `work` takes longer, so a hub that hangs cannot hold a request open. */ function within(milliseconds: number, work: Promise): Promise { return new Promise((resolve, reject) => { @@ -163,7 +177,10 @@ async function chat(request: Request, env: Env, url: URL): Promise { try { response = await within(HUB_TIMEOUT_MS, hub(env, shard).fetch(request)); } catch (error) { - console.error("hub did not take the connection", shard, error); + // The reason has to be part of the message. A second argument is dropped from the + // Workers log line, which otherwise only shows the shard and this file's callsite. + // A URL in the message would name the channel, so one is not kept. + console.error(`hub_unavailable shard=${shard} ${hubFailure(error)}`); return relayUnavailable(env, shard); } if (response.status === 101) { @@ -223,6 +240,19 @@ async function status(request: Request, env: Env): Promise { return json(body); } +/** + * Cheap liveness for an external check. No KV, no hub, no provider: a third-party outage must + * not fail it. `version` is the id of the running deployment, or null where the binding is absent. + */ +export function health(request: Request, env: Pick): Response { + if (request.method !== "GET" && request.method !== "HEAD") { + return refuse("method_not_allowed", 405, { allow: "GET, HEAD" }); + } + const headers = { ...JSON_HEADERS }; + if (request.method === "HEAD") return new Response(null, { status: 200, headers }); + return json({ ok: true, version: env.CF_VERSION_METADATA?.id ?? null }, 200, headers); +} + /** * Handles every request under `/api/`; the rest of the site never reaches this module. * diff --git a/src/worker/entry.ts b/src/worker/entry.ts index 5f62198..aea9209 100644 --- a/src/worker/entry.ts +++ b/src/worker/entry.ts @@ -1,5 +1,5 @@ import nitroWorker from "#petal/nitro-worker"; -import { handleApi } from "./api"; +import { handleApi, health } from "./api"; import type { Env } from "./env"; // Wrangler looks for Durable Object classes among the exports of the Worker's main module. @@ -11,6 +11,8 @@ export default { ...nitroWorker, fetch(request, env, context) { const url = new URL(request.url); + // Before Nitro, so a monitor is not rendered a page and no provider is asked. + if (url.pathname === "/health") return health(request, env); if (url.pathname.startsWith("/api/")) { return handleApi(request, env, context, url); } diff --git a/src/worker/gateway/gateway.ts b/src/worker/gateway/gateway.ts index a4e93d7..98a0493 100644 --- a/src/worker/gateway/gateway.ts +++ b/src/worker/gateway/gateway.ts @@ -125,6 +125,12 @@ const MEMORY_ENTRY_MAX = 1024 * 1024; const PAINTS_BODY_MAX = 32 * 1024; /** A defect fails every request alike, so one log line a minute says as much as all of them. */ const FAILURE_LOG_INTERVAL_MS = 60_000; +/** + * Upstream failures are logged the same way, one line per provider per minute. An outage would + * otherwise write a line per overlay, and the line has to stay distinct from a fault of this + * Worker. + */ +const UPSTREAM_LOG_INTERVAL_MS = 60_000; /** Most telling first: what a response made of several paints reports. */ const OUTCOME_RANK: readonly CacheStatus[] = ["STALE", "MISS", "EXPIRED", "UPDATING", "HIT"]; @@ -147,18 +153,34 @@ export class GatewayError extends Error { /** Both rate limits of the Worker count over a minute, after which a client may try again. */ export const rateLimited = () => new GatewayError(429, "rate_limited", 60); +/** + * Why the provider's answer cannot be stored. The words are the log: a URL is never part of + * them, because the URL of a channel route names the channel. + */ +type UpstreamReason = "timeout" | "network" | "status" | "response" | "abandoned"; + class UpstreamFailure extends Error { + readonly reason: UpstreamReason; + /** Set when the provider did answer and the status was not one we store. */ + readonly status?: number; readonly rateLimited: boolean; /** Milliseconds the provider asked us to stay away, if it said so. */ readonly retryAfterMs?: number; - constructor(reason: string, rateLimited = false, retryAfterMs?: number) { + constructor(reason: UpstreamReason, options?: { status?: number; retryAfterMs?: number }) { super(reason); - this.rateLimited = rateLimited; - this.retryAfterMs = retryAfterMs; + this.reason = reason; + this.status = options?.status; + this.rateLimited = options?.status === 429; + this.retryAfterMs = options?.retryAfterMs; } } +/** `AbortSignal.timeout` rejects with `TimeoutError`; an abort from elsewhere uses `AbortError`. */ +function timedOut(error: unknown): boolean { + return error instanceof Error && (error.name === "TimeoutError" || error.name === "AbortError"); +} + const freshFor = (entry: Entry, policy: CachePolicy) => entry.status === 404 ? (policy.negativeFresh ?? 0) : policy.fresh; @@ -245,6 +267,9 @@ export class DataGateway { /** Keyed by provider for rate limits, which hit every route, and by cache key otherwise. */ readonly #backoff = new Map(); #failureLoggedAt = Number.NEGATIVE_INFINITY; + /** Per provider, so one outage is one line however many routes it takes down. */ + readonly #upstreamLoggedAt = new Map(); + readonly #upstreamSuppressed = new Map(); constructor(options: GatewayOptions) { this.#userAgent = options.userAgent; @@ -350,7 +375,9 @@ export class DataGateway { if (blocked > 0) { throw new GatewayError(503, "upstream_backoff", Math.ceil(blocked / 1000)); } - // Waiting for an answer that is already on its way costs nothing. + // The miss limit is spent only by the request that actually asks the provider. + // Joining a fetch that is already under way is free. A retry that arrives while + // the key is left alone was refused above and never gets here. if (!this.#running(key) && context.admit && !(await context.admit())) { throw rateLimited(); } @@ -365,8 +392,10 @@ export class DataGateway { const age = this.#age(entry); const fresh = freshFor(entry, policy); if (age < fresh) return { entry, cache: "HIT", layer }; + // Any copy still held is served while the provider is left alone, however old it is. + // A 5xx here would send the overlay to the same provider that just failed. + if (blocked > 0) return { entry, cache: "STALE", layer }; const usable = age < fresh + staleIfErrorFor(entry, policy); - if (usable && blocked > 0) return { entry, cache: "STALE", layer }; if (usable && age < fresh + revalidateFor(entry, policy)) { // Failures are recorded by the refresh itself; the next request after the // backoff tries again. @@ -379,17 +408,54 @@ export class DataGateway { ); return { entry: renewed, cache: "EXPIRED", layer: "upstream" }; } catch (error) { - if (usable) return { entry, cache: "STALE", layer }; - throw this.#unavailable(key, route, error); + // Past the stale window the copy is old, and still the last answer the provider gave. + this.#noteAbandoned(route, error); + return { entry, cache: "STALE", layer }; } } #unavailable(key: string, route: GatewayRoute, error: unknown): GatewayError { if (error instanceof GatewayError) return error; + // The refresh logs the failure that rejected it. A follower that gives up while that + // refresh is still unsettled is logged here instead. + this.#noteAbandoned(route, error); const blocked = Math.ceil(this.#blockedFor(key, route) / 1000); return new GatewayError(502, "upstream_unavailable", blocked > 0 ? blocked : undefined); } + /** A follower that gives up is the only upstream failure `#failed` does not already log. */ + #noteAbandoned(route: GatewayRoute, error: unknown): void { + if (error instanceof UpstreamFailure && error.reason === "abandoned") { + this.#logUpstream(route, error); + } + } + + /** + * One line per provider per minute, under the fixed message `upstream_unavailable`, so an + * upstream outage can be told from a fault of this Worker. The route id is not a channel. + */ + #logUpstream(route: GatewayRoute, error: unknown): void { + const provider = route.provider; + const now = this.#now(); + const previous = this.#upstreamLoggedAt.get(provider) ?? Number.NEGATIVE_INFINITY; + if (now - previous < UPSTREAM_LOG_INTERVAL_MS) { + const skipped = this.#upstreamSuppressed.get(provider) ?? 0; + this.#upstreamSuppressed.set(provider, skipped + 1); + return; + } + const suppressed = this.#upstreamSuppressed.get(provider) ?? 0; + this.#upstreamSuppressed.set(provider, 0); + this.#upstreamLoggedAt.set(provider, now); + const failure = error instanceof UpstreamFailure ? error : undefined; + console.warn("upstream_unavailable", { + provider, + route: route.id, + reason: failure?.reason ?? "unknown", + ...(failure?.status === undefined ? {} : { status: failure.status }), + ...(suppressed === 0 ? {} : { suppressed }), + }); + } + /** * Waits for a refresh that another request may own. Should that request be torn down * before its fetch ends, the promise may never settle, so a follower gives up rather than @@ -399,7 +465,7 @@ export class DataGateway { const limit = after(FOLLOW_MS); try { const result = await Promise.race([refresh, limit.elapsed]); - if (!result) throw new UpstreamFailure("the refresh being followed never finished"); + if (!result) throw new UpstreamFailure("abandoned"); return result; } finally { limit.cancel(); @@ -425,6 +491,7 @@ export class DataGateway { return entry; } catch (error) { this.#failed(key, route, error); + this.#logUpstream(route, error); throw error; } finally { // Unless given up and replaced in the meantime. @@ -507,38 +574,37 @@ export class DataGateway { signal: AbortSignal.timeout(timeoutMs), }); } catch (error) { - throw new UpstreamFailure(`fetch failed: ${String(error)}`); + throw new UpstreamFailure(timedOut(error) ? "timeout" : "network"); } if (response.status === 200 || response.status === 404) return response; await response.body?.cancel(); - throw new UpstreamFailure( - `answered ${response.status}`, - response.status === 429, - parseRetryAfter(response.headers.get("retry-after"), this.#now()), - ); + throw new UpstreamFailure("status", { + status: response.status, + retryAfterMs: parseRetryAfter(response.headers.get("retry-after"), this.#now()), + }); } async #json(response: Response, maxBytes: number): Promise { const type = response.headers.get("content-type") ?? ""; if (response.status !== 200 || !/^application\/(?:[\w.-]+\+)?json\b/i.test(type)) { await response.body?.cancel(); - throw new UpstreamFailure(`answered ${response.status} ${type}`); + throw new UpstreamFailure("response"); } let text: string | undefined; try { text = await readCapped(response, maxBytes); - } catch (error) { - throw new UpstreamFailure(`body failed: ${String(error)}`); + } catch { + throw new UpstreamFailure("response"); } - if (text === undefined) throw new UpstreamFailure("body too large"); + if (text === undefined) throw new UpstreamFailure("response"); let json: unknown; try { json = JSON.parse(text); } catch { - throw new UpstreamFailure("body is not JSON"); + throw new UpstreamFailure("response"); } if (typeof json !== "object" || json === null) { - throw new UpstreamFailure("body is not a JSON object or array"); + throw new UpstreamFailure("response"); } return json; } @@ -648,7 +714,7 @@ export class DataGateway { collecting.ids.push(id); collecting.timeoutMs = Math.max(collecting.timeoutMs, timeoutMs); const loaded = (await leaving).get(id); - if (!loaded) throw new UpstreamFailure("paint missing from the answer"); + if (!loaded) throw new UpstreamFailure("response"); return loaded; }; const settled = await Promise.allSettled( @@ -701,7 +767,7 @@ export class DataGateway { }; // GraphQL reports failures with status 200, and one bad field voids them all. const paints = json.data?.paints; - if (json.errors || !paints) throw new UpstreamFailure("GraphQL error"); + if (json.errors || !paints) throw new UpstreamFailure("response"); for (const [index, id] of chunk.entries()) { const paint = paints[`p${index}`]; loaded.set( diff --git a/test/chat/gateway.test.ts b/test/chat/gateway.test.ts index fe596c8..99c0472 100644 --- a/test/chat/gateway.test.ts +++ b/test/chat/gateway.test.ts @@ -42,6 +42,21 @@ for (const [when, answer, result, asked] of [ }); } +it("gives up on the provider itself after 8 seconds", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + t.mock.method(console, "warn", () => {}); + t.after(useLocation("https://chat.shiftbloom.studio/chat/forsen?direct=1")); + const hangs: FetchAnswer = (init) => + new Promise((_, reject) => { + init?.signal?.addEventListener("abort", () => reject(init.signal?.reason)); + }); + const fetches = stubFetch({ [PROVIDER]: hangs }); + t.after(fetches); + const result = fetchJson(PROVIDER); + t.mock.timers.tick(8000); + await assert.rejects(result); +}); + it("asks the provider when the gateway has not answered after 8 seconds", async (t) => { t.mock.timers.enable({ apis: ["setTimeout"] }); /** Never answers, and fails like fetch() does once the request is aborted. */ diff --git a/test/worker/api.test.ts b/test/worker/api.test.ts index 92bb0d1..cb06798 100644 --- a/test/worker/api.test.ts +++ b/test/worker/api.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { describe, it } from "node:test"; -import { handleApi } from "../../src/worker/api"; +import { handleApi, health } from "../../src/worker/api"; import type { Env } from "../../src/worker/env"; import { shardOf } from "../../src/worker/relay/shard"; @@ -96,6 +96,60 @@ describe("handleApi", () => { assert.equal(await call(env, CHANNEL, COUNTED), UPGRADED); }); + it("names the shard and the reason when a hub does not take the connection", async (t) => { + // Enabling the mock clock warns on stderr; mock afterwards so that line is not the log. + t.mock.timers.enable({ apis: ["setTimeout"] }); + const logged = t.mock.method(console, "error", () => {}); + const response = call(fakeEnv({}, () => new Promise(() => {})).env, CHANNEL); + await new Promise((resolve) => setImmediate(resolve)); + t.mock.timers.tick(5000); + assert.deepEqual(await refusal(response), [503, "relay_unavailable"]); + const lines = logged.mock.calls.map((call) => String(call.arguments[0])); + assert.match(lines.join("\n"), /^hub_unavailable shard=\d+ Error: timed out$/m); + }); + + it("does not log the channel when a hub failure names the request", async (t) => { + const logged = t.mock.method(console, "error", () => {}); + const answer = () => + Promise.reject( + new Error("fetch failed https://chat.example/api/irc?channel=somechannel"), + ); + assert.deepEqual(await refusal(call(fakeEnv({}, answer).env, CHANNEL)), [ + 503, + "relay_unavailable", + ]); + const text = logged.mock.calls.map((call) => String(call.arguments[0])).join("\n"); + assert.match(text, /hub_unavailable shard=\d+ Error: fetch failed$/m); + assert.equal(text.includes("somechannel"), false); + assert.equal(text.includes("?"), false); + }); + + it("answers /health from the version binding alone", async () => { + const version = { id: "test-version", tag: "", timestamp: "" }; + const { env, hubs } = fakeEnv({ CF_VERSION_METADATA: version } as Partial); + const response = await health(new Request("https://chat.example/health"), env); + assert.equal(response.status, 200); + assert.equal(response.headers.get("cache-control"), "no-store"); + assert.deepEqual(await response.json(), { ok: true, version: version.id }); + const missing = await health(new Request("https://chat.example/health"), fakeEnv().env); + assert.deepEqual(await missing.json(), { ok: true, version: null }); + const head = await health( + new Request("https://chat.example/health", { method: "HEAD" }), + env, + ); + assert.equal(head.status, 200); + assert.equal(await head.text(), ""); + assert.deepEqual( + await refusal( + Promise.resolve( + health(new Request("https://chat.example/health", { method: "POST" }), env), + ), + ), + [405, "method_not_allowed"], + ); + assert.equal(hubs.length, 0); + }); + it("answers 503 when a hub fails, hangs or does not upgrade, and hands on its refusal", async (t) => { t.mock.method(console, "error", () => {}); t.mock.timers.enable({ apis: ["setTimeout"] }); diff --git a/test/worker/gateway/gateway.test.ts b/test/worker/gateway/gateway.test.ts index 80535c6..217d022 100644 --- a/test/worker/gateway/gateway.test.ts +++ b/test/worker/gateway/gateway.test.ts @@ -7,8 +7,12 @@ import * as paints from "../../../src/worker/gateway/paints"; const BTTV_GLOBAL = "https://api.betterttv.net/3/cached/emotes/global"; const BTTV_USER = "https://api.betterttv.net/3/cached/users/twitch/50985620"; +const FFZ_ROOM_URL = "https://api.frankerfacez.com/v1/room/id/50985620"; +const FFZ_GLOBAL_URL = "https://api.frankerfacez.com/v1/set/global"; const GLOBAL = "/bttv/3/cached/emotes/global"; const USER = "/bttv/3/cached/users/twitch/50985620"; +const FFZ_ROOM = "/ffz/v1/room/id/50985620"; +const FFZ_GLOBAL = "/ffz/v1/set/global"; const START = Date.UTC(2026, 8, 29, 12); const paintId = (n: number) => `01FQB6K5T0000BDD0YMN2${String(n).padStart(5, "0")}`; @@ -145,6 +149,85 @@ describe("provider failures", () => { }); } + it("stores a provider 404 and does not treat it as a failure", async (t) => { + const warned = t.mock.method(console, "warn", () => {}); + const { answers, calls, get, settle } = setup(); + answers.set(FFZ_ROOM_URL, () => new Response("missing", { status: 404 })); + const first = await get(FFZ_ROOM); + assert.equal(first.status, 404); + assert.equal(cacheOf(first), "MISS; layer=upstream"); + assert.equal(await first.text(), "null"); + await settle(); + const second = await get(FFZ_ROOM); + assert.equal(second.status, 404); + assert.equal(cacheOf(second), "HIT; layer=memory"); + assert.equal(calls.length, 1); + assert.equal(warned.mock.calls.length, 0); + }); + + it("does not store a 404 of a list that always exists", async (t) => { + const warned = t.mock.method(console, "warn", () => {}); + const { answers, calls, get, settle, stored } = setup(); + answers.set(FFZ_GLOBAL_URL, () => new Response("nope", { status: 404 })); + const failed = await get(FFZ_GLOBAL); + assert.equal(failed.status, 502); + assert.equal(cacheOf(failed), "ERROR"); + await settle(); + assert.equal(stored.size, 0); + assert.equal(calls.length, 1); + const [message, detail] = warned.mock.calls[0].arguments as [ + string, + { provider: string; route: string; reason: string }, + ]; + assert.equal(message, "upstream_unavailable"); + assert.deepEqual(detail, { provider: "ffz", route: "ffz.global", reason: "response" }); + }); + + it("serves a copy that has outlived its window when the provider fails", async () => { + const { advance, answers, calls, json, settle, get } = setup(); + json(FFZ_ROOM_URL, { room: {}, sets: {} }); + await get(FFZ_ROOM); + await settle(); + advance(8 * 86400); + answers.set(FFZ_ROOM_URL, () => new Response(null, { status: 502 })); + const stale = await get(FFZ_ROOM); + assert.equal(stale.status, 200); + assert.equal(cacheOf(stale), "STALE; layer=memory"); + assert.equal(calls.length, 2); + // The failure leaves the provider alone, so the retry neither fetches nor spends a miss. + const again = await get(FFZ_ROOM); + assert.equal(again.status, 200); + assert.equal(cacheOf(again), "STALE; layer=memory"); + assert.equal(calls.length, 2); + }); + + it("does not spend a miss on a retry while the provider is left alone", async () => { + const { answers, calls, get, misses } = setup(); + answers.set(FFZ_ROOM_URL, () => new Response(null, { status: 503 })); + misses.allowed = 1; + assert.equal((await get(FFZ_ROOM)).status, 502); + assert.equal(misses.allowed, 0); + assert.equal((await get(FFZ_ROOM)).status, 503); + assert.equal(misses.allowed, 0); + assert.equal(calls.length, 1); + }); + + it("logs an upstream timeout with the provider and without the channel", async (t) => { + const warned = t.mock.method(console, "warn", () => {}); + const { answers, get } = setup(); + answers.set(FFZ_ROOM_URL, () => + Promise.reject(Object.assign(new Error("aborted"), { name: "TimeoutError" })), + ); + assert.equal((await get(FFZ_ROOM)).status, 502); + const [message, detail] = warned.mock.calls[0].arguments as [ + string, + { provider: string; route: string; reason: string }, + ]; + assert.equal(message, "upstream_unavailable"); + assert.deepEqual(detail, { provider: "ffz", route: "ffz.room", reason: "timeout" }); + assert.equal(JSON.stringify(warned.mock.calls).includes("50985620"), false); + }); + it("gives up on a provider that never answers", async (t) => { t.mock.timers.enable({ apis: ["setTimeout"] }); const { answers, get } = setup(); From 2a2c278fda6b328f4acf6418830a729a7947901d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 13 Nov 2026 23:03:53 +0000 Subject: [PATCH 2/3] refactor(gateway): name the upstream log reasons One helper writes Worker log lines. Provider failure reasons are a const, and the 7TV test id prefix is named. Co-authored-by: Fabian Zimber --- src/lib/chat/gateway.ts | 6 --- src/worker/api.ts | 23 +++------ src/worker/entry.ts | 1 - src/worker/gateway/gateway.ts | 73 +++++++++++++---------------- src/worker/log.ts | 19 ++++++++ test/worker/api.test.ts | 4 +- test/worker/gateway/gateway.test.ts | 24 ++++------ 7 files changed, 70 insertions(+), 80 deletions(-) create mode 100644 src/worker/log.ts diff --git a/src/lib/chat/gateway.ts b/src/lib/chat/gateway.ts index 4f7940c..0ee13df 100644 --- a/src/lib/chat/gateway.ts +++ b/src/lib/chat/gateway.ts @@ -14,11 +14,6 @@ export const GATEWAY_PREFIXES: Readonly> = { "https://api.ivr.fi/": "/api/data/ivr/", }; -/** - * The gateway itself gives a provider six seconds and answers before this. The direct attempt - * gets the same limit: a provider that hangs must fail the load, so it can be tried again later - * instead of holding it open for the rest of the stream. - */ const GATEWAY_TIMEOUT_MS = 8000; /** On every answer of the gateway, and on nothing else that may answer under its path. */ @@ -43,7 +38,6 @@ async function readJson(response: Response, url: string): Promise { const deadline = new AbortController(); const timer = setTimeout(() => deadline.abort(), GATEWAY_TIMEOUT_MS); diff --git a/src/worker/api.ts b/src/worker/api.ts index cb6d72e..e7da661 100644 --- a/src/worker/api.ts +++ b/src/worker/api.ts @@ -3,10 +3,12 @@ import { record, recordGateway } from "./analytics"; import type { Env } from "./env"; import { handleData } from "./gateway/index"; import type { HubStatus } from "./hub"; +import { logWorkerEvent, WorkerLogEvent } from "./log"; import { withinLimit } from "./rate-limit"; import { normaliseChannel, shardOf } from "./relay/shard"; const DEFAULT_SHARDS = 4; +const MESSAGE_LOCATOR = /https?:\/\/\S+|\?[^ \n]*/g; /** A slip such as `400` would make every status request wake that many hubs. */ const MAX_SHARDS = 64; @@ -84,17 +86,10 @@ function relayUnavailable(env: Env, shard: number): Response { return refuse("relay_unavailable", 503); } -/** - * A short reason for the log. Drops a URL or a query: both can name a channel. The name of the - * error remains, so a failure with nothing else to say is still distinct. - */ function hubFailure(error: unknown): string { const name = error instanceof Error ? error.name : "Error"; const message = error instanceof Error ? error.message : ""; - const detail = message - .replace(/https?:\/\/\S+/g, "") - .replace(/\?[^ \n]*/g, "") - .trim(); + const detail = message.replace(MESSAGE_LOCATOR, "").trim(); return detail ? `${name}: ${detail}` : name; } @@ -177,10 +172,10 @@ async function chat(request: Request, env: Env, url: URL): Promise { try { response = await within(HUB_TIMEOUT_MS, hub(env, shard).fetch(request)); } catch (error) { - // The reason has to be part of the message. A second argument is dropped from the - // Workers log line, which otherwise only shows the shard and this file's callsite. - // A URL in the message would name the channel, so one is not kept. - console.error(`hub_unavailable shard=${shard} ${hubFailure(error)}`); + logWorkerEvent("error", WorkerLogEvent.hubUnavailable, { + shard, + error: hubFailure(error), + }); return relayUnavailable(env, shard); } if (response.status === 101) { @@ -240,10 +235,6 @@ async function status(request: Request, env: Env): Promise { return json(body); } -/** - * Cheap liveness for an external check. No KV, no hub, no provider: a third-party outage must - * not fail it. `version` is the id of the running deployment, or null where the binding is absent. - */ export function health(request: Request, env: Pick): Response { if (request.method !== "GET" && request.method !== "HEAD") { return refuse("method_not_allowed", 405, { allow: "GET, HEAD" }); diff --git a/src/worker/entry.ts b/src/worker/entry.ts index aea9209..8b827c6 100644 --- a/src/worker/entry.ts +++ b/src/worker/entry.ts @@ -11,7 +11,6 @@ export default { ...nitroWorker, fetch(request, env, context) { const url = new URL(request.url); - // Before Nitro, so a monitor is not rendered a page and no provider is asked. if (url.pathname === "/health") return health(request, env); if (url.pathname.startsWith("/api/")) { return handleApi(request, env, context, url); diff --git a/src/worker/gateway/gateway.ts b/src/worker/gateway/gateway.ts index 98a0493..c218f58 100644 --- a/src/worker/gateway/gateway.ts +++ b/src/worker/gateway/gateway.ts @@ -1,3 +1,4 @@ +import { logWorkerEvent, WorkerLogEvent } from "../log"; import { buildPaintsQuery, PAINTS_PER_QUERY, parsePaintsRequest } from "./paints"; import { type CachePolicy, @@ -125,11 +126,6 @@ const MEMORY_ENTRY_MAX = 1024 * 1024; const PAINTS_BODY_MAX = 32 * 1024; /** A defect fails every request alike, so one log line a minute says as much as all of them. */ const FAILURE_LOG_INTERVAL_MS = 60_000; -/** - * Upstream failures are logged the same way, one line per provider per minute. An outage would - * otherwise write a line per overlay, and the line has to stay distinct from a fault of this - * Worker. - */ const UPSTREAM_LOG_INTERVAL_MS = 60_000; /** Most telling first: what a response made of several paints reports. */ @@ -153,15 +149,21 @@ export class GatewayError extends Error { /** Both rate limits of the Worker count over a minute, after which a client may try again. */ export const rateLimited = () => new GatewayError(429, "rate_limited", 60); -/** - * Why the provider's answer cannot be stored. The words are the log: a URL is never part of - * them, because the URL of a channel route names the channel. - */ -type UpstreamReason = "timeout" | "network" | "status" | "response" | "abandoned"; +const UpstreamReason = { + timeout: "timeout", + network: "network", + status: "status", + response: "response", + abandoned: "abandoned", + unknown: "unknown", +} as const; + +type UpstreamReason = (typeof UpstreamReason)[keyof typeof UpstreamReason]; + +const FETCH_ABORT_NAMES = new Set(["TimeoutError", "AbortError"]); class UpstreamFailure extends Error { readonly reason: UpstreamReason; - /** Set when the provider did answer and the status was not one we store. */ readonly status?: number; readonly rateLimited: boolean; /** Milliseconds the provider asked us to stay away, if it said so. */ @@ -176,9 +178,8 @@ class UpstreamFailure extends Error { } } -/** `AbortSignal.timeout` rejects with `TimeoutError`; an abort from elsewhere uses `AbortError`. */ function timedOut(error: unknown): boolean { - return error instanceof Error && (error.name === "TimeoutError" || error.name === "AbortError"); + return error instanceof Error && FETCH_ABORT_NAMES.has(error.name); } const freshFor = (entry: Entry, policy: CachePolicy) => @@ -375,9 +376,6 @@ export class DataGateway { if (blocked > 0) { throw new GatewayError(503, "upstream_backoff", Math.ceil(blocked / 1000)); } - // The miss limit is spent only by the request that actually asks the provider. - // Joining a fetch that is already under way is free. A retry that arrives while - // the key is left alone was refused above and never gets here. if (!this.#running(key) && context.admit && !(await context.admit())) { throw rateLimited(); } @@ -392,8 +390,6 @@ export class DataGateway { const age = this.#age(entry); const fresh = freshFor(entry, policy); if (age < fresh) return { entry, cache: "HIT", layer }; - // Any copy still held is served while the provider is left alone, however old it is. - // A 5xx here would send the overlay to the same provider that just failed. if (blocked > 0) return { entry, cache: "STALE", layer }; const usable = age < fresh + staleIfErrorFor(entry, policy); if (usable && age < fresh + revalidateFor(entry, policy)) { @@ -408,7 +404,6 @@ export class DataGateway { ); return { entry: renewed, cache: "EXPIRED", layer: "upstream" }; } catch (error) { - // Past the stale window the copy is old, and still the last answer the provider gave. this.#noteAbandoned(route, error); return { entry, cache: "STALE", layer }; } @@ -416,24 +411,17 @@ export class DataGateway { #unavailable(key: string, route: GatewayRoute, error: unknown): GatewayError { if (error instanceof GatewayError) return error; - // The refresh logs the failure that rejected it. A follower that gives up while that - // refresh is still unsettled is logged here instead. this.#noteAbandoned(route, error); const blocked = Math.ceil(this.#blockedFor(key, route) / 1000); return new GatewayError(502, "upstream_unavailable", blocked > 0 ? blocked : undefined); } - /** A follower that gives up is the only upstream failure `#failed` does not already log. */ #noteAbandoned(route: GatewayRoute, error: unknown): void { if (error instanceof UpstreamFailure && error.reason === "abandoned") { this.#logUpstream(route, error); } } - /** - * One line per provider per minute, under the fixed message `upstream_unavailable`, so an - * upstream outage can be told from a fault of this Worker. The route id is not a channel. - */ #logUpstream(route: GatewayRoute, error: unknown): void { const provider = route.provider; const now = this.#now(); @@ -447,13 +435,14 @@ export class DataGateway { this.#upstreamSuppressed.set(provider, 0); this.#upstreamLoggedAt.set(provider, now); const failure = error instanceof UpstreamFailure ? error : undefined; - console.warn("upstream_unavailable", { + const fields: Record = { provider, route: route.id, - reason: failure?.reason ?? "unknown", - ...(failure?.status === undefined ? {} : { status: failure.status }), - ...(suppressed === 0 ? {} : { suppressed }), - }); + reason: failure?.reason ?? UpstreamReason.unknown, + }; + if (failure?.status !== undefined) fields.status = failure.status; + if (suppressed > 0) fields.suppressed = suppressed; + logWorkerEvent("warn", WorkerLogEvent.upstreamUnavailable, fields); } /** @@ -465,7 +454,7 @@ export class DataGateway { const limit = after(FOLLOW_MS); try { const result = await Promise.race([refresh, limit.elapsed]); - if (!result) throw new UpstreamFailure("abandoned"); + if (!result) throw new UpstreamFailure(UpstreamReason.abandoned); return result; } finally { limit.cancel(); @@ -574,11 +563,13 @@ export class DataGateway { signal: AbortSignal.timeout(timeoutMs), }); } catch (error) { - throw new UpstreamFailure(timedOut(error) ? "timeout" : "network"); + throw new UpstreamFailure( + timedOut(error) ? UpstreamReason.timeout : UpstreamReason.network, + ); } if (response.status === 200 || response.status === 404) return response; await response.body?.cancel(); - throw new UpstreamFailure("status", { + throw new UpstreamFailure(UpstreamReason.status, { status: response.status, retryAfterMs: parseRetryAfter(response.headers.get("retry-after"), this.#now()), }); @@ -588,23 +579,23 @@ export class DataGateway { const type = response.headers.get("content-type") ?? ""; if (response.status !== 200 || !/^application\/(?:[\w.-]+\+)?json\b/i.test(type)) { await response.body?.cancel(); - throw new UpstreamFailure("response"); + throw new UpstreamFailure(UpstreamReason.response); } let text: string | undefined; try { text = await readCapped(response, maxBytes); } catch { - throw new UpstreamFailure("response"); + throw new UpstreamFailure(UpstreamReason.response); } - if (text === undefined) throw new UpstreamFailure("response"); + if (text === undefined) throw new UpstreamFailure(UpstreamReason.response); let json: unknown; try { json = JSON.parse(text); } catch { - throw new UpstreamFailure("response"); + throw new UpstreamFailure(UpstreamReason.response); } if (typeof json !== "object" || json === null) { - throw new UpstreamFailure("response"); + throw new UpstreamFailure(UpstreamReason.response); } return json; } @@ -714,7 +705,7 @@ export class DataGateway { collecting.ids.push(id); collecting.timeoutMs = Math.max(collecting.timeoutMs, timeoutMs); const loaded = (await leaving).get(id); - if (!loaded) throw new UpstreamFailure("response"); + if (!loaded) throw new UpstreamFailure(UpstreamReason.response); return loaded; }; const settled = await Promise.allSettled( @@ -767,7 +758,7 @@ export class DataGateway { }; // GraphQL reports failures with status 200, and one bad field voids them all. const paints = json.data?.paints; - if (json.errors || !paints) throw new UpstreamFailure("response"); + if (json.errors || !paints) throw new UpstreamFailure(UpstreamReason.response); for (const [index, id] of chunk.entries()) { const paint = paints[`p${index}`]; loaded.set( diff --git a/src/worker/log.ts b/src/worker/log.ts new file mode 100644 index 0000000..06657e7 --- /dev/null +++ b/src/worker/log.ts @@ -0,0 +1,19 @@ +export const WorkerLogEvent = { + upstreamUnavailable: "upstream_unavailable", + hubUnavailable: "hub_unavailable", +} as const; + +export type WorkerLogEventName = (typeof WorkerLogEvent)[keyof typeof WorkerLogEvent]; + +export function logWorkerEvent( + level: "warn" | "error", + event: WorkerLogEventName, + fields: Readonly>, +): void { + const body = Object.entries(fields) + .map(([key, value]) => `${key}=${value}`) + .join(" "); + const line = body.length > 0 ? `${event} ${body}` : event; + if (level === "error") console.error(line); + else console.warn(line); +} diff --git a/test/worker/api.test.ts b/test/worker/api.test.ts index cb06798..094b56e 100644 --- a/test/worker/api.test.ts +++ b/test/worker/api.test.ts @@ -105,7 +105,7 @@ describe("handleApi", () => { t.mock.timers.tick(5000); assert.deepEqual(await refusal(response), [503, "relay_unavailable"]); const lines = logged.mock.calls.map((call) => String(call.arguments[0])); - assert.match(lines.join("\n"), /^hub_unavailable shard=\d+ Error: timed out$/m); + assert.match(lines.join("\n"), /^hub_unavailable shard=\d+ error=Error: timed out$/m); }); it("does not log the channel when a hub failure names the request", async (t) => { @@ -119,7 +119,7 @@ describe("handleApi", () => { "relay_unavailable", ]); const text = logged.mock.calls.map((call) => String(call.arguments[0])).join("\n"); - assert.match(text, /hub_unavailable shard=\d+ Error: fetch failed$/m); + assert.match(text, /hub_unavailable shard=\d+ error=Error: fetch failed$/m); assert.equal(text.includes("somechannel"), false); assert.equal(text.includes("?"), false); }); diff --git a/test/worker/gateway/gateway.test.ts b/test/worker/gateway/gateway.test.ts index 217d022..4129672 100644 --- a/test/worker/gateway/gateway.test.ts +++ b/test/worker/gateway/gateway.test.ts @@ -15,7 +15,8 @@ const FFZ_ROOM = "/ffz/v1/room/id/50985620"; const FFZ_GLOBAL = "/ffz/v1/set/global"; const START = Date.UTC(2026, 8, 29, 12); -const paintId = (n: number) => `01FQB6K5T0000BDD0YMN2${String(n).padStart(5, "0")}`; +const SEVENTV_ULID_PREFIX = "01FQB6K5T0000BDD0YMN2"; +const paintId = (n: number) => `${SEVENTV_ULID_PREFIX}${String(n).padStart(5, "0")}`; const paint = (id: string) => ({ id, name: "Paint", data: { layers: [], shadows: [] } }); /** Answers a paints query the way 7TV does for paints it knows. */ @@ -175,12 +176,10 @@ describe("provider failures", () => { await settle(); assert.equal(stored.size, 0); assert.equal(calls.length, 1); - const [message, detail] = warned.mock.calls[0].arguments as [ - string, - { provider: string; route: string; reason: string }, - ]; - assert.equal(message, "upstream_unavailable"); - assert.deepEqual(detail, { provider: "ffz", route: "ffz.global", reason: "response" }); + assert.equal( + warned.mock.calls[0]?.arguments[0], + "upstream_unavailable provider=ffz route=ffz.global reason=response", + ); }); it("serves a copy that has outlived its window when the provider fails", async () => { @@ -194,7 +193,6 @@ describe("provider failures", () => { assert.equal(stale.status, 200); assert.equal(cacheOf(stale), "STALE; layer=memory"); assert.equal(calls.length, 2); - // The failure leaves the provider alone, so the retry neither fetches nor spends a miss. const again = await get(FFZ_ROOM); assert.equal(again.status, 200); assert.equal(cacheOf(again), "STALE; layer=memory"); @@ -219,12 +217,10 @@ describe("provider failures", () => { Promise.reject(Object.assign(new Error("aborted"), { name: "TimeoutError" })), ); assert.equal((await get(FFZ_ROOM)).status, 502); - const [message, detail] = warned.mock.calls[0].arguments as [ - string, - { provider: string; route: string; reason: string }, - ]; - assert.equal(message, "upstream_unavailable"); - assert.deepEqual(detail, { provider: "ffz", route: "ffz.room", reason: "timeout" }); + assert.equal( + warned.mock.calls[0]?.arguments[0], + "upstream_unavailable provider=ffz route=ffz.room reason=timeout", + ); assert.equal(JSON.stringify(warned.mock.calls).includes("50985620"), false); }); From 9a9ece8af3cd8fa8ef74857145a5277f5bce118c Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 13 Nov 2026 23:12:05 +0000 Subject: [PATCH 3/3] refactor(gateway): type degrade failures and drop narrative comments Name upstream reasons, log Cloudflare platform failures through one error type, and let the cache path read as early returns. Co-authored-by: Fabian Zimber --- docs/ARCHITECTURE.md | 6 +- src/lib/chat/gateway.ts | 72 +++++++-- src/worker/api.ts | 47 +++--- src/worker/entry.ts | 4 +- src/worker/gateway/gateway.ts | 228 +++++++++++++++++----------- src/worker/log.ts | 66 ++++++-- test/chat/gateway.test.ts | 18 ++- test/worker/api.test.ts | 22 ++- test/worker/gateway/gateway.test.ts | 64 +++++--- 9 files changed, 351 insertions(+), 176 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e829973..d44c12f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -535,8 +535,8 @@ not a proxy for arbitrary URLs. The three windows follow each other. While an answer is fresh it is served without asking the provider. In the window after that, lists that every overlay shares are served at once and refreshed in the background. Channel data has no such window: a reload of the overlay must show an emote that -was added a minute ago. In the last window the stored answer is only served when the provider fails. A copy that is -still held after that window is served the same way when the provider fails, instead of a 502. +was added a minute ago. In the last window the stored answer is only served when the provider fails. +A copy still held after that window is served the same way when the provider fails, instead of a 502. A 404 of a provider means that the channel has no account there. It is stored for the time in the table, and served for up to an hour when the provider fails. Lists that always exist have no such @@ -636,7 +636,7 @@ log or to Analytics Engine, and they never appear in an error message or in `/ap | Overlay connection attachment | Channel, anonymous nick, phase of the handshake, time of acceptance | | KV | Answers of the providers: lists of emotes, badges and paints, under keys made of the route id and the id of a channel, an emote set or a paint | | Analytics Engine | Counters: the event (`irc-connect`, `irc-refused`, `data`, `hub-tick`, `hub-paused`, `hub-resumed`), the reason of a refusal or a pause, the route id, cache status and layer, the status code, the shard number, per alarm of a hub, how many overlays, channels, connections to Twitch, relayed lines, accepted and refused connections and failed connections to Twitch it counted and, per pause, the count that exceeded a threshold and the threshold. No channel, no address, no name, no text | -| Workers Logs | What Cloudflare writes per request and per hub event, kept 7 days: among it the URL, which names the channel in `/chat/` and `/api/irc?channel=` and carries the options of an overlay link, and the client address. Alarm invocations are logged by the platform as the scheduled time (`Date.toString()`); the hub's `alarm()` does not log. The code adds error messages, which name a path, a route id or a shard number and never the query; one line `upstream_unavailable` per provider per minute with the provider, the route id and a reason (`timeout`, `network`, `status`, `response`, `abandoned`) and never a channel; and one line when a hub pauses itself, with the reason, the shard, a count and its threshold | +| Workers Logs | What Cloudflare writes per request and per hub event, kept 7 days: among it the URL, which names the channel in `/chat/` and `/api/irc?channel=` and carries the options of an overlay link, and the client address. Alarm invocations are logged by the platform as the scheduled time (`Date.toString()`); the hub's `alarm()` does not log. The code adds error messages, which name a path, a route id or a shard number and never the query; one JSON line `upstream_unavailable` per provider per minute with the provider, the route id and a reason (`timeout`, `network`, `status`, `response`, `abandoned`, `unknown`) and never a channel; and one line when a hub pauses itself, with the reason, the shard, a count and its threshold | | Rate limiter | Counts of requests per client address and endpoint, over one minute, in the memory of Cloudflare's rate limiter | | `/api/status` | Counts per hub, the pause of a hub and the version of the deployment | | Browser, local storage | The theme choice under `theme`, once the button in the header was used. Not on overlay pages | diff --git a/src/lib/chat/gateway.ts b/src/lib/chat/gateway.ts index 0ee13df..d26042c 100644 --- a/src/lib/chat/gateway.ts +++ b/src/lib/chat/gateway.ts @@ -14,7 +14,7 @@ export const GATEWAY_PREFIXES: Readonly> = { "https://api.ivr.fi/": "/api/data/ivr/", }; -const GATEWAY_TIMEOUT_MS = 8000; +export const PROVIDER_FETCH_TIMEOUT_MS = 8_000; /** On every answer of the gateway, and on nothing else that may answer under its path. */ const GATEWAY_HEADER = "x-petal-cache"; @@ -32,28 +32,70 @@ export function gatewayUrl(url: string): string | undefined { return undefined; } +const NOT_FOUND = 404; + async function readJson(response: Response, url: string): Promise { - if (response.status === 404) return undefined; + if (response.status === NOT_FOUND) return undefined; if (!response.ok) throw new Error(`${url} answered ${response.status}`); return (await response.json()) as T; } -function fetchWithinDeadline(url: string, request?: JsonRequest): Promise { - const deadline = new AbortController(); - const timer = setTimeout(() => deadline.abort(), GATEWAY_TIMEOUT_MS); - return fetch(url, { ...request, signal: deadline.signal }).finally(() => clearTimeout(timer)); +interface Deadline { + signal: AbortSignal; + elapsed: Promise; + stop: () => void; } -async function fetchFromGateway(url: string, request?: JsonRequest): Promise { - const response = await fetchWithinDeadline(url, request); - // A 404 from the gateway is the provider's answer. A 404 from whatever else serves - // this origin, such as a deployment without the gateway, says nothing about the data. - const answer = response.ok || response.status === 404; - if (answer && !response.headers.has(GATEWAY_HEADER)) { +function deadlineAfter(milliseconds: number): Deadline { + const controller = new AbortController(); + let stop = () => {}; + const elapsed = new Promise((_, reject) => { + const timer = setTimeout(() => { + controller.abort(); + reject(controller.signal.reason); + }, milliseconds); + stop = () => clearTimeout(timer); + }); + return { signal: controller.signal, elapsed, stop }; +} + +function providerStatus(response: Response): boolean { + return response.ok || response.status === NOT_FOUND; +} + +function requireGatewayResponse(response: Response, url: string): void { + if (providerStatus(response) && !response.headers.has(GATEWAY_HEADER)) { throw new Error(`${url} is not the gateway`); } - // Awaited here, so the deadline covers the body as well. - return readJson(response, url); +} + +function preventUnhandledRejection(work: Promise): Promise { + work.catch(() => undefined); + return work; +} + +async function readWithinTimeout( + url: string, + request: JsonRequest | undefined, + read: (response: Response) => Promise, +): Promise { + const deadline = deadlineAfter(PROVIDER_FETCH_TIMEOUT_MS); + try { + const response = await Promise.race([ + preventUnhandledRejection(fetch(url, { ...request, signal: deadline.signal })), + deadline.elapsed, + ]); + return await Promise.race([preventUnhandledRejection(read(response)), deadline.elapsed]); + } finally { + deadline.stop(); + } +} + +async function fetchFromGateway(url: string, request?: JsonRequest): Promise { + return readWithinTimeout(url, request, async (response) => { + requireGatewayResponse(response, url); + return readJson(response, url); + }); } /** @@ -70,5 +112,5 @@ export async function fetchJson(url: string, request?: JsonRequest): Promise< console.warn("[gateway] no answer, asking the provider directly", error); } } - return readJson(await fetchWithinDeadline(url, request), url); + return readWithinTimeout(url, request, (response) => readJson(response, url)); } diff --git a/src/worker/api.ts b/src/worker/api.ts index e7da661..9d93a19 100644 --- a/src/worker/api.ts +++ b/src/worker/api.ts @@ -3,12 +3,11 @@ import { record, recordGateway } from "./analytics"; import type { Env } from "./env"; import { handleData } from "./gateway/index"; import type { HubStatus } from "./hub"; -import { logWorkerEvent, WorkerLogEvent } from "./log"; +import { logPlatformError, PlatformError, PlatformEvent } from "./log"; import { withinLimit } from "./rate-limit"; import { normaliseChannel, shardOf } from "./relay/shard"; const DEFAULT_SHARDS = 4; -const MESSAGE_LOCATOR = /https?:\/\/\S+|\?[^ \n]*/g; /** A slip such as `400` would make every status request wake that many hubs. */ const MAX_SHARDS = 64; @@ -20,7 +19,7 @@ const MAX_SHARDS = 64; const HUB_LOCATION: DurableObjectLocationHint = "weur"; /** An overlay gives the relay 15 seconds; answering sooner sends it to Twitch sooner. */ -const HUB_TIMEOUT_MS = 5000; +export const HUB_TIMEOUT_MS = 5000; const STATUS_TIMEOUT_MS = 3000; const DATA_PREFIX = "/api/data/"; @@ -86,13 +85,6 @@ function relayUnavailable(env: Env, shard: number): Response { return refuse("relay_unavailable", 503); } -function hubFailure(error: unknown): string { - const name = error instanceof Error ? error.name : "Error"; - const message = error instanceof Error ? error.message : ""; - const detail = message.replace(MESSAGE_LOCATOR, "").trim(); - return detail ? `${name}: ${detail}` : name; -} - /** Rejects when `work` takes longer, so a hub that hangs cannot hold a request open. */ function within(milliseconds: number, work: Promise): Promise { return new Promise((resolve, reject) => { @@ -172,10 +164,7 @@ async function chat(request: Request, env: Env, url: URL): Promise { try { response = await within(HUB_TIMEOUT_MS, hub(env, shard).fetch(request)); } catch (error) { - logWorkerEvent("error", WorkerLogEvent.hubUnavailable, { - shard, - error: hubFailure(error), - }); + logPlatformError(new PlatformError(PlatformEvent.HubUnavailable, error, { shard })); return relayUnavailable(env, shard); } if (response.status === 101) { @@ -210,7 +199,7 @@ async function shardStatus(env: Env, shard: number): Promise { ); return { ...status, shard, available: true }; } catch (error) { - console.error("hub did not report its status", shard, error); + logPlatformError(new PlatformError(PlatformEvent.HubStatusUnavailable, error, { shard })); return { shard, available: false }; } } @@ -235,13 +224,24 @@ async function status(request: Request, env: Env): Promise { return json(body); } +const HEALTH_METHODS = new Set(["GET", "HEAD"]); +const HEALTH_ALLOW = "GET, HEAD"; + +export const HEALTH_PATH = "/health"; + export function health(request: Request, env: Pick): Response { - if (request.method !== "GET" && request.method !== "HEAD") { - return refuse("method_not_allowed", 405, { allow: "GET, HEAD" }); + if (!HEALTH_METHODS.has(request.method)) { + return refuse("method_not_allowed", 405, { allow: HEALTH_ALLOW }); } - const headers = { ...JSON_HEADERS }; - if (request.method === "HEAD") return new Response(null, { status: 200, headers }); - return json({ ok: true, version: env.CF_VERSION_METADATA?.id ?? null }, 200, headers); + if (request.method === "HEAD") { + return new Response(null, { status: 200, headers: JSON_HEADERS }); + } + return json({ ok: true, version: env.CF_VERSION_METADATA?.id ?? null }); +} + +function channelFreeRoute(pathname: string): string { + if (pathname.startsWith(DATA_PREFIX)) return "data"; + return pathname; } /** @@ -272,8 +272,11 @@ export async function handleApi( if (url.pathname === "/api/status") return await status(request, env); return refuse("not_found", 404); } catch (error) { - // Neither the path nor the query is logged: both can name a channel. - console.error("api request failed", data ? "data" : url.pathname, error); + logPlatformError( + new PlatformError(PlatformEvent.ApiFailed, error, { + route: channelFreeRoute(url.pathname), + }), + ); return refuse("unavailable", 503, data ? GATEWAY_REFUSED : undefined); } } diff --git a/src/worker/entry.ts b/src/worker/entry.ts index 8b827c6..85d8235 100644 --- a/src/worker/entry.ts +++ b/src/worker/entry.ts @@ -1,5 +1,5 @@ import nitroWorker from "#petal/nitro-worker"; -import { handleApi, health } from "./api"; +import { HEALTH_PATH, handleApi, health } from "./api"; import type { Env } from "./env"; // Wrangler looks for Durable Object classes among the exports of the Worker's main module. @@ -11,7 +11,7 @@ export default { ...nitroWorker, fetch(request, env, context) { const url = new URL(request.url); - if (url.pathname === "/health") return health(request, env); + if (url.pathname === HEALTH_PATH) return health(request, env); if (url.pathname.startsWith("/api/")) { return handleApi(request, env, context, url); } diff --git a/src/worker/gateway/gateway.ts b/src/worker/gateway/gateway.ts index c218f58..dfe40c1 100644 --- a/src/worker/gateway/gateway.ts +++ b/src/worker/gateway/gateway.ts @@ -1,4 +1,4 @@ -import { logWorkerEvent, WorkerLogEvent } from "../log"; +import { logWarning } from "../log"; import { buildPaintsQuery, PAINTS_PER_QUERY, parsePaintsRequest } from "./paints"; import { type CachePolicy, @@ -124,9 +124,21 @@ const MEMORY_BUDGET = 16 * 1024 * 1024; /** Larger answers are read from KV every time rather than held in every isolate. */ const MEMORY_ENTRY_MAX = 1024 * 1024; const PAINTS_BODY_MAX = 32 * 1024; -/** A defect fails every request alike, so one log line a minute says as much as all of them. */ -const FAILURE_LOG_INTERVAL_MS = 60_000; -const UPSTREAM_LOG_INTERVAL_MS = 60_000; +const LOG_INTERVAL_MS = 60_000; +const HTTP_TOO_MANY_REQUESTS = 429; + +const UpstreamReason = { + Timeout: "timeout", + Network: "network", + Status: "status", + Response: "response", + Abandoned: "abandoned", + Unknown: "unknown", +} as const; + +type UpstreamReason = (typeof UpstreamReason)[keyof typeof UpstreamReason]; + +const TIMEOUT_ERROR_NAMES = new Set(["TimeoutError", "AbortError"]); /** Most telling first: what a response made of several paints reports. */ const OUTCOME_RANK: readonly CacheStatus[] = ["STALE", "MISS", "EXPIRED", "UPDATING", "HIT"]; @@ -149,37 +161,54 @@ export class GatewayError extends Error { /** Both rate limits of the Worker count over a minute, after which a client may try again. */ export const rateLimited = () => new GatewayError(429, "rate_limited", 60); -const UpstreamReason = { - timeout: "timeout", - network: "network", - status: "status", - response: "response", - abandoned: "abandoned", - unknown: "unknown", -} as const; - -type UpstreamReason = (typeof UpstreamReason)[keyof typeof UpstreamReason]; - -const FETCH_ABORT_NAMES = new Set(["TimeoutError", "AbortError"]); +interface UpstreamFailureOptions { + status?: number; + retryAfterMs?: number; +} class UpstreamFailure extends Error { readonly reason: UpstreamReason; readonly status?: number; readonly rateLimited: boolean; - /** Milliseconds the provider asked us to stay away, if it said so. */ readonly retryAfterMs?: number; - constructor(reason: UpstreamReason, options?: { status?: number; retryAfterMs?: number }) { + constructor(reason: UpstreamReason, options: UpstreamFailureOptions = {}) { super(reason); this.reason = reason; - this.status = options?.status; - this.rateLimited = options?.status === 429; - this.retryAfterMs = options?.retryAfterMs; + this.status = options.status; + this.rateLimited = options.status === HTTP_TOO_MANY_REQUESTS; + this.retryAfterMs = options.retryAfterMs; } } function timedOut(error: unknown): boolean { - return error instanceof Error && FETCH_ABORT_NAMES.has(error.name); + return error instanceof Error && TIMEOUT_ERROR_NAMES.has(error.name); +} + +const JSON_CONTENT_TYPE = /^application\/(?:[\w.-]+\+)?json\b/i; + +function isJsonResponse(response: Response): boolean { + const type = response.headers.get("content-type") ?? ""; + return response.status === 200 && JSON_CONTENT_TYPE.test(type); +} + +function parseJson(text: string): unknown { + try { + return JSON.parse(text) as unknown; + } catch { + return undefined; + } +} + +function parseJsonObject(text: string): unknown { + const json = parseJson(text); + if (typeof json === "object" && json !== null) return json; + throw new UpstreamFailure(UpstreamReason.Response); +} + +interface UpstreamLogState { + loggedAt: number; + suppressed: number; } const freshFor = (entry: Entry, policy: CachePolicy) => @@ -268,9 +297,7 @@ export class DataGateway { /** Keyed by provider for rate limits, which hit every route, and by cache key otherwise. */ readonly #backoff = new Map(); #failureLoggedAt = Number.NEGATIVE_INFINITY; - /** Per provider, so one outage is one line however many routes it takes down. */ - readonly #upstreamLoggedAt = new Map(); - readonly #upstreamSuppressed = new Map(); + readonly #upstreamLogs = new Map(); constructor(options: GatewayOptions) { this.#userAgent = options.userAgent; @@ -323,7 +350,7 @@ export class DataGateway { /** Anything thrown that is not a refusal is a fault of the gateway's own code. */ #defect(route: string, error: unknown): GatewayError { - if (this.#now() - this.#failureLoggedAt >= FAILURE_LOG_INTERVAL_MS) { + if (this.#now() - this.#failureLoggedAt >= LOG_INTERVAL_MS) { this.#failureLoggedAt = this.#now(); // The route id only: the path would name a channel. console.error("data gateway failed", route, error); @@ -372,77 +399,100 @@ export class DataGateway { this.#refresh(key, route, context, () => load(timeoutMs)); const blocked = this.#blockedFor(key, route); - if (!entry) { - if (blocked > 0) { - throw new GatewayError(503, "upstream_backoff", Math.ceil(blocked / 1000)); - } - if (!this.#running(key) && context.admit && !(await context.admit())) { - throw rateLimited(); - } - try { - const loaded = await this.#follow(refresh(TIMEOUT_EMPTY_MS)); - return { entry: loaded, cache: "MISS", layer: "upstream" }; - } catch (error) { - throw this.#unavailable(key, route, error); - } - } + if (!entry) return this.#loadMiss(key, route, context, refresh, blocked); const age = this.#age(entry); const fresh = freshFor(entry, policy); if (age < fresh) return { entry, cache: "HIT", layer }; if (blocked > 0) return { entry, cache: "STALE", layer }; - const usable = age < fresh + staleIfErrorFor(entry, policy); - if (usable && age < fresh + revalidateFor(entry, policy)) { - // Failures are recorded by the refresh itself; the next request after the - // backoff tries again. - context.waitUntil(refresh(TIMEOUT_EMPTY_MS).catch(() => {})); + if (this.#refreshBehind(entry, policy, age, fresh)) { + context.waitUntil(refresh(TIMEOUT_EMPTY_MS).catch(() => undefined)); return { entry, cache: "UPDATING", layer }; } + return this.#refreshOrKeep(entry, layer, route, refresh, age, fresh); + } + + async #loadMiss( + key: string, + route: GatewayRoute, + context: RequestContext, + refresh: (timeoutMs: number) => Promise, + blockedMs: number, + ): Promise { + if (blockedMs > 0) { + throw new GatewayError(503, "upstream_backoff", Math.ceil(blockedMs / 1000)); + } + const startsFetch = !this.#running(key); + if (startsFetch && context.admit && !(await context.admit())) throw rateLimited(); try { - const renewed = await this.#follow( - refresh(usable ? TIMEOUT_STALE_MS : TIMEOUT_EMPTY_MS), - ); + const loaded = await this.#follow(refresh(TIMEOUT_EMPTY_MS)); + return { entry: loaded, cache: "MISS", layer: "upstream" }; + } catch (error) { + throw this.#unavailable(key, route, error); + } + } + + #refreshBehind(entry: Entry, policy: CachePolicy, age: number, fresh: number): boolean { + const withinStaleWindow = age < fresh + staleIfErrorFor(entry, policy); + const withinRevalidateWindow = age < fresh + revalidateFor(entry, policy); + return withinStaleWindow && withinRevalidateWindow; + } + + async #refreshOrKeep( + entry: Entry, + layer: CacheLayer, + route: GatewayRoute, + refresh: (timeoutMs: number) => Promise, + age: number, + fresh: number, + ): Promise { + const withinStaleWindow = age < fresh + staleIfErrorFor(entry, route.policy); + const timeoutMs = withinStaleWindow ? TIMEOUT_STALE_MS : TIMEOUT_EMPTY_MS; + try { + const renewed = await this.#follow(refresh(timeoutMs)); return { entry: renewed, cache: "EXPIRED", layer: "upstream" }; } catch (error) { - this.#noteAbandoned(route, error); + this.#logIfAbandoned(route, error); return { entry, cache: "STALE", layer }; } } #unavailable(key: string, route: GatewayRoute, error: unknown): GatewayError { if (error instanceof GatewayError) return error; - this.#noteAbandoned(route, error); + this.#logIfAbandoned(route, error); const blocked = Math.ceil(this.#blockedFor(key, route) / 1000); return new GatewayError(502, "upstream_unavailable", blocked > 0 ? blocked : undefined); } - #noteAbandoned(route: GatewayRoute, error: unknown): void { - if (error instanceof UpstreamFailure && error.reason === "abandoned") { + #logIfAbandoned(route: GatewayRoute, error: unknown): void { + if (error instanceof UpstreamFailure && error.reason === UpstreamReason.Abandoned) { this.#logUpstream(route, error); } } - #logUpstream(route: GatewayRoute, error: unknown): void { - const provider = route.provider; + #claimUpstreamLog(provider: string): { suppressed: number } | undefined { const now = this.#now(); - const previous = this.#upstreamLoggedAt.get(provider) ?? Number.NEGATIVE_INFINITY; - if (now - previous < UPSTREAM_LOG_INTERVAL_MS) { - const skipped = this.#upstreamSuppressed.get(provider) ?? 0; - this.#upstreamSuppressed.set(provider, skipped + 1); - return; + const state = this.#upstreamLogs.get(provider); + if (state && now - state.loggedAt < LOG_INTERVAL_MS) { + state.suppressed += 1; + return undefined; } - const suppressed = this.#upstreamSuppressed.get(provider) ?? 0; - this.#upstreamSuppressed.set(provider, 0); - this.#upstreamLoggedAt.set(provider, now); + const suppressed = state?.suppressed ?? 0; + this.#upstreamLogs.set(provider, { loggedAt: now, suppressed: 0 }); + return { suppressed }; + } + + #logUpstream(route: GatewayRoute, error: unknown): void { + const claimed = this.#claimUpstreamLog(route.provider); + if (!claimed) return; const failure = error instanceof UpstreamFailure ? error : undefined; - const fields: Record = { - provider, + logWarning("upstream_unavailable", { + provider: route.provider, route: route.id, - reason: failure?.reason ?? UpstreamReason.unknown, - }; - if (failure?.status !== undefined) fields.status = failure.status; - if (suppressed > 0) fields.suppressed = suppressed; - logWorkerEvent("warn", WorkerLogEvent.upstreamUnavailable, fields); + reason: failure?.reason ?? UpstreamReason.Unknown, + status: failure?.status, + suppressed: claimed.suppressed > 0 ? claimed.suppressed : undefined, + }); } /** @@ -454,7 +504,7 @@ export class DataGateway { const limit = after(FOLLOW_MS); try { const result = await Promise.race([refresh, limit.elapsed]); - if (!result) throw new UpstreamFailure(UpstreamReason.abandoned); + if (!result) throw new UpstreamFailure(UpstreamReason.Abandoned); return result; } finally { limit.cancel(); @@ -564,40 +614,34 @@ export class DataGateway { }); } catch (error) { throw new UpstreamFailure( - timedOut(error) ? UpstreamReason.timeout : UpstreamReason.network, + timedOut(error) ? UpstreamReason.Timeout : UpstreamReason.Network, ); } if (response.status === 200 || response.status === 404) return response; await response.body?.cancel(); - throw new UpstreamFailure(UpstreamReason.status, { + throw new UpstreamFailure(UpstreamReason.Status, { status: response.status, retryAfterMs: parseRetryAfter(response.headers.get("retry-after"), this.#now()), }); } async #json(response: Response, maxBytes: number): Promise { - const type = response.headers.get("content-type") ?? ""; - if (response.status !== 200 || !/^application\/(?:[\w.-]+\+)?json\b/i.test(type)) { + const text = await this.#jsonText(response, maxBytes); + return parseJsonObject(text); + } + + async #jsonText(response: Response, maxBytes: number): Promise { + if (!isJsonResponse(response)) { await response.body?.cancel(); - throw new UpstreamFailure(UpstreamReason.response); + throw new UpstreamFailure(UpstreamReason.Response); } - let text: string | undefined; - try { - text = await readCapped(response, maxBytes); - } catch { - throw new UpstreamFailure(UpstreamReason.response); - } - if (text === undefined) throw new UpstreamFailure(UpstreamReason.response); - let json: unknown; try { - json = JSON.parse(text); - } catch { - throw new UpstreamFailure(UpstreamReason.response); - } - if (typeof json !== "object" || json === null) { - throw new UpstreamFailure(UpstreamReason.response); + const text = await readCapped(response, maxBytes); + if (text !== undefined) return text; + } catch (error) { + if (error instanceof UpstreamFailure) throw error; } - return json; + throw new UpstreamFailure(UpstreamReason.Response); } #age(entry: Entry): number { @@ -705,7 +749,7 @@ export class DataGateway { collecting.ids.push(id); collecting.timeoutMs = Math.max(collecting.timeoutMs, timeoutMs); const loaded = (await leaving).get(id); - if (!loaded) throw new UpstreamFailure(UpstreamReason.response); + if (!loaded) throw new UpstreamFailure(UpstreamReason.Response); return loaded; }; const settled = await Promise.allSettled( @@ -758,7 +802,7 @@ export class DataGateway { }; // GraphQL reports failures with status 200, and one bad field voids them all. const paints = json.data?.paints; - if (json.errors || !paints) throw new UpstreamFailure(UpstreamReason.response); + if (json.errors || !paints) throw new UpstreamFailure(UpstreamReason.Response); for (const [index, id] of chunk.entries()) { const paint = paints[`p${index}`]; loaded.set( diff --git a/src/worker/log.ts b/src/worker/log.ts index 06657e7..40ac5e3 100644 --- a/src/worker/log.ts +++ b/src/worker/log.ts @@ -1,19 +1,57 @@ -export const WorkerLogEvent = { - upstreamUnavailable: "upstream_unavailable", - hubUnavailable: "hub_unavailable", +const URL_TEXT = /https?:\/\/\S+/g; +const QUERY_TEXT = /\?[^ \n]*/g; + +type LogValue = string | number; + +export function logSafeCause(error: unknown): string { + const name = error instanceof Error ? error.name : "Error"; + const message = error instanceof Error ? error.message : ""; + const detail = message.replace(URL_TEXT, "").replace(QUERY_TEXT, "").trim(); + return detail.length > 0 ? `${name}: ${detail}` : name; +} + +export const PlatformEvent = { + HubUnavailable: "hub_unavailable", + HubStatusUnavailable: "hub_status_unavailable", + ApiFailed: "api_failed", } as const; -export type WorkerLogEventName = (typeof WorkerLogEvent)[keyof typeof WorkerLogEvent]; +export type PlatformEvent = (typeof PlatformEvent)[keyof typeof PlatformEvent]; + +export class PlatformError extends Error { + readonly event: PlatformEvent; + readonly fields: Readonly>; + + constructor(event: PlatformEvent, cause: unknown, fields: Record = {}) { + super(logSafeCause(cause)); + this.name = "PlatformError"; + this.event = event; + this.fields = fields; + } +} -export function logWorkerEvent( - level: "warn" | "error", - event: WorkerLogEventName, - fields: Readonly>, +function eventRecord( + message: string, + fields: Record, +): Record { + const record: Record = { message }; + for (const [key, value] of Object.entries(fields)) { + if (value !== undefined) record[key] = value; + } + return record; +} + +function logError(message: string, fields: Record = {}): void { + console.error(JSON.stringify(eventRecord(message, fields))); +} + +export function logWarning( + message: string, + fields: Record = {}, ): void { - const body = Object.entries(fields) - .map(([key, value]) => `${key}=${value}`) - .join(" "); - const line = body.length > 0 ? `${event} ${body}` : event; - if (level === "error") console.error(line); - else console.warn(line); + console.warn(JSON.stringify(eventRecord(message, fields))); +} + +export function logPlatformError(error: PlatformError): void { + logError(error.event, { ...error.fields, cause: error.message }); } diff --git a/test/chat/gateway.test.ts b/test/chat/gateway.test.ts index 99c0472..f98fe6c 100644 --- a/test/chat/gateway.test.ts +++ b/test/chat/gateway.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { it, type TestContext } from "node:test"; -import { fetchJson } from "../../src/lib/chat/gateway"; +import { fetchJson, PROVIDER_FETCH_TIMEOUT_MS } from "../../src/lib/chat/gateway"; import { type FetchAnswer, stubFetch, useLocation } from "../helpers"; const PROVIDER = "https://api.betterttv.net/3/cached/users/twitch/1"; @@ -53,7 +53,19 @@ it("gives up on the provider itself after 8 seconds", async (t) => { const fetches = stubFetch({ [PROVIDER]: hangs }); t.after(fetches); const result = fetchJson(PROVIDER); - t.mock.timers.tick(8000); + t.mock.timers.tick(PROVIDER_FETCH_TIMEOUT_MS); + await assert.rejects(result); +}); + +it("gives up while a provider body is still open", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + t.mock.method(console, "warn", () => {}); + t.after(useLocation("https://chat.shiftbloom.studio/chat/forsen?direct=1")); + const open: FetchAnswer = () => new Response(new ReadableStream()); + const fetches = stubFetch({ [PROVIDER]: open }); + t.after(fetches); + const result = fetchJson(PROVIDER); + t.mock.timers.tick(PROVIDER_FETCH_TIMEOUT_MS); await assert.rejects(result); }); @@ -66,7 +78,7 @@ it("asks the provider when the gateway has not answered after 8 seconds", async }); const urls = stub(t, hangs); const result = fetchJson(PROVIDER); - t.mock.timers.tick(7999); + t.mock.timers.tick(PROVIDER_FETCH_TIMEOUT_MS - 1); await new Promise((resolve) => setImmediate(resolve)); assert.deepEqual(urls(), [GATEWAY]); t.mock.timers.tick(1); diff --git a/test/worker/api.test.ts b/test/worker/api.test.ts index 094b56e..ae53db9 100644 --- a/test/worker/api.test.ts +++ b/test/worker/api.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { describe, it } from "node:test"; -import { handleApi, health } from "../../src/worker/api"; +import { HUB_TIMEOUT_MS, handleApi, health } from "../../src/worker/api"; import type { Env } from "../../src/worker/env"; import { shardOf } from "../../src/worker/relay/shard"; @@ -102,10 +102,16 @@ describe("handleApi", () => { const logged = t.mock.method(console, "error", () => {}); const response = call(fakeEnv({}, () => new Promise(() => {})).env, CHANNEL); await new Promise((resolve) => setImmediate(resolve)); - t.mock.timers.tick(5000); + t.mock.timers.tick(HUB_TIMEOUT_MS); assert.deepEqual(await refusal(response), [503, "relay_unavailable"]); - const lines = logged.mock.calls.map((call) => String(call.arguments[0])); - assert.match(lines.join("\n"), /^hub_unavailable shard=\d+ error=Error: timed out$/m); + const line = JSON.parse(String(logged.mock.calls[0]?.arguments[0])) as { + message: string; + shard: number; + cause: string; + }; + assert.equal(line.message, "hub_unavailable"); + assert.equal(typeof line.shard, "number"); + assert.equal(line.cause, "Error: timed out"); }); it("does not log the channel when a hub failure names the request", async (t) => { @@ -118,8 +124,10 @@ describe("handleApi", () => { 503, "relay_unavailable", ]); - const text = logged.mock.calls.map((call) => String(call.arguments[0])).join("\n"); - assert.match(text, /hub_unavailable shard=\d+ error=Error: fetch failed$/m); + const text = String(logged.mock.calls[0]?.arguments[0]); + const line = JSON.parse(text) as { message: string; cause: string }; + assert.equal(line.message, "hub_unavailable"); + assert.equal(line.cause, "Error: fetch failed"); assert.equal(text.includes("somechannel"), false); assert.equal(text.includes("?"), false); }); @@ -162,7 +170,7 @@ describe("handleApi", () => { const response = call(fakeEnv({}, answer).env, CHANNEL); // The hub is asked after the rate limit was, which takes a turn of the event loop. await new Promise((resolve) => setImmediate(resolve)); - t.mock.timers.tick(5000); + t.mock.timers.tick(HUB_TIMEOUT_MS); assert.deepEqual(await refusal(response), [503, "relay_unavailable"]); } const full = Response.json({ error: "hub_full" }, { status: 503 }); diff --git a/test/worker/gateway/gateway.test.ts b/test/worker/gateway/gateway.test.ts index 4129672..a0bdb04 100644 --- a/test/worker/gateway/gateway.test.ts +++ b/test/worker/gateway/gateway.test.ts @@ -5,18 +5,24 @@ import { V4_PAINT_FIELDS as CLIENT_PAINT_FIELDS } from "../../../src/lib/chat/pr import { DataGateway, type EntryMetadata, type KVLike } from "../../../src/worker/gateway/gateway"; import * as paints from "../../../src/worker/gateway/paints"; +const TWITCH_USER_ID = "50985620"; const BTTV_GLOBAL = "https://api.betterttv.net/3/cached/emotes/global"; -const BTTV_USER = "https://api.betterttv.net/3/cached/users/twitch/50985620"; -const FFZ_ROOM_URL = "https://api.frankerfacez.com/v1/room/id/50985620"; +const BTTV_USER = `https://api.betterttv.net/3/cached/users/twitch/${TWITCH_USER_ID}`; +const FFZ_ROOM_URL = `https://api.frankerfacez.com/v1/room/id/${TWITCH_USER_ID}`; const FFZ_GLOBAL_URL = "https://api.frankerfacez.com/v1/set/global"; const GLOBAL = "/bttv/3/cached/emotes/global"; -const USER = "/bttv/3/cached/users/twitch/50985620"; -const FFZ_ROOM = "/ffz/v1/room/id/50985620"; +const USER = `/bttv/3/cached/users/twitch/${TWITCH_USER_ID}`; +const FFZ_ROOM = `/ffz/v1/room/id/${TWITCH_USER_ID}`; const FFZ_GLOBAL = "/ffz/v1/set/global"; const START = Date.UTC(2026, 8, 29, 12); +const LOG_INTERVAL_SECONDS = 60; +const SECONDS_PER_DAY = 86_400; +const STALE_IF_ERROR_DAYS = 7; +const SEVENTV_LEGACY_ID_LENGTH = 24; -const SEVENTV_ULID_PREFIX = "01FQB6K5T0000BDD0YMN2"; -const paintId = (n: number) => `${SEVENTV_ULID_PREFIX}${String(n).padStart(5, "0")}`; +function sevenTvId(index: number): string { + return index.toString(16).padStart(SEVENTV_LEGACY_ID_LENGTH, "0"); +} const paint = (id: string) => ({ id, name: "Paint", data: { layers: [], shadows: [] } }); /** Answers a paints query the way 7TV does for paints it knows. */ @@ -176,10 +182,12 @@ describe("provider failures", () => { await settle(); assert.equal(stored.size, 0); assert.equal(calls.length, 1); - assert.equal( - warned.mock.calls[0]?.arguments[0], - "upstream_unavailable provider=ffz route=ffz.global reason=response", - ); + assert.deepEqual(JSON.parse(String(warned.mock.calls[0]?.arguments[0])), { + message: "upstream_unavailable", + provider: "ffz", + route: "ffz.global", + reason: "response", + }); }); it("serves a copy that has outlived its window when the provider fails", async () => { @@ -187,7 +195,7 @@ describe("provider failures", () => { json(FFZ_ROOM_URL, { room: {}, sets: {} }); await get(FFZ_ROOM); await settle(); - advance(8 * 86400); + advance((STALE_IF_ERROR_DAYS + 1) * SECONDS_PER_DAY); answers.set(FFZ_ROOM_URL, () => new Response(null, { status: 502 })); const stale = await get(FFZ_ROOM); assert.equal(stale.status, 200); @@ -217,11 +225,31 @@ describe("provider failures", () => { Promise.reject(Object.assign(new Error("aborted"), { name: "TimeoutError" })), ); assert.equal((await get(FFZ_ROOM)).status, 502); - assert.equal( - warned.mock.calls[0]?.arguments[0], - "upstream_unavailable provider=ffz route=ffz.room reason=timeout", - ); - assert.equal(JSON.stringify(warned.mock.calls).includes("50985620"), false); + assert.deepEqual(JSON.parse(String(warned.mock.calls[0]?.arguments[0])), { + message: "upstream_unavailable", + provider: "ffz", + route: "ffz.room", + reason: "timeout", + }); + assert.equal(JSON.stringify(warned.mock.calls).includes(TWITCH_USER_ID), false); + }); + + it("logs one upstream failure per provider per minute", async (t) => { + const warned = t.mock.method(console, "warn", () => {}); + const { advance, answers, get } = setup(); + const unavailable = () => new Response(null, { status: 503 }); + answers.set(FFZ_ROOM_URL, unavailable); + answers.set(FFZ_GLOBAL_URL, unavailable); + assert.equal((await get(FFZ_ROOM)).status, 502); + assert.equal((await get(FFZ_GLOBAL)).status, 502); + assert.equal(warned.mock.calls.length, 1); + advance(LOG_INTERVAL_SECONDS); + assert.equal((await get(FFZ_GLOBAL)).status, 502); + assert.equal(warned.mock.calls.length, 2); + const again = JSON.parse(String(warned.mock.calls[1]?.arguments[0])) as { + suppressed?: number; + }; + assert.equal(again.suppressed, 1); }); it("gives up on a provider that never answers", async (t) => { @@ -251,7 +279,7 @@ describe("refusals", () => { const { calls, get, post } = setup(); const query = (variables: object, text = "x") => post(JSON.stringify({ query: text, variables })); - const many = Array.from({ length: paints.PAINTS_PER_REQUEST + 1 }, (_, i) => paintId(i)); + const many = Array.from({ length: paints.PAINTS_PER_REQUEST + 1 }, (_, i) => sevenTvId(i)); const cases: [Promise, number][] = [ [get("/bttv/3/cached/users/twitch/1%2F.."), 400], [get("//7tv.io/v3/emote-sets/global"), 400], @@ -284,7 +312,7 @@ describe("refusals", () => { describe("paints", () => { it("splits what it asks into queries 7TV accepts, and sends only its own", async () => { const { calls, post } = setup(); - const ids = Array.from({ length: paints.PAINTS_PER_REQUEST }, (_, i) => paintId(i)); + const ids = Array.from({ length: paints.PAINTS_PER_REQUEST }, (_, i) => sevenTvId(i)); const body = { ...paints.buildPaintsQuery(ids), query: "{ users { id } }" }; const response = await post(JSON.stringify(body)); const { data } = (await response.json()) as { data: { paints: object } };