Merge pull request #3697 from bluewave-labs/fix/jobs-pagination
fix: jobs pagination
This commit is contained in:
12 files changed
+233
-244
No files matched your search
@@ -4,6 +4,7 @@ import Box from "@mui/material/Box";
|
||||
import { TableJobs, TableFailedJobs } from "./components/TableJobs";
|
||||
import { TableWorkers } from "./components/TableWorkers";
|
||||
import { useTheme } from "@mui/material";
|
||||
import { useState } from "react";
|
||||
import { useTranslation, Trans } from "react-i18next";
|
||||
import { useGet, usePost } from "@/Hooks/UseApi";
|
||||
import type { QueueData } from "@/Types/Queue";
|
||||
@@ -15,16 +16,23 @@ import { SPACING } from "@/Utils/Theme/constants";
|
||||
export const TabQueue = () => {
|
||||
const theme = useTheme();
|
||||
const { t } = useTranslation();
|
||||
const [page, setPage] = useState(0);
|
||||
const [rowsPerPage, setRowsPerPage] = useState(10);
|
||||
const {
|
||||
data: queueData,
|
||||
isLoading,
|
||||
// error,
|
||||
refetch,
|
||||
} = useGet<QueueData>("/queue/all-metrics", {}, { refreshInterval: 1000 });
|
||||
} = useGet<QueueData>(
|
||||
`/queue/all-metrics?page=${page}&rowsPerPage=${rowsPerPage}`,
|
||||
{},
|
||||
{ refreshInterval: 1000, keepPreviousData: true }
|
||||
);
|
||||
|
||||
const { post, loading: isFlushing } = usePost();
|
||||
|
||||
const jobs = queueData?.jobs ?? [];
|
||||
const jobsCount = queueData?.count ?? 0;
|
||||
const workers = (queueData?.metrics?.workers ?? []).map((worker, idx) => ({
|
||||
...worker,
|
||||
id: idx,
|
||||
@@ -104,7 +112,17 @@ export const TabQueue = () => {
|
||||
/>
|
||||
</Typography>
|
||||
</Stack>
|
||||
<TableJobs jobs={jobs} />
|
||||
<TableJobs
|
||||
jobs={jobs}
|
||||
count={jobsCount}
|
||||
page={page}
|
||||
rowsPerPage={rowsPerPage}
|
||||
onPageChange={(_, newPage) => setPage(newPage)}
|
||||
onRowsPerPageChange={(event) => {
|
||||
setRowsPerPage(parseInt(event.target.value, 10));
|
||||
setPage(0);
|
||||
}}
|
||||
/>
|
||||
</Stack>
|
||||
<Stack gap={theme.spacing(3)}>
|
||||
<Stack gap={theme.spacing(1)}>
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { Table } from "@/Components/design-elements";
|
||||
import { Table, Pagination } from "@/Components/design-elements";
|
||||
import { Typography, useTheme } from "@mui/material";
|
||||
import prettyMilliseconds from "pretty-ms";
|
||||
import { formatTimestamp } from "@/Utils/TimeUtils";
|
||||
@@ -6,14 +6,27 @@ import { formatTimestamp } from "@/Utils/TimeUtils";
|
||||
import { useTranslation } from "react-i18next";
|
||||
import type { QueueJobFailure, QueueJobSummary, QueueMetrics } from "@/Types/Queue";
|
||||
import type { Header } from "@/Components/design-elements";
|
||||
import type { TablePaginationProps } from "@mui/material/TablePagination";
|
||||
|
||||
type QueueJobWithId = QueueJobSummary & { id: string | number };
|
||||
|
||||
interface TableJobsProps {
|
||||
jobs: QueueJobSummary[];
|
||||
count: number;
|
||||
page: number;
|
||||
rowsPerPage: number;
|
||||
onPageChange: TablePaginationProps["onPageChange"];
|
||||
onRowsPerPageChange: TablePaginationProps["onRowsPerPageChange"];
|
||||
}
|
||||
|
||||
export const TableJobs = ({ jobs }: TableJobsProps) => {
|
||||
export const TableJobs = ({
|
||||
jobs,
|
||||
count,
|
||||
page,
|
||||
rowsPerPage,
|
||||
onPageChange,
|
||||
onRowsPerPageChange,
|
||||
}: TableJobsProps) => {
|
||||
const { t } = useTranslation();
|
||||
const theme = useTheme();
|
||||
|
||||
@@ -151,18 +164,29 @@ export const TableJobs = ({ jobs }: TableJobsProps) => {
|
||||
];
|
||||
|
||||
return (
|
||||
<Table
|
||||
headers={headers}
|
||||
data={jobsWithId}
|
||||
getRowSx={(row) => ({
|
||||
...(row.lockedAt && {
|
||||
"& td": { backgroundColor: theme.palette.rowStatus.running },
|
||||
}),
|
||||
...(row.monitorActive === false && {
|
||||
"& td": { backgroundColor: theme.palette.rowStatus.paused },
|
||||
}),
|
||||
})}
|
||||
/>
|
||||
<>
|
||||
<Table
|
||||
headers={headers}
|
||||
data={jobsWithId}
|
||||
getRowSx={(row) => ({
|
||||
...(row.lockedAt && {
|
||||
"& td": { backgroundColor: theme.palette.rowStatus.running },
|
||||
}),
|
||||
...(row.monitorActive === false && {
|
||||
"& td": { backgroundColor: theme.palette.rowStatus.paused },
|
||||
}),
|
||||
})}
|
||||
/>
|
||||
<Pagination
|
||||
component="div"
|
||||
count={count}
|
||||
page={page}
|
||||
rowsPerPage={rowsPerPage}
|
||||
onPageChange={onPageChange}
|
||||
onRowsPerPageChange={onRowsPerPageChange}
|
||||
itemsOnPage={jobs.length}
|
||||
/>
|
||||
</>
|
||||
);
|
||||
};
|
||||
|
||||
|
||||
@@ -51,5 +51,6 @@ export type QueueJobSummary = {
|
||||
|
||||
export interface QueueData {
|
||||
jobs: QueueJobSummary[];
|
||||
count: number;
|
||||
metrics: QueueMetrics;
|
||||
}
|
||||
Generated
+4
-4
@@ -31,7 +31,7 @@
|
||||
"jmespath": "^0.16.0",
|
||||
"jsdom": "^26.1.0",
|
||||
"jsonwebtoken": "9.0.2",
|
||||
"less-simple-scheduler": "2.4.1",
|
||||
"less-simple-scheduler": "2.6.0",
|
||||
"mailersend": "^2.2.0",
|
||||
"mjml": "^5.0.0-alpha.4",
|
||||
"mongoose": "^8.3.3",
|
||||
@@ -10417,9 +10417,9 @@
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/less-simple-scheduler": {
|
||||
"version": "2.4.1",
|
||||
"resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.4.1.tgz",
|
||||
"integrity": "sha512-7Bgd5qLahFT3rv2QlY8QHrkkXLfUUgUpdDyabAiiy0WeEn1ay7Sacxcbge9TCZ34ZsySp+h1+NHeoTnWVU0vaQ==",
|
||||
"version": "2.6.0",
|
||||
"resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.6.0.tgz",
|
||||
"integrity": "sha512-Vi8ouiodLqe2fyh+DqCmxE5cCJZDI3xOyUAH4jcj9sf7GunLUzH71PBcYJy2P8lxTK3LxEkHtr1reymiogs6JQ==",
|
||||
"license": "ISC",
|
||||
"dependencies": {
|
||||
"mongodb": "7.2.0"
|
||||
|
||||
+1
-1
@@ -53,7 +53,7 @@
|
||||
"jmespath": "^0.16.0",
|
||||
"jsdom": "^26.1.0",
|
||||
"jsonwebtoken": "9.0.2",
|
||||
"less-simple-scheduler": "2.4.1",
|
||||
"less-simple-scheduler": "2.6.0",
|
||||
"mailersend": "^2.2.0",
|
||||
"mjml": "^5.0.0-alpha.4",
|
||||
"mongoose": "^8.3.3",
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { ISuperSimpleQueue } from "@/service/index.js";
|
||||
import { IJobQueue } from "@/service/index.js";
|
||||
import { getQueueJobsQueryValidation } from "@/validation/queueValidation.js";
|
||||
import { Request, Response, NextFunction } from "express";
|
||||
|
||||
const SERVICE_NAME = "JobQueueController";
|
||||
@@ -12,8 +13,8 @@ export interface IJobQueueController {
|
||||
|
||||
class JobQueueController implements IJobQueueController {
|
||||
static SERVICE_NAME = SERVICE_NAME;
|
||||
private jobQueue: ISuperSimpleQueue;
|
||||
constructor(jobQueue: ISuperSimpleQueue) {
|
||||
private jobQueue: IJobQueue;
|
||||
constructor(jobQueue: IJobQueue) {
|
||||
this.jobQueue = jobQueue;
|
||||
}
|
||||
|
||||
@@ -36,11 +37,12 @@ class JobQueueController implements IJobQueueController {
|
||||
|
||||
getJobs = async (req: Request, res: Response, next: NextFunction) => {
|
||||
try {
|
||||
const jobs = await this.jobQueue.getJobs();
|
||||
const pagination = getQueueJobsQueryValidation.parse(req.query);
|
||||
const { jobs, count } = await this.jobQueue.getJobs(pagination);
|
||||
return res.status(200).json({
|
||||
success: true,
|
||||
msg: "Queue jobs fetched successfully",
|
||||
data: jobs,
|
||||
data: { jobs, count },
|
||||
});
|
||||
} catch (error) {
|
||||
next(error);
|
||||
@@ -49,12 +51,13 @@ class JobQueueController implements IJobQueueController {
|
||||
|
||||
getAllMetrics = async (req: Request, res: Response, next: NextFunction) => {
|
||||
try {
|
||||
const jobs = await this.jobQueue.getJobs();
|
||||
const pagination = getQueueJobsQueryValidation.parse(req.query);
|
||||
const { jobs, count } = await this.jobQueue.getJobs(pagination);
|
||||
const metrics = await this.jobQueue.getMetrics();
|
||||
return res.status(200).json({
|
||||
success: true,
|
||||
msg: "Queue metrics fetched successfully",
|
||||
data: { jobs, metrics },
|
||||
data: { jobs, count, metrics },
|
||||
});
|
||||
} catch (error) {
|
||||
next(error);
|
||||
|
||||
@@ -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 }>;
|
||||
}
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
import { z } from "zod";
|
||||
|
||||
export const getQueueJobsQueryValidation = z.object({
|
||||
page: z.coerce.number().int().min(0).optional(),
|
||||
rowsPerPage: z.coerce.number().int().min(1).max(100).optional(),
|
||||
});
|
||||
@@ -26,6 +26,8 @@ const createScheduler = () => {
|
||||
removeJob: jest.fn(),
|
||||
getJob: jest.fn().mockResolvedValue(null),
|
||||
getJobs: jest.fn().mockResolvedValue([]),
|
||||
countJobs: jest.fn().mockResolvedValue(0),
|
||||
getStats: jest.fn().mockResolvedValue({ jobs: 0, activeJobs: 0, failingJobs: 0, totalRuns: 0, totalFailures: 0, jobsWithFailures: [] }),
|
||||
pauseJob: jest.fn().mockResolvedValue(true),
|
||||
resumeJob: jest.fn().mockResolvedValue(true),
|
||||
updateJob: jest.fn(),
|
||||
@@ -81,9 +83,6 @@ const createQueue = (overrides?: Record<string, unknown>) => {
|
||||
return { queue, ...defaults };
|
||||
};
|
||||
|
||||
// A lockedUntil far enough ahead that it is always > Date.now() at assertion time.
|
||||
const FAR_FUTURE = 4_000_000_000_000; // year 2096
|
||||
|
||||
const makeMonitor = (overrides?: Partial<Monitor>): Monitor =>
|
||||
({
|
||||
id: "mon-1",
|
||||
@@ -484,107 +483,55 @@ describe("LessSimpleQueue", () => {
|
||||
expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0, workers: [] });
|
||||
});
|
||||
|
||||
it("aggregates metrics from jobs", async () => {
|
||||
it("maps the store's aggregate counters through", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{
|
||||
id: "m1",
|
||||
template: "monitor-job",
|
||||
runCount: 10,
|
||||
failCount: 2,
|
||||
lastFailedAt: 200,
|
||||
lastFinishedAt: 100,
|
||||
lockedAt: null,
|
||||
lastError: "timeout",
|
||||
data: { url: "http://a.com", type: "http" },
|
||||
},
|
||||
{
|
||||
id: "m2",
|
||||
template: "monitor-job",
|
||||
runCount: 5,
|
||||
failCount: 0,
|
||||
lastFailedAt: 0,
|
||||
lastFinishedAt: 50,
|
||||
// live lock held by a worker
|
||||
lockedBy: "host-1:1234:abcd",
|
||||
lockedUntil: FAR_FUTURE,
|
||||
lockedAt: 999,
|
||||
data: { url: "http://b.com", type: "ping" },
|
||||
},
|
||||
]);
|
||||
(scheduler.getStats as jest.Mock).mockResolvedValue({
|
||||
jobs: 2,
|
||||
activeJobs: 1,
|
||||
failingJobs: 1,
|
||||
totalRuns: 15,
|
||||
totalFailures: 2,
|
||||
jobsWithFailures: [],
|
||||
});
|
||||
const metrics = await queue.getMetrics();
|
||||
|
||||
expect(metrics.jobs).toBe(2);
|
||||
expect(metrics.totalRuns).toBe(15);
|
||||
expect(metrics.totalFailures).toBe(2);
|
||||
expect(metrics.activeJobs).toBe(1);
|
||||
expect(metrics.failingJobs).toBe(1);
|
||||
expect(metrics.jobsWithFailures).toHaveLength(1);
|
||||
expect(metrics.jobsWithFailures[0]).toEqual(
|
||||
expect.objectContaining({ monitorId: "m1", failCount: 2, monitorUrl: "http://a.com", failReason: "timeout" })
|
||||
);
|
||||
expect(metrics).toMatchObject({ jobs: 2, activeJobs: 1, failingJobs: 1, totalRuns: 15, totalFailures: 2 });
|
||||
});
|
||||
|
||||
it("does not count as failing when lastFailedAt < lastFinishedAt", async () => {
|
||||
it("derives monitor fields for failing monitor jobs from job data", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{ id: "m1", template: "monitor-job", runCount: 10, failCount: 1, lastFailedAt: 50, lastFinishedAt: 100, data: { url: "a" } },
|
||||
]);
|
||||
(scheduler.getStats as jest.Mock).mockResolvedValue({
|
||||
jobs: 1,
|
||||
activeJobs: 0,
|
||||
failingJobs: 1,
|
||||
totalRuns: 10,
|
||||
totalFailures: 2,
|
||||
jobsWithFailures: [{ id: "m1", data: { url: "http://a.com", type: "http" }, failedAt: 200, failCount: 2, failReason: "timeout" }],
|
||||
});
|
||||
const metrics = await queue.getMetrics();
|
||||
expect(metrics.failingJobs).toBe(0);
|
||||
expect(metrics.jobsWithFailures).toHaveLength(1);
|
||||
expect(metrics.jobsWithFailures).toEqual([
|
||||
{ monitorId: "m1", monitorUrl: "http://a.com", monitorType: "http", failedAt: 200, failCount: 2, failReason: "timeout" },
|
||||
]);
|
||||
});
|
||||
|
||||
it("leaves monitor fields null for non-monitor (cleanup) jobs with failures", async () => {
|
||||
it("leaves monitor fields null for non-monitor (cleanup) failures", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{
|
||||
id: "cleanup-orphaned",
|
||||
template: "cleanup-orphaned",
|
||||
runCount: 1,
|
||||
failCount: 1,
|
||||
lastFailedAt: 100,
|
||||
lastFinishedAt: 50,
|
||||
lastError: "boom",
|
||||
data: undefined,
|
||||
},
|
||||
]);
|
||||
(scheduler.getStats as jest.Mock).mockResolvedValue({
|
||||
jobs: 1,
|
||||
activeJobs: 0,
|
||||
failingJobs: 1,
|
||||
totalRuns: 1,
|
||||
totalFailures: 1,
|
||||
jobsWithFailures: [{ id: "cleanup-orphaned", data: undefined, failedAt: 100, failCount: 1, failReason: "boom" }],
|
||||
});
|
||||
const metrics = await queue.getMetrics();
|
||||
expect(metrics.jobsWithFailures[0].monitorUrl).toBeNull();
|
||||
expect(metrics.jobsWithFailures[0].monitorType).toBeNull();
|
||||
expect(metrics.jobsWithFailures[0].failReason).toBe("boom");
|
||||
});
|
||||
});
|
||||
|
||||
// ── getMetrics: activeJobs (in-flight) ───────────────────────────────────────
|
||||
|
||||
describe("getMetrics - activeJobs", () => {
|
||||
it("counts jobs held by a live lock as active", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{ id: "m1", template: "monitor-job", lockedBy: "worker-a", lockedUntil: FAR_FUTURE, lockedAt: 100, data: { url: "a" } },
|
||||
{ id: "m2", template: "monitor-job", lockedBy: "worker-b", lockedUntil: FAR_FUTURE, lockedAt: 250, data: { url: "b" } },
|
||||
]);
|
||||
const metrics = await queue.getMetrics();
|
||||
expect(metrics.activeJobs).toBe(2);
|
||||
});
|
||||
|
||||
it("does not count unlocked jobs (lockedBy null)", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{ id: "m1", template: "monitor-job", lockedBy: null, lockedUntil: null, lockedAt: null, data: { url: "a" } },
|
||||
]);
|
||||
const metrics = await queue.getMetrics();
|
||||
expect(metrics.activeJobs).toBe(0);
|
||||
});
|
||||
|
||||
it("does not count jobs whose lock has expired (lockedUntil <= now)", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{ id: "m1", template: "monitor-job", lockedBy: "worker-a", lockedUntil: 1, lockedAt: 1, data: { url: "a" } },
|
||||
]);
|
||||
const metrics = await queue.getMetrics();
|
||||
expect(metrics.activeJobs).toBe(0);
|
||||
expect(metrics.jobsWithFailures[0]).toEqual({
|
||||
monitorId: "cleanup-orphaned",
|
||||
monitorUrl: null,
|
||||
monitorType: null,
|
||||
failedAt: 100,
|
||||
failCount: 1,
|
||||
failReason: "boom",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -627,7 +574,7 @@ describe("LessSimpleQueue", () => {
|
||||
data: { url: "http://a.com", type: "http", interval: 60000, geoCheckInterval: 300000, isActive: true },
|
||||
},
|
||||
]);
|
||||
const jobs = await queue.getJobs();
|
||||
const { jobs } = await queue.getJobs({});
|
||||
|
||||
expect(jobs).toHaveLength(1);
|
||||
expect(jobs[0]).toEqual({
|
||||
@@ -666,20 +613,44 @@ describe("LessSimpleQueue", () => {
|
||||
data: { url: "a", type: "http", interval: 60000 },
|
||||
},
|
||||
]);
|
||||
const jobs = await queue.getJobs();
|
||||
const { jobs } = await queue.getJobs({});
|
||||
expect(jobs[0].lastRunTook).toBeNull();
|
||||
});
|
||||
|
||||
it("leaves monitor fields null for non-monitor jobs", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([{ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true, data: undefined }]);
|
||||
const jobs = await queue.getJobs();
|
||||
const { jobs } = await queue.getJobs({});
|
||||
expect(jobs[0].monitorUrl).toBeNull();
|
||||
expect(jobs[0].monitorType).toBeNull();
|
||||
expect(jobs[0].monitorInterval).toBeNull();
|
||||
expect(jobs[0].monitorGeoInterval).toBeNull();
|
||||
expect(jobs[0].monitorActive).toBeNull();
|
||||
});
|
||||
|
||||
it("pushes pagination to the scheduler and returns its total count", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{ id: "m1", template: "monitor-job", active: true, data: { url: "a", type: "http", interval: 60000 } },
|
||||
]);
|
||||
(scheduler.countJobs as jest.Mock).mockResolvedValue(42);
|
||||
|
||||
const { jobs, count } = await queue.getJobs({ page: 2, rowsPerPage: 10 });
|
||||
|
||||
expect(scheduler.getJobs).toHaveBeenCalledWith({ skip: 20, limit: 10 });
|
||||
expect(count).toBe(42);
|
||||
expect(jobs).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("requests all jobs when rowsPerPage is omitted", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.countJobs as jest.Mock).mockResolvedValue(7);
|
||||
|
||||
const { count } = await queue.getJobs({});
|
||||
|
||||
expect(scheduler.getJobs).toHaveBeenCalledWith(undefined);
|
||||
expect(count).toBe(7);
|
||||
});
|
||||
});
|
||||
|
||||
// ── flushQueues ────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -583,7 +583,7 @@ describe("SuperSimpleQueue", () => {
|
||||
data: { url: "http://a.com", type: "http", interval: 60000, geoCheckInterval: 300000, isActive: true },
|
||||
},
|
||||
]);
|
||||
const jobs = await queue.getJobs();
|
||||
const { jobs } = await queue.getJobs({});
|
||||
|
||||
expect(jobs).toHaveLength(1);
|
||||
expect(jobs[0]).toEqual({
|
||||
@@ -613,14 +613,14 @@ describe("SuperSimpleQueue", () => {
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{ id: "m1", active: true, lockedAt: 999, runCount: 1, data: { url: "a", type: "http", interval: 60000 } },
|
||||
]);
|
||||
const jobs = await queue.getJobs();
|
||||
const { jobs } = await queue.getJobs({});
|
||||
expect(jobs[0].lastRunTook).toBeNull();
|
||||
});
|
||||
|
||||
it("handles jobs with missing data", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([{ id: "m1", active: true, data: null }]);
|
||||
const jobs = await queue.getJobs();
|
||||
const { jobs } = await queue.getJobs({});
|
||||
expect(jobs[0].monitorUrl).toBeNull();
|
||||
expect(jobs[0].monitorType).toBeNull();
|
||||
expect(jobs[0].monitorInterval).toBeNull();
|
||||
@@ -635,7 +635,7 @@ describe("SuperSimpleQueue", () => {
|
||||
{ id: "m1", active: true, data: { url: "a", type: "http", interval: 60000, isActive: true } },
|
||||
{ id: "m2", active: false, data: { url: "b", type: "http", interval: 60000, isActive: false } },
|
||||
]);
|
||||
const jobs = await queue.getJobs();
|
||||
const { jobs } = await queue.getJobs({});
|
||||
expect(jobs[0].monitorActive).toBe(true);
|
||||
expect(jobs[1].monitorActive).toBe(false);
|
||||
});
|
||||
@@ -645,10 +645,33 @@ describe("SuperSimpleQueue", () => {
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue([
|
||||
{ id: "m1", active: true, repeat: 120000, data: { url: "a", type: "http", interval: 60000, geoCheckInterval: 600000 } },
|
||||
]);
|
||||
const jobs = await queue.getJobs();
|
||||
const { jobs } = await queue.getJobs({});
|
||||
expect(jobs[0].repeat).toBe(120000);
|
||||
expect(jobs[0].monitorGeoInterval).toBe(600000);
|
||||
});
|
||||
|
||||
it("paginates jobs and returns the total count", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue(
|
||||
Array.from({ length: 25 }, (_, i) => ({ id: `m${i}`, active: true, data: { url: `a${i}`, type: "http", interval: 60000 } }))
|
||||
);
|
||||
|
||||
const { jobs, count } = await queue.getJobs({ page: 1, rowsPerPage: 10 });
|
||||
expect(count).toBe(25);
|
||||
expect(jobs).toHaveLength(10);
|
||||
expect(jobs[0].monitorId).toBe("m10");
|
||||
});
|
||||
|
||||
it("returns all jobs when rowsPerPage is omitted", async () => {
|
||||
const { queue, scheduler } = createQueue();
|
||||
(scheduler.getJobs as jest.Mock).mockResolvedValue(
|
||||
Array.from({ length: 25 }, (_, i) => ({ id: `m${i}`, active: true, data: { url: `a${i}`, type: "http", interval: 60000 } }))
|
||||
);
|
||||
|
||||
const { jobs, count } = await queue.getJobs({});
|
||||
expect(count).toBe(25);
|
||||
expect(jobs).toHaveLength(25);
|
||||
});
|
||||
});
|
||||
|
||||
// ── flushQueues ──────────────────────────────────────────────────────────
|
||||
|
||||
Reference in new issue
Block a user