diff --git a/src/main/runtime/relay/relay-auth-coordinator-contract.ts b/src/main/runtime/relay/relay-auth-coordinator-contract.ts index 2e31ef899ebb..ed15aad3b890 100644 --- a/src/main/runtime/relay/relay-auth-coordinator-contract.ts +++ b/src/main/runtime/relay/relay-auth-coordinator-contract.ts @@ -33,6 +33,12 @@ export type RelayAuthCoordinatorOptions = { refreshAccessToken: () => Promise }) => Promise onStatus: (status: RelayBrokerStatus, cellUrl?: string) => void + /** + * Test seam, not a policy knob. The only production construction is in `desktop-relay-service.ts` + * and it passes neither this nor `random`, so every shipped build lingers for the hardcoded + * ten-minute default in `relay-auth-coordinator.ts`. Read it as such before tuning it: changing + * the default is the only thing a user can feel. + */ lingerMs?: number random?: () => number } diff --git a/src/main/runtime/relay/relay-auth-coordinator-wait-budget.test.ts b/src/main/runtime/relay/relay-auth-coordinator-wait-budget.test.ts new file mode 100644 index 000000000000..e46431f42f66 --- /dev/null +++ b/src/main/runtime/relay/relay-auth-coordinator-wait-budget.test.ts @@ -0,0 +1,57 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { RelayAuthCoordinator, type RelayAuthContext } from './relay-auth-coordinator' + +const context: RelayAuthContext = { + identity: { userId: 'user-1', profileId: 'profile-1', organizationId: 'org-1' }, + accessToken: 'access-1', + relayEntitled: true +} + +/** + * What the live-broker wait budget does and does not bound. + * + * `LIVE_BROKER_WAIT_BUDGET_MS` is 20s and the deadline is checked only AFTER `await pending`, which + * is deliberately unbounded — cutting a slow-but-succeeding open short would fail a pairing that + * was about to work. So the budget bounds the armed-retry chain and nothing else. + * + * That is worth a test rather than a comment because the consequence lands on the phone: + * `pairing.provisionRelay` reaches this through `requireActiveBroker`, and `readContext`'s own + * ceiling is the cloud refresh timeout (60s, with one retry for a definitive 5xx) — several times + * the phone's request budget. The phone gives up and retries while the desktop is still holding a + * transient demand ref for the call it abandoned. + */ +describe('live-broker wait budget', () => { + afterEach(() => { + vi.useRealTimers() + }) + + it('does not bound a reconcile already in flight, so it can outlast the phone request budget', async () => { + vi.useFakeTimers() + let releaseContext = (): void => {} + const coordinator = new RelayAuthCoordinator({ + readContext: () => + new Promise((resolve) => { + releaseContext = (): void => resolve(context) + }), + openBroker: async () => ({ closeNow: vi.fn() }), + onStatus: vi.fn() + }) + coordinator.reconcile() + await Promise.resolve() + + let settled = false + const wait = coordinator.waitForLiveBrokerResult().then((result) => { + settled = true + return result + }) + + // Well past both the 20s budget and the phone's 30s request budget. + await vi.advanceTimersByTimeAsync(45_000) + expect(settled).toBe(false) + + releaseContext() + await vi.advanceTimersByTimeAsync(0) + await expect(wait).resolves.toEqual({ broker: expect.anything() }) + expect(settled).toBe(true) + }) +}) diff --git a/src/main/runtime/relay/relay-auth-coordinator.ts b/src/main/runtime/relay/relay-auth-coordinator.ts index a284618ac720..b6ab1df3705a 100644 --- a/src/main/runtime/relay/relay-auth-coordinator.ts +++ b/src/main/runtime/relay/relay-auth-coordinator.ts @@ -7,6 +7,7 @@ import type { RelayBrokerStatus } from './relay-session-broker' import { RelayHttpError, shouldRetryRelayConnectionError } from './relay-http-client' import { relayOfflineReasonForOpenFailure, type RelayOfflineReason } from './relay-offline-reason' import { RelayRetrySchedule } from './relay-retry-schedule' +import { RelayAuthority } from './relay-authority' import { withTimeout } from '../../../shared/promise-timeout-fallback' import { relayAuthIdentityKey as identityKey, @@ -24,6 +25,12 @@ export type { RelayAuthIdentity } from './relay-auth-coordinator-contract' +// Bounds only how long a waiter sits through superseded opens and armed retries, never the open +// it arrived on. It does NOT bound the call: a reconcile's own ceiling is readContext's cloud +// refresh (60s) plus the broker open, so a wait can outlive this. relay-auth-coordinator-wait- +// budget.test.ts pins that so the claim cannot drift back. +export const LIVE_BROKER_WAIT_BUDGET_MS = 20_000 + type BrokerOwnership = { identityKey: string broker: CoordinatedRelayBroker | null @@ -31,17 +38,13 @@ type BrokerOwnership = { } export class RelayAuthCoordinator { - // Why 20s: bounds only how long a waiter sits through armed retries, never - // an open already in flight. It spans the first few rungs of the backoff - // ladder and stays inside the phone's 30s request budget, so a sustained - // outage fails the caller with its cause instead of parking the demand ref. - private static readonly LIVE_BROKER_WAIT_BUDGET_MS = 20_000 private readonly options: RelayAuthCoordinatorOptions - private authEpoch = 0 private offlineReason: RelayOfflineReason | null = null private ownership: BrokerOwnership | null = null private readonly pendingOwnerships = new Set() - private latestReconcile: Promise = Promise.resolve() + // The reconcile whose result will be kept. Everything older is discarded, so a waiter parked on + // a superseded one would be held behind an open nobody will use. + private authority = RelayAuthority.idle() private lingerTimer: ReturnType | null = null private readonly retry: RelayRetrySchedule private stopped = false @@ -63,26 +66,31 @@ export class RelayAuthCoordinator { if (this.stopped) { return } - this.retry.cancel() if (resetRetry) { this.retry.reset() } - const epoch = ++this.authEpoch - this.invalidatePendingOwnerships() - const reconcile = this.reconcileEpoch(epoch, expectedIdentityKey, options) - this.latestReconcile = reconcile - void reconcile + this.turnover((authority) => this.reconcileUnder(authority, expectedIdentityKey, options)) + } + + // The only way the authority changes hands, so ending the outgoing one — which is what releases + // every waiter parked on it — cannot be skipped. `run` absent means nothing is running. + private turnover(run?: (authority: RelayAuthority) => Promise): void { + this.retry.cancel() + for (const ownership of this.pendingOwnerships) { + ownership.valid = false + } + this.pendingOwnerships.clear() + this.authority.end() + this.authority = run ? RelayAuthority.running(run) : RelayAuthority.idle() } // hostCloseReason names an auth loss the phone should be told about. Quit, // relaunch and every other fence pass nothing, so the control socket dies // abruptly exactly as before and the cell records no cause. fenceAndCloseNow(hostCloseReason?: RelayHostCloseReason): void { - ++this.authEpoch this.cancelLinger() - this.retry.cancel() this.retry.reset() - this.invalidatePendingOwnerships() + this.turnover() this.invalidateOwnership(hostCloseReason) this.publish('offline', hostCloseReason) } @@ -136,36 +144,33 @@ export class RelayAuthCoordinator { // Why the caller maps offlineReason to a code instead of the coordinator // throwing: the mint-failure vocabulary belongs to the RPC layer. async waitForLiveBrokerResult( - budgetMs = RelayAuthCoordinator.LIVE_BROKER_WAIT_BUDGET_MS + budgetMs = LIVE_BROKER_WAIT_BUDGET_MS ): Promise { const deadline = Date.now() + budgetMs + let joined = false while (!this.stopped) { const broker = this.getLiveBroker() if (broker) { return { broker } } - const pending = this.latestReconcile - // Why unbounded: a reconcile always settles (opens carry HTTP deadlines), - // and cutting a slow-but-succeeding open short would fail a pairing that - // was about to work. The budget bounds only the retry chain below. - await pending - if (pending !== this.latestReconcile) { + // The budget bounds only superseded and retried work; the open the waiter arrived on is + // never cut short, because cutting a slow-but-succeeding one fails a pairing about to work. + if (joined && Date.now() >= deadline) { + break + } + const authority = this.authority + joined = true + await authority.settled + if (authority !== this.authority) { continue } - // Why: a reconcile that failed transiently has already armed its own - // retry; returning now would surface a hiccup fixed moments later. A - // terminal outcome (signed out, unentitled, rejected) arms nothing, so - // its cause returns without waiting. - const armed = this.retry.settled - if (!armed || Date.now() >= deadline) { - return this.settledLiveBrokerResult() + // A transient failure has armed its own retry, so waiting is worthwhile. A terminal outcome + // (signed out, unentitled, rejected) arms nothing, so its cause returns now. + if (!this.retry.pending || Date.now() >= deadline) { + break } - await withTimeout(armed, deadline - Date.now(), undefined) + await withTimeout(authority.ended, deadline - Date.now(), undefined) } - return this.settledLiveBrokerResult() - } - - private settledLiveBrokerResult(): LiveBrokerWaitResult { const broker = this.getLiveBroker() return broker ? { broker } : { broker: null, offlineReason: this.offlineReason } } @@ -175,15 +180,15 @@ export class RelayAuthCoordinator { this.fenceAndCloseNow() } - private async reconcileEpoch( - epoch: number, + private async reconcileUnder( + authority: RelayAuthority, expectedIdentityKey?: string, options?: ReconcileOptions ): Promise { let retryIdentityKey: string | undefined try { const context = await this.options.readContext() - if (!this.isEpochCurrent(epoch)) { + if (!this.isCurrent(authority)) { return } if (!context || !context.relayEntitled) { @@ -241,7 +246,7 @@ export class RelayAuthCoordinator { const isCurrent = (): boolean => ownership.valid && !this.stopped && - (ownership.broker ? this.ownership === ownership : this.isEpochCurrent(epoch)) + (ownership.broker ? this.ownership === ownership : this.isCurrent(authority)) let broker: CoordinatedRelayBroker try { broker = await this.options.openBroker({ @@ -253,7 +258,7 @@ export class RelayAuthCoordinator { this.pendingOwnerships.delete(ownership) } ownership.broker = broker - if (!this.isEpochCurrent(epoch) || !ownership.valid) { + if (!this.isCurrent(authority) || !ownership.valid) { broker.closeNow() return } @@ -261,7 +266,7 @@ export class RelayAuthCoordinator { this.retry.reset() this.publish('registered') } catch (error) { - if (this.isEpochCurrent(epoch)) { + if (this.isCurrent(authority)) { // Why: silent broker-open failures made a dead relay look like standby // during incident diagnosis; the message carries operation + status. console.warn( @@ -272,7 +277,7 @@ export class RelayAuthCoordinator { this.publish('offline', relayOfflineReasonForOpenFailure(retryable, retryIdentityKey)) if (retryable) { const retryAfterMs = error instanceof RelayHttpError ? (error.retryAfterMs ?? 0) : 0 - this.scheduleRetry(epoch, retryIdentityKey, retryAfterMs, options) + this.scheduleRetry(authority, retryIdentityKey, retryAfterMs, options) } } } @@ -281,16 +286,16 @@ export class RelayAuthCoordinator { // options ride along so a policy-change reconcile that failed transiently // still skips the linger when its retry finally observes no demand. private scheduleRetry( - epoch: number, + authority: RelayAuthority, expectedIdentityKey: string | undefined, retryAfterMs: number, options: ReconcileOptions | undefined ): void { - if (!this.isEpochCurrent(epoch)) { + if (!this.isCurrent(authority)) { return } this.retry.schedule(retryAfterMs, () => { - if (this.isEpochCurrent(epoch)) { + if (this.isCurrent(authority)) { // Retry still re-reads entitlement and demand; the timer grants no authority. this.beginReconcile(false, expectedIdentityKey, options) } @@ -304,11 +309,11 @@ export class RelayAuthCoordinator { if (!ownership.valid || this.stopped) { return null } - const epoch = this.authEpoch + const authority = this.authority const context = await this.options.readContext() if ( !ownership.valid || - !this.isEpochCurrent(epoch) || + !this.isCurrent(authority) || !context?.relayEntitled || identityKey(context.identity) !== expectedIdentityKey ) { @@ -351,14 +356,7 @@ export class RelayAuthCoordinator { } } - private invalidatePendingOwnerships(): void { - for (const ownership of this.pendingOwnerships) { - ownership.valid = false - } - this.pendingOwnerships.clear() - } - - private isEpochCurrent(epoch: number): boolean { - return !this.stopped && this.authEpoch === epoch + private isCurrent(authority: RelayAuthority): boolean { + return !this.stopped && this.authority === authority } } diff --git a/src/main/runtime/relay/relay-authority.ts b/src/main/runtime/relay/relay-authority.ts new file mode 100644 index 000000000000..d2603b60391e --- /dev/null +++ b/src/main/runtime/relay/relay-authority.ts @@ -0,0 +1,29 @@ +// One reconcile's tenure as the coordinator's current authority. A waiter parks on the authority +// it joined; either that reconcile returning or a newer authority taking over releases it. Because +// replacing the authority is the only way to end one, there is no wake call to forget. +export class RelayAuthority { + private readonly done = Promise.withResolvers() + private readonly turnover = Promise.withResolvers() + readonly ended: Promise = this.turnover.promise + readonly settled: Promise = Promise.race([this.done.promise, this.turnover.promise]) + + // Nothing is running: before the first reconcile, and after a fence. A waiter that joins one is + // told so on its next pass rather than parking behind an open the fence already abandoned. + static idle(): RelayAuthority { + const authority = new RelayAuthority() + authority.done.resolve() + return authority + } + + static running(run: (authority: RelayAuthority) => Promise): RelayAuthority { + const authority = new RelayAuthority() + void run(authority).finally(() => { + authority.done.resolve() + }) + return authority + } + + end(): void { + this.turnover.resolve() + } +} diff --git a/src/main/runtime/relay/relay-concurrency-broker-wait.test.ts b/src/main/runtime/relay/relay-concurrency-broker-wait.test.ts new file mode 100644 index 000000000000..ed37835cbc88 --- /dev/null +++ b/src/main/runtime/relay/relay-concurrency-broker-wait.test.ts @@ -0,0 +1,298 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + RelayAuthCoordinator, + type CoordinatedRelayBroker, + type LiveBrokerWaitResult, + 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 +} + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise + }) + return { promise, resolve } +} + +// Reads a wait without ever awaiting it, so a waiter that never settles fails an +// assertion instead of hanging out to the runner timeout — which is what every +// leak in this file looks like. +function observe(wait: Promise): () => LiveBrokerWaitResult | 'unsettled' { + let seen: LiveBrokerWaitResult | 'unsettled' = 'unsettled' + void wait.then((result) => { + seen = result + }) + return () => seen +} + +afterEach(() => { + vi.useRealTimers() +}) + +describe('relay live-broker wait under interleaving', () => { + it('hands a waiter the broker a newer reconcile registered instead of the abandoned open', async () => { + // A reconcile settles mid-wait. The waiter joined the previous one, whose + // result is discarded — parking on it held the caller behind an open nobody + // would use, for as long as that open took to time out. + vi.useFakeTimers() + const abandoned = deferred() + const fresh = { closeNow: vi.fn() } + let opens = 0 + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker: async () => { + opens += 1 + return opens === 1 ? await abandoned.promise : fresh + }, + onStatus: vi.fn(), + random: () => 0.5 + }) + + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + expect(opens).toBe(1) + + const seen = observe(coordinator.waitForLiveBrokerResult(20_000)) + await vi.advanceTimersByTimeAsync(0) + expect(seen()).toBe('unsettled') + + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + expect(coordinator.getLiveBroker()).toBe(fresh) + + expect(seen()).toEqual({ broker: fresh }) + coordinator.stop() + }) + + it('follows a superseding reconcile that is still opening rather than answering from the old one', async () => { + // The other half of a mid-wait supersession: waking the waiter is only + // right if it then joins the newer reconcile. Answering as soon as it wakes + // would report "no broker, no cause" while the open that will supply one is + // still in flight. + vi.useFakeTimers() + const opens = [deferred(), deferred()] + const broker = { closeNow: vi.fn() } + let call = 0 + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker: () => opens[call++]!.promise, + onStatus: vi.fn(), + random: () => 0.5 + }) + + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + const seen = observe(coordinator.waitForLiveBrokerResult(20_000)) + await vi.advanceTimersByTimeAsync(0) + + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + expect(call).toBe(2) + expect(seen()).toBe('unsettled') + + opens[1]!.resolve(broker) + await vi.advanceTimersByTimeAsync(0) + expect(seen()).toEqual({ broker }) + coordinator.stop() + }) + + it('does not let the first waiter off an armed retry strand or answer for the second', async () => { + // Two waiters share one armed schedule. The short-budget one giving up must + // neither cancel the retry nor leave the long-budget one on a signal that + // has already fired. + vi.useFakeTimers() + const broker = { closeNow: vi.fn() } + const openBroker = vi + .fn() + .mockRejectedValueOnce(new RelayHttpError('assignment', 500)) + .mockResolvedValue(broker) + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker, + onStatus: vi.fn(), + random: () => 0.5 + }) + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + + const short = observe(coordinator.waitForLiveBrokerResult(100)) + const long = observe(coordinator.waitForLiveBrokerResult(20_000)) + await vi.advanceTimersByTimeAsync(100) + expect(short()).toEqual({ broker: null, offlineReason: 'broker_unavailable' }) + expect(long()).toBe('unsettled') + + // The retry the short waiter walked away from still fires, and its result + // reaches the waiter that stayed. + await vi.advanceTimersByTimeAsync(400) + expect(openBroker).toHaveBeenCalledTimes(2) + expect(long()).toEqual({ broker }) + coordinator.stop() + }) + + it('lets the armed retry take the tie when it fires on the tick the budget expires', async () => { + // attempt 0 with random 0.5 is a 500ms delay; the budget matches it exactly. + // The retry wins the tick, so the caller is told the fresh attempt's cause + // rather than the generic no-cause result. + vi.useFakeTimers() + const openBroker = vi.fn().mockRejectedValue(new RelayHttpError('assignment', 500)) + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker, + onStatus: vi.fn(), + random: () => 0.5 + }) + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + + const seen = observe(coordinator.waitForLiveBrokerResult(500)) + await vi.advanceTimersByTimeAsync(500) + expect(openBroker).toHaveBeenCalledTimes(2) + expect(seen()).toEqual({ broker: null, offlineReason: 'broker_unavailable' }) + coordinator.stop() + }) + + it('settles a waiter parked on an armed retry when the coordinator stops, leaving no timer', async () => { + vi.useFakeTimers() + const openBroker = vi.fn().mockRejectedValue(new RelayHttpError('assignment', 500)) + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker, + onStatus: vi.fn(), + random: () => 0.5 + }) + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + const seen = observe(coordinator.waitForLiveBrokerResult(20_000)) + await vi.advanceTimersByTimeAsync(0) + expect(vi.getTimerCount()).toBeGreaterThan(0) + + coordinator.stop() + await vi.advanceTimersByTimeAsync(0) + expect(seen()).toEqual({ broker: null, offlineReason: null }) + expect(vi.getTimerCount()).toBe(0) + }) + + it('settles a waiter parked on an open that never returns when the coordinator stops', async () => { + // The wait is deliberately unbounded across the open it joined, so nothing + // else can release this waiter; a teardown that did not wake it leaked the + // promise for the life of the process. + vi.useFakeTimers() + const stuck = deferred() + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker: () => stuck.promise, + onStatus: vi.fn(), + random: () => 0.5 + }) + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + const seen = observe(coordinator.waitForLiveBrokerResult(1_000)) + await vi.advanceTimersByTimeAsync(0) + expect(seen()).toBe('unsettled') + + coordinator.stop() + await vi.advanceTimersByTimeAsync(0) + expect(seen()).toEqual({ broker: null, offlineReason: null }) + expect(vi.getTimerCount()).toBe(0) + }) + + it('gives a waiter the terminal cause that landed while its retry was armed', async () => { + vi.useFakeTimers() + let current: RelayAuthContext | null = context + const openBroker = vi.fn().mockRejectedValue(new RelayHttpError('assignment', 500)) + const coordinator = new RelayAuthCoordinator({ + readContext: async () => current, + openBroker, + onStatus: vi.fn(), + random: () => 0.5 + }) + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + const seen = observe(coordinator.waitForLiveBrokerResult(20_000)) + await vi.advanceTimersByTimeAsync(0) + expect(seen()).toBe('unsettled') + + current = null + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + expect(seen()).toEqual({ broker: null, offlineReason: 'signed-out' }) + + // The retryable schedule the terminal cause superseded must arm nothing. + await vi.advanceTimersByTimeAsync(10 * 60_000) + expect(openBroker).toHaveBeenCalledTimes(1) + coordinator.stop() + }) + + it('keeps the budget over a chain of superseding reconciles, not just over armed retries', async () => { + // Every transient-demand acquire and release reconciles, so a busy host can + // supersede a waiter's reconcile faster than opens settle. The budget has to + // survive that or a caller rides the chain indefinitely holding its demand ref. + vi.useFakeTimers() + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker: () => + new Promise((_resolve, reject) => + setTimeout(() => reject(new Error('slow open')), 400) + ), + onStatus: vi.fn(), + random: () => 0.5 + }) + coordinator.reconcile() + const seen = observe(coordinator.waitForLiveBrokerResult(1_000)) + const churn = setInterval(() => coordinator.reconcile(), 200) + + await vi.advanceTimersByTimeAsync(2_000) + clearInterval(churn) + expect(seen()).not.toBe('unsettled') + coordinator.stop() + }) + it('settles a waiter parked on an open that never returns when the coordinator fences', async () => { + // A pre-sign-out fence discards the open the waiter joined. Without the wake + // the waiter sits behind it holding its transient demand ref; the phone gets + // no cause until that abandoned open hits its own ceiling. + vi.useFakeTimers() + const neverReturns = deferred() + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker: async () => await neverReturns.promise, + onStatus: vi.fn(), + random: () => 0.5 + }) + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + const seen = observe(coordinator.waitForLiveBrokerResult(20_000)) + await vi.advanceTimersByTimeAsync(0) + expect(seen()).toBe('unsettled') + + coordinator.fenceAndCloseNow() + await vi.advanceTimersByTimeAsync(0) + expect(seen()).not.toBe('unsettled') + }) + it('settles a waiter that arrives after a fence, without joining the open it abandoned', async () => { + // The fence installs an idle authority, so a caller arriving while the abandoned open is still + // in flight is told the cause now. Joining that open would park it behind readContext's own + // ceiling for an answer the fence already decided. + vi.useFakeTimers() + const neverReturns = deferred() + const coordinator = new RelayAuthCoordinator({ + readContext: async () => context, + openBroker: async () => await neverReturns.promise, + onStatus: vi.fn(), + random: () => 0.5 + }) + coordinator.reconcile() + await vi.advanceTimersByTimeAsync(0) + coordinator.fenceAndCloseNow() + + const seen = observe(coordinator.waitForLiveBrokerResult(20_000)) + await vi.advanceTimersByTimeAsync(0) + expect(seen()).not.toBe('unsettled') + }) +}) diff --git a/src/main/runtime/relay/relay-concurrency-policy-flip-mid-mint.test.ts b/src/main/runtime/relay/relay-concurrency-policy-flip-mid-mint.test.ts new file mode 100644 index 000000000000..0cf398da1016 --- /dev/null +++ b/src/main/runtime/relay/relay-concurrency-policy-flip-mid-mint.test.ts @@ -0,0 +1,185 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { OrcaCloudAuthConfig } from '../../orca-profiles/profile-cloud-auth-config' +import type { MobilePairingConnectionMode } from '../../../shared/mobile-pairing-connection-mode' +import type { OrcaRuntimeRpcServer } from '../runtime-rpc' + +// The policy-flip fix is normally exercised with the flip sequenced before the +// mint. These cover the interleaving it was written for: the flip lands while +// the mint is parked at an await, so the coordinator reaches `standby` and +// clears the offline reason out from under the waiter. +const fakes = vi.hoisted(() => ({ + readRelayAuthContext: vi.fn(), + connect: vi.fn(), + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: an empty literal cannot infer the element type, and vi.hoisted runs before the class that fills it exists. + brokers: [] as { closeNow: ReturnType }[] +})) + +vi.mock('./relay-auth-context', () => ({ readRelayAuthContext: fakes.readRelayAuthContext })) + +vi.mock('./relay-session-broker', () => { + class RelaySessionBroker { + readonly hostId = 'relay-host-1' + readonly ownerIdentityKey = 'user-1\0profile-1\0org-1' + readonly endpoint = { v: 1 as const, relayHostId: 'relay-host-1' } + readonly closeNow = vi.fn() + createPairingRelay = vi.fn(async (relayDeviceId: string) => ({ + v: 1, + relayHostId: this.hostId, + relayDeviceId, + inviteExpiresAt: 0 + })) + isLive(): boolean { + return this.closeNow.mock.calls.length === 0 + } + static connect = fakes.connect + } + return { RelaySessionBroker } +}) + +import { DesktopRelayService } from './desktop-relay-service' +import { RelaySessionBroker } from './relay-session-broker' + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise + }) + return { promise, resolve } +} + +// The real broker's constructor is private, so the mocked class is the only way +// to produce something `instanceof RelaySessionBroker` for the service's gate. +type FakeBroker = { + readonly hostId: string + readonly closeNow: ReturnType + createPairingRelay: (relayDeviceId: string) => Promise +} + +function newBroker(): FakeBroker { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the module mock replaces the class wholesale; only the fake's members are ever called. + const broker = new (RelaySessionBroker as unknown as new () => FakeBroker)() + fakes.brokers.push(broker) + return broker +} + +function service(mode: { current: MobilePairingConnectionMode }): DesktopRelayService { + fakes.readRelayAuthContext.mockResolvedValue({ + identity: { userId: 'user-1', profileId: 'profile-1', organizationId: 'org-1' }, + accessToken: 'access-1', + relayEntitled: true + }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the service reads only these four getters off the rpc server; the rest is never reached. + const runtimeRpc = { + getE2EEKeypair: () => ({ + publicKey: new Uint8Array(32).fill(7), + secretKey: new Uint8Array(32).fill(9), + publicKeyB64: 'x' + }), + getMobileSocketWiring: () => ({ attachTransport: () => () => {} }), + getRelayRevokeOutbox: () => ({ pendingFor: () => [], remove: vi.fn() }), + getDeviceRegistry: () => ({ + listDevices: () => [], + getDevice: () => ({ deviceId: 'device-1', scope: 'mobile' }), + getMobilePairingConnectionMode: () => 'automatic' + }) + } as unknown as OrcaRuntimeRpcServer + return new DesktopRelayService({ + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: RelaySessionBroker.connect reads only these two endpoints off authConfig. + authConfig: { + relayDirectorUrl: 'https://relay.example.test', + relayTokenEndpoint: 'https://login.example.test/relay-token' + } as OrcaCloudAuthConfig, + userDataPath: '/tmp/orca-relay-policy-flip-interleaving', + appVersion: '1.4.188', + runtimeRpc, + onStatus: () => {}, + hostMobilePairingConnectionMode: () => mode.current + }) +} + +beforeEach(() => { + vi.useFakeTimers() + fakes.brokers.length = 0 + fakes.connect.mockReset() +}) + +afterEach(() => { + vi.useRealTimers() +}) + +describe('DesktopRelayService LAN flip landing mid-mint', () => { + it('names the flip when it lands while the mint is parked on an in-flight open', async () => { + const mode: { current: MobilePairingConnectionMode } = { current: 'automatic' } + const open = deferred() + fakes.connect.mockImplementation(() => open.promise) + const relayService = service(mode) + try { + const mint = relayService.createPairingRelay('device-1') + const rejection = expect(mint).rejects.toThrow('relay_disabled_for_device') + await vi.advanceTimersByTimeAsync(0) + expect(fakes.connect).toHaveBeenCalled() + + mode.current = 'local-only' + relayService.pairingPolicyChanged() + await vi.advanceTimersByTimeAsync(0) + await rejection + + // The abandoned open must still be discarded rather than left registered. + open.resolve(newBroker()) + await vi.advanceTimersByTimeAsync(0) + expect(fakes.brokers[0]!.closeNow).toHaveBeenCalled() + } finally { + relayService.stop() + } + }) + + it('names the flip when it lands after the broker is live, mid create_pairing_relay', async () => { + const mode: { current: MobilePairingConnectionMode } = { current: 'automatic' } + const broker = newBroker() + const mintCall = deferred() + const relayService = service(mode) + broker.createPairingRelay = vi.fn(async () => { + mode.current = 'local-only' + relayService.pairingPolicyChanged() + await mintCall.promise + throw new Error('unreachable') + }) + fakes.connect.mockResolvedValue(broker) + try { + const mint = relayService.createPairingRelay('device-1') + const rejection = expect(mint).rejects.toThrow('relay_disabled_for_device') + await vi.advanceTimersByTimeAsync(0) + expect(broker.createPairingRelay).toHaveBeenCalled() + + // The control the flip closed under it fails the call that was in flight. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: resolves a deferred whose value the caller discards; the type is the broker's, not this suite's. + mintCall.resolve(undefined as never) + await vi.advanceTimersByTimeAsync(0) + await rejection + } finally { + relayService.stop() + } + }) + + it('leaves the real failure alone when the flip is reverted before the mint fails', async () => { + // The rewrite reads the live policy at the moment of failure, so a flip that + // is undone in the same window must not claim a failure it did not cause. + const mode: { current: MobilePairingConnectionMode } = { current: 'automatic' } + const broker = newBroker() + broker.createPairingRelay = vi.fn(async () => { + mode.current = 'local-only' + await Promise.resolve() + mode.current = 'automatic' + throw new Error('relay_assignment_failed_500') + }) + fakes.connect.mockResolvedValue(broker) + const relayService = service(mode) + try { + await expect(relayService.createPairingRelay('device-1')).rejects.toThrow( + 'relay_assignment_failed_500' + ) + } finally { + relayService.stop() + } + }) +}) diff --git a/src/main/runtime/relay/relay-demand-ledger-lifecycle.test.ts b/src/main/runtime/relay/relay-demand-ledger-lifecycle.test.ts new file mode 100644 index 000000000000..d10995a15236 --- /dev/null +++ b/src/main/runtime/relay/relay-demand-ledger-lifecycle.test.ts @@ -0,0 +1,231 @@ +import { mkdtempSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { DeviceRegistry } from '../device-registry' +import type { MobilePairingConnectionMode } from '../../../shared/mobile-pairing-connection-mode' +import { isMobileRelayAllowed } from '../../../shared/mobile-relay-policy' +import { RelayDemandLedger } from './relay-demand-ledger' +import { RelayRevokeOutbox, type RelayDeviceBinding } from './relay-revoke-outbox' + +const ownerIdentityKey = 'user-1\0profile-1\0org-1' +const relayHostId = 'relay-host-1' + +/** Mirrors DesktopRelayService.isRelayAllowedForDevice: a live pull, not a snapshot. */ +function fixture(now = 1_000) { + const userDataPath = mkdtempSync(join(tmpdir(), 'orca-relay-demand-lifecycle-')) + const deviceRegistry = new DeviceRegistry(userDataPath) + const revokeOutbox = new RelayRevokeOutbox(userDataPath) + const host: { mode: MobilePairingConnectionMode } = { mode: 'automatic' } + const ledger = new RelayDemandLedger({ + deviceRegistry, + revokeOutbox, + relayHostId, + now: () => now, + isRelayAllowedForDevice: (deviceId) => + isMobileRelayAllowed({ + hostConnectionMode: host.mode, + deviceConnectionMode: deviceRegistry.getMobilePairingConnectionMode(deviceId) ?? null + }) + }) + return { userDataPath, deviceRegistry, revokeOutbox, ledger, host } +} + +function binding(relayDeviceId: string, inviteExpiresAt?: number): RelayDeviceBinding { + return { relayDeviceId, relayHostId, ownerIdentityKey, inviteExpiresAt } +} + +function pairedPhone(fx: ReturnType, name = 'Phone') { + const phone = fx.deviceRegistry.addDevice(name) + fx.deviceRegistry.setMobilePairingConnectionMode(phone.deviceId, 'automatic') + fx.deviceRegistry.setRelayBinding(phone.deviceId, binding(phone.deviceId)) + fx.deviceRegistry.updateLastSeen(phone.deviceId) + return phone.deviceId +} + +describe('RelayDemandLedger transient-ref lifecycle', () => { + afterEach(() => vi.useRealTimers()) + + // Hazard 1: the live policy flips between acquire and release. + it('releases a ref acquired before a policy flip instead of leaking it', () => { + const fx = fixture() + const deviceId = pairedPhone(fx) + const release = fx.ledger.acquireTransient(`endpoints:${deviceId}`, deviceId) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + + fx.host.mode = 'local-only' + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + + // The release path must be policy-blind, or the ref is unreachable forever. + release() + expect(fx.ledger.transientRefsForTests().size).toBe(0) + + fx.host.mode = 'automatic' + // Standing binding demand returns; the released ref must not add a second pin. + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + fx.deviceRegistry.removeDevice(deviceId) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + }) + + it('restores a held ref as demand when the policy flips back', () => { + const fx = fixture() + const deviceId = pairedPhone(fx) + fx.deviceRegistry.removeDevice(deviceId) + const release = fx.ledger.acquireTransient(`pairing:${deviceId}`, deviceId) + + fx.host.mode = 'local-only' + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + // The filter must be re-evaluated per call, not baked in at acquire time. + fx.host.mode = 'automatic' + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + release() + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + }) + + // Hazard 2: the same release closure invoked repeatedly. + it('a repeated release does not decrement a count another holder still owns', () => { + const fx = fixture() + const deviceId = pairedPhone(fx) + fx.deviceRegistry.removeDevice(deviceId) + const key = `endpoints:${deviceId}` + const releaseFirst = fx.ledger.acquireTransient(key, deviceId) + const releaseSecond = fx.ledger.acquireTransient(key, deviceId) + expect(fx.ledger.transientRefsForTests().get(key)?.count).toBe(2) + + releaseFirst() + releaseFirst() + releaseFirst() + expect(fx.ledger.transientRefsForTests().get(key)?.count).toBe(1) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + + releaseSecond() + expect(fx.ledger.transientRefsForTests().size).toBe(0) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + }) + + it('a stale release from a finished ref does not cancel a later acquire of the same key', () => { + const fx = fixture() + const deviceId = pairedPhone(fx) + fx.deviceRegistry.removeDevice(deviceId) + const key = `provision:${deviceId}` + const stale = fx.ledger.acquireTransient(key, deviceId) + stale() + expect(fx.ledger.transientRefsForTests().size).toBe(0) + + const fresh = fx.ledger.acquireTransient(key, deviceId) + stale() + expect(fx.ledger.transientRefsForTests().get(key)?.count).toBe(1) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + fresh() + expect(fx.ledger.transientRefsForTests().size).toBe(0) + }) + + // Hazard 3: an operation throws and its ref is never released. + it('an unreleased ref pins demand and retains its map entry', () => { + const fx = fixture() + const deviceId = pairedPhone(fx) + fx.deviceRegistry.removeDevice(deviceId) + for (let index = 0; index < 5; index += 1) { + fx.ledger.acquireTransient(`pairing:abandoned-${index}`, `abandoned-${index}`) + } + expect(fx.ledger.transientRefsForTests().size).toBe(5) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + }) + + // Hazard 4: two refs for one device released in the opposite order. + it('is release-order independent for refs sharing a key', () => { + const fx = fixture() + const deviceId = pairedPhone(fx) + fx.deviceRegistry.removeDevice(deviceId) + const key = `endpoints:${deviceId}` + const outer = fx.ledger.acquireTransient(key, deviceId) + const inner = fx.ledger.acquireTransient(key, deviceId) + inner() + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + const middle = fx.ledger.acquireTransient(key, deviceId) + outer() + expect(fx.ledger.transientRefsForTests().get(key)?.count).toBe(1) + middle() + expect(fx.ledger.transientRefsForTests().size).toBe(0) + }) + + // Hazard 5: a release for device B must not disturb device A's ref. + it('keeps per-device refs independent across a concurrent release', () => { + const fx = fixture() + const deviceA = fx.deviceRegistry.addDevice('Phone A').deviceId + const deviceB = fx.deviceRegistry.addDevice('Phone B').deviceId + fx.deviceRegistry.setMobilePairingConnectionMode(deviceA, 'local-only') + fx.deviceRegistry.setMobilePairingConnectionMode(deviceB, 'automatic') + + const releaseA = fx.ledger.acquireTransient(`pairing:${deviceA}`, deviceA) + const releaseB = fx.ledger.acquireTransient(`pairing:${deviceB}`, deviceB) + // A is excluded by its own pair-time mode; B alone carries the demand. + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + + releaseB() + expect(fx.ledger.transientRefsForTests().size).toBe(1) + expect(fx.ledger.transientRefsForTests().get(`pairing:${deviceA}`)?.deviceId).toBe(deviceA) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + + releaseA() + expect(fx.ledger.transientRefsForTests().size).toBe(0) + }) + + // Hazard 6: the ledger owns no timers, so teardown is the caller's map only. + it('arms no timers of its own across acquire, query and release', () => { + vi.useFakeTimers() + const fx = fixture() + const deviceId = pairedPhone(fx) + fx.deviceRegistry.setRelayBinding(deviceId, binding(deviceId, 2_000)) + const release = fx.ledger.acquireTransient(`pairing:${deviceId}`, deviceId) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(true) + expect(fx.ledger.nextPendingExpiry()).toBe(2_000) + release() + expect(vi.getTimerCount()).toBe(0) + }) +}) + +describe('RelayDemandLedger nextPendingExpiry scoping', () => { + it('ignores a pending invite bound to a different relay host', () => { + const fx = fixture() + const phone = fx.deviceRegistry.addDevice('Foreign-host phone') + fx.deviceRegistry.setRelayBinding(phone.deviceId, { + ...binding(phone.deviceId, 2_000), + relayHostId: 'relay-host-2' + }) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + expect(fx.ledger.nextPendingExpiry()).toBeNull() + }) + + it('ignores a pending invite on a non-mobile device', () => { + const fx = fixture() + const runtime = fx.deviceRegistry.addDevice('Runtime', 'runtime') + fx.deviceRegistry.setRelayBinding(runtime.deviceId, binding(runtime.deviceId, 2_000)) + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + expect(fx.ledger.nextPendingExpiry()).toBeNull() + }) + + it('ignores a pending invite the live policy excludes, and re-arms when it flips back', () => { + const fx = fixture() + const phone = fx.deviceRegistry.addDevice('LAN phone') + fx.deviceRegistry.setMobilePairingConnectionMode(phone.deviceId, 'automatic') + fx.deviceRegistry.setRelayBinding(phone.deviceId, binding(phone.deviceId, 2_000)) + expect(fx.ledger.nextPendingExpiry()).toBe(2_000) + + fx.host.mode = 'local-only' + expect(fx.ledger.hasDemand(ownerIdentityKey)).toBe(false) + expect(fx.ledger.nextPendingExpiry()).toBeNull() + + fx.host.mode = 'automatic' + expect(fx.ledger.nextPendingExpiry()).toBe(2_000) + }) + + it('still reports the nearest in-scope pending invite', () => { + const fx = fixture() + const near = fx.deviceRegistry.addDevice('Near phone') + const far = fx.deviceRegistry.addDevice('Far phone') + fx.deviceRegistry.setRelayBinding(far.deviceId, binding(far.deviceId, 9_000)) + fx.deviceRegistry.setRelayBinding(near.deviceId, binding(near.deviceId, 4_000)) + expect(fx.ledger.nextPendingExpiry()).toBe(4_000) + }) +}) diff --git a/src/main/runtime/relay/relay-demand-ledger.ts b/src/main/runtime/relay/relay-demand-ledger.ts index 817fbc37e370..a7aad3e397b6 100644 --- a/src/main/runtime/relay/relay-demand-ledger.ts +++ b/src/main/runtime/relay/relay-demand-ledger.ts @@ -1,5 +1,5 @@ -import type { DeviceRegistry } from '../device-registry' -import type { RelayRevokeOutbox } from './relay-revoke-outbox' +import type { DeviceEntry, DeviceRegistry } from '../device-registry' +import type { RelayDeviceBinding, RelayRevokeOutbox } from './relay-revoke-outbox' type RelayDemandLedgerOptions = { deviceRegistry: DeviceRegistry @@ -35,6 +35,10 @@ export class RelayDemandLedger { } released = true const current = this.transientRefs.get(key) + // Keep this guard even though nothing can reach it today: `released` makes the lookup happen + // at most once per closure and no public path empties the map, so mutating it away survives + // the suite. It goes load-bearing the moment the ledger grows a dispose()/clear(), and + // nothing would catch that. if (!current) { return } @@ -47,6 +51,10 @@ export class RelayDemandLedger { } hasDemand(ownerIdentityKey: string): boolean { + // KNOWN GAP: a transient ref carries no owner identity, so this loop answers true for any + // signed-in identity, unlike the owner-filtered branches below. The fix threads identity + // through acquireTransient (call site: desktop-relay-service.ts); inferring the owner here is + // unsafe, because withTransientDemand('provision') rebinds the device mid-operation. for (const ref of this.transientRefs.values()) { if (this.isRelayAllowed(ref.deviceId)) { return true @@ -57,14 +65,8 @@ export class RelayDemandLedger { } const now = (this.options.now ?? Date.now)() return this.options.deviceRegistry.listDevices().some((device) => { - const binding = device.relayBinding - if ( - device.scope !== 'mobile' || - !binding || - binding.ownerIdentityKey !== ownerIdentityKey || - binding.relayHostId !== this.options.relayHostId || - !this.isRelayAllowed(device.deviceId) - ) { + const binding = this.demandCandidateBinding(device) + if (!binding || binding.ownerIdentityKey !== ownerIdentityKey) { return false } // Why: E2EE authentication marks a scanned DeviceEntry as seen before @@ -78,7 +80,9 @@ export class RelayDemandLedger { const now = (this.options.now ?? Date.now)() let next: number | null = null for (const device of this.options.deviceRegistry.listDevices()) { - const expiresAt = device.relayBinding?.inviteExpiresAt + // Why: an expiry that hasDemand would never look at still armed a wake + // timer — another host's binding, a runtime device, a LAN-excluded phone. + const expiresAt = this.demandCandidateBinding(device)?.inviteExpiresAt if (expiresAt && expiresAt > now && (next === null || expiresAt < next)) { next = expiresAt } @@ -86,6 +90,25 @@ export class RelayDemandLedger { return next } + /** Test-only view of the outstanding transient refs. */ + transientRefsForTests(): ReadonlyMap> { + return this.transientRefs + } + + /** The binding when this device could stand as demand on this relay host, else null. */ + private demandCandidateBinding(device: DeviceEntry): RelayDeviceBinding | null { + const binding = device.relayBinding + if ( + device.scope !== 'mobile' || + !binding || + binding.relayHostId !== this.options.relayHostId || + !this.isRelayAllowed(device.deviceId) + ) { + return null + } + return binding + } + private isRelayAllowed(deviceId: string): boolean { return this.options.isRelayAllowedForDevice?.(deviceId) ?? true } diff --git a/src/main/runtime/relay/relay-retry-schedule.test.ts b/src/main/runtime/relay/relay-retry-schedule.test.ts index aff1ce6319a7..936e55c83c7b 100644 --- a/src/main/runtime/relay/relay-retry-schedule.test.ts +++ b/src/main/runtime/relay/relay-retry-schedule.test.ts @@ -6,39 +6,17 @@ afterEach(() => { }) describe('RelayRetrySchedule', () => { - it('exposes no settle signal while nothing is armed', () => { - const schedule = new RelayRetrySchedule(() => 0.5) - expect(schedule.pending).toBe(false) - expect(schedule.settled).toBeNull() - }) - - it('settles after the retry has run, so a woken waiter sees its effect', async () => { - vi.useFakeTimers() - const schedule = new RelayRetrySchedule(() => 0.5) - const order: string[] = [] - schedule.schedule(0, () => order.push('retry')) - const settled = schedule.settled - expect(settled).not.toBeNull() - void settled!.then(() => order.push('settled')) - - await vi.advanceTimersByTimeAsync(500) - expect(order).toEqual(['retry', 'settled']) - expect(schedule.pending).toBe(false) - expect(schedule.settled).toBeNull() - }) - - it('settles on cancel without running the retry', async () => { + it('clears the timer on cancel without running the retry', async () => { vi.useFakeTimers() const schedule = new RelayRetrySchedule(() => 0.5) const retry = vi.fn() schedule.schedule(0, retry) - const settled = schedule.settled! + expect(schedule.pending).toBe(true) schedule.cancel() - await expect(settled).resolves.toBeUndefined() + expect(schedule.pending).toBe(false) await vi.advanceTimersByTimeAsync(10_000) expect(retry).not.toHaveBeenCalled() - expect(schedule.settled).toBeNull() }) it('keeps the attempt count across cancel so a superseded retry does not restart the ladder', async () => { diff --git a/src/main/runtime/relay/relay-retry-schedule.ts b/src/main/runtime/relay/relay-retry-schedule.ts index dcab1069e37e..79becfefb768 100644 --- a/src/main/runtime/relay/relay-retry-schedule.ts +++ b/src/main/runtime/relay/relay-retry-schedule.ts @@ -1,12 +1,10 @@ import { computeRetryDelayMs } from './relay-retry-backoff' -// One armed retry at a time with jittered exponential backoff. `settled` lets a -// waiter sit through the armed retry instead of reading "nothing is running" as -// "nothing more is coming": it resolves when the retry fires or is cancelled. +// One armed retry at a time with jittered exponential backoff. A waiter sits through an armed +// retry by parking on the coordinator's authority, which the retry ends when it fires. export class RelayRetrySchedule { private timer: ReturnType | null = null private attempt = 0 - private armed: PromiseWithResolvers | null = null constructor(private readonly random: () => number = Math.random) {} @@ -14,22 +12,14 @@ export class RelayRetrySchedule { return this.timer !== null } - get settled(): Promise | null { - return this.armed?.promise ?? null - } - schedule(retryAfterMs: number, retry: () => void): void { if (this.timer) { return } const delayMs = computeRetryDelayMs(this.attempt++, retryAfterMs, this.random) - const armed = Promise.withResolvers() - this.armed = armed this.timer = setTimeout(() => { this.timer = null - this.armed = null retry() - armed.resolve() }, delayMs) } @@ -37,15 +27,12 @@ export class RelayRetrySchedule { this.attempt = 0 } - // Why the waiter is woken rather than left parked: a cancelled retry is - // superseded by a fresh reconcile (or a fence), and the waiter must re-check. + // Why no reset(): the coordinator gates that on resetRetry, so a superseded retry keeps its + // place on the backoff ladder rather than restarting it on every reconcile. cancel(): void { if (this.timer) { clearTimeout(this.timer) this.timer = null } - const armed = this.armed - this.armed = null - armed?.resolve() } }