fix: race requests to all URLs in parallel (#53)

gmaclennan · web-flow · commit 101f6a363334 · 2026-04-15T15:42:31.000Z
* fix: race requests to all URLs in parallel

* fix two requests resolving in same tick

* fix race issues with OPTIONS probe

* remove unused fn

* fix remaining windows test failure

* attempted fix 1

* add code comment about aborting downloads
diff --git a/src/lib/download-request.ts b/src/lib/download-request.ts
@@ -5,7 +5,7 @@ import type { DownloadCreateParams } from '../routes/downloads.js'
 import { type DownloadStateUpdate } from '../types.js'
 import { StatusError } from './errors.js'
 import { errors, jsonError } from './errors.js'
-import { secretStreamFetch } from './secret-stream-fetch.js'
+import { anyFetch } from './secret-stream-fetch.js'
 import { StateUpdateEvent } from './state-update-event.js'
 import { throttle } from './throttle.js'
 import { addTrailingSlash, generateId, getErrorCode, noop } from './utils.js'
@@ -80,7 +80,8 @@ export class DownloadRequest extends TypedEventTarget<
 					// namely whether it was canceled, or if a different error occurred on
 					// the server side.
 					try {
-						const response = await secretStreamFetch(mapShareUrls, {
+						// GET /:shareId is idempotent, so racing URLs directly is safe.
+						const response = await anyFetch(mapShareUrls, {
 							dispatcher: this.#dispatcher,
 							signal: AbortSignal.timeout(2000),
 						})
@@ -107,14 +108,42 @@ export class DownloadRequest extends TypedEventTarget<
 		remotePublicKey: Uint8Array
 		keyPair: { publicKey: Uint8Array; secretKey: Uint8Array }
 	}) {
+		// anyFetch probes the URLs in parallel and only sends the real /download
+		// request to the winner. Necessary because the sender only supports one
+		// active download per share, so racing /download against a server
+		// reachable on several URLs would be unsafe.
 		const downloadUrls = mapShareUrls.map(
 			(baseUrl) => new URL('download', addTrailingSlash(baseUrl)),
-		) as unknown as [URL, ...URL[]] // grrrr TS
-		const response = await secretStreamFetch(downloadUrls, {
-			dispatcher: this.#dispatcher,
-		})
+		) as unknown as readonly [URL, ...URL[]]
+		let response: Response
+		try {
+			// We deliberately do NOT pass this.#abortController.signal here. If we
+			// did, a fast abort could fire while anyFetch is still in the probe
+			// phase — before any real /download request has reached the sender —
+			// leaving the sender's MapShare stuck in 'pending' because it never
+			// saw a connection to cancel. Instead we let anyFetch run to
+			// completion, then check the abort flag below (see the
+			// `signal.aborted` check after this try/catch). That way the sender
+			// always gets a real request, and cancelling the response body
+			// closes the connection so the sender can transition its state.
+			response = await anyFetch(downloadUrls, {
+				dispatcher: this.#dispatcher,
+			})
+		} catch (error) {
+			if (error instanceof DOMException && error.name === 'AbortError') {
+				throw error // Handle abort in caller
+			}
+			throw new errors.DOWNLOAD_ERROR({
+				message: 'Could not connect to map share sender',
+				urls: mapShareUrls,
+				cause: error,
+			})
+		}
 		if (!response.body) {
-			throw new errors.DOWNLOAD_ERROR('Could not connect to map share sender')
+			throw new errors.DOWNLOAD_ERROR({
+				message: 'Could not connect to map share sender',
+				urls: mapShareUrls,
+			})
 		}
 		if (!response.ok) {
 			throw new StatusError(response.status, await response.json())
diff --git a/src/lib/secret-stream-fetch.ts b/src/lib/secret-stream-fetch.ts
@@ -1,54 +1,150 @@
 import { fetch as secretStreamFetchOrig } from 'secret-stream-http'
 
-import { errors } from './errors.js'
-import { isArrayReadonly } from './utils.js'
+import { noop } from './utils.js'
 
 const CONNECTION_TIMEOUT_MS = 5000 // 5s
 
+type FetchFn = typeof secretStreamFetchOrig
+type FetchInput = Parameters<FetchFn>[0]
+type FetchInit = NonNullable<Parameters<FetchFn>[1]>
+
+type AnyFetchInit = FetchInit & {
+	/**
+	 * Underlying fetch implementation. Defaults to secret-stream-http's fetch.
+	 * Useful for overriding in tests.
+	 */
+	fetch?: FetchFn
+	/**
+	 * Connection timeout in milliseconds. This is a timeout for establishing the
+	 * connection only, not for the entire request. If the connection is not
+	 * established within this time, the request will be aborted and treated as a
+	 * connection failure. Default is 5000ms (5s), which should be plenty of time
+	 * for a local network request
+	 */
+	timeoutMs?: number
+}
+
 /**
- * A wrapper around secret-stream-http's fetch that tries multiple URLs until one works.
- * This is useful when the server has multiple IPs for different network interfaces.
+ * A fetch-compatible wrapper that accepts an array of URLs that are first
+ * probed in parallel with OPTIONS requests. The real request is then sent to
+ * each URL in series — starting with whichever probe responded first — until
+ * one succeeds. The probe step is what makes this safe against stateful
+ * endpoints: only one real request ever hits the server per call.
+ *
+ * Adds a per-request connection timeout, set by `init.timeoutMs` and defaults
+ * to 5000ms (5s).
+ *
+ * The `init.fetch` option overrides the underlying fetch implementation
+ * and defaults to secret-stream-http's fetch.
  */
-export async function secretStreamFetch(
-	urls: string | URL | readonly [string | URL, ...Array<string | URL>],
-	options: Parameters<typeof secretStreamFetchOrig>[1],
-) {
-	if (!isArrayReadonly(urls)) {
-		urls = [urls]
+export async function anyFetch(
+	inputs: readonly [FetchInput, ...FetchInput[]],
+	init?: AnyFetchInit,
+): Promise<Response> {
+	if (inputs.length === 1) {
+		return await timeoutFetch(inputs[0], init)
 	}
-	let response: Response | undefined
-	let error: unknown
 
-	// The server could have multiple IPs for different network interfaces, and
-	// not all of them may be on the same network as us, so try each URL until
-	// one works
-	for (const url of urls) {
-		const controller = new AbortController()
-		const timeout = setTimeout(() => {
-			controller.abort()
-		}, CONNECTION_TIMEOUT_MS)
-		const signal = options?.signal
-			? AbortSignal.any([options.signal, controller.signal])
-			: controller.signal
+	let winningInput: FetchInput
+	try {
+		winningInput = await raceProbes(inputs, init)
+	} catch (err) {
+		// If the caller aborted mid-probe, propagate that as an AbortError
+		// (via signal.reason) rather than the probe's AggregateError — the
+		// caller needs to distinguish "I cancelled this" from "nothing was
+		// reachable".
+		init?.signal?.throwIfAborted()
+		throw err
+	}
+	const orderedInputs = [
+		winningInput,
+		...inputs.filter((i) => i !== winningInput),
+	]
+
+	const errors: unknown[] = []
+	for (const input of orderedInputs) {
 		try {
-			response = (await secretStreamFetchOrig(url, {
-				...options,
-				signal,
-			})) as unknown as Response // Subtle difference bewteen Undici fetch Response and whatwg Response
-			break // Exit loop on successful fetch
+			return await timeoutFetch(input, init)
 		} catch (err) {
-			error = err
-			// Ignore errors and try the next URL
-		} finally {
-			clearTimeout(timeout)
+			// If the caller aborted mid-fetch, propagate that rather than our
+			// internal connection-timeout abort (or whatever lower-level error).
+			init?.signal?.throwIfAborted()
+			errors.push(err)
 		}
 	}
-	if (!response) {
-		throw new errors.DOWNLOAD_ERROR({
-			message: 'Could not connect to map share sender',
-			urls,
-			cause: error,
-		})
+	throw new AggregateError(errors, 'All fetch attempts failed')
+}
+
+/**
+ * Fetch a single URL with a connection timeout (different than calling WHAT-WG
+ * fetch with a timeout signal, which applies to the entire request, not just
+ * connection establishment).
+ */
+async function timeoutFetch(
+	url: FetchInput,
+	{
+		fetch: innerFetch = secretStreamFetchOrig,
+		timeoutMs = CONNECTION_TIMEOUT_MS,
+		...fetchInit
+	}: AnyFetchInit = {},
+): Promise<Response> {
+	const controller = new AbortController()
+	const timeout = setTimeout(() => {
+		controller.abort(new DOMException('Connection timed out', 'TimeoutError'))
+	}, timeoutMs)
+	const signal = fetchInit.signal
+		? AbortSignal.any([fetchInit.signal, controller.signal])
+		: controller.signal
+	try {
+		return (await innerFetch(url, {
+			...fetchInit,
+			signal,
+		})) as unknown as Response // Subtle difference between Undici fetch Response and whatwg Response
+	} finally {
+		clearTimeout(timeout)
+	}
+}
+
+/**
+ * Probe URLs in parallel with OPTIONS requests. Returns the "winning" URL.
+ * Throws AggregateError if no probe responds before the connection timeout or
+ * if all probes fail with a network error.
+ */
+async function raceProbes(
+	inputs: readonly FetchInput[],
+	fetchInit: FetchInit = {},
+): Promise<FetchInput> {
+	const controllers: AbortController[] = []
+	const probePromises: Promise<FetchInput>[] = []
+	let hasWinner = false
+
+	for (const url of inputs) {
+		const controller = new AbortController()
+		const signal = fetchInit.signal
+			? AbortSignal.any([fetchInit.signal, controller.signal])
+			: controller.signal
+		controllers.push(controller)
+		const probePromise = (async () => {
+			const response = await timeoutFetch(url, {
+				...fetchInit,
+				method: 'OPTIONS',
+				body: undefined,
+				signal,
+			})
+			response.body?.cancel().catch(noop) // We don't care about the response body, and we want to free resources as soon as possible
+			if (hasWinner) {
+				throw new DOMException('Aborted by winner', 'AbortError')
+			}
+			hasWinner = true
+			for (const c of controllers) {
+				if (c !== controller) {
+					c.abort(new DOMException('Aborted by winner', 'AbortError'))
+				}
+			}
+			return url
+		})()
+		probePromises.push(probePromise)
 	}
-	return response
+
+	return Promise.any(probePromises)
 }
diff --git a/src/lib/utils.ts b/src/lib/utils.ts
@@ -107,11 +107,3 @@ function isNonEmptyArray<T>(arr: T[]): arr is [T, ...T[]] {
 export function addTrailingSlash(url: string): string {
 	return url.endsWith('/') ? url : url + '/'
 }
-
-// Typescript's Array.isArray definition does not work as a type guard for
-// readonly arrays, so we need to define our own type guard for that
-export function isArrayReadonly<T, U>(
-	value: U | readonly T[],
-): value is readonly T[] {
-	return Array.isArray(value)
-}
diff --git a/src/routes/map-shares.ts b/src/routes/map-shares.ts
@@ -1,17 +1,15 @@
 import os from 'node:os'
 
 import { IRequestStrict, IttyRouter, type RequestHandler } from 'itty-router'
-import {
-	fetch as secretStreamFetch,
-	Agent as SecretStreamAgent,
-} from 'secret-stream-http'
+import { Agent as SecretStreamAgent } from 'secret-stream-http'
 import { Type as T, type Static } from 'typebox'
 import { Compile } from 'typebox/compile'
 
 import type { Context } from '../context.js'
 import { errors, StatusError } from '../lib/errors.js'
 import { createEventStreamResponse } from '../lib/event-stream-response.js'
 import { MapShare } from '../lib/map-share.js'
+import { anyFetch } from '../lib/secret-stream-fetch.js'
 import { SelfEvictingTimeoutMap } from '../lib/self-evicting-map.js'
 import { addTrailingSlash, timingSafeEqual } from '../lib/utils.js'
 import { localhostOnly } from '../middlewares/localhost-only.js'
@@ -127,6 +125,14 @@ export function MapSharesRouter(
 	router.all('/:shareId', validateRemoteDeviceId)
 	router.all('/:shareId/*', validateRemoteDeviceId)
 
+	// Lightweight non-mutating probe used by remote peers to pick a reachable
+	// URL without triggering stateful handlers like /download or /decline.
+	// Runs under validateRemoteDeviceId so it inherits the same authorization
+	// as every other remote-accessible route.
+	const probeHandler = () => new Response(null, { status: 204 })
+	router.options('/:shareId', probeHandler)
+	router.options('/:shareId/*', probeHandler)
+
 	router.get('/:shareId', async (request): Promise<MapShareState> => {
 		return getMapShare(request.params.shareId).state
 	})
@@ -152,28 +158,27 @@ export function MapSharesRouter(
 		const { senderDeviceId, mapShareUrls, reason } = parsedBody
 		const remotePublicKey = Buffer.from(senderDeviceId, 'hex')
 		const keyPair = ctx.getKeyPair()
-		let response: Response | undefined
-		// The sharer could have multiple IPs for different network interfaces, and
-		// not all of them may be on the same network as us, so try each URL until
-		// one works
-		for (const mapShareUrl of mapShareUrls) {
-			const url = new URL('decline', addTrailingSlash(mapShareUrl))
-			try {
-				response = (await secretStreamFetch(url, {
-					method: 'POST',
-					body: JSON.stringify({ reason }),
-					signal: request.signal,
-					dispatcher: new SecretStreamAgent({ remotePublicKey, keyPair }),
-				})) as unknown as Response // Subtle difference bewteen Undici fetch Response and whatwg Response
-				break // Exit loop on successful fetch
-			} catch (error) {
-				if (error instanceof DOMException && error.name === 'AbortError') {
-					throw error // Handle abort in caller
-				}
-				// Otherwise, try the next URL
+
+		// The sharer could have multiple IPs for different network interfaces
+		// and not all of them may be on the same network as us. anyFetch picks
+		// a reachable URL via an OPTIONS probe before POSTing /decline, so we
+		// don't race a stateful endpoint against a single server reachable on
+		// several URLs.
+		const declineUrls = mapShareUrls.map(
+			(mapShareUrl) => new URL('decline', addTrailingSlash(mapShareUrl)),
+		) as unknown as readonly [URL, ...URL[]]
+		let response: Response
+		try {
+			response = await anyFetch(declineUrls, {
+				method: 'POST',
+				body: JSON.stringify({ reason }),
+				signal: request.signal,
+				dispatcher: new SecretStreamAgent({ remotePublicKey, keyPair }),
+			})
+		} catch (error) {
+			if (error instanceof DOMException && error.name === 'AbortError') {
+				throw error // Handle abort in caller
 			}
-		}
-		if (!response) {
 			throw new errors.DECLINE_CANNOT_CONNECT()
 		}
 		if (!response.ok) {
diff --git a/test/secret-stream-fetch.test.ts b/test/secret-stream-fetch.test.ts
