From 51672890df61bb031588a36cf077d2bbf625b523 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 10:44:39 -0700 Subject: [PATCH] add queue type to less simple queue --- .../JobQueues/LessSimpleQueue.ts | 35 ++++++++++++------- server/src/service/system/settingsService.ts | 4 ++- 2 files changed, 25 insertions(+), 14 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index 1fa3d3e26..00a8db33f 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -4,6 +4,7 @@ import { MongoStore, Scheduler, IJob } from "less-simple-scheduler"; import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; import { Monitor, supportsGeoCheck } from "@/types/monitor.js"; import { type QueueMetrics, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; +import { type QueueMode } from "@/types/settings.js"; import { EnvConfig } from "@/service/system/settingsService.js"; const SERVICE_NAME = "JobQueue"; @@ -49,7 +50,7 @@ export class LessSimpleQueue implements IJobQueue { this.scheduler.on("scheduler:error", (workerId, error) => { this.logger.error({ - message: `w${workerId} error: ${error instanceof Error ? error.message : String(error)}`, + message: `${workerId} error: ${error instanceof Error ? error.message : String(error)}`, service: SERVICE_NAME, stack: error instanceof Error ? error.stack : undefined, }); @@ -57,35 +58,35 @@ export class LessSimpleQueue implements IJobQueue { this.scheduler.on("job:locked", (workerId, job) => { this.logger.debug({ - message: `w${workerId} ${job.id} locked`, + message: `${workerId} ${job.id} locked`, service: SERVICE_NAME, }); }); this.scheduler.on("job:abort", (workerId, job, reason) => { this.logger.warn({ - message: `w${workerId} ${job.id} aborted: ${reason}`, + message: `${workerId} ${job.id} aborted: ${reason}`, service: SERVICE_NAME, }); }); this.scheduler.on("job:attempt", (workerId, job, attempt) => { this.logger.debug({ - message: `w${workerId} ${job.id} attempt ${attempt}`, + message: `${workerId} ${job.id} attempt ${attempt}`, service: SERVICE_NAME, }); }); this.scheduler.on("job:complete", (workerId, job) => { this.logger.debug({ - message: `w${workerId} ${job.id} completed successfully`, + message: `${workerId} ${job.id} completed successfully`, service: SERVICE_NAME, }); }); this.scheduler.on("job:exhausted", (workerId, job, error) => { this.logger.error({ - message: `w${workerId} ${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(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, }); @@ -93,7 +94,7 @@ export class LessSimpleQueue implements IJobQueue { this.scheduler.on("job:fail", (workerId, job, error, attempt) => { this.logger.warn({ - message: `w${workerId} ${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`, + 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, }); @@ -101,13 +102,13 @@ export class LessSimpleQueue implements IJobQueue { this.scheduler.on("job:start", (workerId, job) => { this.logger.debug({ - message: `w${workerId} ${job.id} started`, + message: `${workerId} ${job.id} started`, service: SERVICE_NAME, }); }); }; - static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, envSettings: EnvConfig) { + static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, envSettings: EnvConfig, queueMode: QueueMode) { const store = new MongoStore({ url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db", }); @@ -115,22 +116,30 @@ export class LessSimpleQueue implements IJobQueue { concurrency: 50, }); const instance = new LessSimpleQueue(logger, helper, monitorsRepository, scheduler); - await instance.init(); + await instance.init(queueMode); return instance; } - init = async () => { + 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()); - const monitors = await this.monitorsRepository.findAll(); - if (!monitors) { + + // 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(); for (const monitor of monitors) { const randomOffset = Math.floor(Math.random() * 100); setTimeout(() => { diff --git a/server/src/service/system/settingsService.ts b/server/src/service/system/settingsService.ts index 97960df3a..dfbf6cfdc 100755 --- a/server/src/service/system/settingsService.ts +++ b/server/src/service/system/settingsService.ts @@ -1,5 +1,5 @@ import { ISettingsRepository } from "@/repositories/index.js"; -import { type DbType, type QueueType, Settings, SettingsUpdate } from "@/types/index.js"; +import { type DbType, type QueueType, type QueueMode, Settings, SettingsUpdate } from "@/types/index.js"; import { AppError } from "@/utils/AppError.js"; import { ValidatedEnv } from "@/validation/envValidation.js"; import type { StringValue } from "ms"; @@ -14,6 +14,7 @@ export type EnvConfig = { dbConnectionString: string; dbType: DbType; queueType: QueueType; + queueMode: QueueMode; statusPageThemesEnabled: boolean; }; @@ -41,6 +42,7 @@ export class SettingsService implements ISettingsService { dbConnectionString: env.DB_CONNECTION_STRING, dbType: env.DB_TYPE, queueType: env.QUEUE_TYPE, + queueMode: env.QUEUE_MODE, statusPageThemesEnabled: env.STATUS_PAGE_THEMES_ENABLED, }; }