refactor(scheduler): single continuous loop per run with trace ID
- replace two @Interval jobs (activate + process-one-step) with one job that spawns an async loop per run, eliminating 1-second gaps between steps - track active runs via Set to prevent duplicate processing - wrap each run loop in traceStorage.run() with a fresh UUID so all log lines for a run share a trace ID without an HTTP request context
This commit is contained in:
@@ -1,8 +1,10 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { TraceLogger } from '../common/trace-logger';
|
||||
import { traceStorage } from '../common/trace-context';
|
||||
import { Interval } from '@nestjs/schedule';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import * as crypto from 'crypto';
|
||||
import { chromium } from 'playwright';
|
||||
import type { Browser, BrowserContext, Page } from 'playwright';
|
||||
import { ScenarioRunEntity } from './scenario-run.entity';
|
||||
@@ -27,7 +29,7 @@ interface BrowserHandle {
|
||||
@Injectable()
|
||||
export class ScenarioSchedulerService {
|
||||
private readonly logger = new TraceLogger(ScenarioSchedulerService.name);
|
||||
private isProcessingStep = false;
|
||||
private readonly activeRuns = new Set<number>();
|
||||
private readonly runBrowsers = new Map<number, BrowserHandle>();
|
||||
|
||||
constructor(
|
||||
@@ -44,44 +46,48 @@ export class ScenarioSchedulerService {
|
||||
return (level, msg) => this.logger[level](`StepRun #${stepRunId} script: ${msg}`);
|
||||
}
|
||||
|
||||
// ── Job 1: flip pending runs → in_progress ─────────────────────────────────
|
||||
// ── Job: pick up pending runs and process each to completion ─────────────
|
||||
|
||||
@Interval(1000)
|
||||
async activatePendingRuns(): Promise<void> {
|
||||
async pickUpPendingRuns(): Promise<void> {
|
||||
const pending = await this.runRepo.find({ where: { status: 'pending' } });
|
||||
if (pending.length === 0) return;
|
||||
|
||||
for (const run of pending) {
|
||||
if (this.activeRuns.has(run.id)) continue;
|
||||
this.activeRuns.add(run.id);
|
||||
run.status = 'in_progress';
|
||||
await this.runRepo.save(run);
|
||||
this.logger.log(`Run #${run.id} → in_progress`);
|
||||
const traceId = crypto.randomUUID();
|
||||
void traceStorage.run({ traceId }, () => this.processRunToCompletion(run.id));
|
||||
}
|
||||
}
|
||||
|
||||
// ── Job 2: process one pending step run per tick ───────────────────────────
|
||||
|
||||
@Interval(1000)
|
||||
async processPendingStepRuns(): Promise<void> {
|
||||
if (this.isProcessingStep) return;
|
||||
this.isProcessingStep = true;
|
||||
private async processRunToCompletion(runId: number): Promise<void> {
|
||||
try {
|
||||
await this.processNextPendingStep();
|
||||
let stepRun = await this.runStepRepo.findOne({
|
||||
where: { runId, status: 'pending' },
|
||||
relations: ['scenarioStep'],
|
||||
order: { order: 'ASC' },
|
||||
});
|
||||
while (stepRun) {
|
||||
await this.executeStepRun(stepRun);
|
||||
stepRun = await this.runStepRepo.findOne({
|
||||
where: { runId, status: 'pending' },
|
||||
relations: ['scenarioStep'],
|
||||
order: { order: 'ASC' },
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.error(`Run #${runId}: unexpected error: ${(err as Error).message}`);
|
||||
await this.runRepo.update(runId, { status: 'fail' });
|
||||
} finally {
|
||||
this.isProcessingStep = false;
|
||||
this.activeRuns.delete(runId);
|
||||
}
|
||||
}
|
||||
|
||||
private async processNextPendingStep(): Promise<void> {
|
||||
const stepRun = await this.runStepRepo.findOne({
|
||||
where: { status: 'pending' },
|
||||
relations: ['scenarioStep'],
|
||||
order: { order: 'ASC' },
|
||||
});
|
||||
if (!stepRun) return;
|
||||
|
||||
private async executeStepRun(stepRun: ScenarioRunStepEntity): Promise<void> {
|
||||
const step = stepRun.scenarioStep as ScenarioStepEntity;
|
||||
|
||||
// Mark in_progress
|
||||
stepRun.status = 'in_progress';
|
||||
await this.runStepRepo.save(stepRun);
|
||||
this.logger.log(`StepRun #${stepRun.id} (run #${stepRun.runId}, step #${step.id} order=${step.order}) → in_progress`);
|
||||
@@ -247,7 +253,6 @@ export class ScenarioSchedulerService {
|
||||
if (nextStep) {
|
||||
nextStep.status = 'pending';
|
||||
await this.runStepRepo.save(nextStep);
|
||||
this.logger.log(`StepRun #${nextStep.id} → pending (next in sequence)`);
|
||||
} else {
|
||||
// No more waiting steps — check if any are still in_progress/pending (shouldn't be, but guard anyway)
|
||||
const remaining = await this.runStepRepo.count({
|
||||
|
||||
Reference in New Issue
Block a user