diff --git a/admin/app/controllers/benchmark_controller.ts b/admin/app/controllers/benchmark_controller.ts index 714c61a8..d5e0ffa8 100644 --- a/admin/app/controllers/benchmark_controller.ts +++ b/admin/app/controllers/benchmark_controller.ts @@ -3,6 +3,7 @@ import type { HttpContext } from '@adonisjs/core/http' import { BenchmarkService } from '#services/benchmark_service' import { runBenchmarkValidator, submitBenchmarkValidator } from '#validators/benchmark' import { RunBenchmarkJob } from '#jobs/run_benchmark_job' +import { resolveBenchmarkStatus } from '../utils/benchmark_status.js' import type { BenchmarkType } from '../../types/benchmark.js' import { randomUUID } from 'node:crypto' import logger from '@adonisjs/core/services/logger' @@ -19,8 +20,9 @@ export default class BenchmarkController { const benchmarkType: BenchmarkType = payload.benchmark_type || 'full' const runSync = request.input('sync') === 'true' || request.input('sync') === true - // Check if a benchmark is already running - const status = this.benchmarkService.getStatus() + // Check if a benchmark is already running. A sync run is the no-Redis path, and + // an unreachable Redis makes a queue read wait rather than fail, so it doesn't ask. + const status = runSync ? this.benchmarkService.getStatus() : await this.currentStatus() if (status.status !== 'idle') { return response.status(409).send({ success: false, @@ -83,7 +85,7 @@ export default class BenchmarkController { * Run a system-only benchmark (CPU, memory, disk) */ async runSystem({ response }: HttpContext) { - const status = this.benchmarkService.getStatus() + const status = await this.currentStatus() if (status.status !== 'idle') { return response.status(409).send({ success: false, @@ -109,7 +111,7 @@ export default class BenchmarkController { * Run an AI-only benchmark */ async runAI({ response }: HttpContext) { - const status = this.benchmarkService.getStatus() + const status = await this.currentStatus() if (status.status !== 'idle') { return response.status(409).send({ success: false, @@ -258,7 +260,22 @@ export default class BenchmarkController { * Get current benchmark status */ async status({}: HttpContext) { - return this.benchmarkService.getStatus() + return this.currentStatus() + } + + /** + * The run status as both processes see it: benchmarks execute in the queue + * worker, so it comes from the queue, not this process's BenchmarkService (#1326). + * A queue read that errors falls back to the local status, as before this change. + */ + private async currentStatus() { + const local = this.benchmarkService.getStatus() + try { + return resolveBenchmarkStatus(await RunBenchmarkJob.getInFlight(), local) + } catch (error) { + logger.warn({ err: error }, '[BenchmarkController] Could not read the benchmark queue') + return local + } } /** diff --git a/admin/app/jobs/run_benchmark_job.ts b/admin/app/jobs/run_benchmark_job.ts index 962e663d..bee244bc 100644 --- a/admin/app/jobs/run_benchmark_job.ts +++ b/admin/app/jobs/run_benchmark_job.ts @@ -4,6 +4,7 @@ import { BenchmarkService } from '#services/benchmark_service' import type { RunBenchmarkJobParams } from '../../types/benchmark.js' import logger from '@adonisjs/core/services/logger' import { DockerService } from '#services/docker_service' +import type { InFlightBenchmark } from '../utils/benchmark_status.js' export class RunBenchmarkJob { static get queue() { @@ -21,6 +22,14 @@ export class RunBenchmarkJob { const dockerService = new DockerService() const benchmarkService = new BenchmarkService(dockerService) + // Publish each stage on the job so the web process can answer /api/benchmark/status + // and the double-run guard (#1326). Best-effort: a missed update only makes the + // reported stage lag, and must not fail the benchmark. + benchmarkService.onStatusChange = (status) => { + job.updateProgress({ status }).catch((err) => { + logger.warn(`[RunBenchmarkJob] Could not publish stage ${status}: ${err.message}`) + }) + } try { let result @@ -94,6 +103,22 @@ export class RunBenchmarkJob { return await queue.getJob(benchmarkId) } + /** Benchmark jobs that are running or still queued to run. */ + static async getInFlight(): Promise { + const queue = QueueService.getInstance().getQueue(this.queue) + const [active, waiting, delayed] = await Promise.all([ + queue.getJobs(['active']), + queue.getJobs(['waiting']), + queue.getJobs(['delayed']), + ]) + const tag = (state: InFlightBenchmark['state']) => (job: Job) => ({ + state, + benchmarkId: (job.data as RunBenchmarkJobParams | undefined)?.benchmark_id ?? job.id ?? null, + progress: job.progress, + }) + return [...active.map(tag('active')), ...waiting.map(tag('waiting')), ...delayed.map(tag('delayed'))] + } + static async getJobState(benchmarkId: string): Promise { const job = await this.getJob(benchmarkId) return job ? await job.getState() : undefined diff --git a/admin/app/services/benchmark_service.ts b/admin/app/services/benchmark_service.ts index a2d4e23a..dad969a7 100644 --- a/admin/app/services/benchmark_service.ts +++ b/admin/app/services/benchmark_service.ts @@ -212,6 +212,12 @@ export class BenchmarkService { private currentStatus: BenchmarkStatus = 'idle' private currentStages: BenchmarkStageDescriptor[] = [] private telemetry: BenchmarkTelemetrySampler | null = null + /** + * Called on every stage change. RunBenchmarkJob uses it to publish the stage on the + * BullMQ job, where the web process can read it — this instance's own status is + * only visible inside the worker (#1326). + */ + public onStatusChange?: (status: BenchmarkStatus) => void constructor(private dockerService: DockerService) {} @@ -1733,6 +1739,7 @@ export class BenchmarkService { private _updateStatus(status: BenchmarkStatus, message: string) { this.currentStatus = status this.telemetry?.setStage(status) + this.onStatusChange?.(status) const progress: BenchmarkProgress = { status, diff --git a/admin/app/utils/benchmark_status.ts b/admin/app/utils/benchmark_status.ts new file mode 100644 index 00000000..d635797b --- /dev/null +++ b/admin/app/utils/benchmark_status.ts @@ -0,0 +1,38 @@ +import type { BenchmarkStatus } from '../../types/benchmark.js' + +/** A benchmark job still in the queue, as RunBenchmarkJob.getInFlight() reports it. */ +export interface InFlightBenchmark { + state: 'active' | 'waiting' | 'delayed' + benchmarkId: string | null + /** The job's BullMQ progress; the worker writes `{ status }` into it at every stage. */ + progress: unknown +} + +const NOT_RUNNING: ReadonlySet = new Set(['idle', 'completed', 'error']) + +/** + * Decide the current benchmark status from the queue. + * + * Benchmarks run in the queue worker, so the web process's BenchmarkService + * never sees them and its in-memory status always reads idle (#1326). The queue + * is the one place both processes can see: a job that is active or still + * waiting means a run is in flight, and the worker reports its stage through + * the job's progress. `local` is the web process's own status, which is only + * ever non-idle for a `sync=true` run executing in this process. + */ +export function resolveBenchmarkStatus( + inFlight: InFlightBenchmark[], + local: { status: BenchmarkStatus; benchmarkId: string | null } +): { status: BenchmarkStatus; benchmarkId: string | null } { + const active = inFlight.find((j) => j.state === 'active') + if (active) { + const reported = (active.progress as { status?: BenchmarkStatus } | null)?.status + // Before the worker's first stage update, or in the moment between it reporting + // `completed` and the job leaving the active set, the run is still in flight + const status = reported && !NOT_RUNNING.has(reported) ? reported : 'starting' + return { status, benchmarkId: active.benchmarkId } + } + const queued = inFlight.find((j) => j.state === 'waiting' || j.state === 'delayed') + if (queued) return { status: 'starting', benchmarkId: queued.benchmarkId } + return local +} diff --git a/admin/tests/unit/benchmark_status.spec.ts b/admin/tests/unit/benchmark_status.spec.ts new file mode 100644 index 00000000..5da87967 --- /dev/null +++ b/admin/tests/unit/benchmark_status.spec.ts @@ -0,0 +1,67 @@ +/** + * Tests for the benchmark status decision — what /api/benchmark/status and the + * double-run guard report while a benchmark is queued or running. + * + * Benchmarks run in the queue worker, so the web process's in-memory status + * always read idle and a second run was accepted mid-flight (#1326). Status now + * comes from the queue; this is the pure decision over what the queue holds. + * + * Pure functions only — no MySQL, Redis, or queue worker needed: + * npm run test:unit + */ +import * as assert from 'node:assert/strict' +import { test } from 'node:test' + +import { resolveBenchmarkStatus } from '../../app/utils/benchmark_status.js' + +const IDLE = { status: 'idle' as const, benchmarkId: null } + +test('benchmark status: an empty queue falls back to the local status', () => { + assert.deepEqual(resolveBenchmarkStatus([], IDLE), IDLE) + // A sync=true run executes in this process and is only visible locally + const localRun = { status: 'running_cpu' as const, benchmarkId: 'local-1' } + assert.deepEqual(resolveBenchmarkStatus([], localRun), localRun) +}) + +test('benchmark status: an active job reports the stage the worker published', () => { + assert.deepEqual( + resolveBenchmarkStatus( + [{ state: 'active', benchmarkId: 'b1', progress: { status: 'running_disk_read' } }], + IDLE + ), + { status: 'running_disk_read', benchmarkId: 'b1' } + ) +}) + +test('benchmark status: an active job with no usable stage yet is starting, never idle', () => { + // BullMQ progress starts at 0 before the worker's first update + for (const progress of [0, null, {}, { status: 'idle' }, { status: 'completed' }, { status: 'error' }]) { + assert.deepEqual( + resolveBenchmarkStatus([{ state: 'active', benchmarkId: 'b1', progress }], IDLE), + { status: 'starting', benchmarkId: 'b1' }, + `progress ${JSON.stringify(progress)}` + ) + } +}) + +test('benchmark status: a job still waiting in the queue counts as starting', () => { + for (const state of ['waiting', 'delayed'] as const) { + assert.deepEqual( + resolveBenchmarkStatus([{ state, benchmarkId: 'b2', progress: 0 }], IDLE), + { status: 'starting', benchmarkId: 'b2' } + ) + } +}) + +test('benchmark status: the running job wins over one queued behind it', () => { + assert.deepEqual( + resolveBenchmarkStatus( + [ + { state: 'waiting', benchmarkId: 'queued', progress: 0 }, + { state: 'active', benchmarkId: 'running', progress: { status: 'running_ai' } }, + ], + IDLE + ), + { status: 'running_ai', benchmarkId: 'running' } + ) +})