diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8c4fc73..d44c12f 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 | @@ -532,6 +536,7 @@ The three windows follow each other. While an answer is fresh it is served witho 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 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 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/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..d26042c 100644 --- a/src/lib/chat/gateway.ts +++ b/src/lib/chat/gateway.ts @@ -14,8 +14,7 @@ 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. */ -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"; @@ -33,30 +32,72 @@ 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; } -async function fetchFromGateway(url: string, request?: JsonRequest): Promise { - const deadline = new AbortController(); - const timer = setTimeout(() => deadline.abort(), GATEWAY_TIMEOUT_MS); +interface Deadline { + signal: AbortSignal; + elapsed: Promise; + stop: () => void; +} + +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`); + } +} + +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 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); + const response = await Promise.race([ + preventUnhandledRejection(fetch(url, { ...request, signal: deadline.signal })), + deadline.elapsed, + ]); + return await Promise.race([preventUnhandledRejection(read(response)), deadline.elapsed]); } finally { - clearTimeout(timer); + deadline.stop(); } } +async function fetchFromGateway(url: string, request?: JsonRequest): Promise { + return readWithinTimeout(url, request, async (response) => { + requireGatewayResponse(response, url); + return readJson(response, url); + }); +} + /** * Fetches provider data as JSON, undefined when the provider answers 404. The gateway is asked * first; whenever it fails to give the provider's answer, the provider is asked directly, which @@ -71,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 fetch(url, request), url); + return readWithinTimeout(url, request, (response) => readJson(response, url)); } diff --git a/src/worker/api.ts b/src/worker/api.ts index 2a53aaa..9d93a19 100644 --- a/src/worker/api.ts +++ b/src/worker/api.ts @@ -3,6 +3,7 @@ import { record, recordGateway } from "./analytics"; import type { Env } from "./env"; import { handleData } from "./gateway/index"; import type { HubStatus } from "./hub"; +import { logPlatformError, PlatformError, PlatformEvent } from "./log"; import { withinLimit } from "./rate-limit"; import { normaliseChannel, shardOf } from "./relay/shard"; @@ -18,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/"; @@ -163,7 +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) { - console.error("hub did not take the connection", shard, error); + logPlatformError(new PlatformError(PlatformEvent.HubUnavailable, error, { shard })); return relayUnavailable(env, shard); } if (response.status === 101) { @@ -198,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 }; } } @@ -223,6 +224,26 @@ 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 (!HEALTH_METHODS.has(request.method)) { + return refuse("method_not_allowed", 405, { allow: HEALTH_ALLOW }); + } + 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; +} + /** * Handles every request under `/api/`; the rest of the site never reaches this module. * @@ -251,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 5f62198..85d8235 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 { 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,6 +11,7 @@ export default { ...nitroWorker, fetch(request, env, context) { const url = new URL(request.url); + 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 a4e93d7..dfe40c1 100644 --- a/src/worker/gateway/gateway.ts +++ b/src/worker/gateway/gateway.ts @@ -1,3 +1,4 @@ +import { logWarning } from "../log"; import { buildPaintsQuery, PAINTS_PER_QUERY, parsePaintsRequest } from "./paints"; import { type CachePolicy, @@ -123,8 +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 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"]; @@ -147,18 +161,56 @@ 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); +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: string, rateLimited = false, retryAfterMs?: number) { + constructor(reason: UpstreamReason, options: UpstreamFailureOptions = {}) { super(reason); - this.rateLimited = rateLimited; - this.retryAfterMs = retryAfterMs; + this.reason = reason; + 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 && 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) => entry.status === 404 ? (policy.negativeFresh ?? 0) : policy.fresh; @@ -245,6 +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; + readonly #upstreamLogs = new Map(); constructor(options: GatewayOptions) { this.#userAgent = options.userAgent; @@ -297,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); @@ -346,50 +399,102 @@ 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)); - } - // Waiting for an answer that is already on its way costs nothing. - 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 }; - 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. - context.waitUntil(refresh(TIMEOUT_EMPTY_MS).catch(() => {})); + if (blocked > 0) return { entry, cache: "STALE", layer }; + 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), - ); - return { entry: renewed, cache: "EXPIRED", layer: "upstream" }; + const loaded = await this.#follow(refresh(TIMEOUT_EMPTY_MS)); + return { entry: loaded, cache: "MISS", layer: "upstream" }; } catch (error) { - if (usable) return { entry, cache: "STALE", layer }; 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.#logIfAbandoned(route, error); + return { entry, cache: "STALE", layer }; + } + } + #unavailable(key: string, route: GatewayRoute, error: unknown): GatewayError { if (error instanceof GatewayError) return error; + this.#logIfAbandoned(route, error); const blocked = Math.ceil(this.#blockedFor(key, route) / 1000); return new GatewayError(502, "upstream_unavailable", blocked > 0 ? blocked : undefined); } + #logIfAbandoned(route: GatewayRoute, error: unknown): void { + if (error instanceof UpstreamFailure && error.reason === UpstreamReason.Abandoned) { + this.#logUpstream(route, error); + } + } + + #claimUpstreamLog(provider: string): { suppressed: number } | undefined { + const now = this.#now(); + const state = this.#upstreamLogs.get(provider); + if (state && now - state.loggedAt < LOG_INTERVAL_MS) { + state.suppressed += 1; + return undefined; + } + 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; + logWarning("upstream_unavailable", { + provider: route.provider, + route: route.id, + reason: failure?.reason ?? UpstreamReason.Unknown, + status: failure?.status, + suppressed: claimed.suppressed > 0 ? claimed.suppressed : undefined, + }); + } + /** * 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 +504,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(UpstreamReason.Abandoned); return result; } finally { limit.cancel(); @@ -425,6 +530,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,40 +613,35 @@ export class DataGateway { signal: AbortSignal.timeout(timeoutMs), }); } catch (error) { - throw new UpstreamFailure(`fetch failed: ${String(error)}`); + 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( - `answered ${response.status}`, - response.status === 429, - parseRetryAfter(response.headers.get("retry-after"), this.#now()), - ); + 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(`answered ${response.status} ${type}`); + throw new UpstreamFailure(UpstreamReason.Response); } - let text: string | undefined; try { - text = await readCapped(response, maxBytes); + const text = await readCapped(response, maxBytes); + if (text !== undefined) return text; } catch (error) { - throw new UpstreamFailure(`body failed: ${String(error)}`); - } - if (text === undefined) throw new UpstreamFailure("body too large"); - let json: unknown; - try { - json = JSON.parse(text); - } catch { - throw new UpstreamFailure("body is not JSON"); - } - if (typeof json !== "object" || json === null) { - throw new UpstreamFailure("body is not a JSON object or array"); + if (error instanceof UpstreamFailure) throw error; } - return json; + throw new UpstreamFailure(UpstreamReason.Response); } #age(entry: Entry): number { @@ -648,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("paint missing from the answer"); + if (!loaded) throw new UpstreamFailure(UpstreamReason.Response); return loaded; }; const settled = await Promise.allSettled( @@ -701,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("GraphQL error"); + 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..40ac5e3 --- /dev/null +++ b/src/worker/log.ts @@ -0,0 +1,57 @@ +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 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; + } +} + +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 { + 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 fe596c8..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"; @@ -42,6 +42,33 @@ 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(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); +}); + 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. */ @@ -51,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 92bb0d1..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 } 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"; @@ -96,6 +96,68 @@ 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(HUB_TIMEOUT_MS); + assert.deepEqual(await refusal(response), [503, "relay_unavailable"]); + 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) => { + 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 = 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); + }); + + 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"] }); @@ -108,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 80535c6..a0bdb04 100644 --- a/test/worker/gateway/gateway.test.ts +++ b/test/worker/gateway/gateway.test.ts @@ -5,13 +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 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 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 paintId = (n: number) => `01FQB6K5T0000BDD0YMN2${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. */ @@ -145,6 +156,102 @@ 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); + 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 () => { + const { advance, answers, calls, json, settle, get } = setup(); + json(FFZ_ROOM_URL, { room: {}, sets: {} }); + await get(FFZ_ROOM); + await settle(); + 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); + assert.equal(cacheOf(stale), "STALE; layer=memory"); + assert.equal(calls.length, 2); + 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); + 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) => { t.mock.timers.enable({ apis: ["setTimeout"] }); const { answers, get } = setup(); @@ -172,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], @@ -205,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 } };