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

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<string>): 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<string, Group[] | undefined>[] } }
+    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([]))
+  ]
+}

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<string | null> {
+  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)
+  }
+}

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 ('<op>:<outcome>'); 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<string, number>
+// `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`
+  }
+]

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 ('<op>:<outcome>'); 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<string, number>
-// `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<Counts> {
   const [counts, latency] = await Promise.all([
@@ -137,34 +45,53 @@ async function post(env: AlertsEnv, text: string): Promise<void> {
   })
 }
 
+async function probes(env: AlertsEnv): Promise<string[]> {
+  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<string | null> {
+  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<string[]> {
   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 {

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

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<boolean> {
+  let timer: ReturnType<typeof setTimeout> | undefined
+  const timeout = new Promise<boolean>((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<Response> {
+  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' } })
+}

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 &&

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<ReplayRefusal | { outcome: 'idle' } | { outcome: 'drained'; controls: number }> {
-    // 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<ReplayRefusal | null> {
     if (!replay) { return null }
     // Only storage calls between the read and the write: no other event runs in between.

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)
       })
     }
 

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<JSONWebKeySet | null> {
+  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()
+      }
+    }
+  }
+}

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<EdgeClaims | null> {
+export type RelayJwtResult = { ok: true; claims: EdgeClaims } | { ok: false; reason: 'invalid' | 'keys_unavailable' }
+
+export async function verifyRelayJwt(env: Env, token: string): Promise<RelayJwtResult> {
   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<EdgeClaim
     })
     const claims = ClaimsSchema.parse(payload)
     return {
-      userId: claims.sub,
-      profileId: claims.prof,
-      organizationId: claims.org ?? '',
-      relayHostId: claims.relayHostId,
-      exp: claims.exp * 1000
+      ok: true,
+      claims: {
+        userId: claims.sub,
+        profileId: claims.prof,
+        organizationId: claims.org ?? '',
+        relayHostId: claims.relayHostId,
+        exp: claims.exp * 1000
+      }
     }
-  } catch {
-    return null
+  } catch (error) {
+    return { ok: false, reason: error instanceof KeysUnavailableError ? 'keys_unavailable' : 'invalid' }
   }
 }
 

cloud/apps/relay-edge/src/metrics.ts

@@ -19,27 +19,34 @@ export type MetricEvent =
   | 'admin'
   // Admin call accepted from a token without a jti (transitional, admin.ts): label '<op>'.
   | '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 '<continent>/<colo>' (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.

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<boolean> {
+export async function admitPreAuth(env: Env, request: Request, kind: 'phone' | 'data' | 'health'): Promise<boolean> {
   if (!env.PREAUTH_LIMITER) { return true }
   try {
     const { success } = await env.PREAUTH_LIMITER.limit({ key: `${kind}:${sourceKey(request)}` })

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<Row>, before: Partial<Row> = {}, graph: Graph = {}) {
+  const posts: string[] = []
+  const graphql: { query: string; variables: Record<string, string> }[] = []
+  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'])
+  })
+})

cloud/apps/relay-edge/test/alerts-small-volume.test.ts

@@ -19,6 +19,10 @@ function stub(now: Partial<Row>, before: Partial<Row> = {}, 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 () => {

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<string, unknown>[] }
+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<string, Response> {
+  const store = new Map<string, Response>()
+  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)
+  })
+})

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<void> {
   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<void> {
   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<void> {
+  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)}`)
+}

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<void> {
@@ -110,6 +110,7 @@ async function main(): Promise<void> {
     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'))

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"
   }
 }