diff --git a/docker/dev/docker-compose.yaml b/docker/dev/docker-compose.yaml index a2c66c027..09c12fe18 100755 --- a/docker/dev/docker-compose.yaml +++ b/docker/dev/docker-compose.yaml @@ -20,9 +20,35 @@ services: - server.env depends_on: - mongodb + worker: + image: uptime_server:latest + restart: always + ports: + - "52346:52346" + env_file: + - server.env + environment: + QUEUE_MODE: "worker" + HEALTH_PORT: "52346" + stop_grace_period: 30s + healthcheck: + test: + [ + "CMD", + "node", + "-e", + "require('http').get('http://localhost:52346/livez',r=>process.exit(r.statusCode===200?0:1)).on('error',()=>process.exit(1))", + ] + interval: 15s + timeout: 3s + retries: 3 + depends_on: + - mongodb mongodb: image: uptime_mongo:latest restart: always + ports: + - "27017:27017" command: ["mongod", "--quiet", "--bind_ip_all"] volumes: - ./mongo/data:/data/db diff --git a/server/src/config/envValidation.ts b/server/src/config/envValidation.ts index bdfacf906..05471e690 100644 --- a/server/src/config/envValidation.ts +++ b/server/src/config/envValidation.ts @@ -6,6 +6,8 @@ import { ILogger } from "@/utils/logger.js"; const envSchema = z.object({ // Server Configuration PORT: z.string().default("52345"), + // Port for worker health checks + HEALTH_PORT: z.string().default("52346"), NODE_ENV: z.enum(["development", "production", "test"]).default("development"), LOG_LEVEL: z.enum(["error", "warn", "info", "debug"]).default("debug"), diff --git a/server/src/config/services.worker.ts b/server/src/config/services.worker.ts index 93233cd37..44997af5f 100644 --- a/server/src/config/services.worker.ts +++ b/server/src/config/services.worker.ts @@ -9,6 +9,7 @@ import jmespath from "jmespath"; import * as grpc from "@grpc/grpc-js"; import * as protoLoader from "@grpc/proto-loader"; import WebSocket from "ws"; +import mongoose from "mongoose"; import { EnvConfig } from "@/domain/app-settings/app-settings.service.js"; import { SharedServices } from "@/config/services.shared.js"; @@ -132,10 +133,12 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig const worker = await DBQueueWorker.create( logger, + () => mongoose.connection.readyState === 1, jobsRepository, monitorsRepository, checksRepository, checkService, + bufferService, checkProducer, checkEvaluator, geoCheckPipeline, diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index 190323eed..d253523fa 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -72,4 +72,5 @@ export interface IJobsRepository { // ******************** findPage(pagination: JobPageQuery): Promise; findAll(): Promise; + countDueBacklog(now: number): Promise; // Scaling signal } diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index 89d8dfd04..60bd4cd77 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -301,6 +301,15 @@ class MongoJobsRepository implements IJobsRepository { const docs = await JobModel.find().lean(); return docs.map(this.toEntity); }; + + countDueBacklog = async (now: number) => { + return JobModel.countDocuments({ + type: "check", + isActive: true, + nextScheduledAt: { $lte: now }, + $or: [{ lockedUntil: null }, { lockedUntil: { $lt: now } }], // free or expired lock + }); + }; } export default MongoJobsRepository; diff --git a/server/src/domain/queue-workers/queue-worker.model.ts b/server/src/domain/queue-workers/queue-worker.model.ts index d1a7aa86b..0dd90b5c5 100644 --- a/server/src/domain/queue-workers/queue-worker.model.ts +++ b/server/src/domain/queue-workers/queue-worker.model.ts @@ -10,6 +10,7 @@ const QueueWorkerSchema = new Schema( { _id: { type: String, required: true }, mode: { type: String, enum: QueueModes, required: true }, + processesJobs: { type: Boolean, required: true }, lastSeenAt: { type: Date, default: Date.now, expires: WORKER_TTL_SECONDS }, }, { timestamps: true } diff --git a/server/src/domain/queue-workers/queue-worker.repository.interface.ts b/server/src/domain/queue-workers/queue-worker.repository.interface.ts index e4522c916..4da6f736e 100644 --- a/server/src/domain/queue-workers/queue-worker.repository.interface.ts +++ b/server/src/domain/queue-workers/queue-worker.repository.interface.ts @@ -2,7 +2,7 @@ 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; + upsert(workerId: string, mode: QueueMode, processesJobs: boolean): 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 index 35565bd03..da62d9850 100644 --- a/server/src/domain/queue-workers/queue-worker.repository.mongo.ts +++ b/server/src/domain/queue-workers/queue-worker.repository.mongo.ts @@ -11,12 +11,13 @@ class MongoQueueWorkersRepository implements IQueueWorkersRepository { return { workerId: doc._id, mode: doc.mode, + processesJobs: doc.processesJobs, lastSeenAt: doc.lastSeenAt.getTime(), }; }; - upsert = async (workerId: string, mode: QueueMode): Promise => { - await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true }); + upsert = async (workerId: string, mode: QueueMode, processesJobs: boolean): Promise => { + await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, processesJobs, lastSeenAt: new Date() } }, { upsert: true }); }; findRecent = async (maxAgeMs: number): Promise => { diff --git a/server/src/domain/queue-workers/queue-worker.type.ts b/server/src/domain/queue-workers/queue-worker.type.ts index a83c831ad..b750f77d7 100644 --- a/server/src/domain/queue-workers/queue-worker.type.ts +++ b/server/src/domain/queue-workers/queue-worker.type.ts @@ -1,11 +1,9 @@ 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; + processesJobs: boolean; lastSeenAt: Date; createdAt: Date; updatedAt: Date; @@ -14,5 +12,6 @@ export interface QueueWorkerDocument { export type QueueWorker = { workerId: string; // hostname:pid:uuid mode: QueueMode; + processesJobs: boolean; lastSeenAt: number; // epoch ms of the last heartbeat }; diff --git a/server/src/index.ts b/server/src/index.ts index 3b0ab6ba4..44536ce8b 100755 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -14,6 +14,7 @@ import Logger, { ILogger } from "@/utils/logger.js"; import { SettingsService } from "@/domain/app-settings/app-settings.service.js"; import { JobScheduler } from "@/worker/worker.job-scheduler.js"; import { IJobScheduler } from "@/worker/worker.interface.js"; +import { HealthServer } from "@/worker/worker.health-server.js"; const SERVICE_NAME = "Server"; let logger: ILogger; @@ -36,7 +37,7 @@ const startApp = async () => { }); logger.info({ - message: `Process: ${queuePrimaryProcesses}`, + message: `Process jobs: ${queueMode === "worker" ? true : queuePrimaryProcesses}`, service: SERVICE_NAME, method: "startApp", }); @@ -51,8 +52,10 @@ const startApp = async () => { // *********************** if (queueMode === "worker") { const { worker } = await buildWorker(shared, envSettings); + const healthServer = new HealthServer(logger, env.HEALTH_PORT, worker); + await healthServer.listen(); logger.info({ message: "Worker instance started. API will not be started", service: SERVICE_NAME }); - initShutdownListener(null, { worker, db: shared.db, logger }); + initShutdownListener(null, { worker, db: shared.db, logger, healthServer }); return; } @@ -72,7 +75,8 @@ const startApp = async () => { // Primary node does not process jobs, only need scheduler // *********************** else { - scheduler = new JobScheduler(shared.jobsRepository, shared.queueWorkersRepository, shared.workerId); + scheduler = new JobScheduler(shared.jobsRepository, shared.queueWorkersRepository, shared.monitorsRepository, queueMode, logger, shared.workerId); + await scheduler.init(); // register in the queue_workers registry + reconcile (seed the jobs collection) } const services = buildApi(shared, scheduler); diff --git a/server/src/service/bufferService.ts b/server/src/service/bufferService.ts index 989ebf936..d47a6472e 100755 --- a/server/src/service/bufferService.ts +++ b/server/src/service/bufferService.ts @@ -13,6 +13,7 @@ export interface IBufferService { scheduleNextFlush(): void; flushBuffer(): Promise; flushGeoBuffer(): Promise; + shutdown(): Promise; } export class BufferService implements IBufferService { @@ -152,4 +153,13 @@ export class BufferService implements IBufferService { this.geoBuffer = []; } } + + async shutdown() { + if (this.bufferTimer) { + clearTimeout(this.bufferTimer); + this.bufferTimer = null; + } + await this.flushBuffer(); + await this.flushGeoBuffer(); + } } diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index 625547e85..a25d96c0e 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -3,15 +3,17 @@ import type { ILogger } from "@/utils/logger.js"; import type { Server } from "http"; import { Mongoose } from "mongoose"; import { IJobScheduler } from "@/worker/worker.interface.js"; +import { IHealthServer } from "@/worker/worker.health-server.js"; type ShutdownTargets = { worker: IJobScheduler; db: IDb; logger: ILogger; + healthServer?: IHealthServer; }; -export const initShutdownListener = (server: Server | null, services: ShutdownTargets) => { +export const initShutdownListener = (server: Server | null, shutdownTargets: ShutdownTargets) => { const SERVICE_NAME = "Server"; - const logger = services.logger; + const logger = shutdownTargets.logger; let isShuttingDown = false; @@ -24,8 +26,10 @@ export const initShutdownListener = (server: Server | null, services: ShutdownTa try { server?.close(); - await services.worker.shutdown(); - await services.db.disconnect(); + await shutdownTargets.worker.drain(); + await shutdownTargets.worker.shutdown(); + await shutdownTargets.healthServer?.close(); + await shutdownTargets.db.disconnect(); logger.info({ message: "Graceful shutdown complete" }); process.exit(0); } catch (error) { diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index ec7d1cf94..cdcb2de44 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -1,6 +1,5 @@ -import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; import { IQueueWorker } from "@/worker/worker.interface.js"; -import { type Job, type JobSeed, type JobType, jobId, LOCK_MS } from "@/domain/jobs/job.type.js"; +import { type Job, type JobType, LOCK_MS } from "@/domain/jobs/job.type.js"; import { ILogger } from "@/utils/logger.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; @@ -15,13 +14,15 @@ import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.rep import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; import { JobScheduler } from "@/worker/worker.job-scheduler.js"; +import { IBufferService } from "@/service/bufferService.js"; const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; // base poll interval while a loop is actively claiming work const POLL_MAX_MS = 5000; // back off poll interval to 5s if no jobs const WORKER_STALE_MS = WORKER_TTL_SECONDS * 1000; // a worker counts as alive if seen within this window -const HEARTBEAT_MS = WORKER_STALE_MS / 3; // Worker can miss two beats without being considered stale const LOCK_RENEW_MS = LOCK_MS / 3; // renew an in-flight job's lock at 1/3 the lease, so it survives two missed renewals -const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; +const DRAIN_TIMEOUT_MS = 25_000; +const DRAIN_POLL_MS = 200; + const CONCURRENCY: Record = { check: 500, "geo-check": 20, @@ -32,6 +33,10 @@ const CONCURRENCY: Record = { export class DBQueueWorker extends JobScheduler implements IQueueWorker { static SERVICE_NAME = SERVICE_NAME; + + private initComplete: boolean = false; + private lastTickAt: number | null = null; + private inFlight: Record = { check: 0, "geo-check": 0, @@ -41,22 +46,25 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { }; constructor( - private logger: ILogger, - jobsRepository: IJobsRepository, // ← no modifier: forwarded to super - private monitorsRepository: IMonitorsRepository, + logger: ILogger, // forwarded to super + private isDbConnected: () => boolean, + jobsRepository: IJobsRepository, // forwarded + monitorsRepository: IMonitorsRepository, // forwarded private checksRepository: IChecksRepository, private checkService: ICheckService, + private bufferService: IBufferService, private checkProducer: ICheckProducer, private checkEvaluator: ICheckEvaluator, private geoCheckPipeline: ICheckPipeline, private dispatcher: IReactorDispatcher, private helper: IWorkerHelper, - queueWorkersRepository: IQueueWorkersRepository, // ← no modifier: forwarded - private queueMode: QueueMode, - private queuePrimaryProcesses: boolean, + queueWorkersRepository: IQueueWorkersRepository, // forwarded + queueMode: QueueMode, // forwarded + queuePrimaryProcesses: boolean, // consumed immediately workerId: string ) { - super(jobsRepository, queueWorkersRepository, workerId); + super(jobsRepository, queueWorkersRepository, monitorsRepository, queueMode, logger, workerId); + this.processesJobs = queuePrimaryProcesses === true || queueMode === "worker"; } get serviceName() { @@ -65,10 +73,12 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { static async create( logger: ILogger, + isDbConnected: () => boolean, jobsRepository: IJobsRepository, monitorsRepository: IMonitorsRepository, checksRepository: IChecksRepository, checkService: ICheckService, + bufferService: IBufferService, checkProducer: ICheckProducer, checkEvaluator: ICheckEvaluator, geoCheckPipeline: ICheckPipeline, @@ -81,10 +91,12 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { ): Promise { const instance = new DBQueueWorker( logger, + isDbConnected, jobsRepository, monitorsRepository, checksRepository, checkService, + bufferService, checkProducer, checkEvaluator, geoCheckPipeline, @@ -99,14 +111,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { return instance; } - 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 getInFlightCount = () => Object.values(this.inFlight).reduce((sum, n) => sum + n, 0); // ******************** // Stage 1: This can eventually be offloaded to a separate service @@ -204,6 +209,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { private startLoop = (type: JobType) => { this.pollMs[type] = POLL_MS; const tick = async () => { + this.lastTickAt = Date.now(); if (this.stopped) return; this.ticking[type] = true; try { @@ -236,55 +242,71 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { // Wake a worker loop after adding a job that's due immediately. // Otherwise the loop could be delayed by POLL_MAX_MS if it is already fully backed off - // 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)); - } - } - // Both cleanup jobs run immediately on every startup, then once a day - await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-orphaned", now)); - await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-retention", now)); - }; - - // Register worker - private heartbeat = async () => { - try { - await this.queueWorkersRepository.upsert(this.workerId, this.queueMode); - } catch (error: unknown) { - this.logger.warn({ - message: error instanceof Error ? error.message : String(error), - service: SERVICE_NAME, - method: "heartbeat", - }); - } - }; - - init = async () => { - this.stopped = false; - - await this.heartbeat(); - this.timers.set("heartbeat", setInterval(this.heartbeat, HEARTBEAT_MS)); - - if (this.queueMode === "primary") { - await this.reconcile(); - } - - if (this.queuePrimaryProcesses === true || this.queueMode === "worker") { + override async init() { + await super.init(); // heartbeat + (primary?) reconcile + if (this.processesJobs) { for (const type of Object.keys(this.inFlight) as JobType[]) { this.startLoop(type); } } + this.initComplete = true; return true; - }; + } override flushQueues = async () => { await this.shutdown(); const ok = await this.init(); return { success: ok }; }; + + override drain = async () => { + this.draining = true; + this.stopped = true; + const cutoff = Date.now() + DRAIN_TIMEOUT_MS; + while (this.getInFlightCount() > 0 && Date.now() < cutoff) { + await new Promise((resolve) => setTimeout(resolve, DRAIN_POLL_MS)); // Wait for DRAIN_POLL_MS for jobs to finish + } + + const remainingJobs = this.getInFlightCount(); + if (remainingJobs > 0) { + this.logger.warn({ + message: `Draining timed out with ${remainingJobs} in-flight. Locks will expire and jobs will be reclaimed`, + service: SERVICE_NAME, + method: "drain", + }); + } else { + this.logger.info({ + message: `${this.workerId} drained`, + service: SERVICE_NAME, + method: "drain", + }); + } + + await this.bufferService.shutdown(); // Flush buffers + }; + + // ************************ + // Observability + // ************************ + + getHealth = () => { + return { + workerId: this.workerId, + mode: this.queueMode, + dbConnected: this.isDbConnected(), + initComplete: this.initComplete, + draining: this.draining, + lastTickAt: this.lastTickAt, + inFlight: this.getInFlightCount(), + }; + }; + + countDueBacklog = () => { + return this.jobsRepository.countDueBacklog(Date.now()); + }; + + countAliveWorkers = async () => { + const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); + return workers.filter((worker) => worker.processesJobs).length; + }; } diff --git a/server/src/worker/worker.health-server.ts b/server/src/worker/worker.health-server.ts new file mode 100644 index 000000000..19bd22754 --- /dev/null +++ b/server/src/worker/worker.health-server.ts @@ -0,0 +1,129 @@ +import http from "http"; +import { IQueueWorker, WorkerHealth } from "@/worker/worker.interface.js"; +import { ILogger } from "@/utils/logger.js"; + +const SERVICE_NAME = "HealthServer"; +const DEFAULT_HEALTH_PORT = 52346; +const MAX_PORT_ATTEMPTS = 20; +const LIVENESS_STALE_MS = 30_000; // 30s ~6x POLL_MAX_MS so we don't restart workers +const CONTENT_TYPE_METRICS = "text/plain; version=0.0.4; charset=utf-8"; +const gauge = (name: string, help: string, value: number): string => `# HELP ${name} ${help}\n# TYPE ${name} gauge\n${name} ${value}\n`; + +export interface IHealthServer { + listen(): Promise; + close(): Promise; + address(): ReturnType; +} + +const isLive = (health: WorkerHealth): boolean => { + if (!health.initComplete) return false; // Not init yet + if (health.draining) return true; // Draining workers are still healthy + + if (health.lastTickAt === null) return false; // Never ticked + + if (Date.now() - health.lastTickAt >= LIVENESS_STALE_MS) return false; // Stale if last tick exceed LIVENESS_STALE_MS + + return true; +}; + +const isReady = (health: WorkerHealth): boolean => { + if (health.draining) return false; // Draining workers are not ready + return health.initComplete && health.dbConnected; +}; + +export class HealthServer implements IHealthServer { + private server: http.Server; + + constructor( + private logger: ILogger, + private port: string, + private worker: IQueueWorker + ) { + this.server = http.createServer(this.handle); + } + + private metrics = async (): Promise => { + const health = this.worker.getHealth(); + const [dueBacklog, aliveWorkers] = await Promise.all([this.worker.countDueBacklog(), this.worker.countAliveWorkers()]); + return ( + gauge("checkmate_worker_jobs_in_flight", "Jobs currently being processed by this worker", health.inFlight) + + gauge("checkmate_worker_due_backlog", "Due check jobs not currently locked (cluster-wide)", dueBacklog) + + gauge("checkmate_worker_alive_total", "Job-processing nodes seen within the TTL window (cluster-wide)", aliveWorkers) + + gauge("checkmate_worker_draining", "1 if this worker is draining, else 0", health.draining ? 1 : 0) + ); + }; + + private handle = async (req: http.IncomingMessage, res: http.ServerResponse) => { + const path = (req.url ?? "").split("?")[0]; // Strip query params + + // Metrics + if (path === "/metrics") { + try { + const body = await this.metrics(); + res.writeHead(200, { "Content-Type": CONTENT_TYPE_METRICS }); + res.end(body); + } catch (error: unknown) { + this.logger.warn({ + message: error instanceof Error ? error.message : String(error), + service: SERVICE_NAME, + method: "metrics", + }); + res.writeHead(500).end(); + } + return; + } + + if (path !== "/livez" && path !== "/readyz") { + res.writeHead(404).end(); + return; + } + + // Get health + const health = this.worker.getHealth(); + const ok = path === "/livez" ? isLive(health) : isReady(health); + const code = ok ? 200 : 503; + res.writeHead(code, { "Content-Type": "application/json" }); + res.end(JSON.stringify(health)); + }; + + listen = () => { + return new Promise((resolve, reject) => { + let port = Number(this.port) || DEFAULT_HEALTH_PORT; + const lastPort = port + MAX_PORT_ATTEMPTS; + + const onError = (error: NodeJS.ErrnoException) => { + if (error.code === "EADDRINUSE" && port < lastPort) { + this.logger.warn({ + message: `Health port ${port} in use, retrying on ${port + 1}`, + service: SERVICE_NAME, + method: "listen", + }); + port++; + this.server.listen(port); + return; + } + this.server.removeListener("error", onError); + reject(error); + }; + + this.server.on("error", onError); + this.server.listen(port, () => { + this.server.removeListener("error", onError); + this.logger.info({ + message: `Health server listening on ${port}`, + service: SERVICE_NAME, + method: "listen", + }); + resolve(); + }); + }); + }; + + address = () => this.server.address(); + + close = () => { + return new Promise((resolve) => { + this.server.close(() => resolve()); + }); + }; +} diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index 522ed3204..98120fa47 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -4,6 +4,7 @@ import { JobType } from "@/domain/jobs/job.type.js"; import { MonitorPayloadMap, MonitorStatusResponse, StatusChangeResult } from "@/types/network.js"; import { MonitorActionDecision } from "@/worker/worker.helper.js"; import { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js"; +import type { QueueMode } from "@/domain/app-settings/app-settings.type.js"; export type MonitorEvaluation = { monitor: Monitor; @@ -56,6 +57,15 @@ export type WorkerJobsPage = { count: number; }; +export type WorkerHealth = { + workerId: string; + mode: QueueMode; + dbConnected: boolean; // mongoose connection readyState === 1 + initComplete: boolean; // init() finished, loops armed + draining: boolean; // SIGTERM received, no longer claiming + lastTickAt: number | null; // newest tick across all loops (epoch ms) + inFlight: number; // total in-flight jobs across types +}; export interface IJobScheduler { readonly serviceName: string; addJob(monitorId: string, monitor: Monitor): Promise; @@ -67,10 +77,14 @@ export interface IJobScheduler { getMetrics(): Promise; getJobs(pagination: WorkerJobsPagination): Promise; flushQueues(): Promise<{ success: boolean }>; + init(): Promise; + drain(): Promise; shutdown(): Promise; } export interface IQueueWorker extends IJobScheduler { readonly serviceName: string; - init(): Promise; + getHealth(): WorkerHealth; + countDueBacklog(): Promise; + countAliveWorkers(): Promise; } diff --git a/server/src/worker/worker.job-scheduler.ts b/server/src/worker/worker.job-scheduler.ts index 4e05e4c38..458db1d08 100644 --- a/server/src/worker/worker.job-scheduler.ts +++ b/server/src/worker/worker.job-scheduler.ts @@ -1,28 +1,39 @@ import { IJobScheduler } from "@/worker/worker.interface.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; +import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; import { type Job, type JobSeed, type JobType, jobId } from "@/domain/jobs/job.type.js"; import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; import { WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; +import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; +import { ILogger } from "@/utils/logger.js"; const POLL_MS = 250; // base poll interval while a loop is actively claiming work const SERVICE_NAME = "JobScheduler"; const WORKER_STALE_MS = WORKER_TTL_SECONDS * 1000; // a worker counts as alive if seen within this window +const HEARTBEAT_MS = WORKER_STALE_MS / 3; // Worker can miss two beats without being considered stale +const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; export class JobScheduler implements IJobScheduler { static SERVICE_NAME = SERVICE_NAME; constructor( protected jobsRepository: IJobsRepository, protected queueWorkersRepository: IQueueWorkersRepository, + protected monitorsRepository: IMonitorsRepository, + protected queueMode: QueueMode, + protected logger: ILogger, protected readonly workerId: string // same id MongoJobsRepository stamps into lockedBy ) {} get serviceName() { return JobScheduler.SERVICE_NAME; } + private heartbeatInFlight: Promise = Promise.resolve(); + protected processesJobs = false; // scheduler-only by default; DBQueueWorker flips this on when it runs job loops protected stopped = false; + protected draining = false; protected ticking: Partial> = {}; // Whether a loop is mid tick protected tickFns = new Map void>(); // Store tick fns so we can run them immediately on wake protected timers = new Map(); @@ -52,6 +63,15 @@ export class JobScheduler implements IJobScheduler { intervalMs: monitor.geoCheckInterval!, }); + protected 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, monitorType: job.type, @@ -111,7 +131,8 @@ export class JobScheduler implements IJobScheduler { getMetrics = async (): Promise => { const rows = await this.jobsRepository.findAll(); - const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); + const recent = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); + const workers = recent.filter((worker) => worker.processesJobs); const now = Date.now(); return rows.reduce( (acc, job) => { @@ -154,6 +175,60 @@ export class JobScheduler implements IJobScheduler { return { success: true }; }; + // init is the scheduler's startup: register in the registry + (if primary) seed the queue. + // Subclasses override to ALSO start processing, calling super.init() first. + // Declared as a prototype method (not an arrow field) so subclasses can reach it via super.init(). + async init(): Promise { + this.stopped = false; + await this.heartbeat(); + this.timers.set("heartbeat", setInterval(this.heartbeat, HEARTBEAT_MS)); + if (this.queueMode === "primary") { + await this.reconcile(); + } + return true; + } + + // Seed queue + protected 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)); + } + } + // Both cleanup jobs run immediately on every startup, then once a day + await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-orphaned", now)); + await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-retention", now)); + }; + + // Register worker + protected heartbeat = async () => { + try { + if (this.stopped) return; // Don't re-register if shutting down + this.heartbeatInFlight = this.queueWorkersRepository.upsert(this.workerId, this.queueMode, this.processesJobs).catch((error: unknown) => { + this.logger.warn({ + message: error instanceof Error ? error.message : String(error), + service: SERVICE_NAME, + method: "heartbeat", + }); + }); + await this.heartbeatInFlight; + } catch (error: unknown) { + this.logger.warn({ + message: error instanceof Error ? error.message : String(error), + service: SERVICE_NAME, + method: "heartbeat", + }); + } + }; + + drain = async () => { + this.stopped = true; + this.draining = true; + }; + shutdown = async () => { this.stopped = true; // Stop all timers @@ -161,6 +236,7 @@ export class JobScheduler implements IJobScheduler { clearTimeout(timer); } this.timers.clear(); // Clear the map out + await this.heartbeatInFlight; // Wait for in flight jobs // Remove workers from registry await this.queueWorkersRepository.deleteById(this.workerId); }; diff --git a/server/test/integration/jobsRepository.test.ts b/server/test/integration/jobsRepository.test.ts index 02b0ecc5d..78ac17b64 100644 --- a/server/test/integration/jobsRepository.test.ts +++ b/server/test/integration/jobsRepository.test.ts @@ -497,4 +497,40 @@ describe("MongoJobsRepository", () => { expect(await repo.findAll()).toHaveLength(2); }); }); + + // ── countDueBacklog (scaling signal) ───────────────────────────────────────── + describe("countDueBacklog", () => { + it("counts a due, active, unlocked check job", async () => { + await seedJob(); + expect(await repo.countDueBacklog(NOW)).toBe(1); + }); + + it("counts a check job whose lock has expired (reclaimable work)", async () => { + await seedJob({ lockedBy: "dead-worker", lockedUntil: NOW - 1 }); + expect(await repo.countDueBacklog(NOW)).toBe(1); + }); + + it("excludes not-yet-due, inactive, live-locked, and non-check rows", async () => { + await seedJob({ _id: "check:due", refId: "due" }); // counts + await seedJob({ _id: "check:future", refId: "future", nextScheduledAt: NOW + 1 }); // not due + await seedJob({ _id: "check:paused", refId: "paused", isActive: false }); // inactive + await seedJob({ _id: "check:locked", refId: "locked", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // live lock + await seedJob({ _id: "geo:mon", refId: "mon", type: "geo-check" }); // wrong type + await seedJob({ _id: "evaluate:mon", refId: "mon", type: "evaluate" }); // wrong type + + expect(await repo.countDueBacklog(NOW)).toBe(1); // only check:due + }); + + it("counts every qualifying check row across monitors", async () => { + await seedJob({ _id: "check:a", refId: "a" }); + await seedJob({ _id: "check:b", refId: "b" }); + await seedJob({ _id: "check:c", refId: "c", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // excluded + expect(await repo.countDueBacklog(NOW)).toBe(2); + }); + + it("returns 0 when nothing is due", async () => { + await seedJob({ nextScheduledAt: NOW + 10_000 }); + expect(await repo.countDueBacklog(NOW)).toBe(0); + }); + }); }); diff --git a/server/test/integration/schedulerReconcile.test.ts b/server/test/integration/schedulerReconcile.test.ts new file mode 100644 index 000000000..8b1a48f79 --- /dev/null +++ b/server/test/integration/schedulerReconcile.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, it, beforeAll, afterAll, beforeEach } from "@jest/globals"; +import mongoose from "mongoose"; +import { MongoMemoryServer } from "mongodb-memory-server"; +import { JobScheduler } from "../../src/worker/worker.job-scheduler.ts"; +import MongoJobsRepository from "../../src/domain/jobs/job.repository.mongo.ts"; +import MongoQueueWorkersRepository from "../../src/domain/queue-workers/queue-worker.repository.mongo.ts"; +import MongoMonitorsRepository from "../../src/domain/monitors/monitor.repository.mongo.ts"; +import JobModel, { type JobDocument } from "../../src/domain/jobs/job.model.ts"; +import { MonitorModel } from "../../src/domain/monitors/monitor.model.ts"; +import { QueueWorkerModel } from "../../src/domain/queue-workers/queue-worker.model.ts"; +import { createMockLogger } from "../helpers/createMockLogger.ts"; + +// ── Real-Mongo harness ───────────────────────────────────────────────────────── +// The bug: a scheduler-only primary (queueMode=primary, queuePrimaryProcesses=false) +// never seeded the jobs collection, so a fresh start with existing monitors left the +// workers idle. The fix moved reconcile/heartbeat onto the base JobScheduler.init(). +// This proves init() against a live mongod actually writes the expected job rows. + +let mongod: MongoMemoryServer; + +beforeAll(async () => { + mongod = await MongoMemoryServer.create(); + await mongoose.connect(mongod.getUri()); + await JobModel.init(); +}, 120_000); + +afterAll(async () => { + await mongoose.disconnect(); + await mongod.stop(); +}); + +beforeEach(async () => { + await JobModel.deleteMany({}); + await MonitorModel.collection.deleteMany({}); + await QueueWorkerModel.deleteMany({}); +}); + +const WORKER_ID = "scheduler-primary-1"; + +const seedMonitor = (overrides: Record = {}) => + MonitorModel.create({ + userId: new mongoose.Types.ObjectId(), + teamId: new mongoose.Types.ObjectId(), + name: "mon", + type: "http", + url: "https://example.com", + interval: 60_000, + isActive: true, + ...overrides, + }); + +const makeScheduler = () => + new JobScheduler( + new MongoJobsRepository(WORKER_ID), + new MongoQueueWorkersRepository(), + new MongoMonitorsRepository(), + "primary", + createMockLogger(), + WORKER_ID + ); + +const idsOf = async () => (await JobModel.find().lean()).map((j) => j._id); + +describe("JobScheduler.init (scheduler-only primary reconcile)", () => { + it("seeds a check row per monitor plus the two cleanup rows", async () => { + const m1 = await seedMonitor(); + const m2 = await seedMonitor({ name: "mon-2" }); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); // stop the heartbeat interval + + const ids = await idsOf(); + expect(ids).toContain(`check:${m1.id}`); + expect(ids).toContain(`check:${m2.id}`); + expect(ids).toContain("cleanup-orphaned"); + expect(ids).toContain("cleanup-retention"); + }); + + it("seeds a geo-check row only for geo-enabled monitors", async () => { + const geo = await seedMonitor({ name: "geo", geoCheckEnabled: true }); + const plain = await seedMonitor({ name: "plain", geoCheckEnabled: false }); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); + + const ids = await idsOf(); + expect(ids).toContain(`geo-check:${geo.id}`); + expect(ids).not.toContain(`geo-check:${plain.id}`); + }); + + it("registers the primary in the queue_workers registry", async () => { + await seedMonitor(); + + const scheduler = makeScheduler(); + await scheduler.init(); + await scheduler.shutdown(); + + // shutdown deregisters, so assert mid-flight: re-init then read before shutdown. + const scheduler2 = makeScheduler(); + await scheduler2.init(); + const row = await QueueWorkerModel.findById(WORKER_ID).lean(); + expect(row?.mode).toBe("primary"); + await scheduler2.shutdown(); + }); +}); diff --git a/server/test/unit/services/bufferService.test.ts b/server/test/unit/services/bufferService.test.ts index 7c344c086..d994e5f24 100644 --- a/server/test/unit/services/bufferService.test.ts +++ b/server/test/unit/services/bufferService.test.ts @@ -482,4 +482,32 @@ describe("BufferService", () => { ); }); }); + + // ── shutdown ───────────────────────────────────────────────────────────── + + describe("shutdown", () => { + it("stops the flush timer and flushes both buffers", async () => { + const { service, checkService, geoChecksService } = createService(); + service.addToBuffer(makeCheck()); + service.addGeoCheckToBuffer(makeGeoCheck()); + + await service.shutdown(); + + expect(checkService.createChecks).toHaveBeenCalledTimes(1); + expect(geoChecksService.createGeoChecks).toHaveBeenCalledTimes(1); + + // Timer is cleared: advancing past the flush interval triggers no further flush. + await jest.advanceTimersByTimeAsync(60 * 1000); + expect(checkService.createChecks).toHaveBeenCalledTimes(1); + }); + + it("is safe to call with empty buffers", async () => { + const { service, checkService, geoChecksService } = createService(); + + await service.shutdown(); + + expect(checkService.createChecks).not.toHaveBeenCalled(); + expect(geoChecksService.createGeoChecks).not.toHaveBeenCalled(); + }); + }); }); diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 66ca6753d..cabf90e31 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -20,6 +20,8 @@ import { createMockLogger } from "../../helpers/createMockLogger.ts"; const POLL_MS = 250; const LOCK_RENEW_MS = LOCK_MS / 3; +const DRAIN_TIMEOUT_MS = 25_000; // mirror of DBQueueWorker DRAIN_TIMEOUT_MS +const DRAIN_POLL_MS = 200; // mirror of DBQueueWorker DRAIN_POLL_MS const deferred = () => { let resolve!: (value: T) => void; @@ -77,6 +79,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses updateScheduleById: jest.fn().mockResolvedValue(true), findAll: jest.fn().mockResolvedValue([]), findPage: jest.fn().mockResolvedValue({ jobs: [], count: 0 }), + countDueBacklog: jest.fn().mockResolvedValue(0), }; const monitorsRepository = { findByIds: jest.fn().mockResolvedValue([makeMonitor()]), @@ -91,6 +94,14 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses toStatusResponse: jest.fn().mockReturnValue({ status: "up" }), toLastEvaluatedAt: jest.fn().mockReturnValue(12345), }; + const bufferService = { + addToBuffer: jest.fn(), + addGeoCheckToBuffer: jest.fn(), + scheduleNextFlush: jest.fn(), + flushBuffer: jest.fn().mockResolvedValue(undefined), + flushGeoBuffer: jest.fn().mockResolvedValue(undefined), + shutdown: jest.fn().mockResolvedValue(undefined), + }; const checkProducer = { produce: jest.fn().mockResolvedValue({ status: { status: "up" }, check: { id: "c1" } }), }; @@ -109,12 +120,15 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses findRecent: jest.fn().mockResolvedValue([]), }; const logger = createMockLogger(); + const isDbConnected = jest.fn(() => true); const mocks = { + isDbConnected, jobsRepository, monitorsRepository, checksRepository, checkService, + bufferService, checkProducer, checkEvaluator, geoCheckPipeline, @@ -127,10 +141,12 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses const worker = new DBQueueWorker( mocks.logger as any, + mocks.isDbConnected as any, mocks.jobsRepository as any, mocks.monitorsRepository as any, mocks.checksRepository as any, mocks.checkService as any, + mocks.bufferService as any, mocks.checkProducer as any, mocks.checkEvaluator as any, mocks.geoCheckPipeline as any, @@ -171,7 +187,7 @@ describe("DBQueueWorker", () => { describe("init", () => { it("registers the worker in the heartbeat registry on startup", async () => { const { mocks } = await start({ queueMode: "worker" }); - expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker"); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", true); }); it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => { @@ -229,10 +245,12 @@ describe("DBQueueWorker", () => { const { mocks } = createWorker(); // reuse the mock set; the un-init'd instance is discarded const worker = await DBQueueWorker.create( mocks.logger as any, + mocks.isDbConnected as any, mocks.jobsRepository as any, mocks.monitorsRepository as any, mocks.checksRepository as any, mocks.checkService as any, + mocks.bufferService as any, mocks.checkProducer as any, mocks.checkEvaluator as any, mocks.geoCheckPipeline as any, @@ -246,7 +264,7 @@ describe("DBQueueWorker", () => { active = worker; await jest.advanceTimersByTimeAsync(1); - expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker"); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", true); }); }); @@ -619,6 +637,161 @@ describe("DBQueueWorker", () => { }); }); + // ── drain ───────────────────────────────────────────────────────────────────── + + describe("drain", () => { + // Hand the "check" loop exactly one job, then nothing. + const claimOnce = (job: Job) => { + let handed = false; + return jest.fn(async (type: string) => { + if (type === job.type && !handed) { + handed = true; + return [job]; + } + return []; + }); + }; + + it("stops the loops and flushes the buffer when nothing is in flight", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + const claimsBefore = mocks.jobsRepository.claimDueBatch.mock.calls.length; + + await worker.drain(); + + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + expect(mocks.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("drained") })); + // Loops must not claim again after draining. + await jest.advanceTimersByTimeAsync(POLL_MS * 4); + expect(mocks.jobsRepository.claimDueBatch.mock.calls.length).toBe(claimsBefore); + }); + + it("waits for an in-flight job to finish before flushing", async () => { + const job = makeJob({ type: "check" }); + const gate = deferred<{ status: unknown; check: unknown }>(); + const slowProducer = { produce: jest.fn().mockReturnValue(gate.promise) }; + const { worker, mocks } = await start({ + mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checkProducer: slowProducer }, + }); + + // Job is parked on produce(); drain must not flush while it is still in flight. + const draining = worker.drain(); + await jest.advanceTimersByTimeAsync(DRAIN_POLL_MS * 3); + expect(mocks.bufferService.shutdown).not.toHaveBeenCalled(); + + // Let the job finish; drain observes inFlight == 0, then flushes. + gate.resolve({ status: { status: "up" }, check: { id: "c1" } }); + await jest.advanceTimersByTimeAsync(DRAIN_POLL_MS); + await draining; + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + }); + + it("times out after DRAIN_TIMEOUT_MS, warns, and still flushes", async () => { + const job = makeJob({ type: "check" }); + const gate = deferred<{ status: unknown; check: unknown }>(); + const stuckProducer = { produce: jest.fn().mockReturnValue(gate.promise) }; + const { worker, mocks } = await start({ + mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOnce(job) }, checkProducer: stuckProducer }, + }); + + const draining = worker.drain(); + await jest.advanceTimersByTimeAsync(DRAIN_TIMEOUT_MS + DRAIN_POLL_MS); + await draining; + + expect(mocks.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("timed out") })); + expect(mocks.bufferService.shutdown).toHaveBeenCalledTimes(1); + + gate.resolve({ status: { status: "up" }, check: { id: "c1" } }); // release the stuck job for cleanup + }); + }); + + // ── getHealth ─────────────────────────────────────────────────────────────────── + + describe("getHealth", () => { + // Hand the "check" loop exactly one job, then nothing — keeps it parked in-flight. + const claimOneCheck = (job: Job) => { + let handed = false; + return jest.fn(async (type: string) => { + if (type === job.type && !handed) { + handed = true; + return [job]; + } + return []; + }); + }; + + it("reports un-init state before init(): not complete, not draining, never ticked", () => { + const { worker } = createWorker({ queueMode: "worker" }); + active = worker; // afterEach shutdown() is safe on an un-init'd worker + expect(worker.getHealth()).toMatchObject({ + workerId: "worker-1", + mode: "worker", + initComplete: false, + draining: false, + lastTickAt: null, + inFlight: 0, + }); + }); + + it("flips initComplete and stamps lastTickAt once the loops run", async () => { + const { worker } = await start({ queueMode: "worker" }); + const health = worker.getHealth(); + expect(health.initComplete).toBe(true); + expect(typeof health.lastTickAt).toBe("number"); + }); + + it("reflects the injected dbConnected probe", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + expect(worker.getHealth().dbConnected).toBe(true); + mocks.isDbConnected.mockReturnValue(false); + expect(worker.getHealth().dbConnected).toBe(false); + }); + + it("sets draining true after drain()", async () => { + const { worker } = await start({ queueMode: "worker" }); + expect(worker.getHealth().draining).toBe(false); + await worker.drain(); + expect(worker.getHealth().draining).toBe(true); + }); + + it("counts in-flight jobs while they run and releases on completion", async () => { + const job = makeJob({ type: "check" }); + const gate = deferred<{ status: unknown; check: unknown }>(); + const slowProducer = { produce: jest.fn().mockReturnValue(gate.promise) }; + const { worker } = await start({ + mocks: { jobsRepository: { ...createWorker().mocks.jobsRepository, claimDueBatch: claimOneCheck(job) }, checkProducer: slowProducer }, + }); + + expect(worker.getHealth().inFlight).toBe(1); // parked on produce() + + gate.resolve({ status: { status: "up" }, check: { id: "c1" } }); + await jest.advanceTimersByTimeAsync(1); + expect(worker.getHealth().inFlight).toBe(0); + }); + }); + + // ── metrics accessors ─────────────────────────────────────────────────────────── + + describe("metrics accessors", () => { + it("countDueBacklog delegates to the repo with the current time", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + mocks.jobsRepository.countDueBacklog.mockResolvedValue(7); + + expect(await worker.countDueBacklog()).toBe(7); + expect(mocks.jobsRepository.countDueBacklog).toHaveBeenCalledWith(expect.any(Number)); + }); + + it("countAliveWorkers counts only recently-seen nodes that process jobs", async () => { + const { worker, mocks } = await start({ queueMode: "worker" }); + mocks.queueWorkersRepository.findRecent.mockResolvedValue([ + { processesJobs: true }, + { processesJobs: true }, + { processesJobs: false }, // scheduler-only primary: alive but not a processor + ]); + + expect(await worker.countAliveWorkers()).toBe(2); + }); + }); + // ── flushQueues ─────────────────────────────────────────────────────────────── describe("flushQueues", () => { @@ -650,7 +823,10 @@ describe("DBQueueWorker", () => { const { worker, mocks } = createWorker(); active = worker; mocks.jobsRepository.findAll.mockResolvedValue(rows); - mocks.queueWorkersRepository.findRecent.mockResolvedValue([{ workerId: "w" }]); + mocks.queueWorkersRepository.findRecent.mockResolvedValue([ + { workerId: "w", processesJobs: true }, + { workerId: "scheduler", processesJobs: false }, // scheduler-only primary excluded from metrics + ]); const metrics = await worker.getMetrics(); @@ -661,7 +837,7 @@ describe("DBQueueWorker", () => { expect(metrics.failingJobs).toBe(1); expect(metrics.jobsWithFailures).toHaveLength(1); expect(metrics.jobsWithFailures[0]).toEqual(expect.objectContaining({ failCount: 3, failReason: "boom" })); - expect(metrics.workers).toEqual([{ workerId: "w" }]); + expect(metrics.workers).toEqual([{ workerId: "w", processesJobs: true }]); }); }); diff --git a/server/test/unit/services/healthServer.test.ts b/server/test/unit/services/healthServer.test.ts new file mode 100644 index 000000000..ccf9d535c --- /dev/null +++ b/server/test/unit/services/healthServer.test.ts @@ -0,0 +1,199 @@ +import { afterEach, describe, expect, it } from "@jest/globals"; +import http from "http"; +import net from "net"; +import { HealthServer } from "../../../src/worker/worker.health-server.ts"; +import type { WorkerHealth } from "../../../src/worker/worker.interface.ts"; +import { createMockLogger } from "../../helpers/createMockLogger.ts"; + +// ── Notes ────────────────────────────────────────────────────────────────────── +// +// HealthServer is a dumb bare-http listener over the worker's getHealth() snapshot: +// it maps the snapshot to 200/503 per endpoint and serves the snapshot as the body. +// We drive it end-to-end with real sockets and a stub worker, so REAL timers are used +// (Date.now() drives the /livez staleness check) — no fake timers in this file. +// +// listen() binds `Number(port) || 52346`, so "0" would resolve to 52346, not an +// ephemeral port. We discover a free port first and pass it as a concrete string. + +const freePort = (): Promise => + new Promise((resolve, reject) => { + const srv = net.createServer(); + srv.once("error", reject); + srv.listen(0, () => { + const addr = srv.address(); + const port = typeof addr === "object" && addr ? addr.port : 0; + srv.close(() => resolve(port)); + }); + }); + +const get = (port: number, path: string): Promise<{ status: number; body: string; contentType: string }> => + new Promise((resolve, reject) => { + http + .get({ host: "127.0.0.1", port, path }, (res) => { + let body = ""; + res.on("data", (chunk) => (body += chunk)); + res.on("end", () => resolve({ status: res.statusCode ?? 0, body, contentType: String(res.headers["content-type"] ?? "") })); + }) + .on("error", reject); + }); + +// Fresh, fully-healthy snapshot; override per test. lastTickAt is stamped at call time +// so it is "recent" relative to the request that follows a moment later. +const makeHealth = (over?: Partial): WorkerHealth => ({ + workerId: "worker-1", + mode: "worker", + dbConnected: true, + initComplete: true, + draining: false, + lastTickAt: Date.now(), + inFlight: 0, + ...over, +}); + +describe("HealthServer", () => { + let active: HealthServer | null = null; + + afterEach(async () => { + if (active) await active.close(); + active = null; + }); + + // The metrics endpoint also calls countDueBacklog()/countAliveWorkers(); default them + // to 0 and let tests override (including to a rejection, to exercise the 500 path). + const setup = async (health: WorkerHealth, workerOverrides: Record = {}) => { + const port = await freePort(); + const worker = { + getHealth: () => health, + countDueBacklog: async () => 0, + countAliveWorkers: async () => 0, + ...workerOverrides, + } as any; + const server = new HealthServer(createMockLogger(), String(port), worker); + await server.listen(); + active = server; + return { port }; + }; + + describe("/livez", () => { + it("200 when init complete and a tick is recent", async () => { + const { port } = await setup(makeHealth()); + expect((await get(port, "/livez")).status).toBe(200); + }); + + it("503 before init completes", async () => { + const { port } = await setup(makeHealth({ initComplete: false })); + expect((await get(port, "/livez")).status).toBe(503); + }); + + it("503 when it has never ticked", async () => { + const { port } = await setup(makeHealth({ lastTickAt: null })); + expect((await get(port, "/livez")).status).toBe(503); + }); + + it("503 when the last tick is stale and not draining", async () => { + const { port } = await setup(makeHealth({ lastTickAt: Date.now() - 31_000 })); + expect((await get(port, "/livez")).status).toBe(503); + }); + + it("200 when draining even if ticks are stale (drain must read as live)", async () => { + const { port } = await setup(makeHealth({ draining: true, lastTickAt: Date.now() - 60_000 })); + expect((await get(port, "/livez")).status).toBe(200); + }); + }); + + describe("/readyz", () => { + it("200 when init complete, db connected, and not draining", async () => { + const { port } = await setup(makeHealth()); + expect((await get(port, "/readyz")).status).toBe(200); + }); + + it("503 while draining", async () => { + const { port } = await setup(makeHealth({ draining: true })); + expect((await get(port, "/readyz")).status).toBe(503); + }); + + it("503 when the db is disconnected", async () => { + const { port } = await setup(makeHealth({ dbConnected: false })); + expect((await get(port, "/readyz")).status).toBe(503); + }); + + it("503 before init completes", async () => { + const { port } = await setup(makeHealth({ initComplete: false })); + expect((await get(port, "/readyz")).status).toBe(503); + }); + }); + + describe("routing & body", () => { + it("serves the WorkerHealth snapshot as the JSON body", async () => { + const { port } = await setup(makeHealth({ inFlight: 3 })); + const res = await get(port, "/readyz"); + expect(JSON.parse(res.body)).toMatchObject({ workerId: "worker-1", mode: "worker", inFlight: 3 }); + }); + + it("404s unknown paths", async () => { + const { port } = await setup(makeHealth()); + expect((await get(port, "/nope")).status).toBe(404); + }); + + it("ignores query params when matching the route", async () => { + const { port } = await setup(makeHealth()); + expect((await get(port, "/livez?probe=1")).status).toBe(200); + }); + }); + + describe("/metrics", () => { + it("200 with prometheus content-type and all four gauges", async () => { + const { port } = await setup(makeHealth({ inFlight: 3, draining: false }), { + countDueBacklog: async () => 5, + countAliveWorkers: async () => 2, + }); + const res = await get(port, "/metrics"); + + expect(res.status).toBe(200); + expect(res.contentType).toMatch(/text\/plain/); + expect(res.body).toContain("# TYPE checkmate_worker_due_backlog gauge"); + expect(res.body).toContain("checkmate_worker_jobs_in_flight 3"); + expect(res.body).toContain("checkmate_worker_due_backlog 5"); + expect(res.body).toContain("checkmate_worker_alive_total 2"); + expect(res.body).toContain("checkmate_worker_draining 0"); + }); + + it("reports draining as 1 when the worker is draining", async () => { + const { port } = await setup(makeHealth({ draining: true })); + expect((await get(port, "/metrics")).body).toContain("checkmate_worker_draining 1"); + }); + + it("500 when a metrics query fails", async () => { + const { port } = await setup(makeHealth(), { + countDueBacklog: async () => { + throw new Error("db down"); + }, + }); + expect((await get(port, "/metrics")).status).toBe(500); + }); + }); + + describe("listen", () => { + it("walks to the next port when the configured one is in use", async () => { + const base = await freePort(); + // Hold the base port so the health server's first bind hits EADDRINUSE. + const blocker = net.createServer(); + await new Promise((resolve) => blocker.listen(base, resolve)); + + try { + const worker = { getHealth: () => makeHealth(), countDueBacklog: async () => 0, countAliveWorkers: async () => 0 } as any; + const server = new HealthServer(createMockLogger(), String(base), worker); + await server.listen(); + active = server; + + // base is still held by the blocker, so the health server retried onto a higher port and is reachable there. + const addr = server.address(); + const bound = typeof addr === "object" && addr ? addr.port : 0; + expect(bound).toBeGreaterThan(base); + expect((await get(bound, "/livez")).status).toBe(200); + } finally { + await new Promise((resolve) => blocker.close(() => resolve())); + } + }); + }); +}); diff --git a/server/test/unit/services/jobScheduler.test.ts b/server/test/unit/services/jobScheduler.test.ts new file mode 100644 index 000000000..90d7a2b99 --- /dev/null +++ b/server/test/unit/services/jobScheduler.test.ts @@ -0,0 +1,145 @@ +import { describe, expect, it, jest, beforeEach, afterEach } from "@jest/globals"; +import { JobScheduler } from "../../../src/worker/worker.job-scheduler.ts"; +import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts"; +import type { QueueMode } from "../../../src/domain/app-settings/app-settings.type.ts"; +import { createMockLogger } from "../../helpers/createMockLogger.ts"; + +// ── Notes ────────────────────────────────────────────────────────────────────── +// +// JobScheduler is the scheduler-only primary base class: it owns the queue +// registry/heartbeat AND, on a primary, the reconcile that seeds the jobs +// collection — scheduling is its job whether or not it also processes. It runs no +// polling loops and has no buffer, so drain() is a no-op beyond setting the stop +// flag — it must NOT tear down (clear timers or deregister the worker); that stays +// in shutdown(). The processing tier (poll loops, buffer flush) lives in +// DBQueueWorker and is covered separately. + +const makeMonitor = (overrides?: Partial): Monitor => + ({ + id: "m1", + teamId: "team", + type: "http", + interval: 60000, + status: "up", + isActive: true, + geoCheckEnabled: false, + geoCheckInterval: 300000, + lastEvaluatedAt: 0, + ...overrides, + }) as Monitor; + +const createScheduler = (overrides?: { queueMode?: QueueMode; mocks?: Record }) => { + const jobsRepository = { + upsertJob: jest.fn().mockResolvedValue(true), + upsertCleanupJob: jest.fn().mockResolvedValue(true), + }; + const queueWorkersRepository = { + upsert: jest.fn().mockResolvedValue(undefined), + deleteById: jest.fn().mockResolvedValue(undefined), + }; + const monitorsRepository = { + findAll: jest.fn().mockResolvedValue([]), + }; + const logger = createMockLogger(); + const mocks = { jobsRepository, queueWorkersRepository, monitorsRepository, logger, ...overrides?.mocks }; + const scheduler = new JobScheduler( + mocks.jobsRepository as any, + mocks.queueWorkersRepository as any, + mocks.monitorsRepository as any, + overrides?.queueMode ?? "primary", + mocks.logger as any, + "worker-1" + ); + return { scheduler, mocks }; +}; + +describe("JobScheduler", () => { + describe("init", () => { + beforeEach(() => { + jest.useFakeTimers(); + }); + + afterEach(async () => { + jest.clearAllTimers(); + jest.useRealTimers(); + }); + + it("registers the worker in the heartbeat registry on startup", async () => { + const { scheduler, mocks } = createScheduler({ queueMode: "primary" }); + await scheduler.init(); + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "primary", false); + }); + + it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "primary", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor()]) } }, + }); + await scheduler.init(); + + const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type); + expect(seededTypes).toContain("check"); + const cleanupTypes = mocks.jobsRepository.upsertCleanupJob.mock.calls.map((c: any[]) => c[0].type); + expect(cleanupTypes).toContain("cleanup-orphaned"); + expect(cleanupTypes).toContain("cleanup-retention"); + }); + + it("primary mode also seeds a geo-check row for geo-enabled monitors", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "primary", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor({ geoCheckEnabled: true })]) } }, + }); + await scheduler.init(); + + const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type); + expect(seededTypes).toContain("geo-check"); + }); + + it("worker mode heartbeats but does not reconcile (no monitor reads, no seeds)", async () => { + const { scheduler, mocks } = createScheduler({ + queueMode: "worker", + mocks: { monitorsRepository: { findAll: jest.fn().mockResolvedValue([makeMonitor()]) } }, + }); + await scheduler.init(); + + expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", false); + expect(mocks.monitorsRepository.findAll).not.toHaveBeenCalled(); + expect(mocks.jobsRepository.upsertJob).not.toHaveBeenCalled(); + expect(mocks.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled(); + }); + + it("arms a heartbeat interval timer that shutdown() clears and deregisters", async () => { + const { scheduler, mocks } = createScheduler({ queueMode: "primary" }); + await scheduler.init(); + expect((scheduler as any).timers.has("heartbeat")).toBe(true); + + await scheduler.shutdown(); + expect((scheduler as any).timers.has("heartbeat")).toBe(false); + expect(mocks.queueWorkersRepository.deleteById).toHaveBeenCalledWith("worker-1"); + }); + }); + + describe("drain", () => { + it("sets the stop and draining flags without tearing down (no deregister)", async () => { + const { scheduler, mocks } = createScheduler(); + + await scheduler.drain(); + + expect((scheduler as any).stopped).toBe(true); + expect((scheduler as any).draining).toBe(true); + // drain must not do shutdown's teardown work + expect(mocks.queueWorkersRepository.deleteById).not.toHaveBeenCalled(); + }); + + it("stops wake() from arming a tick once drained", async () => { + const { scheduler } = createScheduler(); + const tick = jest.fn(); + (scheduler as any).tickFns.set("check", tick); + + await scheduler.drain(); + scheduler.wake("check"); + + expect((scheduler as any).timers.has("check")).toBe(false); + }); + }); +});