TRX/Checkmate
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

implement pagination, update metrics to use scheduler's aggregation

Browse Source
This commit is contained in:
Alex Holliday committed 2026-06-03 13:28:53 -07:00
1 parent 63fd3f6afd
commit 174fb048f7
3 files changed
+57 -114
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/service/infrastructure/JobQueues/IJobQueue.ts
+11 -1
View File
@@ -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<boolean>;
@@ -51,6 +61,6 @@ export interface IJobQueue {
updateJob(monitor: Monitor): Promise<void>;
shutdown(): Promise<void>;
getMetrics(): Promise<QueueMetrics>;
getJobs(): Promise<QueueJobSummary[]>;
getJobs(pagination: QueueJobsPagination): Promise<QueueJobsPage>;
flushQueues(): Promise<{ success: boolean }>;
}
server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts
+36 -59
View File
@@ -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<Scheduler["addJobs"]>[0][number];
// For discriminating job types
type MonitorJob = Omit<IJob, "data"> & { data: Monitor };
const MONITOR_TEMPLATES = new Set<string>(["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<QueueMetrics> => {
const stats = await this.scheduler.getStats();
const metrics = jobs.reduce<QueueMetrics>(
(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<QueueJobsPage> => {
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 () => {
server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts
+10 -54
View File
@@ -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<boolean>;
addJob(monitorId: string, monitor: Monitor): Promise<void>;
deleteJob(monitor: Monitor): Promise<void>;
pauseJob(monitor: Monitor): Promise<void>;
resumeJob(monitor: Monitor): Promise<void>;
updateJob(monitor: Monitor): Promise<void>;
shutdown(): Promise<void>;
getMetrics(): Promise<QueueMetrics>;
getJobs(): Promise<QueueJobSummary[]>;
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 () => {