fix(relay): back off overloaded assignments (#10894)

Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>
2026-07-27 02:17:24 -07:00 · 2026-07-27 02:17:24 -07:00 · c53a12e11d
parent 3830851a83
commit c53a12e11d
8 changed files with 365 additions and 14 deletions
--- a/config/reliability-gates.jsonc
+++ b/config/reliability-gates.jsonc
@ -1,6 +1,6 @@
 {
  "schemaVersion": 1,
-  "updatedAt": "2026-07-25",
+  "updatedAt": "2026-07-26",
  "policy": {
    "maturityLevels": [
      "experimental",
@ -141,6 +141,108 @@
      ],
      "demotionRule": "Keep experimental or demote if focused tests flake, HOST_OFFLINE can park indefinitely, direct probes serialize configured endpoints, loser cleanup leaks clients, or physical-device radio cost exceeds the measured budget."
    },
+    {
+      "id": "desktop-relay.assignment-backpressure",
+      "title": "Desktop relay drain recovery cannot amplify a director outage",
+      "maturity": "experimental",
+      "protection": "partial",
+      "owner": "desktop-runtime",
+      "layer": "main-relay-state-machine",
+      "surfaces": [
+        "desktop relay drain recovery",
+        "director assignment overload",
+        "relay broker shutdown"
+      ],
+      "platforms": [
+        "macos",
+        "linux",
+        "windows"
+      ],
+      "providers": [
+        "cloud-relay"
+      ],
+      "coveredPlatforms": [
+        "macos"
+      ],
+      "coveredProviders": [
+        "cloud-relay"
+      ],
+      "coverageNotes": "Deterministic main-process tests cover duplicate drain notifications, full-jitter backoff, Retry-After during initial setup and drain recovery, successful recovery, and broker-close cleanup. Packaged desktop, mixed-version fleets, GFE, and production Cloud SQL remain live-test gaps.",
+      "motivatingLinks": [
+        "https://github.com/stablyai/orca-cloud/actions/runs/30223521062"
+      ],
+      "invariant": "One relay host may have at most one assignment attempt or retry timer per recovery path. Sustained director failure must increase the retry window up to five minutes, a bounded Retry-After must be respected during initial setup and drain recovery, shutdown must cancel pending work, and recovery must activate the authoritative assigned origin.",
+      "oracle": "Inject duplicate drain events, deterministic randomness, fake time, repeated assignment failures, a 30-second Retry-After during initial setup and drain recovery, broker close, and eventual director recovery. Count every assignment call, require 500 ms then 1,000 ms retry windows, reject duplicate fanout, require no pre-hint retry or post-close work, and prove the recovered cell becomes authoritative.",
+      "commands": [
+        "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/relay/relay-session-broker.test.ts src/main/runtime/relay/relay-http-client.test.ts src/main/runtime/relay/relay-auth-coordinator-recovery.test.ts --reporter=dot"
+      ],
+      "testFiles": [
+        "src/main/runtime/relay/relay-session-broker.test.ts",
+        "src/main/runtime/relay/relay-http-client.test.ts",
+        "src/main/runtime/relay/relay-auth-coordinator-recovery.test.ts"
+      ],
+      "assertionRefs": [
+        {
+          "file": "src/main/runtime/relay/relay-session-broker.test.ts",
+          "assertions": [
+            "duplicate drain notifications share one exponentially backed-off retry schedule",
+            "Retry-After suppresses early assignment requests",
+            "broker close prevents retry resurrection",
+            "a later successful assignment activates the new origin"
+          ]
+        },
+        {
+          "file": "src/main/runtime/relay/relay-http-client.test.ts",
+          "assertions": [
+            "assignment overload preserves a bounded Retry-After hint"
+          ]
+        },
+        {
+          "file": "src/main/runtime/relay/relay-auth-coordinator-recovery.test.ts",
+          "assertions": [
+            "initial relay setup does not retry before Retry-After expires"
+          ]
+        }
+      ],
+      "evidenceRuns": [
+        {
+          "date": "2026-07-26",
+          "runner": "local",
+          "platform": "macos",
+          "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/relay/relay-session-broker.test.ts src/main/runtime/relay/relay-http-client.test.ts src/main/runtime/relay/relay-auth-coordinator-recovery.test.ts --reporter=dot",
+          "result": "passed",
+          "durationSeconds": 0.49,
+          "summary": "Three focused relay files passed with 25 assertions."
+        }
+      ],
+      "runtimeBudget": {
+        "p95Seconds": 5,
+        "scope": "focused desktop relay state-machine tests"
+      },
+      "flakeHistory": {
+        "status": "unknown",
+        "evidence": "One deterministic local run exists; CI and soak history are not yet available."
+      },
+      "redGreenEvidence": {
+        "status": "complete",
+        "evidence": "The prior fixed-delay implementation issued a duplicate assignment within 499 ms and ignored Retry-After, while the candidate passes the byte-identical timer and call-count oracle."
+      },
+      "performanceBudget": {
+        "required": true,
+        "evidence": "One host retains at most one assignment attempt or retry timer, retry windows grow to a five-minute cap, duplicate drain events add no calls, and close leaves no timer-driven work."
+      },
+      "promotionCriteria": [
+        "Collect 100 consecutive CI passes or 14 days of soak history.",
+        "Run a mixed-version load test with at least the incident-scale desktop population.",
+        "Verify production director request rate decays during an injected assignment outage."
+      ],
+      "knownGaps": [
+        "No packaged desktop or physical phone was exercised.",
+        "The deterministic seam does not measure production GFE, carrier NAT, DNS, TLS, or Cloud SQL behavior.",
+        "Legacy desktop versions remain dependent on server-side overload protection."
+      ],
+      "demotionRule": "Keep experimental or demote if assignment calls overlap, duplicate drain events bypass backoff, Retry-After is ignored, close resurrects work, or mixed-version request rate exceeds the reviewed director budget."
+    },
    {
      "id": "git-worktree.refresh-event-semantics",
      "title": "Index-only Git metadata cannot trigger structural worktree refresh fanout",
--- a/src/main/runtime/relay/relay-auth-coordinator-recovery.test.ts
+++ b/src/main/runtime/relay/relay-auth-coordinator-recovery.test.ts
@ -39,6 +39,28 @@ describe('RelayAuthCoordinator transient recovery', () => {
    expect(statuses.at(-1)).toBe('registered')
  })

+  it('does not retry initial relay setup before the server Retry-After window', async () => {
+    vi.useFakeTimers()
+    const broker = { closeNow: vi.fn() }
+    const openBroker = vi
+      .fn()
+      .mockRejectedValueOnce(new RelayHttpError('assignment', 503, 30_000))
+      .mockResolvedValueOnce(broker)
+    const coordinator = new RelayAuthCoordinator({
+      readContext: async () => context,
+      openBroker,
+      onStatus: vi.fn(),
+      random: () => 0.5
+    })
+
+    coordinator.reconcile()
+    await vi.advanceTimersByTimeAsync(29_999)
+    expect(openBroker).toHaveBeenCalledOnce()
+    await vi.advanceTimersByTimeAsync(1)
+    expect(openBroker).toHaveBeenCalledTimes(2)
+    expect(coordinator.getActiveBroker()).toBe(broker)
+  })
+
  it('retries when cloud-session refresh fails before identity can be read', async () => {
    vi.useFakeTimers()
    const broker = { closeNow: vi.fn() }
--- a/src/main/runtime/relay/relay-auth-coordinator.ts
+++ b/src/main/runtime/relay/relay-auth-coordinator.ts
@ -1,5 +1,5 @@
 import type { RelayBrokerStatus } from './relay-session-broker'
-import { shouldRetryRelayConnectionError } from './relay-http-client'
+import { RelayHttpError, shouldRetryRelayConnectionError } from './relay-http-client'

 export type RelayAuthIdentity = {
  userId: string
@ -183,13 +183,14 @@ export class RelayAuthCoordinator {
      if (this.isEpochCurrent(epoch)) {
        this.options.onStatus('offline')
        if (shouldRetryRelayConnectionError(error)) {
-          this.scheduleRetry(epoch, retryIdentityKey)
+          const retryAfterMs = error instanceof RelayHttpError ? (error.retryAfterMs ?? 0) : 0
+          this.scheduleRetry(epoch, retryIdentityKey, retryAfterMs)
        }
      }
    }
  }

-  private scheduleRetry(epoch: number, expectedIdentityKey?: string): void {
+  private scheduleRetry(epoch: number, expectedIdentityKey?: string, retryAfterMs = 0): void {
    if (this.retryTimer || !this.isEpochCurrent(epoch)) {
      return
    }
@ -203,7 +204,7 @@ export class RelayAuthCoordinator {
    )
    this.retryAttempt++
    const random = this.options.random ?? Math.random
-    const delayMs = Math.floor(random() * (capMs + 1))
+    const delayMs = Math.max(Math.floor(random() * (capMs + 1)), retryAfterMs)
    this.retryTimer = setTimeout(() => {
      this.retryTimer = null
      if (this.isEpochCurrent(epoch)) {
--- a/src/main/runtime/relay/relay-drain-retry-schedule.ts
+++ b/src/main/runtime/relay/relay-drain-retry-schedule.ts
@ -0,0 +1,42 @@
+const RETRY_BASE_MS = 1_000
+const RETRY_MAX_MS = 5 * 60_000
+
+export class RelayDrainRetrySchedule {
+  private timer: ReturnType<typeof setTimeout> | null = null
+  private attempt = 0
+
+  constructor(private readonly random: () => number = Math.random) {}
+
+  get pending(): boolean {
+    return this.timer !== null
+  }
+
+  schedule(retryAfterMs: number, retry: () => void): void {
+    if (this.timer) {
+      return
+    }
+    const exponent = Math.min(this.attempt, Math.ceil(Math.log2(RETRY_MAX_MS / RETRY_BASE_MS)))
+    const capMs = Math.min(RETRY_MAX_MS, RETRY_BASE_MS * 2 ** exponent)
+    this.attempt++
+    const jitterMs = Math.floor(this.random() * (capMs + 1))
+    this.timer = setTimeout(
+      () => {
+        this.timer = null
+        retry()
+      },
+      Math.max(jitterMs, retryAfterMs)
+    )
+  }
+
+  reset(): void {
+    this.attempt = 0
+  }
+
+  cancel(): void {
+    if (this.timer) {
+      clearTimeout(this.timer)
+      this.timer = null
+    }
+    this.reset()
+  }
+}
--- a/src/main/runtime/relay/relay-http-client.test.ts
+++ b/src/main/runtime/relay/relay-http-client.test.ts
@ -147,4 +147,38 @@ describe('relay HTTP client', () => {
    ).rejects.toThrow()
    expect(cancelledBodies).toBe(2)
  })
+
+  it('preserves a bounded Retry-After hint on assignment overload', async () => {
+    const fetch = vi.fn<typeof globalThis.fetch>(
+      async () => new Response(null, { status: 503, headers: { 'retry-after': '30' } })
+    )
+
+    await expect(
+      requestRelayAssignment({
+        directorUrl: 'https://relay.example',
+        relayToken: 'scoped-token',
+        relayHostId: 'AbCdEf0123_-xyZ9',
+        fetch
+      })
+    ).rejects.toMatchObject({
+      operation: 'assignment',
+      statusCode: 503,
+      retryAfterMs: 30_000
+    })
+  })
+
+  it('caps an excessive Retry-After hint at five minutes', async () => {
+    const fetch = vi.fn<typeof globalThis.fetch>(
+      async () => new Response(null, { status: 503, headers: { 'retry-after': '999999' } })
+    )
+
+    await expect(
+      requestRelayAssignment({
+        directorUrl: 'https://relay.example',
+        relayToken: 'scoped-token',
+        relayHostId: 'AbCdEf0123_-xyZ9',
+        fetch
+      })
+    ).rejects.toMatchObject({ retryAfterMs: 5 * 60_000 })
+  })
 })
--- a/src/main/runtime/relay/relay-http-client.ts
+++ b/src/main/runtime/relay/relay-http-client.ts
@ -4,6 +4,7 @@ import type { E2EEKeypair } from '../e2ee-keypair'
 import { cancelUnreadResponseBody } from '../../lib/unread-response-body'

 const RELAY_HTTP_REQUEST_DEADLINE_MS = 15_000
+const RELAY_RETRY_AFTER_MAX_MS = 5 * 60_000

 const RelayTokenResponseSchema = z
  .object({
@ -33,12 +34,25 @@ export type RelayAssignment = z.infer<typeof AssignmentResponseSchema>
 export class RelayHttpError extends Error {
  constructor(
    readonly operation: 'token-exchange' | 'assignment',
-    readonly statusCode: number
+    readonly statusCode: number,
+    readonly retryAfterMs: number | null = null
  ) {
    super(`relay_${operation}_failed_${statusCode}`)
  }
 }

+function relayRetryAfterMs(value: string | null, nowMs = Date.now()): number | null {
+  if (!value) {
+    return null
+  }
+  const seconds = Number(value)
+  const delayMs = Number.isFinite(seconds) ? seconds * 1_000 : Date.parse(value) - nowMs
+  if (!Number.isFinite(delayMs) || delayMs <= 0) {
+    return null
+  }
+  return Math.min(RELAY_RETRY_AFTER_MAX_MS, Math.ceil(delayMs))
+}
+
 export function shouldRetryRelayConnectionError(error: unknown): boolean {
  if (!(error instanceof RelayHttpError)) {
    return true
@ -87,8 +101,9 @@ export async function exchangeRelayAuthorization(input: {
    body: JSON.stringify({ relayHostId, hostPublicKeyB64: input.keypair.publicKeyB64 })
  })
  if (!response.ok) {
+    const retryAfterMs = relayRetryAfterMs(response.headers.get('retry-after'))
    await cancelUnreadResponseBody(response)
-    throw new RelayHttpError('token-exchange', response.status)
+    throw new RelayHttpError('token-exchange', response.status, retryAfterMs)
  }
  const parsed = RelayTokenResponseSchema.safeParse(await response.json())
  if (!parsed.success) {
@ -117,8 +132,9 @@ export async function requestRelayAssignment(input: {
    body: JSON.stringify({ v: 1, relayHostId: input.relayHostId })
  })
  if (!response.ok) {
+    const retryAfterMs = relayRetryAfterMs(response.headers.get('retry-after'))
    await cancelUnreadResponseBody(response)
-    throw new RelayHttpError('assignment', response.status)
+    throw new RelayHttpError('assignment', response.status, retryAfterMs)
  }
  const parsed = AssignmentResponseSchema.safeParse(await response.json())
  if (!parsed.success || !isAllowedRelayOrigin(parsed.data.cellUrl)) {
--- a/src/main/runtime/relay/relay-origin-pool.ts
+++ b/src/main/runtime/relay/relay-origin-pool.ts
@ -4,7 +4,8 @@ import type { MobileSocketWiring } from '../rpc/mobile-socket-wiring'
 import { RelayControlOrigin } from './relay-control-origin'
 import type { RelayControlClient } from './relay-control-client'
 import type { RelayDrainMessage } from './relay-control-protocol'
-import { requestRelayAssignment, type RelayAssignment } from './relay-http-client'
+import { RelayDrainRetrySchedule } from './relay-drain-retry-schedule'
+import { RelayHttpError, requestRelayAssignment, type RelayAssignment } from './relay-http-client'
 import type { RelayBrokerStatus, RelayIdentity } from './relay-session-broker-contract'

 type RelayOriginPoolOptions = {
@ -34,10 +35,12 @@ export class RelayOriginPool {
  private relayJwt: string | null = null
  private rotationTimer: ReturnType<typeof setTimeout> | null = null
  private rotationPromise: Promise<void> | null = null
+  private readonly drainRetry: RelayDrainRetrySchedule
  private closed = false

  constructor(options: RelayOriginPoolOptions) {
    this.options = options
+    this.drainRetry = new RelayDrainRetrySchedule(options.random)
  }

  get activeAssignment(): RelayAssignment | null {
@ -79,6 +82,7 @@ export class RelayOriginPool {
      clearTimeout(this.rotationTimer)
      this.rotationTimer = null
    }
+    this.drainRetry.cancel()
    for (const timer of this.drainTimers.values()) {
      clearTimeout(timer)
    }
@ -135,7 +139,7 @@ export class RelayOriginPool {
    origin.markDraining()
    this.drainingOrigins.add(origin)
    this.options.onStatus('draining')
-    if (!this.rotationPromise) {
+    if (!this.rotationPromise && !this.drainRetry.pending) {
      this.rotationPromise = this.resolveDrainTarget(origin, message).finally(() => {
        this.rotationPromise = null
      })
@ -178,11 +182,12 @@ export class RelayOriginPool {
        await this.activateTarget(origin, assignment, this.relayJwt, message.graceMs)
      }
      this.options.onStatus('registered')
+      this.drainRetry.reset()
      this.scheduleControlRotation()
-    } catch {
-      if (this.isCurrent()) {
-        const random = this.options.random ?? Math.random
-        setTimeout(() => this.handleDrain(origin, message), 250 + Math.floor(random() * 751))
+    } catch (error) {
+      if (this.isCurrent() && origin === this.activeOrigin) {
+        const retryAfterMs = error instanceof RelayHttpError ? (error.retryAfterMs ?? 0) : 0
+        this.drainRetry.schedule(retryAfterMs, () => this.handleDrain(origin, message))
      }
    }
  }
--- a/src/main/runtime/relay/relay-session-broker.test.ts
+++ b/src/main/runtime/relay/relay-session-broker.test.ts
@ -87,6 +87,7 @@ vi.mock('../rpc/relay-transport', () => ({
 }))

 import { RelaySessionBroker, StaleRelayBrokerError } from './relay-session-broker'
+import { RelayHttpError } from './relay-http-client'

 function deferred<T>() {
  let resolve!: (value: T) => void
@ -299,6 +300,134 @@ describe('RelaySessionBroker lifecycle ownership', () => {
    await vi.waitFor(() => expect(onStatus).toHaveBeenLastCalledWith('registered'))
    expect(broker.endpoint?.cellUrl).toBe('https://relay.example.test')
  })
+
+  it('backs off drain resolution failures without duplicate retries or post-close work', async () => {
+    vi.useFakeTimers()
+    try {
+      const ack: RelayHostHelloAckMessage = {
+        type: 'host-hello-ack',
+        v: 1,
+        generation: 1,
+        controlResumeSecret: 'R'.repeat(43),
+        leaseExpiresAt: 1_000_000,
+        activeConnIds: [],
+        pendingConns: []
+      }
+      fakes.controlConnect.mockResolvedValue(ack)
+      fakes.assign
+        .mockResolvedValueOnce({
+          cellUrl: 'https://relay.example.test',
+          assignmentEpoch: 1,
+          leaseExpiresAt: 1_000_000
+        })
+        .mockRejectedValue(new Error('director_unavailable'))
+      const broker = await RelaySessionBroker.connect(brokerOptions({ random: () => 0.5 }))
+      const drain = {
+        type: 'drain' as const,
+        graceMs: 5_000,
+        recovery: 'resolve-director' as const
+      }
+
+      fakes.controls[0]!.options.onDrain(drain)
+      await vi.advanceTimersByTimeAsync(0)
+      expect(fakes.assign).toHaveBeenCalledTimes(2)
+      fakes.controls[0]!.options.onDrain(drain)
+      await vi.advanceTimersByTimeAsync(499)
+      expect(fakes.assign).toHaveBeenCalledTimes(2)
+      await vi.advanceTimersByTimeAsync(1)
+      expect(fakes.assign).toHaveBeenCalledTimes(3)
+      await vi.advanceTimersByTimeAsync(999)
+      expect(fakes.assign).toHaveBeenCalledTimes(3)
+      await vi.advanceTimersByTimeAsync(1)
+      expect(fakes.assign).toHaveBeenCalledTimes(4)
+
+      broker.closeNow()
+      await vi.advanceTimersByTimeAsync(5 * 60_000)
+      expect(fakes.assign).toHaveBeenCalledTimes(4)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('does not retry drain resolution before the director Retry-After window', async () => {
+    vi.useFakeTimers()
+    try {
+      const ack: RelayHostHelloAckMessage = {
+        type: 'host-hello-ack',
+        v: 1,
+        generation: 1,
+        controlResumeSecret: 'R'.repeat(43),
+        leaseExpiresAt: 1_000_000,
+        activeConnIds: [],
+        pendingConns: []
+      }
+      fakes.controlConnect.mockResolvedValue(ack)
+      fakes.assign
+        .mockResolvedValueOnce({
+          cellUrl: 'https://relay.example.test',
+          assignmentEpoch: 1,
+          leaseExpiresAt: 1_000_000
+        })
+        .mockRejectedValue(new RelayHttpError('assignment', 503, 30_000))
+      const broker = await RelaySessionBroker.connect(brokerOptions({ random: () => 0.5 }))
+
+      fakes.controls[0]!.options.onDrain({
+        type: 'drain',
+        graceMs: 5_000,
+        recovery: 'resolve-director'
+      })
+      await vi.advanceTimersByTimeAsync(29_999)
+      expect(fakes.assign).toHaveBeenCalledTimes(2)
+      await vi.advanceTimersByTimeAsync(1)
+      expect(fakes.assign).toHaveBeenCalledTimes(3)
+      broker.closeNow()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('recovers through a new origin after the director failure clears', async () => {
+    vi.useFakeTimers()
+    try {
+      const ack: RelayHostHelloAckMessage = {
+        type: 'host-hello-ack',
+        v: 1,
+        generation: 1,
+        controlResumeSecret: 'R'.repeat(43),
+        leaseExpiresAt: 1_000_000,
+        activeConnIds: [],
+        pendingConns: []
+      }
+      fakes.controlConnect.mockResolvedValue(ack)
+      fakes.assign
+        .mockResolvedValueOnce({
+          cellUrl: 'https://relay-c1.example.test',
+          assignmentEpoch: 1,
+          leaseExpiresAt: 1_000_000
+        })
+        .mockRejectedValueOnce(new Error('director_unavailable'))
+        .mockResolvedValueOnce({
+          cellUrl: 'https://relay-c2.example.test',
+          assignmentEpoch: 2,
+          leaseExpiresAt: 2_000_000
+        })
+      const broker = await RelaySessionBroker.connect(brokerOptions({ random: () => 0.5 }))
+
+      fakes.controls[0]!.options.onDrain({
+        type: 'drain',
+        graceMs: 5_000,
+        recovery: 'resolve-director'
+      })
+      await vi.advanceTimersByTimeAsync(499)
+      expect(broker.endpoint?.cellUrl).toBe('https://relay-c1.example.test')
+      await vi.advanceTimersByTimeAsync(1)
+      expect(broker.endpoint?.cellUrl).toBe('https://relay-c2.example.test')
+      expect(fakes.assign).toHaveBeenCalledTimes(3)
+      broker.closeNow()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
 })

 function brokerBasisIds(broker: RelaySessionBroker): string[] {