Move report throttle out of reporting logic

Change-type: patch
Signed-off-by: 20k-ultra <3946250+20k-ultra@users.noreply.github.com>

package-lock.json

@@ -773,12 +773,12 @@
       "dev": true
     },
     "@types/sinon": {
-      "version": "10.0.0",
+      "version": "10.0.11",
-      "resolved": "https://registry.npmjs.org/@types/sinon/-/sinon-10.0.0.tgz",
+      "resolved": "https://registry.npmjs.org/@types/sinon/-/sinon-10.0.11.tgz",
-      "integrity": "sha512-jDZ55oCKxqlDmoTBBbBBEx+N8ZraUVhggMZ9T5t+6/Dh8/4NiOjSUfpLrPiEwxQDlAe3wpAkoXhWvE6LibtsMQ==",
+      "integrity": "sha512-dmZsHlBsKUtBpHriNjlK0ndlvEh8dcb9uV9Afsbt89QIyydpC7NcR+nWlAhASfy3GHnxTl4FX/aKE7XZUt/B4g==",
       "dev": true,
       "requires": {
-        "@sinonjs/fake-timers": "^7.0.4"
+        "@types/sinonjs__fake-timers": "*"
       }
     },
     "@types/sinon-chai": {
@@ -791,6 +791,12 @@
         "@types/sinon": "*"
       }
     },
