mirror of
https://github.com/Crosstalk-Solutions/project-nomad.git
synced 2026-10-16 21:00:03 +02:00
Merge 2aa55a43ec into 54ee30a449
This commit is contained in:
commit
e29bccde8b
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<InFlightBenchmark[]> {
|
||||
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<string | undefined> {
|
||||
const job = await this.getJob(benchmarkId)
|
||||
return job ? await job.getState() : undefined
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
38
admin/app/utils/benchmark_status.ts
Normal file
38
admin/app/utils/benchmark_status.ts
Normal file
|
|
@ -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<BenchmarkStatus> = 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
|
||||
}
|
||||
67
admin/tests/unit/benchmark_status.spec.ts
Normal file
67
admin/tests/unit/benchmark_status.spec.ts
Normal file
|
|
@ -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' }
|
||||
)
|
||||
})
|
||||
Loading…
Reference in New Issue
Block a user