From 174fb048f739bb449253eff3737adf93214d02a9 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 3 Jun 2026 13:28:53 -0700 Subject: [PATCH] implement pagination, update metrics to use scheduler's aggregation --- .../infrastructure/JobQueues/IJobQueue.ts | 12 ++- .../JobQueues/LessSimpleQueue.ts | 95 +++++++------------ .../JobQueues/SuperSimpleQueue.ts | 64 ++----------- 3 files changed, 57 insertions(+), 114 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/IJobQueue.ts b/server/src/service/infrastructure/JobQueues/IJobQueue.ts index 371a7c74a..ccbca625a 100644 --- a/server/src/service/infrastructure/JobQueues/IJobQueue.ts +++ b/server/src/service/infrastructure/JobQueues/IJobQueue.ts @@ -41,6 +41,16 @@ export type QueueJobSummary = { repeat: number | null; }; +export type QueueJobsPagination = { + page?: number; + rowsPerPage?: number; +}; + +export type QueueJobsPage = { + jobs: QueueJobSummary[]; + count: number; +}; + export interface IJobQueue { readonly serviceName: string; init(): Promise; @@ -51,6 +61,6 @@ export interface IJobQueue { updateJob(monitor: Monitor): Promise; shutdown(): Promise; getMetrics(): Promise; - getJobs(): Promise; + getJobs(pagination: QueueJobsPagination): Promise; flushQueues(): Promise<{ success: boolean }>; } diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index 51dcd9460..963b41267 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -1,22 +1,16 @@ import { IMonitorsRepository, IQueueWorkersRepository } from "@/repositories/index.js"; import { ILogger } from "@/utils/logger.js"; -import { MongoStore, Scheduler, IJob } from "less-simple-scheduler"; +import { MongoStore, Scheduler, IJob, AddJobInput } from "less-simple-scheduler"; import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; import { Monitor, supportsGeoCheck } from "@/types/monitor.js"; -import { type QueueMetrics, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; +import { type QueueMetrics, type QueueJobsPagination, type QueueJobsPage, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; import { type QueueMode } from "@/types/settings.js"; import { EnvConfig } from "@/service/system/settingsService.js"; const SERVICE_NAME = "JobQueue"; -// How long after its last heartbeat a worker is still considered alive. The -// model's TTL index uses the same window to garbage-collect stale records. -// Heartbeats are driven by the scheduler's `scheduler:heartbeat` event (emitted -// every lockMs/3, so ~5s with the default lockMs), comfortably inside this TTL. +// How long after its last heartbeat a worker is still considered alive. Must be longer than the scheduler's heartbeat const WORKER_TTL_MS = 30_000; -// Derived from the scheduler -type SchedulerJobInput = Parameters[0][number]; - // For discriminating job types type MonitorJob = Omit & { data: Monitor }; const MONITOR_TEMPLATES = new Set(["monitor-job", "geo-check-job"]); @@ -215,8 +209,8 @@ export class LessSimpleQueue implements IJobQueue { // Builds the scheduler inputs for a monitor: always a monitor-job, plus a // geo-check job when the monitor supports and enables geo checks. - private buildJobInputs = (monitor: Monitor): SchedulerJobInput[] => { - const inputs: SchedulerJobInput[] = [ + private buildJobInputs = (monitor: Monitor): AddJobInput[] => { + const inputs: AddJobInput[] = [ { id: monitor.id, template: "monitor-job", @@ -355,58 +349,39 @@ export class LessSimpleQueue implements IJobQueue { this.scheduler.stop(); }; - getMetrics = async () => { - const jobs = await this.scheduler.getJobs(); - const now = Date.now(); + getMetrics = async (): Promise => { + const stats = await this.scheduler.getStats(); - const metrics = jobs.reduce( - (acc, job) => { - const runCount = job.runCount ?? 0; - const failCount = job.failCount ?? 0; - const lastFailedAt = job.lastFailedAt ?? 0; - const lastFinishedAt = job.lastFinishedAt ?? 0; - acc.totalRuns += runCount; - acc.totalFailures += failCount; - acc.jobs++; - if (failCount > 0 && lastFailedAt >= lastFinishedAt) { - acc.failingJobs++; - } + const jobsWithFailures = stats.jobsWithFailures.map((failure) => { + // `data` is untyped at the store level: for monitor jobs it's the Monitor, + // for system jobs (e.g. cleanup) it has no url/type => null + const monitor = failure.data as Monitor | undefined; + return { + monitorId: failure.id, + monitorUrl: monitor?.url ?? null, + monitorType: monitor?.type ?? null, + failedAt: failure.failedAt, + failCount: failure.failCount, + failReason: failure.failReason, + }; + }); - // A job currently held by a live lock is in flight on some worker - if (job.lockedBy && job.lockedUntil && job.lockedUntil > now) { - acc.activeJobs++; - } - - if (failCount > 0) { - const monitor = isMonitorJob(job) ? job.data : undefined; - acc.jobsWithFailures.push({ - monitorId: job.id, - monitorUrl: monitor?.url || null, - monitorType: monitor?.type || null, - failedAt: job.lastFailedAt ?? null, - failCount, - failReason: job.lastError ?? null, - }); - } - return acc; - }, - { - jobs: 0, - activeJobs: 0, - failingJobs: 0, - jobsWithFailures: [], - totalRuns: 0, - totalFailures: 0, - workers: [], - } - ); - metrics.workers = await this.workersRepository.findRecent(WORKER_TTL_MS); - return metrics; + const workers = await this.workersRepository.findRecent(WORKER_TTL_MS); + return { + jobs: stats.jobs, + activeJobs: stats.activeJobs, + failingJobs: stats.failingJobs, + totalRuns: stats.totalRuns, + totalFailures: stats.totalFailures, + jobsWithFailures, + workers, + }; }; - getJobs = async () => { - const jobs = await this.scheduler.getJobs(); - return jobs.map((job) => { + getJobs = async ({ page = 0, rowsPerPage = 0 }: QueueJobsPagination): Promise => { + const listOptions = rowsPerPage > 0 ? { skip: Math.max(page, 0) * rowsPerPage, limit: rowsPerPage } : undefined; + const [jobs, count] = await Promise.all([this.scheduler.getJobs(listOptions), this.scheduler.countJobs()]); + const mapped = jobs.map((job) => { const monitor = isMonitorJob(job) ? job.data : undefined; return { @@ -430,6 +405,8 @@ export class LessSimpleQueue implements IJobQueue { lastFailedAt: job.lastFailedAt ?? null, }; }); + + return { jobs: mapped, count }; }; flushQueues = async () => { diff --git a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index ee542d2be..42409696b 100644 --- a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -6,61 +6,11 @@ import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js"; import { hostname } from "node:os"; import { randomUUID } from "node:crypto"; import { QueueMode } from "@/types/settings.js"; +import { IJobQueue, type QueueJobSummary } from "@/service/infrastructure/JobQueues/IJobQueue.js"; const SERVICE_NAME = "JobQueue"; -type QueueJobFailure = { - monitorId: string | number; - monitorUrl: string | null; - monitorType: string | null; - failedAt: number | null; - failCount: number; - failReason: string | null; -}; - -type QueueMetrics = { - jobs: number; - activeJobs: number; - failingJobs: number; - jobsWithFailures: QueueJobFailure[]; - totalRuns: number; - totalFailures: number; -}; - -type QueueJobSummary = { - monitorId: string | number; - monitorUrl: string | null; - monitorType: string | null; - monitorInterval: number | null; - monitorGeoInterval: number | null; - monitorActive: boolean | null; - active: boolean; - lockedAt: number | null; - runCount: number; - failCount: number; - failReason: string | null; - lastRunAt: number | null; - lastFinishedAt: number | null; - lastRunTook: number | null; - lastFailedAt: number | null; - repeat: number | null; -}; - -export interface ISuperSimpleQueue { - readonly serviceName: string; - init(): Promise; - addJob(monitorId: string, monitor: Monitor): Promise; - deleteJob(monitor: Monitor): Promise; - pauseJob(monitor: Monitor): Promise; - resumeJob(monitor: Monitor): Promise; - updateJob(monitor: Monitor): Promise; - shutdown(): Promise; - getMetrics(): Promise; - getJobs(): Promise; - flushQueues(): Promise<{ success: boolean }>; -} - -export class SuperSimpleQueue implements ISuperSimpleQueue { +export class SuperSimpleQueue implements IJobQueue { static SERVICE_NAME = SERVICE_NAME; private logger: ILogger; @@ -391,9 +341,9 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { return metrics; }; - getJobs = async () => { + getJobs = async ({ page = 0, rowsPerPage = 0 }: { page?: number; rowsPerPage?: number }): Promise<{ jobs: QueueJobSummary[]; count: number }> => { const jobs = await this.scheduler.getJobs(); - return jobs.map((job) => { + const mapped = jobs.map((job) => { return { monitorId: job.id, monitorUrl: job.data?.url || null, @@ -415,6 +365,12 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { lastFailedAt: job.lastFailedAt ?? null, }; }); + + const start = Math.max(page, 0) * rowsPerPage; + return { + jobs: rowsPerPage > 0 ? mapped.slice(start, start + rowsPerPage) : mapped, + count: mapped.length, + }; }; flushQueues = async () => {