+    "@types/sinonjs__fake-timers": {
+      "version": "8.1.1",
+      "resolved": "https://registry.npmjs.org/@types/sinonjs__fake-timers/-/sinonjs__fake-timers-8.1.1.tgz",
+      "integrity": "sha512-0kSuKjAS0TrGLJ0M/+8MaFkGsQhZpB6pxOmvS3K8FYI72K//YmdfoW9X2qPsAKh1mkwxGD5zib9s1FIFed6E8g==",
+      "dev": true
+    },
     "@types/source-list-map": {
       "version": "0.1.2",
       "resolved": "https://registry.npmjs.org/@types/source-list-map/-/source-list-map-0.1.2.tgz",

package.json

@@ -67,7 +67,7 @@
     "@types/rwlock": "^5.0.2",
     "@types/semver": "^7.3.3",
     "@types/shell-quote": "^1.7.0",
-    "@types/sinon": "^10.0.0",
+    "@types/sinon": "^10.0.11",
     "@types/sinon-chai": "^3.2.5",
     "@types/supertest": "^2.0.11",
     "@types/terser-webpack-plugin": "^3.0.0",

src/device-state/current-state.ts

@@ -3,6 +3,7 @@ import * as url from 'url';
 import { CoreOptions } from 'request';
 
 import * as constants from '../lib/constants';
+import { withBackoff, OnFailureInfo } from '../lib/backoff';
 import { log } from '../lib/supervisor-console';
 import { InternalInconsistencyError, StatusError } from '../lib/errors';
 import { getRequestInstance } from '../lib/request';
@@ -14,9 +15,6 @@ import { SchemaTypeKey, SchemaReturn } from '../config/schema-type';
 import * as eventTracker from '../event-tracker';
 import * as deviceState from '../device-state';
 
-// The exponential backoff starts at 15s
-const MINIMUM_BACKOFF_DELAY = 15000;
-
 const INTERNAL_STATE_KEYS = [
 	'update_pending',
 	'update_downloaded',
@@ -28,10 +26,6 @@ const lastReportedState: DeviceStatus = {
 	local: {},
 	dependent: {},
 };
-const stateForReport: DeviceStatus = {
-	local: {},
-	dependent: {},
-};
 let reportPending = false;
 
 type CurrentStateReportConf = {
@@ -43,6 +37,7 @@ type CurrentStateReportConf = {
 		| 'deviceApiKey'
 		| 'deviceId'
 		| 'localMode'
+		| 'appUpdatePollInterval'
 		| 'hardwareMetrics'
 	>]: SchemaReturn<key>;
 };
@@ -64,7 +59,7 @@ const stripDeviceStateInLocalMode = (state: DeviceStatus): DeviceStatus => {
 	};
 };
 
-const report = async ({ stateDiff, conf }: StateReport) => {
+async function report({ stateDiff, conf }: StateReport): Promise<boolean> {
 	let body = stateDiff;
 	const { apiEndpoint, apiTimeout, deviceApiKey, localMode, uuid } = conf;
 	if (localMode) {
@@ -73,7 +68,7 @@ const report = async ({ stateDiff, conf }: StateReport) => {
 
 	if (_.isEmpty(body.local)) {
 		// Nothing to send.
-		return;
+		return false;
 	}
 
 	if (conf.uuid == null || conf.apiEndpoint == null) {
@@ -104,9 +99,11 @@ const report = async ({ stateDiff, conf }: StateReport) => {
 			headers['retry-after'] ? parseInt(headers['retry-after'], 10) : undefined,
 		);
 	}
-};
+	// State was reported
+	return true;
+}
 
-const getStateDiff = (): DeviceStatus => {
+function newStateDiff(stateForReport: DeviceStatus): DeviceStatus {
 	const lastReportedLocal = lastReportedState.local;
 	const lastReportedDependent = lastReportedState.dependent;
 	if (lastReportedLocal == null || lastReportedDependent == null) {
@@ -138,128 +135,110 @@ const getStateDiff = (): DeviceStatus => {
 	};
 
 	return _.omitBy(diff, _.isEmpty);
-};
-
-const throttledReport = _.throttle(
-	// We define the throttled function this way to avoid UncaughtPromise exceptions
-	// for exceptions thrown from the report function
-	(opts: StateReport, resolve: () => void, reject: (e: Error) => void) =>
-		report(opts).then(resolve).catch(reject),
-	constants.maxReportFrequency,
-);
-
-/**
- * Perform exponential backoff on the function, increasing the attempts
- * counter on each call
- *
- * If attempts is 0 then, it will delay for min(minDelay, maxDelay)
- */
-const backoff = (
-	retry: (attempts: number) => void,
-	attempts = 0,
-	maxDelay: number,
-	minDelay = MINIMUM_BACKOFF_DELAY,
-) => {
-	const delay = Math.min(2 ** attempts * minDelay, maxDelay);
-	log.info(`Retrying request in ${delay / 1000} seconds`);
-	setTimeout(() => retry(attempts + 1), delay);
-};
-
-function reportCurrentState(attempts = 0) {
-	(async () => {
-		const {
-			hardwareMetrics,
-			appUpdatePollInterval: maxDelay,
-			...conf
-		} = await config.getMany([
-			'deviceApiKey',
-			'apiTimeout',
-			'apiEndpoint',
-			'uuid',
-			'localMode',
-			'appUpdatePollInterval',
-			'hardwareMetrics',
-		]);
-
-		reportPending = true;
-		const currentDeviceState = await deviceState.getStatus();
-		// If hardwareMetrics is false, send null patch for system metrics to cloud API
-		const info = {
-			...(hardwareMetrics
-				? await sysInfo.getSystemMetrics()
-				: {
-						cpu_usage: null,
-						memory_usage: null,
-						memory_total: null,
-						storage_usage: null,
-						storage_total: null,
-						storage_block_device: null,
-						cpu_temp: null,
-						cpu_id: null,
-				  }),
-			...(await sysInfo.getSystemChecks()),
-		};
-
-		stateForReport.local = {
-			...stateForReport.local,
-			...currentDeviceState.local,
-			...info,
-		};
-		stateForReport.dependent = {
-			...stateForReport.dependent,
-			...currentDeviceState.dependent,
-		};
-
-		const stateDiff = getStateDiff();
-
-		// report state diff
-		throttledReport(
-			{ stateDiff, conf },
-			() => {
-				// If the patch succeeds update lastReport and reset the error counter
-				_.assign(lastReportedState.local, stateDiff.local);
-				_.assign(lastReportedState.dependent, stateDiff.dependent);
-				stateReportErrors = 0;
-
-				reportPending = false;
-			},
-			(e) => {
-				if (e instanceof StatusError) {
-					// We don't want these errors to be classed as a report error, as this will cause
-					// the watchdog to kill the supervisor - and killing the supervisor will
-					// not help in this situation
-					log.error(
-						`Device state report failure! Status code: ${e.statusCode} - message:`,
-						e.message,
-					);
-
-					// We want to backoff on all errors, but without having the healthchecks
-					// get triggered.
-					// This will use retryAfter as maxDelay if the header is present in the
-					// response by the API
-					backoff(
-						reportCurrentState,
-						// Do not do exponential backoff if the API reported a retryAfter
-						e.retryAfter ? 0 : attempts,
-						maxDelay,
-						// Start the polling at the value given by the API if any
-						e.retryAfter ?? MINIMUM_BACKOFF_DELAY,
-					);
-				} else {
-					eventTracker.track('Device state report failure', {
-						error: e.message,
-					});
-					backoff(reportCurrentState, stateReportErrors++, maxDelay);
-				}
-			},
-		);
-	})();
 }
 
-export const startReporting = () => {
+async function reportCurrentState(conf: CurrentStateReportConf) {
-	const doReport = () => {
+	// Ensure no other report starts
+	reportPending = true;
+	// Wrap the report with fetching of state so report always has the latest state diff
+	const getStateAndReport = async () => {
+		// Get state to report
+		const stateToReport = await generateStateForReport();
+		// Get diff from last reported state
+		const stateDiff = newStateDiff(stateToReport);
+		// Report diff
+		if (await report({ stateDiff, conf })) {
+			// Update lastReportedState
+			_.assign(lastReportedState.local, stateDiff.local);
+			_.assign(lastReportedState.dependent, stateDiff.dependent);
+			// Log that we successfully reported the current state
+			log.info('Reported current state to the cloud');
+		}
+	};
+	// Create a report that will backoff on errors
+	const reportWithBackoff = withBackoff(getStateAndReport, {
+		maxDelay: conf.appUpdatePollInterval,
+		minDelay: 15000,
+		onFailure: handleRetry,
+	});
+	// Run in try block to avoid throwing any exceptions
+	try {
+		await reportWithBackoff();
+		stateReportErrors = 0;
+	} catch (e) {
+		log.error(e);
+	}
+	reportPending = false;
+}
+
+function handleRetry(retryInfo: OnFailureInfo) {
+	if (retryInfo.error instanceof StatusError) {
+		log.error(
+			`Device state report failure! Status code: ${retryInfo.error.statusCode} - message:`,
+			retryInfo.error?.message ?? retryInfo.error,
+		);
+	} else {
+		eventTracker.track('Device state report failure', {
+			error: retryInfo.error?.message ?? retryInfo.error,
+		});
+	}
+	stateReportErrors++;
+	log.info(
+		`Retrying current state report in ${retryInfo.delay / 1000} seconds`,
+	);
+}
+
+async function generateStateForReport() {
+	const { hardwareMetrics } = await config.getMany(['hardwareMetrics']);
+
+	const currentDeviceState = await deviceState.getStatus();
+
+	// If hardwareMetrics is false, send null patch for system metrics to cloud API
+	const info = {
+		...(hardwareMetrics
+			? await sysInfo.getSystemMetrics()
+			: {
+					cpu_usage: null,
+					memory_usage: null,
+					memory_total: null,
+					storage_usage: null,
+					storage_total: null,
+					storage_block_device: null,
+					cpu_temp: null,
+					cpu_id: null,
+			  }),
+		...(await sysInfo.getSystemChecks()),
+	};
+
+	return {
+		local: {
+			...currentDeviceState.local,
+			...info,
+		},
+		dependent: currentDeviceState.dependent,
+	};
+}
+
+export async function startReporting() {
+	// Get configs needed to make a report
+	const reportConfigs = (await config.getMany([
+		'uuid',
+		'apiEndpoint',
+		'apiTimeout',
+		'deviceApiKey',
+		'deviceId',
+		'localMode',
+		'appUpdatePollInterval',
+		'hardwareMetrics',
+	])) as CurrentStateReportConf;
+	// Throttle reportCurrentState so we don't query device or hit API excessively
+	const throttledReport = _.throttle(
+		reportCurrentState,
+		constants.maxReportFrequency,
+	);
+	const doReport = async () => {
 		if (!reportPending) {
-			reportCurrentState();
+			throttledReport(reportConfigs);
 		}
 	};
 
@@ -270,4 +249,4 @@ export const startReporting = () => {
 	setInterval(doReport, constants.maxReportFrequency);
 	// Try to perform a report right away
 	return doReport();
-};
+}

src/lib/backoff.ts

@@ -0,0 +1,105 @@
+import { strict as assert } from 'assert';
+import { promisify } from 'util';
+
+export type OnFailureInfo = {
+	failures: number;
+	delay: number;
+	error: any;
+};
+
+export type Options = {
+	retryCount: number;
+	maxDelay: number;
+	minDelay: number;
+	maxRetries: number;
+	onFailure?: (info: OnFailureInfo) => void;
+};
+
+const DEFAULT_OPTIONS: Partial<Options> = {
+	retryCount: 0,
+	maxDelay: 15000,
+	minDelay: 15000,
+	maxRetries: Infinity,
+};
+
+/**
+ * Retries a function with exponential delay between calls.
+ * If attempts is 0 then, it will delay for min(minDelay, maxDelay)
+ *
+ * Supports exceptions which return `retryAfter` value to specify the backoff duration.
+ * Will call `retryCallback` if passed in options as a way for callers to react to retries
+ */
+export function withBackoff<T extends (...args: any[]) => any>(
+	func: T,
+	options: Partial<Options> = {},
+) {
+	assert(typeof func === 'function', 'expected a function as parameter');
+	// If the function returns a promise, unwrap the promise return type
+	// otherwise use the actual return
+	type TReturn = ReturnType<T> extends Promise<infer R> ? R : ReturnType<T>;
+
+	// TODO use standard lib async setTimout (requires node 16)
+	const sleep = promisify(setTimeout);
+
+	const normalizedOptions: Options = {
+		...DEFAULT_OPTIONS,
+		...options,
+	} as Options;
+
+	return async function wrapped(...args: Parameters<T>): Promise<TReturn> {
+		try {
+			return await func(...args);
+		} catch (error) {
+			if (normalizedOptions.retryCount >= normalizedOptions.maxRetries) {
+				throw new Error(
+					`Reached max number of retries: ${normalizedOptions.maxRetries}`,
+				);
+			}
+
+			const wait = isRetryError(error)
+				? error.retryAfter
+				: exponentialRange(
+						normalizedOptions.retryCount,
+						normalizedOptions.minDelay,
+						normalizedOptions.maxDelay,
+				  );
+
+			normalizedOptions.retryCount++;
+
+			if (normalizedOptions.onFailure) {
+				normalizedOptions.onFailure({
+					failures: normalizedOptions.retryCount,
+					delay: wait,
+					error,
+				});
+			}
+
+			await sleep(wait);
+
+			return wrapped(...args);
+		}
+	};
+}
+
+// This is the expected interface for the error type
+export interface RetryError extends Error {
+	retryAfter: number;
+}
+
+// Type guard only checks that retryAfter is present
+const isRetryError = (x: unknown): x is RetryError =>
+	x != null &&
+	x instanceof Error &&
+	Number.isInteger((x as RetryError).retryAfter);
+
+/**
+ * Calculates a number with exponential growth given {retryCount} within
+ * a range of {minDelay} and {maxDelay} using {minDelay} as base.
+ */
+function exponentialRange(
+	retryCount: number,
+	minDelay: number,
+	maxDelay: number,
+): number {
+	return Math.min(2 ** retryCount * minDelay, maxDelay);
+}

test/src/lib/backoff.spec.ts

@@ -0,0 +1,216 @@
+import { assert, expect } from 'chai';
+import * as sinon from 'sinon';
+
+import { StatusError } from '../../../src/lib/errors';
+import { withBackoff, OnFailureInfo } from '../../../src/lib/backoff';
+
+const DEFAULT_OPTIONS = {
+	maxRetries: 5,
+	maxDelay: 900000, // 15 minutes
+	minDelay: 10000, // 10 seconds
+};
+
+describe('lib/backoff', async () => {
+	let clock: sinon.SinonFakeTimers;
+
+	beforeEach(() => {
+		clock = sinon.useFakeTimers();
+	});
+
+	afterEach(() => {
+		clock.restore();
+	});
+
+	it('resolves after 3 retries', async () => {
+		// Create a function that will fail 3 times so it succeeds on the 4th
+		const failer = new Failer(3);
+		// Wrap function withBackoff
+		const fnWithBackoff = withBackoff(async () => {
+			return failer.willResolve('fails 3 times then resolves on 4th');
+		}, DEFAULT_OPTIONS);
+		// Call function and allow clock to trigger all events
+		clock.runAllAsync();
+		await expect(fnWithBackoff()).to.eventually.equal(
+			'fails 3 times then resolves on 4th',
+		);
+		// Check that function was called 4 times (failed 3 times, succeeds on 4th)
+		expect(failer.callCount).to.equal(4);
+	});
+
+	it('should not call the function before minDelay', async () => {
+		const failer = new Failer(3);
+		const minDelay = Math.floor(Math.random() * 1000);
+		const myBackoffFunc = withBackoff(
+			async () => {
+				return failer.willResolve('ok');
+			},
+			{
+				minDelay,
+			},
+		);
+		// Function should have been called 0 times to start
+		expect(failer.callCount).to.equal(0);
+		// Call function
+		myBackoffFunc();
+		// Check that function was run at least once
+		expect(failer.callCount).to.equal(1);
+		// Elapse some time but not enough to be minDelay
+		await clock.tickAsync(minDelay - 1);
+		// Check that the function still has only been called 1 time
+		expect(failer.callCount).to.equal(1);
+		// Elapse exactly minDelay so function is called once more
+		await clock.tickAsync(minDelay + 1);
+		// Check that function was called twice
+		expect(failer.callCount).to.equal(2);
+	});
+
+	it('backs off with exponential delay', async () => {
+		const failer = new Failer(3);
+		const minDelay = Math.floor(Math.random() * 1000);
+		const myBackoffFunc = withBackoff(
+			async () => {
+				return failer.willResolve('ok');
+			},
+			{
+				minDelay,
+				maxDelay: 5000000,
+			},
+		);
+		expect(failer.callCount).to.equal(0);
+		// Call function
+		const p = myBackoffFunc();
+		// Function is called immediately
+		expect(failer.callCount).to.equal(1);
+		// First delay is equal to minDelay
+		await clock.tickAsync(minDelay);
+		// Should have been called again
+		expect(failer.callCount).to.equal(2);
+		// Tick exponential time
+		await clock.tickAsync(minDelay * 2);
+		// Should have been called again
+		expect(failer.callCount).to.equal(3);
+		// Tick exponential time
+		await clock.tickAsync(minDelay * 2 * 2);
+		// Function should be fulfilled by now
+		await expect(p).to.be.fulfilled;
+		expect(failer.callCount).to.equal(4);
+	});
+
+	it('never exceeds maxRetries', async () => {
+		// Make the failer fail 100 times (more then maxRetries)
+		const failer = new Failer(100);
+		const fnWithBackoff = withBackoff(async () => {
+			return failer.willResolve('ok');
+		}, DEFAULT_OPTIONS);
+		clock.runAllAsync();
+		// Call the function
+		await expect(fnWithBackoff()).to.eventually.be.rejectedWith(
+			`Reached max number of retries: ${DEFAULT_OPTIONS.maxRetries}`,
+		);
+	});
+
+	it('provides correct info within onFailure callback', async () => {
+		// Create a function that will fail 3 times so it succeeds on the 4th
+		const failer = new Failer(3);
+		const minDelay = Math.floor(Math.random() * 1000);
+		let counter = 0;
+		// Wrap function withBackoff
+		const fnWithBackoff = withBackoff(
+			async () => {
+				return failer.willResolve('ok');
+			},
+			{
+				minDelay,
+				onFailure: (data: OnFailureInfo) => {
+					counter++;
+					expect(data).to.deep.equal({
+						failures: counter,
+						delay:
+							counter === 1 ? minDelay : exponentialize(minDelay, counter - 1),
+						error: 'Not ready!',
+					});
+				},
+			},
+		);
+		// Call function and allow clock to trigger all events
+		clock.runAllAsync();
+		await fnWithBackoff();
+	});
+
+	it('uses RetryAfter from exception thrown', async () => {
+		const retryAfter = 50000;
+		const minDelay = 1000;
+		const failer = new Failer(
+			1,
+			new StatusError(503, 'Service Unavailable', retryAfter),
+		);
+		const fnWithBackoff = withBackoff(
+			async () => {
+				return failer.willResolve('ok');
+			},
+			{
+				minDelay,
+			},
+		);
+		assert(retryAfter > minDelay, 'retryAfter must be greater than minDelay');
+		expect(failer.callCount).to.equal(0);
+		// Start calling function that fails with retryAfter
+		const p = fnWithBackoff();
+		// Check that function was only called once (it runs right away)
+		expect(failer.callCount).to.equal(1);
+		// Tick clock by minDelay
+		// This will not be enough to call the function because retryAfter was in the
+		// exception thrown and it is greater then minDelay
+		await clock.tickAsync(minDelay);
+		// Check that function wasn't called yet
+		expect(failer.callCount).to.equal(1);
+		// Tick clock again just before retryAfter
+		await clock.tickAsync(retryAfter - minDelay - 1);
+		// Check that call count is still only 1 since retryAfter time has not elapsed
+		expect(failer.callCount).to.equal(1);
+		// Elapse enough time to trigger function call
+		await clock.tickAsync(1);
+		// Check that function has now been called once more
+		expect(failer.callCount).to.equal(2);
+		// Failure was set to only fail once so function should be fulfilled after 2 executions
+		await expect(p).to.be.fulfilled;
+	});
+});
+
+function exponentialize(n: number, timesToExpo: number): number {
+	let product = 0;
+	let exponentialized = 0;
+	function expo(): number {
+		if (exponentialized >= timesToExpo) {
+			return product;
+		}
+		exponentialized++;
+		product = product + n * 2;
+		return expo();
+	}
+	return expo();
+}
+
+class Failer {
+	public maxFails: number;
+	public callCount: number;
+	public customError?: Error;
+
+	public constructor(maxFails: number, error?: Error) {
+		this.maxFails = maxFails;
+		this.callCount = 0;
+		this.customError = error;
+	}
+
+	public async willResolve(resolvesWith: string): Promise<string> {
+		return new Promise((resolve, reject) => {
+			if (++this.callCount <= this.maxFails) {
+				if (this.customError) {
+					return reject(this.customError);
+				}
+				return reject('Not ready!');
+			}
+			return resolve(resolvesWith);
+		});
+	}
+}