From 6fe396e49b76e3a817b47010032c8cb73d8a3742 Mon Sep 17 00:00:00 2001 From: Gabe Debes Date: Sat, 10 Oct 2026 01:25:35 -0700 Subject: [PATCH] [build-tools] Supervise device-session tunnels Co-Authored-By: GPT-6.1-Sol --- .changeset/serve-sim-pr-4526.md | 5 + .../startAgentDeviceRemoteSession.ts | 1 + .../functions/startAppiumRemoteSession.ts | 1 + .../functions/startArgentRemoteSession.ts | 1 + .../steps/utils/__tests__/ngrokTunnel.test.ts | 637 ++++++++++++++++++ .../__tests__/remoteDeviceRunSession.test.ts | 8 +- .../src/steps/utils/deviceSessionHost.ts | 1 + .../src/steps/utils/ngrokTunnel.ts | 315 +++++++++ .../src/steps/utils/remoteDeviceRunSession.ts | 60 +- 9 files changed, 965 insertions(+), 64 deletions(-) create mode 100644 .changeset/serve-sim-pr-4526.md create mode 100644 packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts create mode 100644 packages/build-tools/src/steps/utils/ngrokTunnel.ts diff --git a/.changeset/serve-sim-pr-4526.md b/.changeset/serve-sim-pr-4526.md new file mode 100644 index 0000000000..9cc7ee01be --- /dev/null +++ b/.changeset/serve-sim-pr-4526.md @@ -0,0 +1,5 @@ +--- +'@expo/build-tools': patch +--- + +Recover unhealthy device-session tunnels without changing their URLs. diff --git a/packages/build-tools/src/steps/functions/startAgentDeviceRemoteSession.ts b/packages/build-tools/src/steps/functions/startAgentDeviceRemoteSession.ts index 67658e54e2..f56585cf3d 100644 --- a/packages/build-tools/src/steps/functions/startAgentDeviceRemoteSession.ts +++ b/packages/build-tools/src/steps/functions/startAgentDeviceRemoteSession.ts @@ -167,6 +167,7 @@ export async function runAgentDeviceRemoteSessionAsync( subdomainPrefix: 'agent-device', baseDomain: ngrokTunnelDomain, authtoken: ngrokAuthtoken, + healthCheck: { path: '/health' }, logger: taskLogger, }); tasks.signal.throwIfAborted(); diff --git a/packages/build-tools/src/steps/functions/startAppiumRemoteSession.ts b/packages/build-tools/src/steps/functions/startAppiumRemoteSession.ts index 4eb0073ad9..d12bce5fbc 100644 --- a/packages/build-tools/src/steps/functions/startAppiumRemoteSession.ts +++ b/packages/build-tools/src/steps/functions/startAppiumRemoteSession.ts @@ -263,6 +263,7 @@ export function createStartAppiumRemoteSessionBuildFunction( subdomainPrefix: 'appium', baseDomain: ngrokTunnelDomain, authtoken: ngrokAuthtoken, + healthCheck: { path: '/status' }, logger, }); diff --git a/packages/build-tools/src/steps/functions/startArgentRemoteSession.ts b/packages/build-tools/src/steps/functions/startArgentRemoteSession.ts index 2e7be1cc7b..742d20eb69 100644 --- a/packages/build-tools/src/steps/functions/startArgentRemoteSession.ts +++ b/packages/build-tools/src/steps/functions/startArgentRemoteSession.ts @@ -340,6 +340,7 @@ export function createStartArgentRemoteSessionBuildFunction( baseDomain: ngrokTunnelDomain, authtoken: ngrokAuthtoken, rewriteHostHeader: true, + healthCheck: { path: '/tools', statuses: [200, 401] }, logger, }); const publicToolsUrl = toolsTunnel.url; diff --git a/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts b/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts new file mode 100644 index 0000000000..83eb2d4caa --- /dev/null +++ b/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts @@ -0,0 +1,637 @@ +import { type bunyan } from '@expo/logger'; +import * as ngrok from '@ngrok/ngrok'; +import { createServer } from 'node:http'; +import { type AddressInfo } from 'node:net'; + +import { turtleFetch } from '../../../utils/turtleFetch'; +import { startNgrokTunnelAsync } from '../ngrokTunnel'; + +jest.mock('@ngrok/ngrok'); +jest.mock('node:timers', () => ({ + setTimeout: (...args: Parameters) => setTimeout(...args), + clearTimeout: (...args: Parameters) => clearTimeout(...args), +})); +jest.mock('../../../utils/turtleFetch'); + +const { Response } = jest.requireActual('node-fetch'); +const url = 'https://web-preview-fixed.example.test'; +const logger = { info: jest.fn(), warn: jest.fn() } as unknown as bunyan; +const options = { + port: 4321, + subdomainPrefix: 'web-preview', + subdomainId: 'fixed', + baseDomain: 'example.test', + authtoken: 'token', + rewriteHostHeader: true, + healthCheck: { path: '/healthz' }, + logger, +}; + +function listener(publicUrl = url) { + return { url: () => publicUrl, close: jest.fn().mockResolvedValue(undefined) }; +} + +let initial: ReturnType; +let replacement: ReturnType; + +beforeEach(() => { + jest.useFakeTimers(); + jest.clearAllMocks(); + initial = listener(); + replacement = listener(); + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockResolvedValue(replacement as never); + jest + .mocked(turtleFetch) + .mockReset() + .mockResolvedValue(new Response(undefined, { status: 200 })); +}); + +afterEach(() => { + jest.useRealTimers(); +}); + +function failPublicProbes() { + jest.mocked(turtleFetch).mockImplementation( + async target => + new Response(undefined, { + status: target.startsWith('https:') ? 503 : 200, + }) + ); +} + +it('keeps tunnels without a health check unsupervised and closes them on stop', async () => { + const tunnel = await startNgrokTunnelAsync({ ...options, healthCheck: undefined }); + await jest.advanceTimersByTimeAsync(90_000); + expect(turtleFetch).not.toHaveBeenCalled(); + expect(ngrok.forward).toHaveBeenCalledTimes(1); + await tunnel.stopAsync(); + expect(initial.close).toHaveBeenCalledTimes(1); +}); + +it('probes the public service and keeps a healthy listener open', async () => { + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(45_000); + expect(turtleFetch).toHaveBeenCalledTimes(3); + expect(turtleFetch).toHaveBeenCalledWith( + `${url}/healthz`, + 'GET', + expect.objectContaining({ + retries: 0, + redirect: 'manual', + headers: expect.objectContaining({ 'ngrok-skip-browser-warning': 'true' }), + }) + ); + expect(ngrok.forward).toHaveBeenCalledTimes(1); + } finally { + await tunnel.stopAsync(); + } + expect(initial.close).toHaveBeenCalledTimes(1); + await jest.advanceTimersByTimeAsync(60_000); + expect(turtleFetch).toHaveBeenCalledTimes(3); +}); + +it('reopens after three failed public probes with a healthy local service, preserving the endpoint', async () => { + failPublicProbes(); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(30_000); + expect(initial.close).not.toHaveBeenCalled(); + await jest.advanceTimersByTimeAsync(15_000); + expect(initial.close).toHaveBeenCalledTimes(1); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + expect(jest.mocked(ngrok.forward).mock.calls[1][0]).toEqual({ + addr: 4321, + authtoken: 'token', + domain: 'web-preview-fixed.example.test', + request_header_add: ['Host:localhost:4321'], + force_new_session: true, + }); + expect(tunnel.url).toBe(url); + jest.mocked(turtleFetch).mockResolvedValue(new Response(undefined, { status: 200 })); + await jest.advanceTimersByTimeAsync(15_000); + expect(logger.info).toHaveBeenCalledWith(expect.stringContaining('healthy again')); + } finally { + await tunnel.stopAsync(); + } + expect(replacement.close).toHaveBeenCalledTimes(1); +}); + +it('does not reopen when the local service is unhealthy', async () => { + jest.mocked(turtleFetch).mockResolvedValue(new Response(undefined, { status: 503 })); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(90_000); + expect(ngrok.forward).toHaveBeenCalledTimes(1); + expect(initial.close).not.toHaveBeenCalled(); + } finally { + await tunnel.stopAsync(); + } +}); + +it('resets consecutive failures after a successful public probe', async () => { + failPublicProbes(); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(30_000); + jest.mocked(turtleFetch).mockResolvedValueOnce(new Response(undefined, { status: 200 })); + await jest.advanceTimersByTimeAsync(15_000); + await jest.advanceTimersByTimeAsync(30_000); + expect(ngrok.forward).toHaveBeenCalledTimes(1); + } finally { + await tunnel.stopAsync(); + } +}); + +it('retries a failed reopen with backoff', async () => { + failPublicProbes(); + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockRejectedValueOnce(new Error('offline')) + .mockResolvedValue(replacement as never); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(45_000); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(1_999); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(1); + expect(ngrok.forward).toHaveBeenCalledTimes(3); + } finally { + await tunnel.stopAsync(); + } +}); + +it('closes a listener that finishes opening after shutdown, with idempotent cleanup', async () => { + failPublicProbes(); + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockReturnValueOnce( + new Promise(resolve => { + resolveOpening = resolve; + }) + ); + const tunnel = await startNgrokTunnelAsync(options); + await jest.advanceTimersByTimeAsync(45_000); + const stopping = tunnel.stopAsync(); + expect(tunnel.stopAsync()).toBe(stopping); + const stopped = jest.fn(); + void stopping.then(stopped); + await jest.advanceTimersByTimeAsync(0); + expect(stopped).not.toHaveBeenCalled(); + resolveOpening(replacement as never); + await stopping; + expect(replacement.close).toHaveBeenCalledTimes(1); + await jest.advanceTimersByTimeAsync(120_000); + expect(ngrok.forward).toHaveBeenCalledTimes(2); +}); + +it.each([0, 4, 5, 10])('retires a reopen cancelled after %s promise turns', async turns => { + failPublicProbes(); + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockReturnValueOnce(new Promise(resolve => (resolveOpening = resolve))); + const tunnel = await startNgrokTunnelAsync(options); + await jest.advanceTimersByTimeAsync(45_000); + resolveOpening(replacement as never); + for (let turn = 0; turn < turns; turn++) { + await Promise.resolve(); + } + await tunnel.stopAsync(); + await jest.advanceTimersByTimeAsync(0); + expect(replacement.close).toHaveBeenCalledTimes(1); +}); + +it('retries a rejected close when a reopen finishes after shutdown', async () => { + failPublicProbes(); + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockReturnValueOnce(new Promise(resolve => (resolveOpening = resolve))); + const tunnel = await startNgrokTunnelAsync(options); + await jest.advanceTimersByTimeAsync(45_000); + const stopping = tunnel.stopAsync(); + replacement.close.mockRejectedValueOnce(new Error('temporary unlisten failure')); + resolveOpening(replacement as never); + await jest.advanceTimersByTimeAsync(250); + await stopping; + expect(replacement.close).toHaveBeenCalledTimes(2); + const warnings = jest.mocked(logger.warn).mock.calls.length; + await jest.advanceTimersByTimeAsync(120_000); + expect(logger.warn).toHaveBeenCalledTimes(warnings); +}); + +it('fails stop if an in-flight reopen still has not settled by the stop deadline', async () => { + failPublicProbes(); + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockReturnValueOnce(new Promise(resolve => (resolveOpening = resolve))); + const tunnel = await startNgrokTunnelAsync(options); + await jest.advanceTimersByTimeAsync(45_000); + const rejected = expect(tunnel.stopAsync()).rejects.toThrow( + 'Ngrok tunnel web-preview-fixed.example.test stop timed out after 4000ms' + ); + await jest.advanceTimersByTimeAsync(4_000); + await rejected; + expect(ngrok.forward).toHaveBeenCalledTimes(2); + resolveOpening(replacement as never); + await jest.advanceTimersByTimeAsync(0); + expect(replacement.close).toHaveBeenCalledTimes(1); +}); + +it('fails stop if a pending reopen arrives but cannot be retired', async () => { + failPublicProbes(); + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockReturnValueOnce(new Promise(resolve => (resolveOpening = resolve))); + replacement.close.mockRejectedValue(new Error('unlisten failed')); + const tunnel = await startNgrokTunnelAsync(options); + await jest.advanceTimersByTimeAsync(45_000); + const rejected = expect(tunnel.stopAsync()).rejects.toThrow( + 'Ngrok tunnel web-preview-fixed.example.test stop timed out' + ); + resolveOpening(replacement as never); + await jest.advanceTimersByTimeAsync(4_000); + await rejected; + expect(replacement.close.mock.calls.length).toBeGreaterThan(1); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ err: expect.any(Error) }), + 'Could not retire late ngrok tunnel web-preview-fixed.example.test.' + ); + const warnings = jest.mocked(logger.warn).mock.calls.length; + await jest.advanceTimersByTimeAsync(120_000); + expect(logger.warn).toHaveBeenCalledTimes(warnings); + expect(ngrok.forward).toHaveBeenCalledTimes(2); +}); + +it('rejects a changed public URL and closes the replacement', async () => { + failPublicProbes(); + const changed = listener('https://different.example.test'); + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockResolvedValue(changed as never); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(45_000); + expect(changed.close).toHaveBeenCalledTimes(1); + expect(tunnel.url).toBe(url); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ err: expect.any(Error) }), + expect.stringContaining('Could not reopen') + ); + } finally { + await tunnel.stopAsync(); + } +}); + +it('treats ngrok error responses as failures even if their status is accepted', async () => { + jest.mocked(turtleFetch).mockImplementation( + async target => + new Response(undefined, { + status: 200, + headers: target.startsWith('https:') ? { 'ngrok-error-code': 'ERR_NGROK_3200' } : {}, + }) + ); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(45_000); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + } finally { + await tunnel.stopAsync(); + } +}); + +it('reopens on a fresh agent session even if closing the old listener fails', async () => { + failPublicProbes(); + initial.close.mockRejectedValueOnce(new Error('agent disconnected')); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(45_000); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + expect(ngrok.forward).toHaveBeenLastCalledWith( + expect.objectContaining({ force_new_session: true }) + ); + expect(ngrok.disconnect).not.toHaveBeenCalled(); + } finally { + await tunnel.stopAsync(); + } + expect(initial.close).toHaveBeenCalledTimes(2); + expect(replacement.close).toHaveBeenCalledTimes(1); +}); + +it('retries a rejected close before reopening an endpoint that remains bound', async () => { + failPublicProbes(); + let domainOccupied = true; + initial.close + .mockRejectedValueOnce(new Error('agent disconnected')) + .mockImplementation(async () => { + domainOccupied = false; + }); + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockImplementation(async () => { + if (domainOccupied) { + throw new Error('endpoint is still bound'); + } + return replacement as never; + }); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(45_000); + expect(initial.close).toHaveBeenCalledTimes(1); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(2_000); + expect(initial.close).toHaveBeenCalledTimes(2); + expect(ngrok.forward).toHaveBeenCalledTimes(3); + expect(tunnel.url).toBe(url); + } finally { + await tunnel.stopAsync(); + } + expect(replacement.close).toHaveBeenCalledTimes(1); +}); + +it('retires an old listener even after its replacement is healthy', async () => { + failPublicProbes(); + initial.close.mockRejectedValueOnce(new Error('agent disconnected')); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(45_000); + expect(initial.close).toHaveBeenCalledTimes(1); + jest.mocked(turtleFetch).mockResolvedValue(new Response(undefined, { status: 200 })); + await jest.advanceTimersByTimeAsync(15_000); + expect(initial.close).toHaveBeenCalledTimes(2); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + } finally { + await tunnel.stopAsync(); + } +}); + +it('retries an old listener without repeating its close warning on every probe', async () => { + failPublicProbes(); + initial.close.mockRejectedValue(new Error('old session disconnected')); + const tunnel = await startNgrokTunnelAsync(options); + await jest.advanceTimersByTimeAsync(45_000); + const warnings = jest.mocked(logger.warn).mock.calls.length; + expect(warnings).toBeGreaterThan(0); + jest.mocked(turtleFetch).mockResolvedValue(new Response(undefined, { status: 200 })); + await jest.advanceTimersByTimeAsync(120_000); + expect(initial.close.mock.calls.length).toBeGreaterThan(1); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + expect(logger.warn).toHaveBeenCalledTimes(warnings); + const rejected = expect(tunnel.stopAsync()).rejects.toThrow( + 'Ngrok tunnel web-preview-fixed.example.test stop timed out' + ); + await jest.advanceTimersByTimeAsync(4_000); + await rejected; + expect(replacement.close).toHaveBeenCalledTimes(1); +}); + +it('retries a rejected close during shutdown', async () => { + const failure = new Error('agent disconnected'); + initial.close.mockRejectedValueOnce(failure); + const tunnel = await startNgrokTunnelAsync(options); + const stopping = tunnel.stopAsync(); + await jest.advanceTimersByTimeAsync(250); + await stopping; + expect(initial.close).toHaveBeenCalledTimes(2); + expect(logger.warn).toHaveBeenCalledTimes(1); + expect(logger.warn).toHaveBeenCalledWith( + { err: failure }, + 'Could not stop ngrok tunnel web-preview-fixed.example.test.' + ); +}); + +it('logs the first close rejection once while bounded shutdown retries it', async () => { + const failure = new Error('agent disconnected'); + initial.close.mockRejectedValue(failure); + const tunnel = await startNgrokTunnelAsync(options); + const rejected = expect(tunnel.stopAsync()).rejects.toThrow('stop timed out'); + await jest.advanceTimersByTimeAsync(4_000); + await rejected; + expect(initial.close.mock.calls.length).toBeGreaterThan(1); + expect(logger.warn).toHaveBeenCalledTimes(1); + expect(logger.warn).toHaveBeenCalledWith( + { err: failure }, + 'Could not stop ngrok tunnel web-preview-fixed.example.test.' + ); +}); + +it('bounds shutdown during recovery without duplicating a pending native close', async () => { + failPublicProbes(); + let finishClose!: () => void; + initial.close.mockReturnValueOnce(new Promise(resolve => (finishClose = resolve))); + const tunnel = await startNgrokTunnelAsync(options); + await jest.advanceTimersByTimeAsync(45_000); + const stopping = tunnel.stopAsync(); + expect(tunnel.stopAsync()).toBe(stopping); + const rejected = expect(stopping).rejects.toThrow( + 'Ngrok tunnel web-preview-fixed.example.test stop timed out after 4000ms' + ); + await jest.advanceTimersByTimeAsync(4_000); + await rejected; + expect(initial.close).toHaveBeenCalledTimes(1); + expect(ngrok.forward).toHaveBeenCalledTimes(1); + const warnings = jest.mocked(logger.warn).mock.calls.length; + finishClose(); + await jest.advanceTimersByTimeAsync(120_000); + expect(logger.warn).toHaveBeenCalledTimes(warnings); + expect(initial.close).toHaveBeenCalledTimes(1); +}); + +it('bounds a stalled listener close before reopening', async () => { + failPublicProbes(); + initial.close.mockImplementationOnce(async () => await new Promise(() => {})); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(49_999); + expect(ngrok.forward).toHaveBeenCalledTimes(1); + await jest.advanceTimersByTimeAsync(1); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + } finally { + const stopping = tunnel.stopAsync(); + const rejected = expect(stopping).rejects.toThrow( + 'Ngrok tunnel web-preview-fixed.example.test stop timed out' + ); + await jest.advanceTimersByTimeAsync(4_000); + await rejected; + } +}); + +it('does not queue more native opens while a timed-out open is still pending', async () => { + failPublicProbes(); + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockReturnValueOnce( + new Promise(resolve => { + resolveOpening = resolve; + }) + ) + .mockResolvedValue(replacement as never); + const tunnel = await startNgrokTunnelAsync(options); + const late = listener(); + try { + await jest.advanceTimersByTimeAsync(180_000); + expect(ngrok.forward).toHaveBeenCalledTimes(2); + resolveOpening(late as never); + await jest.advanceTimersByTimeAsync(30_000); + expect(late.close).toHaveBeenCalledTimes(1); + expect(ngrok.forward).toHaveBeenCalledTimes(3); + } finally { + await tunnel.stopAsync(); + } + expect(replacement.close).toHaveBeenCalledTimes(1); +}); + +it('bounds initial tunnel creation and retires a late listener', async () => { + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockReturnValueOnce( + new Promise(resolve => { + resolveOpening = resolve; + }) + ); + const opening = startNgrokTunnelAsync(options); + const rejected = expect(opening).rejects.toThrow('Ngrok tunnel open timed out'); + await jest.advanceTimersByTimeAsync(15_000); + await rejected; + resolveOpening(initial as never); + await jest.advanceTimersByTimeAsync(0); + expect(initial.close).toHaveBeenCalledTimes(1); +}); + +it('retries a rejected close after initial tunnel creation timed out', async () => { + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockReturnValueOnce(new Promise(resolve => (resolveOpening = resolve))); + const opening = startNgrokTunnelAsync(options); + const rejected = expect(opening).rejects.toThrow('Ngrok tunnel open timed out'); + await jest.advanceTimersByTimeAsync(15_000); + await rejected; + initial.close.mockRejectedValueOnce(new Error('temporary unlisten failure')); + resolveOpening(initial as never); + await jest.advanceTimersByTimeAsync(250); + expect(initial.close).toHaveBeenCalledTimes(2); +}); + +it('retires only the late listener without blaming it for an older stuck listener', async () => { + failPublicProbes(); + initial.close.mockRejectedValue(new Error('old session disconnected')); + let resolveOpening!: (value: ngrok.Listener) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockReturnValueOnce(new Promise(resolve => (resolveOpening = resolve))) + .mockResolvedValue(replacement as never); + const tunnel = await startNgrokTunnelAsync(options); + try { + await jest.advanceTimersByTimeAsync(60_000); + const late = listener(); + resolveOpening(late as never); + jest.mocked(turtleFetch).mockResolvedValue(new Response(undefined, { status: 200 })); + await jest.advanceTimersByTimeAsync(4_000); + expect(late.close).toHaveBeenCalledTimes(1); + expect( + jest + .mocked(logger.warn) + .mock.calls.some(([, message]) => message?.includes('Could not retire late')) + ).toBe(false); + } finally { + initial.close.mockResolvedValue(undefined); + await tunnel.stopAsync(); + } +}); + +it('recovers using actual HTTP health probes with a local ngrok adapter', async () => { + const actualFetch = jest.requireActual('node-fetch').default; + const realSetTimeout = jest.requireActual('node:timers').setTimeout; + let broken = true; + const local = createServer((_req, res) => { + res.writeHead(200); + res.end('healthy'); + }); + const publicServer = createServer((_req, res) => { + res.writeHead(broken ? 503 : 200); + res.end('probe'); + }); + await Promise.all( + [local, publicServer].map( + server => new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + ) + ); + const publicUrl = `http://127.0.0.1:${(publicServer.address() as AddressInfo).port}`; + let completedRequests = 0; + jest.mocked(turtleFetch).mockImplementation(async (target, method, requestOptions) => { + const response = await actualFetch(target, { ...requestOptions, method }); + completedRequests++; + return response; + }); + const first = listener(publicUrl); + const next = listener(publicUrl); + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(first as never) + .mockImplementation(async () => { + broken = false; + return next as never; + }); + const tunnel = await startNgrokTunnelAsync({ + ...options, + port: (local.address() as AddressInfo).port, + }); + try { + for (let count = 0; count < 3; count++) { + await jest.advanceTimersByTimeAsync(15_000); + for (let turns = 0; turns < 50 && completedRequests < (count + 1) * 2; turns++) { + await new Promise(resolve => realSetTimeout(resolve, 10)); + } + expect(completedRequests).toBe((count + 1) * 2); + } + expect(ngrok.forward).toHaveBeenCalledTimes(2); + expect(first.close).toHaveBeenCalledTimes(1); + expect((await actualFetch(`${tunnel.url}/healthz`)).status).toBe(200); + } finally { + await tunnel.stopAsync(); + await Promise.all( + [local, publicServer].map( + server => + new Promise((resolve, reject) => + server.close(err => (err ? reject(err) : resolve())) + ) + ) + ); + } + expect(next.close).toHaveBeenCalledTimes(1); +}); diff --git a/packages/build-tools/src/steps/utils/__tests__/remoteDeviceRunSession.test.ts b/packages/build-tools/src/steps/utils/__tests__/remoteDeviceRunSession.test.ts index 0e94c21d54..cec59e9ec7 100644 --- a/packages/build-tools/src/steps/utils/__tests__/remoteDeviceRunSession.test.ts +++ b/packages/build-tools/src/steps/utils/__tests__/remoteDeviceRunSession.test.ts @@ -916,7 +916,7 @@ describe(waitForWebPreviewReadyAsync, () => { describe(startNgrokTunnelAsync, () => { it.each(['rejected', 'stalled'] as const)( - 'retains a %s close failure on repeated stops', + 'bounds %s listener retirement and reuses the stop promise', async mode => { jest.useFakeTimers(); try { @@ -938,14 +938,12 @@ describe(startNgrokTunnelAsync, () => { logger: createLoggerMock(), }); const stopping = tunnel.stopAsync(); - const assertion = expect(stopping).rejects.toThrow( - mode === 'rejected' ? 'close failed' : 'Ngrok tunnel stop timed out after 4000ms.' - ); + const assertion = expect(stopping).rejects.toThrow(/stop timed out after 4000ms/); expect(tunnel.stopAsync()).toBe(stopping); await jest.advanceTimersByTimeAsync(4_000); await assertion; await expect(tunnel.stopAsync()).rejects.toThrow(); - expect(close).toHaveBeenCalledTimes(1); + expect(close).toHaveBeenCalled(); } finally { jest.useRealTimers(); } diff --git a/packages/build-tools/src/steps/utils/deviceSessionHost.ts b/packages/build-tools/src/steps/utils/deviceSessionHost.ts index adc087fe8b..6326e7e76d 100644 --- a/packages/build-tools/src/steps/utils/deviceSessionHost.ts +++ b/packages/build-tools/src/steps/utils/deviceSessionHost.ts @@ -607,6 +607,7 @@ async function startDeviceSessionHostInternalAsync( subdomainId, baseDomain, authtoken, + healthCheck: { path: isAndroid ? '/readyz' : '/healthz' }, logger, }); if (finishTask) { diff --git a/packages/build-tools/src/steps/utils/ngrokTunnel.ts b/packages/build-tools/src/steps/utils/ngrokTunnel.ts new file mode 100644 index 0000000000..c93c44028c --- /dev/null +++ b/packages/build-tools/src/steps/utils/ngrokTunnel.ts @@ -0,0 +1,315 @@ +import { SystemError } from '@expo/eas-build-job'; +import { type bunyan } from '@expo/logger'; +import * as ngrok from '@ngrok/ngrok'; +import { randomBytes } from 'node:crypto'; +import { clearTimeout, setTimeout } from 'node:timers'; + +import { turtleFetch } from '../../utils/turtleFetch'; +import { withDeviceRunSessionTimeoutAsync } from './deviceRunSessionTimeout'; + +const PROBE_INTERVAL_MS = 15_000; +const FAILURE_THRESHOLD = 3; +const STOP_TIMEOUT_MS = 4_000; + +export type NgrokTunnelHandle = { + url: string; + subdomainId: string; + stopAsync: () => Promise; +}; + +export async function startNgrokTunnelAsync({ + port, + subdomainPrefix, + subdomainId = randomBytes(16).toString('hex'), + baseDomain, + authtoken, + rewriteHostHeader, + healthCheck, + logger, +}: { + port: number; + subdomainPrefix: string; + subdomainId?: string; + baseDomain: string; + authtoken: string; + rewriteHostHeader?: boolean; + healthCheck?: { path: string; statuses?: number[] }; + logger: bunyan; +}): Promise { + const domain = `${subdomainPrefix}-${subdomainId}.${baseDomain}`; + const controller = new AbortController(); + const { signal } = controller; + const config: ngrok.Config = { + addr: port, + authtoken, + domain, + ...(rewriteHostHeader ? { request_header_add: [`Host:localhost:${port}`] } : {}), + }; + + const retiredListeners = new Set(); + const closingListeners = new Map>(); + + async function closeAsync( + listener: ngrok.Listener, + closeSignal: AbortSignal, + logFailure = true + ): Promise { + retiredListeners.add(listener); + let closing = closingListeners.get(listener); + if (!closing) { + closing = Promise.resolve().then(async () => await listener.close()); + closingListeners.set(listener, closing); + void closing.then( + () => { + closingListeners.delete(listener); + retiredListeners.delete(listener); + }, + () => closingListeners.delete(listener) + ); + } + try { + await withDeviceRunSessionTimeoutAsync( + { name: 'Ngrok tunnel close', timeoutMs: 5_000, signal: closeSignal }, + async () => await closing + ); + } catch (err) { + if (logFailure && !closeSignal.aborted) { + logger.warn({ err }, `Could not stop ngrok tunnel ${domain}.`); + } + } + } + + async function retireListenersAsync(retirementSignal: AbortSignal): Promise { + let logFailure = true; + while (retiredListeners.size > 0) { + retirementSignal.throwIfAborted(); + await Promise.all( + [...retiredListeners].map(retired => closeAsync(retired, retirementSignal, logFailure)) + ); + logFailure = false; + if (retiredListeners.size > 0) { + await waitAsync(250, retirementSignal); + } + } + } + + async function retireLateListenerAsync(listener: ngrok.Listener): Promise { + retiredListeners.add(listener); + try { + await withDeviceRunSessionTimeoutAsync( + { name: 'Late ngrok tunnel retirement', timeoutMs: STOP_TIMEOUT_MS }, + async retirementSignal => { + while (retiredListeners.has(listener)) { + retirementSignal.throwIfAborted(); + await closeAsync(listener, retirementSignal, false); + if (retiredListeners.has(listener)) { + await waitAsync(250, retirementSignal); + } + } + } + ); + } catch (err) { + logger.warn({ err }, `Could not retire late ngrok tunnel ${domain}.`); + } + } + + let pendingOpen: Promise | undefined; + async function openAsync(forceNewSession = false): Promise { + if (pendingOpen) { + throw new SystemError(`A previous ngrok tunnel open for ${domain} is still pending.`); + } + let acquired: ngrok.Listener | undefined; + let retirement: Promise | undefined; + const retireAcquired = () => { + if (acquired) { + retirement ??= retireLateListenerAsync(acquired); + } + }; + try { + return await withDeviceRunSessionTimeoutAsync( + { name: 'Ngrok tunnel open', timeoutMs: 15_000, signal }, + async openSignal => { + openSignal.throwIfAborted(); + const opening = ngrok + .forward({ + ...config, + ...(forceNewSession ? { force_new_session: true } : {}), + }) + .finally(() => { + pendingOpen = undefined; + }); + pendingOpen = opening; + void opening + .then(listener => { + acquired = listener; + if (openSignal.aborted) { + retireAcquired(); + } + }) + .catch(() => {}); + const listener = await opening; + openSignal.throwIfAborted(); + return listener; + } + ); + } catch (err) { + retireAcquired(); + throw err; + } + } + + logger.info(`Starting ngrok tunnel ${domain} -> http://localhost:${port}.`); + let currentListener: ngrok.Listener | undefined = await openAsync(); + const publicUrl = currentListener.url(); + if (!publicUrl) { + await closeAsync(currentListener, signal); + throw new SystemError(`ngrok tunnel for ${domain} did not return a public URL.`); + } + + const url = publicUrl; + + async function probeAsync( + baseUrl: string, + { path, statuses = [200] }: NonNullable + ): Promise { + try { + return await withDeviceRunSessionTimeoutAsync( + { name: 'Ngrok tunnel health probe', timeoutMs: 5_000, signal }, + async probeSignal => { + const requestController = new AbortController(); + try { + const response = await turtleFetch(new URL(path, baseUrl).href, 'GET', { + retries: 0, + redirect: 'manual', + shouldThrowOnNotOk: false, + headers: { 'ngrok-skip-browser-warning': 'true', 'Cache-Control': 'no-cache' }, + signal: AbortSignal.any([probeSignal, requestController.signal]), + }); + return !response.headers.has('ngrok-error-code') && statuses.includes(response.status); + } finally { + requestController.abort(); + } + } + ); + } catch { + return false; + } + } + + async function superviseAsync(check: NonNullable): Promise { + let failures = 0; + let attempts = 0; + let localUnhealthy = false; + while ( + await waitAsync( + currentListener || localUnhealthy + ? PROBE_INTERVAL_MS + : Math.min(1_000 * 2 ** Math.min(attempts, 5), 30_000), + signal + ) + ) { + await Promise.all([...retiredListeners].map(retired => closeAsync(retired, signal, false))); + if (signal.aborted) { + return; + } + if (currentListener && (await probeAsync(url, check))) { + if (failures > 0 || attempts > 0 || localUnhealthy) { + logger.info(`Ngrok tunnel ${domain} is healthy again.`); + } + failures = 0; + attempts = 0; + localUnhealthy = false; + continue; + } + if (signal.aborted) { + return; + } + if (!(await probeAsync(`http://127.0.0.1:${port}`, check))) { + if (!localUnhealthy && !signal.aborted) { + logger.warn(`Local service on port ${port} is unhealthy; waiting for it to recover.`); + } + localUnhealthy = true; + failures = 0; + continue; + } + localUnhealthy = false; + if (currentListener) { + failures++; + if (failures < FAILURE_THRESHOLD) { + continue; + } + logger.warn(`Ngrok tunnel ${domain} failed ${failures} health probes; reopening it.`); + await closeAsync(currentListener, signal); + currentListener = undefined; + } + if (signal.aborted) { + return; + } + attempts++; + try { + const reopened = await openAsync(/* forceNewSession */ true); + if (reopened.url() !== url) { + await closeAsync(reopened, signal); + throw new SystemError( + `Reopened ngrok tunnel for ${domain} returned a different public URL.` + ); + } + currentListener = reopened; + failures = 0; + } catch (err) { + if (!signal.aborted && (attempts === 1 || attempts % 5 === 0)) { + logger.warn( + { err, attempt: attempts }, + `Could not reopen ngrok tunnel ${domain}; will retry.` + ); + } + } + } + } + + const supervision = healthCheck + ? superviseAsync(healthCheck).catch(err => { + if (!signal.aborted) { + logger.warn({ err }, `Ngrok tunnel supervision stopped for ${domain}.`); + } + }) + : Promise.resolve(); + let stopTask: Promise | undefined; + return { + url, + subdomainId, + stopAsync: () => + (stopTask ??= (async () => { + controller.abort(); + await withDeviceRunSessionTimeoutAsync( + { name: `Ngrok tunnel ${domain} stop`, timeoutMs: STOP_TIMEOUT_MS }, + async stopSignal => { + await supervision; + await pendingOpen?.catch(() => {}); + if (currentListener) { + retiredListeners.add(currentListener); + currentListener = undefined; + } + await retireListenersAsync(stopSignal); + } + ); + })()), + }; +} + +function waitAsync(ms: number, signal: AbortSignal): Promise { + if (signal.aborted) { + return Promise.resolve(false); + } + return new Promise(resolve => { + const onAbort = () => { + clearTimeout(timer); + resolve(false); + }; + const timer = setTimeout(() => { + signal.removeEventListener('abort', onAbort); + resolve(true); + }, ms); + signal.addEventListener('abort', onAbort, { once: true }); + }); +} diff --git a/packages/build-tools/src/steps/utils/remoteDeviceRunSession.ts b/packages/build-tools/src/steps/utils/remoteDeviceRunSession.ts index c75babbce3..f39f9336c3 100644 --- a/packages/build-tools/src/steps/utils/remoteDeviceRunSession.ts +++ b/packages/build-tools/src/steps/utils/remoteDeviceRunSession.ts @@ -9,9 +9,7 @@ import { spawnAsync, } from '@expo/steps'; import spawn from '@expo/turtle-spawn'; -import * as ngrok from '@ngrok/ngrok'; import { z } from 'zod'; -import { randomBytes } from 'node:crypto'; import fs from 'node:fs'; import { createServer } from 'node:net'; import path from 'node:path'; @@ -946,63 +944,7 @@ export async function findAvailablePortAsync(): Promise { return address.port; } -export type NgrokTunnelHandle = { - url: string; - subdomainId: string; - stopAsync: () => Promise; -}; - -export async function startNgrokTunnelAsync({ - port, - subdomainPrefix, - subdomainId: subdomainIdArg, - baseDomain, - authtoken, - rewriteHostHeader, - logger, -}: { - port: number; - subdomainPrefix: string; - subdomainId?: string; - baseDomain: string; - authtoken: string; - rewriteHostHeader?: boolean; - logger: bunyan; -}): Promise { - const subdomainId = subdomainIdArg ?? randomBytes(16).toString('hex'); - const domain = `${subdomainPrefix}-${subdomainId}.${baseDomain}`; - logger.info(`Starting ngrok tunnel ${domain} -> http://localhost:${port}.`); - // Run the ngrok agent in-process via the SDK; it keeps the session alive until - // the process exits, and the step blocks forever to hold it open. - const listener = await ngrok.forward({ - addr: port, - authtoken, - domain, - ...(rewriteHostHeader ? { request_header_add: [`Host:localhost:${port}`] } : {}), - }); - const url = listener.url(); - if (!url) { - await listener.close(); - throw new SystemError(`ngrok tunnel for ${domain} did not return a public URL.`); - } - let stopTask: Promise | undefined; - return { - url, - subdomainId, - stopAsync: () => - (stopTask ??= (async () => { - try { - await withDeviceRunSessionTimeoutAsync( - { name: 'Ngrok tunnel stop', timeoutMs: 4_000 }, - async () => await listener.close() - ); - } catch (error) { - logger.warn({ err: error }, `Could not stop ngrok tunnel ${domain}.`); - throw error; - } - })()), - }; -} +export { type NgrokTunnelHandle, startNgrokTunnelAsync } from './ngrokTunnel'; export async function waitForFileAsync({ filePath,