Force update on device connects to avoid potential cross-instance races

src/api.ts

@@ -82,7 +82,6 @@ export const apiFactory = (serviceId: number) => {
 		// Immediately respond to minimize time in the client-connect script
 		res.status(200).end();
 
-		const workerId = parseInt(req.params.worker, 10);
 		const uuid = req.body.common_name;
 
 		const startingRefCount = clientRefCount.get(uuid) ?? 0;
@@ -97,7 +96,7 @@ export const apiFactory = (serviceId: number) => {
 		if (startingRefCount >= 0) {
 			// Only set the device as connected if the starting ref count is >= 0, this handles the case where
 			// the disconnect comes before the first connect so we go 0 -> -1 -> 0
-			setConnected(uuid, serviceId, workerId, true, logger);
+			setConnected(uuid, serviceId, true, logger);
 		}
 	});
 
@@ -161,7 +160,7 @@ export const apiFactory = (serviceId: number) => {
 
 		metrics.dec(Metrics.OnlineDevices);
 
-		setConnected(uuid, serviceId, workerId, false, logger);
+		setConnected(uuid, serviceId, false, logger);
 	});
 
 	return api;

src/app.ts

@@ -73,18 +73,21 @@ if (cluster.isPrimary) {
 	process.on('SIGUSR2', () => {
 		masterLogger.notice('caught SIGUSR2, toggling log verbosity');
 		verbose = !verbose;
-		_.each(cluster.workers, (clusterWorker) => {
-			if (clusterWorker != null) {
-				clusterWorker.send('toggleVerbosity');
-			}
-		});
+		if (cluster.workers == null) {
+			return;
+		}
+		for (const clusterWorker of Object.values(cluster.workers)) {
+			clusterWorker?.send('toggleVerbosity');
+		}
 	});
 
 	process.on('SIGTERM', () => {
 		masterLogger.notice('received SIGTERM');
-		_.each(cluster.workers, (clusterWorker) => {
-			clusterWorker?.send('prepareShutdown');
-		});
+		if (cluster.workers != null) {
+			for (const clusterWorker of Object.values(cluster.workers)) {
+				clusterWorker?.send('prepareShutdown');
+			}
+		}
 		masterLogger.notice(
 			`waiting ${DEFAULT_SIGTERM_TIMEOUT}ms for workers to finish`,
 		);
@@ -154,7 +157,7 @@ if (cluster.isPrimary) {
 						VPN_INSTANCE_COUNT > 1 ? 's' : ''
 					}`,
 				);
-				_.times(VPN_INSTANCE_COUNT, (i) => {
+				for (let i = 0; i < VPN_INSTANCE_COUNT; i++) {
 					const workerId = i + 1;
 					const restartWorker = (code?: number, signal?: string) => {
 						if (signal != null) {
@@ -174,7 +177,7 @@ if (cluster.isPrimary) {
 						cluster.fork(env).on('exit', restartWorker);
 					};
 					restartWorker();
-				});
+				}
 
 				const aggregatorRegistry = new prometheus.AggregatorRegistry();
 

src/utils/clients.ts

@@ -41,9 +41,8 @@ import { getPassthrough } from './index.js';
 
 interface DeviceStateTracker {
 	targetConnected: boolean;
-	targetWorkerId: number;
-	currentConnected?: boolean;
-	currentWorkerId?: number;
+	currentConnected: boolean;
+	forceUpdate: boolean;
 }
 
 export const setConnected = (() => {
@@ -86,10 +85,7 @@ export const setConnected = (() => {
 					for (const uuid of uuidChunk) {
 						const deviceState = deviceStates.get(uuid)!;
 						deviceState.currentConnected = connected;
-						// We mark the currentWorkerId to match the target since the key is that we have sent a new connect
-						// event since the worker has connected in case it went from eg `us-1` -> `other-1` -> `us-2` and
-						// `other` is seen as being in charge of the device when it actually should be us
-						deviceState.currentWorkerId = deviceState.targetWorkerId;
+						deviceState.forceUpdate = false;
 						logger.debug(
 							`successfully updated state for device: uuid=${uuid} connected=${connected}`,
 						);
@@ -117,18 +113,12 @@ export const setConnected = (() => {
 			const disconnects = [];
 			const connects = [];
 			for (const uuid of pendingUpdates) {
-				const {
-					targetConnected,
-					targetWorkerId,
-					currentConnected,
-					currentWorkerId,
-				} = deviceStates.get(uuid)!;
+				const { targetConnected, currentConnected, forceUpdate } =
+					deviceStates.get(uuid)!;
 				// We only try to update those where the target/current state differs, any where it matches
-				// will naturally be dropped from pending updates as expected as there is no pending update
-				if (
-					targetConnected !== currentConnected ||
-					targetWorkerId !== currentWorkerId
-				) {
+				// will naturally be dropped from pending updates as expected as there is no pending update,
+				// with the exception being if a force update has been marked
+				if (targetConnected !== currentConnected || forceUpdate === true) {
 					if (targetConnected) {
 						connects.push(uuid);
 					} else {
@@ -153,19 +143,24 @@ export const setConnected = (() => {
 	return (
 		uuid: string,
 		serviceId: number,
-		workerId: number,
 		connected: boolean,
 		logger: Logger,
 	) => {
 		const deviceState = deviceStates.get(uuid);
 		if (deviceState == null) {
 			deviceStates.set(uuid, {
 				targetConnected: connected,
-				targetWorkerId: workerId,
+				currentConnected: false,
+				forceUpdate: false,
 			});
 		} else {
 			deviceState.targetConnected = connected;
-			deviceState.targetWorkerId = workerId;
+			if (deviceState.currentConnected === true) {
+				// If we think the device is already connected but are marking it as connected again then we force an
+				// update in case the device has gone from eg `us` -> `other` -> `us` and `other` is seen as being in
+				// charge of the device when it actually should be us
+				deviceState.forceUpdate = true;
+			}
 		}
 		pendingUpdates.add(uuid);
 		void updateLoop(serviceId, logger);