From b0c02a1cdc114ef12e3813e909c61d5939dc6b4e Mon Sep 17 00:00:00 2001
From: Andrii Arsenin <ars9@pm.me>
Date: Wed, 8 Apr 2026 17:21:41 +0300
Subject: [PATCH] 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
---
 src/scenario/scenario-scheduler.service.ts | 51 ++++++++++++----------
 1 file changed, 28 insertions(+), 23 deletions(-)

diff --git a/src/scenario/scenario-scheduler.service.ts b/src/scenario/scenario-scheduler.service.ts
index a0d0890..018b493 100644
--- a/src/scenario/scenario-scheduler.service.ts
+++ b/src/scenario/scenario-scheduler.service.ts
@@ -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({