From 30172f150adeccd182e1631e2ca5ee8058d51d2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Jesus?= Date: Fri, 28 Aug 2026 17:19:25 +0100 Subject: [PATCH 1/3] feat: add parallel bandwidth measurements --- README.md | 22 + src/Results/MeasurementCalculations.ts | 12 +- src/config/defaultConfig.ts | 8 + .../BandwidthEngine/BandwidthEngine.ts | 515 +++++++++++------- .../BandwidthEngine/LoggingBandwidthEngine.ts | 3 +- src/index.ts | 54 +- src/types.ts | 3 + src/utils/parallelism.ts | 10 + .../Results/MeasurementCalculations.test.ts | 21 + tests/unit/config/defaultConfig.test.ts | 5 + tests/unit/engines/parallelism.test.ts | 246 +++++++++ 11 files changed, 693 insertions(+), 206 deletions(-) create mode 100644 src/utils/parallelism.ts create mode 100644 tests/unit/engines/parallelism.test.ts diff --git a/README.md b/README.md index 90d063ab..0b0592bb 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,8 @@ them. | **autoStart**: *boolean* | Whether to automatically start the measurements on instantiation. | `true` | | **downloadApiUrl**: *string* | The URL of the API for performing download GET requests. | `https://speed.cloudflare.com/__down` | | **uploadApiUrl**: *string* | The URL of the API for performing upload POST requests. | `https://speed.cloudflare.com/__up` | +| **bandwidthOrigins**: *string[]* | Origins used for bandwidth requests. The engine appends `/__down` or `/__up` and distributes parallel requests across the origins. When omitted, `downloadApiUrl` and `uploadApiUrl` are used. | `[]` | +| **parallelism**: *number* | Maximum number of concurrent requests in each download or upload step. Must be a positive integer. | `1` | | **turnServerUri**: *string* | The URI of the TURN server used to measure packet loss. | `turn.cloudflare.com:3478` | | **turnServerCredsApiUrl**: *string* | A URI that returns TURN server credentials. Expects a JSON response with `username` and `credential` keys. | - | | **turnServerUser**: *string* | The username for the TURN server credentials. | - | @@ -136,6 +138,26 @@ Each of these measurement sets are bound to a specific file size. The engine fol | **bytes**: *number* | yes | The file size to request from the download API, or post to the upload API. The bandwidth (calculated as bits per second, or bps) for each request is calculated by dividing the `transferSize` (in bits) by the request duration (excluding the server processing time). | - | | **count**: *number* | yes | The number of requests to perform for this file size. | - | | **bypassMinDuration**: *boolean* | no | Whether the `bandwidthMinRequestDuration` check should be ignored, and the engine is instructed to proceed with the measurements of this direction in any case. | `false` | +| **parallelism**: *number* | no | Overrides the global `parallelism` for this step. `count` remains the total number of requests, which are divided into batches of this size. | global value | + +Parallel requests in one batch are reported as one bandwidth point. Its `bytes` value is the total payload across the requests, and its `bps` value is calculated across the complete overlapping transfer. When a `sessionId` is configured, the maximum concurrency expected from the configured steps is appended as `parallel=n`. + +```js +new SpeedTest({ + bandwidthOrigins: [ + 'https://speed-0.example.com', + 'https://speed-1.example.com', + 'https://speed-2.example.com', + 'https://speed-3.example.com' + ], + parallelism: 4, + measurements: [ + { type: 'download', bytes: 1e7, count: 8 }, + { type: 'upload', bytes: 1e7, count: 8 }, + { type: 'download', bytes: 2.5e7, count: 2, parallelism: 1 } + ] +}); +``` #### packetLoss diff --git a/src/Results/MeasurementCalculations.ts b/src/Results/MeasurementCalculations.ts index dc87fe1f..2dc73c9e 100644 --- a/src/Results/MeasurementCalculations.ts +++ b/src/Results/MeasurementCalculations.ts @@ -82,8 +82,16 @@ class MeasurementCalculations { Object.entries(bandwidthResults) .map(([bytes, { timings }]) => timings.map( - ({ bps, duration, ping, measTime, serverTime, transferSize }) => ({ - bytes: +bytes, + ({ + bps, + duration, + ping, + measTime, + serverTime, + transferSize, + transferredBytes + }) => ({ + bytes: transferredBytes ?? +bytes, bps, duration, ping, diff --git a/src/config/defaultConfig.ts b/src/config/defaultConfig.ts index 4105efc9..8936f600 100644 --- a/src/config/defaultConfig.ts +++ b/src/config/defaultConfig.ts @@ -10,6 +10,8 @@ export interface BandwidthMeasurementConfig { bytes: number; /** Number of requests to issue at this payload size. */ count: number; + /** Maximum requests to run concurrently for this step. Overrides the global value. */ + parallelism?: number; /** If `true`, skip the minimum-duration filter for this round. */ bypassMinDuration?: boolean; } @@ -49,6 +51,8 @@ export interface Config { downloadApiUrl: string; /** URL for upload requests. Default: `https://speed.cloudflare.com/__up`. */ uploadApiUrl: string; + /** Origins used for bandwidth requests. `/__down` or `/__up` is appended automatically. */ + bandwidthOrigins: string[]; /** URL for per-measurement logging. Set to `null` to disable. Default: `null`. */ logMeasurementApiUrl: string | null; /** URL for logging test results. Set to `null` to disable. Default: `https://speed.cloudflare.com/__results`. */ @@ -65,6 +69,8 @@ export interface Config { rpkiInvalidHost: string; /** Whether to include credentials (cookies) in fetch requests. Default: `false`. */ includeCredentials: boolean; + /** Maximum concurrent requests in each bandwidth step. Default: `1`. */ + parallelism: number; /** Optional session ID attached to measurement logs. */ sessionId: string | undefined; /** @@ -162,6 +168,7 @@ const defaultConfig: Config = { // APIs downloadApiUrl: `${REL_API_URL}/__down`, uploadApiUrl: `${REL_API_URL}/__up`, + bandwidthOrigins: [], logMeasurementApiUrl: null, logAimApiUrl: `${REL_API_URL}/__results`, turnServerUri: 'turn.speed.cloudflare.com:50000', @@ -170,6 +177,7 @@ const defaultConfig: Config = { turnServerPass: null, rpkiInvalidHost: 'invalid.rpki.cloudflare.com', includeCredentials: false, + parallelism: 1, sessionId: undefined, authorizationToken: null, authorizationEnabled: undefined, diff --git a/src/engines/BandwidthEngine/BandwidthEngine.ts b/src/engines/BandwidthEngine/BandwidthEngine.ts index a3353dff..d21494cb 100644 --- a/src/engines/BandwidthEngine/BandwidthEngine.ts +++ b/src/engines/BandwidthEngine/BandwidthEngine.ts @@ -77,6 +77,57 @@ const calcUploadSpeed = ( return !secs ? undefined : bits / secs; }; +export const aggregateRequestTimings = ( + timings: RequestTiming[], + isDown: boolean, + numBytes: number +): BandwidthMeasurementTiming => { + if (timings.length === 1) return timings[0]; + + const requestStart = Math.min(...timings.map(timing => timing.requestStart)); + const responseStart = Math.min( + ...timings.map(timing => timing.responseStart) + ); + const responseEnd = Math.max(...timings.map(timing => timing.responseEnd)); + const duration = isDown + ? responseEnd - responseStart + : Math.max(...timings.map(timing => timing.responseStart)) - requestStart; + const transferSize = timings.reduce( + (total, timing) => total + timing.transferSize, + 0 + ); + const transferredBytes = numBytes * timings.length; + const effectiveTransferSize = timings.reduce( + (total, timing) => + total + + (timing.transferSize || numBytes * (1 + ESTIMATED_HEADER_FRACTION)), + 0 + ); + const serverTimes = timings + .map(timing => timing.serverTime) + .filter(serverTime => serverTime >= 0); + + return { + transferSize, + transferredBytes, + ttfb: responseStart - requestStart, + payloadDownloadTime: isDown ? duration : 0, + serverTime: serverTimes.length + ? serverTimes.reduce((total, serverTime) => total + serverTime, 0) / + serverTimes.length + : -1, + measTime: new Date(), + ping: Math.min(...timings.map(timing => timing.ping)), + duration, + bps: isDown + ? calcDownloadSpeed( + { duration, transferSize: effectiveTransferSize }, + transferredBytes + ) + : calcUploadSpeed({ duration }, transferredBytes) + }; +}; + const genContent = (() => { const cache = new Map(); return (numBytes: number): string => { @@ -103,6 +154,13 @@ export interface BandwidthMeasurementTiming { ping: number; duration: number; bps: number | undefined; + transferredBytes?: number; +} + +export interface RequestTiming extends BandwidthMeasurementTiming { + requestStart: number; + responseStart: number; + responseEnd: number; } export interface BandwidthTimingResult extends BandwidthMeasurementTiming { @@ -129,6 +187,9 @@ export interface ResponseHookPayload { export interface BandwidthEngineOptions { downloadApiUrl?: string; uploadApiUrl?: string; + downloadApiUrls?: string[]; + uploadApiUrls?: string[]; + parallelism?: number; throttleMs?: number; estimatedServerTime?: number; serverTimeDelta?: number; @@ -136,7 +197,7 @@ export interface BandwidthEngineOptions { } /** - * Measures download and upload bandwidth via sequential HTTP requests. + * Measures download and upload bandwidth via configurable HTTP request batches. * Each request's timing is extracted from the browser's PerformanceResourceTiming * API, providing accurate transfer duration independent of JS execution overhead. * Supports configurable retry logic and abort thresholds. @@ -147,6 +208,9 @@ class BandwidthMeasurementEngine implements Engine { { downloadApiUrl, uploadApiUrl, + downloadApiUrls, + uploadApiUrls, + parallelism = 1, throttleMs = 0, estimatedServerTime = 0, serverTimeDelta = 0, @@ -154,12 +218,22 @@ class BandwidthMeasurementEngine implements Engine { }: BandwidthEngineOptions = {} ) { if (!measurements) throw new Error('Missing measurements argument'); - if (!downloadApiUrl) throw new Error('Missing downloadApiUrl argument'); - if (!uploadApiUrl) throw new Error('Missing uploadApiUrl argument'); + if (!downloadApiUrl && !downloadApiUrls?.length) { + throw new Error('Missing download API URL argument'); + } + if (!uploadApiUrl && !uploadApiUrls?.length) { + throw new Error('Missing upload API URL argument'); + } + if (!Number.isInteger(parallelism) || parallelism < 1) { + throw new Error('parallelism must be a positive integer'); + } this.#measurements = measurements; - this.#downloadApi = downloadApiUrl; - this.#uploadApi = uploadApiUrl; + this.#downloadApis = downloadApiUrls?.length + ? downloadApiUrls + : [downloadApiUrl!]; + this.#uploadApis = uploadApiUrls?.length ? uploadApiUrls : [uploadApiUrl!]; + this.#parallelism = parallelism; this.#throttleMs = throttleMs; this.#estimatedServerTime = Math.max(0, estimatedServerTime); this.#serverTimeDelta = Math.max(0, serverTimeDelta); @@ -226,6 +300,10 @@ class BandwidthMeasurementEngine implements Engine { ) { this.#onMeasurementResult = f; } + #onRequestResult: (result: BandwidthTimingResult) => void = () => {}; + set onRequestResult(f: (result: BandwidthTimingResult) => void) { + this.#onRequestResult = f; + } #onFinished: (results: BandwidthEngineResults) => void = () => {}; // callback invoked when all the measurements are finished set onFinished(f: (results: BandwidthEngineResults) => void) { this.#onFinished = f; @@ -250,15 +328,16 @@ class BandwidthMeasurementEngine implements Engine { // Internal state #measurements: BandwidthMeasurement[]; - #downloadApi: string; - #uploadApi: string; + #downloadApis: string[]; + #uploadApis: string[]; + #parallelism: number; #running: boolean = false; #finished: Record = { down: false, up: false }; #results: BandwidthEngineResults = { down: {}, up: {} }; #measIdx: number = 0; #counter: number = 0; - #retries: number = 0; + #requestId: number = 0; #minDuration: number = -Infinity; // of current measurement #throttleMs: number = 0; #estimatedServerTime: number = 0; @@ -294,10 +373,10 @@ class BandwidthMeasurementEngine implements Engine { ? results[dir][bytes] : { timings: [], - // count all measurements with same bytes and direction + // Count logical batches with the same bytes and direction. numMeasurements: this.#measurements .filter(({ bytes: b, dir: d }) => bytes === b && dir === d) - .map(m => m.count) + .map(m => Math.ceil(m.count / this.#parallelism)) .reduce((agg, cnt) => agg + cnt, 0) }; @@ -320,11 +399,26 @@ class BandwidthMeasurementEngine implements Engine { ); }); } else { - this.#onNewMeasurementStarted(this.#measurements[measIdx], results); + this.#onNewMeasurementStarted( + { + ...this.#measurements[measIdx], + count: Math.ceil( + this.#measurements[measIdx].count / this.#parallelism + ) + }, + results + ); } } #nextMeasurement(): void { + this.#runNextMeasurement().catch(error => { + this.#setRunning(false); + this.#onConnectionError(String(error)); + }); + } + + async #runNextMeasurement(): Promise { const measurements = this.#measurements; let meas = measurements[this.#measIdx]; @@ -373,210 +467,233 @@ class BandwidthMeasurementEngine implements Engine { const { bytes: numBytes, dir } = meas; const isDown = dir === 'down'; + const apis = isDown ? this.#downloadApis : this.#uploadApis; + const batchSize = Math.min(this.#parallelism, meas.count - this.#counter); + + this.#currentAbortController?.abort('restarting engine'); + this.#currentAbortController = new AbortController(); + const abortController = this.#currentAbortController; + let abortTimeout: ReturnType | undefined; + if (this.abortRequestDuration) { + abortTimeout = setTimeout(() => { + const errorMessage = `${isDown ? 'Download' : 'Upload'} measurement of ${numBytes} bytes aborted. Measurement exceeded bandwidthAbortRequestDuration (${this.abortRequestDuration}ms)`; + this.#cancelCurrentMeasurement(errorMessage); + this.#setRunning(false); + this.#onConnectionError(errorMessage); + }, this.abortRequestDuration); + abortController.signal.addEventListener('abort', () => + clearTimeout(abortTimeout) + ); + } - const apiUrl = isDown ? this.#downloadApi : this.#uploadApi; - const qsParams: Record = Object.assign({}, this.#qsParams); - qsParams.bytes = `${numBytes}`; + try { + const timings = await Promise.all( + Array.from({ length: batchSize }, (_, offset) => { + const apiUrl = apis[(this.#counter + offset) % apis.length]; + return this.#fetchMeasurement( + apiUrl, + numBytes, + isDown, + abortController, + `${this.#measIdx}-${this.#requestId++}` + ); + }) + ); + clearTimeout(abortTimeout); + if (abortController.signal.aborted) return; + + const timing = aggregateRequestTimings(timings, isDown, numBytes); + this.#saveMeasurementResults(measIdx, timing); + this.#minDuration = + this.#minDuration < 0 + ? timing.duration + : Math.min(this.#minDuration, timing.duration); + this.#counter += batchSize; + + if (this.#throttleMs) { + const throttleTimeout = setTimeout( + () => this.#nextMeasurement(), + this.#throttleMs + ); + abortController.signal.addEventListener('abort', () => + clearTimeout(throttleTimeout) + ); + } else { + this.#nextMeasurement(); + } + } catch (error) { + clearTimeout(abortTimeout); + if (abortController.signal.aborted) return; + this.#setRunning(false); + this.#onConnectionError(String(error)); + } + } + async #fetchMeasurement( + apiUrl: string, + numBytes: number, + isDown: boolean, + abortController: AbortController, + requestId: string + ): Promise { + const qsParams: Record = { + ...this.#qsParams, + bytes: `${numBytes}`, + ...(this.#parallelism > 1 && { + __cf_speedtest_request: requestId + }) + }; const urlObj = new URL(apiUrl, window.location.origin); - Object.entries(qsParams).forEach(([k, v]) => urlObj.searchParams.set(k, v)); + Object.entries(qsParams).forEach(([key, value]) => + urlObj.searchParams.set(key, value) + ); const url = urlObj.href; - - const fetchOpt: RequestInit = withAuthorizationHeader( - Object.assign( - {}, - isDown - ? {} - : { - method: 'POST', - body: genContent(numBytes) - }, - this.#fetchOptions - ), + const fetchOptions = withAuthorizationHeader( + { + ...(isDown ? {} : { method: 'POST', body: genContent(numBytes) }), + ...this.#fetchOptions + }, this.#authorization, url ); - if (this.#retries === 0) { - // abort existing abort controller - this.#currentAbortController?.abort('restarting engine'); - - // create new abort controller - this.#currentAbortController = new AbortController(); - if (this.abortRequestDuration) { - const abortTimeout = setTimeout(() => { - const errorMessage = `${isDown ? 'Download' : 'Upload'} measurement of ${numBytes} bytes aborted. Measurement exceeded bandwidthAbortRequestDuration (${this.abortRequestDuration}ms)`; - this.#cancelCurrentMeasurement(errorMessage); - this.#retries = 0; - this.#setRunning(false); - this.#onConnectionError(errorMessage); - }, this.abortRequestDuration); - this.#currentAbortController.signal.addEventListener('abort', () => - clearTimeout(abortTimeout) + let lastError: unknown; + for (let retry = 0; retry <= MAX_RETRIES; retry += 1) { + try { + return await this.#performFetch( + url, + fetchOptions, + numBytes, + isDown, + qsParams, + abortController.signal ); + } catch (error) { + if (abortController.signal.aborted) throw error; + lastError = error; + console.warn(`Error fetching ${url}: ${error}`); } } - let serverTime: number | undefined; - fetch(url, { - ...fetchOpt, - signal: this.#currentAbortController!.signal - }) - .then(r => { - if (r.ok) return r; - throw Error(r.statusText); - }) - .then(r => { - this.getServerTime && (serverTime = this.getServerTime(r)); - return r; - }) - .then(r => - r.text().then(body => { - this.#responseHook({ - url, - headers: r.headers, - body - }); - - return body; - }) - ) - .then(() => { - const perf = performance - .getEntriesByName(url) - .slice(-1)[0] as PerformanceResourceTiming; // get latest perf timing - const timing: BandwidthMeasurementTiming = { - transferSize: perf.transferSize, - ttfb: getTtfb(perf), - payloadDownloadTime: getPayloadDownload(perf), - serverTime: serverTime || -1, - measTime: new Date(), - ping: 0, - duration: 0, - bps: undefined - }; - // Detect new TCP connection from handshake timings. - let connectTime = 0; - if (perf.secureConnectionStart > perf.connectStart) { - connectTime = perf.secureConnectionStart - perf.connectStart; - } else { - connectTime = perf.connectEnd - perf.connectStart; - } - - const protoMatch = perf.nextHopProtocol.match(/([0-9.]+)/); - const httpVersion = protoMatch ? +protoMatch[1] : 0; - - // Calibrate serverTimeDelta from new TCP connections (HTTP/1.1) - if (serverTime && connectTime && httpVersion > 0 && httpVersion < 2) { - const derivedTotalServerTime = Math.max(0, timing.ttfb - connectTime); - const delta = derivedTotalServerTime - serverTime; - if ( - delta > 0 && - delta <= SERVER_TIME_DELTA_MAX && - delta <= serverTime && - serverTime <= SERVER_TIME_CALIBRATION_MAX - ) { - this.#serverTimeDelta = - this.#serverTimeDelta * (1 - SERVER_TIME_DELTA_WEIGHT) + - delta * SERVER_TIME_DELTA_WEIGHT; - console.log( - `serverTimeDelta (estimated): ${this.#serverTimeDelta.toFixed(2)}ms` - ); - } else if (delta > 0) { - console.log(`serverTimeDelta (skipped): ${delta.toFixed(2)}ms`); - } - } - - const baseServerTime = serverTime || this.#estimatedServerTime; - timing.ping = timing.ttfb - baseServerTime - this.#serverTimeDelta; - - // Discard the delta adjustment if it would collapse the ping - if (timing.ping <= 1) { - timing.ping = Math.max(0, timing.ttfb - baseServerTime); - } - timing.duration = (isDown ? calcDownloadDuration : calcUploadDuration)( - timing - ); - timing.bps = (isDown ? calcDownloadSpeed : calcUploadSpeed)( - timing, - numBytes + throw new Error( + `Connection failed to ${url}. Gave up after ${MAX_RETRIES} retries: ${lastError}` + ); + } + + async #performFetch( + url: string, + fetchOptions: RequestInit, + numBytes: number, + isDown: boolean, + qsParams: Record, + signal: AbortSignal + ): Promise { + const response = await fetch(url, { ...fetchOptions, signal }); + if (!response.ok) throw Error(response.statusText); + + const serverTime = this.getServerTime?.(response); + const body = await response.text(); + this.#responseHook({ url, headers: response.headers, body }); + + const perf = performance.getEntriesByName(url).slice(-1)[0] as + | PerformanceResourceTiming + | undefined; + if (!perf) throw new Error(`Missing resource timing for ${url}`); + + const timing: RequestTiming = { + transferSize: perf.transferSize, + ttfb: getTtfb(perf), + payloadDownloadTime: getPayloadDownload(perf), + serverTime: serverTime || -1, + measTime: new Date(), + ping: 0, + duration: 0, + bps: undefined, + requestStart: perf.requestStart, + responseStart: perf.responseStart, + responseEnd: perf.responseEnd + }; + + let connectTime = 0; + if (perf.secureConnectionStart > perf.connectStart) { + connectTime = perf.secureConnectionStart - perf.connectStart; + } else { + connectTime = perf.connectEnd - perf.connectStart; + } + const protoMatch = perf.nextHopProtocol.match(/([0-9.]+)/); + const httpVersion = protoMatch ? +protoMatch[1] : 0; + if (serverTime && connectTime && httpVersion > 0 && httpVersion < 2) { + const derivedTotalServerTime = Math.max(0, timing.ttfb - connectTime); + const delta = derivedTotalServerTime - serverTime; + if ( + delta > 0 && + delta <= SERVER_TIME_DELTA_MAX && + delta <= serverTime && + serverTime <= SERVER_TIME_CALIBRATION_MAX + ) { + this.#serverTimeDelta = + this.#serverTimeDelta * (1 - SERVER_TIME_DELTA_WEIGHT) + + delta * SERVER_TIME_DELTA_WEIGHT; + console.log( + `serverTimeDelta (estimated): ${this.#serverTimeDelta.toFixed(2)}ms` ); + } else if (delta > 0) { + console.log(`serverTimeDelta (skipped): ${delta.toFixed(2)}ms`); + } + } - // Log measurement details - const delta = this.#serverTimeDelta; - if (+numBytes === 0) { - console.log('latency', { - phase: `during ${qsParams.during || 'idle'}`, - ttfb: timing.ttfb, - serverTime: baseServerTime, - ...(delta && { serverTimeDelta: delta }), - ping: timing.ping - }); - } else { - console.log(isDown ? 'download' : 'upload', { - bytes: +numBytes, - bps: timing.bps, - ttfb: timing.ttfb, - serverTime: baseServerTime, - ...(delta && { serverTimeDelta: delta }), - ping: timing.ping - }); - } - - if (isDown && numBytes) { - const reqSize = +numBytes; - if ( - timing.transferSize && - (timing.transferSize < reqSize || - timing.transferSize / reqSize > 1.05) - ) { - // log if transferSize is too different from requested size - console.warn( - `Requested ${reqSize}B but received ${timing.transferSize}B (${ - Math.round((timing.transferSize / reqSize) * 1e4) / 1e2 - }%).` - ); - } - } - - this.#saveMeasurementResults(measIdx, timing); - const requestDuration = timing.duration; - this.#minDuration = - this.#minDuration < 0 - ? requestDuration - : Math.min(this.#minDuration, requestDuration); // carry minimum request duration - - this.#counter += 1; - this.#retries = 0; - - if (this.#throttleMs) { - const throttleTimeout = setTimeout( - () => this.#nextMeasurement(), - this.#throttleMs - ); - this.#currentAbortController!.signal.addEventListener('abort', () => - clearTimeout(throttleTimeout) - ); - } else { - this.#nextMeasurement(); - } - }) - .catch(error => { - if (this.#currentAbortController!.signal.aborted) { - return; - } - console.warn(`Error fetching ${url}: ${error}`); + const baseServerTime = serverTime || this.#estimatedServerTime; + timing.ping = timing.ttfb - baseServerTime - this.#serverTimeDelta; + if (timing.ping <= 1) { + timing.ping = Math.max(0, timing.ttfb - baseServerTime); + } + timing.duration = (isDown ? calcDownloadDuration : calcUploadDuration)( + timing + ); + timing.bps = (isDown ? calcDownloadSpeed : calcUploadSpeed)( + timing, + numBytes + ); - if (this.#retries++ < MAX_RETRIES) { - this.#nextMeasurement(); // keep trying - } else { - this.#retries = 0; - this.#setRunning(false); - this.#onConnectionError( - `Connection failed to ${url}. Gave up after ${MAX_RETRIES} retries.` - ); - } + const delta = this.#serverTimeDelta; + if (numBytes === 0) { + console.log('latency', { + phase: `during ${qsParams.during || 'idle'}`, + ttfb: timing.ttfb, + serverTime: baseServerTime, + ...(delta && { serverTimeDelta: delta }), + ping: timing.ping }); + } else { + console.log(isDown ? 'download' : 'upload', { + bytes: numBytes, + bps: timing.bps, + ttfb: timing.ttfb, + serverTime: baseServerTime, + ...(delta && { serverTimeDelta: delta }), + ping: timing.ping + }); + } + + if ( + isDown && + numBytes && + timing.transferSize && + (timing.transferSize < numBytes || timing.transferSize / numBytes > 1.05) + ) { + console.warn( + `Requested ${numBytes}B but received ${timing.transferSize}B (${ + Math.round((timing.transferSize / numBytes) * 1e4) / 1e2 + }%).` + ); + } + + this.#onRequestResult({ + type: isDown ? 'down' : 'up', + bytes: numBytes, + ...timing + }); + return timing; } #cancelCurrentMeasurement(reason?: string): void { diff --git a/src/engines/BandwidthEngine/LoggingBandwidthEngine.ts b/src/engines/BandwidthEngine/LoggingBandwidthEngine.ts index a8127cfe..06c1575e 100644 --- a/src/engines/BandwidthEngine/LoggingBandwidthEngine.ts +++ b/src/engines/BandwidthEngine/LoggingBandwidthEngine.ts @@ -45,7 +45,7 @@ class LoggingBandwidthEngine extends BandwidthEngine { super.qsParams = logApiUrl ? { measId: this.#measurementId! } : {}; super.responseHook = (r: ResponseHookPayload) => this.#loggingResponseHook(r); - super.onMeasurementResult = (meas: BandwidthTimingResult) => + super.onRequestResult = (meas: BandwidthTimingResult) => this.#logMeasurement(meas); } @@ -74,7 +74,6 @@ class LoggingBandwidthEngine extends BandwidthEngine { ...restArgs: [BandwidthEngineResults] ) => { onMeasurementResult(meas, ...restArgs); - this.#logMeasurement(meas); }; } diff --git a/src/index.ts b/src/index.ts index d38ee3a8..b1e8e2cf 100644 --- a/src/index.ts +++ b/src/index.ts @@ -16,6 +16,7 @@ import logFinalResults, { type AimLogResponse } from './logging/logFinalResults'; import type { AuthorizationOptions } from './utils/authorization'; +import { appendParallelism } from './utils/parallelism'; const DEFAULT_OPTIMAL_DOWNLOAD_SIZE = 1e6; const DEFAULT_OPTIMAL_UPLOAD_SIZE = 1e6; @@ -50,6 +51,8 @@ interface MeasurementStep { count?: number; /** Skip the minimum-duration filter for this round (download/upload types). */ bypassMinDuration?: boolean; + /** Maximum concurrent requests for this bandwidth step. */ + parallelism?: number; /** Number of packets sent per batch (packetLoss types). */ batchSize?: number; /** Delay between batches in ms (packetLoss types). */ @@ -108,6 +111,24 @@ const pausableTypes: MeasurementType[] = [ // TODO: consider replacing with crypto.randomUUID() for better uniqueness const genMeasId = (): string => `${Math.round(Math.random() * 1e16)}`; +const validateParallelism = (parallelism: number): number => { + if (!Number.isInteger(parallelism) || parallelism < 1) { + throw new Error('parallelism must be a positive integer'); + } + return parallelism; +}; + +const getMaximumParallelism = (config: SpeedTestConfig): number => + config.measurements.reduce((maximum, measurement) => { + if (measurement.type !== 'download' && measurement.type !== 'upload') { + return maximum; + } + const parallelism = validateParallelism( + measurement.parallelism ?? config.parallelism + ); + return Math.max(maximum, Math.min(parallelism, measurement.count ?? 1)); + }, 1); + /** * Core speed test engine that orchestrates measurement phases (latency, * download, upload, packet loss, reachability) and exposes results via @@ -129,6 +150,15 @@ class MeasurementEngine { userConfig, internalConfig ) as SpeedTestConfig; + validateParallelism(this.#config.parallelism); + this.#config.measurements.forEach(measurement => { + if ( + (measurement.type === 'download' || measurement.type === 'upload') && + measurement.parallelism !== undefined + ) { + validateParallelism(measurement.parallelism); + } + }); // Built once: the insecure-transport warning is latched per object, so a // fresh one per access would warn on every request. this.#authorization = { @@ -154,6 +184,13 @@ class MeasurementEngine { return this.#authorization; } + protected get loggingSessionId(): string | undefined { + return appendParallelism( + this.#config.sessionId, + getMaximumParallelism(this.#config) + ); + } + /** Not paused and not finished. */ get isRunning(): boolean { return this.#running; @@ -279,6 +316,14 @@ class MeasurementEngine { : this.#config.measurements[this.#curMsmIdx].type; } + #bandwidthApiUrls(type: 'download' | 'upload'): string[] | undefined { + if (!this.#config.bandwidthOrigins.length) return undefined; + const path = type === 'download' ? '/__down' : '/__up'; + return this.#config.bandwidthOrigins.map(origin => + new URL(path, origin).toString() + ); + } + #curTypeResults(): MeasurementResult | undefined { const type = this.#curType(); if (!type) return undefined; @@ -510,7 +555,7 @@ class MeasurementEngine { serverTimeDelta: this.#serverTimeDelta, logApiUrl: this.#config.logMeasurementApiUrl ?? undefined, measurementId: this.#measurementId, - sessionId: this.#config.sessionId, + sessionId: this.loggingSessionId, authorization: this.authorization, // if under load @@ -593,13 +638,16 @@ class MeasurementEngine { { downloadApiUrl, uploadApiUrl, + downloadApiUrls: this.#bandwidthApiUrls('download'), + uploadApiUrls: this.#bandwidthApiUrls('upload'), + parallelism: msmConfig.parallelism ?? this.#config.parallelism, estimatedServerTime, serverTimeDelta: this.#serverTimeDelta, logApiUrl: this.#config.logMeasurementApiUrl ?? undefined, measurementId: this.#measurementId, measureParallelLatency, parallelLatencyThrottleMs: this.#config.loadedLatencyThrottle, - sessionId: this.#config.sessionId, + sessionId: this.loggingSessionId, authorization: this.authorization } ) as Engine; @@ -793,7 +841,7 @@ class SpeedTestEngine extends MeasurementEngine { } logFinalResults(results, { apiUrl, - sessionId: this.config.sessionId, + sessionId: this.loggingSessionId, authorization: this.authorization }).then(response => { this.onResultsLogged(response); diff --git a/src/types.ts b/src/types.ts index 3660ee9d..174acce3 100644 --- a/src/types.ts +++ b/src/types.ts @@ -37,6 +37,9 @@ export interface BandwidthTiming { /** Actual number of bytes transferred (from `PerformanceResourceTiming`). */ transferSize: number; + + /** Total payload bytes represented by an aggregated parallel sample. */ + transferredBytes?: number; } /** diff --git a/src/utils/parallelism.ts b/src/utils/parallelism.ts new file mode 100644 index 00000000..5152332c --- /dev/null +++ b/src/utils/parallelism.ts @@ -0,0 +1,10 @@ +export const appendParallelism = ( + sessionId: string | undefined, + parallelism: number +): string | undefined => { + if (!sessionId || parallelism <= 1) return sessionId; + const fields = sessionId + .split('&') + .filter(field => !field.startsWith('parallel=')); + return [...fields, `parallel=${parallelism}`].join('&'); +}; diff --git a/tests/unit/Results/MeasurementCalculations.test.ts b/tests/unit/Results/MeasurementCalculations.test.ts index c727d351..a4c42391 100644 --- a/tests/unit/Results/MeasurementCalculations.test.ts +++ b/tests/unit/Results/MeasurementCalculations.test.ts @@ -119,6 +119,27 @@ describe('MeasurementCalculations', () => { expect(result[0].bytes).toBe(100000); expect(result[1].bytes).toBe(1000000); }); + + it('uses the aggregate byte count from parallel samples', () => { + const calc = createCalc(); + const [result] = calc.getBandwidthPoints({ + 100000: { + timings: [ + { + bps: 10e6, + duration: 100, + ping: 10, + measTime: new Date(100), + serverTime: 5, + transferSize: 400000, + transferredBytes: 400000 + } + ] + } + }); + + expect(result.bytes).toBe(400000); + }); }); describe('getBandwidth', () => { diff --git a/tests/unit/config/defaultConfig.test.ts b/tests/unit/config/defaultConfig.test.ts index d94c68c8..2e829b4e 100644 --- a/tests/unit/config/defaultConfig.test.ts +++ b/tests/unit/config/defaultConfig.test.ts @@ -52,6 +52,11 @@ describe('defaultConfig', () => { expect(defaultConfig.includeCredentials).toBe(false); }); + it('runs bandwidth requests sequentially by default', () => { + expect(defaultConfig.parallelism).toBe(1); + expect(defaultConfig.bandwidthOrigins).toEqual([]); + }); + it('has null values for optional TURN server credentials', () => { expect(defaultConfig.turnServerUser).toBeNull(); expect(defaultConfig.turnServerPass).toBeNull(); diff --git a/tests/unit/engines/parallelism.test.ts b/tests/unit/engines/parallelism.test.ts new file mode 100644 index 00000000..74affbc8 --- /dev/null +++ b/tests/unit/engines/parallelism.test.ts @@ -0,0 +1,246 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import SpeedTest from '../../../src/index.ts'; +import { appendParallelism } from '../../../src/utils/parallelism.ts'; +import BandwidthEngine, { + aggregateRequestTimings, + type RequestTiming +} from '../../../src/engines/BandwidthEngine/BandwidthEngine.ts'; + +const timing = ( + requestStart: number, + responseStart: number, + responseEnd: number +): RequestTiming => ({ + requestStart, + responseStart, + responseEnd, + transferSize: 1000, + ttfb: responseStart - requestStart, + payloadDownloadTime: responseEnd - responseStart, + serverTime: 2, + measTime: new Date(), + ping: 8, + duration: responseEnd - requestStart, + bps: 1 +}); + +describe('parallel bandwidth aggregation', () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('measures downloads from the first response byte to the last completion', () => { + const result = aggregateRequestTimings( + [timing(0, 10, 110), timing(5, 20, 120)], + true, + 1000 + ); + + expect(result.duration).toBe(110); + expect(result.transferredBytes).toBe(2000); + expect(result.transferSize).toBe(2000); + expect(result.bps).toBeCloseTo(16000 / 0.11); + }); + + it('estimates bytes missing from resource timing', () => { + const hiddenTiming = { ...timing(5, 20, 120), transferSize: 0 }; + const result = aggregateRequestTimings( + [timing(0, 10, 110), hiddenTiming], + true, + 1000 + ); + + expect(result.transferSize).toBe(1000); + expect(result.bps).toBeCloseTo(((1000 + 1005) * 8) / 0.11); + }); + + it('measures uploads from the first request start to the last response', () => { + const result = aggregateRequestTimings( + [timing(0, 100, 105), timing(10, 120, 125)], + false, + 1000 + ); + + expect(result.duration).toBe(120); + expect(result.transferredBytes).toBe(2000); + expect(result.bps).toBeCloseTo(16080 / 0.12); + }); + + it('preserves sequential timing calculations', () => { + const singleTiming = timing(0, 10, 110); + expect(aggregateRequestTimings([singleTiming], true, 1000)).toBe( + singleTiming + ); + }); + + it('starts a batch across all configured origins', async () => { + const releases: Array<() => void> = []; + const fetchMock = vi.fn( + (_url: RequestInfo | URL) => + new Promise(resolve => { + releases.push(() => resolve(new Response('body'))); + }) + ); + vi.stubGlobal('fetch', fetchMock); + vi.stubGlobal('window', { location: { origin: 'https://app.example' } }); + vi.stubGlobal('performance', { + clearResourceTimings: vi.fn(), + getEntriesByName: (url: string) => { + const index = Number( + new URL(url).searchParams.get('__cf_speedtest_request')?.split('-')[1] + ); + return [ + { + transferSize: 1000, + requestStart: 0, + responseStart: 10 + index, + responseEnd: 110 + index, + connectStart: 0, + connectEnd: 0, + secureConnectionStart: 0, + nextHopProtocol: 'h2' + } + ]; + } + }); + + const origins = Array.from( + { length: 4 }, + (_, index) => `https://t${index}.example/__down` + ); + const engine = new BandwidthEngine( + [{ dir: 'down', bytes: 1000, count: 6 }], + { + downloadApiUrls: origins, + uploadApiUrl: 'https://upload.example/__up', + parallelism: 4 + } + ); + const onRequestResult = vi.fn(); + const onMeasurementResult = vi.fn(); + engine.onRequestResult = onRequestResult; + engine.onMeasurementResult = onMeasurementResult; + const finished = new Promise(resolve => { + engine.onFinished = resolve; + }); + + engine.play(); + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(4)); + expect( + fetchMock.mock.calls.map(([url]) => new URL(url.toString()).origin) + ).toEqual(origins.map(origin => new URL(origin).origin)); + + releases.slice(0, 4).forEach(release => release()); + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(6)); + releases.slice(4).forEach(release => release()); + await finished; + + expect(engine.results.down[1000].timings).toHaveLength(2); + expect(engine.results.down[1000].timings[0].transferredBytes).toBe(4000); + expect(engine.results.down[1000].timings[1].transferredBytes).toBe(2000); + expect(onRequestResult).toHaveBeenCalledTimes(6); + await vi.waitFor(() => + expect(onMeasurementResult).toHaveBeenCalledTimes(2) + ); + }); + + it('applies global and step parallelism through the public API', async () => { + const resultsUrl = 'https://results.example/__results'; + const fetchMock = vi.fn((url: RequestInfo | URL, _init?: RequestInit) => + Promise.resolve( + new Response(url.toString() === resultsUrl ? '{}' : url.toString()) + ) + ); + vi.stubGlobal('fetch', fetchMock); + vi.stubGlobal('window', { location: { origin: 'https://app.example' } }); + vi.stubGlobal('performance', { + now: vi.fn(() => 1), + clearResourceTimings: vi.fn(), + setResourceTimingBufferSize: vi.fn(), + getEntriesByName: (url: string) => { + const index = Number( + new URL(url).searchParams.get('__cf_speedtest_request')?.split('-')[1] + ); + return [ + { + transferSize: 1000, + requestStart: 0, + responseStart: 10 + index, + responseEnd: 110 + index, + connectStart: 0, + connectEnd: 0, + secureConnectionStart: 0, + nextHopProtocol: 'h2' + } + ]; + } + }); + + const engine = new SpeedTest({ + autoStart: false, + bandwidthOrigins: ['https://speed-0.example', 'https://speed-1.example'], + parallelism: 4, + measurements: [ + { type: 'download', bytes: 1000, count: 2, parallelism: 2 }, + { type: 'upload', bytes: 1000, count: 4 } + ], + measureDownloadLoadedLatency: false, + measureUploadLoadedLatency: false, + logAimApiUrl: resultsUrl, + sessionId: 'session=abc' + }); + const finished = new Promise((resolve, reject) => { + engine.onFinish = resolve; + engine.onError = reject; + }); + const logged = new Promise(resolve => { + engine.onResultsLogged = resolve; + }); + + engine.play(); + const results = await finished; + await logged; + + const measurementCalls = fetchMock.mock.calls.filter( + ([url]) => url.toString() !== resultsUrl + ); + const urls = measurementCalls.map(([url]) => new URL(url.toString())); + expect(urls.map(url => `${url.origin}${url.pathname}`)).toEqual([ + 'https://speed-0.example/__down', + 'https://speed-1.example/__down', + 'https://speed-0.example/__up', + 'https://speed-1.example/__up', + 'https://speed-0.example/__up', + 'https://speed-1.example/__up' + ]); + expect(results.getDownloadBandwidthPoints()[0].bytes).toBe(2000); + expect(results.getUploadBandwidthPoints()[0].bytes).toBe(4000); + + const resultsCall = fetchMock.mock.calls.find( + ([url]) => url.toString() === resultsUrl + ); + const body = JSON.parse(resultsCall?.[1]?.body as string); + expect(body.sessionId).toBe('session=abc¶llel=4'); + expect(body.download).toEqual([expect.objectContaining({ bytes: 2000 })]); + expect(body.upload).toEqual([expect.objectContaining({ bytes: 4000 })]); + }); +}); + +describe('parallel session metadata', () => { + it('appends the maximum parallelism', () => { + expect(appendParallelism('session=abc&tier=test', 4)).toBe( + 'session=abc&tier=test¶llel=4' + ); + }); + + it('replaces an existing parallelism value', () => { + expect(appendParallelism('session=abc¶llel=2', 4)).toBe( + 'session=abc¶llel=4' + ); + }); + + it('leaves sequential and absent sessions unchanged', () => { + expect(appendParallelism('session=abc', 1)).toBe('session=abc'); + expect(appendParallelism(undefined, 4)).toBeUndefined(); + }); +}); From bb3eb1812c360c3de6325aa04e838dffad8a38f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Jesus?= Date: Wed, 2 Sep 2026 14:52:22 +0100 Subject: [PATCH 2/3] refactor: use target-based bandwidth parallelism --- README.md | 27 +- src/config/defaultConfig.ts | 29 ++- .../BandwidthEngine/BandwidthEngine.ts | 231 ++++++++++++------ .../BandwidthEngine/ParallelLatency.ts | 3 + src/index.ts | 91 ++++--- src/utils/parallelism.ts | 8 +- tests/unit/config/defaultConfig.test.ts | 3 +- tests/unit/engines/parallelism.test.ts | 174 +++++++++++-- 8 files changed, 393 insertions(+), 173 deletions(-) diff --git a/README.md b/README.md index 0b0592bb..176c6f3c 100644 --- a/README.md +++ b/README.md @@ -46,10 +46,9 @@ them. | Config option | Description | Default | | --- | --- | :--: | | **autoStart**: *boolean* | Whether to automatically start the measurements on instantiation. | `true` | -| **downloadApiUrl**: *string* | The URL of the API for performing download GET requests. | `https://speed.cloudflare.com/__down` | -| **uploadApiUrl**: *string* | The URL of the API for performing upload POST requests. | `https://speed.cloudflare.com/__up` | -| **bandwidthOrigins**: *string[]* | Origins used for bandwidth requests. The engine appends `/__down` or `/__up` and distributes parallel requests across the origins. When omitted, `downloadApiUrl` and `uploadApiUrl` are used. | `[]` | -| **parallelism**: *number* | Maximum number of concurrent requests in each download or upload step. Must be a positive integer. | `1` | +| **downloadApiUrl**: *string* | Deprecated fallback URL for download GET requests when `measurementTargets` is empty. | `https://speed.cloudflare.com/__down` | +| **uploadApiUrl**: *string* | Deprecated fallback URL for upload POST requests when `measurementTargets` is empty. | `https://speed.cloudflare.com/__up` | +| **measurementTargets**: *string[]* | Origins used for latency, download, and upload requests. The engine appends `/__down` or `/__up` and distributes requests across the targets, starting at a random target. Duplicate targets are preserved. | `[]` | | **turnServerUri**: *string* | The URI of the TURN server used to measure packet loss. | `turn.cloudflare.com:3478` | | **turnServerCredsApiUrl**: *string* | A URI that returns TURN server credentials. Expects a JSON response with `username` and `credential` keys. | - | | **turnServerUser**: *string* | The username for the TURN server credentials. | - | @@ -138,26 +137,6 @@ Each of these measurement sets are bound to a specific file size. The engine fol | **bytes**: *number* | yes | The file size to request from the download API, or post to the upload API. The bandwidth (calculated as bits per second, or bps) for each request is calculated by dividing the `transferSize` (in bits) by the request duration (excluding the server processing time). | - | | **count**: *number* | yes | The number of requests to perform for this file size. | - | | **bypassMinDuration**: *boolean* | no | Whether the `bandwidthMinRequestDuration` check should be ignored, and the engine is instructed to proceed with the measurements of this direction in any case. | `false` | -| **parallelism**: *number* | no | Overrides the global `parallelism` for this step. `count` remains the total number of requests, which are divided into batches of this size. | global value | - -Parallel requests in one batch are reported as one bandwidth point. Its `bytes` value is the total payload across the requests, and its `bps` value is calculated across the complete overlapping transfer. When a `sessionId` is configured, the maximum concurrency expected from the configured steps is appended as `parallel=n`. - -```js -new SpeedTest({ - bandwidthOrigins: [ - 'https://speed-0.example.com', - 'https://speed-1.example.com', - 'https://speed-2.example.com', - 'https://speed-3.example.com' - ], - parallelism: 4, - measurements: [ - { type: 'download', bytes: 1e7, count: 8 }, - { type: 'upload', bytes: 1e7, count: 8 }, - { type: 'download', bytes: 2.5e7, count: 2, parallelism: 1 } - ] -}); -``` #### packetLoss diff --git a/src/config/defaultConfig.ts b/src/config/defaultConfig.ts index 8936f600..a8b825de 100644 --- a/src/config/defaultConfig.ts +++ b/src/config/defaultConfig.ts @@ -10,8 +10,12 @@ export interface BandwidthMeasurementConfig { bytes: number; /** Number of requests to issue at this payload size. */ count: number; - /** Maximum requests to run concurrently for this step. Overrides the global value. */ - parallelism?: number; + /** + * Runs this step using one continuously replenished request lane per target. + * + * @experimental Unstable — may change or be removed in any release. + */ + parallel?: boolean; /** If `true`, skip the minimum-duration filter for this round. */ bypassMinDuration?: boolean; } @@ -47,12 +51,20 @@ export interface Config { /** Whether to start the test immediately on construction. Default: `true`. */ autoStart: boolean; - /** URL for download requests. Default: `https://speed.cloudflare.com/__down`. */ + /** + * URL for download requests. + * + * @deprecated Use {@link measurementTargets}. This remains the fallback when no targets are configured. + */ downloadApiUrl: string; - /** URL for upload requests. Default: `https://speed.cloudflare.com/__up`. */ + /** + * URL for upload requests. + * + * @deprecated Use {@link measurementTargets}. This remains the fallback when no targets are configured. + */ uploadApiUrl: string; - /** Origins used for bandwidth requests. `/__down` or `/__up` is appended automatically. */ - bandwidthOrigins: string[]; + /** Origins used for latency, download, and upload requests. */ + measurementTargets: string[]; /** URL for per-measurement logging. Set to `null` to disable. Default: `null`. */ logMeasurementApiUrl: string | null; /** URL for logging test results. Set to `null` to disable. Default: `https://speed.cloudflare.com/__results`. */ @@ -69,8 +81,6 @@ export interface Config { rpkiInvalidHost: string; /** Whether to include credentials (cookies) in fetch requests. Default: `false`. */ includeCredentials: boolean; - /** Maximum concurrent requests in each bandwidth step. Default: `1`. */ - parallelism: number; /** Optional session ID attached to measurement logs. */ sessionId: string | undefined; /** @@ -168,7 +178,7 @@ const defaultConfig: Config = { // APIs downloadApiUrl: `${REL_API_URL}/__down`, uploadApiUrl: `${REL_API_URL}/__up`, - bandwidthOrigins: [], + measurementTargets: [], logMeasurementApiUrl: null, logAimApiUrl: `${REL_API_URL}/__results`, turnServerUri: 'turn.speed.cloudflare.com:50000', @@ -177,7 +187,6 @@ const defaultConfig: Config = { turnServerPass: null, rpkiInvalidHost: 'invalid.rpki.cloudflare.com', includeCredentials: false, - parallelism: 1, sessionId: undefined, authorizationToken: null, authorizationEnabled: undefined, diff --git a/src/engines/BandwidthEngine/BandwidthEngine.ts b/src/engines/BandwidthEngine/BandwidthEngine.ts index d21494cb..21954a9a 100644 --- a/src/engines/BandwidthEngine/BandwidthEngine.ts +++ b/src/engines/BandwidthEngine/BandwidthEngine.ts @@ -90,7 +90,7 @@ export const aggregateRequestTimings = ( ); const responseEnd = Math.max(...timings.map(timing => timing.responseEnd)); const duration = isDown - ? responseEnd - responseStart + ? responseEnd - requestStart : Math.max(...timings.map(timing => timing.responseStart)) - requestStart; const transferSize = timings.reduce( (total, timing) => total + timing.transferSize, @@ -189,7 +189,9 @@ export interface BandwidthEngineOptions { uploadApiUrl?: string; downloadApiUrls?: string[]; uploadApiUrls?: string[]; - parallelism?: number; + getDownloadApiUrl?: () => string; + getUploadApiUrl?: () => string; + parallel?: boolean; throttleMs?: number; estimatedServerTime?: number; serverTimeDelta?: number; @@ -197,7 +199,7 @@ export interface BandwidthEngineOptions { } /** - * Measures download and upload bandwidth via configurable HTTP request batches. + * Measures download and upload bandwidth via configurable HTTP requests. * Each request's timing is extracted from the browser's PerformanceResourceTiming * API, providing accurate transfer duration independent of JS execution overhead. * Supports configurable retry logic and abort thresholds. @@ -210,7 +212,9 @@ class BandwidthMeasurementEngine implements Engine { uploadApiUrl, downloadApiUrls, uploadApiUrls, - parallelism = 1, + getDownloadApiUrl, + getUploadApiUrl, + parallel = false, throttleMs = 0, estimatedServerTime = 0, serverTimeDelta = 0, @@ -218,22 +222,22 @@ class BandwidthMeasurementEngine implements Engine { }: BandwidthEngineOptions = {} ) { if (!measurements) throw new Error('Missing measurements argument'); - if (!downloadApiUrl && !downloadApiUrls?.length) { + if (!downloadApiUrl && !downloadApiUrls?.length && !getDownloadApiUrl) { throw new Error('Missing download API URL argument'); } - if (!uploadApiUrl && !uploadApiUrls?.length) { + if (!uploadApiUrl && !uploadApiUrls?.length && !getUploadApiUrl) { throw new Error('Missing upload API URL argument'); } - if (!Number.isInteger(parallelism) || parallelism < 1) { - throw new Error('parallelism must be a positive integer'); - } this.#measurements = measurements; this.#downloadApis = downloadApiUrls?.length ? downloadApiUrls : [downloadApiUrl!]; this.#uploadApis = uploadApiUrls?.length ? uploadApiUrls : [uploadApiUrl!]; - this.#parallelism = parallelism; + this.#getDownloadApiUrl = + getDownloadApiUrl ?? (() => this.#downloadApis[0]); + this.#getUploadApiUrl = getUploadApiUrl ?? (() => this.#uploadApis[0]); + this.#parallel = parallel; this.#throttleMs = throttleMs; this.#estimatedServerTime = Math.max(0, estimatedServerTime); this.#serverTimeDelta = Math.max(0, serverTimeDelta); @@ -330,7 +334,9 @@ class BandwidthMeasurementEngine implements Engine { #measurements: BandwidthMeasurement[]; #downloadApis: string[]; #uploadApis: string[]; - #parallelism: number; + #getDownloadApiUrl: () => string; + #getUploadApiUrl: () => string; + #parallel: boolean; #running: boolean = false; #finished: Record = { down: false, up: false }; @@ -338,6 +344,7 @@ class BandwidthMeasurementEngine implements Engine { #measIdx: number = 0; #counter: number = 0; #requestId: number = 0; + #parallelTimings: RequestTiming[] = []; #minDuration: number = -Infinity; // of current measurement #throttleMs: number = 0; #estimatedServerTime: number = 0; @@ -373,10 +380,10 @@ class BandwidthMeasurementEngine implements Engine { ? results[dir][bytes] : { timings: [], - // Count logical batches with the same bytes and direction. + // Parallel steps produce one logical result for all physical requests. numMeasurements: this.#measurements .filter(({ bytes: b, dir: d }) => bytes === b && dir === d) - .map(m => Math.ceil(m.count / this.#parallelism)) + .map(m => (this.#parallel ? 1 : m.count)) .reduce((agg, cnt) => agg + cnt, 0) }; @@ -402,9 +409,7 @@ class BandwidthMeasurementEngine implements Engine { this.#onNewMeasurementStarted( { ...this.#measurements[measIdx], - count: Math.ceil( - this.#measurements[measIdx].count / this.#parallelism - ) + count: this.#parallel ? 1 : this.#measurements[measIdx].count }, results ); @@ -439,6 +444,7 @@ class BandwidthMeasurementEngine implements Engine { // clear settings this.#counter = 0; this.#minDuration = -Infinity; + this.#parallelTimings = []; performance.clearResourceTimings(); do { @@ -467,48 +473,43 @@ class BandwidthMeasurementEngine implements Engine { const { bytes: numBytes, dir } = meas; const isDown = dir === 'down'; - const apis = isDown ? this.#downloadApis : this.#uploadApis; - const batchSize = Math.min(this.#parallelism, meas.count - this.#counter); this.#currentAbortController?.abort('restarting engine'); this.#currentAbortController = new AbortController(); const abortController = this.#currentAbortController; - let abortTimeout: ReturnType | undefined; - if (this.abortRequestDuration) { - abortTimeout = setTimeout(() => { - const errorMessage = `${isDown ? 'Download' : 'Upload'} measurement of ${numBytes} bytes aborted. Measurement exceeded bandwidthAbortRequestDuration (${this.abortRequestDuration}ms)`; - this.#cancelCurrentMeasurement(errorMessage); - this.#setRunning(false); - this.#onConnectionError(errorMessage); - }, this.abortRequestDuration); - abortController.signal.addEventListener('abort', () => - clearTimeout(abortTimeout) - ); - } try { - const timings = await Promise.all( - Array.from({ length: batchSize }, (_, offset) => { - const apiUrl = apis[(this.#counter + offset) % apis.length]; - return this.#fetchMeasurement( - apiUrl, - numBytes, - isDown, - abortController, - `${this.#measIdx}-${this.#requestId++}` - ); - }) - ); - clearTimeout(abortTimeout); - if (abortController.signal.aborted) return; + let timing: BandwidthMeasurementTiming; + if (this.#parallel) { + const timings = await this.#runParallelPool( + meas, + isDown, + abortController + ); + if (abortController.signal.aborted) return; + timing = aggregateRequestTimings(timings, isDown, numBytes); + this.#counter = meas.count; + this.#minDuration = Math.min(...timings.map(timing => timing.duration)); + } else { + const apiUrl = isDown + ? this.#getDownloadApiUrl() + : this.#getUploadApiUrl(); + timing = await this.#fetchMeasurement( + apiUrl, + numBytes, + isDown, + abortController, + `${this.#measIdx}-${this.#requestId++}` + ); + if (abortController.signal.aborted) return; + this.#counter += 1; + this.#minDuration = + this.#minDuration < 0 + ? timing.duration + : Math.min(this.#minDuration, timing.duration); + } - const timing = aggregateRequestTimings(timings, isDown, numBytes); this.#saveMeasurementResults(measIdx, timing); - this.#minDuration = - this.#minDuration < 0 - ? timing.duration - : Math.min(this.#minDuration, timing.duration); - this.#counter += batchSize; if (this.#throttleMs) { const throttleTimeout = setTimeout( @@ -522,24 +523,73 @@ class BandwidthMeasurementEngine implements Engine { this.#nextMeasurement(); } } catch (error) { - clearTimeout(abortTimeout); if (abortController.signal.aborted) return; this.#setRunning(false); this.#onConnectionError(String(error)); } } + async #runParallelPool( + measurement: BandwidthMeasurement, + isDown: boolean, + abortController: AbortController + ): Promise { + const configuredApis = isDown ? this.#downloadApis : this.#uploadApis; + const apis = configuredApis.length + ? configuredApis + : [isDown ? this.#getDownloadApiUrl() : this.#getUploadApiUrl()]; + let nextRequest = this.#parallelTimings.length; + const runLane = async (apiUrl: string): Promise => { + while ( + !abortController.signal.aborted && + this.#currentAbortController === abortController && + nextRequest < measurement.count + ) { + const requestId = `${this.#measIdx}-${this.#requestId++}`; + nextRequest += 1; + await this.#fetchMeasurement( + apiUrl, + measurement.bytes, + isDown, + abortController, + requestId, + completedTiming => { + if (this.#currentAbortController !== abortController) return false; + this.#parallelTimings.push(completedTiming); + return true; + } + ); + if ( + abortController.signal.aborted || + this.#currentAbortController !== abortController + ) { + return; + } + } + }; + + await Promise.all( + apis.slice(0, measurement.count).map(apiUrl => runLane(apiUrl)) + ); + return this.#parallelTimings; + } + async #fetchMeasurement( apiUrl: string, numBytes: number, isDown: boolean, abortController: AbortController, - requestId: string + requestId: string, + recordCompletion?: (timing: RequestTiming) => boolean ): Promise { + if (abortController.signal.aborted) { + throw new Error(String(abortController.signal.reason)); + } + const qsParams: Record = { ...this.#qsParams, bytes: `${numBytes}`, - ...(this.#parallelism > 1 && { + ...(this.#parallel && { __cf_speedtest_request: requestId }) }; @@ -557,27 +607,59 @@ class BandwidthMeasurementEngine implements Engine { url ); - let lastError: unknown; - for (let retry = 0; retry <= MAX_RETRIES; retry += 1) { - try { - return await this.#performFetch( - url, - fetchOptions, - numBytes, - isDown, - qsParams, - abortController.signal - ); - } catch (error) { - if (abortController.signal.aborted) throw error; - lastError = error; - console.warn(`Error fetching ${url}: ${error}`); + const requestController = new AbortController(); + const abortRequest = () => + requestController.abort(abortController.signal.reason); + abortController.signal.addEventListener('abort', abortRequest, { + once: true + }); + const timeoutMessage = `${isDown ? 'Download' : 'Upload'} measurement of ${numBytes} bytes aborted. Measurement exceeded bandwidthAbortRequestDuration (${this.abortRequestDuration}ms)`; + const abortTimeout = this.abortRequestDuration + ? setTimeout( + () => requestController.abort(timeoutMessage), + this.abortRequestDuration + ) + : undefined; + + try { + let lastError: unknown; + for (let retry = 0; retry <= MAX_RETRIES; retry += 1) { + try { + const timing = await this.#performFetch( + url, + fetchOptions, + numBytes, + isDown, + qsParams, + requestController.signal + ); + if (recordCompletion && !recordCompletion(timing)) return timing; + this.#onRequestResult({ + type: isDown ? 'down' : 'up', + bytes: numBytes, + ...timing + }); + return timing; + } catch (error) { + if (requestController.signal.aborted) { + throw new Error( + typeof requestController.signal.reason === 'string' + ? requestController.signal.reason + : String(error) + ); + } + lastError = error; + console.warn(`Error fetching ${url}: ${error}`); + } } - } - throw new Error( - `Connection failed to ${url}. Gave up after ${MAX_RETRIES} retries: ${lastError}` - ); + throw new Error( + `Connection failed to ${url}. Gave up after ${MAX_RETRIES} retries: ${lastError}` + ); + } finally { + clearTimeout(abortTimeout); + abortController.signal.removeEventListener('abort', abortRequest); + } } async #performFetch( @@ -688,11 +770,6 @@ class BandwidthMeasurementEngine implements Engine { ); } - this.#onRequestResult({ - type: isDown ? 'down' : 'up', - bytes: numBytes, - ...timing - }); return timing; } diff --git a/src/engines/BandwidthEngine/ParallelLatency.ts b/src/engines/BandwidthEngine/ParallelLatency.ts index 6db72283..1261ff33 100644 --- a/src/engines/BandwidthEngine/ParallelLatency.ts +++ b/src/engines/BandwidthEngine/ParallelLatency.ts @@ -8,6 +8,7 @@ import type { export interface ParallelLatencyOptions extends BandwidthEngineOptions { measureParallelLatency?: boolean; parallelLatencyThrottleMs?: number; + getLoadedLatencyApiUrl?: () => string; } /** @@ -24,6 +25,7 @@ class BandwidthWithParallelLatencyEngine extends BandwidthEngine { parallelLatencyThrottleMs = 100, downloadApiUrl, uploadApiUrl, + getLoadedLatencyApiUrl, estimatedServerTime = 0, serverTimeDelta = 0, authorization = null, @@ -52,6 +54,7 @@ class BandwidthWithParallelLatencyEngine extends BandwidthEngine { { downloadApiUrl, uploadApiUrl, + getDownloadApiUrl: getLoadedLatencyApiUrl, estimatedServerTime, serverTimeDelta, authorization, diff --git a/src/index.ts b/src/index.ts index b1e8e2cf..481d6654 100644 --- a/src/index.ts +++ b/src/index.ts @@ -51,8 +51,8 @@ interface MeasurementStep { count?: number; /** Skip the minimum-duration filter for this round (download/upload types). */ bypassMinDuration?: boolean; - /** Maximum concurrent requests for this bandwidth step. */ - parallelism?: number; + /** Whether this bandwidth step uses one request lane per target. */ + parallel?: boolean; /** Number of packets sent per batch (packetLoss types). */ batchSize?: number; /** Delay between batches in ms (packetLoss types). */ @@ -111,23 +111,12 @@ const pausableTypes: MeasurementType[] = [ // TODO: consider replacing with crypto.randomUUID() for better uniqueness const genMeasId = (): string => `${Math.round(Math.random() * 1e16)}`; -const validateParallelism = (parallelism: number): number => { - if (!Number.isInteger(parallelism) || parallelism < 1) { - throw new Error('parallelism must be a positive integer'); - } - return parallelism; -}; - -const getMaximumParallelism = (config: SpeedTestConfig): number => - config.measurements.reduce((maximum, measurement) => { - if (measurement.type !== 'download' && measurement.type !== 'upload') { - return maximum; - } - const parallelism = validateParallelism( - measurement.parallelism ?? config.parallelism - ); - return Math.max(maximum, Math.min(parallelism, measurement.count ?? 1)); - }, 1); +const hasParallelMeasurement = (config: SpeedTestConfig): boolean => + config.measurements.some( + measurement => + (measurement.type === 'download' || measurement.type === 'upload') && + measurement.parallel === true + ); /** * Core speed test engine that orchestrates measurement phases (latency, @@ -150,15 +139,12 @@ class MeasurementEngine { userConfig, internalConfig ) as SpeedTestConfig; - validateParallelism(this.#config.parallelism); - this.#config.measurements.forEach(measurement => { - if ( - (measurement.type === 'download' || measurement.type === 'upload') && - measurement.parallelism !== undefined - ) { - validateParallelism(measurement.parallelism); - } - }); + this.#targetIndex = this.#config.measurementTargets.length + ? Math.floor(Math.random() * this.#config.measurementTargets.length) + : 0; + this.#loadedLatencyTargetIndex = this.#config.measurementTargets.length + ? Math.floor(Math.random() * this.#config.measurementTargets.length) + : 0; // Built once: the insecure-transport warning is latched per object, so a // fresh one per access would warn on every request. this.#authorization = { @@ -187,7 +173,9 @@ class MeasurementEngine { protected get loggingSessionId(): string | undefined { return appendParallelism( this.#config.sessionId, - getMaximumParallelism(this.#config) + hasParallelMeasurement(this.#config) + ? Math.max(1, this.#config.measurementTargets.length) + : undefined ); } @@ -266,6 +254,8 @@ class MeasurementEngine { #curEngine: Engine | undefined; #optimalDownloadChunkSize: number = DEFAULT_OPTIMAL_DOWNLOAD_SIZE; #optimalUploadChunkSize: number = DEFAULT_OPTIMAL_UPLOAD_SIZE; + #targetIndex: number; + #loadedLatencyTargetIndex: number; /** * High-resolution timestamp (from performance.now()) of the test start or @@ -316,14 +306,34 @@ class MeasurementEngine { : this.#config.measurements[this.#curMsmIdx].type; } - #bandwidthApiUrls(type: 'download' | 'upload'): string[] | undefined { - if (!this.#config.bandwidthOrigins.length) return undefined; + #measurementApiUrls(type: 'download' | 'upload'): string[] | undefined { + if (!this.#config.measurementTargets.length) return undefined; const path = type === 'download' ? '/__down' : '/__up'; - return this.#config.bandwidthOrigins.map(origin => + return this.#config.measurementTargets.map(origin => new URL(path, origin).toString() ); } + #nextMeasurementApiUrl = (type: 'download' | 'upload'): string => { + const urls = this.#measurementApiUrls(type); + if (!urls) { + return type === 'download' + ? this.#config.downloadApiUrl + : this.#config.uploadApiUrl; + } + const url = urls[this.#targetIndex % urls.length]; + this.#targetIndex += 1; + return url; + }; + + #nextLoadedLatencyApiUrl = (): string => { + const urls = this.#measurementApiUrls('download'); + if (!urls) return this.#config.downloadApiUrl; + const url = urls[this.#loadedLatencyTargetIndex % urls.length]; + this.#loadedLatencyTargetIndex += 1; + return url; + }; + #curTypeResults(): MeasurementResult | undefined { const type = this.#curType(); if (!type) return undefined; @@ -339,6 +349,12 @@ class MeasurementEngine { this.#measurementId = genMeasId(); this.#curMsmIdx = -1; this.#curEngine = undefined; + this.#targetIndex = this.#config.measurementTargets.length + ? Math.floor(Math.random() * this.#config.measurementTargets.length) + : 0; + this.#loadedLatencyTargetIndex = this.#config.measurementTargets.length + ? Math.floor(Math.random() * this.#config.measurementTargets.length) + : 0; this.#setRunning(false); this.#setFinished(false); @@ -551,6 +567,8 @@ class MeasurementEngine { { downloadApiUrl, uploadApiUrl, + getDownloadApiUrl: () => this.#nextMeasurementApiUrl('download'), + getUploadApiUrl: () => this.#nextMeasurementApiUrl('upload'), estimatedServerTime, serverTimeDelta: this.#serverTimeDelta, logApiUrl: this.#config.logMeasurementApiUrl ?? undefined, @@ -638,9 +656,12 @@ class MeasurementEngine { { downloadApiUrl, uploadApiUrl, - downloadApiUrls: this.#bandwidthApiUrls('download'), - uploadApiUrls: this.#bandwidthApiUrls('upload'), - parallelism: msmConfig.parallelism ?? this.#config.parallelism, + downloadApiUrls: this.#measurementApiUrls('download'), + uploadApiUrls: this.#measurementApiUrls('upload'), + getDownloadApiUrl: () => this.#nextMeasurementApiUrl('download'), + getUploadApiUrl: () => this.#nextMeasurementApiUrl('upload'), + getLoadedLatencyApiUrl: this.#nextLoadedLatencyApiUrl, + parallel: msmConfig.parallel === true, estimatedServerTime, serverTimeDelta: this.#serverTimeDelta, logApiUrl: this.#config.logMeasurementApiUrl ?? undefined, diff --git a/src/utils/parallelism.ts b/src/utils/parallelism.ts index 5152332c..ea028396 100644 --- a/src/utils/parallelism.ts +++ b/src/utils/parallelism.ts @@ -1,10 +1,12 @@ export const appendParallelism = ( sessionId: string | undefined, - parallelism: number + parallelism: number | undefined ): string | undefined => { - if (!sessionId || parallelism <= 1) return sessionId; + if (!sessionId) return sessionId; const fields = sessionId .split('&') .filter(field => !field.startsWith('parallel=')); - return [...fields, `parallel=${parallelism}`].join('&'); + return parallelism === undefined + ? fields.join('&') + : [...fields, `parallel=${parallelism}`].join('&'); }; diff --git a/tests/unit/config/defaultConfig.test.ts b/tests/unit/config/defaultConfig.test.ts index 2e829b4e..4b213b02 100644 --- a/tests/unit/config/defaultConfig.test.ts +++ b/tests/unit/config/defaultConfig.test.ts @@ -53,8 +53,7 @@ describe('defaultConfig', () => { }); it('runs bandwidth requests sequentially by default', () => { - expect(defaultConfig.parallelism).toBe(1); - expect(defaultConfig.bandwidthOrigins).toEqual([]); + expect(defaultConfig.measurementTargets).toEqual([]); }); it('has null values for optional TURN server credentials', () => { diff --git a/tests/unit/engines/parallelism.test.ts b/tests/unit/engines/parallelism.test.ts index 74affbc8..df244ba0 100644 --- a/tests/unit/engines/parallelism.test.ts +++ b/tests/unit/engines/parallelism.test.ts @@ -27,19 +27,20 @@ const timing = ( describe('parallel bandwidth aggregation', () => { afterEach(() => { vi.unstubAllGlobals(); + vi.restoreAllMocks(); }); - it('measures downloads from the first response byte to the last completion', () => { + it('measures downloads from the first request to the last completion', () => { const result = aggregateRequestTimings( [timing(0, 10, 110), timing(5, 20, 120)], true, 1000 ); - expect(result.duration).toBe(110); + expect(result.duration).toBe(120); expect(result.transferredBytes).toBe(2000); expect(result.transferSize).toBe(2000); - expect(result.bps).toBeCloseTo(16000 / 0.11); + expect(result.bps).toBeCloseTo(16000 / 0.12); }); it('estimates bytes missing from resource timing', () => { @@ -51,7 +52,7 @@ describe('parallel bandwidth aggregation', () => { ); expect(result.transferSize).toBe(1000); - expect(result.bps).toBeCloseTo(((1000 + 1005) * 8) / 0.11); + expect(result.bps).toBeCloseTo(((1000 + 1005) * 8) / 0.12); }); it('measures uploads from the first request start to the last response', () => { @@ -73,7 +74,7 @@ describe('parallel bandwidth aggregation', () => { ); }); - it('starts a batch across all configured origins', async () => { + it('continuously replenishes one request lane per target', async () => { const releases: Array<() => void> = []; const fetchMock = vi.fn( (_url: RequestInfo | URL) => @@ -113,7 +114,7 @@ describe('parallel bandwidth aggregation', () => { { downloadApiUrls: origins, uploadApiUrl: 'https://upload.example/__up', - parallelism: 4 + parallel: true } ); const onRequestResult = vi.fn(); @@ -130,21 +131,145 @@ describe('parallel bandwidth aggregation', () => { fetchMock.mock.calls.map(([url]) => new URL(url.toString()).origin) ).toEqual(origins.map(origin => new URL(origin).origin)); - releases.slice(0, 4).forEach(release => release()); + releases[0](); + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(5)); + expect(new URL(fetchMock.mock.calls[4][0].toString()).origin).toBe( + new URL(origins[0]).origin + ); + releases[4](); await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(6)); - releases.slice(4).forEach(release => release()); + expect(new URL(fetchMock.mock.calls[5][0].toString()).origin).toBe( + new URL(origins[0]).origin + ); + releases.slice(1, 4).forEach(release => release()); + releases[5](); await finished; - expect(engine.results.down[1000].timings).toHaveLength(2); - expect(engine.results.down[1000].timings[0].transferredBytes).toBe(4000); - expect(engine.results.down[1000].timings[1].transferredBytes).toBe(2000); + expect(engine.results.down[1000].timings).toHaveLength(1); + expect(engine.results.down[1000].timings[0].transferredBytes).toBe(6000); expect(onRequestResult).toHaveBeenCalledTimes(6); - await vi.waitFor(() => - expect(onMeasurementResult).toHaveBeenCalledTimes(2) + await vi.waitFor(() => expect(onMeasurementResult).toHaveBeenCalledOnce()); + }); + + it('retains completed requests when a parallel step resumes', async () => { + const releases: Array<() => void> = []; + const fetchMock = vi.fn( + (_url: RequestInfo | URL, init?: RequestInit) => + new Promise((resolve, reject) => { + releases.push(() => resolve(new Response('body'))); + init?.signal?.addEventListener( + 'abort', + () => reject(init.signal?.reason), + { once: true } + ); + }) + ); + vi.stubGlobal('fetch', fetchMock); + vi.stubGlobal('window', { location: { origin: 'https://app.example' } }); + vi.stubGlobal('performance', { + clearResourceTimings: vi.fn(), + getEntriesByName: () => [ + { + transferSize: 1000, + requestStart: 0, + responseStart: 10, + responseEnd: 110, + connectStart: 0, + connectEnd: 0, + secureConnectionStart: 0, + nextHopProtocol: 'h2' + } + ] + }); + + const engine = new BandwidthEngine( + [{ dir: 'down', bytes: 1000, count: 4 }], + { + downloadApiUrls: [ + 'https://speed-0.example/__down', + 'https://speed-1.example/__down' + ], + uploadApiUrl: 'https://upload.example/__up', + parallel: true + } + ); + const onRequestResult = vi.fn(); + engine.onRequestResult = onRequestResult; + onRequestResult.mockImplementationOnce(() => engine.pause()); + const finished = new Promise(resolve => { + engine.onFinished = resolve; + }); + + engine.play(); + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2)); + releases[0](); + await vi.waitFor(() => expect(onRequestResult).toHaveBeenCalledOnce()); + engine.play(); + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(4)); + releases[2](); + releases[3](); + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(5)); + releases[4](); + await finished; + + expect(onRequestResult).toHaveBeenCalledTimes(4); + expect(engine.results.down[1000].timings[0].transferredBytes).toBe(4000); + }); + + it('rotates sequential requests from a randomized target', async () => { + const fetchMock = vi.fn((url: RequestInfo | URL) => + Promise.resolve(new Response(url.toString())) ); + vi.stubGlobal('fetch', fetchMock); + vi.stubGlobal('window', { location: { origin: 'https://app.example' } }); + vi.stubGlobal('performance', { + now: vi.fn(() => 1), + clearResourceTimings: vi.fn(), + setResourceTimingBufferSize: vi.fn(), + getEntriesByName: () => [ + { + transferSize: 1000, + requestStart: 0, + responseStart: 10, + responseEnd: 110, + connectStart: 0, + connectEnd: 0, + secureConnectionStart: 0, + nextHopProtocol: 'h2' + } + ] + }); + vi.spyOn(Math, 'random').mockReturnValue(0.5); + + const engine = new SpeedTest({ + autoStart: false, + measurementTargets: [ + 'https://speed-0.example', + 'https://speed-1.example', + 'https://speed-1.example' + ], + measurements: [{ type: 'download', bytes: 1000, count: 3 }], + measureDownloadLoadedLatency: false, + logAimApiUrl: null + }); + const finished = new Promise((resolve, reject) => { + engine.onFinish = resolve; + engine.onError = reject; + }); + + engine.play(); + await finished; + + expect( + fetchMock.mock.calls.map(([url]) => new URL(url.toString()).origin) + ).toEqual([ + 'https://speed-1.example', + 'https://speed-1.example', + 'https://speed-0.example' + ]); }); - it('applies global and step parallelism through the public API', async () => { + it('applies measurement targets and step parallelism through the public API', async () => { const resultsUrl = 'https://results.example/__results'; const fetchMock = vi.fn((url: RequestInfo | URL, _init?: RequestInit) => Promise.resolve( @@ -175,14 +300,17 @@ describe('parallel bandwidth aggregation', () => { ]; } }); + vi.spyOn(Math, 'random').mockReturnValue(0); const engine = new SpeedTest({ autoStart: false, - bandwidthOrigins: ['https://speed-0.example', 'https://speed-1.example'], - parallelism: 4, + measurementTargets: [ + 'https://speed-0.example', + 'https://speed-1.example' + ], measurements: [ - { type: 'download', bytes: 1000, count: 2, parallelism: 2 }, - { type: 'upload', bytes: 1000, count: 4 } + { type: 'download', bytes: 1000, count: 2, parallel: true }, + { type: 'upload', bytes: 1000, count: 4, parallel: true } ], measureDownloadLoadedLatency: false, measureUploadLoadedLatency: false, @@ -220,14 +348,14 @@ describe('parallel bandwidth aggregation', () => { ([url]) => url.toString() === resultsUrl ); const body = JSON.parse(resultsCall?.[1]?.body as string); - expect(body.sessionId).toBe('session=abc¶llel=4'); + expect(body.sessionId).toBe('session=abc¶llel=2'); expect(body.download).toEqual([expect.objectContaining({ bytes: 2000 })]); expect(body.upload).toEqual([expect.objectContaining({ bytes: 4000 })]); }); }); describe('parallel session metadata', () => { - it('appends the maximum parallelism', () => { + it('appends the target count', () => { expect(appendParallelism('session=abc&tier=test', 4)).toBe( 'session=abc&tier=test¶llel=4' ); @@ -239,8 +367,10 @@ describe('parallel session metadata', () => { ); }); - it('leaves sequential and absent sessions unchanged', () => { - expect(appendParallelism('session=abc', 1)).toBe('session=abc'); + it('removes metadata for sequential sessions', () => { + expect(appendParallelism('session=abc¶llel=2', undefined)).toBe( + 'session=abc' + ); expect(appendParallelism(undefined, 4)).toBeUndefined(); }); }); From 200ec8d307d4159e48230ac40508ce33be80f563 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Jesus?= Date: Thu, 3 Sep 2026 00:34:58 +0100 Subject: [PATCH 3/3] fix: correct parallel bandwidth aggregation --- .../BandwidthEngine/BandwidthEngine.ts | 78 ++++++++++++++++-- src/index.ts | 27 ++++++- tests/unit/engines/parallelism.test.ts | 80 +++++++++++++++---- 3 files changed, 162 insertions(+), 23 deletions(-) diff --git a/src/engines/BandwidthEngine/BandwidthEngine.ts b/src/engines/BandwidthEngine/BandwidthEngine.ts index 21954a9a..b168a76b 100644 --- a/src/engines/BandwidthEngine/BandwidthEngine.ts +++ b/src/engines/BandwidthEngine/BandwidthEngine.ts @@ -77,10 +77,38 @@ const calcUploadSpeed = ( return !secs ? undefined : bits / secs; }; +interface TimeInterval { + start: number; + end: number; +} + +const getCoveredDuration = ( + intervals: TimeInterval[], + rangeStart: number, + rangeEnd: number +): number => { + const clipped = intervals + .map(({ start, end }) => ({ + start: Math.max(start, rangeStart), + end: Math.min(end, rangeEnd) + })) + .filter(({ start, end }) => end > start) + .sort((a, b) => a.start - b.start); + let covered = 0; + let currentEnd = rangeStart; + clipped.forEach(({ start, end }) => { + if (end <= currentEnd) return; + covered += end - Math.max(start, currentEnd); + currentEnd = end; + }); + return covered; +}; + export const aggregateRequestTimings = ( timings: RequestTiming[], isDown: boolean, - numBytes: number + numBytes: number, + pausedIntervals: TimeInterval[] = [] ): BandwidthMeasurementTiming => { if (timings.length === 1) return timings[0]; @@ -89,9 +117,30 @@ export const aggregateRequestTimings = ( ...timings.map(timing => timing.responseStart) ); const responseEnd = Math.max(...timings.map(timing => timing.responseEnd)); - const duration = isDown - ? responseEnd - requestStart - : Math.max(...timings.map(timing => timing.responseStart)) - requestStart; + const intervalEnd = isDown + ? responseEnd + : Math.max(...timings.map(timing => timing.responseStart)); + const serverIntervals = isDown + ? timings.map(timing => { + const rawDuration = timing.responseEnd - timing.requestStart; + const adjustment = Math.min( + Math.max(0, rawDuration - timing.duration), + timing.responseStart - timing.requestStart + ); + return { + start: timing.responseStart - adjustment, + end: timing.responseStart + }; + }) + : []; + const duration = + intervalEnd - + requestStart - + getCoveredDuration( + [...serverIntervals, ...pausedIntervals], + requestStart, + intervalEnd + ); const transferSize = timings.reduce( (total, timing) => total + timing.transferSize, 0 @@ -319,12 +368,22 @@ class BandwidthMeasurementEngine implements Engine { // Public methods pause(): void { + if (this.#parallel && this.#running && this.#pauseStartedAt === undefined) { + this.#pauseStartedAt = performance.now(); + } this.#cancelCurrentMeasurement(`pause()`); this.#setRunning(false); } play(): void { if (!this.#running) { + if (this.#pauseStartedAt !== undefined) { + this.#pausedIntervals.push({ + start: this.#pauseStartedAt, + end: performance.now() + }); + this.#pauseStartedAt = undefined; + } this.#setRunning(true); this.#nextMeasurement(); } @@ -345,6 +404,8 @@ class BandwidthMeasurementEngine implements Engine { #counter: number = 0; #requestId: number = 0; #parallelTimings: RequestTiming[] = []; + #pausedIntervals: TimeInterval[] = []; + #pauseStartedAt: number | undefined; #minDuration: number = -Infinity; // of current measurement #throttleMs: number = 0; #estimatedServerTime: number = 0; @@ -445,6 +506,8 @@ class BandwidthMeasurementEngine implements Engine { this.#counter = 0; this.#minDuration = -Infinity; this.#parallelTimings = []; + this.#pausedIntervals = []; + this.#pauseStartedAt = undefined; performance.clearResourceTimings(); do { @@ -487,7 +550,12 @@ class BandwidthMeasurementEngine implements Engine { abortController ); if (abortController.signal.aborted) return; - timing = aggregateRequestTimings(timings, isDown, numBytes); + timing = aggregateRequestTimings( + timings, + isDown, + numBytes, + this.#pausedIntervals + ); this.#counter = meas.count; this.#minDuration = Math.min(...timings.map(timing => timing.duration)); } else { diff --git a/src/index.ts b/src/index.ts index 481d6654..9be26cc1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -326,6 +326,17 @@ class MeasurementEngine { return url; }; + #parallelMeasurementApiUrls( + type: 'download' | 'upload', + count: number + ): string[] | undefined { + const urls = this.#measurementApiUrls(type); + if (!urls) return undefined; + const startIndex = this.#targetIndex % urls.length; + this.#targetIndex += count; + return [...urls.slice(startIndex), ...urls.slice(0, startIndex)]; + } + #nextLoadedLatencyApiUrl = (): string => { const urls = this.#measurementApiUrls('download'); if (!urls) return this.#config.downloadApiUrl; @@ -656,8 +667,20 @@ class MeasurementEngine { { downloadApiUrl, uploadApiUrl, - downloadApiUrls: this.#measurementApiUrls('download'), - uploadApiUrls: this.#measurementApiUrls('upload'), + downloadApiUrls: + msmConfig.parallel === true && type === 'download' + ? this.#parallelMeasurementApiUrls( + 'download', + msmConfig.count ?? 1 + ) + : undefined, + uploadApiUrls: + msmConfig.parallel === true && type === 'upload' + ? this.#parallelMeasurementApiUrls( + 'upload', + msmConfig.count ?? 1 + ) + : undefined, getDownloadApiUrl: () => this.#nextMeasurementApiUrl('download'), getUploadApiUrl: () => this.#nextMeasurementApiUrl('upload'), getLoadedLatencyApiUrl: this.#nextLoadedLatencyApiUrl, diff --git a/tests/unit/engines/parallelism.test.ts b/tests/unit/engines/parallelism.test.ts index df244ba0..d294bc5f 100644 --- a/tests/unit/engines/parallelism.test.ts +++ b/tests/unit/engines/parallelism.test.ts @@ -43,6 +43,36 @@ describe('parallel bandwidth aggregation', () => { expect(result.bps).toBeCloseTo(16000 / 0.12); }); + it('subtracts overlapping server and delta adjustments', () => { + const first = { + ...timing(0, 20, 120), + serverTime: 8, + duration: 110 + }; + const second = { + ...timing(0, 25, 125), + serverTime: 8, + duration: 115 + }; + + const result = aggregateRequestTimings([first, second], true, 1000); + + expect(result.duration).toBe(110); + expect(result.bps).toBeCloseTo(16000 / 0.11); + }); + + it('subtracts paused intervals from the aggregate duration', () => { + const result = aggregateRequestTimings( + [timing(0, 10, 100), timing(300, 310, 400)], + true, + 1000, + [{ start: 100, end: 300 }] + ); + + expect(result.duration).toBe(200); + expect(result.bps).toBeCloseTo(16000 / 0.2); + }); + it('estimates bytes missing from resource timing', () => { const hiddenTiming = { ...timing(5, 20, 120), transferSize: 0 }; const result = aggregateRequestTimings( @@ -152,6 +182,7 @@ describe('parallel bandwidth aggregation', () => { }); it('retains completed requests when a parallel step resumes', async () => { + let clock = 0; const releases: Array<() => void> = []; const fetchMock = vi.fn( (_url: RequestInfo | URL, init?: RequestInit) => @@ -167,19 +198,26 @@ describe('parallel bandwidth aggregation', () => { vi.stubGlobal('fetch', fetchMock); vi.stubGlobal('window', { location: { origin: 'https://app.example' } }); vi.stubGlobal('performance', { + now: () => clock, clearResourceTimings: vi.fn(), - getEntriesByName: () => [ - { - transferSize: 1000, - requestStart: 0, - responseStart: 10, - responseEnd: 110, - connectStart: 0, - connectEnd: 0, - secureConnectionStart: 0, - nextHopProtocol: 'h2' - } - ] + getEntriesByName: (url: string) => { + const index = Number( + new URL(url).searchParams.get('__cf_speedtest_request')?.split('-')[1] + ); + const requestStart = index === 0 ? 0 : index < 4 ? 300 : 400; + return [ + { + transferSize: 1000, + requestStart, + responseStart: requestStart + 10, + responseEnd: requestStart + 100, + connectStart: 0, + connectEnd: 0, + secureConnectionStart: 0, + nextHopProtocol: 'h2' + } + ]; + } }); const engine = new BandwidthEngine( @@ -195,7 +233,10 @@ describe('parallel bandwidth aggregation', () => { ); const onRequestResult = vi.fn(); engine.onRequestResult = onRequestResult; - onRequestResult.mockImplementationOnce(() => engine.pause()); + onRequestResult.mockImplementationOnce(() => { + clock = 100; + engine.pause(); + }); const finished = new Promise(resolve => { engine.onFinished = resolve; }); @@ -204,6 +245,7 @@ describe('parallel bandwidth aggregation', () => { await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2)); releases[0](); await vi.waitFor(() => expect(onRequestResult).toHaveBeenCalledOnce()); + clock = 300; engine.play(); await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(4)); releases[2](); @@ -214,9 +256,10 @@ describe('parallel bandwidth aggregation', () => { expect(onRequestResult).toHaveBeenCalledTimes(4); expect(engine.results.down[1000].timings[0].transferredBytes).toBe(4000); + expect(engine.results.down[1000].timings[0].duration).toBe(300); }); - it('rotates sequential requests from a randomized target', async () => { + it('rotates sequential and parallel requests from a randomized target', async () => { const fetchMock = vi.fn((url: RequestInfo | URL) => Promise.resolve(new Response(url.toString())) ); @@ -248,7 +291,10 @@ describe('parallel bandwidth aggregation', () => { 'https://speed-1.example', 'https://speed-1.example' ], - measurements: [{ type: 'download', bytes: 1000, count: 3 }], + measurements: [ + { type: 'download', bytes: 1000, count: 3 }, + { type: 'download', bytes: 2000, count: 2, parallel: true } + ], measureDownloadLoadedLatency: false, logAimApiUrl: null }); @@ -265,7 +311,9 @@ describe('parallel bandwidth aggregation', () => { ).toEqual([ 'https://speed-1.example', 'https://speed-1.example', - 'https://speed-0.example' + 'https://speed-0.example', + 'https://speed-1.example', + 'https://speed-1.example' ]); });