From 8b9c41464a0996a9eea201dff3230eda90b3778c Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 12 Jun 2026 15:43:36 -0700 Subject: [PATCH] dbQueueWorker --- server/scripts/seed-monitors.js | 5 +- server/src/config/services.ts | 18 + .../domain/app-settings/app-settings.type.ts | 2 +- server/src/domain/checks/check.service.ts | 5 + .../domain/jobs/job.repository.interface.ts | 11 +- .../src/domain/jobs/job.repository.mongo.ts | 31 +- server/src/domain/jobs/job.type.ts | 7 + server/src/worker/worker.db-queue.ts | 324 ++++++++++++++++++ 8 files changed, 394 insertions(+), 9 deletions(-) create mode 100644 server/src/worker/worker.db-queue.ts diff --git a/server/scripts/seed-monitors.js b/server/scripts/seed-monitors.js index b316eec4f..e8419f73f 100644 --- a/server/scripts/seed-monitors.js +++ b/server/scripts/seed-monitors.js @@ -1,6 +1,6 @@ import mongoose from "mongoose"; -import { MonitorModel } from "../dist/db/models/Monitor.js"; -import { UserModel } from "../dist/db/models/User.js"; +import { MonitorModel } from "../dist/domain/monitors/monitor.model.js"; +import { UserModel } from "../dist/domain/users/user.model.js"; // Seeds a large number of HTTP monitors against real, resolving domains so the // scheduler can be exercised under load. Run after `npm run build` (imports the @@ -241,6 +241,7 @@ async function run() { url: `https://${domain}`, isActive: true, interval, + lastEvaluatedAt: 0, // evaluator watermark; without it findUnevaluatedByMonitorId gets new Date(undefined) }); if (docs.length === batchSize) { diff --git a/server/src/config/services.ts b/server/src/config/services.ts index db6104b83..e7741f378 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -109,6 +109,8 @@ import { IncidentReactor } from "@/worker/reactors/reactor.incident.js"; import { CheckPipeline, GeoChecksPipeline } from "@/worker/worker.check-pipeline.js"; import { CheckProducer } from "@/worker/worker.check-producer.js"; import { CheckEvaluator } from "@/worker/worker.check-evaluator.js"; +import { DBQueueWorker } from "@/worker/worker.db-queue.js"; +import MongoJobsRepository from "@/domain/jobs/job.repository.mongo.js"; export type InitializedServices = { settingsService: ISettingsService; @@ -233,6 +235,7 @@ export const initializeServices = async ({ 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) (settingsService as SettingsService).setRepository(settingsRepository); @@ -360,6 +363,21 @@ export const initializeServices = async ({ 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); } diff --git a/server/src/domain/app-settings/app-settings.type.ts b/server/src/domain/app-settings/app-settings.type.ts index d1b4c254b..7b0216548 100644 --- a/server/src/domain/app-settings/app-settings.type.ts +++ b/server/src/domain/app-settings/app-settings.type.ts @@ -1,7 +1,7 @@ export const DbTypes = ["mongodb"] as const; export type DbType = (typeof DbTypes)[number]; -export const QueueTypes = ["lessSimpleQueue", "superSimpleQueue"] as const; +export const QueueTypes = ["lessSimpleQueue", "superSimpleQueue", "dbQueue"] as const; export type QueueType = (typeof QueueTypes)[number]; export const QueueModes = ["primary", "worker"] as const; diff --git a/server/src/domain/checks/check.service.ts b/server/src/domain/checks/check.service.ts index e7fabe5be..ca7fca457 100644 --- a/server/src/domain/checks/check.service.ts +++ b/server/src/domain/checks/check.service.ts @@ -11,6 +11,7 @@ const SERVICE_NAME = "checkService"; export interface ICheckService { createChecks(checks: Check[]): Promise; + toLastEvaluatedAt(check: Check): number; toCheck(statusResponse: MonitorStatusResponse): Check | undefined; toStatusResponse(check: Check): MonitorStatusResponse; getChecksByMonitor(params: { @@ -57,6 +58,10 @@ export class CheckService implements ICheckService { return this.checksRepository.createChecks(checks); }; + toLastEvaluatedAt = (check: Check) => { + return new Date(check.createdAt).getTime(); + }; + toCheck = (statusResponse: MonitorStatusResponse): Check | undefined => { const { monitorId, teamId, type, status, responseTime, code, message, payload, timings } = statusResponse; diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index 044ea709e..e431675a0 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -1,4 +1,4 @@ -import { Job, JobType } from "@/domain/jobs/job.type.js"; +import { Job, JobSeed, JobType } from "@/domain/jobs/job.type.js"; export type JobPageQuery = { page?: number; @@ -23,6 +23,8 @@ export interface IJobsRepository { // Record failure, release lease, bump fail count, record reason, reset nextScheduledAt for retry recordFailure(id: string, error: unknown, now: number): Promise; + recordOneShot(id: string, now: number): Promise; + // ******************** // Hand off to evaluator // This is accomplished by creating an "evaluation" type job // ******************** @@ -32,8 +34,8 @@ export interface IJobsRepository { // Job CRUD // ******************** - // Create jobs - upsertJob(job: Job): Promise; + // Create jobs. Takes a JobSeed — the repo owns volatile lease/counter defaults on insert. + upsertJob(job: JobSeed): Promise; // Pause/Resume setActiveById(refId: string, isActive: boolean): Promise; @@ -44,6 +46,9 @@ export interface IJobsRepository { // Delete, drop all rows deleteById(refId: string): Promise; + // Delete a single typed row for a monitor (e.g. drop just the geo row, keep check/evaluate) + deleteByIdAndType(refId: string, type: JobType): Promise; + // Get job findById(refId: string): Promise; diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index 1e0fb9d35..a3a80f221 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -1,6 +1,6 @@ import { IJobsRepository, JobPageQuery } from "@/domain/jobs/job.repository.interface.js"; import JobModel, { JobDocument } from "@/domain/jobs/job.model.js"; -import { JobType, Job, BACKOFF_MS, LOCK_MS } from "@/domain/jobs/job.type.js"; +import { JobType, Job, JobSeed, BACKOFF_MS, LOCK_MS, PARKED, jobId } from "@/domain/jobs/job.type.js"; import { hostname } from "node:os"; import { randomUUID } from "node:crypto"; @@ -98,11 +98,31 @@ class MongoJobsRepository implements IJobsRepository { return res.modifiedCount === 1; }; + recordOneShot = async (id: string, now: number) => { + const res = await JobModel.updateOne( + { + _id: id, + }, + { + $set: { + nextScheduledAt: PARKED, + lockedBy: null, // Remove lock + lockedUntil: null, + lastFinishedAt: now, + }, + $inc: { + runCount: 1, // Increment run count + }, + } + ); + return res.modifiedCount === 1; + }; + upsertEvaluate = async (monitorId: string, now: number) => { const filter = { type: "evaluate", refId: monitorId }; const update = { $setOnInsert: { - _id: `evaluate:${monitorId}`, + _id: jobId("evaluate", monitorId), // refId and Type inferred from filter isActive: true, intervalMs: null, @@ -130,7 +150,7 @@ class MongoJobsRepository implements IJobsRepository { throw error; } }; - upsertJob = async (job: Job) => { + upsertJob = async (job: JobSeed) => { const res = await JobModel.updateOne( { _id: job.id }, { @@ -172,6 +192,11 @@ class MongoJobsRepository implements IJobsRepository { return res.deletedCount > 0; }; + deleteByIdAndType = async (refId: string, type: JobType) => { + const res = await JobModel.deleteOne({ refId, type }); // Delete a single typed row, e.g. just the geo row + return res.deletedCount > 0; + }; + findById = async (refId: string) => { const docs = await JobModel.find({ refId }).lean(); return docs.map(this.toEntity); diff --git a/server/src/domain/jobs/job.type.ts b/server/src/domain/jobs/job.type.ts index 1a9098160..7ffd0f2f8 100644 --- a/server/src/domain/jobs/job.type.ts +++ b/server/src/domain/jobs/job.type.ts @@ -2,6 +2,10 @@ export const JOB_TYPES = ["check", "geo-check", "evaluate", "cleanup-orphaned", export type JobType = (typeof JOB_TYPES)[number]; export const LOCK_MS = 30_000; export const BACKOFF_MS = 5_000; +export const PARKED = Number.MAX_SAFE_INTEGER; + +// Canonical _id scheme: monitor-bound rows are `${type}:${refId}`; global rows (refId null) are just the type. +export const jobId = (type: JobType, refId: string | null): string => (refId === null ? type : `${type}:${refId}`); export type Job = { id: string; @@ -23,3 +27,6 @@ export type Job = { lastFinishedAt: number | null; lastFailReason: string | null; }; + +// These are the only values to set when creating a job. Repo owns the rest +export type JobSeed = Pick; diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts new file mode 100644 index 000000000..cb56ac477 --- /dev/null +++ b/server/src/worker/worker.db-queue.ts @@ -0,0 +1,324 @@ +import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; +import { IWorker, WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; +import { type Job, type JobSeed, type JobType, jobId } from "@/domain/jobs/job.type.js"; +import { randomUUID } from "node:crypto"; +import { hostname } from "node:os"; +import { ILogger } from "@/utils/logger.js"; +import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; +import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; +import { IChecksRepository } from "@/domain/checks/check.repository.interface.js"; +import { ICheckService } from "@/domain/checks/check.service.js"; +import { ICheckProducer } from "@/worker/worker.check-producer.js"; +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"; +const SERVICE_NAME = "JobQueue"; +const POLL_MS = 250; +const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; +const CONCURRENCY: Record = { + check: 50, + "geo-check": 20, + evaluate: 20, + "cleanup-orphaned": 1, + "cleanup-retention": 1, +}; + +export class DBQueueWorker implements IWorker { + static SERVICE_NAME = SERVICE_NAME; + private readonly workerId = `${hostname()}:${process.pid}:${randomUUID().slice(0, 8)}`; + private stopped = false; + private timers = new Map(); + private inFlight: Record = { + check: 0, + "geo-check": 0, + evaluate: 0, + "cleanup-orphaned": 0, + "cleanup-retention": 0, + }; + + constructor( + private logger: ILogger, + private jobsRepository: IJobsRepository, + private monitorsRepository: IMonitorsRepository, + private checksRepository: IChecksRepository, + private checkService: ICheckService, // toStatusResponse + toLastEvaluatedAt + private checkProducer: ICheckProducer, + private checkEvaluator: ICheckEvaluator, + private geoCheckPipeline: ICheckPipeline, + private dispatcher: IReactorDispatcher, + private helper: IWorkerHelper + ) {} + + get serviceName() { + return DBQueueWorker.SERVICE_NAME; + } + + static async create( + logger: ILogger, + jobsRepository: IJobsRepository, + monitorsRepository: IMonitorsRepository, + checksRepository: IChecksRepository, + checkService: ICheckService, + checkProducer: ICheckProducer, + checkEvaluator: ICheckEvaluator, + geoCheckPipeline: ICheckPipeline, + dispatcher: IReactorDispatcher, + helper: IWorkerHelper + ): Promise { + const instance = new DBQueueWorker( + logger, + jobsRepository, + monitorsRepository, + checksRepository, + checkService, + checkProducer, + checkEvaluator, + geoCheckPipeline, + dispatcher, + helper + ); + await instance.init(); + return instance; + } + + // Helpers, builds job seeds + private toCheckJob = (monitor: Monitor, now: number): JobSeed => ({ + id: jobId("check", monitor.id), + type: "check", + refId: monitor.id, + isActive: monitor.isActive, + nextScheduledAt: now + Math.floor(Math.random() * monitor.interval), // Avoid herd + intervalMs: monitor.interval, + }); + + private toGeoCheckJob = (monitor: Monitor, now: number): JobSeed => ({ + ...this.toCheckJob(monitor, now), + id: jobId("geo-check", monitor.id), + type: "geo-check", + intervalMs: monitor.geoCheckInterval!, + }); + + private toCleanupJob = (type: "cleanup-orphaned" | "cleanup-retention", now: number): JobSeed => ({ + id: jobId(type, null), + type, + refId: null, + isActive: true, + nextScheduledAt: now, + intervalMs: CLEANUP_INTERVAL_MS, + }); + + private toSummary = (job: Job): WorkerJobSummary => ({ + monitorId: job.refId ?? job.id, + monitorUrl: null, // Jobs don't have URL + monitorType: job.type, + monitorInterval: job.intervalMs, + monitorGeoInterval: null, + monitorActive: job.isActive, + active: job.isActive, + lockedBy: job.lockedBy, + lockedUntil: job.lockedUntil, + lockedAt: null, // no lockedAt + runCount: job.runCount, + failCount: job.failCount, + failReason: job.lastFailReason, + lastRunAt: null, // not tracked + lastFinishedAt: job.lastFinishedAt, + lastRunTook: null, // not tracked + lastFailedAt: null, // not tracked + repeat: job.intervalMs, + }); + + // ******************** + // Stage 1: This can eventually be offloaded to a separate service + // ******************** + + private runCheck = async (job: Job) => { + if (!job.refId) return; + const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId + if (!monitor) return; + const produced = await this.checkProducer.produce(monitor); + if (produced) { + await this.jobsRepository.upsertEvaluate(monitor.id, Date.now()); + } + }; + + // ******************** + // Stage 2: Evaluator loop stays here + // ******************** + + private runEvaluate = async (job: Job) => { + if (!job.refId) return; + const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId + if (!monitor) return; + const checks = await this.checksRepository.findUnevaluatedByMonitorId(job.refId, monitor.lastEvaluatedAt); + for (const check of checks) { + const status = this.checkService.toStatusResponse(check); + const evaluation = await this.checkEvaluator.evaluate(status, check); + await this.dispatcher.dispatch(evaluation); // Handle incidents and notifications + await this.monitorsRepository.updateById(job.refId, monitor.teamId, { lastEvaluatedAt: this.checkService.toLastEvaluatedAt(check) }); + } + }; + + private runGeoCheck = async (job: Job) => { + const [monitor] = await this.monitorsRepository.findByIds([job.refId!]); + if (monitor) await this.geoCheckPipeline.run(monitor); // returns null; no evaluate handoff + }; + + private runJob = async (job: Job) => { + try { + switch (job.type) { + case "check": + await this.runCheck(job); + break; + case "evaluate": + await this.runEvaluate(job); + break; + case "geo-check": + await this.runGeoCheck(job); + break; + case "cleanup-orphaned": + await this.helper.getCleanupOrphanedJob()(); // Get job and execute + break; + case "cleanup-retention": + await this.helper.getCleanupRetentionJob()(); // Get job and execute + break; + } + + if (job.intervalMs === null) { + // One-shot job + await this.jobsRepository.recordOneShot(job.id, Date.now()); + } else { + await this.jobsRepository.recordSuccess(job.id, job.nextScheduledAt, job.intervalMs, Date.now()); + } + } catch (error: unknown) { + await this.jobsRepository.recordFailure(job.id, error, Date.now()); // Record failure and release lock + this.logger.warn({ + message: error instanceof Error ? error.message : String(error), + service: SERVICE_NAME, + method: `runJob:${job.type}`, + }); + } + }; + + private startLoop = (type: JobType) => { + const tick = async () => { + if (this.stopped) return; + try { + // Drain rows up to concurrency limit + while (this.inFlight[type] < CONCURRENCY[type]) { + const job = await this.jobsRepository.claimDue(type, Date.now()); // Claim a job + if (!job) break; // we're done, no jobs due + this.inFlight[type]++; // increment in flight jobs to stay within concurrency limits + this.runJob(job).finally(() => { + this.inFlight[type]--; // job is done, decrement in flight count + }); + } + } catch (error: unknown) { + this.logger.error({ + message: error instanceof Error ? error.message : String(error), + service: SERVICE_NAME, + method: `loop:${type}`, + }); + } + if (!this.stopped) this.timers.set(type, setTimeout(tick, POLL_MS)); // re-arm unless shutting down + }; + tick(); + }; + + // Seed queue + private reconcile = async () => { + const now = Date.now(); + const monitors = (await this.monitorsRepository.findAll()) ?? []; + for (const monitor of monitors) { + await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now)); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now)); + } + } + await this.jobsRepository.upsertJob(this.toCleanupJob("cleanup-orphaned", now)); + await this.jobsRepository.upsertJob(this.toCleanupJob("cleanup-retention", now)); + }; + + init = async () => { + this.stopped = false; + await this.reconcile(); + for (const type of Object.keys(this.inFlight) as JobType[]) { + this.startLoop(type); + } + return true; + }; + addJob = async (_monitorId: string, monitor: Monitor) => { + const now = Date.now(); + await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now)); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now)); + }; + deleteJob = async (monitor: Monitor) => { + await this.jobsRepository.deleteById(monitor.id); + }; + pauseJob = async (monitor: Monitor) => { + await this.jobsRepository.setActiveById(monitor.id, false); + }; + resumeJob = async (monitor: Monitor) => { + await this.jobsRepository.setActiveById(monitor.id, true); + }; + updateJob = async (monitor: Monitor) => { + await this.jobsRepository.updateScheduleById(monitor.id, "check", monitor.interval); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, Date.now())); + } else { + await this.jobsRepository.deleteByIdAndType(monitor.id, "geo-check"); + } + }; + shutdown = async () => { + this.stopped = true; + // Stop all timers + for (const timer of this.timers.values()) { + clearTimeout(timer); + } + this.timers.clear(); // Clear the map out + }; + getMetrics = async (): Promise => { + const rows = await this.jobsRepository.findAll(); + const now = Date.now(); + return rows.reduce( + (acc, job) => { + acc.jobs++; + acc.totalRuns += job.runCount; + acc.totalFailures += job.failCount; + if (job.lockedBy !== null && job.lockedUntil !== null && job.lockedUntil > now) acc.activeJobs++; + if (job.failCount > 0) { + acc.failingJobs++; + acc.jobsWithFailures.push({ + monitorId: job.refId ?? job.id, + monitorUrl: null, + monitorType: job.type, + failedAt: job.lastFinishedAt, + failCount: job.failCount, + failReason: job.lastFailReason, + }); + } + return acc; + }, + { + jobs: 0, + activeJobs: 0, + failingJobs: 0, + jobsWithFailures: [], + totalRuns: 0, + totalFailures: 0, + workers: [{ workerId: this.workerId, mode: "primary", lastSeenAt: now }], + } + ); + }; + + getJobs = async (pagination: WorkerJobsPagination) => { + const { jobs, count } = await this.jobsRepository.findPage(pagination); + return { jobs: jobs.map(this.toSummary), count }; + }; + flushQueues = async () => { + await this.shutdown(); + const ok = await this.init(); + return { success: ok }; + }; +}