From d2461ef118a7766ee210050c93ef06554e85b307 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 15 Jun 2026 10:39:48 -0700 Subject: [PATCH] remove old queues --- server/package-lock.json | 113 ----- server/package.json | 2 - server/src/config/services.ts | 104 +---- .../queue-workers/queue-worker.model.ts | 18 - .../queue-worker.repository.interface.ts | 8 - .../queue-worker.repository.mongo.ts | 33 -- .../domain/queue-workers/queue-worker.type.ts | 18 - server/src/index.ts | 2 +- server/src/worker/worker.db-queue.ts | 22 +- server/src/worker/worker.less-simple.ts | 427 ------------------ server/src/worker/worker.super-simple.ts | 384 ---------------- 11 files changed, 29 insertions(+), 1102 deletions(-) delete mode 100644 server/src/domain/queue-workers/queue-worker.model.ts delete mode 100644 server/src/domain/queue-workers/queue-worker.repository.interface.ts delete mode 100644 server/src/domain/queue-workers/queue-worker.repository.mongo.ts delete mode 100644 server/src/domain/queue-workers/queue-worker.type.ts delete mode 100644 server/src/worker/worker.less-simple.ts delete mode 100644 server/src/worker/worker.super-simple.ts diff --git a/server/package-lock.json b/server/package-lock.json index e6bd58c1b..7c69b6c1e 100644 --- a/server/package-lock.json +++ b/server/package-lock.json @@ -31,7 +31,6 @@ "jmespath": "^0.16.0", "jsdom": "^26.1.0", "jsonwebtoken": "9.0.2", - "less-simple-scheduler": "2.6.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", @@ -40,7 +39,6 @@ "ping": "0.4.4", "sharp": "0.33.5", "ssl-checker": "2.0.10", - "super-simple-scheduler": "1.9.2", "swagger-ui-express": "5.0.1", "winston": "^3.13.0", "ws": "^8.19.0", @@ -10608,95 +10606,6 @@ "integrity": "sha512-Xq9nH7KlWZmXAtodXDDRE7vs6DU1gTU8zYDHDiWLSip45Egwq3plLHzPn27NgvzL2r1LMPC1vdqh98sQxtqj4A==", "license": "MIT" }, - "node_modules/less-simple-scheduler": { - "version": "2.6.0", - "resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.6.0.tgz", - "integrity": "sha512-Vi8ouiodLqe2fyh+DqCmxE5cCJZDI3xOyUAH4jcj9sf7GunLUzH71PBcYJy2P8lxTK3LxEkHtr1reymiogs6JQ==", - "license": "ISC", - "dependencies": { - "mongodb": "7.2.0" - }, - "engines": { - "node": ">=18" - } - }, - "node_modules/less-simple-scheduler/node_modules/@types/whatwg-url": { - "version": "13.0.0", - "resolved": "https://registry.npmjs.org/@types/whatwg-url/-/whatwg-url-13.0.0.tgz", - "integrity": "sha512-N8WXpbE6Wgri7KUSvrmQcqrMllKZ9uxkYWMt+mCSGwNc0Hsw9VQTW7ApqI4XNrx6/SaM2QQJCzMPDEXE058s+Q==", - "license": "MIT", - "dependencies": { - "@types/webidl-conversions": "*" - } - }, - "node_modules/less-simple-scheduler/node_modules/bson": { - "version": "7.2.0", - "resolved": "https://registry.npmjs.org/bson/-/bson-7.2.0.tgz", - "integrity": "sha512-YCEo7KjMlbNlyHhz7zAZNDpIpQbd+wOEHJYezv0nMYTn4x31eIUM2yomNNubclAt63dObUzKHWsBLJ9QcZNSnQ==", - "license": "Apache-2.0", - "engines": { - "node": ">=20.19.0" - } - }, - "node_modules/less-simple-scheduler/node_modules/mongodb": { - "version": "7.2.0", - "resolved": "https://registry.npmjs.org/mongodb/-/mongodb-7.2.0.tgz", - "integrity": "sha512-F/2+BMZtLVhY30ioZp0dAmZ+IRZMBqI+nrv6t5+9/1AIwCa8sMRC3jBf81lpxMhnZgqq8CoUD503Z1oZWq1/sw==", - "license": "Apache-2.0", - "dependencies": { - "@mongodb-js/saslprep": "^1.3.0", - "bson": "^7.2.0", - "mongodb-connection-string-url": "^7.0.0" - }, - "engines": { - "node": ">=20.19.0" - }, - "peerDependencies": { - "@aws-sdk/credential-providers": "^3.806.0", - "@mongodb-js/zstd": "^7.0.0", - "gcp-metadata": "^7.0.1", - "kerberos": "^7.0.0", - "mongodb-client-encryption": ">=7.0.0 <7.1.0", - "snappy": "^7.3.2", - "socks": "^2.8.6" - }, - "peerDependenciesMeta": { - "@aws-sdk/credential-providers": { - "optional": true - }, - "@mongodb-js/zstd": { - "optional": true - }, - "gcp-metadata": { - "optional": true - }, - "kerberos": { - "optional": true - }, - "mongodb-client-encryption": { - "optional": true - }, - "snappy": { - "optional": true - }, - "socks": { - "optional": true - } - } - }, - "node_modules/less-simple-scheduler/node_modules/mongodb-connection-string-url": { - "version": "7.0.1", - "resolved": "https://registry.npmjs.org/mongodb-connection-string-url/-/mongodb-connection-string-url-7.0.1.tgz", - "integrity": "sha512-h0AZ9A7IDVwwHyMxmdMXKy+9oNlF0zFoahHiX3vQ8e3KFcSP3VmsmfvtRSuLPxmyv2vjIDxqty8smTgie/SNRQ==", - "license": "Apache-2.0", - "dependencies": { - "@types/whatwg-url": "^13.0.0", - "whatwg-url": "^14.1.0" - }, - "engines": { - "node": ">=20.19.0" - } - }, "node_modules/leven": { "version": "3.1.0", "resolved": "https://registry.npmjs.org/leven/-/leven-3.1.0.tgz", @@ -14754,28 +14663,6 @@ "postcss": "^8.4.32" } }, - "node_modules/super-simple-scheduler": { - "version": "1.9.2", - "resolved": "https://registry.npmjs.org/super-simple-scheduler/-/super-simple-scheduler-1.9.2.tgz", - "integrity": "sha512-dfeI1R8jFLeHXIrNsDRxqvdJN2r4BjwKrvv/pPThu0X4RQa008/ubkFno96esHad04J6KyWYTK+P/qgOcoaReg==", - "license": "MIT", - "dependencies": { - "uuid": "11.1.1" - } - }, - "node_modules/super-simple-scheduler/node_modules/uuid": { - "version": "11.1.1", - "resolved": "https://registry.npmjs.org/uuid/-/uuid-11.1.1.tgz", - "integrity": "sha512-vIYxrBCC/N/K+Js3qSN88go7kIfNPssr/hHCesKCQNAjmgvYS2oqr69kIufEG+O4+PfezOH4EbIeHCfFov8ZgQ==", - "funding": [ - "https://github.com/sponsors/broofa", - "https://github.com/sponsors/ctavan" - ], - "license": "MIT", - "bin": { - "uuid": "dist/esm/bin/uuid" - } - }, "node_modules/supports-color": { "version": "7.2.0", "resolved": "https://registry.npmjs.org/supports-color/-/supports-color-7.2.0.tgz", diff --git a/server/package.json b/server/package.json index 66ca955eb..8341bfb2c 100755 --- a/server/package.json +++ b/server/package.json @@ -53,7 +53,6 @@ "jmespath": "^0.16.0", "jsdom": "^26.1.0", "jsonwebtoken": "9.0.2", - "less-simple-scheduler": "2.6.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", @@ -62,7 +61,6 @@ "ping": "0.4.4", "sharp": "0.33.5", "ssl-checker": "2.0.10", - "super-simple-scheduler": "1.9.2", "swagger-ui-express": "5.0.1", "winston": "^3.13.0", "ws": "^8.19.0", diff --git a/server/src/config/services.ts b/server/src/config/services.ts index e7741f378..74a73ad7e 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -30,8 +30,6 @@ import { EmailService, IEmailService } from "@/service/emailService.js"; import { GlobalPingService } from "@/service/globalPingService.js"; import { WorkerHelper } from "@/worker/worker.helper.js"; import { IWorker } from "@/worker/worker.interface.js"; -import { LessSimpleWorker } from "@/worker/worker.less-simple.js"; -import { SuperSimpleQueue } from "@/worker/worker.super-simple.js"; import { INetworkService, NetworkService } from "@/service/networkService.js"; import { IStatusService, StatusService } from "@/service/statusService.js"; import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; @@ -89,7 +87,6 @@ import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interf import MongoMonitorsRepository from "@/domain/monitors/monitor.repository.mongo.js"; import { INotificationsRepository } from "@/domain/notifications/notification.repository.interface.js"; import MongoNotificationsRepository from "@/domain/notifications/notification.repository.mongo.js"; -import MongoQueueWorkersRepository from "@/domain/queue-workers/queue-worker.repository.mongo.js"; import { IRecoveryTokensRepository } from "@/domain/recovery-tokens/recovery-token.repository.interface.js"; import MongoRecoveryTokensRepository from "@/domain/recovery-tokens/recovery-token.repository.mongo.js"; import { IStatusPagesRepository } from "@/domain/status-pages/status-page-repository.interface.js"; @@ -101,7 +98,6 @@ import MongoTeamsRepository from "@/domain/teams/team.repository.model.js"; import { IUsersRepository } from "@/domain/users/user.repository.interface.js"; import MongoUsersRepository from "@/domain/users/user.repository.mongo.js"; import { ILogger } from "@/utils/logger.js"; -import { AppError } from "@/utils/AppError.js"; import { type QueueMode } from "@/domain/app-settings/app-settings.type.js"; import { NotificationReactor } from "@/worker/reactors/reactor.notification.js"; import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; @@ -155,12 +151,10 @@ export const initializeServices = async ({ logger, envSettings, settingsService, - queueMode = "primary", }: { logger: ILogger; envSettings: EnvConfig; settingsService: ISettingsService; - queueMode?: QueueMode; }): Promise => { // Create DB @@ -168,58 +162,6 @@ export const initializeServices = async ({ db = new MongoDB(logger, envSettings); await db.connect(); - // ** NOTE ** - // const dbType = envSettings.dbType; - // DB type has been fixed to MongoDB for now - - // if (dbType === "mongodb") { - // db = new MongoDB(logger, envSettings); - // } - - // if (!db) { - // throw new AppError({ message: "Unsupported database type", status: 500 }); - // } - - // await db.connect(); - - // Repositories - - // ** NOTE ** - // DB Type fixed to MongoDB for now - // let monitorsRepository: IMonitorsRepository; - // let checksRepository: IChecksRepository; - // let geoChecksRepository: IGeoChecksRepository; - // let monitorStatsRepository: IMonitorStatsRepository; - // let statusPagesRepository: IStatusPagesRepository; - // let usersRepository: IUsersRepository; - // let invitesRepository: IInvitesRepository; - // let recoveryTokensRepository: IRecoveryTokensRepository; - // let settingsRepository: ISettingsRepository; - // let notificationsRepository: INotificationsRepository; - // let tagsRepository: ITagsRepository; - // let incidentsRepository: IIncidentsRepository; - // let teamsRepository: ITeamsRepository; - // let maintenanceWindowsRepository: IMaintenanceWindowsRepository; - - // if (dbType === "mongodb") { - // monitorsRepository = new MongoMonitorsRepository(); - // checksRepository = new MongoChecksRepository(logger); - // geoChecksRepository = new MongoGeoChecksRepository(logger); - // monitorStatsRepository = new MongoMonitorStatsRepository(); - // statusPagesRepository = new MongoStatusPagesRepository(); - // usersRepository = new MongoUsersRepository(); - // invitesRepository = new MongoInvitesRepository(); - // recoveryTokensRepository = new MongoRecoveryTokensRepository(); - // settingsRepository = new MongoSettingsRepository(); - // notificationsRepository = new MongoNotificationsRepository(); - // tagsRepository = new MongoTagsRepository(); - // incidentsRepository = new MongoIncidentsRepository(); - // teamsRepository = new MongoTeamsRepository(); - // maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository(); - // } else { - // throw new AppError({ message: "Unsupported database type", status: 500 }); - // } - const monitorsRepository = new MongoMonitorsRepository(); const checksRepository = new MongoChecksRepository(logger); const geoChecksRepository = new MongoGeoChecksRepository(logger); @@ -234,7 +176,6 @@ export const initializeServices = async ({ const incidentsRepository = new MongoIncidentsRepository(); const teamsRepository = new MongoTeamsRepository(); const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository(); - const queueWorkersRepository = new MongoQueueWorkersRepository(); const jobsRepository = new MongoJobsRepository(); // Inject settings repository into settings service (now that DB is connected) @@ -349,38 +290,19 @@ export const initializeServices = async ({ geoCheckPipeline ); - if (queueMode === "worker" && envSettings.queueType !== "lessSimpleQueue") { - throw new AppError({ - message: `Worker mode requires QUEUE_TYPE="lessSimpleQueue" got ${envSettings.queueType}`, - status: 500, - service: "config services", - method: "initializeServices", - }); - } - - let worker: IWorker; - switch (envSettings.queueType) { - case "lessSimpleQueue": - worker = await LessSimpleWorker.create(logger, workerHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode); - break; - case "dbQueue": - worker = await DBQueueWorker.create( - logger, - jobsRepository, - monitorsRepository, - checksRepository, - checkService, - checkProducer, - checkEvaluator, - geoCheckPipeline, - reactorDispatcher, - workerHelper - ); - break; - - default: - worker = await SuperSimpleQueue.create(logger, workerHelper, monitorsRepository); - } + const worker = await DBQueueWorker.create( + logger, + jobsRepository, + monitorsRepository, + checksRepository, + checkService, + checkProducer, + checkEvaluator, + geoCheckPipeline, + reactorDispatcher, + workerHelper, + envSettings.queueMode + ); // Business services const userService = new UserService({ diff --git a/server/src/domain/queue-workers/queue-worker.model.ts b/server/src/domain/queue-workers/queue-worker.model.ts deleted file mode 100644 index 06299d339..000000000 --- a/server/src/domain/queue-workers/queue-worker.model.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { Schema, model } from "mongoose"; -import { QueueModes } from "@/domain/app-settings/app-settings.type.js"; -import { QueueWorkerDocument } from "@/domain/queue-workers/queue-worker.type.js"; - -const QueueWorkerSchema = new Schema( - { - _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("QueueWorker", QueueWorkerSchema); - -export type { QueueWorkerDocument }; -export { QueueWorkerModel }; -export default QueueWorkerModel; diff --git a/server/src/domain/queue-workers/queue-worker.repository.interface.ts b/server/src/domain/queue-workers/queue-worker.repository.interface.ts deleted file mode 100644 index e4522c916..000000000 --- a/server/src/domain/queue-workers/queue-worker.repository.interface.ts +++ /dev/null @@ -1,8 +0,0 @@ -import type { QueueMode } from "@/domain/app-settings/app-settings.type.js"; -import type { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js"; - -export interface IQueueWorkersRepository { - upsert(workerId: string, mode: QueueMode): Promise; - findRecent(maxAgeMs: number): Promise; - deleteById(workerId: string): Promise; -} diff --git a/server/src/domain/queue-workers/queue-worker.repository.mongo.ts b/server/src/domain/queue-workers/queue-worker.repository.mongo.ts deleted file mode 100644 index 35565bd03..000000000 --- a/server/src/domain/queue-workers/queue-worker.repository.mongo.ts +++ /dev/null @@ -1,33 +0,0 @@ -import type { QueueMode } from "@/domain/app-settings/app-settings.type.js"; -import type { IQueueWorkersRepository } from "./queue-worker.repository.interface.js"; -import type { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js"; -import { QueueWorkerModel, type QueueWorkerDocument } from "@/domain/queue-workers/queue-worker.model.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 => { - await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true }); - }; - - findRecent = async (maxAgeMs: number): Promise => { - 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 => { - await QueueWorkerModel.deleteOne({ _id: workerId }); - }; -} - -export default MongoQueueWorkersRepository; diff --git a/server/src/domain/queue-workers/queue-worker.type.ts b/server/src/domain/queue-workers/queue-worker.type.ts deleted file mode 100644 index a83c831ad..000000000 --- a/server/src/domain/queue-workers/queue-worker.type.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { type QueueMode } from "@/domain/app-settings/app-settings.type.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. -export interface QueueWorkerDocument { - _id: string; - mode: QueueMode; - lastSeenAt: Date; - createdAt: Date; - updatedAt: Date; -} - -export type QueueWorker = { - workerId: string; // hostname:pid:uuid - mode: QueueMode; - lastSeenAt: number; // epoch ms of the last heartbeat -}; diff --git a/server/src/index.ts b/server/src/index.ts index cd3a39cdd..37d4e59d8 100755 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -24,7 +24,7 @@ const startApp = async () => { logger = new Logger({ envSettings }); // Initialize services (connects DB, creates repositories, injects settingsRepository) - const services = await initializeServices({ logger, envSettings, settingsService, queueMode: envSettings.queueMode }); + const services = await initializeServices({ logger, envSettings, settingsService }); // If this is a worker instance, we're done. No need for express if (envSettings.queueMode === "worker") { diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index c333491df..62d2ca7e5 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -13,6 +13,7 @@ import { ICheckEvaluator } from "@/worker/worker.check-evaluator.js"; import { ICheckPipeline } from "@/worker/worker.check-pipeline.js"; import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; import { IWorkerHelper } from "@/worker/worker.helper.js"; +import { QueueMode, QueueModes } from "@/domain/app-settings/app-settings.type.js"; const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; @@ -47,7 +48,8 @@ export class DBQueueWorker implements IWorker { private checkEvaluator: ICheckEvaluator, private geoCheckPipeline: ICheckPipeline, private dispatcher: IReactorDispatcher, - private helper: IWorkerHelper + private helper: IWorkerHelper, + private mode: QueueMode ) {} get serviceName() { @@ -64,7 +66,8 @@ export class DBQueueWorker implements IWorker { checkEvaluator: ICheckEvaluator, geoCheckPipeline: ICheckPipeline, dispatcher: IReactorDispatcher, - helper: IWorkerHelper + helper: IWorkerHelper, + mode: QueueMode ): Promise { const instance = new DBQueueWorker( logger, @@ -76,7 +79,8 @@ export class DBQueueWorker implements IWorker { checkEvaluator, geoCheckPipeline, dispatcher, - helper + helper, + mode ); await instance.init(); return instance; @@ -242,9 +246,13 @@ export class DBQueueWorker implements IWorker { init = async () => { this.stopped = false; - await this.reconcile(); - for (const type of Object.keys(this.inFlight) as JobType[]) { - this.startLoop(type); + if (this.mode === "primary") { + await this.reconcile(); + return true; + } else { + for (const type of Object.keys(this.inFlight) as JobType[]) { + this.startLoop(type); + } } return true; }; @@ -307,7 +315,7 @@ export class DBQueueWorker implements IWorker { jobsWithFailures: [], totalRuns: 0, totalFailures: 0, - workers: [{ workerId: this.workerId, mode: "primary", lastSeenAt: now }], + workers: [{ workerId: this.workerId, mode: this.mode, lastSeenAt: now }], } ); }; diff --git a/server/src/worker/worker.less-simple.ts b/server/src/worker/worker.less-simple.ts deleted file mode 100644 index bd4752c6e..000000000 --- a/server/src/worker/worker.less-simple.ts +++ /dev/null @@ -1,427 +0,0 @@ -import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; -import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; -import { ILogger } from "@/utils/logger.js"; -import { MongoStore, Scheduler, IJob, AddJobInput } from "less-simple-scheduler"; -import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; -import { IWorkerHelper } from "@/worker/worker.helper.js"; -import { type WorkerMetrics, type WorkerJobsPagination, type WorkerJobsPage, IWorker } from "@/worker/worker.interface.js"; -import { type QueueMode } from "@/domain/app-settings/app-settings.type.js"; -import { EnvConfig } from "@/domain/app-settings/app-settings.service.js"; -const SERVICE_NAME = "JobQueue"; - -// How long after its last heartbeat a worker is still considered alive. Must be longer than the scheduler's heartbeat -const WORKER_TTL_MS = 30_000; - -// For discriminating job types -type MonitorJob = Omit & { data: Monitor }; -const MONITOR_TEMPLATES = new Set(["monitor-job", "geo-check-job"]); -const isMonitorJob = (job: IJob): job is MonitorJob => MONITOR_TEMPLATES.has(job.template); - -export class LessSimpleWorker implements IWorker { - static SERVICE_NAME = SERVICE_NAME; - - private logger: ILogger; - private helper: IWorkerHelper; - private monitorsRepository: IMonitorsRepository; - private workersRepository: IQueueWorkersRepository; - private readonly scheduler: Scheduler; - private queueMode: QueueMode; - private heartbeatListener: ((workerId: string) => void) | null = null; - - constructor( - logger: ILogger, - helper: IWorkerHelper, - 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(); - } - - get serviceName() { - return LessSimpleWorker.SERVICE_NAME; - } - - private registerListeners = () => { - this.scheduler.on("scheduler:start", () => { - this.logger.info({ - message: "Scheduler started", - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("scheduler:stop", () => { - this.logger.info({ - message: "Scheduler stopped", - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("scheduler:error", (workerId, error) => { - this.logger.error({ - message: `${workerId} error: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - stack: error instanceof Error ? error.stack : undefined, - }); - }); - - this.scheduler.on("job:locked", (workerId, job) => { - this.logger.debug({ - message: `${workerId} ${job.id} locked`, - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("job:abort", (workerId, job, reason) => { - this.logger.warn({ - message: `${workerId} ${job.id} aborted: ${reason}`, - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("job:attempt", (workerId, job, attempt) => { - this.logger.debug({ - message: `${workerId} ${job.id} attempt ${attempt}`, - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("job:complete", (workerId, job) => { - this.logger.debug({ - message: `${workerId} ${job.id} completed successfully`, - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("job:exhausted", (workerId, job, error) => { - this.logger.error({ - message: `${workerId} ${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - stack: error instanceof Error ? error.stack : undefined, - }); - }); - - this.scheduler.on("job:fail", (workerId, job, error, attempt) => { - this.logger.warn({ - message: `${workerId} ${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - stack: error instanceof Error ? error.stack : undefined, - }); - }); - - this.scheduler.on("job:start", (workerId, job) => { - this.logger.debug({ - message: `${workerId} ${job.id} started`, - service: SERVICE_NAME, - }); - }); - }; - - static async create( - logger: ILogger, - helper: IWorkerHelper, - 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: envSettings.queueMode === "primary" ? 0 : 50, - processEvery: 100, - }); - const instance = new LessSimpleWorker(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(); - this.scheduler.addTemplate("monitor-job", this.helper.getHeartbeatJob()); - this.scheduler.addTemplate("geo-check-job", this.helper.getHeartbeatGeoJob()); - 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({ - message: `Worker queue: ${this.scheduler.workerId} initialized`, - service: SERVICE_NAME, - method: "init", - }); - return true; - } - - const monitors = await this.monitorsRepository.findAll(); - if (!monitors) { - return true; - } - const jobInputs = monitors.flatMap((monitor) => this.buildJobInputs(monitor)); - if (jobInputs.length > 0) { - await this.scheduler.addJobs(jobInputs); - } - - this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true, upsert: true }); - this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000, upsert: true }); - - return true; - } catch (error: unknown) { - this.logger.error({ - message: `Failed to initialize LessSimpleQueue: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - method: "init", - }); - return false; - } - }; - - // Builds the scheduler inputs for a monitor: always a monitor-job, plus a - // geo-check job when the monitor supports and enables geo checks. - private buildJobInputs = (monitor: Monitor): AddJobInput[] => { - const inputs: AddJobInput[] = [ - { - id: monitor.id, - template: "monitor-job", - repeat: monitor.interval, - active: monitor.isActive, - data: monitor, - upsert: true, - jitter: true, - }, - ]; - - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { - inputs.push({ - id: `${monitor.id}-geo`, - template: "geo-check-job", - repeat: monitor.geoCheckInterval, - active: monitor.isActive, - data: monitor, - upsert: true, - jitter: true, - }); - } - - return inputs; - }; - - addJob = async (_monitorId: string, monitor: Monitor) => { - await this.scheduler.addJobs(this.buildJobInputs(monitor)); - }; - - deleteJob = async (monitor: Monitor) => { - this.scheduler.removeJob(monitor.id); - const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); - if (geoJob) await this.scheduler.removeJob(`${monitor.id}-geo`); - }; - - pauseJob = async (monitor: Monitor) => { - const result = await this.scheduler.pauseJob(monitor.id); - if (result === false) { - throw new Error("Failed to pause monitor"); - } - this.scheduler.updateJob(monitor.id, { data: monitor }); - - const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); - if (geoJob) { - const geoRes = await this.scheduler.pauseJob(`${monitor.id}-geo`); - if (geoRes === false) { - this.logger.error({ - message: `Failed to pause geo check job for monitor ${monitor.id}`, - service: SERVICE_NAME, - method: "pauseJob", - }); - } - this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); - } - - this.logger.debug({ - message: `Paused monitor ${monitor.id}`, - service: SERVICE_NAME, - method: "pauseJob", - }); - }; - - resumeJob = async (monitor: Monitor) => { - const result = await this.scheduler.resumeJob(monitor.id); - if (result === false) { - throw new Error("Failed to resume monitor"); - } - this.scheduler.updateJob(monitor.id, { data: monitor }); - - const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); - if (geoJob) { - const geoRes = await this.scheduler.resumeJob(`${monitor.id}-geo`); - if (geoRes === false) { - this.logger.error({ - message: `Failed to resume geo check job for monitor ${monitor.id}`, - service: SERVICE_NAME, - method: "resumeJob", - }); - } - this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); - } - this.logger.debug({ - message: `Resumed monitor ${monitor.id}`, - service: SERVICE_NAME, - method: "resumeJob", - }); - }; - - private syncGeoJob = async (monitor: Monitor) => { - const geoJobId = `${monitor.id}-geo`; - const existingGeoJob = await this.scheduler.getJob(geoJobId); - - // If geoChecks have been disabled, or the monitor type doesn't support them, remove - if (!monitor.geoCheckEnabled || !supportsGeoCheck(monitor.type)) { - if (existingGeoJob) this.scheduler.removeJob(geoJobId); - return; - } - - // If the job exists, update it - if (existingGeoJob) { - this.scheduler.updateJob(geoJobId, { repeat: monitor.geoCheckInterval, active: monitor.isActive, data: monitor }); - return; - } - - // Otherwise, create it - this.scheduler.addJob({ - id: geoJobId, - template: "geo-check-job", - repeat: monitor.geoCheckInterval, - active: monitor.isActive, - data: monitor, - upsert: true, - }); - }; - - updateJob = async (monitor: Monitor) => { - this.scheduler.updateJob(monitor.id, { repeat: monitor.interval, data: monitor }); - await this.syncGeoJob(monitor); - }; - - 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 (): Promise => { - const stats = await this.scheduler.getStats(); - - const jobsWithFailures = stats.jobsWithFailures.map((failure) => { - // `data` is untyped at the store level: for monitor jobs it's the Monitor, - // for system jobs (e.g. cleanup) it has no url/type => null - const monitor = failure.data as Monitor | undefined; - return { - monitorId: failure.id, - monitorUrl: monitor?.url ?? null, - monitorType: monitor?.type ?? null, - failedAt: failure.failedAt, - failCount: failure.failCount, - failReason: failure.failReason, - }; - }); - - const workers = await this.workersRepository.findRecent(WORKER_TTL_MS); - return { - jobs: stats.jobs, - activeJobs: stats.activeJobs, - failingJobs: stats.failingJobs, - totalRuns: stats.totalRuns, - totalFailures: stats.totalFailures, - jobsWithFailures, - workers, - }; - }; - - getJobs = async ({ page = 0, rowsPerPage = 0 }: WorkerJobsPagination): Promise => { - const listOptions = rowsPerPage > 0 ? { skip: Math.max(page, 0) * rowsPerPage, limit: rowsPerPage } : undefined; - const [jobs, count] = await Promise.all([this.scheduler.getJobs(listOptions), this.scheduler.countJobs()]); - const mapped = jobs.map((job) => { - const monitor = isMonitorJob(job) ? job.data : undefined; - - return { - monitorId: job.id, - monitorUrl: monitor?.url || null, - monitorType: monitor?.type || null, - monitorInterval: monitor?.interval || null, - monitorGeoInterval: monitor?.geoCheckInterval || null, - 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, - failReason: job.lastError ?? null, - lastRunAt: job.lastStartedAt ?? null, - lastFinishedAt: job.lastFinishedAt ?? null, - lastRunTook: job.lockedAt || !job.lastFinishedAt || !job.lastStartedAt ? null : job.lastFinishedAt - job.lastStartedAt, - lastFailedAt: job.lastFailedAt ?? null, - }; - }); - - return { jobs: mapped, count }; - }; - - flushQueues = async () => { - 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), - }; - }; -} diff --git a/server/src/worker/worker.super-simple.ts b/server/src/worker/worker.super-simple.ts deleted file mode 100644 index 86b9f152a..000000000 --- a/server/src/worker/worker.super-simple.ts +++ /dev/null @@ -1,384 +0,0 @@ -import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; -import { ILogger } from "@/utils/logger.js"; -import Scheduler from "super-simple-scheduler"; -import { Monitor, MonitorType, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; -import { hostname } from "node:os"; -import { randomUUID } from "node:crypto"; -import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; -import { IWorker, WorkerJobSummary } from "@/worker/worker.interface.js"; -import { IWorkerHelper } from "@/worker/worker.helper.js"; - -const SERVICE_NAME = "JobQueue"; - -export class SuperSimpleQueue implements IWorker { - static SERVICE_NAME = SERVICE_NAME; - - private logger: ILogger; - private helper: IWorkerHelper; - private workerId: string; - private monitorsRepository: IMonitorsRepository; - private readonly scheduler: Scheduler; - - constructor(logger: ILogger, helper: IWorkerHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { - this.logger = logger; - this.helper = helper; - this.monitorsRepository = monitorsRepository; - this.scheduler = scheduler; - this.registerListeners(); - this.workerId = `${hostname()}:${process.pid}:${randomUUID()}`; - } - - get serviceName() { - return SuperSimpleQueue.SERVICE_NAME; - } - - private registerListeners = () => { - this.scheduler.on("scheduler:start", () => { - this.logger.info({ - message: "Scheduler started", - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("scheduler:stop", () => { - this.logger.info({ - message: "Scheduler stopped", - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("scheduler:error", (error) => { - this.logger.error({ - message: `Scheduler error: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - stack: error instanceof Error ? error.stack : undefined, - }); - }); - - this.scheduler.on("job:abort", (job, reason) => { - this.logger.warn({ - message: `${job.id} aborted: ${reason}`, - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("job:attempt", (job, attempt) => { - this.logger.debug({ - message: `${job.id} attempt ${attempt}`, - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("job:complete", (job) => { - this.logger.debug({ - message: `${job.id} completed successfully`, - service: SERVICE_NAME, - }); - }); - - this.scheduler.on("job:exhausted", (job, error) => { - this.logger.error({ - message: `${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - stack: error instanceof Error ? error.stack : undefined, - }); - }); - - this.scheduler.on("job:fail", (job, error, attempt) => { - this.logger.warn({ - message: `${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - stack: error instanceof Error ? error.stack : undefined, - }); - }); - - this.scheduler.on("job:start", (job) => { - this.logger.debug({ - message: `${job.id} started`, - service: SERVICE_NAME, - }); - }); - }; - - static async create(logger: ILogger, helper: IWorkerHelper, monitorsRepository: IMonitorsRepository) { - const scheduler = new Scheduler({ - concurrency: 50, - // storeType: "mongo", - // storeType: "redis", - // dbUri: envSettings.dbConnectionString, - }); - const instance = new SuperSimpleQueue(logger, helper, monitorsRepository, scheduler); - await instance.init(); - - return instance; - } - - init = async () => { - try { - this.scheduler.start(); - this.scheduler.addTemplate("monitor-job", this.helper.getHeartbeatJob()); - this.scheduler.addTemplate("geo-check-job", this.helper.getHeartbeatGeoJob()); - this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob()); - this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob()); - const monitors = await this.monitorsRepository.findAll(); - if (!monitors) { - return true; - } - for (const monitor of monitors) { - const randomOffset = Math.floor(Math.random() * 100); - setTimeout(() => { - this.addJob(monitor.id, monitor); - }, randomOffset); - } - - this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true }); - this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000 }); - - return true; - } catch (error: unknown) { - this.logger.error({ - message: `Failed to initialize SuperSimpleQueue: ${error instanceof Error ? error.message : String(error)}`, - service: SERVICE_NAME, - method: "init", - }); - return false; - } - }; - - addJob = async (monitorId: string, monitor: Monitor) => { - this.scheduler.addJob({ - id: monitorId, - template: "monitor-job", - repeat: monitor.interval, - active: monitor.isActive, - data: monitor, - }); - - // Return early if we don't need geo checks - if (!supportsGeoCheck(monitor.type)) { - return; - } - - // Add geo check job if enabled for HTTP monitors - if (monitor.geoCheckEnabled) { - this.scheduler.addJob({ - id: `${monitorId}-geo`, - template: "geo-check-job", - repeat: monitor.geoCheckInterval, - active: monitor.isActive, - data: monitor, - }); - } - }; - - deleteJob = async (monitor: Monitor) => { - this.scheduler.removeJob(monitor.id); - const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); - if (geoJob) await this.scheduler.removeJob(`${monitor.id}-geo`); - }; - - pauseJob = async (monitor: Monitor) => { - const result = await this.scheduler.pauseJob(monitor.id); - if (result === false) { - throw new Error("Failed to pause monitor"); - } - this.scheduler.updateJob(monitor.id, { data: monitor }); - - const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); - if (geoJob) { - const geoRes = await this.scheduler.pauseJob(`${monitor.id}-geo`); - if (geoRes === false) { - this.logger.error({ - message: `Failed to pause geo check job for monitor ${monitor.id}`, - service: SERVICE_NAME, - method: "pauseJob", - }); - } - this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); - } - - this.logger.debug({ - message: `Paused monitor ${monitor.id}`, - service: SERVICE_NAME, - method: "pauseJob", - }); - }; - - resumeJob = async (monitor: Monitor) => { - const result = await this.scheduler.resumeJob(monitor.id); - if (result === false) { - throw new Error("Failed to resume monitor"); - } - this.scheduler.updateJob(monitor.id, { data: monitor }); - - const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); - if (geoJob) { - const geoRes = await this.scheduler.resumeJob(`${monitor.id}-geo`); - if (geoRes === false) { - this.logger.error({ - message: `Failed to resume geo check job for monitor ${monitor.id}`, - service: SERVICE_NAME, - method: "resumeJob", - }); - } - this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); - } - this.logger.debug({ - message: `Resumed monitor ${monitor.id}`, - service: SERVICE_NAME, - method: "resumeJob", - }); - }; - - private syncGeoJob = async (monitor: Monitor) => { - const geoJobId = `${monitor.id}-geo`; - const existingGeoJob = await this.scheduler.getJob(geoJobId); - - // If geoChecks have been disabled, or the monitor type doesn't support them, remove - if (!monitor.geoCheckEnabled || !supportsGeoCheck(monitor.type)) { - if (existingGeoJob) this.scheduler.removeJob(geoJobId); - return; - } - - // If the job exists, update it - if (existingGeoJob) { - this.scheduler.updateJob(geoJobId, { repeat: monitor.geoCheckInterval, active: monitor.isActive, data: monitor }); - return; - } - - // Otherwise, create it - this.scheduler.addJob({ - id: geoJobId, - template: "geo-check-job", - repeat: monitor.geoCheckInterval, - active: monitor.isActive, - data: monitor, - }); - }; - - updateJob = async (monitor: Monitor) => { - this.scheduler.updateJob(monitor.id, { repeat: monitor.interval, data: monitor }); - await this.syncGeoJob(monitor); - }; - - shutdown = async () => { - this.scheduler.stop(); - }; - - getMetrics = async () => { - const jobs = await this.scheduler.getJobs(); - 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; - const lastFailedAt = job.lastFailedAt ?? 0; - const lastRunAt = job.lastRunAt ?? 0; - acc.totalRuns += runCount; - acc.totalFailures += failCount; - acc.jobs++; - if (failCount > 0 && lastFailedAt >= lastRunAt) { - acc.failingJobs++; - } - - if (job.lockedAt) { - acc.activeJobs++; - } - - if (failCount > 0) { - acc.jobsWithFailures.push({ - monitorId: job.id, - monitorUrl: job.data?.url || null, - monitorType: job.data?.type || null, - failedAt: job.lastFailedAt ?? null, - failCount, - failReason: job.lastFailReason ?? null, - }); - } - return acc; - }, - { - jobs: 0, - activeJobs: 0, - failingJobs: 0, - jobsWithFailures: [] as Array<{ - monitorId: string | number; - monitorUrl: string; - monitorType: MonitorType; - failedAt: number | null; - failCount: number; - failReason: string | null; - }>, - totalRuns: 0, - totalFailures: 0, - workers: [ - { - workerId: this.workerId, - mode: "primary", - lastSeenAt: Date.now(), - }, - ], - } - ); - return metrics; - }; - - getJobs = async ({ page = 0, rowsPerPage = 0 }: { page?: number; rowsPerPage?: number }): Promise<{ jobs: WorkerJobSummary[]; count: number }> => { - const jobs = await this.scheduler.getJobs(); - const mapped = jobs.map((job) => { - return { - monitorId: job.id, - monitorUrl: job.data?.url || null, - monitorType: job.data?.type || null, - monitorInterval: job.data?.interval || null, - monitorGeoInterval: job.data?.geoCheckInterval || null, - 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, - failReason: job.lastFailReason ?? null, - lastRunAt: job.lastRunAt ?? null, - lastFinishedAt: job.lastFinishedAt ?? null, - lastRunTook: job.lockedAt ? null : (job.lastFinishedAt ?? 0) - (job.lastRunAt ?? 0), - lastFailedAt: job.lastFailedAt ?? null, - }; - }); - - const start = Math.max(page, 0) * rowsPerPage; - return { - jobs: rowsPerPage > 0 ? mapped.slice(start, start + rowsPerPage) : mapped, - count: mapped.length, - }; - }; - - flushQueues = async () => { - const stopRes = await this.scheduler.stop(); - const flushRes = await this.scheduler.flushJobs(); - const initRes = await this.init(); - return { - success: Boolean(stopRes && flushRes && initRes), - }; - }; -}