fix: retry transient relay broker startup

This commit is contained in:
OrcaWin
2026-07-23 02:13:16 -07:00
parent 7ab601487c
commit 2d56194767
3 changed files with 186 additions and 3 deletions
@@ -0,0 +1,114 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { RelayAuthCoordinator, type RelayAuthContext } from './relay-auth-coordinator'
import { RelayHttpError } from './relay-http-client'
const context: RelayAuthContext = {
identity: { userId: 'user-1', profileId: 'profile-1', organizationId: 'org-1' },
accessToken: 'access-1',
relayEntitled: true
}
afterEach(() => {
vi.useRealTimers()
})
describe('RelayAuthCoordinator transient recovery', () => {
it('retries a transient assignment failure and activates without an external event', async () => {
vi.useFakeTimers()
const broker = { closeNow: vi.fn() }
const openBroker = vi
.fn()
.mockRejectedValueOnce(new RelayHttpError('assignment', 500))
.mockResolvedValueOnce(broker)
const statuses: string[] = []
const coordinator = new RelayAuthCoordinator({
readContext: async () => context,
openBroker,
onStatus: (status) => statuses.push(status),
random: () => 0.5
})
coordinator.reconcile()
await vi.advanceTimersByTimeAsync(0)
expect(openBroker).toHaveBeenCalledOnce()
expect(statuses.at(-1)).toBe('offline')
await vi.advanceTimersByTimeAsync(501)
expect(openBroker).toHaveBeenCalledTimes(2)
expect(coordinator.getActiveBroker()).toBe(broker)
expect(statuses.at(-1)).toBe('registered')
})
it('does not retry a permanent authorization response', async () => {
vi.useFakeTimers()
const openBroker = vi.fn().mockRejectedValue(new RelayHttpError('token-exchange', 403))
const coordinator = new RelayAuthCoordinator({
readContext: async () => context,
openBroker,
onStatus: vi.fn(),
random: () => 0
})
coordinator.reconcile()
await vi.advanceTimersByTimeAsync(120_000)
expect(openBroker).toHaveBeenCalledOnce()
expect(coordinator.getActiveBroker()).toBeNull()
})
it('cancels a pending retry as soon as demand disappears', async () => {
vi.useFakeTimers()
let demanded = true
const openBroker = vi.fn().mockRejectedValue(new Error('temporary control open failure'))
const coordinator = new RelayAuthCoordinator({
readContext: async () => context,
hasDemand: () => demanded,
openBroker,
onStatus: vi.fn(),
random: () => 0.75
})
coordinator.reconcile()
await vi.advanceTimersByTimeAsync(0)
expect(openBroker).toHaveBeenCalledOnce()
demanded = false
coordinator.reconcile()
await vi.advanceTimersByTimeAsync(120_000)
expect(openBroker).toHaveBeenCalledOnce()
expect(coordinator.getActiveBroker()).toBeNull()
})
it('does not carry a pending retry across an identity switch', async () => {
vi.useFakeTimers()
let current = context
const broker = { closeNow: vi.fn() }
const openBroker = vi
.fn()
.mockRejectedValueOnce(new Error('temporary control open failure'))
.mockResolvedValueOnce(broker)
const coordinator = new RelayAuthCoordinator({
readContext: async () => current,
openBroker,
onStatus: vi.fn(),
random: () => 0.75
})
coordinator.reconcile()
await vi.advanceTimersByTimeAsync(0)
expect(openBroker).toHaveBeenCalledOnce()
current = {
...context,
identity: { ...context.identity, profileId: 'profile-2' }
}
coordinator.reconcile()
await vi.advanceTimersByTimeAsync(0)
expect(openBroker).toHaveBeenCalledTimes(2)
expect(coordinator.getActiveBroker()).toBe(broker)
await vi.advanceTimersByTimeAsync(120_000)
expect(openBroker).toHaveBeenCalledTimes(2)
})
})
@@ -1,4 +1,5 @@
import type { RelayBrokerStatus } from './relay-session-broker'
import { shouldRetryRelayConnectionError } from './relay-http-client'
export type RelayAuthIdentity = {
userId: string
@@ -26,6 +27,7 @@ type RelayAuthCoordinatorOptions = {
}) => Promise<CoordinatedRelayBroker>
onStatus: (status: RelayBrokerStatus) => void
lingerMs?: number
random?: () => number
}
type BrokerOwnership = {
@@ -39,12 +41,16 @@ function identityKey(identity: RelayAuthIdentity): string {
}
export class RelayAuthCoordinator {
private static readonly RETRY_BASE_MS = 1_000
private static readonly RETRY_MAX_MS = 60_000
private readonly options: RelayAuthCoordinatorOptions
private authEpoch = 0
private ownership: BrokerOwnership | null = null
private readonly pendingOwnerships = new Set<BrokerOwnership>()
private latestReconcile: Promise<void> = Promise.resolve()
private lingerTimer: ReturnType<typeof setTimeout> | null = null
private retryTimer: ReturnType<typeof setTimeout> | null = null
private retryAttempt = 0
private stopped = false
constructor(options: RelayAuthCoordinatorOptions) {
@@ -52,12 +58,20 @@ export class RelayAuthCoordinator {
}
reconcile(): void {
this.beginReconcile(true)
}
private beginReconcile(resetRetry: boolean, expectedIdentityKey?: string): void {
if (this.stopped) {
return
}
this.cancelRetry()
if (resetRetry) {
this.retryAttempt = 0
}
const epoch = ++this.authEpoch
this.invalidatePendingOwnerships()
const reconcile = this.reconcileEpoch(epoch)
const reconcile = this.reconcileEpoch(epoch, expectedIdentityKey)
this.latestReconcile = reconcile
void reconcile
}
@@ -65,6 +79,8 @@ export class RelayAuthCoordinator {
fenceAndCloseNow(): void {
++this.authEpoch
this.cancelLinger()
this.cancelRetry()
this.retryAttempt = 0
this.invalidatePendingOwnerships()
this.invalidateOwnership()
this.options.onStatus('offline')
@@ -94,7 +110,8 @@ export class RelayAuthCoordinator {
this.fenceAndCloseNow()
}
private async reconcileEpoch(epoch: number): Promise<void> {
private async reconcileEpoch(epoch: number, expectedIdentityKey?: string): Promise<void> {
let retryIdentityKey: string | null = null
try {
const context = await this.options.readContext()
if (!this.isEpochCurrent(epoch)) {
@@ -102,12 +119,19 @@ export class RelayAuthCoordinator {
}
if (!context || !context.relayEntitled) {
this.cancelLinger()
this.retryAttempt = 0
this.invalidateOwnership()
this.options.onStatus('offline')
return
}
const nextIdentityKey = identityKey(context.identity)
if (expectedIdentityKey && nextIdentityKey !== expectedIdentityKey) {
this.retryAttempt = 0
this.options.onStatus('offline')
return
}
if (!(this.options.hasDemand?.(context) ?? true)) {
this.retryAttempt = 0
if (this.ownership?.valid && this.ownership.identityKey !== nextIdentityKey) {
this.cancelLinger()
this.invalidateOwnership()
@@ -119,9 +143,11 @@ export class RelayAuthCoordinator {
}
this.cancelLinger()
if (this.ownership?.valid && this.ownership.identityKey === nextIdentityKey) {
this.retryAttempt = 0
this.options.onStatus('registered')
return
}
retryIdentityKey = nextIdentityKey
this.invalidateOwnership()
this.options.onStatus('connecting')
const ownership: BrokerOwnership = {
@@ -150,14 +176,38 @@ export class RelayAuthCoordinator {
return
}
this.ownership = ownership
this.retryAttempt = 0
this.options.onStatus('registered')
} catch {
} catch (error) {
if (this.isEpochCurrent(epoch)) {
this.options.onStatus('offline')
if (retryIdentityKey && shouldRetryRelayConnectionError(error)) {
this.scheduleRetry(epoch, retryIdentityKey)
}
}
}
}
private scheduleRetry(epoch: number, expectedIdentityKey: string): void {
if (this.retryTimer || !this.isEpochCurrent(epoch)) {
return
}
const capMs = Math.min(
RelayAuthCoordinator.RETRY_MAX_MS,
RelayAuthCoordinator.RETRY_BASE_MS * 2 ** Math.min(this.retryAttempt, 6)
)
this.retryAttempt++
const random = this.options.random ?? Math.random
const delayMs = Math.floor(random() * (capMs + 1))
this.retryTimer = setTimeout(() => {
this.retryTimer = null
if (this.isEpochCurrent(epoch)) {
// Retry still re-reads entitlement and demand; the timer grants no authority.
this.beginReconcile(false, expectedIdentityKey)
}
}, delayMs)
}
private async refreshAccessToken(
ownership: { valid: boolean },
expectedIdentityKey: string
@@ -212,6 +262,13 @@ export class RelayAuthCoordinator {
}
}
private cancelRetry(): void {
if (this.retryTimer) {
clearTimeout(this.retryTimer)
this.retryTimer = null
}
}
private invalidatePendingOwnerships(): void {
for (const ownership of this.pendingOwnerships) {
ownership.valid = false
@@ -37,6 +37,18 @@ export class RelayHttpError extends Error {
}
}
export function shouldRetryRelayConnectionError(error: unknown): boolean {
if (!(error instanceof RelayHttpError)) {
return true
}
return (
error.statusCode >= 500 ||
error.statusCode === 408 ||
error.statusCode === 425 ||
error.statusCode === 429
)
}
export function deriveRelayHostId(publicKey: Uint8Array): string {
return createHash('sha256').update(publicKey).digest('base64url').slice(0, 16)
}