Merge pull request #3533 from bluewave-labs/fix/monitor-status-atomicity

fix: monitor status atomicity

server/src/repositories/monitors/IMonitorsRepository.ts

@@ -1,4 +1,4 @@
-import { type MonitorType, type Monitor, type MonitorsSummary } from "@/types/index.js";
+import { type MonitorType, type Monitor, type MonitorsSummary, CheckSnapshot } from "@/types/index.js";
 
 export interface TeamQueryConfig {
 	limit?: number;
@@ -29,6 +29,15 @@ export interface IMonitorsRepository {
 
 	// update
 	updateById(monitorId: string, teamId: string, updates: Partial<Monitor>): Promise<Monitor>;
+	updateStatusWindowAndChecks(
+		monitorId: string,
+		teamId: string,
+		status: boolean,
+		checkSnapshot: CheckSnapshot,
+		windowSize: number,
+		maxRecentChecks: number,
+		statusPatch?: Partial<Monitor>
+	): Promise<Monitor>;
 	togglePauseById(monitorId: string, teamId: string): Promise<Monitor>;
 	// delete
 	deleteById(monitorId: string, teamId: string): Promise<Monitor>;

server/src/repositories/monitors/MongoMonitorsRepository.ts

@@ -182,6 +182,33 @@ class MongoMonitorsRepository implements IMonitorsRepository {
 		return this.toEntity(updatedMonitor);
 	};
 
+	updateStatusWindowAndChecks = async (
+		monitorId: string,
+		teamId: string,
+		status: boolean,
+		checkSnapshot: CheckSnapshot,
+		windowSize: number,
+		maxRecentChecks: number,
+		statusPatch?: Partial<Monitor>
+	): Promise<Monitor> => {
+		const updatedMonitor = await MonitorModel.findOneAndUpdate(
+			{ _id: monitorId, teamId },
+			{
+				$push: {
+					statusWindow: { $each: [status], $slice: -windowSize },
+					recentChecks: { $each: [checkSnapshot], $slice: -maxRecentChecks },
+				},
+				...(statusPatch && { $set: statusPatch }),
+			},
+			{ returnDocument: "after" }
+		);
+
+		if (!updatedMonitor) {
+			throw new AppError({ message: `Failed to update status and checks for monitor with id ${monitorId}`, status: 500 });
+		}
+		return this.toEntity(updatedMonitor);
+	};
+
 	togglePauseById = async (monitorId: string, teamId: string) => {
 		const monitor = await MonitorModel.findOneAndUpdate(
 			{ _id: monitorId, teamId },

server/src/repositories/monitors/TimescaleMonitorsRepository.ts

@@ -1,5 +1,5 @@
 import type { Pool } from "pg";
-import type { Monitor, MonitorsSummary, MonitorStatus, MonitorType, MonitorMatchMethod, GeoContinent } from "@/types/monitor.js";
+import type { Monitor, MonitorsSummary, MonitorStatus, MonitorType, MonitorMatchMethod, GeoContinent, CheckSnapshot } from "@/types/monitor.js";
 import type { IMonitorsRepository, TeamQueryConfig, SummaryConfig } from "./IMonitorsRepository.js";
 import { AppError } from "@/utils/AppError.js";
 
@@ -641,6 +641,55 @@ export class TimescaleMonitorsRepository implements IMonitorsRepository {
 		return entity;
 	};
 
+	updateStatusWindowAndChecks = async (
+		monitorId: string,
+		teamId: string,
+		status: boolean,
+		_checkSnapshot: CheckSnapshot,
+		windowSize: number,
+		_maxRecentChecks: number,
+		statusPatch?: Partial<Monitor>
+	): Promise<Monitor> => {
+		// In TimescaleDB, recentChecks live in the checks table (inserted by the buffer service),
+		// so we only need to atomically update status_window and any statusPatch fields.
+		const sets: string[] = [
+			`status_window = (array_append(COALESCE(status_window, ARRAY[]::boolean[]), $1))[array_length(array_append(COALESCE(status_window, ARRAY[]::boolean[]), $1), 1) - $2 + 1:]`,
+			`updated_at = NOW()`,
+		];
+		const values: unknown[] = [status, windowSize];
+		let paramIndex = 3;
+
+		if (statusPatch) {
+			const patchFieldMap: [keyof Monitor, string][] = [
+				["status", "status"],
+				["cpuAlertCounter", "cpu_alert_counter"],
+				["memoryAlertCounter", "memory_alert_counter"],
+				["diskAlertCounter", "disk_alert_counter"],
+				["tempAlertCounter", "temp_alert_counter"],
+			];
+			for (const [key, column] of patchFieldMap) {
+				if (statusPatch[key] !== undefined) {
+					sets.push(`${column} = $${paramIndex++}`);
+					values.push(statusPatch[key]);
+				}
+			}
+		}
+
+		values.push(monitorId, teamId);
+		const result = await this.pool.query<MonitorRow>(
+			`UPDATE monitors SET ${sets.join(", ")} WHERE id = $${paramIndex++} AND team_id = $${paramIndex}
+			 RETURNING ${MONITOR_COLUMNS}`,
+			values
+		);
+		const row = result.rows[0];
+		if (!row) {
+			throw new AppError({ message: `Failed to update status and checks for monitor with id ${monitorId}`, status: 500 });
+		}
+		const entity = this.toEntity(row);
+		entity.notifications = await this.fetchNotificationIds([monitorId]).then((m) => m.get(monitorId) ?? []);
+		return entity;
+	};
+
 	togglePauseById = async (monitorId: string, teamId: string): Promise<Monitor> => {
 		const result = await this.pool.query<MonitorRow>(
 			`UPDATE monitors SET

server/src/service/infrastructure/statusService.ts

@@ -102,14 +102,6 @@ export class StatusService implements IStatusService {
 		}
 	};
 
-	private appendToWindow = (monitor: Monitor, status: boolean) => {
-		monitor.statusWindow = monitor.statusWindow || [];
-		monitor.statusWindow.push(status);
-		while (monitor.statusWindow.length > monitor.statusWindowSize) {
-			monitor.statusWindow.shift();
-		}
-	};
-
 	private toCheckSnapshot = (check: Check): CheckSnapshot => {
 		return {
 			id: check.id,
@@ -134,15 +126,6 @@ export class StatusService implements IStatusService {
 		};
 	};
 
-	private appendToRecentChecks = (monitor: Monitor, check: Check) => {
-		monitor.recentChecks = monitor.recentChecks || [];
-		const checkSnapshot = this.toCheckSnapshot(check);
-		monitor.recentChecks.push(checkSnapshot);
-		while (monitor.recentChecks.length > MAX_RECENT_CHECKS) {
-			monitor.recentChecks.shift();
-		}
-	};
-
 	private computeReachability = (
 		currentStatus: MonitorStatus,
 		window: Array<boolean>,
@@ -238,16 +221,31 @@ export class StatusService implements IStatusService {
 			// Update running stats
 			await this.tryUpdateRunningStats(monitor, statusResponse);
 
-			// Update the sliding window and recent checks
-			this.appendToWindow(monitor, status);
-			this.appendToRecentChecks(monitor, check);
-
 			const prevStatus = monitor.status;
+			const checkSnapshot = this.toCheckSnapshot(check);
+
+			// Project the window as it will look after updating DB
+			// This is done because we need the updated status window to compute new status, but we don't
+			// want an an extra DB write just to get the window.
+			const projectedWindow = [...(monitor.statusWindow || []), check.status].slice(-monitor.statusWindowSize);
+
+			// Build the status patch — computed against the projected window
+			const patch: Partial<Monitor> = {};
+
+			// Not enough data points yet
+			if (projectedWindow.length < monitor.statusWindowSize) {
+				patch.status = status === true ? "up" : "down";
+
+				const updated = await this.monitorsRepository.updateStatusWindowAndChecks(
+					monitor.id,
+					monitor.teamId,
+					check.status,
+					checkSnapshot,
+					monitor.statusWindowSize,
+					MAX_RECENT_CHECKS,
+					patch
+				);
 
-			// Return early if not enough data points
-			if (monitor.statusWindow.length < monitor.statusWindowSize) {
-				monitor.status = status === true ? "up" : "down";
-				const updated = await this.monitorsRepository.updateById(monitor.id, monitor.teamId, monitor);
 				return {
 					monitor: updated,
 					statusChanged: false,
@@ -257,7 +255,8 @@ export class StatusService implements IStatusService {
 				};
 			}
 
-			// With a full window, a single raw check must not change UNLESS we are initializing. Otherwise, only the sliding-window threshold can trigger a transition.
+			// With a full window, a single raw check must not change UNLESS we are initializing.
+			// Otherwise, only the sliding-window threshold can trigger a transition.
 			let newStatus: MonitorStatus;
 			if (monitor.status === "initializing") {
 				newStatus = status === true ? "up" : "down";
@@ -267,8 +266,9 @@ export class StatusService implements IStatusService {
 
 			let statusChanged = false;
 
-			// First evaluate reachability-based status changes, which apply to all monitor types and take precedence over hardware breaches.
-			const reachabilityResult = this.computeReachability(newStatus, monitor.statusWindow, monitor.statusWindowThreshold);
+			// First evaluate reachability-based status changes, which apply to all monitor types
+			// and take precedence over hardware breaches.
+			const reachabilityResult = this.computeReachability(newStatus, projectedWindow, monitor.statusWindowThreshold);
 			if (reachabilityResult.transitioned) {
 				newStatus = reachabilityResult.nextStatus;
 				statusChanged = true;
@@ -296,10 +296,10 @@ export class StatusService implements IStatusService {
 					},
 				});
 
-				monitor.cpuAlertCounter = hardware.nextCounters.cpu;
-				monitor.memoryAlertCounter = hardware.nextCounters.memory;
-				monitor.diskAlertCounter = hardware.nextCounters.disk;
-				monitor.tempAlertCounter = hardware.nextCounters.temp;
+				patch.cpuAlertCounter = hardware.nextCounters.cpu;
+				patch.memoryAlertCounter = hardware.nextCounters.memory;
+				patch.diskAlertCounter = hardware.nextCounters.disk;
+				patch.tempAlertCounter = hardware.nextCounters.temp;
 				thresholdBreaches = hardware.breaches;
 				if (hardware.transitioned) {
 					newStatus = hardware.nextStatus;
@@ -307,10 +307,18 @@ export class StatusService implements IStatusService {
 				}
 			}
 
-			// Apply the final status
-			monitor.status = newStatus;
+			patch.status = newStatus;
 
-			const updated = await this.monitorsRepository.updateById(monitor.id, monitor.teamId, monitor);
+			// Single atomic write: push arrays + set status/counters
+			const updated = await this.monitorsRepository.updateStatusWindowAndChecks(
+				monitor.id,
+				monitor.teamId,
+				check.status,
+				checkSnapshot,
+				monitor.statusWindowSize,
+				MAX_RECENT_CHECKS,
+				patch
+			);
 
 			return {
 				monitor: updated,

server/test/helpers/InMemoryMonitorsRepository.ts

@@ -1,5 +1,5 @@
 import type { IMonitorsRepository, TeamQueryConfig, SummaryConfig } from "../../src/repositories/monitors/IMonitorsRepository.ts";
-import type { Monitor, MonitorsSummary } from "../../src/types/index.ts";
+import type { Monitor, MonitorsSummary, CheckSnapshot } from "../../src/types/index.ts";
 
 export class InMemoryMonitorsRepository implements IMonitorsRepository {
 	private monitors: Monitor[] = [];
@@ -49,6 +49,37 @@ export class InMemoryMonitorsRepository implements IMonitorsRepository {
 		return { ...updated };
 	}
 
+	async updateStatusWindowAndChecks(
+		monitorId: string,
+		teamId: string,
+		status: boolean,
+		checkSnapshot: CheckSnapshot,
+		windowSize: number,
+		maxRecentChecks: number,
+		statusPatch?: Partial<Monitor>
+	): Promise<Monitor> {
+		const index = this.monitors.findIndex((m) => m.id === monitorId && m.teamId === teamId);
+		if (index === -1) {
+			throw new Error(`Monitor ${monitorId} not found`);
+		}
+		const monitor = this.monitors[index];
+		monitor.statusWindow = monitor.statusWindow || [];
+		monitor.statusWindow.push(status);
+		while (monitor.statusWindow.length > windowSize) {
+			monitor.statusWindow.shift();
+		}
+		monitor.recentChecks = monitor.recentChecks || [];
+		monitor.recentChecks.push(checkSnapshot);
+		while (monitor.recentChecks.length > maxRecentChecks) {
+			monitor.recentChecks.shift();
+		}
+		if (statusPatch) {
+			Object.assign(monitor, statusPatch);
+		}
+		this.monitors[index] = monitor;
+		return { ...monitor };
+	}
+
 	async togglePauseById(monitorId: string, teamId: string): Promise<Monitor> {
 		const monitor = await this.findById(monitorId, teamId);
 		const newStatus = monitor.status === "paused" ? "up" : "paused";

server/test/unit/services/statusService.test.ts

@@ -11,11 +11,49 @@ const createBuffer = (): jest.Mocked<Pick<IBufferService, "addToBuffer">> => ({
 	addToBuffer: jest.fn(),
 });
 
-const createMonitorsRepo = () =>
-	({
-		findById: jest.fn(),
+const createMonitorsRepo = () => {
+	const findById = jest.fn();
+	const updateStatusWindowAndChecks = jest
+		.fn()
+		.mockImplementation(
+			(
+				_id: unknown,
+				_tid: unknown,
+				status: boolean,
+				checkSnapshot: unknown,
+				windowSize: number,
+				maxRecentChecks: number,
+				statusPatch?: Partial<Monitor>
+			) => {
+				// Grab the monitor that findById was configured to resolve with
+				const monitor = findById.mock.results.at(-1)?.value;
+				if (monitor && typeof monitor.then === "function") {
+					return monitor.then((m: any) => {
+						m.statusWindow = m.statusWindow || [];
+						m.statusWindow.push(status);
+						while (m.statusWindow.length > windowSize) {
+							m.statusWindow.shift();
+						}
+						m.recentChecks = m.recentChecks || [];
+						m.recentChecks.push(checkSnapshot);
+						while (m.recentChecks.length > maxRecentChecks) {
+							m.recentChecks.shift();
+						}
+						if (statusPatch) {
+							Object.assign(m, statusPatch);
+						}
+						return { ...m };
+					});
+				}
+				return Promise.reject(new Error("findById not mocked"));
+			}
+		);
+	return {
+		findById,
 		updateById: jest.fn(),
-	}) as unknown as jest.Mocked<IMonitorsRepository>;
+		updateStatusWindowAndChecks,
+	} as unknown as jest.Mocked<IMonitorsRepository>;
+};
 
 const createMonitorStatsRepo = () =>
 	({
@@ -233,11 +271,18 @@ describe("StatusService", () => {
 			(monitorsRepository.findById as jest.Mock).mockResolvedValue(monitor);
 			(monitorsRepository.updateById as jest.Mock).mockImplementation((_id: unknown, _tid: unknown, m: unknown) => Promise.resolve(m));
 
-			await service.updateMonitorStatus(makeStatusResponse({ status: false }), makeCheck());
+			await service.updateMonitorStatus(makeStatusResponse({ status: false }), makeCheck({ status: false }));
 
-			// Window should have shifted: [true, true, true, true, false]
-			expect(monitor.statusWindow).toHaveLength(5);
-			expect(monitor.statusWindow[4]).toBe(false);
+			// Atomic push should have been called with the correct status and window size
+			expect(monitorsRepository.updateStatusWindowAndChecks).toHaveBeenCalledWith(
+				"mon-1",
+				"team-1",
+				false,
+				expect.any(Object),
+				5,
+				25,
+				expect.objectContaining({ status: expect.any(String) })
+			);
 		});
 
 		it("pushes check snapshot to recentChecks and trims to 25", async () => {
@@ -542,7 +587,8 @@ describe("StatusService", () => {
 
 				expect(result.thresholdBreaches?.cpu).toBe(true);
 				expect(result.thresholdBreaches?.memory).toBe(false);
-				expect(monitor.cpuAlertCounter).toBe(0);
+				const patch = (monitorsRepository.updateStatusWindowAndChecks as jest.Mock).mock.calls.at(-1)?.[6];
+				expect(patch.cpuAlertCounter).toBe(0);
 				expect(result.statusChanged).toBe(true);
 				expect(result.monitor.status).toBe("breached");
 			});
@@ -557,7 +603,8 @@ describe("StatusService", () => {
 				const result = await service.updateMonitorStatus(response, makeCheck());
 
 				expect(result.thresholdBreaches?.memory).toBe(true);
-				expect(monitor.memoryAlertCounter).toBe(0);
+				const patch = (monitorsRepository.updateStatusWindowAndChecks as jest.Mock).mock.calls.at(-1)?.[6];
+				expect(patch.memoryAlertCounter).toBe(0);
 				expect(result.monitor.status).toBe("breached");
 			});
 
@@ -573,7 +620,8 @@ describe("StatusService", () => {
 				const result = await service.updateMonitorStatus(response, makeCheck());
 
 				expect(result.thresholdBreaches?.disk).toBe(true);
-				expect(monitor.diskAlertCounter).toBe(0);
+				const patch = (monitorsRepository.updateStatusWindowAndChecks as jest.Mock).mock.calls.at(-1)?.[6];
+				expect(patch.diskAlertCounter).toBe(0);
 				expect(result.monitor.status).toBe("breached");
 			});
 
@@ -589,7 +637,8 @@ describe("StatusService", () => {
 				const result = await service.updateMonitorStatus(response, makeCheck());
 
 				expect(result.thresholdBreaches?.temp).toBe(true);
-				expect(monitor.tempAlertCounter).toBe(0);
+				const patch = (monitorsRepository.updateStatusWindowAndChecks as jest.Mock).mock.calls.at(-1)?.[6];
+				expect(patch.tempAlertCounter).toBe(0);
 				expect(result.monitor.status).toBe("breached");
 			});
 
@@ -609,10 +658,11 @@ describe("StatusService", () => {
 				} as any);
 				await service.updateMonitorStatus(response, makeCheck());
 
-				expect(monitor.cpuAlertCounter).toBe(5);
-				expect(monitor.memoryAlertCounter).toBe(5);
-				expect(monitor.diskAlertCounter).toBe(5);
-				expect(monitor.tempAlertCounter).toBe(5);
+				const patch = (monitorsRepository.updateStatusWindowAndChecks as jest.Mock).mock.calls.at(-1)?.[6];
+				expect(patch.cpuAlertCounter).toBe(5);
+				expect(patch.memoryAlertCounter).toBe(5);
+				expect(patch.diskAlertCounter).toBe(5);
+				expect(patch.tempAlertCounter).toBe(5);
 			});
 
 			it("stays breached without statusChanged when already breached and counter still at 0", async () => {
@@ -841,7 +891,8 @@ describe("StatusService", () => {
 				const response = makeHardwareResponse({ data: { cpu: { usage_percent: 0.9 }, memory: { usage_percent: 0.1 }, disk: [], host: {} } } as any);
 				const result = await service.updateMonitorStatus(response, makeCheck());
 
-				expect(monitor.cpuAlertCounter).toBe(2);
+				const patch = (monitorsRepository.updateStatusWindowAndChecks as jest.Mock).mock.calls.at(-1)?.[6];
+				expect(patch.cpuAlertCounter).toBe(2);
 				expect(result.thresholdBreaches?.cpu).toBe(true);
 				expect(result.statusChanged).toBe(false);
 				expect(result.monitor.status).toBe("up");