From ae6313206caf4991db9ca0bb19a5a2c3c470129b Mon Sep 17 00:00:00 2001 From: Gabe Debes Date: Sat, 10 Oct 2026 01:25:35 -0700 Subject: [PATCH 1/5] [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 70c169af6e..c9d4ae0599 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 e848dc5631..82a17eb6b9 100644 --- a/packages/build-tools/src/steps/functions/startAppiumRemoteSession.ts +++ b/packages/build-tools/src/steps/functions/startAppiumRemoteSession.ts @@ -274,6 +274,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 b24986e78b..d39cc92bd4 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 6fc8170446..384eb33370 100644 --- a/packages/build-tools/src/steps/utils/__tests__/remoteDeviceRunSession.test.ts +++ b/packages/build-tools/src/steps/utils/__tests__/remoteDeviceRunSession.test.ts @@ -932,7 +932,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 { @@ -954,14 +954,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 f4056ec12c..e14953c471 100644 --- a/packages/build-tools/src/steps/utils/deviceSessionHost.ts +++ b/packages/build-tools/src/steps/utils/deviceSessionHost.ts @@ -623,6 +623,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 ce779b21d1..06d3a2b4c7 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'; @@ -967,63 +965,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, From 3c607129fe9ad11b0b7e7e1dc72d26506496f50a Mon Sep 17 00:00:00 2001 From: Gabe Debes Date: Thu, 29 Oct 2026 15:43:58 -0700 Subject: [PATCH 2/5] [build-tools] Retire tunnel listeners while reopening --- .../steps/utils/__tests__/ngrokTunnel.test.ts | 25 +++++++++++++++++++ .../src/steps/utils/ngrokTunnel.ts | 3 +-- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts b/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts index 83eb2d4caa..fa12ba9224 100644 --- a/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts +++ b/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts @@ -256,6 +256,31 @@ it('fails stop if an in-flight reopen still has not settled by the stop deadline expect(replacement.close).toHaveBeenCalledTimes(1); }); +it('retries an old listener while a reopen outlasts shutdown', async () => { + failPublicProbes(); + initial.close.mockRejectedValueOnce(new Error('agent disconnected')); + 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); + expect(initial.close).toHaveBeenCalledTimes(1); + 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; + const oldCloseCalls = initial.close.mock.calls.length; + resolveOpening(replacement as never); + await jest.advanceTimersByTimeAsync(0); + expect(oldCloseCalls).toBe(2); + expect(replacement.close).toHaveBeenCalledTimes(1); + expect(ngrok.forward).toHaveBeenCalledTimes(2); +}); + it('fails stop if a pending reopen arrives but cannot be retired', async () => { failPublicProbes(); let resolveOpening!: (value: ngrok.Listener) => void; diff --git a/packages/build-tools/src/steps/utils/ngrokTunnel.ts b/packages/build-tools/src/steps/utils/ngrokTunnel.ts index c93c44028c..1778692f17 100644 --- a/packages/build-tools/src/steps/utils/ngrokTunnel.ts +++ b/packages/build-tools/src/steps/utils/ngrokTunnel.ts @@ -285,12 +285,11 @@ export async function startNgrokTunnelAsync({ { 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); + await Promise.all([pendingOpen?.catch(() => {}), retireListenersAsync(stopSignal)]); } ); })()), From 209016a8630cb525acc2658c7829a34beedf26c2 Mon Sep 17 00:00:00 2001 From: Gabe Debes Date: Thu, 29 Oct 2026 15:55:48 -0700 Subject: [PATCH 3/5] [build-tools] Wait for replacement tunnel retirement --- .../steps/utils/__tests__/ngrokTunnel.test.ts | 27 +++++++++++++++++++ .../src/steps/utils/ngrokTunnel.ts | 1 + 2 files changed, 28 insertions(+) diff --git a/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts b/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts index fa12ba9224..d57352b54a 100644 --- a/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts +++ b/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts @@ -195,6 +195,33 @@ it('closes a listener that finishes opening after shutdown, with idempotent clea expect(ngrok.forward).toHaveBeenCalledTimes(2); }); +it('waits for a replacement acquired during shutdown to finish closing', async () => { + failPublicProbes(); + let resolveOpening!: (value: ngrok.Listener) => void; + let resolveClose!: () => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockResolvedValueOnce(initial as never) + .mockReturnValueOnce(new Promise(resolve => (resolveOpening = resolve))); + replacement.close.mockReturnValue(new Promise(resolve => (resolveClose = resolve))); + const tunnel = await startNgrokTunnelAsync(options); + await jest.advanceTimersByTimeAsync(45_000); + let stopped = false; + const stopping = tunnel.stopAsync().then(() => { + stopped = true; + }); + await jest.advanceTimersByTimeAsync(0); + resolveOpening(replacement as never); + await jest.advanceTimersByTimeAsync(0); + const stoppedBeforeClose = stopped; + resolveClose(); + await jest.advanceTimersByTimeAsync(250); + await stopping; + expect(stoppedBeforeClose).toBe(false); + expect(replacement.close).toHaveBeenCalledTimes(1); +}); + it.each([0, 4, 5, 10])('retires a reopen cancelled after %s promise turns', async turns => { failPublicProbes(); let resolveOpening!: (value: ngrok.Listener) => void; diff --git a/packages/build-tools/src/steps/utils/ngrokTunnel.ts b/packages/build-tools/src/steps/utils/ngrokTunnel.ts index 1778692f17..302447ff6c 100644 --- a/packages/build-tools/src/steps/utils/ngrokTunnel.ts +++ b/packages/build-tools/src/steps/utils/ngrokTunnel.ts @@ -290,6 +290,7 @@ export async function startNgrokTunnelAsync({ currentListener = undefined; } await Promise.all([pendingOpen?.catch(() => {}), retireListenersAsync(stopSignal)]); + await retireListenersAsync(stopSignal); } ); })()), From 9b549c69d5aabb73dffbea60cf56770ce15cc02b Mon Sep 17 00:00:00 2001 From: Gabe Debes Date: Thu, 29 Oct 2026 16:53:13 -0700 Subject: [PATCH 4/5] [build-tools] Block tunnel forwarding when shutdown starts --- .changeset/serve-sim-pr-4526.md | 2 + .../steps/utils/__tests__/ngrokTunnel.test.ts | 190 +++++++++++++++++- .../__tests__/remoteDeviceRunSession.test.ts | 12 +- .../src/steps/utils/ngrokTunnel.ts | 73 ++++++- 4 files changed, 261 insertions(+), 16 deletions(-) diff --git a/.changeset/serve-sim-pr-4526.md b/.changeset/serve-sim-pr-4526.md index 9cc7ee01be..1c35f0165d 100644 --- a/.changeset/serve-sim-pr-4526.md +++ b/.changeset/serve-sim-pr-4526.md @@ -3,3 +3,5 @@ --- Recover unhealthy device-session tunnels without changing their URLs. + +Block further tunnel traffic as soon as local shutdown starts, even if ngrok cleanup times out. diff --git a/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts b/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts index d57352b54a..6612cb3c33 100644 --- a/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts +++ b/packages/build-tools/src/steps/utils/__tests__/ngrokTunnel.test.ts @@ -1,7 +1,9 @@ import { type bunyan } from '@expo/logger'; import * as ngrok from '@ngrok/ngrok'; +import { once } from 'node:events'; import { createServer } from 'node:http'; -import { type AddressInfo } from 'node:net'; +import WebSocket, { WebSocketServer } from 'ws'; +import { type AddressInfo, createConnection, createServer as createTcpServer } from 'node:net'; import { turtleFetch } from '../../../utils/turtleFetch'; import { startNgrokTunnelAsync } from '../ngrokTunnel'; @@ -35,7 +37,7 @@ let initial: ReturnType; let replacement: ReturnType; beforeEach(() => { - jest.useFakeTimers(); + jest.useFakeTimers({ doNotFake: ['nextTick', 'setImmediate'] }); jest.clearAllMocks(); initial = listener(); replacement = listener(); @@ -105,7 +107,7 @@ it('reopens after three failed public probes with a healthy local service, prese expect(initial.close).toHaveBeenCalledTimes(1); expect(ngrok.forward).toHaveBeenCalledTimes(2); expect(jest.mocked(ngrok.forward).mock.calls[1][0]).toEqual({ - addr: 4321, + addr: expect.any(Number), authtoken: 'token', domain: 'web-preview-fixed.example.test', request_header_add: ['Host:localhost:4321'], @@ -687,3 +689,185 @@ it('recovers using actual HTTP health probes with a local ngrok adapter', async } expect(next.close).toHaveBeenCalledTimes(1); }); + +it('blocks HTTP and existing WebSockets on Stop even when ngrok cannot close', async () => { + jest.useRealTimers(); + const actualFetch = jest.requireActual('node-fetch').default; + let requests = 0; + let messages = 0; + const backend = createServer((_request, response) => { + requests++; + response.end('owned backend'); + }); + const websockets = new WebSocketServer({ server: backend }); + websockets.on('connection', socket => { + socket.on('message', data => { + messages++; + socket.send(data); + }); + }); + backend.listen(0, '127.0.0.1'); + await once(backend, 'listening'); + const backendPort = (backend.address() as AddressInfo).port; + const tunnel = await startNgrokTunnelAsync({ + ...options, + port: backendPort, + healthCheck: undefined, + }); + const targetPort = (jest.mocked(ngrok.forward).mock.calls[0][0] as ngrok.Config).addr; + const targetUrl = `http://127.0.0.1:${targetPort}`; + const socket = new WebSocket(`ws://127.0.0.1:${targetPort}`); + const opened = once(socket, 'open'); + try { + expect(await (await actualFetch(targetUrl)).text()).toBe('owned backend'); + await opened; + const echoed = once(socket, 'message'); + socket.send('before-stop'); + expect((await echoed)[0].toString()).toBe('before-stop'); + const closed = once(socket, 'close'); + initial.close.mockRejectedValue(new Error('agent disconnected')); + const stopping = tunnel.stopAsync(); + const rejected = expect(stopping).rejects.toThrow('stop timed out after 4000ms'); + await expect(actualFetch(targetUrl, { timeout: 1_000 })).rejects.toThrow(); + await closed; + expect(requests).toBe(1); + expect(messages).toBe(1); + expect(await (await actualFetch(`http://127.0.0.1:${backendPort}`)).text()).toBe( + 'owned backend' + ); + await rejected; + await expect(actualFetch(targetUrl, { timeout: 1_000 })).rejects.toThrow(); + expect(requests).toBe(2); + expect(messages).toBe(1); + const competing = createTcpServer(); + const conflict = once(competing, 'error'); + competing.listen(targetPort as number, '127.0.0.1'); + expect((await conflict)[0]).toMatchObject({ code: 'EADDRINUSE' }); + } finally { + socket.terminate(); + for (const client of websockets.clients) { + client.terminate(); + } + websockets.close(); + backend.closeAllConnections(); + await new Promise(resolve => backend.close(() => resolve())); + await tunnel.stopAsync().catch(() => {}); + } +}); + +it('releases its local target when initial tunnel creation fails', async () => { + jest.useRealTimers(); + jest.mocked(ngrok.forward).mockReset().mockRejectedValue(new Error('forward failed')); + await expect(startNgrokTunnelAsync(options)).rejects.toThrow('forward failed'); + const targetPort = (jest.mocked(ngrok.forward).mock.calls[0][0] as ngrok.Config).addr; + const server = createServer(); + server.listen(targetPort as number, '127.0.0.1'); + await once(server, 'listening'); + await new Promise(resolve => server.close(() => resolve())); +}); + +it('preserves a full response for a half-closed slow reader', async () => { + jest.useRealTimers(); + const payload = Buffer.alloc(4 * 1024 * 1024, 37); + const backend = createTcpServer({ allowHalfOpen: true }, socket => { + socket.resume(); + socket.on('end', () => socket.end(payload)); + }); + backend.listen(0, '127.0.0.1'); + await once(backend, 'listening'); + const tunnel = await startNgrokTunnelAsync({ + ...options, + port: (backend.address() as AddressInfo).port, + healthCheck: undefined, + }); + const targetPort = (jest.mocked(ngrok.forward).mock.calls[0][0] as ngrok.Config).addr as number; + const socket = createConnection({ host: '127.0.0.1', port: targetPort, allowHalfOpen: true }); + try { + const chunks: Buffer[] = []; + socket.on('data', chunk => chunks.push(chunk)); + const ended = once(socket, 'end'); + socket.pause(); + socket.end('request'); + await new Promise(resolve => setTimeout(resolve, 100)); + socket.resume(); + await ended; + expect(Buffer.concat(chunks).equals(payload)).toBe(true); + } finally { + socket.destroy(); + await tunnel.stopAsync(); + await new Promise(resolve => backend.close(() => resolve())); + } +}); + +it.each(['resolve', 'reject'])( + 'releases the target after timed-out startup later %s', + async outcome => { + let resolveOpening!: (value: ngrok.Listener) => void; + let rejectOpening!: (error: Error) => void; + jest + .mocked(ngrok.forward) + .mockReset() + .mockReturnValue( + new Promise((resolve, reject) => { + resolveOpening = resolve; + rejectOpening = reject; + }) + ); + const opening = startNgrokTunnelAsync(options); + const rejected = expect(opening).rejects.toThrow('open timed out'); + await new Promise(resolve => setImmediate(resolve)); + await jest.advanceTimersByTimeAsync(15_000); + await rejected; + const targetPort = (jest.mocked(ngrok.forward).mock.calls[0][0] as ngrok.Config).addr as number; + if (outcome === 'resolve') { + resolveOpening(replacement as never); + } else { + rejectOpening(new Error('forward failed')); + } + await jest.advanceTimersByTimeAsync(0); + const server = createTcpServer(); + server.listen(targetPort, '127.0.0.1'); + await once(server, 'listening'); + await new Promise(resolve => server.close(() => resolve())); + } +); + +it('drains a full upload when the backend ends its response early', async () => { + jest.useRealTimers(); + const payload = Buffer.alloc(4 * 1024 * 1024, 51); + let received!: Buffer; + let receivedRequest!: () => void; + const completed = new Promise(resolve => { + receivedRequest = resolve; + }); + const backend = createTcpServer({ allowHalfOpen: true }, socket => { + const chunks: Buffer[] = []; + socket.on('data', chunk => chunks.push(chunk)); + socket.on('end', () => { + received = Buffer.concat(chunks); + receivedRequest(); + }); + socket.pause(); + socket.end('early response'); + setTimeout(() => socket.resume(), 100); + }); + backend.listen(0, '127.0.0.1'); + await once(backend, 'listening'); + const tunnel = await startNgrokTunnelAsync({ + ...options, + port: (backend.address() as AddressInfo).port, + healthCheck: undefined, + }); + const targetPort = (jest.mocked(ngrok.forward).mock.calls[0][0] as ngrok.Config).addr as number; + const socket = createConnection({ host: '127.0.0.1', port: targetPort, allowHalfOpen: true }); + try { + socket.resume(); + socket.end(payload); + await completed; + expect(received.equals(payload)).toBe(true); + } finally { + socket.destroy(); + await tunnel.stopAsync(); + await new Promise(resolve => backend.close(() => resolve())); + } +}); 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 384eb33370..1539d79385 100644 --- a/packages/build-tools/src/steps/utils/__tests__/remoteDeviceRunSession.test.ts +++ b/packages/build-tools/src/steps/utils/__tests__/remoteDeviceRunSession.test.ts @@ -983,7 +983,7 @@ describe(startNgrokTunnelAsync, () => { expect(ngrok.forward).toHaveBeenCalledWith( expect.objectContaining({ - addr: 4321, + addr: expect.any(Number), authtoken: 'token', domain: expect.stringMatching(/^web-preview-[a-f0-9]{32}\.eas-simulator\.ngrok\.dev$/), }) @@ -1300,7 +1300,10 @@ describe(startDeviceSessionHostAsync, () => { recordingDirectory, }), ]); - expect(ngrok.forward).toHaveBeenCalledWith(expect.objectContaining({ addr: port })); + expect(ngrok.forward).toHaveBeenCalledWith( + expect.objectContaining({ addr: expect.any(Number) }) + ); + expect(ngrok.forward).not.toHaveBeenCalledWith(expect.objectContaining({ addr: port })); expect(preview.apiUrl).toBe('https://android-preview.example.test'); await host.finishAsync(); @@ -1436,7 +1439,10 @@ describe(startDeviceSessionHostAsync, () => { packageVersion, }), ]); - expect(ngrok.forward).toHaveBeenCalledWith(expect.objectContaining({ addr: port })); + expect(ngrok.forward).toHaveBeenCalledWith( + expect.objectContaining({ addr: expect.any(Number) }) + ); + expect(ngrok.forward).not.toHaveBeenCalledWith(expect.objectContaining({ addr: port })); expect(preview.apiUrl).toBe('https://ios-preview.example.test'); await host.finishAsync(); diff --git a/packages/build-tools/src/steps/utils/ngrokTunnel.ts b/packages/build-tools/src/steps/utils/ngrokTunnel.ts index 302447ff6c..5c581bcd6b 100644 --- a/packages/build-tools/src/steps/utils/ngrokTunnel.ts +++ b/packages/build-tools/src/steps/utils/ngrokTunnel.ts @@ -2,6 +2,8 @@ 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 { once } from 'node:events'; +import { type AddressInfo, type Socket, createConnection, createServer } from 'node:net'; import { clearTimeout, setTimeout } from 'node:timers'; import { turtleFetch } from '../../utils/turtleFetch'; @@ -39,15 +41,52 @@ export async function startNgrokTunnelAsync({ const domain = `${subdomainPrefix}-${subdomainId}.${baseDomain}`; const controller = new AbortController(); const { signal } = controller; + const sockets = new Set(); + const target = createServer({ allowHalfOpen: true }, client => { + if (signal.aborted) { + client.destroy(); + return; + } + const upstream = createConnection({ host: '127.0.0.1', port, allowHalfOpen: true }); + for (const socket of [client, upstream]) { + sockets.add(socket); + socket.on('error', () => { + client.destroy(); + upstream.destroy(); + }); + socket.on('close', () => sockets.delete(socket)); + } + client.pipe(upstream).pipe(client); + }); + target.listen(0, '127.0.0.1'); + await once(target, 'listening'); + target.unref(); + signal.addEventListener( + 'abort', + () => { + for (const socket of sockets) { + socket.destroy(); + } + }, + { once: true } + ); const config: ngrok.Config = { - addr: port, + addr: (target.address() as AddressInfo).port, authtoken, domain, ...(rewriteHostHeader ? { request_header_add: [`Host:localhost:${port}`] } : {}), }; + const ownedListeners = new Set(); const retiredListeners = new Set(); const closingListeners = new Map>(); + let pendingOpen: Promise | undefined; + let currentListener: ngrok.Listener | undefined; + function releaseTarget(): void { + if (signal.aborted && !pendingOpen && ownedListeners.size === 0 && target.listening) { + target.close(); + } + } async function closeAsync( listener: ngrok.Listener, @@ -63,6 +102,8 @@ export async function startNgrokTunnelAsync({ () => { closingListeners.delete(listener); retiredListeners.delete(listener); + ownedListeners.delete(listener); + releaseTarget(); }, () => closingListeners.delete(listener) ); @@ -113,7 +154,6 @@ export async function startNgrokTunnelAsync({ } } - 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.`); @@ -135,18 +175,22 @@ export async function startNgrokTunnelAsync({ ...config, ...(forceNewSession ? { force_new_session: true } : {}), }) + .then(listener => { + acquired = listener; + ownedListeners.add(listener); + return listener; + }) .finally(() => { pendingOpen = undefined; }); pendingOpen = opening; void opening - .then(listener => { - acquired = listener; + .then(() => { if (openSignal.aborted) { retireAcquired(); } }) - .catch(() => {}); + .catch(() => releaseTarget()); const listener = await opening; openSignal.throwIfAborted(); return listener; @@ -159,11 +203,19 @@ export async function startNgrokTunnelAsync({ } 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.`); + let publicUrl: string | null; + try { + currentListener = await openAsync(); + publicUrl = currentListener.url(); + if (!publicUrl) { + await closeAsync(currentListener, signal); + currentListener = undefined; + throw new SystemError(`ngrok tunnel for ${domain} did not return a public URL.`); + } + } catch (error) { + controller.abort(); + releaseTarget(); + throw error; } const url = publicUrl; @@ -293,6 +345,7 @@ export async function startNgrokTunnelAsync({ await retireListenersAsync(stopSignal); } ); + releaseTarget(); })()), }; } From 6073478ed34826006d09737a71f6eadf9cebd2a0 Mon Sep 17 00:00:00 2001 From: Gabe Debes Date: Fri, 30 Oct 2026 15:34:21 -0700 Subject: [PATCH 5/5] [build-tools] Shorten tunnel recovery changeset --- .changeset/serve-sim-pr-4526.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/.changeset/serve-sim-pr-4526.md b/.changeset/serve-sim-pr-4526.md index 1c35f0165d..af4db16cba 100644 --- a/.changeset/serve-sim-pr-4526.md +++ b/.changeset/serve-sim-pr-4526.md @@ -2,6 +2,4 @@ '@expo/build-tools': patch --- -Recover unhealthy device-session tunnels without changing their URLs. - -Block further tunnel traffic as soon as local shutdown starts, even if ngrok cleanup times out. +Recover unhealthy device-session tunnels without changing their URLs and block further traffic when shutdown starts, even if ngrok cleanup times out.