Re try failed jobs

.circleci/config.yml

@@ -76,6 +76,7 @@ workflows:
                               - develop
                               - feat/ai-workflows
                               - pm-1955_2
+                              - re-try-failed-jobs
 
 
             - 'build-prod':

package.json

@@ -43,7 +43,7 @@
     "lodash": "^4.17.21",
     "multer": "^2.0.1",
     "nanoid": "~5.1.2",
-    "pg-boss": "^11.0.2",
+    "pg-boss": "^11.0.5",
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.1",
     "tc-core-library-js": "appirio-tech/tc-core-library-js.git#v3.0.1"
@@ -104,4 +104,4 @@
     "coverageDirectory": "../coverage",
     "testEnvironment": "node"
   }
-}
+}

src/shared/modules/global/queue-scheduler.service.ts

@@ -14,6 +14,7 @@ import { policies, Queue } from 'pg-boss';
 export class QueueSchedulerService implements OnModuleInit, OnModuleDestroy {
   private readonly logger: Logger = new Logger(QueueSchedulerService.name);
   private boss: PgBoss;
+  private $start;
 
   private jobsHandlersMap = new Map<
     string,
@@ -46,7 +47,7 @@ export class QueueSchedulerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
-    await this.boss.start();
+    await (this.$start = this.boss.start());
   }
 
   async onModuleDestroy() {
@@ -114,14 +115,35 @@ export class QueueSchedulerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
+    if (resolution === 'fail') {
+      // IMPORTANT!
+      // thes 4 operations will update the cache for the active singletons in the database
+      // and will allow the jobs queue to go next or retry
+      await this.boss.cancel(queueName, jobId);
+      await this.boss.getQueueStats(queueName);
+      await this.boss.supervise(queueName);
+      await this.boss.resume(queueName, jobId);
+    }
+
     if (this.jobsHandlersMap.has(jobId)) {
+      this.logger.log(
+        `Found job handler for ${jobId}. Calling with '${resolution}' resolution.`,
+      );
       this.jobsHandlersMap.get(jobId)?.call(null, resolution);
       this.jobsHandlersMap.delete(jobId);
+      this.logger.log('JobHandlers left:', [...this.jobsHandlersMap.keys()]);
     } else {
       await this.boss[resolution](queueName, jobId);
     }
 
     this.logger.log(`Job ${jobId} ${resolution} called.`);
+
+    if (resolution === 'fail') {
+      const bossJob = await this.boss.getJobById(queueName, jobId);
+      if (bossJob && bossJob.retryCount >= bossJob.retryLimit) {
+        throw new Error('Job failed! Retry limit reached!');
+      }
+    }
   }
 
   async handleWorkForQueues<T>(
@@ -135,7 +157,7 @@ export class QueueSchedulerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
-    await this.boss.start();
+    await this.$start;
     return Promise.all(
       queuesNames.map(async (queueName) => {
         const queue = await this.boss.getQueue(queueName);
@@ -155,6 +177,7 @@ export class QueueSchedulerService implements OnModuleInit, OnModuleDestroy {
     jobId: string,
     handler: (resolution?: string, result?: any) => void,
   ) {
+    this.logger.log(`Registering job handler for job ${jobId}.`);
     this.jobsHandlersMap.set(jobId, handler);
   }
 }

src/shared/modules/global/workflow-queue.handler.ts

@@ -100,14 +100,19 @@ export class WorkflowQueueHandler implements OnModuleInit {
     // return not-resolved promise,
     // this will put a pause on the job
     // until it is marked as completed via webhook call
-    return new Promise<void>((resolve, reject) => {
+    await new Promise<void>((resolve, reject) => {
       this.scheduler.registerJobHandler(
         job.id,
         (resolution: string = 'complete', result: any) => {
+          this.logger.log(
+            `Job handler called with ${resolution} and ${result}`,
+          );
           (resolution === 'fail' ? reject : resolve)(result);
         },
       );
     });
+
+    this.logger.log(`Job ${job.id} promise finished.`);
   }
 
   async handleWorkflowRunEvents(event: {
@@ -263,24 +268,6 @@ export class WorkflowQueueHandler implements OnModuleInit {
           break;
         }
 
-        if (conclusion === 'FAILURE') {
-          await this.scheduler.completeJob(
-            (aiWorkflowRun as any).workflow.gitWorkflowId,
-            aiWorkflowRun.scheduledJobId as string,
-            'fail',
-          );
-
-          this.logger.log({
-            message: 'Workflow job failed. Calling retry.',
-            aiWorkflowRunId: aiWorkflowRun.id,
-            gitRunId: event.workflow_job.run_id,
-            jobId: event.workflow_job.id,
-            status: conclusion,
-            timestamp: new Date().toISOString(),
-          });
-          break;
-        }
-
         await this.prisma.aiWorkflowRun.update({
           where: { id: aiWorkflowRun.id },
           data: {
@@ -289,13 +276,31 @@ export class WorkflowQueueHandler implements OnModuleInit {
             completedJobs: { increment: 1 },
           },
         });
-        await this.scheduler.completeJob(
-          (aiWorkflowRun as any).workflow.gitWorkflowId,
-          aiWorkflowRun.scheduledJobId as string,
-        );
+
+        try {
+          await this.scheduler.completeJob(
+            (aiWorkflowRun as any).workflow.gitWorkflowId,
+            aiWorkflowRun.scheduledJobId as string,
+            conclusion === 'FAILURE' ? 'fail' : 'complete',
+          );
+
+          if (conclusion === 'FAILURE') {
+            this.logger.log({
+              message: `Workflow job ${aiWorkflowRun.id} failed. Retrying!`,
+              aiWorkflowRunId: aiWorkflowRun.id,
+              gitRunId: event.workflow_job.run_id,
+              jobId: event.workflow_job.id,
+              status: conclusion,
+              timestamp: new Date().toISOString(),
+            });
+            return;
+          }
+        } catch (e) {
+          this.logger.log(aiWorkflowRun.id, e.message);
+        }
 
         this.logger.log({
-          message: 'Workflow job completed',
+          message: `Workflow job ${aiWorkflowRun.id} completed with conclusion: ${conclusion}`,
           aiWorkflowRunId: aiWorkflowRun.id,
           gitRunId: event.workflow_job.run_id,
           jobId: event.workflow_job.id,