Merge pull request #3693 from bluewave-labs/feat/worker-stats

feat: worker stats

client/src/Pages/Logs/TabQueue.tsx

@@ -2,7 +2,7 @@ import Stack from "@mui/material/Stack";
 import Typography from "@mui/material/Typography";
 import Box from "@mui/material/Box";
 import { TableJobs, TableFailedJobs } from "./components/TableJobs";
-
+import { TableWorkers } from "./components/TableWorkers";
 import { useTheme } from "@mui/material";
 import { useTranslation, Trans } from "react-i18next";
 import { useGet, usePost } from "@/Hooks/UseApi";
@@ -25,6 +25,10 @@ export const TabQueue = () => {
 	const { post, loading: isFlushing } = usePost();
 
 	const jobs = queueData?.jobs ?? [];
+	const workers = (queueData?.metrics?.workers ?? []).map((worker, idx) => ({
+		...worker,
+		id: idx,
+	}));
 	const metrics = queueData?.metrics ?? null;
 	const hasNeverRun = !isLoading && metrics !== null && (metrics.totalRuns ?? 0) === 0;
 	const noQueueData = !isLoading && metrics === null && jobs.length === 0;
@@ -46,6 +50,34 @@ export const TabQueue = () => {
 	return (
 		<Stack gap={theme.spacing(16)}>
 			<Metrics metrics={metrics} />
+			<Stack gap={theme.spacing(3)}>
+				<Stack gap={theme.spacing(1)}>
+					<Typography
+						variant="eyebrow"
+						color={theme.palette.text.secondary}
+					>
+						{t("pages.logs.workers")}
+					</Typography>
+					<Typography color={theme.palette.text.secondary}>
+						<Trans
+							i18nKey="pages.logs.workersExplainers"
+							components={{
+								highlight: (
+									<Box
+										component="span"
+										bgcolor={theme.palette.rowStatus.running}
+										color={theme.palette.text.primary}
+										px={SPACING.SM}
+										py={SPACING.XXS}
+										borderRadius={theme.shape.borderRadius}
+									/>
+								),
+							}}
+						/>
+					</Typography>
+				</Stack>
+				<TableWorkers queueWorkers={workers} />
+			</Stack>
 			<Stack gap={theme.spacing(3)}>
 				<Stack gap={theme.spacing(1)}>
 					<Typography

client/src/Pages/Logs/components/TableJobs.tsx

@@ -127,6 +127,27 @@ export const TableJobs = ({ jobs }: TableJobsProps) => {
 				);
 			},
 		},
+		{
+			id: "lockedBy",
+			content: t("pages.logs.table.headers.lockedBy"),
+			render: (row) => {
+				const lockedBy = row.lockedBy;
+				if (!lockedBy) {
+					return <Typography sx={cellSx}>-</Typography>;
+				}
+
+				const workerId = lockedBy.split(":")[2];
+				const shortWorkerId = workerId.slice(workerId.length - 8);
+				return (
+					<Typography
+						title={workerId}
+						sx={cellSx}
+					>
+						{`...${shortWorkerId}`}
+					</Typography>
+				);
+			},
+		},
 	];
 
 	return (

client/src/Pages/Logs/components/TableWorkers.tsx

@@ -0,0 +1,94 @@
+import { Table } from "@/Components/design-elements";
+import { Typography, useTheme } from "@mui/material";
+import { formatTimestamp } from "@/Utils/TimeUtils";
+
+import { useTranslation } from "react-i18next";
+import type { QueueWorker } from "@/Types/Queue";
+import type { Header } from "@/Components/design-elements";
+
+type QueueWorkerWithId = QueueWorker & { id: number };
+
+interface TableWorkersProps {
+	queueWorkers: QueueWorkerWithId[];
+}
+
+export const TableWorkers = ({ queueWorkers }: TableWorkersProps) => {
+	const { t } = useTranslation();
+	const theme = useTheme();
+
+	const cellSx = {
+		whiteSpace: "nowrap" as const,
+		overflow: "hidden",
+		textOverflow: "ellipsis",
+		maxWidth: 220,
+	};
+
+	const headers: Header<QueueWorkerWithId>[] = [
+		{
+			id: "host",
+			content: t("pages.logs.table.headers.host"),
+			render: (row) => {
+				const host = row.workerId.split(":")[0];
+				return (
+					<Typography
+						fontFamily={theme.typography.fontFamilyMonospace}
+						sx={cellSx}
+					>
+						{host}
+					</Typography>
+				);
+			},
+		},
+		{
+			id: "processId",
+			content: t("pages.logs.table.headers.processId"),
+			render: (row) => {
+				const processId = row.workerId.split(":")[1];
+				return (
+					<Typography
+						fontFamily={theme.typography.fontFamilyMonospace}
+						sx={cellSx}
+					>
+						{processId}
+					</Typography>
+				);
+			},
+		},
+		{
+			id: "id",
+			content: t("pages.logs.table.headers.workerId"),
+			render: (row) => {
+				const workerId = row.workerId.split(":")[2];
+				const shortWorkerId = workerId.slice(workerId.length - 8);
+				return (
+					<Typography
+						fontFamily={theme.typography.fontFamilyMonospace}
+						title={workerId}
+						sx={cellSx}
+					>
+						{`...${shortWorkerId}`}
+					</Typography>
+				);
+			},
+		},
+		{
+			id: "mode",
+			content: t("pages.logs.table.headers.workerMode"),
+			render: (row) => <Typography sx={cellSx}>{row.mode}</Typography>,
+		},
+		{
+			id: "lastSeen",
+			content: t("pages.logs.table.headers.lastSeenAt"),
+			render: (row) => (
+				<Typography sx={cellSx}>{formatTimestamp(row.lastSeenAt)}</Typography>
+			),
+		},
+	];
+
+	return (
+		<Table
+			headers={headers}
+			data={queueWorkers}
+		/>
+	);
+};

client/src/Pages/Logs/index.tsx

@@ -8,7 +8,7 @@ import { useTranslation } from "react-i18next";
 
 const LogsPage = () => {
 	const { t } = useTranslation();
-	const [activeTab, setActiveTab] = useState<number>(0);
+	const [activeTab, setActiveTab] = useState<number>(1);
 	return (
 		<BasePage headerKey="logs">
 			<Tabs

client/src/Types/Queue.ts

@@ -1,5 +1,14 @@
 import type { MonitorType } from "./Monitor";
 
+export const QueueModes = ["primary", "worker"] as const;
+export type QueueMode = (typeof QueueModes)[number];
+
+export type QueueWorker = {
+	workerId: string; // hostname:pid:uuid
+	mode: QueueMode;
+	lastSeenAt: number; // epoch ms of the last heartbeat
+};
+
 export interface QueueJobFailure {
 	monitorId: string | number;
 	monitorUrl: string | null;
@@ -9,23 +18,26 @@ export interface QueueJobFailure {
 	failReason: string | null;
 }
 
-export interface QueueMetrics {
+export type QueueMetrics = {
 	jobs: number;
 	activeJobs: number;
 	failingJobs: number;
 	jobsWithFailures: QueueJobFailure[];
 	totalRuns: number;
 	totalFailures: number;
-}
+	workers: QueueWorker[];
+};
 
-export interface QueueJobSummary {
+export type QueueJobSummary = {
 	monitorId: string | number;
 	monitorUrl: string | null;
-	monitorType: MonitorType | null;
+	monitorType: string | null;
 	monitorInterval: number | null;
 	monitorGeoInterval: number | null;
 	monitorActive: boolean | null;
 	active: boolean;
+	lockedBy: string | null;
+	lockedUntil: number | null;
 	lockedAt: number | null;
 	runCount: number;
 	failCount: number;
@@ -35,7 +47,7 @@ export interface QueueJobSummary {
 	lastRunTook: number | null;
 	lastFailedAt: number | null;
 	repeat: number | null;
-}
+};
 
 export interface QueueData {
 	jobs: QueueJobSummary[];

client/src/locales/en.json

@@ -933,6 +933,8 @@
 			"logLevelSelect": {
 				"label": "Log level"
 			},
+			"workers": "Workers",
+			"workersExplainers": "Shows worker processes.  Worker IDs are truncated, hover for full ID",
 			"jobQueue": "Job queue",
 			"jobQueueExplainer": "<highlight>Highlighted rows</highlight> are jobs currently being processed by a worker.",
 			"failedJobs": "Failed jobs",
@@ -1000,6 +1002,11 @@
 			},
 			"table": {
 				"headers": {
+					"host": "Host",
+					"processId": "Process ID",
+					"workerId": "Worker ID",
+					"workerMode": "Worker mode",
+					"lastSeenAt": "Last seen at",
 					"state": "State",
 					"timestamp": "Timestamp",
 					"level": "Level",
@@ -1010,6 +1017,7 @@
 					"failCount": "Count",
 					"lastRunAt": "Last run at",
 					"lockedAt": "Locked at",
+					"lockedBy": "Locked by",
 					"lastFinishedAt": "Last finished at",
 					"lastRunTook": "Last run took",
 					"lastFailedAt": "Last failed",

server/package-lock.json

@@ -31,7 +31,7 @@
 				"jmespath": "^0.16.0",
 				"jsdom": "^26.1.0",
 				"jsonwebtoken": "9.0.2",
-				"less-simple-scheduler": "2.2.0",
+				"less-simple-scheduler": "2.3.0",
 				"mailersend": "^2.2.0",
 				"mjml": "^5.0.0-alpha.4",
 				"mongoose": "^8.3.3",
@@ -10417,9 +10417,9 @@
 			"license": "MIT"
 		},
 		"node_modules/less-simple-scheduler": {
-			"version": "2.2.0",
-			"resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.2.0.tgz",
-			"integrity": "sha512-0bbFohlkcWZY2wfkGfDiYEbDxkshj0RuskkL0RLUDMn1Y3WjoPqy29PAnKNtnHG+DQqn0hMjZ0e3hVFJ4k76oA==",
+			"version": "2.3.0",
+			"resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.3.0.tgz",
+			"integrity": "sha512-HL8f8YCcNav03oNTWFQxkWmoqtEzsIXNOgLcVmOEAgFxsALukOk4bDlem6SlZmjVAsip8Y50QUu6ZsblxqQYmA==",
 			"license": "ISC",
 			"dependencies": {
 				"mongodb": "7.2.0"

server/package.json

@@ -53,7 +53,7 @@
 		"jmespath": "^0.16.0",
 		"jsdom": "^26.1.0",
 		"jsonwebtoken": "9.0.2",
-		"less-simple-scheduler": "2.2.0",
+		"less-simple-scheduler": "2.3.0",
 		"mailersend": "^2.2.0",
 		"mjml": "^5.0.0-alpha.4",
 		"mongoose": "^8.3.3",

server/src/config/services.ts

@@ -107,6 +107,7 @@ import {
 	MongoTeamsRepository,
 	MongoMaintenanceWindowsRepository,
 	MongoSettingsRepository,
+	MongoQueueWorkersRepository,
 	IMonitorsRepository,
 	IChecksRepository,
 	IGeoChecksRepository,
@@ -248,6 +249,7 @@ export const initializeServices = async ({
 	const incidentsRepository = new MongoIncidentsRepository();
 	const teamsRepository = new MongoTeamsRepository();
 	const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository();
+	const queueWorkersRepository = new MongoQueueWorkersRepository();
 
 	// Inject settings repository into settings service (now that DB is connected)
 	(settingsService as SettingsService).setRepository(settingsRepository);
@@ -360,7 +362,7 @@ export const initializeServices = async ({
 	let jobQueue: IJobQueue;
 	switch (envSettings.queueType) {
 		case "lessSimpleQueue":
-			jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, envSettings, queueMode);
+			jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode);
 			break;
 		default:
 			jobQueue = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository);

server/src/db/models/QueueWorker.ts

@@ -0,0 +1,28 @@
+import { Schema, model } from "mongoose";
+import { QueueModes, type QueueMode } from "@/types/settings.js";
+
+// Heartbeat presence record for a scheduler instance. _id is the scheduler's
+// workerId (hostname:pid:uuid). The TTL index on lastSeenAt garbage-collects
+// records for instances that have stopped heartbeating.
+interface QueueWorkerDocument {
+	_id: string;
+	mode: QueueMode;
+	lastSeenAt: Date;
+	createdAt: Date;
+	updatedAt: Date;
+}
+
+const QueueWorkerSchema = new Schema<QueueWorkerDocument>(
+	{
+		_id: { type: String, required: true },
+		mode: { type: String, enum: QueueModes, required: true },
+		lastSeenAt: { type: Date, default: Date.now, expires: 30 },
+	},
+	{ timestamps: true }
+);
+
+const QueueWorkerModel = model<QueueWorkerDocument>("QueueWorker", QueueWorkerSchema);
+
+export type { QueueWorkerDocument };
+export { QueueWorkerModel };
+export default QueueWorkerModel;

server/src/db/models/index.ts

@@ -41,3 +41,6 @@ export * from "@/db/models/DLQItem.js";
 export { default as DLQItemModel } from "@/db/models/DLQItem.js";
 export * from "@/db/models/Tag.js";
 export { default as TagModel } from "@/db/models/Tag.js";
+
+export * from "@/db/models/QueueWorker.js";
+export { default as QueueWorkerModel } from "@/db/models/QueueWorker.js";

server/src/repositories/index.ts

@@ -42,3 +42,6 @@ export { default as MongoGeoChecksRepository } from "@/repositories/geo-checks/M
 
 export * from "@/repositories/dlq/IDLQRepository.js";
 export { default as MongoDLQRepository } from "@/repositories/dlq/MongoDLQRepository.js";
+
+export * from "@/repositories/queue-workers/IQueueWorkersRepository.js";
+export { default as MongoQueueWorkersRepository } from "@/repositories/queue-workers/MongoQueueWorkersRepository.js";

server/src/repositories/queue-workers/IQueueWorkersRepository.ts

@@ -0,0 +1,7 @@
+import type { QueueMode, QueueWorker } from "@/types/settings.js";
+
+export interface IQueueWorkersRepository {
+	upsert(workerId: string, mode: QueueMode): Promise<void>;
+	findRecent(maxAgeMs: number): Promise<QueueWorker[]>;
+	deleteById(workerId: string): Promise<void>;
+}

server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts

@@ -0,0 +1,33 @@
+import type { QueueMode, QueueWorker } from "@/types/settings.js";
+import type { IQueueWorkersRepository } from "./IQueueWorkersRepository.js";
+import type { QueueWorkerDocument } from "@/db/models/QueueWorker.js";
+import { QueueWorkerModel } from "@/db/models/QueueWorker.js";
+const SERVICE_NAME = "MongoQueueWorkersRepository";
+
+class MongoQueueWorkersRepository implements IQueueWorkersRepository {
+	static SERVICE_NAME = SERVICE_NAME;
+
+	protected toEntity = (doc: QueueWorkerDocument): QueueWorker => {
+		return {
+			workerId: doc._id,
+			mode: doc.mode,
+			lastSeenAt: doc.lastSeenAt.getTime(),
+		};
+	};
+
+	upsert = async (workerId: string, mode: QueueMode): Promise<void> => {
+		await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true });
+	};
+
+	findRecent = async (maxAgeMs: number): Promise<QueueWorker[]> => {
+		const cutoff = new Date(Date.now() - maxAgeMs);
+		const docs = await QueueWorkerModel.find({ lastSeenAt: { $gte: cutoff } });
+		return docs.map(this.toEntity);
+	};
+
+	deleteById = async (workerId: string): Promise<void> => {
+		await QueueWorkerModel.deleteOne({ _id: workerId });
+	};
+}
+
+export default MongoQueueWorkersRepository;

server/src/service/infrastructure/JobQueues/IJobQueue.ts

@@ -1,4 +1,5 @@
 import { Monitor } from "@/types/monitor.js";
+import { QueueWorker } from "@/types/settings.js";
 
 export type QueueJobFailure = {
 	monitorId: string | number;
@@ -16,6 +17,7 @@ export type QueueMetrics = {
 	jobsWithFailures: QueueJobFailure[];
 	totalRuns: number;
 	totalFailures: number;
+	workers: QueueWorker[];
 };
 
 export type QueueJobSummary = {
@@ -26,6 +28,8 @@ export type QueueJobSummary = {
 	monitorGeoInterval: number | null;
 	monitorActive: boolean | null;
 	active: boolean;
+	lockedBy: string | null;
+	lockedUntil: number | null;
 	lockedAt: number | null;
 	runCount: number;
 	failCount: number;

server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts

@@ -1,4 +1,4 @@
-import { IMonitorsRepository } from "@/repositories/index.js";
+import { IMonitorsRepository, IQueueWorkersRepository } from "@/repositories/index.js";
 import { ILogger } from "@/utils/logger.js";
 import { MongoStore, Scheduler, IJob } from "less-simple-scheduler";
 import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js";
@@ -8,6 +8,12 @@ import { type QueueMode } from "@/types/settings.js";
 import { EnvConfig } from "@/service/system/settingsService.js";
 const SERVICE_NAME = "JobQueue";
 
+// How long after its last heartbeat a worker is still considered alive. The
+// model's TTL index uses the same window to garbage-collect stale records.
+// Heartbeats are driven by the scheduler's `scheduler:heartbeat` event (emitted
+// every lockMs/3, so ~5s with the default lockMs), comfortably inside this TTL.
+const WORKER_TTL_MS = 30_000;
+
 // For discriminating job types
 type MonitorJob = Omit<IJob, "data"> & { data: Monitor };
 const MONITOR_TEMPLATES = new Set<string>(["monitor-job", "geo-check-job"]);
@@ -19,13 +25,25 @@ export class LessSimpleQueue implements IJobQueue {
 	private logger: ILogger;
 	private helper: IQueueHelper;
 	private monitorsRepository: IMonitorsRepository;
+	private workersRepository: IQueueWorkersRepository;
 	private readonly scheduler: Scheduler;
+	private queueMode: QueueMode;
+	private heartbeatListener: ((workerId: string) => void) | null = null;
 
-	constructor(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) {
+	constructor(
+		logger: ILogger,
+		helper: IQueueHelper,
+		monitorsRepository: IMonitorsRepository,
+		workersRepository: IQueueWorkersRepository,
+		scheduler: Scheduler,
+		queueMode: QueueMode
+	) {
 		this.logger = logger;
 		this.helper = helper;
 		this.monitorsRepository = monitorsRepository;
+		this.workersRepository = workersRepository;
 		this.scheduler = scheduler;
+		this.queueMode = queueMode;
 		this.registerListeners();
 	}
 
@@ -108,19 +126,45 @@ export class LessSimpleQueue implements IJobQueue {
 		});
 	};
 
-	static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, envSettings: EnvConfig, queueMode: QueueMode) {
+	static async create(
+		logger: ILogger,
+		helper: IQueueHelper,
+		monitorsRepository: IMonitorsRepository,
+		workersRepository: IQueueWorkersRepository,
+		envSettings: EnvConfig,
+		queueMode: QueueMode
+	) {
 		const store = new MongoStore({
 			url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db",
 		});
 		const scheduler = new Scheduler(store, {
 			concurrency: 50,
 		});
-		const instance = new LessSimpleQueue(logger, helper, monitorsRepository, scheduler);
+		const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode);
 		await instance.init(queueMode);
 
 		return instance;
 	}
 
+	private startHeartbeat = (queueMode: QueueMode) => {
+		const beat = (
+			workerId: string // This is the heartbeat function that registers workers
+		) =>
+			this.workersRepository.upsert(workerId, queueMode).catch((error: unknown) => {
+				this.logger.warn({
+					message: `Worker heartbeat failed: ${error instanceof Error ? error.message : String(error)}`,
+					service: SERVICE_NAME,
+					method: "startHeartbeat",
+				});
+			});
+		beat(this.scheduler.workerId); // Beat immediately to register
+		if (this.heartbeatListener) {
+			this.scheduler.off("scheduler:heartbeat", this.heartbeatListener); // remove old listeners
+		}
+		this.heartbeatListener = beat; /// Keep a reference to the listener
+		this.scheduler.on("scheduler:heartbeat", beat); // beat on every heartbeat
+	};
+
 	init = async (queueMode: QueueMode = "primary") => {
 		try {
 			await this.scheduler.start();
@@ -129,6 +173,9 @@ export class LessSimpleQueue implements IJobQueue {
 			this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob());
 			this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob());
 
+			// Register in the worker registry
+			this.startHeartbeat(queueMode);
+
 			// If this is a worker, return early - primary will populate queue
 			if (queueMode === "worker") {
 				this.logger.info({
@@ -284,11 +331,26 @@ export class LessSimpleQueue implements IJobQueue {
 	};
 
 	shutdown = async () => {
+		// Stop beating before we remove this worker, so a heartbeat firing between
+		// the remove and the scheduler stopping can't re-insert it.
+		if (this.heartbeatListener) {
+			this.scheduler.off("scheduler:heartbeat", this.heartbeatListener);
+			this.heartbeatListener = null;
+		}
+		await this.workersRepository.deleteById(this.scheduler.workerId).catch((error: unknown) => {
+			this.logger.warn({
+				message: `Failed to deregister worker on shutdown: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				method: "shutdown",
+			});
+		});
 		this.scheduler.stop();
 	};
 
 	getMetrics = async () => {
 		const jobs = await this.scheduler.getJobs();
+		const now = Date.now();
+
 		const metrics = jobs.reduce<QueueMetrics>(
 			(acc, job) => {
 				const runCount = job.runCount ?? 0;
@@ -302,7 +364,8 @@ export class LessSimpleQueue implements IJobQueue {
 					acc.failingJobs++;
 				}
 
-				if (job.lockedAt) {
+				// A job currently held by a live lock is in flight on some worker
+				if (job.lockedBy && job.lockedUntil && job.lockedUntil > now) {
 					acc.activeJobs++;
 				}
 
@@ -326,8 +389,10 @@ export class LessSimpleQueue implements IJobQueue {
 				jobsWithFailures: [],
 				totalRuns: 0,
 				totalFailures: 0,
+				workers: [],
 			}
 		);
+		metrics.workers = await this.workersRepository.findRecent(WORKER_TTL_MS);
 		return metrics;
 	};
 
@@ -345,6 +410,8 @@ export class LessSimpleQueue implements IJobQueue {
 				monitorActive: monitor?.isActive ?? null,
 				active: job.active,
 				repeat: job.repeat ?? null,
+				lockedBy: job.lockedBy ?? null,
+				lockedUntil: job.lockedUntil ?? null,
 				lockedAt: job.lockedAt ?? null,
 				runCount: job.runCount ?? 0,
 				failCount: job.failCount ?? 0,
@@ -358,8 +425,14 @@ export class LessSimpleQueue implements IJobQueue {
 	};
 
 	flushQueues = async () => {
-		const stopRes = await this.scheduler.stop();
-		const flushRes = await this.scheduler.flushJobs();
+		if (this.queueMode !== "primary") {
+			// Don't allow workers to flush, though this isn't currently reachable since workers have no API endpoints
+			this.logger.warn({ message: "Ignoring flush on non-primary worker", service: SERVICE_NAME, method: "flushQueues" });
+			return { success: false };
+		}
+
+		const flushRes = await this.scheduler.flushJobs(); // Less Simple Queue must flush before stopping as it needs DB
+		const stopRes = await this.scheduler.stop(); // Destructive, other workers will abort their jobs
 		const initRes = await this.init();
 		return {
 			success: Boolean(stopRes && flushRes && initRes),

server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts

@@ -3,6 +3,10 @@ import { ILogger } from "@/utils/logger.js";
 import Scheduler from "super-simple-scheduler";
 import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js";
 import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js";
+import { hostname } from "node:os";
+import { randomUUID } from "node:crypto";
+import { QueueMode } from "@/types/settings.js";
+
 const SERVICE_NAME = "JobQueue";
 
 type QueueJobFailure = {
@@ -61,6 +65,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 
 	private logger: ILogger;
 	private helper: IQueueHelper;
+	private workerId: string;
 	private monitorsRepository: IMonitorsRepository;
 	private readonly scheduler: Scheduler;
 
@@ -70,6 +75,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 		this.monitorsRepository = monitorsRepository;
 		this.scheduler = scheduler;
 		this.registerListeners();
+		this.workerId = `${hostname()}:${process.pid}:${randomUUID()}`;
 	}
 
 	get serviceName() {
@@ -311,7 +317,26 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 
 	getMetrics = async () => {
 		const jobs = await this.scheduler.getJobs();
-		const metrics = jobs.reduce(
+		const metrics: {
+			jobs: number;
+			activeJobs: number;
+			failingJobs: number;
+			jobsWithFailures: Array<{
+				monitorId: string | number;
+				monitorUrl: string;
+				monitorType: MonitorType;
+				failedAt: number | null;
+				failCount: number;
+				failReason: string | null;
+			}>;
+			totalRuns: number;
+			totalFailures: number;
+			workers: Array<{
+				workerId: string;
+				mode: QueueMode;
+				lastSeenAt: number;
+			}>;
+		} = jobs.reduce(
 			(acc, job) => {
 				const runCount = job.runCount ?? 0;
 				const failCount = job.failCount ?? 0;
@@ -354,6 +379,13 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 				}>,
 				totalRuns: 0,
 				totalFailures: 0,
+				workers: [
+					{
+						workerId: this.workerId,
+						mode: "primary",
+						lastSeenAt: Date.now(),
+					},
+				],
 			}
 		);
 		return metrics;
@@ -371,6 +403,8 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 				monitorActive: job.data?.isActive ?? null,
 				active: job.active,
 				repeat: job.repeat ?? null,
+				lockedBy: job.lockedAt ? this.workerId : null,
+				lockedUntil: null,
 				lockedAt: job.lockedAt ?? null,
 				runCount: job.runCount ?? 0,
 				failCount: job.failCount ?? 0,

server/src/types/settings.ts

@@ -7,6 +7,12 @@ export type QueueType = (typeof QueueTypes)[number];
 export const QueueModes = ["primary", "worker"] as const;
 export type QueueMode = (typeof QueueModes)[number];
 
+export type QueueWorker = {
+	workerId: string; // hostname:pid:uuid
+	mode: QueueMode;
+	lastSeenAt: number; // epoch ms of the last heartbeat
+};
+
 export interface SettingsThresholds {
 	cpu?: number;
 	memory?: number;

server/test/unit/services/lessSimpleQueue.test.ts

@@ -12,6 +12,9 @@ const createScheduler = () => {
 			listeners[event] = listeners[event] || [];
 			listeners[event].push(cb);
 		}),
+		off: jest.fn((event: string, cb: Function) => {
+			listeners[event] = (listeners[event] || []).filter((fn) => fn !== cb);
+		}),
 		emit: (event: string, ...args: unknown[]) => {
 			(listeners[event] || []).forEach((cb) => cb(...args));
 		},
@@ -51,18 +54,35 @@ const createMonitorsRepo = () => ({
 	findAll: jest.fn().mockResolvedValue([]),
 });
 
+const createWorkersRepo = () => ({
+	upsert: jest.fn().mockResolvedValue(undefined),
+	findRecent: jest.fn().mockResolvedValue([]),
+	deleteById: jest.fn().mockResolvedValue(undefined),
+});
+
 const createQueue = (overrides?: Record<string, unknown>) => {
 	const logger = createMockLogger();
 	const helper = createQueueHelper();
 	const monitorsRepository = createMonitorsRepo();
+	const workersRepository = createWorkersRepo();
 	const scheduler = createScheduler();
 
-	const defaults = { logger, helper, monitorsRepository, scheduler, ...overrides };
+	const defaults = { logger, helper, monitorsRepository, workersRepository, scheduler, queueMode: "primary", ...overrides };
 
-	const queue = new LessSimpleQueue(defaults.logger as any, defaults.helper as any, defaults.monitorsRepository as any, defaults.scheduler as any);
+	const queue = new LessSimpleQueue(
+		defaults.logger as any,
+		defaults.helper as any,
+		defaults.monitorsRepository as any,
+		defaults.workersRepository as any,
+		defaults.scheduler as any,
+		defaults.queueMode as any
+	);
 	return { queue, ...defaults };
 };
 
+// A lockedUntil far enough ahead that it is always > Date.now() at assertion time.
+const FAR_FUTURE = 4_000_000_000_000; // year 2096
+
 const makeMonitor = (overrides?: Partial<Monitor>): Monitor =>
 	({
 		id: "mon-1",
@@ -112,7 +132,8 @@ describe("LessSimpleQueue", () => {
 			(monitorsRepository.findAll as jest.Mock).mockResolvedValue(monitors);
 
 			await queue.init();
-			jest.runAllTimers();
+			// advance past the <100ms seed offsets
+			jest.advanceTimersByTime(100);
 
 			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m1", template: "monitor-job" }));
 			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m2", template: "monitor-job" }));
@@ -454,7 +475,7 @@ describe("LessSimpleQueue", () => {
 		it("returns empty metrics when no jobs", async () => {
 			const { queue } = createQueue();
 			const metrics = await queue.getMetrics();
-			expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0 });
+			expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0, workers: [] });
 		});
 
 		it("aggregates metrics from jobs", async () => {
@@ -478,6 +499,9 @@ describe("LessSimpleQueue", () => {
 					failCount: 0,
 					lastFailedAt: 0,
 					lastFinishedAt: 50,
+					// live lock held by a worker
+					lockedBy: "host-1:1234:abcd",
+					lockedUntil: FAR_FUTURE,
 					lockedAt: 999,
 					data: { url: "http://b.com", type: "ping" },
 				},
@@ -526,6 +550,56 @@ describe("LessSimpleQueue", () => {
 		});
 	});
 
+	// ── getMetrics: activeJobs (in-flight) ───────────────────────────────────────
+
+	describe("getMetrics - activeJobs", () => {
+		it("counts jobs held by a live lock as active", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([
+				{ id: "m1", template: "monitor-job", lockedBy: "worker-a", lockedUntil: FAR_FUTURE, lockedAt: 100, data: { url: "a" } },
+				{ id: "m2", template: "monitor-job", lockedBy: "worker-b", lockedUntil: FAR_FUTURE, lockedAt: 250, data: { url: "b" } },
+			]);
+			const metrics = await queue.getMetrics();
+			expect(metrics.activeJobs).toBe(2);
+		});
+
+		it("does not count unlocked jobs (lockedBy null)", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([
+				{ id: "m1", template: "monitor-job", lockedBy: null, lockedUntil: null, lockedAt: null, data: { url: "a" } },
+			]);
+			const metrics = await queue.getMetrics();
+			expect(metrics.activeJobs).toBe(0);
+		});
+
+		it("does not count jobs whose lock has expired (lockedUntil <= now)", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([
+				{ id: "m1", template: "monitor-job", lockedBy: "worker-a", lockedUntil: 1, lockedAt: 1, data: { url: "a" } },
+			]);
+			const metrics = await queue.getMetrics();
+			expect(metrics.activeJobs).toBe(0);
+		});
+	});
+
+	// ── getMetrics: worker registry ──────────────────────────────────────────────
+
+	describe("getMetrics - worker registry", () => {
+		it("returns the alive workers from the registry", async () => {
+			const { queue, workersRepository } = createQueue();
+			const alive = [
+				{ workerId: "host:1:aaa", mode: "primary", lastSeenAt: 1000 },
+				{ workerId: "host:2:bbb", mode: "worker", lastSeenAt: 2000 },
+			];
+			(workersRepository.findRecent as jest.Mock).mockResolvedValue(alive);
+
+			const metrics = await queue.getMetrics();
+
+			expect(workersRepository.findRecent).toHaveBeenCalledWith(30000);
+			expect(metrics.workers).toEqual(alive);
+		});
+	});
+
 	// ── getJobs ────────────────────────────────────────────────────────────────
 
 	describe("getJobs", () => {
@@ -559,6 +633,8 @@ describe("LessSimpleQueue", () => {
 				monitorActive: true,
 				active: true,
 				repeat: 60000,
+				lockedBy: null,
+				lockedUntil: null,
 				lockedAt: null,
 				runCount: 5,
 				failCount: 1,
@@ -611,6 +687,29 @@ describe("LessSimpleQueue", () => {
 			expect(result.success).toBe(true);
 		});
 
+		it("flushes before stopping, so the store is still connected for removeAll", async () => {
+			const { queue, scheduler } = createQueue();
+			const order: string[] = [];
+			(scheduler.flushJobs as jest.Mock).mockImplementation(async () => {
+				order.push("flush");
+				return true;
+			});
+			(scheduler.stop as jest.Mock).mockImplementation(async () => {
+				order.push("stop");
+				return true;
+			});
+			await queue.flushQueues();
+			expect(order).toEqual(["flush", "stop"]);
+		});
+
+		it("refuses to flush on a non-primary worker without touching the scheduler", async () => {
+			const { queue, scheduler } = createQueue({ queueMode: "worker" });
+			const result = await queue.flushQueues();
+			expect(result.success).toBe(false);
+			expect(scheduler.flushJobs).not.toHaveBeenCalled();
+			expect(scheduler.stop).not.toHaveBeenCalled();
+		});
+
 		it("returns false when flush fails", async () => {
 			const { queue, scheduler } = createQueue();
 			(scheduler.flushJobs as jest.Mock).mockResolvedValue(false);
@@ -628,8 +727,16 @@ describe("LessSimpleQueue", () => {
 			const logger = createMockLogger();
 			const helper = createQueueHelper();
 			const monitorsRepository = createMonitorsRepo();
+			const workersRepository = createWorkersRepo();
 
-			const instance = await LessSimpleQueue.create(logger as any, helper as any, monitorsRepository as any, envSettings, "primary");
+			const instance = await LessSimpleQueue.create(
+				logger as any,
+				helper as any,
+				monitorsRepository as any,
+				workersRepository as any,
+				envSettings,
+				"primary"
+			);
 
 			expect(MockMongoStore).toHaveBeenCalledWith(expect.objectContaining({ url: "mongodb://localhost:27017/test_db" }));
 			expect(MockScheduler).toHaveBeenCalled();
@@ -642,11 +749,66 @@ describe("LessSimpleQueue", () => {
 			const logger = createMockLogger();
 			const helper = createQueueHelper();
 			const monitorsRepository = createMonitorsRepo();
+			const workersRepository = createWorkersRepo();
 
-			const instance = await LessSimpleQueue.create(logger as any, helper as any, monitorsRepository as any, envSettings, "worker");
+			const instance = await LessSimpleQueue.create(
+				logger as any,
+				helper as any,
+				monitorsRepository as any,
+				workersRepository as any,
+				envSettings,
+				"worker"
+			);
 
 			expect(instance).toBeInstanceOf(LessSimpleQueue);
 			expect(monitorsRepository.findAll).not.toHaveBeenCalled();
 		});
 	});
+
+	// ── worker heartbeat / deregistration ────────────────────────────────────────
+
+	describe("worker heartbeat", () => {
+		it("writes a heartbeat on init with the workerId and mode (primary)", async () => {
+			const { queue, workersRepository } = createQueue();
+			await queue.init("primary");
+			expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "primary");
+		});
+
+		it("writes a heartbeat on init with the workerId and mode (worker)", async () => {
+			const { queue, workersRepository } = createQueue();
+			await queue.init("worker");
+			expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "worker");
+		});
+
+		it("beats again on each scheduler:heartbeat event", async () => {
+			const { queue, scheduler, workersRepository } = createQueue();
+			await queue.init("worker");
+			(workersRepository.upsert as jest.Mock).mockClear();
+
+			scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd");
+			scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd");
+
+			expect(workersRepository.upsert).toHaveBeenCalledTimes(2);
+			expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "worker");
+		});
+
+		it("stops beating after shutdown", async () => {
+			const { queue, scheduler, workersRepository } = createQueue();
+			await queue.init("worker");
+			await queue.shutdown();
+			(workersRepository.upsert as jest.Mock).mockClear();
+
+			scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd");
+
+			expect(workersRepository.upsert).not.toHaveBeenCalled();
+		});
+
+		it("deregisters the worker and stops the scheduler on shutdown", async () => {
+			const { queue, scheduler, workersRepository } = createQueue();
+			await queue.init("worker");
+			await queue.shutdown();
+			expect(workersRepository.deleteById).toHaveBeenCalledWith("host-1:1234:abcd");
+			expect(scheduler.stop).toHaveBeenCalled();
+		});
+	});
 });

server/test/unit/services/superSimpleQueue.test.ts

@@ -486,10 +486,25 @@ describe("SuperSimpleQueue", () => {
 	// ── getMetrics ───────────────────────────────────────────────────────────
 
 	describe("getMetrics", () => {
-		it("returns empty metrics when no jobs", async () => {
+		it("returns empty job metrics plus this process as the sole worker when no jobs", async () => {
 			const { queue } = createQueue();
 			const metrics = await queue.getMetrics();
-			expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0 });
+			expect(metrics).toEqual({
+				jobs: 0,
+				activeJobs: 0,
+				failingJobs: 0,
+				jobsWithFailures: [],
+				totalRuns: 0,
+				totalFailures: 0,
+				workers: [
+					{
+						// hostname:pid:uuid, matching less-simple-scheduler's workerId format
+						workerId: expect.stringMatching(/^.+:\d+:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/),
+						mode: "primary",
+						lastSeenAt: expect.any(Number),
+					},
+				],
+			});
 		});
 
 		it("aggregates metrics from jobs", async () => {
@@ -580,6 +595,8 @@ describe("SuperSimpleQueue", () => {
 				monitorActive: true,
 				active: true,
 				repeat: 60000,
+				lockedBy: null,
+				lockedUntil: null,
 				lockedAt: null,
 				runCount: 5,
 				failCount: 1,