From 90bbc764fc9efb1e8aed57aa40e71bfd9a3f6b05 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 15 Jun 2026 11:26:49 -0700 Subject: [PATCH] readd queue-worker domain, add heartbewat to db-queue --- server/src/config/services.ts | 3 ++ .../queue-workers/queue-worker.model.ts | 22 +++++++++++++ .../queue-worker.repository.interface.ts | 8 +++++ .../queue-worker.repository.mongo.ts | 33 +++++++++++++++++++ .../domain/queue-workers/queue-worker.type.ts | 18 ++++++++++ server/src/worker/worker.db-queue.ts | 30 +++++++++++++++-- server/src/worker/worker.interface.ts | 1 + 7 files changed, 113 insertions(+), 2 deletions(-) create mode 100644 server/src/domain/queue-workers/queue-worker.model.ts create mode 100644 server/src/domain/queue-workers/queue-worker.repository.interface.ts create mode 100644 server/src/domain/queue-workers/queue-worker.repository.mongo.ts create mode 100644 server/src/domain/queue-workers/queue-worker.type.ts diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 341e7da2f..2afd263c9 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -107,6 +107,7 @@ 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"; +import MongoQueueWorkersRepository from "@/domain/queue-workers/queue-worker.repository.mongo.js"; export type InitializedServices = { settingsService: ISettingsService; @@ -177,6 +178,7 @@ export const initializeServices = async ({ const teamsRepository = new MongoTeamsRepository(); const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository(); const jobsRepository = new MongoJobsRepository(); + const queueWorkersRepository = new MongoQueueWorkersRepository(); // Inject settings repository into settings service (now that DB is connected) (settingsService as SettingsService).setRepository(settingsRepository); @@ -301,6 +303,7 @@ export const initializeServices = async ({ geoCheckPipeline, reactorDispatcher, workerHelper, + queueWorkersRepository, envSettings.queueMode, envSettings.queuePrimaryProcesses ); diff --git a/server/src/domain/queue-workers/queue-worker.model.ts b/server/src/domain/queue-workers/queue-worker.model.ts new file mode 100644 index 000000000..d1a7aa86b --- /dev/null +++ b/server/src/domain/queue-workers/queue-worker.model.ts @@ -0,0 +1,22 @@ +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"; + +// Single source of truth for worker liveness: the TTL that GCs stale records and +// the window readers use to count a worker as alive. Heartbeat cadence must stay well below it. +export const WORKER_TTL_SECONDS = 180; + +const QueueWorkerSchema = new Schema( + { + _id: { type: String, required: true }, + mode: { type: String, enum: QueueModes, required: true }, + lastSeenAt: { type: Date, default: Date.now, expires: WORKER_TTL_SECONDS }, + }, + { 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 new file mode 100644 index 000000000..e4522c916 --- /dev/null +++ b/server/src/domain/queue-workers/queue-worker.repository.interface.ts @@ -0,0 +1,8 @@ +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 new file mode 100644 index 000000000..35565bd03 --- /dev/null +++ b/server/src/domain/queue-workers/queue-worker.repository.mongo.ts @@ -0,0 +1,33 @@ +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 new file mode 100644 index 000000000..a83c831ad --- /dev/null +++ b/server/src/domain/queue-workers/queue-worker.type.ts @@ -0,0 +1,18 @@ +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/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 9331a6927..b26bb4e93 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -13,9 +13,13 @@ 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 { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; +import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; +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; const CONCURRENCY: Record = { check: 500, @@ -29,7 +33,7 @@ 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 timers = new Map(); private inFlight: Record = { check: 0, "geo-check": 0, @@ -49,6 +53,7 @@ export class DBQueueWorker implements IWorker { private geoCheckPipeline: ICheckPipeline, private dispatcher: IReactorDispatcher, private helper: IWorkerHelper, + private queueWorkersRepository: IQueueWorkersRepository, private queueMode: QueueMode, private queuePrimaryProcesses: boolean ) {} @@ -68,6 +73,7 @@ export class DBQueueWorker implements IWorker { geoCheckPipeline: ICheckPipeline, dispatcher: IReactorDispatcher, helper: IWorkerHelper, + queueWorkersRepository: IQueueWorkersRepository, queueMode: QueueMode, queuePrimaryProcesses: boolean ): Promise { @@ -82,6 +88,7 @@ export class DBQueueWorker implements IWorker { geoCheckPipeline, dispatcher, helper, + queueWorkersRepository, queueMode, queuePrimaryProcesses ); @@ -247,9 +254,25 @@ export class DBQueueWorker implements IWorker { await this.jobsRepository.upsertJob(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(); } @@ -291,9 +314,12 @@ export class DBQueueWorker implements IWorker { clearTimeout(timer); } this.timers.clear(); // Clear the map out + // Deregister promptly so the worker drops off diagnostics before the TTL would + await this.queueWorkersRepository.deleteById(this.workerId); }; getMetrics = async (): Promise => { const rows = await this.jobsRepository.findAll(); + const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); const now = Date.now(); return rows.reduce( (acc, job) => { @@ -321,7 +347,7 @@ export class DBQueueWorker implements IWorker { jobsWithFailures: [], totalRuns: 0, totalFailures: 0, - workers: [{ workerId: this.workerId, mode: this.queueMode, lastSeenAt: now }], + workers, } ); }; diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index 3b5eb0bd7..ccaa79d12 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -2,6 +2,7 @@ import { Monitor } from "@/domain/monitors/monitor.types.js"; import { Check } from "@/domain/checks/check.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"; export type MonitorEvaluation = { monitor: Monitor;