Add jitter() setting to DatumLoader to add random delays on parallel requests, and help avoid rate throttling errors.

msqr · msqr · commit 5a1c3d90bcb8 · 2025-04-24T08:14:22.000+12:00
diff --git a/src/main/net/jsonClientSupport.ts b/src/main/net/jsonClientSupport.ts
@@ -55,14 +55,16 @@ export default abstract class JsonClientSupport<API extends UrlHelper, T> {
 	 *
 	 * @param url the URL to request.
 	 * @param signUrl the URL to sign (might be different to `url` if a proxy is used)
+	 * @param delay an optional number of milliseconds to sleep before initiating the request
 	 * @returns a function that accepts a callback argument
 	 */
 	protected requestor<V>(
 		url: string,
-		signUrl?: string
+		signUrl?: string,
+		delay?: number
 	): (cb: LoaderDataCallbackFn<V>) => void {
 		const auth = this.authBuilder;
-		return (cb: LoaderDataCallbackFn<V>) => {
+		const fn = (cb: LoaderDataCallbackFn<V>) => {
 			const headers: any = {
 				Accept: "application/json",
 			};
@@ -101,5 +103,13 @@ export default abstract class JsonClientSupport<API extends UrlHelper, T> {
 				}, errorHandler);
 			}, errorHandler);
 		};
+		if (delay && delay > 0) {
+			return (cb: LoaderDataCallbackFn<V>) => {
+				setTimeout(() => {
+					fn.call(this, cb);
+				}, delay);
+			};
+		}
+		return fn;
 	}
 }
diff --git a/src/main/tool/datumLoader.ts b/src/main/tool/datumLoader.ts
@@ -21,6 +21,7 @@ interface QueryResultsData {
 }
 
 const DEFAULT_PAGE_SIZE: number = 1000;
+const DEFAULT_JITTER: number = 150;
 
 /**
  * An enumeration of loader state values.
@@ -88,6 +89,12 @@ export default class DatumLoader
 	 */
 	#concurrency: number;
 
+	/**
+	 * When > 0 then add a random amount of milliseconds up to this amount before initiating
+	 * parallel requests (thus #concurrency must also be configured).
+	 */
+	#jitter: number;
+
 	/**
 	 * A queue to use for parallel mode, when `concurrency` configured > 0.
 	 */
@@ -123,6 +130,7 @@ export default class DatumLoader
 		this.#readingsMode = false;
 		this.#proxyUrl = null;
 		this.#concurrency = 0;
+		this.#jitter = DEFAULT_JITTER;
 		this.#state = DatumLoaderState.Ready;
 	}
 
@@ -152,12 +160,41 @@ export default class DatumLoader
 		if (value === undefined) {
 			return this.#concurrency;
 		}
-		if (!isNaN(value) && Number(value) > 0) {
+		if (!isNaN(value) && Number(value) >= 0) {
 			this.#concurrency = Number(value);
 		}
 		return this;
 	}
 
