diff --git a/README.md b/README.md index 90d063a..0b0592b 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 dc87fe1..2dc73c9 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 4105efc..8936f60 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 a3353df..d21494c 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 a8127cf..06c1575 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 d38ee3a..b1e8e2c 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 3660ee9..174acce 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 0000000..5152332 --- /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 c727d35..a4c4239 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 d94c68c..2e829b4 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 0000000..74affbc --- /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(); + }); +});