ars9/liqa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(scenario-scheduler): parallelize run pickup with effect, add cache cleanup

Browse Source 
- replace sequential for-loop in pickUpPendingRuns with Effect.forEach concurrency:10
- extract CONCURRENCY_LIMIT constant shared by both startup and pickup paths
- clear all run caches (creds, snippets, env, timeouts, activeRuns) on setup failure
- call failRun on setup error to leave the db run in a consistent failed state
This commit is contained in:
ars9
2026-04-17 14:48:55 +03:00
parent a199700878
commit b4cfe050fa
1 changed files with 57 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/scenario/scenario-scheduler.service.ts
+57 -33
View File
@@ -46,6 +46,7 @@ export class ScenarioSchedulerService implements OnModuleInit {
private static readonly DEFAULT_SCENARIO_TIMEOUT_SEC = 600; private static readonly DEFAULT_SCENARIO_TIMEOUT_SEC = 600;
private static readonly DEFAULT_STEP_TIMEOUT_SEC = 60; private static readonly DEFAULT_STEP_TIMEOUT_SEC = 60;
private static readonly CONCURRENCY_LIMIT = 10;
constructor( constructor(
@InjectRepository(ScenarioRunEntity) @InjectRepository(ScenarioRunEntity)
@@ -97,7 +98,7 @@ export class ScenarioSchedulerService implements OnModuleInit {
), ),
), ),
), ),
{ concurrency: 10 }, { concurrency: ScenarioSchedulerService.CONCURRENCY_LIMIT },
), ),
); );
} }
@@ -168,38 +169,61 @@ export class ScenarioSchedulerService implements OnModuleInit {
@Interval(1000) @Interval(1000)
async pickUpPendingRuns(): Promise<void> { async pickUpPendingRuns(): Promise<void> {
const pending = await this.runRepo.find({ where: { status: "pending" } }); const pending = await this.runRepo.find({ where: { status: "pending" } });
for (const run of pending) { const newRuns = pending.filter((run) => !this.activeRuns.has(run.id));
if (this.activeRuns.has(run.id)) continue; if (newRuns.length === 0) return;
this.activeRuns.add(run.id);
run.status = "in_progress"; await Effect.runPromise(
await this.runRepo.save(run); Effect.forEach(
this.logger.log(`Run #${run.id} → in_progress`); newRuns,
// Pre-load credential map for the scenario (run) => {
const credMap = await this.scenarioService this.activeRuns.add(run.id);
.buildCredentialMap(run.scenarioId) return Effect.tryPromise(async () => {
.catch(() => ({}) as Record<string, unknown>); run.status = "in_progress";
this.runCredentials.set(run.id, credMap); await this.runRepo.save(run);
// Pre-load snippet map this.logger.log(`Run #${run.id} → in_progress`);
const snippetMap = await this.snippetService const credMap = await this.scenarioService
.buildSnippetMap() .buildCredentialMap(run.scenarioId)
.catch(() => ({}) as Record<string, string>); .catch(() => ({}) as Record<string, unknown>);
this.runSnippets.set(run.id, snippetMap); this.runCredentials.set(run.id, credMap);
// Pre-load selected environment data const snippetMap = await this.snippetService
const environmentData = await this.environmentRepo .buildSnippetMap()
.findOneBy({ id: run.environmentId }) .catch(() => ({}) as Record<string, string>);
.then((env) => env?.data ?? {}) this.runSnippets.set(run.id, snippetMap);
.catch(() => ({}) as EnvironmentData); const environmentData = await this.environmentRepo
this.runEnvironments.set(run.id, environmentData); .findOneBy({ id: run.environmentId })
// Cache scenario-level timeout for the run .then((env) => env?.data ?? {})
const scenario = await this.scenarioService .catch(() => ({}) as EnvironmentData);
.findOne(run.scenarioId) this.runEnvironments.set(run.id, environmentData);
.catch(() => null); const scenario = await this.scenarioService
this.runScenarioTimeouts.set(run.id, scenario?.timeoutSeconds ?? null); .findOne(run.scenarioId)
const traceId = crypto.randomUUID(); .catch(() => null);
void traceStorage.run({ traceId }, () => this.runScenarioTimeouts.set(run.id, scenario?.timeoutSeconds ?? null);
this.processRunToCompletion(run.id), const traceId = crypto.randomUUID();
); void traceStorage.run({ traceId }, () =>
} this.processRunToCompletion(run.id),
);
}).pipe(
Effect.catchAll((err) =>
Effect.promise(async () => {
this.activeRuns.delete(run.id);
this.runCredentials.delete(run.id);
this.runSnippets.delete(run.id);
this.runEnvironments.delete(run.id);
this.runScenarioTimeouts.delete(run.id);
this.logger.error(
`Run #${run.id}: setup failed — ${String(err)}`,
);
await this.failRun(
run.id,
`Run setup failed: ${String(err)}`,
).catch(() => {});
}),
),
);
},
{ concurrency: ScenarioSchedulerService.CONCURRENCY_LIMIT },
),
);
} }
private async processRunToCompletion(runId: string): Promise<void> { private async processRunToCompletion(runId: string): Promise<void> {