+	/**
+	 * Get the concurrency jitter value to use for parallel requests.
+	 *
+	 * @returns the current concurrency jitter value (milliseconds); defaults to `150`
+	 */
+	jitter(): number;
+
+	/**
+	 * Set the concurrency jitter amount to use for parallel requests.
+	 *
+	 * When parallel mode is enabled by setting `concurrency()` to a positive value, a random amount
+	 * of "pause" time can be added before parallel requests are made by configuring this
+	 * to a positive value. This can be helpful to avoid API rate limiting errors.
+	 *
+	 * @param value the concurrency jitter amount to use, in milliseconds, or `0` to disable
+	 * @returns this object
+	 */
+	jitter(value: number): this;
+
+	jitter(value?: number): number | this {
+		if (value === undefined) {
+			return this.#jitter;
+		}
+		if (!isNaN(value) && Number(value) >= 0) {
+			this.#jitter = Number(value);
+		}
+		return this;
+	}
+
 	/**
 	 * Get the optional callback function.
 	 *
@@ -485,7 +522,13 @@ export default class DatumLoader
 			? url.replace(/^[^:]+:\/\/[^/]+/, this.#proxyUrl)
 			: url;
 
-		const query = this.requestor<QueryResultsData>(reqUrl, url);
+		// add delay for parallel mode if jitter configured
+		const delay =
+			q && this.#jitter > 0
+				? Math.floor(Math.random() * this.#jitter) + 1
+				: 0;
+
+		const query = this.requestor<QueryResultsData>(reqUrl, url, delay);
 
 		const handler = (error?: Error, data?: QueryResultsData) => {
 			if (error) {
diff --git a/src/test/net/jsonClientSupportTests.ts b/src/test/net/jsonClientSupportTests.ts
@@ -15,10 +15,12 @@ log.level = LogLevel.DEBUG;
 
 class TestClient extends JsonClientSupport<SolarQueryApi, any> {
 	#url: string;
+	#jitter?: number;
 
-	constructor(api: SolarQueryApi, url: string) {
+	constructor(api: SolarQueryApi, url: string, jitter?: number) {
 		super(api);
 		this.#url = url;
+		this.#jitter = jitter;
 	}
 
 	/**
@@ -28,7 +30,11 @@ class TestClient extends JsonClientSupport<SolarQueryApi, any> {
 	 */
 	fetch(): Promise<any> {
 		return new Promise((resolve, reject) => {
-			this.requestor(this.#url)((error, results) => {
+			this.requestor(
+				this.#url,
+				undefined,
+				this.#jitter
+			)((error, results) => {
 				if (error) {
 					reject(error);
 				} else {
@@ -71,6 +77,31 @@ test.serial("fetch:http404", async (t) => {
 	);
 });
 
+test.serial("fetch:jitter", async (t) => {
+	// GIVEN
+	const url = "http://localhost/foo";
+	const http = t.context.agent.get("http://localhost");
+	http.intercept({
+		path: "/foo",
+		method: "GET",
+	}).reply(200, {
+		success: true,
+		data: "hi",
+	});
+
+	// WHEN
+	const client = new TestClient(t.context.api, url, 1000);
+	const start = new Date();
+	const result = await client.fetch();
+	const end = new Date();
+
+	t.is(result, "hi", "Requested URL with delay returned data result.");
+	t.true(
+		end.getTime() - start.getTime() >= 1000,
+		"At least delay ms has elapsed."
+	);
+});
+
 test.serial("fetch:result:error:message", async (t) => {
 	// GIVEN
 	const url = "http://localhost/foo";
diff --git a/src/test/tool/datumLoaderTests.ts b/src/test/tool/datumLoaderTests.ts
@@ -1008,6 +1008,20 @@ test.serial("load:parallel:multiPage:gapDataFirstPage", async (t) => {
 	t.deepEqual(results, allResults.slice(0, 4));
 });
 
+test.serial("jitter", (t) => {
+	const filter = testFilter();
+	const loader = new DatumLoader(t.context.api, filter);
+	t.truthy(loader);
+	t.is(loader.jitter("Ten" as any).jitter(), 150, "NaN ignored");
+	t.is(loader.jitter(-1).jitter(), 150, "Negative ignored");
+	t.is(
+		loader.jitter("500" as any).jitter(),
+		500,
+		"Number string parsed as number"
+	);
+	t.is(loader.jitter(1).jitter(), 1, "Positive set");
+});
+
 test.serial("load:proxy:multiPage:parallel", (t) => {
 	const http = t.context.agent.get("https://query-proxy");
 	const allResults = [
@@ -1092,6 +1106,111 @@ test.serial("load:proxy:multiPage:parallel", (t) => {
 		.proxyUrl(proxyUrl)
 		.paginationSize(2)
 		.concurrency(Infinity)
+		.jitter(500)
+		.includeTotalResultsCount(true);
+	t.truthy(loader);
+	t.is(loader.jitter(), 500);
+	t.is(loader.proxyUrl(), proxyUrl);
+	t.is(loader.includeTotalResultsCount(), true);
+
+	return new Promise((resolve, reject) => {
+		loader.load((error, results) => {
+			try {
+				t.falsy(error, "No error generated.");
+				t.deepEqual(results, allResults, "All pages returned.");
+				resolve();
+			} catch (error) {
+				reject(error);
+			}
+		});
+	});
+});
+
+test.serial("load:proxy:multiPage:parallel:noJitter", (t) => {
+	const http = t.context.agent.get("https://query-proxy");
+	const allResults = [
+		{
+			created: "2017-07-04 12:00:00.000Z",
+			nodeId: 123,
+			sourceId: "test-source",
+			val: 0,
+		},
+		{
+			created: "2017-07-04 13:00:00.000Z",
+			nodeId: 123,
+			sourceId: "test-source",
+			val: 1,
+		},
+		{
+			created: "2017-07-04 14:00:00.000Z",
+			nodeId: 123,
+			sourceId: "test-source",
+			val: 0,
+		},
+		{
+			created: "2017-07-04 15:00:00.000Z",
+			nodeId: 123,
+			sourceId: "test-source",
+			val: 1,
+		},
+		{
+			created: "2017-07-04 16:00:00.000Z",
+			nodeId: 123,
+			sourceId: "test-source",
+			val: 0,
+		},
+		{
+			created: "2017-07-04 17:00:00.000Z",
+			nodeId: 123,
+			sourceId: "test-source",
+			val: 1,
+		},
+	];
+	// expect 3 page queries: one for first page and total result count, 2 more for remaining pages
+	http.intercept({
+		path: "/path/solarquery/api/v1/pub/datum/list?nodeId=123&sourceId=test-source&startDate=2017-04-01T12%3A00&endDate=2017-05-01T12%3A00&aggregation=Hour&withoutTotalResultsCount=false&max=2",
+		method: "GET",
+	}).reply(200, {
+		success: true,
+		data: {
+			totalResults: 6,
+			startingOffset: 0,
+			returnedResultCount: 2,
+			results: allResults.slice(0, 2),
+		},
+	});
+	http.intercept({
+		path: "/path/solarquery/api/v1/pub/datum/list?nodeId=123&sourceId=test-source&startDate=2017-04-01T12%3A00&endDate=2017-05-01T12%3A00&aggregation=Hour&withoutTotalResultsCount=true&max=2&offset=2",
+		method: "GET",
+	}).reply(200, {
+		success: true,
+		data: {
+			totalResults: null,
+			startingOffset: 2,
+			returnedResultCount: 2,
+			results: allResults.slice(2, 4),
+		},
+	});
+	http.intercept({
+		path: "/path/solarquery/api/v1/pub/datum/list?nodeId=123&sourceId=test-source&startDate=2017-04-01T12%3A00&endDate=2017-05-01T12%3A00&aggregation=Hour&withoutTotalResultsCount=true&max=2&offset=4",
+		method: "GET",
+	}).reply(200, {
+		success: true,
+		data: {
+			totalResults: null,
+			startingOffset: 4,
+			returnedResultCount: 2,
+			results: allResults.slice(4, 6),
+		},
+	});
+
+	const proxyUrl = "https://query-proxy/path";
+	const filter = testFilter();
+	const loader = new DatumLoader(t.context.api, filter)
+		.proxyUrl(proxyUrl)
+		.paginationSize(2)
+		.concurrency(Infinity)
+		.jitter(0)
 		.includeTotalResultsCount(true);
 	t.truthy(loader);
 	t.is(loader.proxyUrl(), proxyUrl);
