diff --git a/src/main/runtime/push/desktop-push-service.test.ts b/src/main/runtime/push/desktop-push-service.test.ts index 5258ceb6d56..4cb47651d74 100644 --- a/src/main/runtime/push/desktop-push-service.test.ts +++ b/src/main/runtime/push/desktop-push-service.test.ts @@ -14,7 +14,14 @@ const REGISTER_INPUT = { filter: { sources: ['agent-task-complete'] as const, agentStates: ['finished'] as const } } -function createService(options: { registerFails?: boolean; deleteFails?: boolean } = {}): { +function createService( + options: { + registerFails?: boolean + deleteFails?: boolean + /** Runs before each delete resolves, so a suite can queue work mid-flush. */ + onDelete?: (registrationId: string) => void + } = {} +): { service: DesktopPushService registry: DeviceRegistry outbox: PushUnregisterOutbox @@ -22,6 +29,7 @@ function createService(options: { registerFails?: boolean; deleteFails?: boolean deletes: string[] send: ReturnType dispatch: (event: MobileNotificationEvent) => void + retries: { run: () => void; delayMs: number }[] } { const userDataPath = mkdtempSync(join(tmpdir(), 'orca-push-service-')) const registry = new DeviceRegistry(userDataPath) @@ -54,17 +62,22 @@ function createService(options: { registerFails?: boolean; deleteFails?: boolean ), deleteDevice: vi.fn(async (registrationId: string) => { deletes.push(registrationId) + options.onDelete?.(registrationId) return options.deleteFails ? { deleted: false, retryable: true } : { deleted: true, retryable: false } }), send: vi.fn(async () => ({ ok: true, results: [] }) as const) } + const retries: { run: () => void; delayMs: number }[] = [] const service = DesktopPushService.create({ runtime: runtime as never, runtimeRpc: runtimeRpc as never, gatewayUrl: 'https://push.onorca.dev', - client: client as never + client: client as never, + scheduleRetry: (run, delayMs) => { + retries.push({ run, delayMs }) + } })! service.start() @@ -75,7 +88,8 @@ function createService(options: { registerFails?: boolean; deleteFails?: boolean deviceId: device.deviceId, deletes, send: client.send, - dispatch: (event) => listener?.(event) + dispatch: (event) => listener?.(event), + retries } } @@ -150,6 +164,83 @@ describe('DesktopPushService', () => { expect(harness.outbox.pending()).toEqual([]) }) + it('unregisters at the gateway when the device stopped being a phone mid-register', async () => { + const harness = createService() + vi.spyOn(harness.registry, 'setPushRegistration').mockReturnValue(false) + + expect( + await harness.service.register({ deviceId: harness.deviceId, ...REGISTER_INPUT }) + ).toEqual({ registered: false, reason: 'not_mobile' }) + // register() kicks the flush off without awaiting it; join the same run. + await harness.service.flushUnregisterOutbox() + expect(harness.deletes).toEqual(['reg-1']) + expect(harness.outbox.pending()).toEqual([]) + }) + + it('unregisters at the gateway when the registration cannot be written', async () => { + const harness = createService({ deleteFails: true }) + vi.spyOn(harness.registry, 'setPushRegistration').mockImplementation(() => { + throw new Error('disk full') + }) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + + expect( + await harness.service.register({ deviceId: harness.deviceId, ...REGISTER_INPUT }) + ).toEqual({ registered: false, reason: 'registration_storage_failed' }) + // The gateway kept the token, so the delete stays queued until it lands. + expect(harness.outbox.pending()).toEqual([ + expect.objectContaining({ registrationId: 'reg-1', deviceId: harness.deviceId }) + ]) + warn.mockRestore() + }) + + it('drains a delete queued while a flush is already running', async () => { + let queued = false + const harness = createService({ + onDelete: () => { + if (queued) { + return + } + queued = true + harness.outbox.enqueue({ registrationId: 'reg-late', deviceId: 'device-late' }) + // Mirrors unregister(): the trigger arrives while the flush is mid-await. + void harness.service.flushUnregisterOutbox() + } + }) + harness.outbox.enqueue({ registrationId: 'reg-first', deviceId: 'device-first' }) + + await harness.service.flushUnregisterOutbox() + + expect(harness.deletes).toEqual(['reg-first', 'reg-late']) + expect(harness.outbox.pending()).toEqual([]) + }) + + it('retries a failed drain on a capped backoff instead of waiting for a relaunch', async () => { + const harness = createService({ deleteFails: true }) + harness.outbox.enqueue({ registrationId: 'reg-stuck', deviceId: 'device-1' }) + + await harness.service.flushUnregisterOutbox() + expect(harness.retries.map((entry) => entry.delayMs)).toEqual([30_000]) + + harness.retries[0]?.run() + await new Promise((resolve) => setImmediate(resolve)) + expect(harness.deletes).toEqual(['reg-stuck', 'reg-stuck']) + expect(harness.retries.map((entry) => entry.delayMs)).toEqual([30_000, 60_000]) + expect(harness.outbox.pending()).toHaveLength(1) + }) + + it('stops re-arming the retry once the service is stopped', async () => { + const harness = createService({ deleteFails: true }) + harness.outbox.enqueue({ registrationId: 'reg-stuck', deviceId: 'device-1' }) + await harness.service.flushUnregisterOutbox() + + harness.service.stop() + harness.retries[0]?.run() + await new Promise((resolve) => setImmediate(resolve)) + + expect(harness.retries).toHaveLength(1) + }) + it('pushes a dispatched notification through the subscribed dispatcher', async () => { const harness = createService() await harness.service.register({ deviceId: harness.deviceId, ...REGISTER_INPUT }) diff --git a/src/main/runtime/push/desktop-push-service.ts b/src/main/runtime/push/desktop-push-service.ts index 7b462d4f793..da67c3fc41c 100644 --- a/src/main/runtime/push/desktop-push-service.ts +++ b/src/main/runtime/push/desktop-push-service.ts @@ -13,12 +13,19 @@ import { PushDispatcher } from './push-dispatcher' import { PushGatewayClient } from './push-gateway-client' import type { PushUnregisterOutbox } from './push-unregister-outbox' +const OUTBOX_RETRY_BASE_MS = 30_000 +const OUTBOX_RETRY_MAX_MS = 10 * 60_000 + +type RegisterStorageFailure = 'not_mobile' | 'registration_storage_failed' + type DesktopPushServiceOptions = { runtime: OrcaRuntimeService runtimeRpc: OrcaRuntimeRpcServer gatewayUrl: string /** Test seam: lets a suite drive the service without a live gateway. */ client?: PushGatewayClient + /** Test seam: lets a suite drive the outbox backoff without real timers. */ + scheduleRetry?: (run: () => void, delayMs: number) => void } export class DesktopPushService { @@ -28,8 +35,13 @@ export class DesktopPushService { private readonly outbox: PushUnregisterOutbox private readonly client: PushGatewayClient private readonly dispatcher: PushDispatcher + private readonly scheduleRetry: (run: () => void, delayMs: number) => void private unsubscribe: (() => void) | null = null - private flushing = false + private flushLoop: Promise | null = null + private flushRequested = false + private retryArmed = false + private retryDelayMs = OUTBOX_RETRY_BASE_MS + private stopped = false private constructor( options: DesktopPushServiceOptions, @@ -42,6 +54,12 @@ export class DesktopPushService { this.client = client this.outbox = options.runtimeRpc.getPushUnregisterOutbox() this.dispatcher = new PushDispatcher({ client, registry }) + this.scheduleRetry = + options.scheduleRetry ?? + ((run, delayMs) => { + // Why: a queued gateway delete must never hold the app open at quit. + setTimeout(run, delayMs).unref?.() + }) } /** Returns null when the mobile runtime never came up, so there is nothing to push for. */ @@ -57,6 +75,7 @@ export class DesktopPushService { } start(): void { + this.stopped = false this.runtime.setMobilePushRegistrar(this) this.unsubscribe = this.runtime.onNotificationDispatched((event) => { this.dispatcher.enqueue(event) @@ -70,6 +89,7 @@ export class DesktopPushService { } stop(): void { + this.stopped = true this.unsubscribe?.() this.unsubscribe = null this.runtimeRpc.setOnPushUnregisterQueued(null) @@ -87,14 +107,16 @@ export class DesktopPushService { reason: result.reason === 'unreachable' ? 'gateway_unreachable' : 'gateway_rejected' } } - this.registry.setPushRegistration(input.deviceId, { - registrationId: result.registrationId, - platform: input.platform, - filter: input.filter, - registeredAt: Date.now() - }) + const failure = this.storeRegistration(input, result.registrationId) + if (failure) { + // Why: the gateway now holds a token this host will never push to. Queue its + // delete instead of leaking it until the phone happens to register again. + this.outbox.enqueue({ registrationId: result.registrationId, deviceId: input.deviceId }) + } void this.flushUnregisterOutbox() - return { registered: true, registrationId: result.registrationId } + return failure + ? { registered: false, reason: failure } + : { registered: true, registrationId: result.registrationId } } async unregister(deviceId: string): Promise<{ unregistered: boolean }> { @@ -110,23 +132,85 @@ export class DesktopPushService { return { unregistered: true } } + /** Joining an in-flight drain still waits for the item this call queued. */ async flushUnregisterOutbox(): Promise { - if (this.flushing) { - return + this.flushRequested = true + this.flushLoop ??= this.runFlushLoop().finally(() => { + this.flushLoop = null + }) + await this.flushLoop + } + + private async runFlushLoop(): Promise { + while (this.flushRequested && !this.stopped) { + // Cleared before the pass, so a delete queued mid-drain earns another one. + this.flushRequested = false + if (await this.drainPending()) { + this.scheduleFlushRetry() + } else { + this.retryDelayMs = OUTBOX_RETRY_BASE_MS + } } - this.flushing = true + } + + /** Returns the refusal reason when a gateway-accepted registration cannot be stored. */ + private storeRegistration( + input: MobilePushRegisterInput, + registrationId: string + ): RegisterStorageFailure | null { try { - // Safe to iterate while removing: the outbox swaps in a new array per write. - for (const item of this.outbox.pending()) { + const stored = this.registry.setPushRegistration(input.deviceId, { + registrationId, + platform: input.platform, + filter: input.filter, + registeredAt: Date.now() + }) + // False means the device was removed or left mobile scope while the gateway + // call was in flight. + return stored ? null : 'not_mobile' + } catch (error) { + console.warn('[push] Failed to persist a push registration:', error) + return 'registration_storage_failed' + } + } + + /** Returns true when the pass left behind an item the gateway may still accept. */ + private async drainPending(): Promise { + const attempted = new Set() + let retryable = false + for (;;) { + // Re-read per item: a snapshot taken at loop entry misses anything queued + // while an await was in flight, and the outbox swaps arrays on every write. + const item = this.outbox.pending().find((candidate) => !attempted.has(candidate.reqId)) + if (!item) { + return retryable + } + attempted.add(item.reqId) + try { const result = await this.client.deleteDevice(item.registrationId) if (result.deleted || !result.retryable) { this.outbox.remove(item.reqId) + } else { + retryable = true } + } catch (error) { + // One bad delete must not strand the rest of the queue. + console.warn('[push] Failed to drain the push unregister outbox:', error) + retryable = true } - } catch (error) { - console.warn('[push] Failed to drain the push unregister outbox:', error) - } finally { - this.flushing = false } } + + private scheduleFlushRetry(): void { + if (this.retryArmed || this.stopped) { + return + } + this.retryArmed = true + const delayMs = this.retryDelayMs + this.retryDelayMs = Math.min(delayMs * 2, OUTBOX_RETRY_MAX_MS) + this.scheduleRetry(() => { + this.retryArmed = false + void this.flushUnregisterOutbox() + }, delayMs) + } } diff --git a/src/main/runtime/push/push-dispatcher.test.ts b/src/main/runtime/push/push-dispatcher.test.ts index 40c634ccb8d..5a30115e364 100644 --- a/src/main/runtime/push/push-dispatcher.test.ts +++ b/src/main/runtime/push/push-dispatcher.test.ts @@ -132,6 +132,43 @@ describe('PushDispatcher', () => { }) }) + it('fans out past the per-request cap instead of starving the extra devices', async () => { + const devices = Array.from({ length: 25 }, (_, index) => ({ + deviceId: `device-${index}`, + pushRegistration: registration({ registrationId: `reg-${index}` }) + })) + const harness = createHarness({ devices }) + + harness.dispatcher.enqueue(notification()) + await flush() + + expect(harness.sends).toHaveLength(2) + expect(harness.sends[0]?.registrationIds).toHaveLength(20) + expect(harness.sends[1]?.registrationIds).toEqual([ + 'reg-20', + 'reg-21', + 'reg-22', + 'reg-23', + 'reg-24' + ]) + }) + + it('drops a dead registration reported by a later chunk', async () => { + const devices = Array.from({ length: 25 }, (_, index) => ({ + deviceId: `device-${index}`, + pushRegistration: registration({ registrationId: `reg-${index}` }) + })) + const harness = createHarness({ + devices, + results: [{ registrationId: 'reg-24', status: 'dead' }] + }) + + harness.dispatcher.enqueue(notification()) + await flush() + + expect(harness.cleared).toEqual(['device-24']) + }) + it('never pushes a dismissal', async () => { const harness = createHarness({ devices: [{ deviceId: 'a', pushRegistration: registration() }] diff --git a/src/main/runtime/push/push-dispatcher.ts b/src/main/runtime/push/push-dispatcher.ts index 76b52b9b306..f3f9d9af78c 100644 --- a/src/main/runtime/push/push-dispatcher.ts +++ b/src/main/runtime/push/push-dispatcher.ts @@ -11,8 +11,8 @@ import type { MobileNotificationEvent } from '../runtime-mobile-notification-con import type { PushGatewayClient, PushSendNotification } from './push-gateway-client' const PUSH_RETRY_DELAY_MS = 2_000 -// The gateway rejects a whole request above this; a host with more paired phones -// than this still pushes to the first 20 rather than to none. +// The gateway rejects a whole request above this, so a host with more paired +// phones fans out across several sends rather than starving the extras. const MAX_REGISTRATIONS_PER_SEND = 20 const PUSH_TITLE_MAX_LENGTH = 80 const PUSH_BODY_MAX_LENGTH = 180 @@ -76,8 +76,13 @@ export class PushDispatcher { enqueue(event: MobileNotificationEvent): void { try { const plan = this.planSend(event) - if (plan) { - void this.deliver(plan.targets, plan.notification, 0) + if (!plan) { + return + } + // Each chunk is its own request, so its retry and dead-drop are independent. + for (let start = 0; start < plan.targets.length; start += MAX_REGISTRATIONS_PER_SEND) { + const chunk = plan.targets.slice(start, start + MAX_REGISTRATIONS_PER_SEND) + void this.deliver(chunk, plan.notification, 0) } } catch (error) { console.warn('[push] Failed to prepare a push notification:', error) @@ -99,19 +104,16 @@ export class PushDispatcher { if (agentState === undefined) { return null } - const targets = this.registry - .listDevices() - .flatMap((device) => { - const registration = device.pushRegistration - if (!registration || !registration.filter.sources.includes(source)) { - return [] - } - if (agentState !== null && !registration.filter.agentStates.includes(agentState)) { - return [] - } - return [{ deviceId: device.deviceId, registrationId: registration.registrationId }] - }) - .slice(0, MAX_REGISTRATIONS_PER_SEND) + const targets = this.registry.listDevices().flatMap((device) => { + const registration = device.pushRegistration + if (!registration || !registration.filter.sources.includes(source)) { + return [] + } + if (agentState !== null && !registration.filter.agentStates.includes(agentState)) { + return [] + } + return [{ deviceId: device.deviceId, registrationId: registration.registrationId }] + }) if (targets.length === 0) { return null } diff --git a/src/main/runtime/push/push-gateway-client.test.ts b/src/main/runtime/push/push-gateway-client.test.ts index 565370c1580..b73b904de45 100644 --- a/src/main/runtime/push/push-gateway-client.test.ts +++ b/src/main/runtime/push/push-gateway-client.test.ts @@ -19,7 +19,9 @@ function jsonResponse(status: number, body: unknown): Response { }) } -function createFakeGateway(options: { sessionTtlMs?: number; devicesStatus?: number } = {}): { +function createFakeGateway( + options: { sessionTtlMs?: number; devicesStatus?: number; rejectBearer?: boolean } = {} +): { client: PushGatewayClient calls: Recorded[] expireSession: () => void @@ -30,6 +32,7 @@ function createFakeGateway(options: { sessionTtlMs?: number; devicesStatus?: num const now = { value: NOW } const calls: Recorded[] = [] const liveTokens = new Set() + const knownRegistrations = new Set() let issued = 0 let pendingProof: string | null = null @@ -68,18 +71,27 @@ function createFakeGateway(options: { sessionTtlMs?: number; devicesStatus?: num }) } const bearer = headers.get('authorization')?.replace('Bearer ', '') ?? '' - if (!liveTokens.has(bearer)) { + if (options.rejectBearer || !liveTokens.has(bearer)) { return jsonResponse(401, { error: 'session_expired' }) } if (url.endsWith('/v1/devices')) { - return options.devicesStatus - ? jsonResponse(options.devicesStatus, { error: 'nope' }) - : jsonResponse(200, { registrationId: 'reg-1' }) + if (options.devicesStatus) { + return jsonResponse(options.devicesStatus, { error: 'nope' }) + } + knownRegistrations.add('reg-1') + return jsonResponse(200, { registrationId: 'reg-1' }) } if (url.endsWith('/v1/send')) { return jsonResponse(200, { results: [{ registrationId: 'reg-1', status: 'queued' }] }) } - return new Response(null, { status: 204 }) + // Why explicit: a catch-all 204 would report every delete as accepted and + // leave the 404 branch of deleteDevice untested. + const deleted = /\/v1\/devices\/([^/]+)$/.exec(url) + if (deleted && init?.method === 'DELETE') { + const registrationId = decodeURIComponent(deleted[1] ?? '') + return new Response(null, { status: knownRegistrations.has(registrationId) ? 204 : 404 }) + } + throw new Error(`unexpected request: ${init?.method ?? 'GET'} ${url}`) }) as unknown as typeof globalThis.fetch return { @@ -193,8 +205,37 @@ describe('PushGatewayClient', () => { }) }) + it('treats a delete the gateway accepted as done', async () => { + const gateway = createFakeGateway() + await gateway.client.registerDevice(REGISTER_INPUT) + + expect(await gateway.client.deleteDevice('reg-1')).toEqual({ deleted: true, retryable: false }) + expect(gateway.calls.at(-1)).toMatchObject({ method: 'DELETE' }) + }) + it('treats a delete of an unknown registration as done', async () => { const gateway = createFakeGateway() - expect(await gateway.client.deleteDevice('reg-1')).toEqual({ deleted: true, retryable: false }) + + expect(await gateway.client.deleteDevice('reg-gone')).toEqual({ + deleted: true, + retryable: false + }) + }) + + it('reports a 401 that survives the forced re-auth as unreachable', async () => { + const gateway = createFakeGateway({ rejectBearer: true }) + + expect(await gateway.client.registerDevice(REGISTER_INPUT)).toEqual({ + ok: false, + reason: 'unreachable' + }) + // Exactly one forced re-auth, not a handshake loop. + expect(gateway.calls.filter((call) => call.url.endsWith('/v1/host/challenge'))).toHaveLength(2) + }) + + it('keeps an unreachable-classified 401 retryable for a queued delete', async () => { + const gateway = createFakeGateway({ rejectBearer: true }) + + expect(await gateway.client.deleteDevice('reg-1')).toEqual({ deleted: false, retryable: true }) }) }) diff --git a/src/main/runtime/push/push-gateway-client.ts b/src/main/runtime/push/push-gateway-client.ts index ec1d62a1529..0073632e7d5 100644 --- a/src/main/runtime/push/push-gateway-client.ts +++ b/src/main/runtime/push/push-gateway-client.ts @@ -4,7 +4,6 @@ import { z } from 'zod' import { cancelUnreadResponseBody } from '../../lib/unread-response-body' import type { E2EEKeypair } from '../e2ee-keypair' -import { deriveRelayHostId } from '../relay/relay-http-client' import type { MobilePushAgentState, MobilePushApnsEnvironment, @@ -12,33 +11,16 @@ import type { MobilePushPlatform, MobilePushSource } from '../../../shared/mobile-push-contract' -import { answerPushHostChallenge, type PushHostChallenge } from './push-host-proof' +import { + PUSH_REQUEST_DEADLINE_MS, + readPushGatewayJson, + type PushGatewayFailure, + type PushGatewayResponse, + type PushGatewayResult +} from './push-gateway-response' +import { PushGatewaySession } from './push-gateway-session' -const PUSH_REQUEST_DEADLINE_MS = 15_000 -// Re-auth a little early so a send never spends its one retry on a token that -// expired between the check and the request. -const SESSION_RENEWAL_MARGIN_MS = 60_000 - -const ChallengeResponseSchema = z - .object({ - challengeId: z.string().min(1).max(512), - gatewayEphemeralPublicKeyB64: z.string().min(1).max(128), - nonceB64: z.string().min(1).max(128), - ciphertextB64: z - .string() - .min(1) - .max(8 * 1024), - expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER) - }) - .strict() - -const SessionResponseSchema = z - .object({ - sessionToken: z.string().min(1).max(1024), - expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER), - hostFingerprint: z.string().min(1).max(64) - }) - .strict() +export type { PushGatewayFailure, PushGatewayResult } const RegisterResponseSchema = z.object({ registrationId: z.string().min(1).max(512) }) @@ -55,9 +37,6 @@ const SendResponseSchema = z.object({ export type PushSendResult = z.infer['results'][number] -export type PushGatewayFailure = { ok: false; reason: 'unreachable' | 'rejected' } -export type PushGatewayResult = ({ ok: true } & T) | PushGatewayFailure - export type PushSendNotification = { notificationId?: string notificationSeq: number @@ -76,23 +55,24 @@ type PushGatewayClientOptions = { now?: () => number } -type CachedSession = { token: string; expiresAt: number } +type AuthorizedResponse = { ok: true; response: Response; token: string } | PushGatewayFailure export class PushGatewayClient { private readonly origin: string - private readonly keypair: E2EEKeypair private readonly fetchImpl: typeof globalThis.fetch - private readonly now: () => number + private readonly session: PushGatewaySession readonly hostFingerprint: string - private session: CachedSession | null = null - private pendingSession: Promise | null = null constructor(options: PushGatewayClientOptions) { this.origin = new URL(options.gatewayUrl).origin - this.keypair = options.keypair this.fetchImpl = options.fetch ?? globalThis.fetch - this.now = options.now ?? Date.now - this.hostFingerprint = deriveRelayHostId(options.keypair.publicKey) + this.session = new PushGatewaySession({ + origin: this.origin, + keypair: options.keypair, + fetchImpl: this.fetchImpl, + now: options.now ?? Date.now + }) + this.hostFingerprint = this.session.hostFingerprint } async registerDevice(input: { @@ -113,7 +93,7 @@ export class PushGatewayClient { filter: { sources: [...input.filter.sources], agentStates: [...input.filter.agentStates] } } }) - const parsed = await readJson(response, RegisterResponseSchema) + const parsed = await readPushGatewayJson(response, RegisterResponseSchema) return parsed.ok ? { ok: true, registrationId: parsed.value.registrationId } : parsed } @@ -143,143 +123,53 @@ export class PushGatewayClient { notification: input.notification } }) - const parsed = await readJson(response, SendResponseSchema) + const parsed = await readPushGatewayJson(response, SendResponseSchema) return parsed.ok ? { ok: true, results: parsed.value.results } : parsed } private async authorized( path: string, init: { method: string; body?: unknown } - ): Promise<{ ok: true; response: Response } | PushGatewayFailure> { - const first = await this.sendAuthorized(path, init, false) - // A 401 means the cached session died server-side; one forced re-auth, then give up. - if (first.ok && first.response.status === 401) { - await cancelUnreadResponseBody(first.response) - this.session = null - return await this.sendAuthorized(path, init, true) + ): Promise { + const first = await this.sendAuthorized(path, init, null) + if (!first.ok || first.response.status !== 401) { + return first } - return first + // A 401 means that one session died server-side; one forced re-auth, then stop. + await cancelUnreadResponseBody(first.response) + const retried = await this.sendAuthorized(path, init, first.token) + if (retried.ok && retried.response.status === 401) { + await cancelUnreadResponseBody(retried.response) + // A 401 that survives a freshly minted session is the gateway being unusable + // right now, not this request being wrong: register should report it as + // unreachable, and send should still spend its one retry. + return { ok: false, reason: 'unreachable' } + } + return retried } private async sendAuthorized( path: string, init: { method: string; body?: unknown }, - forceReauth: boolean - ): Promise<{ ok: true; response: Response } | PushGatewayFailure> { - const session = await this.ensureSession(forceReauth) - if (!session) { - return { ok: false, reason: 'unreachable' } + staleToken: string | null + ): Promise { + const outcome = await this.session.ensure(staleToken) + if (!outcome.ok) { + return outcome } try { const response = await this.fetchImpl(`${this.origin}${path}`, { method: init.method, headers: { - authorization: `Bearer ${session.token}`, + authorization: `Bearer ${outcome.session.token}`, ...(init.body === undefined ? {} : { 'content-type': 'application/json' }) }, signal: AbortSignal.timeout(PUSH_REQUEST_DEADLINE_MS), ...(init.body === undefined ? {} : { body: JSON.stringify(init.body) }) }) - return { ok: true, response } - } catch { - return { ok: false, reason: 'unreachable' } - } - } - - private async ensureSession(forceReauth: boolean): Promise { - if (forceReauth) { - this.session = null - } - const cached = this.session - if (cached && cached.expiresAt - SESSION_RENEWAL_MARGIN_MS > this.now()) { - return cached - } - // Concurrent sends must not each burn a challenge; share one handshake. - this.pendingSession ??= this.openSession().finally(() => { - this.pendingSession = null - }) - return await this.pendingSession - } - - private async openSession(): Promise { - const challenge = await this.requestChallenge() - if (!challenge) { - return null - } - const proofB64 = answerPushHostChallenge(challenge, { - gatewayOrigin: this.origin, - hostFingerprint: this.hostFingerprint, - hostPublicKey: this.keypair.publicKey, - hostSecretKey: this.keypair.secretKey, - now: this.now - }) - if (!proofB64) { - return null - } - const response = await this.post('/v1/host/session', { - v: 1, - challengeId: challenge.challengeId, - proofB64 - }) - const parsed = await readJson(response, SessionResponseSchema) - if (!parsed.ok || parsed.value.hostFingerprint !== this.hostFingerprint) { - return null - } - this.session = { token: parsed.value.sessionToken, expiresAt: parsed.value.expiresAt } - return this.session - } - - private async requestChallenge(): Promise { - const response = await this.post('/v1/host/challenge', { - v: 1, - hostPublicKeyB64: this.keypair.publicKeyB64 - }) - const parsed = await readJson(response, ChallengeResponseSchema) - return parsed.ok ? parsed.value : null - } - - private async post( - path: string, - body: unknown - ): Promise<{ ok: true; response: Response } | PushGatewayFailure> { - try { - const response = await this.fetchImpl(`${this.origin}${path}`, { - method: 'POST', - headers: { 'content-type': 'application/json' }, - signal: AbortSignal.timeout(PUSH_REQUEST_DEADLINE_MS), - body: JSON.stringify(body) - }) - return { ok: true, response } + return { ok: true, response, token: outcome.session.token } } catch { return { ok: false, reason: 'unreachable' } } } } - -async function readJson( - result: { ok: true; response: Response } | PushGatewayFailure, - schema: TSchema -): Promise<{ ok: true; value: z.infer } | PushGatewayFailure> { - if (!result.ok) { - return result - } - const { response } = result - if (!response.ok) { - await cancelUnreadResponseBody(response) - // 5xx and 429 are worth another attempt later; anything else is the gateway - // refusing this request as written. - return { - ok: false, - reason: response.status >= 500 || response.status === 429 ? 'unreachable' : 'rejected' - } - } - let payload: unknown - try { - payload = await response.json() - } catch { - await cancelUnreadResponseBody(response) - return { ok: false, reason: 'unreachable' } - } - const parsed = schema.safeParse(payload) - return parsed.success ? { ok: true, value: parsed.data } : { ok: false, reason: 'rejected' } -} diff --git a/src/main/runtime/push/push-gateway-response.ts b/src/main/runtime/push/push-gateway-response.ts new file mode 100644 index 00000000000..fcc9e724bb1 --- /dev/null +++ b/src/main/runtime/push/push-gateway-response.ts @@ -0,0 +1,58 @@ +// Why: the authorized request path and the handshake that authorizes it must +// classify a gateway response identically — otherwise the same 503 means "retry" +// on one leg and "give up" on the other, and register/send disagree about why. +import type { z } from 'zod' +import { cancelUnreadResponseBody } from '../../lib/unread-response-body' + +export const PUSH_REQUEST_DEADLINE_MS = 15_000 + +export type PushGatewayFailure = { ok: false; reason: 'unreachable' | 'rejected' } +export type PushGatewayResult = ({ ok: true } & T) | PushGatewayFailure +export type PushGatewayResponse = { ok: true; response: Response } | PushGatewayFailure + +/** Unauthenticated POST; the handshake legs run before any session exists. */ +export async function postPushGatewayJson( + fetchImpl: typeof globalThis.fetch, + url: string, + body: unknown +): Promise { + try { + const response = await fetchImpl(url, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + signal: AbortSignal.timeout(PUSH_REQUEST_DEADLINE_MS), + body: JSON.stringify(body) + }) + return { ok: true, response } + } catch { + return { ok: false, reason: 'unreachable' } + } +} + +export async function readPushGatewayJson( + result: PushGatewayResponse, + schema: TSchema +): Promise<{ ok: true; value: z.infer } | PushGatewayFailure> { + if (!result.ok) { + return result + } + const { response } = result + if (!response.ok) { + await cancelUnreadResponseBody(response) + // 5xx and 429 are worth another attempt later; anything else is the gateway + // refusing this request as written. + return { + ok: false, + reason: response.status >= 500 || response.status === 429 ? 'unreachable' : 'rejected' + } + } + let payload: unknown + try { + payload = await response.json() + } catch { + await cancelUnreadResponseBody(response) + return { ok: false, reason: 'unreachable' } + } + const parsed = schema.safeParse(payload) + return parsed.success ? { ok: true, value: parsed.data } : { ok: false, reason: 'rejected' } +} diff --git a/src/main/runtime/push/push-gateway-session.test.ts b/src/main/runtime/push/push-gateway-session.test.ts new file mode 100644 index 00000000000..8527430365a --- /dev/null +++ b/src/main/runtime/push/push-gateway-session.test.ts @@ -0,0 +1,169 @@ +import { createHash } from 'node:crypto' +import { describe, expect, it, vi } from 'vitest' +import { buildPushChallengeFixture, createPushHostKeypair } from './push-host-challenge-fixtures' +import { PushGatewaySession, type PushSessionOutcome } from './push-gateway-session' + +const GATEWAY_ORIGIN = 'https://push.onorca.dev' +const NOW = 1_770_000_000_000 + +function jsonResponse(status: number, body: unknown): Response { + return new Response(JSON.stringify(body), { + status, + headers: { 'content-type': 'application/json' } + }) +} + +function tokenOf(outcome: PushSessionOutcome): string | null { + return outcome.ok ? outcome.session.token : null +} + +function createSessionHarness( + options: { sessionStatus?: number; challengeStatus?: number; wrongFingerprint?: boolean } = {} +): { + session: PushGatewaySession + challenges: () => number + requests: () => number + now: { value: number } +} { + const hostKeypair = createPushHostKeypair() + const hostFingerprint = createHash('sha256') + .update(hostKeypair.publicKey) + .digest('base64url') + .slice(0, 16) + const now = { value: NOW } + let issued = 0 + let requests = 0 + let pendingProof: string | null = null + + const fetchImpl = (async (input: string, init?: RequestInit): Promise => { + const url = String(input) + requests += 1 + if (url.endsWith('/v1/host/challenge')) { + if (options.challengeStatus) { + return jsonResponse(options.challengeStatus, { error: 'rate_limited' }) + } + const built = buildPushChallengeFixture({ + hostKeypair, + gatewayOrigin: GATEWAY_ORIGIN, + hostFingerprint, + issuedAt: now.value, + challengeId: `challenge-${++issued}` + }) + pendingProof = built.proof + return jsonResponse(200, built.challenge) + } + if (options.sessionStatus) { + return jsonResponse(options.sessionStatus, { error: 'nope' }) + } + const body = init?.body ? (JSON.parse(String(init.body)) as { proofB64: string }) : null + if (body?.proofB64 !== pendingProof) { + return jsonResponse(401, { error: 'bad_proof' }) + } + return jsonResponse(200, { + sessionToken: `session-${issued}`, + expiresAt: now.value + 24 * 60 * 60_000, + hostFingerprint: options.wrongFingerprint ? 'someone-else' : hostFingerprint + }) + }) as unknown as typeof globalThis.fetch + + return { + session: new PushGatewaySession({ + origin: GATEWAY_ORIGIN, + keypair: hostKeypair, + fetchImpl, + now: () => now.value + }), + challenges: () => issued, + requests: () => requests, + now + } +} + +describe('PushGatewaySession', () => { + it('reuses the cached session until it nears expiry', async () => { + const harness = createSessionHarness() + + expect(tokenOf(await harness.session.ensure(null))).toBe('session-1') + expect(tokenOf(await harness.session.ensure(null))).toBe('session-1') + expect(harness.challenges()).toBe(1) + }) + + it('drops only the exact session that received the 401', async () => { + const harness = createSessionHarness() + expect(tokenOf(await harness.session.ensure(null))).toBe('session-1') + + // A request that 401ed on session-1 forces a fresh handshake. + expect(tokenOf(await harness.session.ensure('session-1'))).toBe('session-2') + // A second request whose 401 also named session-1 must keep the new token. + expect(tokenOf(await harness.session.ensure('session-1'))).toBe('session-2') + expect(harness.challenges()).toBe(2) + }) + + it('reports a refused handshake as rejected rather than unreachable', async () => { + const harness = createSessionHarness({ sessionStatus: 403 }) + + expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'rejected' }) + }) + + it('reports a session minted for another host as rejected', async () => { + const harness = createSessionHarness({ wrongFingerprint: true }) + + expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'rejected' }) + }) + + it('caches a refusal briefly instead of re-handshaking on every call', async () => { + const harness = createSessionHarness({ sessionStatus: 403 }) + + await harness.session.ensure(null) + await harness.session.ensure(null) + expect(harness.challenges()).toBe(1) + + harness.now.value += 30_000 + await harness.session.ensure(null) + expect(harness.challenges()).toBe(2) + }) + + it('never caches a transport failure, which may clear on the next try', async () => { + const fetchImpl = vi.fn(async () => { + throw new Error('network down') + }) as unknown as typeof globalThis.fetch + const session = new PushGatewaySession({ + origin: GATEWAY_ORIGIN, + keypair: createPushHostKeypair(), + fetchImpl, + now: () => NOW + }) + + expect(await session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' }) + expect(await session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' }) + expect(fetchImpl).toHaveBeenCalledTimes(2) + }) + + it('reports a rate-limited challenge as unreachable and backs off', async () => { + const harness = createSessionHarness({ challengeStatus: 429 }) + + expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' }) + expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' }) + expect(harness.requests()).toBe(1) + + harness.now.value += 60_000 + await harness.session.ensure(null) + expect(harness.requests()).toBe(2) + }) + + it('reports a rate-limited session mint as unreachable, not refused', async () => { + const harness = createSessionHarness({ sessionStatus: 429 }) + + expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' }) + // Cached for a minute, so the next dispatch does not spend more of the bucket. + expect(await harness.session.ensure(null)).toEqual({ ok: false, reason: 'unreachable' }) + expect(harness.challenges()).toBe(1) + }) + + it('shares one handshake across concurrent callers', async () => { + const harness = createSessionHarness() + + await Promise.all([harness.session.ensure(null), harness.session.ensure(null)]) + expect(harness.challenges()).toBe(1) + }) +}) diff --git a/src/main/runtime/push/push-gateway-session.ts b/src/main/runtime/push/push-gateway-session.ts new file mode 100644 index 00000000000..dd50b813f1d --- /dev/null +++ b/src/main/runtime/push/push-gateway-session.ts @@ -0,0 +1,157 @@ +// Why: the challenge/proof handshake every push request rides on, split out of +// push-gateway-client.ts so the session cache and its refusal cache stay readable +// next to the request methods rather than buried under them. +import { z } from 'zod' +import { cancelUnreadResponseBody } from '../../lib/unread-response-body' +import type { E2EEKeypair } from '../e2ee-keypair' +import { deriveRelayHostId } from '../relay/relay-http-client' +import { answerPushHostChallenge } from './push-host-proof' +import { + postPushGatewayJson, + readPushGatewayJson, + type PushGatewayFailure +} from './push-gateway-response' + +// Re-auth a little early so a send never spends its one retry on a token that +// expired between the check and the request. +const SESSION_RENEWAL_MARGIN_MS = 60_000 +// Why: a gateway that refuses this host's proof refuses the identical next one, +// so without this every dispatch pays two full handshake round trips to relearn it. +const HANDSHAKE_REFUSAL_TTL_MS = 30_000 +// Why: the handshake routes sit behind a per-IP bucket. Backing off keeps this +// host from spending the whole bucket on challenges it will never get to use. +const HANDSHAKE_RATE_LIMIT_TTL_MS = 60_000 + +const ChallengeResponseSchema = z + .object({ + challengeId: z.string().min(1).max(512), + gatewayEphemeralPublicKeyB64: z.string().min(1).max(128), + nonceB64: z.string().min(1).max(128), + ciphertextB64: z + .string() + .min(1) + .max(8 * 1024), + expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER) + }) + .strict() + +const SessionResponseSchema = z + .object({ + sessionToken: z.string().min(1).max(1024), + expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER), + hostFingerprint: z.string().min(1).max(64) + }) + .strict() + +export type PushSession = { token: string; expiresAt: number } +export type PushSessionOutcome = { ok: true; session: PushSession } | PushGatewayFailure + +type PushGatewaySessionOptions = { + origin: string + keypair: E2EEKeypair + fetchImpl: typeof globalThis.fetch + now: () => number +} + +export class PushGatewaySession { + private readonly origin: string + private readonly keypair: E2EEKeypair + private readonly fetchImpl: typeof globalThis.fetch + private readonly now: () => number + readonly hostFingerprint: string + private session: PushSession | null = null + private pending: Promise | null = null + private negative: { until: number; reason: PushGatewayFailure['reason'] } | null = null + + constructor(options: PushGatewaySessionOptions) { + this.origin = options.origin + this.keypair = options.keypair + this.fetchImpl = options.fetchImpl + this.now = options.now + this.hostFingerprint = deriveRelayHostId(options.keypair.publicKey) + } + + /** + * `staleToken` is the token that just received a 401. Only that exact session is + * dropped: a concurrent request may already have installed a good one, and + * clearing unconditionally would throw it away and re-handshake for nothing. + */ + async ensure(staleToken: string | null): Promise { + if (staleToken !== null && this.session?.token === staleToken) { + this.session = null + } + const cached = this.session + if (cached && cached.expiresAt - SESSION_RENEWAL_MARGIN_MS > this.now()) { + return { ok: true, session: cached } + } + if (this.negative && this.negative.until > this.now()) { + return { ok: false, reason: this.negative.reason } + } + // Concurrent sends must not each burn a challenge; share one handshake. + this.pending ??= this.open().finally(() => { + this.pending = null + }) + return await this.pending + } + + private async open(): Promise { + const challenge = await this.handshakePost( + '/v1/host/challenge', + { v: 1, hostPublicKeyB64: this.keypair.publicKeyB64 }, + ChallengeResponseSchema + ) + if (!challenge.ok) { + return this.remember(challenge) + } + const proofB64 = answerPushHostChallenge(challenge.value, { + gatewayOrigin: this.origin, + hostFingerprint: this.hostFingerprint, + hostPublicKey: this.keypair.publicKey, + hostSecretKey: this.keypair.secretKey, + now: this.now + }) + if (!proofB64) { + // A challenge this host cannot answer is a refusal, not a dropped packet. + return this.remember({ ok: false, reason: 'rejected' }) + } + const parsed = await this.handshakePost( + '/v1/host/session', + { v: 1, challengeId: challenge.value.challengeId, proofB64 }, + SessionResponseSchema + ) + if (!parsed.ok) { + return this.remember(parsed) + } + if (parsed.value.hostFingerprint !== this.hostFingerprint) { + // The gateway answered for some other host; that token is never usable here. + return this.remember({ ok: false, reason: 'rejected' }) + } + this.session = { token: parsed.value.sessionToken, expiresAt: parsed.value.expiresAt } + this.negative = null + return { ok: true, session: this.session } + } + + private async handshakePost( + path: string, + body: unknown, + schema: TSchema + ): Promise<{ ok: true; value: z.infer } | PushGatewayFailure> { + const response = await postPushGatewayJson(this.fetchImpl, `${this.origin}${path}`, body) + if (response.ok && response.response.status === 429) { + await cancelUnreadResponseBody(response.response) + // Rate limiting refuses the moment, not this host: back off, stay retryable + // so register reports gateway_unreachable and send keeps its one retry. + this.negative = { until: this.now() + HANDSHAKE_RATE_LIMIT_TTL_MS, reason: 'unreachable' } + return { ok: false, reason: 'unreachable' } + } + return await readPushGatewayJson(response, schema) + } + + /** Caches refusals only: a transport failure may clear on the very next try. */ + private remember(failure: PushGatewayFailure): PushGatewayFailure { + if (failure.reason === 'rejected') { + this.negative = { until: this.now() + HANDSHAKE_REFUSAL_TTL_MS, reason: 'rejected' } + } + return failure + } +} diff --git a/src/main/startup/main-process-quit.ts b/src/main/startup/main-process-quit.ts index a4149e13ba7..de580bf7d95 100644 --- a/src/main/startup/main-process-quit.ts +++ b/src/main/startup/main-process-quit.ts @@ -72,6 +72,9 @@ function installBeforeQuitHandler(): void { } state.isQuitting = true state.desktopRelayService?.fenceAndCloseNow() + // Why: drops the notification subscription so a late dispatch cannot start a + // push (and its unref'd outbox retry) on the way out. + state.desktopPushService?.stop() state.runtimeRpc?.setMobileRelayPairingProvider(null) state.unsubscribeAgentAwakeStatusChanges?.() state.unsubscribeAgentAwakeStatusChanges = null diff --git a/src/shared/mobile-push-contract.ts b/src/shared/mobile-push-contract.ts index bb7ba902fda..fd05ea1e509 100644 --- a/src/shared/mobile-push-contract.ts +++ b/src/shared/mobile-push-contract.ts @@ -38,7 +38,17 @@ export type MobilePushRegisterInput = { export type MobilePushRegisterResult = | { registered: true; registrationId: string } - | { registered: false; reason: 'gateway_unreachable' | 'gateway_rejected' | 'not_mobile' } + | { + registered: false + // `registration_storage_failed`: the gateway accepted the token but the host + // could not persist it, so the phone must register again rather than believe + // a push route that does not exist. + reason: + | 'gateway_unreachable' + | 'gateway_rejected' + | 'not_mobile' + | 'registration_storage_failed' + } function isStringMember(value: unknown, members: readonly T[]): value is T { return typeof value === 'string' && (members as readonly string[]).includes(value)