From f740eef0fad0c45676b103fc3068586475e2fb5e Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Thu, 18 Jun 2026 11:05:57 -0700 Subject: [PATCH] thread prefetched worker, remove dead code --- server/src/worker/worker.check-evaluator.ts | 7 +-- server/src/worker/worker.check-pipeline.ts | 18 -------- server/src/worker/worker.db-queue.ts | 13 +++--- server/src/worker/worker.helper.ts | 49 +-------------------- 4 files changed, 13 insertions(+), 74 deletions(-) diff --git a/server/src/worker/worker.check-evaluator.ts b/server/src/worker/worker.check-evaluator.ts index bbae96d69..34c444271 100644 --- a/server/src/worker/worker.check-evaluator.ts +++ b/server/src/worker/worker.check-evaluator.ts @@ -1,11 +1,12 @@ import { MonitorStatusResponse } from "@/types/network.js"; import { MonitorEvaluation } from "@/worker/worker.interface.js"; import { Check } from "@/domain/checks/check.type.js"; +import { Monitor } from "@/domain/monitors/monitor.types.js"; import { IMonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; import { IStatusService } from "@/service/statusService.js"; export interface ICheckEvaluator { - evaluate(status: MonitorStatusResponse, check: Check): Promise; + evaluate(status: MonitorStatusResponse, check: Check, monitor: Monitor): Promise; } export class CheckEvaluator implements ICheckEvaluator { @@ -13,11 +14,11 @@ export class CheckEvaluator implements ICheckEvaluator { private statusService: IStatusService, private monitorStatusPolicy: IMonitorStatusPolicy ) {} - evaluate = async (status: MonitorStatusResponse, check: Check) => { + evaluate = async (status: MonitorStatusResponse, check: Check, monitor: Monitor) => { // **************************** // Step 3: Evaluate and return result to reactors // **************************** - const statusChangeResult = await this.statusService.updateMonitorStatus(status, check); + const statusChangeResult = await this.statusService.updateMonitorStatus(status, check, monitor); // Step 5. Get decisions and create an evaluation obj const decision = this.monitorStatusPolicy.evaluate(statusChangeResult); diff --git a/server/src/worker/worker.check-pipeline.ts b/server/src/worker/worker.check-pipeline.ts index c472c4bfa..045cea1ad 100644 --- a/server/src/worker/worker.check-pipeline.ts +++ b/server/src/worker/worker.check-pipeline.ts @@ -7,8 +7,6 @@ import { IBufferService } from "@/service/bufferService.js"; import { AppError } from "@/utils/AppError.js"; import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; import { IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js"; -import { ICheckProducer } from "@/worker/worker.check-producer.js"; -import { ICheckEvaluator } from "@/worker/worker.check-evaluator.js"; const SERVICE_NAME = "CheckPipeline"; @@ -16,22 +14,6 @@ export interface ICheckPipeline { run(monitor: Monitor): Promise; // null = skipped } -export class CheckPipeline implements ICheckPipeline { - constructor( - private checkProducer: ICheckProducer, - private checkEvaluator: ICheckEvaluator - ) {} - - run = async (monitor: Monitor): Promise => { - const checkResult = await this.checkProducer.produce(monitor); - if (!checkResult) { - return null; // skipped due to maintenance / geo guards / other guards - } - const evaluation = await this.checkEvaluator.evaluate(checkResult.status, checkResult.check); - return evaluation; - }; -} - export class GeoChecksPipeline implements ICheckPipeline { constructor( private maintenanceWindowsRepository: IMaintenanceWindowsRepository, diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 9088dc3b3..bef7a838e 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -151,7 +151,7 @@ export class DBQueueWorker implements IWorker { private runCheck = async (job: Job) => { if (!job.refId) return; - const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId + const monitor = await this.monitorsRepository.findByIdLean(job.refId); // job row has no teamId if (!monitor) return; await this.checkProducer.produce(monitor); }; @@ -162,19 +162,22 @@ export class DBQueueWorker implements IWorker { private runEvaluate = async (job: Job) => { if (!job.refId) return; - const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId + const monitor = await this.monitorsRepository.findByIdLean(job.refId); // job row has no teamId if (!monitor) return; const checks = await this.checksRepository.findUnevaluatedByMonitorId(job.refId, monitor.lastEvaluatedAt); + + let current = monitor; for (const check of checks) { const status = this.checkService.toStatusResponse(check); - const evaluation = await this.checkEvaluator.evaluate(status, check); + const evaluation = await this.checkEvaluator.evaluate(status, check, current); await this.dispatcher.dispatch(evaluation); // Handle incidents and notifications - await this.monitorsRepository.updateById(job.refId, monitor.teamId, { lastEvaluatedAt: this.checkService.toLastEvaluatedAt(check) }); + await this.monitorsRepository.updateById(job.refId, current.teamId, { lastEvaluatedAt: this.checkService.toLastEvaluatedAt(check) }); + current = evaluation.statusChange.monitor; // fresh statusWindow/status/counters for the next check } }; private runGeoCheck = async (job: Job) => { - const [monitor] = await this.monitorsRepository.findByIds([job.refId!]); + const monitor = await this.monitorsRepository.findByIdLean(job.refId!); if (monitor) await this.geoCheckPipeline.run(monitor); // returns null; no evaluate handoff }; diff --git a/server/src/worker/worker.helper.ts b/server/src/worker/worker.helper.ts index 7b60a26ba..3d2d2e183 100644 --- a/server/src/worker/worker.helper.ts +++ b/server/src/worker/worker.helper.ts @@ -1,5 +1,4 @@ const SERVICE_NAME = "JobQueueHelper"; -import type { Monitor } from "@/domain/monitors/monitor.types.js"; import { ISettingsService } from "@/domain/app-settings/app-settings.service.js"; import { ICheckService } from "@/domain/checks/check.service.js"; import { CHECK_TTL_SENTINEL } from "@/domain/checks/check.type.js"; @@ -11,13 +10,9 @@ import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interf import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js"; import { ILogger } from "@/utils/logger.js"; -import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; -import { ICheckPipeline } from "@/worker/worker.check-pipeline.js"; export interface IWorkerHelper { readonly serviceName: string; - getHeartbeatJob(): (monitor: Monitor) => Promise; - getHeartbeatGeoJob(): (monitor: Monitor) => Promise; getCleanupOrphanedJob(): () => Promise; getCleanupRetentionJob(): () => Promise; } @@ -49,9 +44,6 @@ export class WorkerHelper implements IWorkerHelper { private checksRepository: IChecksRepository; private incidentsRepository: IIncidentsRepository; private geoChecksRepository: IGeoChecksRepository; - private reactorDispatcher: IReactorDispatcher; - private checkPipeline: ICheckPipeline; - private geoCheckPipeline: ICheckPipeline; constructor( logger: ILogger, @@ -63,10 +55,7 @@ export class WorkerHelper implements IWorkerHelper { monitorStatsRepository: IMonitorStatsRepository, checksRepository: IChecksRepository, incidentsRepository: IIncidentsRepository, - geoChecksRepository: IGeoChecksRepository, - reactorDispatcher: IReactorDispatcher, - checkPipeline: ICheckPipeline, - geoCheckPipeline: ICheckPipeline + geoChecksRepository: IGeoChecksRepository ) { this.logger = logger; this.checkService = checkService; @@ -78,48 +67,12 @@ export class WorkerHelper implements IWorkerHelper { this.checksRepository = checksRepository; this.incidentsRepository = incidentsRepository; this.geoChecksRepository = geoChecksRepository; - this.reactorDispatcher = reactorDispatcher; - this.checkPipeline = checkPipeline; - this.geoCheckPipeline = geoCheckPipeline; } get serviceName() { return WorkerHelper.SERVICE_NAME; } - getHeartbeatJob = () => { - return async (monitor: Monitor) => { - try { - const evaluation = await this.checkPipeline.run(monitor); - if (evaluation) await this.reactorDispatcher.dispatch(evaluation); - } catch (error: unknown) { - this.logger.warn({ - message: error instanceof Error ? error.message : "Unknown error", - service: SERVICE_NAME, - method: "getMonitorJob", - stack: error instanceof Error ? error.stack : undefined, - }); - throw error; - } - }; - }; - - getHeartbeatGeoJob = () => { - return async (monitor: Monitor) => { - try { - await this.geoCheckPipeline.run(monitor); - } catch (error: unknown) { - this.logger.error({ - message: error instanceof Error ? error.message : "Unknown error", - service: SERVICE_NAME, - method: "getHeartbeatGeoJob", - stack: error instanceof Error ? error.stack : undefined, - }); - // Don't throw - geo check failures shouldn't crash the job scheduler - } - }; - }; - getCleanupOrphanedJob = () => { return async () => { try {