From c362cb183efd2e5c99fe168b29c2e5a2e70f69af Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Sat, 7 Nov 2026 23:38:28 -0800 Subject: [PATCH] relay-edge: keep valid desktops through JWKS outages, deep object probe, cohort-scaled alerts (#39772) * relay-edge: keep valid desktops on JWKS outages, deep object probe, cohort-scaled alerts - auth: tell 'token invalid' from 'keys unavailable'. An auth-refresh that cannot get keys keeps the control and its session unchanged (metric auth_refresh:keys_unavailable); the exp+grace close still applies. A control upgrade without keys gets 503 instead of 401 (shipped desktops treat both as a retried connect failure). The last good JWKS is kept in the colo's Cache API (24 h max age) as a fallback for cold isolates; a stored set lacking the token's key never reads as invalid. - admin export ignores closing controls, like drain, so a just-drained host is not busy. - GET /__health/object: canary HostRelay reads its SQLite (SELECT 1), no writes; per-source limited; 503 on failure or after 3 s. The alerts Worker probes it (object-probe, > 2 s alerts). - alerts: MIN_VOLUME stays configurable, reconnect-surge rule (control hellos >= max(3x previous, RECONNECT_SURGE_MIN)), optional cloudflare-errors rule from the GraphQL Analytics API (CF_GRAPHQL_TOKEN). Rules moved to alerts-rules.ts. - metrics: heartbeat events carry their 1-in-10 sampling factor as their last double. * relay-edge: persisted JWKS trusted for 30 min, as the cells' stale window * relay-edge alerts: opt-in reuse of the Analytics Engine read token for the GraphQL error rule --- .../apps/relay-edge/src/alerts-cloudflare.ts | 78 +++++++++ cloud/apps/relay-edge/src/alerts-probe.ts | 28 ++++ cloud/apps/relay-edge/src/alerts-rules.ts | 150 +++++++++++++++++ cloud/apps/relay-edge/src/alerts.ts | 153 +++++------------- cloud/apps/relay-edge/src/env.ts | 3 +- cloud/apps/relay-edge/src/health.ts | 37 +++++ .../apps/relay-edge/src/host-relay-control.ts | 10 +- cloud/apps/relay-edge/src/host-relay.ts | 18 ++- cloud/apps/relay-edge/src/index.ts | 19 ++- cloud/apps/relay-edge/src/jwks-persist.ts | 99 ++++++++++++ cloud/apps/relay-edge/src/jwt.ts | 31 ++-- cloud/apps/relay-edge/src/metrics.ts | 19 ++- cloud/apps/relay-edge/src/preauth.ts | 5 +- .../relay-edge/test/alerts-cohort.test.ts | 147 +++++++++++++++++ .../test/alerts-small-volume.test.ts | 7 +- cloud/apps/relay-edge/test/jwt-keys.test.ts | 124 ++++++++++++++ .../relay-edge/test/mvp-object-hardening.ts | 40 ++++- cloud/apps/relay-edge/test/mvp-object.ts | 3 +- cloud/apps/relay-edge/wrangler.alerts.jsonc | 8 +- 19 files changed, 830 insertions(+), 149 deletions(-) create mode 100644 cloud/apps/relay-edge/src/alerts-cloudflare.ts create mode 100644 cloud/apps/relay-edge/src/alerts-rules.ts create mode 100644 cloud/apps/relay-edge/src/health.ts create mode 100644 cloud/apps/relay-edge/src/jwks-persist.ts create mode 100644 cloud/apps/relay-edge/test/alerts-cohort.test.ts create mode 100644 cloud/apps/relay-edge/test/jwt-keys.test.ts diff --git a/cloud/apps/relay-edge/src/alerts-cloudflare.ts b/cloud/apps/relay-edge/src/alerts-cloudflare.ts new file mode 100644 index 00000000000..f7db2fee1df --- /dev/null +++ b/cloud/apps/relay-edge/src/alerts-cloudflare.ts @@ -0,0 +1,78 @@ +// Worker and Durable Object errors from Cloudflare's GraphQL Analytics API, for the +// cloudflare-errors rule (alerts-rules.ts). Catches what the edge's own metrics cannot: an isolate +// that throws, exceeds its limits or hits an internal error never writes its data point. +// One small query covers both windows. Statuses other than success and a client going away count +// as errors (websocket peers disconnecting is normal traffic). +import type { AlertsEnv, Counts } from './alerts-rules' + +const GRAPHQL_URL = 'https://api.cloudflare.com/client/v4/graphql' +const OK_STATUSES = new Set(['success', 'clientDisconnected', 'responseStreamDisconnected']) +export const DEFAULT_SCRIPT_NAME = 'orca-relay-edge' + +type Group = { sum?: { requests?: number }; dimensions?: { status?: string; namespaceId?: string } } + +const workers = (alias: string, range: string) => + `${alias}:workersInvocationsAdaptive(limit:50,filter:{scriptName:$script,${range}}){sum{requests} dimensions{status}}` +const objects = (alias: string, range: string) => + `${alias}:durableObjectsInvocationsAdaptiveGroups(limit:200,filter:{${range}}){sum{requests} dimensions{namespaceId status}}` + +export function cloudflareQuery(withObjects: boolean): string { + const now = 'datetime_geq:$s0,datetime_lt:$e0' + const before = 'datetime_geq:$s1,datetime_lt:$e1' + const parts = [workers('w0', now), workers('w1', before), ...(withObjects ? [objects('o0', now), objects('o1', before)] : [])] + return `query($a:String!,$script:String!,$s0:Time!,$e0:Time!,$s1:Time!,$e1:Time!){viewer{accounts(filter:{accountTag:$a}){${parts.join(' ')}}}}` +} + +function tally(groups: Group[] | undefined, namespaces?: Set): Counts { + let invocations = 0 + let errors = 0 + for (const group of groups ?? []) { + if (namespaces && !namespaces.has(group.dimensions?.namespaceId ?? '')) { continue } + const requests = group.sum?.requests ?? 0 + invocations += requests + if (!OK_STATUSES.has(group.dimensions?.status ?? '')) { errors += requests } + } + return { cf_invocations: invocations, cf_errors: errors } +} + +const add = (a: Counts, b: Counts): Counts => ({ + cf_invocations: a.cf_invocations + b.cf_invocations, + cf_errors: a.cf_errors + b.cf_errors +}) + +// [current window, previous window], with the same 2-min ingestion lag as the Analytics Engine +// queries; null when the rule is not configured. Throws when the API cannot be read. +export async function cloudflareErrors(env: AlertsEnv, windowMin: number, lagMin: number, nowMs = Date.now()): Promise<[Counts, Counts] | null> { + // The Analytics Engine read token also carries Account Analytics: Read when CF_GRAPHQL_USE_AE_TOKEN=1. + const token = env.CF_GRAPHQL_TOKEN ?? (env.CF_GRAPHQL_USE_AE_TOKEN === '1' ? env.AE_READ_TOKEN : undefined) + if (!token) { return null } + const namespaceIds = (env.CF_DO_NAMESPACE_IDS ?? '').split(',').map((id) => id.trim()).filter(Boolean) + const namespaces = namespaceIds.length > 0 ? new Set(namespaceIds) : undefined + const at = (minutes: number) => new Date(nowMs - minutes * 60_000).toISOString() + const response = await fetch(GRAPHQL_URL, { + method: 'POST', + headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' }, + body: JSON.stringify({ + query: cloudflareQuery(namespaces !== undefined), + variables: { + a: env.ACCOUNT_ID, + script: env.CF_SCRIPT_NAME || DEFAULT_SCRIPT_NAME, + s0: at(windowMin + lagMin), + e0: at(lagMin), + s1: at(2 * windowMin + lagMin), + e1: at(windowMin + lagMin) + } + }) + }) + if (!response.ok) { throw new Error(`cloudflare analytics query failed: ${response.status}`) } + const body = (await response.json()) as { + data?: { viewer?: { accounts?: Record[] } } + errors?: unknown[] | null + } + const account = body.data?.viewer?.accounts?.[0] + if ((body.errors && body.errors.length > 0) || !account) { throw new Error('cloudflare analytics query returned errors') } + return [ + add(tally(account.w0), namespaces ? tally(account.o0, namespaces) : tally([])), + add(tally(account.w1), namespaces ? tally(account.o1, namespaces) : tally([])) + ] +} diff --git a/cloud/apps/relay-edge/src/alerts-probe.ts b/cloud/apps/relay-edge/src/alerts-probe.ts index e3fdf5dddd9..9988277fdf1 100644 --- a/cloud/apps/relay-edge/src/alerts-probe.ts +++ b/cloud/apps/relay-edge/src/alerts-probe.ts @@ -2,6 +2,8 @@ // refused 401 by the edge's router, which proves the Worker is up, routing and verifying tokens // without touching any object or database. Stateless, like the rules: one failed run alerts (with // the status), and every later failing run alerts again until the edge answers. +// probeObject goes one step deeper: GET /__health/object makes the edge call a canary HostRelay +// object that reads its own storage, so a broken object runtime (or storage) alerts too. export const PROBE_TIMEOUT_MS = 5_000 @@ -40,3 +42,29 @@ export async function probeEdge(baseUrl: string, now: () => number = Date.now): clearTimeout(timer) } } + +// Slower than this (or not {ok:true} with 200) alerts as 'object-probe'. +export const OBJECT_PROBE_SLOW_MS = 2_000 + +// null when the canary object answered 200 {ok:true} within OBJECT_PROBE_SLOW_MS. +export async function probeObject(baseUrl: string, now: () => number = Date.now): Promise { + const url = `${baseUrl.replace(/\/+$/, '')}/__health/object` + const started = now() + const abort = new AbortController() + const timer = setTimeout(() => abort.abort(new DOMException('probe timed out', 'TimeoutError')), PROBE_TIMEOUT_MS) + try { + const response = await fetch(url, { signal: abort.signal }) + const body = (await response.json().catch(() => null)) as { ok?: unknown } | null + const elapsed = now() - started + if (response.status !== 200 || body?.ok !== true) { return `object health check got ${response.status}, expected 200` } + if (elapsed > OBJECT_PROBE_SLOW_MS) { return `object health check took ${elapsed} ms` } + return null + } catch (error) { + const name = (error as Error).name + return name === 'TimeoutError' || name === 'AbortError' + ? `object health check timed out after ${PROBE_TIMEOUT_MS} ms` + : `object health check failed: ${(error as Error).message}` + } finally { + clearTimeout(timer) + } +} diff --git a/cloud/apps/relay-edge/src/alerts-rules.ts b/cloud/apps/relay-edge/src/alerts-rules.ts new file mode 100644 index 00000000000..4b0da8938a4 --- /dev/null +++ b/cloud/apps/relay-edge/src/alerts-rules.ts @@ -0,0 +1,150 @@ +// Edge alert rules over per-window counts (alerts.ts reads them). Every message carries counts and +// shares only: never a user, host, device or connection id. + +export type AlertsEnv = { + ACCOUNT_ID: string + DATASET: string + ENVIRONMENT: string + AE_READ_TOKEN: string + ALERT_WEBHOOK_URL: string + // Below this many hellos (phone_open + control_active) per window, ratio rules stay quiet. + // Default 200; scale it with the cohort on the edge. + MIN_VOLUME?: string + // The edge's public origin; each run probes it (alerts-probe.ts). Absent: no probe. + EDGE_PROBE_URL?: string + // reconnect-surge fires at max(3x the previous window, this) control hellos. Default 50. + RECONNECT_SURGE_MIN?: string + // cloudflare-errors (alerts-cloudflare.ts): skipped while CF_GRAPHQL_TOKEN (secret, Account + // Analytics: Read) is unset. Script defaults to 'orca-relay-edge'; CF_DO_NAMESPACE_IDS + // (comma-separated) adds the edge's Durable Object namespaces. Fires above CF_ERRORS_MIN (20) + // errors and CF_ERRORS_PCT (1) percent of invocations per window. + CF_GRAPHQL_TOKEN?: string + // '1': use AE_READ_TOKEN for the GraphQL query when CF_GRAPHQL_TOKEN is unset. + CF_GRAPHQL_USE_AE_TOKEN?: string + CF_SCRIPT_NAME?: string + CF_DO_NAMESPACE_IDS?: string + CF_ERRORS_MIN?: string + CF_ERRORS_PCT?: string +} + +export type Thresholds = { minVolume: number; surgeMin: number; cfErrorsMin: number; cfErrorsPct: number } + +function positive(value: string | undefined, fallback: number): number { + const parsed = Number(value) + return value !== undefined && value !== '' && Number.isFinite(parsed) && parsed >= 0 ? parsed : fallback +} + +export function thresholds(env: AlertsEnv): Thresholds { + return { + minVolume: positive(env.MIN_VOLUME, 200), + surgeMin: positive(env.RECONNECT_SURGE_MIN, 50), + cfErrorsMin: positive(env.CF_ERRORS_MIN, 20), + cfErrorsPct: positive(env.CF_ERRORS_PCT, 1) + } +} + +export const WINDOW_MIN = 15 +// Per window, whatever the volume. +const ADMIN_FAILURES_MIN = 3 +const LOW_VOLUME_4401_MIN = 5 +// Admin labels that are an answer the director expects (':'); anything else (a status +// such as assign:401, or stale / conflict / error) is a failure. +export const ADMIN_SUCCESS_LABELS = [ + 'assign:applied', + 'assign:unchanged', + 'export:exported', + 'export:busy', + 'erase:erased', + 'erase:busy', + 'erase:mismatch', + 'drain:drained', + 'drain:idle' +] + +export type Counts = Record +// `prior` is the previous window; rules that compare against it never count as breaching "before". +export type Rule = { id: string; describe: (now: Counts) => string; breach: (c: Counts, t: Thresholds, prior?: Counts) => boolean } + +// Counts per window, weighted by Analytics Engine sampling. +export const SQL = (dataset: string, fromMin: number, toMin: number) => ` +SELECT + sumIf(_sample_interval, blob1 = 'phone_open') AS phone_open, + sumIf(_sample_interval, blob1 = 'control_active') AS control_active, + sumIf(_sample_interval, blob1 = 'control_active' AND blob2 IN ('new', 'rebind')) AS reconnects, + sumIf(_sample_interval, blob1 = 'refuse' AND endsWith(blob2, ':4401')) AS refuse_4401, + sumIf(_sample_interval, blob1 = 'refuse' AND (endsWith(blob2, ':4429') OR endsWith(blob2, ':4503'))) AS refuse_busy, + sumIf(_sample_interval, blob1 = 'control_close') AS control_close, + sumIf(_sample_interval, blob1 = 'db_op') AS db_ops, + sumIf(_sample_interval, blob1 = 'db_op' AND blob2 != 'ok') AS db_failures, + sumIf(_sample_interval, blob1 = 'admin' AND blob2 NOT IN (${ADMIN_SUCCESS_LABELS.map((l) => `'${l}'`).join(', ')})) AS admin_failures, + sum(_sample_interval) AS events +FROM ${dataset} +WHERE timestamp > NOW() - INTERVAL '${fromMin}' MINUTE AND timestamp <= NOW() - INTERVAL '${toMin}' MINUTE +FORMAT JSON` + +// Analytics Engine has no conditional quantile, so database latency is its own query. +export const DB_P95_SQL = (dataset: string, fromMin: number, toMin: number) => ` +SELECT quantileExactWeighted(0.95)(double1, _sample_interval) AS db_p95_ms +FROM ${dataset} +WHERE blob1 = 'db_op' AND timestamp > NOW() - INTERVAL '${fromMin}' MINUTE AND timestamp <= NOW() - INTERVAL '${toMin}' MINUTE +FORMAT JSON` + +const pct = (a: number, b: number) => (b > 0 ? (100 * a) / b : 0) +const hellos = (c: Counts) => c.phone_open + c.control_active + +export const RULES: Rule[] = [ + { + // 4401 disables phone credentials; a spike means the edge is wrongly refusing good clients. + id: 'auth-refusals', + breach: (c, t) => hellos(c) >= t.minVolume && pct(c.refuse_4401, hellos(c)) > 0.5, + describe: (c) => `4401 refusals ${c.refuse_4401} = ${pct(c.refuse_4401, hellos(c)).toFixed(2)}% of hellos` + }, + { + id: 'busy-refusals', + breach: (c, t) => hellos(c) >= t.minVolume && pct(c.refuse_busy, hellos(c)) > 5, + describe: (c) => `4429/4503 refusals ${c.refuse_busy} = ${pct(c.refuse_busy, hellos(c)).toFixed(1)}% of hellos` + }, + { + id: 'database', + breach: (c) => c.db_ops >= 20 && (pct(c.db_failures, c.db_ops) > 1 || c.db_p95_ms > 300), + describe: (c) => `database failures ${pct(c.db_failures, c.db_ops).toFixed(1)}% of ${c.db_ops} ops, p95 ${Math.round(c.db_p95_ms)} ms` + }, + { + // Mass drop: control closes jump to 3x the previous window. + id: 'mass-drop', + breach: (c, t, prior) => prior !== undefined && c.control_close >= t.minVolume && c.control_close > 3 * Math.max(1, prior.control_close), + describe: (c) => `control closes ${c.control_close} in ${WINDOW_MIN} min, over 3x the previous window` + }, + { + // Mass drop with no close events (object reset or eviction drops sockets silently): the + // desktops come back, so control hellos jump. Floor keeps a small cohort's noise quiet. + id: 'reconnect-surge', + breach: (c, t, prior) => prior !== undefined && (c.reconnects ?? 0) >= Math.max(3 * (prior.reconnects ?? 0), t.surgeMin), + describe: (c) => `control hellos ${c.reconnects ?? 0} in ${WINDOW_MIN} min, at least 3x the previous window` + }, + { + // The director's calls failing (bad signature, stale or conflicting pushes, errors). + id: 'admin-failures', + breach: (c) => c.admin_failures >= ADMIN_FAILURES_MIN, + describe: (c) => `${c.admin_failures} failed admin calls in ${WINDOW_MIN} min` + }, + { + // Below the share rules' volume a handful of 4401s is still worth a look. + id: 'auth-refusals-low-volume', + breach: (c, t) => hellos(c) < t.minVolume && c.refuse_4401 >= LOW_VOLUME_4401_MIN, + describe: (c) => `4401 refusals ${c.refuse_4401} in ${WINDOW_MIN} min at low volume (${hellos(c)} hellos)` + }, + { + // Dead man: traffic stopped (the edge or its metrics went silent) after a window that had some. + id: 'silent', + breach: (c, t, prior) => prior !== undefined && prior.events >= t.minVolume && c.events === 0, + describe: () => `no edge events in ${WINDOW_MIN} min` + }, + { + // Worker / Durable Object invocations that ended in an error (Cloudflare GraphQL; absent when + // CF_GRAPHQL_TOKEN is unset, and then never breaching). + id: 'cloudflare-errors', + breach: (c, t) => (c.cf_errors ?? 0) > t.cfErrorsMin && pct(c.cf_errors ?? 0, c.cf_invocations ?? 0) > t.cfErrorsPct, + describe: (c) => `Worker/Durable Object errors ${c.cf_errors} = ${pct(c.cf_errors ?? 0, c.cf_invocations ?? 0).toFixed(1)}% of ${c.cf_invocations} invocations in ${WINDOW_MIN} min` + } +] diff --git a/cloud/apps/relay-edge/src/alerts.ts b/cloud/apps/relay-edge/src/alerts.ts index 6016902517f..6debe292713 100644 --- a/cloud/apps/relay-edge/src/alerts.ts +++ b/cloud/apps/relay-edge/src/alerts.ts @@ -5,110 +5,18 @@ // Secrets (set from stdin, never in config or logs): AE_READ_TOKEN (Account Analytics: Read), // ALERT_WEBHOOK_URL (Slack-compatible incoming webhook: POST {text}). // Share rules need MIN_VOLUME hellos per window; the absolute-count rules (admin failures, 4401s -// below that volume) and the synthetic probe (EDGE_PROBE_URL, alerts-probe.ts) also cover a quiet edge. -import { probeEdge } from './alerts-probe' +// below that volume), the reconnect surge floor and the synthetic probes (EDGE_PROBE_URL, +// alerts-probe.ts: the router's 401 and a canary object's storage read) also cover a quiet edge. +// Optional: Worker/Durable Object errors from Cloudflare's GraphQL API (alerts-cloudflare.ts). +// Rules, thresholds and their env vars: alerts-rules.ts. +import { probeEdge, probeObject } from './alerts-probe' +import { cloudflareErrors } from './alerts-cloudflare' +import { DB_P95_SQL, RULES, SQL, WINDOW_MIN, thresholds, type AlertsEnv, type Counts } from './alerts-rules' -export type AlertsEnv = { - ACCOUNT_ID: string - DATASET: string - ENVIRONMENT: string - AE_READ_TOKEN: string - ALERT_WEBHOOK_URL: string - // Below this many hellos (phone_open + control_active) per window, ratio rules stay quiet. - MIN_VOLUME?: string - // The edge's public origin; each run probes it (alerts-probe.ts). Absent: no probe. - EDGE_PROBE_URL?: string -} +export { ADMIN_SUCCESS_LABELS, RULES, type AlertsEnv } from './alerts-rules' -const WINDOW_MIN = 15 -// Per window, whatever the volume. -const ADMIN_FAILURES_MIN = 3 -const LOW_VOLUME_4401_MIN = 5 -// Admin labels that are an answer the director expects (':'); anything else (a status -// such as assign:401, or stale / conflict / error) is a failure. -export const ADMIN_SUCCESS_LABELS = [ - 'assign:applied', - 'assign:unchanged', - 'export:exported', - 'export:busy', - 'erase:erased', - 'erase:busy', - 'erase:mismatch', - 'drain:drained', - 'drain:idle' -] - -type Counts = Record -// `prior` is the previous window; rules that compare against it never count as breaching "before". -type Rule = { id: string; describe: (now: Counts) => string; breach: (c: Counts, minVolume: number, prior?: Counts) => boolean } - -// Counts per window, weighted by Analytics Engine sampling. -const SQL = (dataset: string, fromMin: number, toMin: number) => ` -SELECT - sumIf(_sample_interval, blob1 = 'phone_open') AS phone_open, - sumIf(_sample_interval, blob1 = 'control_active') AS control_active, - sumIf(_sample_interval, blob1 = 'refuse' AND endsWith(blob2, ':4401')) AS refuse_4401, - sumIf(_sample_interval, blob1 = 'refuse' AND (endsWith(blob2, ':4429') OR endsWith(blob2, ':4503'))) AS refuse_busy, - sumIf(_sample_interval, blob1 = 'control_close') AS control_close, - sumIf(_sample_interval, blob1 = 'db_op') AS db_ops, - sumIf(_sample_interval, blob1 = 'db_op' AND blob2 != 'ok') AS db_failures, - sumIf(_sample_interval, blob1 = 'admin' AND blob2 NOT IN (${ADMIN_SUCCESS_LABELS.map((l) => `'${l}'`).join(', ')})) AS admin_failures, - sum(_sample_interval) AS events -FROM ${dataset} -WHERE timestamp > NOW() - INTERVAL '${fromMin}' MINUTE AND timestamp <= NOW() - INTERVAL '${toMin}' MINUTE -FORMAT JSON` - -// Analytics Engine has no conditional quantile, so database latency is its own query. -const DB_P95_SQL = (dataset: string, fromMin: number, toMin: number) => ` -SELECT quantileExactWeighted(0.95)(double1, _sample_interval) AS db_p95_ms -FROM ${dataset} -WHERE blob1 = 'db_op' AND timestamp > NOW() - INTERVAL '${fromMin}' MINUTE AND timestamp <= NOW() - INTERVAL '${toMin}' MINUTE -FORMAT JSON` - -const pct = (a: number, b: number) => (b > 0 ? (100 * a) / b : 0) - -export const RULES: Rule[] = [ - { - // 4401 disables phone credentials; a spike means the edge is wrongly refusing good clients. - id: 'auth-refusals', - breach: (c, min) => c.phone_open + c.control_active >= min && pct(c.refuse_4401, c.phone_open + c.control_active) > 0.5, - describe: (c) => `4401 refusals ${c.refuse_4401} = ${pct(c.refuse_4401, c.phone_open + c.control_active).toFixed(2)}% of hellos` - }, - { - id: 'busy-refusals', - breach: (c, min) => c.phone_open + c.control_active >= min && pct(c.refuse_busy, c.phone_open + c.control_active) > 5, - describe: (c) => `4429/4503 refusals ${c.refuse_busy} = ${pct(c.refuse_busy, c.phone_open + c.control_active).toFixed(1)}% of hellos` - }, - { - id: 'database', - breach: (c) => c.db_ops >= 20 && (pct(c.db_failures, c.db_ops) > 1 || c.db_p95_ms > 300), - describe: (c) => `database failures ${pct(c.db_failures, c.db_ops).toFixed(1)}% of ${c.db_ops} ops, p95 ${Math.round(c.db_p95_ms)} ms` - }, - { - // Mass drop: control closes jump to 3x the previous window. - id: 'mass-drop', - breach: (c, min, prior) => prior !== undefined && c.control_close >= min && c.control_close > 3 * Math.max(1, prior.control_close), - describe: (c) => `control closes ${c.control_close} in ${WINDOW_MIN} min, over 3x the previous window` - }, - { - // The director's calls failing (bad signature, stale or conflicting pushes, errors). - id: 'admin-failures', - breach: (c) => c.admin_failures >= ADMIN_FAILURES_MIN, - describe: (c) => `${c.admin_failures} failed admin calls in ${WINDOW_MIN} min` - }, - { - // Below the share rules' volume a handful of 4401s is still worth a look. - id: 'auth-refusals-low-volume', - breach: (c, min) => c.phone_open + c.control_active < min && c.refuse_4401 >= LOW_VOLUME_4401_MIN, - describe: (c) => `4401 refusals ${c.refuse_4401} in ${WINDOW_MIN} min at low volume (${c.phone_open + c.control_active} hellos)` - }, - { - // Dead man: traffic stopped (the edge or its metrics went silent) after a window that had some. - id: 'silent', - breach: (c, min, prior) => prior !== undefined && prior.events >= min && c.events === 0, - describe: () => `no edge events in ${WINDOW_MIN} min` - } -] +// Analytics Engine ingestion is not instant. +const LAG_MIN = 2 async function window(env: AlertsEnv, fromMin: number, toMin: number): Promise { const [counts, latency] = await Promise.all([ @@ -137,34 +45,53 @@ async function post(env: AlertsEnv, text: string): Promise { }) } +async function probes(env: AlertsEnv): Promise { + if (!env.EDGE_PROBE_URL) { return [] } + const [edge, object] = await Promise.all([probeEdge(env.EDGE_PROBE_URL), probeObject(env.EDGE_PROBE_URL)]) + return [...(edge ? [`FIRING edge-probe: ${edge}`] : []), ...(object ? [`FIRING object-probe: ${object}`] : [])] +} + +// Cloudflare's counts merged into each window; an unreadable API skips the rule and says so hourly. +async function withCloudflare(env: AlertsEnv, now: Counts, before: Counts, hourly: boolean): Promise { + try { + const counts = await cloudflareErrors(env, WINDOW_MIN, LAG_MIN) + if (counts) { + Object.assign(now, counts[0]) + Object.assign(before, counts[1]) + } + return null + } catch (error) { + return hourly ? `alerts cannot read Cloudflare analytics: ${(error as Error).message}` : null + } +} + export async function evaluate(env: AlertsEnv, scheduledTime: number): Promise { if (!env.AE_READ_TOKEN || !env.ALERT_WEBHOOK_URL) { console.warn('relay-edge alerts: secrets not set; skipping') return [] } - const minVolume = Number(env.MIN_VOLUME ?? '200') - const probeFailure = env.EDGE_PROBE_URL ? await probeEdge(env.EDGE_PROBE_URL) : null - const probeAlert = probeFailure ? [`FIRING edge-probe: ${probeFailure}`] : [] - if (probeAlert.length > 0) { await post(env, probeAlert[0]) } + const limits = thresholds(env) + const probeAlerts = await probes(env) + for (const text of probeAlerts) { await post(env, text) } let now: Counts let before: Counts try { - // 2-min lag: Analytics Engine ingestion is not instant. - ;[now, before] = await Promise.all([window(env, WINDOW_MIN + 2, 2), window(env, 2 * WINDOW_MIN + 2, WINDOW_MIN + 2)]) + ;[now, before] = await Promise.all([window(env, WINDOW_MIN + LAG_MIN, LAG_MIN), window(env, 2 * WINDOW_MIN + LAG_MIN, WINDOW_MIN + LAG_MIN)]) } catch (error) { const text = `alerts cannot read metrics: ${(error as Error).message}` await post(env, text) - return [...probeAlert, text] + return [...probeAlerts, text] } const hourly = new Date(scheduledTime).getUTCMinutes() < 5 - const fired: string[] = [] + const unreadable = await withCloudflare(env, now, before, hourly) + const fired: string[] = unreadable ? [unreadable] : [] for (const rule of RULES) { - if (!rule.breach(now, minVolume, before)) { continue } - const started = !rule.breach(before, minVolume) + if (!rule.breach(now, limits, before)) { continue } + const started = !rule.breach(before, limits) if (started || hourly) { fired.push(`${started ? 'FIRING' : 'STILL FIRING'} ${rule.id}: ${rule.describe(now)}`) } } for (const text of fired) { await post(env, text) } - return [...probeAlert, ...fired] + return [...probeAlerts, ...fired] } export default { diff --git a/cloud/apps/relay-edge/src/env.ts b/cloud/apps/relay-edge/src/env.ts index d61c4d692f4..31de529480a 100644 --- a/cloud/apps/relay-edge/src/env.ts +++ b/cloud/apps/relay-edge/src/env.ts @@ -11,7 +11,8 @@ export type Env = { // The cell id directors write into relay_assignments for hosts placed on the edge. EDGE_CELL_ID?: string // Relay JWTs: verified against the auth service's JWKS (fetched, cached per isolate, served - // stale for up to 30 min if auth is unreachable). RELAY_JWKS (inline set) is for local tests and + // stale for up to 30 min if auth is unreachable; a cold isolate falls back to the colo's last + // good set in the Cache API for up to 24 h, jwks-persist.ts). RELAY_JWKS (inline set) is for local tests and // the spike only. RELAY_JWKS_URL?: string RELAY_JWKS?: string diff --git a/cloud/apps/relay-edge/src/health.ts b/cloud/apps/relay-edge/src/health.ts new file mode 100644 index 00000000000..a685a761fcb --- /dev/null +++ b/cloud/apps/relay-edge/src/health.ts @@ -0,0 +1,37 @@ +// GET /__health/object: the alerts Worker's deep probe (alerts-probe.ts). Unauthenticated and +// side-effect free: it calls one fixed canary HostRelay object, which reads its own SQLite +// (SELECT 1) and answers. 200 {ok:true}, or 503 {ok:false} on any failure or after +// OBJECT_HEALTH_TIMEOUT_MS. Bounded per source per colo by PREAUTH_LIMITER (429, its own 'health:' +// key). The canary holds no host data, so even a distributed flood that overloads it affects no host. +import type { Env } from './env' +import { admitPreAuth } from './preauth' + +// Can never be a relay host id (HOST_ID allows no ':'), so it never shares an object with a host. +export const HEALTH_CANARY_OBJECT = 'health:canary' +export const OBJECT_HEALTH_TIMEOUT_MS = 3_000 + +async function canaryAnswers(env: Env): Promise { + let timer: ReturnType | undefined + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(false), OBJECT_HEALTH_TIMEOUT_MS) + }) + try { + const stub = env.HOST_RELAY.get(env.HOST_RELAY.idFromName(HEALTH_CANARY_OBJECT)) + const check = stub.healthCheck().then( + (result) => result.ok === true, + () => false + ) + return await Promise.race([check, timeout]) + } catch { + return false + } finally { + clearTimeout(timer) + } +} + +export async function objectHealth(env: Env, request: Request): Promise { + if (request.method !== 'GET') { return new Response(null, { status: 405 }) } + if (!(await admitPreAuth(env, request, 'health'))) { return new Response(null, { status: 429 }) } + const ok = await canaryAnswers(env) + return Response.json({ ok }, { status: ok ? 200 : 503, headers: { 'cache-control': 'no-store' } }) +} diff --git a/cloud/apps/relay-edge/src/host-relay-control.ts b/cloud/apps/relay-edge/src/host-relay-control.ts index 9f214ceb1f7..dddb466028d 100644 --- a/cloud/apps/relay-edge/src/host-relay-control.ts +++ b/cloud/apps/relay-edge/src/host-relay-control.ts @@ -57,10 +57,18 @@ export abstract class HostRelayControl extends HostRelayHandshake { metric(this.env, 'auth_refresh', 1, 'invalid', [], '', this.attachment(ws)?.place) return } - const claims = await verifyRelayJwt(this.env, parsed.data.relayJwt) + const verified = await verifyRelayJwt(this.env, parsed.data.relayJwt) // Re-read after the await: the attachment may have moved on (pongs, pings) meanwhile. const current = this.attachment(ws) if (!current || current.k !== 'control') { return } + // Keys unavailable (cold isolate, auth unreachable) says nothing about the token: keep the + // socket and its current session unchanged; the next refresh retries. The alarm still closes + // it at exp + grace, so this never extends access. + if (!verified.ok && verified.reason === 'keys_unavailable') { + metric(this.env, 'auth_refresh', 1, 'keys_unavailable', [], '', current.place) + return + } + const claims = verified.ok ? verified.claims : null const same = claims && claims.userId === current.claims.userId && diff --git a/cloud/apps/relay-edge/src/host-relay.ts b/cloud/apps/relay-edge/src/host-relay.ts index f2360870b3b..397b9dd3844 100644 --- a/cloud/apps/relay-edge/src/host-relay.ts +++ b/cloud/apps/relay-edge/src/host-relay.ts @@ -48,7 +48,7 @@ export class HostRelay extends HostRelayPhones { > { const refused = await this.claimAdminJti(replay) if (refused) { return refused } - if (this.ctx.getWebSockets('control').length > 0) { return { outcome: 'busy' } } + if (this.liveControls().length > 0) { return { outcome: 'busy' } } const devices = this.store.freezeForExport() await this.closeSplices(() => true, RELAY_CLOSE_CODE.WRONG_CELL, 'host moving') await this.reschedule() @@ -76,8 +76,7 @@ export class HostRelay extends HostRelayPhones { // desktop re-asks the director), and goes through the same path as that expiry. The assignment, // devices and splices are left as they are. async adminDrain(replay?: AdminReplayGuard | null): Promise { - // A socket whose attachment is cleared is already gone (closing, still listed). - const controls = this.ctx.getWebSockets('control').filter((ws) => this.attachment(ws) !== null) + const controls = this.liveControls() // An idle drain of a host nobody pushed changes nothing and must not give the object storage, // so its jti is not recorded; a copy of that token can at most drain a control that arrived // since, which sends the desktop back to the director once. @@ -92,6 +91,19 @@ export class HostRelay extends HostRelayPhones { return { outcome: 'drained', controls: controls.length } } + // A socket whose attachment is cleared is already gone (closing, still listed): drain and export + // both ignore it, so a just-drained desktop does not make the export that follows answer busy. + private liveControls(): WebSocket[] { + return this.ctx.getWebSockets('control').filter((ws) => this.attachment(ws) !== null) + } + + // Health probe (GET /__health/object, health.ts), only ever called on the canary object: one + // read of the object's SQLite. No writes, no alarm, no storage created. + healthCheck(): { ok: true } { + this.ctx.storage.sql.exec('SELECT 1').one() + return { ok: true } + } + private async claimAdminJti(replay: AdminReplayGuard | null | undefined): Promise { if (!replay) { return null } // Only storage calls between the read and the write: no other event runs in between. diff --git a/cloud/apps/relay-edge/src/index.ts b/cloud/apps/relay-edge/src/index.ts index 4b64e3a78b5..c12f4c1979f 100644 --- a/cloud/apps/relay-edge/src/index.ts +++ b/cloud/apps/relay-edge/src/index.ts @@ -7,6 +7,7 @@ import { HOST_ID, readBearer, verifyRelayJwt } from './jwt' import { SPLICE_SLOTS, makeConnId, parseConnId } from './splice-slot' import { metric, requestPlace } from './metrics' import { admitPreAuth, notOnEdge } from './preauth' +import { objectHealth } from './health' import { AssignBodySchema, DrainBodySchema, EraseBodySchema, ExportBodySchema, verifyAdminRequest, type AdminOp } from './admin' export { HostRelay, SpliceRelay } @@ -96,6 +97,7 @@ export default { return Response.json({ ok: true, connectionCapacityProtocol: 2 }) } if (url.search) { return reject(400) } + if (url.pathname === '/__health/object') { return objectHealth(env, request) } const admin = url.pathname.match(/^\/__admin\/(assign|export|erase|drain)$/) if (admin) { return handleAdmin(env, request, admin[1] as AdminOp) } // Spike-only: force the object to reset (simulates eviction / runtime restart). @@ -146,11 +148,20 @@ export default { if (url.pathname === '/v1/host/control') { if (!isUpgrade) { return reject(426) } const token = readBearer(request.headers.get('authorization')) - const claims = token ? await verifyRelayJwt(env, token) : null - if (!claims) { return reject(401) } - return forward(env, request, claims.relayHostId, { + const verified = token ? await verifyRelayJwt(env, token) : null + if (!verified?.ok) { + // No keys to judge the token (cold isolate, auth unreachable): 503, not 401. Shipped + // desktops treat any refused upgrade the same (a network error, retried through the + // director), so this changes no client behavior; it keeps an outage out of the 401s. + if (verified?.reason === 'keys_unavailable') { + metric(env, 'control_upgrade', 1, 'keys_unavailable', [], '', requestPlace(request)) + return reject(503, 5) + } + return reject(401) + } + return forward(env, request, verified.claims.relayHostId, { 'x-edge-kind': 'control', - 'x-edge-claims': JSON.stringify(claims) + 'x-edge-claims': JSON.stringify(verified.claims) }) } diff --git a/cloud/apps/relay-edge/src/jwks-persist.ts b/cloud/apps/relay-edge/src/jwks-persist.ts new file mode 100644 index 00000000000..96759d96979 --- /dev/null +++ b/cloud/apps/relay-edge/src/jwks-persist.ts @@ -0,0 +1,99 @@ +// Last good relay JWKS, kept in the colo's Cache API (shared by every isolate in the colo, no object +// storage writes). A fresh isolate whose fetch of the keys fails falls back to it, so a cold isolate +// during an auth outage does not refuse valid desktops. Never used while auth answers: a fetched +// set always wins, and a token whose key the stored set lacks reads as "keys unavailable", not invalid. +import { createLocalJWKSet, type FetchImplementation, type JSONWebKeySet, type JWTVerifyGetKey } from 'jose' + +// The same 30 min the cells trust a stale key set for: a rotated or leaked key is not trusted +// longer here during an auth outage. Covers only isolates that have no set at all, and never +// extends a live socket past its token's own expiry (+60 s grace). +export const PERSISTED_JWKS_MAX_STALE_MS = 30 * 60 * 1000 +const STORED_AT_HEADER = 'x-orca-jwks-stored-at' + +function edgeCache(): Cache | null { + try { + return typeof caches === 'undefined' ? null : (caches as unknown as { default: Cache }).default + } catch { + return null + } +} + +function isKeySet(value: unknown): value is JSONWebKeySet { + return typeof value === 'object' && value !== null && Array.isArray((value as { keys?: unknown }).keys) +} + +// The key set's fetch, plus a copy of every 200 answer into the Cache API (best effort; never fails a fetch). +export function persistingFetch(url: string): FetchImplementation { + return async (target, options) => { + const response = await fetch(target, options) + if (response.status !== 200) { return response } + const body = await response.text() + const cache = edgeCache() + if (cache) { + try { + if (isKeySet(JSON.parse(body))) { + await cache.put( + new Request(url), + new Response(body, { + headers: { + 'content-type': 'application/json', + 'cache-control': `max-age=${PERSISTED_JWKS_MAX_STALE_MS / 1000}`, + [STORED_AT_HEADER]: String(Date.now()) + } + }) + ) + } + } catch { + // Persistence is a fallback only. + } + } + return new Response(body, { status: 200, headers: response.headers }) + } +} + +export async function readPersistedJwks(url: string, now = Date.now()): Promise { + const cache = edgeCache() + if (!cache) { return null } + try { + const stored = await cache.match(new Request(url)) + const storedAt = Number(stored?.headers.get(STORED_AT_HEADER)) + if (!stored || !Number.isFinite(storedAt) || now - storedAt > PERSISTED_JWKS_MAX_STALE_MS) { return null } + const jwks: unknown = await stored.json() + return isKeySet(jwks) ? jwks : null + } catch { + return null + } +} + +// The verifier could not obtain keys to judge the token: not a verdict on the token. +export class KeysUnavailableError extends Error { + constructor() { + super('relay_jwks_unavailable') + } +} + +function isVerdict(error: unknown): boolean { + const code = (error as { code?: unknown } | null)?.code + return code === 'ERR_JWKS_NO_MATCHING_KEY' || code === 'ERR_JWKS_MULTIPLE_MATCHING_KEYS' +} + +// Wraps the remote (stale-if-error) key lookup: a set auth served decides; anything else (fetch +// error, timeout, non-200, unparsable set) tries the persisted set and otherwise throws +// KeysUnavailableError. +export function withPersistedFallback(url: string, remote: JWTVerifyGetKey): JWTVerifyGetKey { + return async (protectedHeader, token) => { + try { + return await remote(protectedHeader, token) + } catch (error) { + if (isVerdict(error)) { throw error } + const persisted = await readPersistedJwks(url) + if (!persisted) { throw new KeysUnavailableError() } + try { + return await createLocalJWKSet(persisted)(protectedHeader, token) + } catch { + // An old stored set lacking a rotated key is not an answer from auth. + throw new KeysUnavailableError() + } + } + } +} diff --git a/cloud/apps/relay-edge/src/jwt.ts b/cloud/apps/relay-edge/src/jwt.ts index eba24fd227a..2edb497f810 100644 --- a/cloud/apps/relay-edge/src/jwt.ts +++ b/cloud/apps/relay-edge/src/jwt.ts @@ -1,9 +1,12 @@ // Relay JWT verification: same claims, algorithm, issuer and audience as the GCE cells // (cloud/apps/relay/src/authentication/relay-token-verifier.ts), and the same stale-if-error JWKS cache, reused -// unchanged, so an auth outage does not turn valid hosts away with 4401. +// unchanged, so an auth outage does not turn valid hosts away with 4401. A cold isolate also falls +// back to the colo's last good set (jwks-persist.ts), and "could not get keys" is reported apart +// from "token invalid" so callers never treat an outage as a bad credential. import { createLocalJWKSet, jwtVerify, type JSONWebKeySet, type JWTVerifyGetKey } from 'jose' import { z } from 'zod' import { createStaleIfErrorJWKSet } from '../../relay/src/authentication/stale-if-error-jwks' +import { KeysUnavailableError, persistingFetch, withPersistedFallback } from './jwks-persist' import type { EdgeClaims, Env } from './env' export const HOST_ID = /^[A-Za-z0-9_-]{16}$/ @@ -24,15 +27,18 @@ function keysFor(env: Env): JWTVerifyGetKey { const source = env.RELAY_JWKS_URL ?? `inline:${env.RELAY_JWKS ?? ''}` let keys = keySets.get(source) if (!keys) { - keys = env.RELAY_JWKS_URL - ? createStaleIfErrorJWKSet(new URL(env.RELAY_JWKS_URL)) + const url = env.RELAY_JWKS_URL + keys = url + ? withPersistedFallback(url, createStaleIfErrorJWKSet(new URL(url), { fetch: persistingFetch(url) })) : createLocalJWKSet(JSON.parse(env.RELAY_JWKS ?? '{"keys":[]}') as JSONWebKeySet) keySets.set(source, keys) } return keys } -export async function verifyRelayJwt(env: Env, token: string): Promise { +export type RelayJwtResult = { ok: true; claims: EdgeClaims } | { ok: false; reason: 'invalid' | 'keys_unavailable' } + +export async function verifyRelayJwt(env: Env, token: string): Promise { try { const { payload } = await jwtVerify(token, keysFor(env), { audience: env.RELAY_JWT_AUDIENCE, @@ -41,14 +47,17 @@ export async function verifyRelayJwt(env: Env, token: string): Promise'. | 'admin_no_jti' - // Host auth-refresh frame: label 'extended' | 'unchanged' | 'invalid' | 'identity_changed'. + // Host auth-refresh frame: label 'extended' | 'unchanged' | 'invalid' | 'identity_changed' | + // 'keys_unavailable' (no relay JWKS to judge it; the socket stays, unchanged). | 'auth_refresh' + // Control upgrade answered 503 because no relay JWKS could be obtained: label 'keys_unavailable'. + | 'control_upgrade' // Control closed 4401 for expired authorization: value = seconds since its last extending // refresh, label 'refreshed'; -1 and label 'never' when this socket never extended its token. | 'auth_expired_close' - // Pre-auth socket refused by the per-source limiter: label 'phone' | 'data'. Unpushed-host phones: 'not_on_edge'. + // Pre-auth socket refused by the per-source limiter: label 'phone' | 'data' | 'health'. Unpushed-host phones: 'not_on_edge'. | 'preauth_limited' | 'not_on_edge' // blobs: [event, label, version, reason, place]; doubles: [value, ...extra]. place is the // client's ingress '/' (e.g. 'AS/HKG'), '' where no client is involved. One index per event for sampling. -// Heartbeat events fire per host every ping interval; keep 1 in 10 (multiply counts by 10). -const HEARTBEAT_SAMPLE = 0.1 +// Heartbeat events (alarm_lag_ms, pong_rtt_ms) fire per host every ping interval; only 1 in +// HEARTBEAT_FACTOR is written, and Analytics Engine's _sample_interval does not know that. Those +// two events carry the factor as their last double (double2, as they pass no extra), so a count +// is sum(_sample_interval * double2). No other event has it. +const HEARTBEAT_FACTOR = 10 export function metric(env: Env, event: MetricEvent, value = 1, label = '', extra: number[] = [], reason = '', place = ''): void { - if ((event === 'alarm_lag_ms' || event === 'pong_rtt_ms') && Math.random() >= HEARTBEAT_SAMPLE) { return } + const heartbeat = event === 'alarm_lag_ms' || event === 'pong_rtt_ms' + if (heartbeat && Math.random() >= 1 / HEARTBEAT_FACTOR) { return } try { env.METRICS?.writeDataPoint({ indexes: [event], blobs: [event, label, env.EDGE_VERSION ?? 'dev', reason, place], - doubles: [value, ...extra] + doubles: heartbeat ? [value, ...extra, HEARTBEAT_FACTOR] : [value, ...extra] }) } catch { // Metrics must never affect the relay path. diff --git a/cloud/apps/relay-edge/src/preauth.ts b/cloud/apps/relay-edge/src/preauth.ts index bae0206b784..2f7e50a94e8 100644 --- a/cloud/apps/relay-edge/src/preauth.ts +++ b/cloud/apps/relay-edge/src/preauth.ts @@ -1,4 +1,5 @@ -// Limits on sockets that have not authenticated yet (phone connects, host data sockets). Desktops +// Limits on sockets that have not authenticated yet (phone connects, host data sockets), and on +// the unauthenticated object health probe (health.ts). Desktops // authenticate with a relay JWT before the upgrade; phones and data sockets prove a credential in // their first frame, so until then they are only bounded here. import { RELAY_CLOSE_CODE } from '../../../packages/relay-contract/src/close-codes' @@ -23,7 +24,7 @@ export function sourceKey(request: Request): string { // Counted per source per colo by the Rate Limiting binding. Fails open: a limiter outage must not // refuse live users. Without a binding (local spike runs) nothing is limited. -export async function admitPreAuth(env: Env, request: Request, kind: 'phone' | 'data'): Promise { +export async function admitPreAuth(env: Env, request: Request, kind: 'phone' | 'data' | 'health'): Promise { if (!env.PREAUTH_LIMITER) { return true } try { const { success } = await env.PREAUTH_LIMITER.limit({ key: `${kind}:${sourceKey(request)}` }) diff --git a/cloud/apps/relay-edge/test/alerts-cohort.test.ts b/cloud/apps/relay-edge/test/alerts-cohort.test.ts new file mode 100644 index 00000000000..c2c15c626f0 --- /dev/null +++ b/cloud/apps/relay-edge/test/alerts-cohort.test.ts @@ -0,0 +1,147 @@ +// Alert rules that scale with the cohort: configurable MIN_VOLUME, the reconnect surge, Cloudflare +// GraphQL errors and the canary object probe. Analytics Engine, GraphQL, edge and webhook stubbed. +import { afterEach, describe, expect, it, vi } from 'vitest' +import { evaluate } from '../src/alerts' +import { cloudflareQuery } from '../src/alerts-cloudflare' +import { probeObject } from '../src/alerts-probe' + +const quiet = { phone_open: 20, control_active: 20, reconnects: 20, refuse_4401: 0, refuse_busy: 0, control_close: 5, db_ops: 0, db_failures: 0, admin_failures: 0, db_p95_ms: 0, events: 60 } +const env = { ACCOUNT_ID: 'acct', DATASET: 'd', ENVIRONMENT: 'test', AE_READ_TOKEN: 't', ALERT_WEBHOOK_URL: 'https://hook.invalid' } +const GRAPHQL = 'https://api.cloudflare.com/client/v4/graphql' +const PROBE = 'https://edge.invalid' +type Row = typeof quiet +type Group = { sum: { requests: number }; dimensions: { status: string; namespaceId?: string } } +type Graph = { w0?: Group[]; w1?: Group[]; o0?: Group[]; o1?: Group[] } | 'errors' | 'down' + +const at17 = Date.UTC(2026, 10, 6, 10, 17) +const at02 = Date.UTC(2026, 10, 6, 10, 2) +const group = (status: string, requests: number, namespaceId?: string): Group => ({ sum: { requests }, dimensions: { status, ...(namespaceId ? { namespaceId } : {}) } }) + +function stub(now: Partial, before: Partial = {}, graph: Graph = {}) { + const posts: string[] = [] + const graphql: { query: string; variables: Record }[] = [] + vi.stubGlobal('fetch', async (url: string, init: RequestInit = {}) => { + if (url === env.ALERT_WEBHOOK_URL) { + posts.push(JSON.parse(String(init.body)).text) + return new Response('ok') + } + if (url === GRAPHQL) { + graphql.push(JSON.parse(String(init.body))) + if (graph === 'down') { return new Response('no', { status: 403 }) } + if (graph === 'errors') { return Response.json({ data: null, errors: [{ message: 'unknown field' }] }) } + return Response.json({ data: { viewer: { accounts: [graph] } }, errors: null }) + } + const sql = String(init.body) + const row = sql.includes("INTERVAL '32' MINUTE") ? { ...quiet, ...before } : { ...quiet, ...now } + return Response.json({ data: [sql.includes('db_p95_ms') && !sql.includes('sumIf') ? { db_p95_ms: row.db_p95_ms } : row] }) + }) + return { posts, graphql } +} + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('MIN_VOLUME', () => { + it('scales the share rules to the cohort', async () => { + stub({ refuse_4401: 1 }) + expect(await evaluate(env, at17)).toEqual([]) + stub({ refuse_4401: 1 }) + expect(await evaluate({ ...env, MIN_VOLUME: '20' }, at17)).toEqual(['FIRING auth-refusals: 4401 refusals 1 = 2.50% of hellos']) + }) +}) + +describe('reconnect surge', () => { + it('fires at 3x the previous window once above the floor', async () => { + stub({ reconnects: 60 }, { reconnects: 20 }) + expect(await evaluate(env, at17)).toEqual(['FIRING reconnect-surge: control hellos 60 in 15 min, at least 3x the previous window']) + }) + + it('stays quiet below the floor or below 3x', async () => { + stub({ reconnects: 49 }, { reconnects: 0 }) + expect(await evaluate(env, at17)).toEqual([]) + stub({ reconnects: 100 }, { reconnects: 40 }) + expect(await evaluate(env, at17)).toEqual([]) + }) + + it('takes its floor from RECONNECT_SURGE_MIN', async () => { + stub({ reconnects: 6 }, { reconnects: 2 }) + expect(await evaluate({ ...env, RECONNECT_SURGE_MIN: '5' }, at17)).toHaveLength(1) + }) +}) + +describe('cloudflare errors', () => { + it('is skipped without CF_GRAPHQL_TOKEN', async () => { + const { graphql } = stub({}) + expect(await evaluate(env, at17)).toEqual([]) + expect(graphql).toHaveLength(0) + }) + + it('fires above 20 errors and 1% of invocations, for the edge script', async () => { + const { graphql, posts } = stub({}, {}, { w0: [group('success', 2000), group('clientDisconnected', 500), group('scriptThrewException', 30)], w1: [group('success', 2000)] }) + const fired = await evaluate({ ...env, CF_GRAPHQL_TOKEN: 'g' }, at17) + expect(fired).toEqual(['FIRING cloudflare-errors: Worker/Durable Object errors 30 = 1.2% of 2530 invocations in 15 min']) + expect(graphql[0].variables.script).toBe('orca-relay-edge') + expect(graphql[0].variables.a).toBe('acct') + expect(graphql[0].query).not.toContain('durableObjects') + expect(posts[0]).not.toMatch(/acct|orca-relay-edge/) + }) + + it('stays quiet under either threshold, and honours the env thresholds', async () => { + stub({}, {}, { w0: [group('success', 20000), group('exceededResources', 30)] }) + expect(await evaluate({ ...env, CF_GRAPHQL_TOKEN: 'g' }, at17)).toEqual([]) + stub({}, {}, { w0: [group('success', 100), group('internalError', 20)] }) + expect(await evaluate({ ...env, CF_GRAPHQL_TOKEN: 'g' }, at17)).toEqual([]) + stub({}, {}, { w0: [group('success', 100), group('internalError', 20)] }) + expect(await evaluate({ ...env, CF_GRAPHQL_TOKEN: 'g', CF_ERRORS_MIN: '10' }, at17)).toHaveLength(1) + }) + + it('counts only the configured Durable Object namespaces', async () => { + const { graphql } = stub({}, {}, { w0: [group('success', 100)], o0: [group('error', 50, 'ns-edge'), group('error', 500, 'ns-other'), group('success', 900, 'ns-edge')] }) + const fired = await evaluate({ ...env, CF_GRAPHQL_TOKEN: 'g', CF_DO_NAMESPACE_IDS: 'ns-edge' }, at17) + expect(fired).toEqual(['FIRING cloudflare-errors: Worker/Durable Object errors 50 = 4.8% of 1050 invocations in 15 min']) + expect(graphql[0].query).toContain('durableObjectsInvocationsAdaptiveGroups') + }) + + it('reports an unreadable API hourly only, and keeps the other rules running', async () => { + stub({ admin_failures: 3 }, {}, 'down') + expect(await evaluate({ ...env, CF_GRAPHQL_TOKEN: 'g' }, at17)).toEqual(['FIRING admin-failures: 3 failed admin calls in 15 min']) + stub({}, {}, 'errors') + expect(await evaluate({ ...env, CF_GRAPHQL_TOKEN: 'g' }, at02)).toEqual(['alerts cannot read Cloudflare analytics: cloudflare analytics query returned errors']) + }) + + it('keeps the query small: one call, both windows, documented fields only', () => { + expect(cloudflareQuery(false)).toMatch(/w0:workersInvocationsAdaptive\(limit:50,filter:\{scriptName:\$script,datetime_geq:\$s0,datetime_lt:\$e0\}\)\{sum\{requests\} dimensions\{status\}\}/) + expect(cloudflareQuery(true)).toContain('o1:durableObjectsInvocationsAdaptiveGroups') + }) +}) + +describe('object probe', () => { + const answer = (status: number, body: unknown) => vi.stubGlobal('fetch', async () => Response.json(body, { status })) + + it('passes on 200 {ok:true} and alerts on 503 or a wrong body', async () => { + answer(200, { ok: true }) + expect(await probeObject(PROBE)).toBeNull() + answer(503, { ok: false }) + expect(await probeObject(PROBE)).toBe('object health check got 503, expected 200') + answer(200, { ok: false }) + expect(await probeObject(PROBE)).toBe('object health check got 200, expected 200') + }) + + it('alerts over 2 s', async () => { + answer(200, { ok: true }) + let t = 0 + expect(await probeObject(PROBE, () => (t += 2_001))).toBe('object health check took 2001 ms') + }) + + it('runs with the edge probe each cycle and fires as object-probe', async () => { + vi.stubGlobal('fetch', async (url: string, init: RequestInit = {}) => { + if (url === env.ALERT_WEBHOOK_URL) { return new Response('ok') } + if (url === `${PROBE}/v1/host/control`) { return new Response(null, { status: 401 }) } + if (url === `${PROBE}/__health/object`) { return Response.json({ ok: false }, { status: 503 }) } + const sql = String(init.body) + return Response.json({ data: [sql.includes('sumIf') ? quiet : { db_p95_ms: 0 }] }) + }) + expect(await evaluate({ ...env, EDGE_PROBE_URL: PROBE }, at17)).toEqual(['FIRING object-probe: object health check got 503, expected 200']) + }) +}) diff --git a/cloud/apps/relay-edge/test/alerts-small-volume.test.ts b/cloud/apps/relay-edge/test/alerts-small-volume.test.ts index c61b6832fb2..3937d8d3b88 100644 --- a/cloud/apps/relay-edge/test/alerts-small-volume.test.ts +++ b/cloud/apps/relay-edge/test/alerts-small-volume.test.ts @@ -19,6 +19,10 @@ function stub(now: Partial, before: Partial = {}, edge: EdgeAnswer = 4 posts.push(JSON.parse(String(init.body)).text) return new Response('ok') } + if (url === `${PROBE}/__health/object`) { + if (edge === 'network') { throw new TypeError('connection refused') } + return Response.json({ ok: true }) + } if (url === `${PROBE}/v1/host/control`) { probes.push(init) if (edge === 'network') { throw new TypeError('connection refused') } @@ -114,7 +118,8 @@ describe('edge probe', () => { }) const fired = await evaluate({ ...env, EDGE_PROBE_URL: PROBE }, at17) expect(fired[0]).toContain('FIRING edge-probe') - expect(fired[1]).toContain('cannot read metrics') + expect(fired[1]).toContain('FIRING object-probe') + expect(fired[2]).toContain('cannot read metrics') }) it('alerts after 5 s without an answer', async () => { diff --git a/cloud/apps/relay-edge/test/jwt-keys.test.ts b/cloud/apps/relay-edge/test/jwt-keys.test.ts new file mode 100644 index 00000000000..22524f3f0f7 --- /dev/null +++ b/cloud/apps/relay-edge/test/jwt-keys.test.ts @@ -0,0 +1,124 @@ +// Relay JWT verification when keys cannot be fetched: "token invalid" and "no keys to judge it" +// stay apart, and a cold isolate falls back to the colo's persisted set (Cache API, stubbed). +import { SignJWT, exportJWK, generateKeyPair, type CryptoKey } from 'jose' +import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest' +import type { Env } from '../src/env' +import { PERSISTED_JWKS_MAX_STALE_MS } from '../src/jwks-persist' + +const ISSUER = 'https://auth.test.invalid' +const HOST = 'AAAAAAAAAAAAAAAA' +let privateKey: CryptoKey +let otherKey: CryptoKey +let jwks: { keys: Record[] } +let urlSeq = 0 + +beforeAll(async () => { + const pair = await generateKeyPair('ES256', { extractable: true }) + privateKey = pair.privateKey + otherKey = (await generateKeyPair('ES256')).privateKey + jwks = { keys: [{ ...(await exportJWK(pair.publicKey)), kid: 'k1', alg: 'ES256', use: 'sig' }] } +}) + +const mint = (key: CryptoKey = privateKey, kid = 'k1') => + new SignJWT({ prof: 'p', relayHostId: HOST, purpose: 'host-control' }) + .setProtectedHeader({ alg: 'ES256', kid }) + .setSubject('u') + .setIssuer(ISSUER) + .setAudience('orca-relay') + .setExpirationTime('5m') + .sign(key) + +// A Cache API stand-in shared by every "isolate" (module instance) in the test. +function stubCache(): Map { + const store = new Map() + vi.stubGlobal('caches', { + default: { + put: async (request: Request, response: Response) => { + store.set(request.url, new Response(await response.text(), { headers: response.headers })) + }, + match: async (request: Request) => store.get(request.url)?.clone() + } + }) + return store +} + +// Each call is a fresh isolate: a new module instance with an empty in-memory key cache. +async function coldVerifier() { + vi.resetModules() + return (await import('../src/jwt')).verifyRelayJwt +} + +function envFor(url: string): Env { + return { RELAY_JWKS_URL: url, RELAY_AUTH_ISSUER: ISSUER, RELAY_JWT_AUDIENCE: 'orca-relay' } as Env +} + +function authUp(up: () => boolean) { + vi.stubGlobal('fetch', async () => { + if (!up()) { throw new TypeError('connection refused') } + return Response.json(jwks) + }) +} + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('relay JWT key availability', () => { + it('reports keys_unavailable, not invalid, when a cold isolate cannot fetch and nothing is persisted', async () => { + stubCache() + authUp(() => false) + const verify = await coldVerifier() + expect(await verify(envFor(`https://auth.test.invalid/jwks/${++urlSeq}`), await mint())).toEqual({ ok: false, reason: 'keys_unavailable' }) + }) + + it('reports keys_unavailable on a non-200 answer', async () => { + stubCache() + vi.stubGlobal('fetch', async () => new Response('down', { status: 502 })) + const verify = await coldVerifier() + expect(await verify(envFor(`https://auth.test.invalid/jwks/${++urlSeq}`), await mint())).toEqual({ ok: false, reason: 'keys_unavailable' }) + }) + + it('still says invalid for a bad token when keys are available', async () => { + stubCache() + authUp(() => true) + const verify = await coldVerifier() + const env = envFor(`https://auth.test.invalid/jwks/${++urlSeq}`) + expect(await verify(env, await mint(otherKey))).toEqual({ ok: false, reason: 'invalid' }) + expect(await verify(env, 'not-a-jwt')).toEqual({ ok: false, reason: 'invalid' }) + expect(await verify(env, await mint(privateKey, 'unknown-kid'))).toEqual({ ok: false, reason: 'invalid' }) + }) + + it('a cold isolate verifies from the persisted set while auth is down', async () => { + const store = stubCache() + let up = true + authUp(() => up) + const env = envFor(`https://auth.test.invalid/jwks/${++urlSeq}`) + const warm = await coldVerifier() + expect((await warm(env, await mint())).ok).toBe(true) + expect(store.size).toBe(1) + up = false + const cold = await coldVerifier() + const result = await cold(env, await mint()) + expect(result.ok && result.claims.relayHostId).toBe(HOST) + // A forged token is still refused from the persisted set. + expect(await cold(env, await mint(otherKey))).toEqual({ ok: false, reason: 'invalid' }) + }) + + it('ignores a persisted set past its max age, and one lacking the token key', async () => { + const store = stubCache() + authUp(() => false) + const url = `https://auth.test.invalid/jwks/${++urlSeq}` + const stored = (storedAt: number, keys: unknown) => + new Response(JSON.stringify(keys), { headers: { 'x-orca-jwks-stored-at': String(storedAt) } }) + store.set(url, stored(Date.now() - PERSISTED_JWKS_MAX_STALE_MS - 1000, jwks)) + expect(await (await coldVerifier())(envFor(url), await mint())).toEqual({ ok: false, reason: 'keys_unavailable' }) + store.set(url, stored(Date.now(), { keys: [] })) + expect(await (await coldVerifier())(envFor(url), await mint())).toEqual({ ok: false, reason: 'keys_unavailable' }) + }) + + it('works without a Cache API (local tests)', async () => { + authUp(() => true) + const verify = await coldVerifier() + expect((await verify(envFor(`https://auth.test.invalid/jwks/${++urlSeq}`), await mint())).ok).toBe(true) + }) +}) diff --git a/cloud/apps/relay-edge/test/mvp-object-hardening.ts b/cloud/apps/relay-edge/test/mvp-object-hardening.ts index 9349cf242fe..a304dabe9e4 100644 --- a/cloud/apps/relay-edge/test/mvp-object-hardening.ts +++ b/cloud/apps/relay-edge/test/mvp-object-hardening.ts @@ -1,8 +1,11 @@ -// mvp-object.ts checks (local only) for control liveness on any inbound frame and the director's -// signed `drain` op. +// mvp-object.ts checks (local only) for control liveness on any inbound frame, the director's +// signed `drain` op, and the object health route. import { randomBytes } from 'node:crypto' +import { readdirSync } from 'node:fs' +import { join } from 'node:path' +import { DatabaseSync } from 'node:sqlite' import type { KeyLike } from 'jose' -import { EPOCH, adminCall, assign, control, mint, newHost, record, sleep, waitFor } from './harness-mvp-object' +import { EPOCH, ORIGIN, adminCall, assign, control, mint, newHost, record, sleep, waitFor } from './harness-mvp-object' // Local timing (mvp-object.ts): ping every 3 s, silence timeout 9 s. const SILENCE_MS = 9_000 @@ -58,6 +61,9 @@ export async function drainCheck(key: KeyLike): Promise { const live = control(host, await mint(key, id)) await live.client.connect() const drained = await adminCall('drain', { relayHostId: id }) + // Straight after the drain: the closing control must not make the export answer busy. + const exported = await adminCall('export', { relayHostId: id }) + record('F5 export right after a drain: exported, not busy', exported.json?.outcome === 'exported', `${exported.status}:${JSON.stringify(exported.json?.outcome)}`) await waitFor(() => live.events.closes.length > 0, 3_000) const frame = live.events.drains[0] record( @@ -69,3 +75,31 @@ export async function drainCheck(key: KeyLike): Promise { const after = await adminCall('drain', { relayHostId: id }) record('F5 drain again after the control left: idle', after.json?.outcome === 'idle', JSON.stringify(after.json)) } + +// H1: GET /__health/object reads the canary object's storage and answers 200 {ok:true}; it writes +// nothing, only GET is served, and it is limited per source like the other unauthenticated routes. +export async function objectHealthCheck(stateDir: string): Promise { + const list = () => readdirSync(stateDir, { recursive: true }).map(String).filter((f) => f.endsWith('.sqlite')) + const beforeList = new Set(list()) + const first = await fetch(`${ORIGIN}/__health/object`) + const body = (await first.json().catch(() => null)) as { ok?: unknown } | null + const post = (await fetch(`${ORIGIN}/__health/object`, { method: 'POST' })).status + const tables = list() + .filter((f) => !beforeList.has(f)) + .flatMap((f) => { + const db = new DatabaseSync(join(stateDir, f), { readOnly: true }) + try { + return db.prepare("SELECT name FROM sqlite_master WHERE type = 'table'").all().map((r) => String(r.name)).filter((name) => !name.startsWith('__')) + } finally { + db.close() + } + }) + record( + 'H1 object health: 200 {ok:true}, POST 405, nothing written to object storage', + first.status === 200 && body?.ok === true && post === 405 && tables.length === 0, + `status=${first.status} body=${JSON.stringify(body)} post=${post} tables=${tables.join(',') || 'none'}` + ) + const statuses: number[] = [] + for (let i = 0; i < 70; i++) { statuses.push((await fetch(`${ORIGIN}/__health/object`)).status) } + record('H1 object health past the per-source limit: 429', statuses.at(-1) === 429, `limitedFrom=${statuses.indexOf(429)}`) +} diff --git a/cloud/apps/relay-edge/test/mvp-object.ts b/cloud/apps/relay-edge/test/mvp-object.ts index b4813e7f8b1..8183bd102ba 100644 --- a/cloud/apps/relay-edge/test/mvp-object.ts +++ b/cloud/apps/relay-edge/test/mvp-object.ts @@ -41,7 +41,7 @@ import { waitFor, type Host } from './harness-mvp-object' -import { drainCheck, heartbeatCheck } from './mvp-object-hardening' +import { drainCheck, heartbeatCheck, objectHealthCheck } from './mvp-object-hardening' import { replayCheck, rotationCheck } from './mvp-object-replay' async function mainRemote(): Promise { @@ -110,6 +110,7 @@ async function main(): Promise { await Promise.all([heartbeatCheck(privateKey), drainCheck(privateKey), replayCheck(), rotationCheck()]) await sweepCheck() await preAuthCheck(stateDir) + await objectHealthCheck(stateDir) } catch (error) { console.error(error) console.error(devLog.split('\n').filter((l) => /error|exception/i.test(l)).slice(-20).join('\n')) diff --git a/cloud/apps/relay-edge/wrangler.alerts.jsonc b/cloud/apps/relay-edge/wrangler.alerts.jsonc index bbccda93e3d..531b9b01e32 100644 --- a/cloud/apps/relay-edge/wrangler.alerts.jsonc +++ b/cloud/apps/relay-edge/wrangler.alerts.jsonc @@ -1,8 +1,9 @@ { "$schema": "node_modules/wrangler/config-schema.json", // Edge alerts (src/alerts.ts): separate from the edge so they keep running when it is broken. - // Secrets, set from stdin only: AE_READ_TOKEN, ALERT_WEBHOOK_URL. Local only; deployed - // environments live in the private deploy config. + // Secrets, set from stdin only: AE_READ_TOKEN, ALERT_WEBHOOK_URL, and optionally CF_GRAPHQL_TOKEN + // (enables the cloudflare-errors rule). Optional vars and their defaults: src/alerts-rules.ts. + // Local only; deployed environments live in the private deploy config. "name": "orca-relay-edge-alerts-local", "main": "src/alerts.ts", "compatibility_date": "2026-11-01", @@ -14,7 +15,8 @@ "ENVIRONMENT": "local", "MIN_VOLUME": "200", // Optional: the edge origin probed each run (an unauthenticated control upgrade must get 401; - // any other status, a network error or over 5 s alerts, on every failing run). Omit to skip. + // any other status, a network error or over 5 s alerts, on every failing run; and + // GET /__health/object must answer 200 {ok:true} within 2 s). Omit to skip. "EDGE_PROBE_URL": "http://127.0.0.1:8787" } }